You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2019/03/19 00:46:44 UTC
[incubator-iceberg] branch master updated: Support case sensitivity
configuration (#89)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 5e37ff4 Support case sensitivity configuration (#89)
5e37ff4 is described below
commit 5e37ff424a7be3a5ac107421d41e5df8c0dbe33c
Author: Xabriel J. Collazo Mojica <xc...@adobe.com>
AuthorDate: Mon Mar 18 17:46:39 2019 -0700
Support case sensitivity configuration (#89)
---
api/src/main/java/com/netflix/iceberg/Schema.java | 45 ++++++++-
.../main/java/com/netflix/iceberg/TableScan.java | 19 +++-
.../com/netflix/iceberg/expressions/Evaluator.java | 4 +
.../expressions/InclusiveManifestEvaluator.java | 9 +-
.../expressions/InclusiveMetricsEvaluator.java | 10 +-
.../netflix/iceberg/expressions/Projections.java | 64 +++++++++++--
.../iceberg/expressions/ResidualEvaluator.java | 8 +-
.../java/com/netflix/iceberg/types/TypeUtil.java | 9 ++
.../iceberg/expressions/TestEvaluatior.java | 18 ++++
.../TestInclusiveManifestEvaluator.java | 31 +++++++
.../expressions/TestInclusiveMetricsEvaluator.java | 29 ++++++
.../netflix/iceberg/transforms/TestProjection.java | 103 +++++++++++++++++++++
.../netflix/iceberg/transforms/TestResiduals.java | 59 +++++++++++-
.../iceberg/types/TestReadabilityChecks.java | 19 ++++
.../java/com/netflix/iceberg/BaseTableScan.java | 83 ++++++++++++-----
.../java/com/netflix/iceberg/FilteredManifest.java | 22 +++--
.../java/com/netflix/iceberg/ManifestReader.java | 48 +++++++++-
...TestFilterFiles.java => TestBaseTableScan.java} | 68 ++++++--------
.../java/com/netflix/iceberg/TestFilterFiles.java | 40 ++++++++
.../com/netflix/iceberg/data/IcebergGenerics.java | 15 ++-
.../netflix/iceberg/data/TableScanIterable.java | 10 +-
.../com/netflix/iceberg/spark/SparkSchemaUtil.java | 6 +-
.../iceberg/spark/source/IcebergSource.java | 5 +-
.../com/netflix/iceberg/spark/source/Reader.java | 29 ++++--
.../iceberg/spark/source/TestFilteredScan.java | 31 +++++++
25 files changed, 673 insertions(+), 111 deletions(-)
diff --git a/api/src/main/java/com/netflix/iceberg/Schema.java b/api/src/main/java/com/netflix/iceberg/Schema.java
index 5662c12..ee21439 100644
--- a/api/src/main/java/com/netflix/iceberg/Schema.java
+++ b/api/src/main/java/com/netflix/iceberg/Schema.java
@@ -31,6 +31,7 @@ import java.io.Serializable;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
@@ -46,6 +47,7 @@ public class Schema implements Serializable {
private transient BiMap<String, Integer> aliasToId = null;
private transient Map<Integer, Types.NestedField> idToField = null;
private transient BiMap<String, Integer> nameToId = null;
+ private transient BiMap<String, Integer> lowerCaseNameToId = null;
public Schema(List<Types.NestedField> columns, Map<String, Integer> aliases) {
this.struct = Types.StructType.of(columns);
@@ -70,6 +72,13 @@ public class Schema implements Serializable {
return nameToId;
}
+ private BiMap<String, Integer> lazyLowerCaseNameToId() {
+ if (lowerCaseNameToId == null) {
+ this.lowerCaseNameToId = ImmutableBiMap.copyOf(TypeUtil.indexByLowerCaseName(struct));
+ }
+ return lowerCaseNameToId;
+ }
+
public Schema(Types.NestedField... columns) {
this(Arrays.asList(columns));
}
@@ -207,13 +216,47 @@ public class Schema implements Serializable {
* @return a projection schema from this schema, by name
*/
public Schema select(Collection<String> names) {
+ return internalSelect(names, true);
+ }
+
+ /**
+ * Creates a projection schema for a subset of columns, selected by case insensitive name
+ * <p>
+ * Names that identify nested fields will select part or all of the field's top-level column.
+ *
+ * @param names a List of String names for selected columns
+ * @return a projection schema from this schema, by name
+ */
+ public Schema caseInsensitiveSelect(String... names) {
+ return caseInsensitiveSelect(Arrays.asList(names));
+ }
+
+ /**
+ * Creates a projection schema for a subset of columns, selected by case insensitive name
+ * <p>
+ * Names that identify nested fields will select part or all of the field's top-level column.
+ *
+ * @param names a List of String names for selected columns
+ * @return a projection schema from this schema, by name
+ */
+ public Schema caseInsensitiveSelect(Collection<String> names) {
+ return internalSelect(names, false);
+ }
+
+ private Schema internalSelect(Collection<String> names, boolean caseSensitive) {
if (names.contains(ALL_COLUMNS)) {
return this;
}
Set<Integer> selected = Sets.newHashSet();
for (String name : names) {
- Integer id = lazyNameToId().get(name);
+ Integer id;
+ if (caseSensitive) {
+ id = lazyNameToId().get(name);
+ } else {
+ id = lazyLowerCaseNameToId().get(name.toLowerCase(Locale.ROOT));
+ }
+
if (id != null) {
selected.add(id);
}
diff --git a/api/src/main/java/com/netflix/iceberg/TableScan.java b/api/src/main/java/com/netflix/iceberg/TableScan.java
index 11df3be..eaab978 100644
--- a/api/src/main/java/com/netflix/iceberg/TableScan.java
+++ b/api/src/main/java/com/netflix/iceberg/TableScan.java
@@ -67,6 +67,15 @@ public interface TableScan {
TableScan project(Schema schema);
/**
+ * Create a new {@link TableScan} from this that, if data columns where selected
+ * via {@link #select(java.util.Collection)}, controls whether the match to the schema will be done
+ * with case sensitivity.
+ *
+ * @return a new scan based on this with case sensitivity as stated
+ */
+ TableScan caseSensitive(boolean caseSensitive);
+
+ /**
* Create a new {@link TableScan} from this that will read the given data columns. This produces
* an expected schema that includes all fields that are either selected or used by this scan's
* filter expression.
@@ -83,8 +92,8 @@ public interface TableScan {
* an expected schema that includes all fields that are either selected or used by this scan's
* filter expression.
*
- * @param columns column names from the manifest file schema
- * @return a new scan based on this with the given manifest columns
+ * @param columns column names from the table's schema
+ * @return a new scan based on this with the given projection columns
*/
TableScan select(Collection<String> columns);
@@ -136,4 +145,10 @@ public interface TableScan {
* @return this scan's filter expression
*/
Expression filter();
+
+ /**
+ * Returns whether this scan should apply column name case sensitiveness as per {@link #caseSensitive(boolean)}.
+ * @return true if case sensitive, false otherwise.
+ */
+ boolean isCaseSensitive();
}
diff --git a/api/src/main/java/com/netflix/iceberg/expressions/Evaluator.java b/api/src/main/java/com/netflix/iceberg/expressions/Evaluator.java
index fc6eacd..dacaba9 100644
--- a/api/src/main/java/com/netflix/iceberg/expressions/Evaluator.java
+++ b/api/src/main/java/com/netflix/iceberg/expressions/Evaluator.java
@@ -47,6 +47,10 @@ public class Evaluator implements Serializable {
this.expr = Binder.bind(struct, unbound, true);
}
+ public Evaluator(Types.StructType struct, Expression unbound, boolean caseSensitive) {
+ this.expr = Binder.bind(struct, unbound, caseSensitive);
+ }
+
public boolean eval(StructLike data) {
return visitor().eval(data);
}
diff --git a/api/src/main/java/com/netflix/iceberg/expressions/InclusiveManifestEvaluator.java b/api/src/main/java/com/netflix/iceberg/expressions/InclusiveManifestEvaluator.java
index 6493273..203ae78 100644
--- a/api/src/main/java/com/netflix/iceberg/expressions/InclusiveManifestEvaluator.java
+++ b/api/src/main/java/com/netflix/iceberg/expressions/InclusiveManifestEvaluator.java
@@ -53,8 +53,15 @@ public class InclusiveManifestEvaluator {
}
public InclusiveManifestEvaluator(PartitionSpec spec, Expression rowFilter) {
+ this(spec, rowFilter, true);
+ }
+
+ public InclusiveManifestEvaluator(PartitionSpec spec, Expression rowFilter, boolean caseSensitive) {
this.struct = spec.partitionType();
- this.expr = Binder.bind(struct, rewriteNot(Projections.inclusive(spec).project(rowFilter)), true);
+ this.expr = Binder.bind(
+ struct,
+ rewriteNot(Projections.inclusive(spec, caseSensitive).project(rowFilter)),
+ caseSensitive);
}
/**
diff --git a/api/src/main/java/com/netflix/iceberg/expressions/InclusiveMetricsEvaluator.java b/api/src/main/java/com/netflix/iceberg/expressions/InclusiveMetricsEvaluator.java
index 54cc0be..b8b0c48 100644
--- a/api/src/main/java/com/netflix/iceberg/expressions/InclusiveMetricsEvaluator.java
+++ b/api/src/main/java/com/netflix/iceberg/expressions/InclusiveMetricsEvaluator.java
@@ -44,6 +44,7 @@ public class InclusiveMetricsEvaluator {
private final Schema schema;
private final StructType struct;
private final Expression expr;
+ private final boolean caseSensitive;
private transient ThreadLocal<MetricsEvalVisitor> visitors = null;
private MetricsEvalVisitor visitor() {
@@ -53,10 +54,15 @@ public class InclusiveMetricsEvaluator {
return visitors.get();
}
- public InclusiveMetricsEvaluator(Schema schema, Expression unbound) {
+ InclusiveMetricsEvaluator(Schema schema, Expression unbound) {
+ this(schema, unbound, true);
+ }
+
+ public InclusiveMetricsEvaluator(Schema schema, Expression unbound, boolean caseSensitive) {
this.schema = schema;
this.struct = schema.asStruct();
- this.expr = Binder.bind(struct, rewriteNot(unbound), true);
+ this.caseSensitive = caseSensitive;
+ this.expr = Binder.bind(struct, rewriteNot(unbound), caseSensitive);
}
/**
diff --git a/api/src/main/java/com/netflix/iceberg/expressions/Projections.java b/api/src/main/java/com/netflix/iceberg/expressions/Projections.java
index b811e27..d54a33f 100644
--- a/api/src/main/java/com/netflix/iceberg/expressions/Projections.java
+++ b/api/src/main/java/com/netflix/iceberg/expressions/Projections.java
@@ -54,7 +54,8 @@ public class Projections {
}
/**
- * Creates an inclusive {@code ProjectionEvaluator} for the {@link PartitionSpec spec}.
+ * Creates an inclusive {@code ProjectionEvaluator} for the {@link PartitionSpec spec}, defaulting
+ * to case sensitive mode.
* <p>
* An evaluator is used to project expressions for a table's data rows into expressions on the
* table's partition values. The evaluator returned by this function is inclusive and will build
@@ -69,11 +70,32 @@ public class Projections {
* @see Transform#project(String, BoundPredicate) Inclusive transform used for each predicate
*/
public static ProjectionEvaluator inclusive(PartitionSpec spec) {
- return new InclusiveProjection(spec);
+ return new InclusiveProjection(spec, true);
}
/**
- * Creates a strict {@code ProjectionEvaluator} for the {@link PartitionSpec spec}.
+ * Creates an inclusive {@code ProjectionEvaluator} for the {@link PartitionSpec spec}.
+ * <p>
+ * An evaluator is used to project expressions for a table's data rows into expressions on the
+ * table's partition values. The evaluator returned by this function is inclusive and will build
+ * expressions with the following guarantee: if the original expression matches a row, then the
+ * projected expression will match that row's partition.
+ * <p>
+ * Each predicate in the expression is projected using
+ * {@link Transform#project(String, BoundPredicate)}.
+ *
+ * @param spec a partition spec
+ * @param caseSensitive whether the Projection should consider case sensitivity on column names or not.
+ * @return an inclusive projection evaluator for the partition spec
+ * @see Transform#project(String, BoundPredicate) Inclusive transform used for each predicate
+ */
+ public static ProjectionEvaluator inclusive(PartitionSpec spec, boolean caseSensitive) {
+ return new InclusiveProjection(spec, caseSensitive);
+ }
+
+ /**
+ * Creates a strict {@code ProjectionEvaluator} for the {@link PartitionSpec spec}, defaulting
+ * to case sensitive mode.
* <p>
* An evaluator is used to project expressions for a table's data rows into expressions on the
* table's partition values. The evaluator returned by this function is strict and will build
@@ -88,14 +110,36 @@ public class Projections {
* @see Transform#projectStrict(String, BoundPredicate) Strict transform used for each predicate
*/
public static ProjectionEvaluator strict(PartitionSpec spec) {
- return new StrictProjection(spec);
+ return new StrictProjection(spec, true);
+ }
+
+ /**
+ * Creates a strict {@code ProjectionEvaluator} for the {@link PartitionSpec spec}.
+ * <p>
+ * An evaluator is used to project expressions for a table's data rows into expressions on the
+ * table's partition values. The evaluator returned by this function is strict and will build
+ * expressions with the following guarantee: if the projected expression matches a partition,
+ * then the original expression will match all rows in that partition.
+ * <p>
+ * Each predicate in the expression is projected using
+ * {@link Transform#projectStrict(String, BoundPredicate)}.
+ *
+ * @param spec a partition spec
+ * @param caseSensitive whether the Projection should consider case sensitivity on column names or not.
+ * @return a strict projection evaluator for the partition spec
+ * @see Transform#projectStrict(String, BoundPredicate) Strict transform used for each predicate
+ */
+ public static ProjectionEvaluator strict(PartitionSpec spec, boolean caseSensitive) {
+ return new StrictProjection(spec, caseSensitive);
}
private static class BaseProjectionEvaluator extends ProjectionEvaluator {
final PartitionSpec spec;
+ final boolean caseSensitive;
- private BaseProjectionEvaluator(PartitionSpec spec) {
+ private BaseProjectionEvaluator(PartitionSpec spec, boolean caseSensitive) {
this.spec = spec;
+ this.caseSensitive = caseSensitive;
}
@Override
@@ -135,7 +179,7 @@ public class Projections {
@Override
public <T> Expression predicate(UnboundPredicate<T> pred) {
- Expression bound = pred.bind(spec.schema().asStruct(), true);
+ Expression bound = pred.bind(spec.schema().asStruct(), caseSensitive);
if (bound instanceof BoundPredicate) {
return predicate((BoundPredicate<?>) bound);
@@ -146,8 +190,8 @@ public class Projections {
}
private static class InclusiveProjection extends BaseProjectionEvaluator {
- private InclusiveProjection(PartitionSpec spec) {
- super(spec);
+ private InclusiveProjection(PartitionSpec spec, boolean caseSensitive) {
+ super(spec, caseSensitive);
}
@Override
@@ -171,8 +215,8 @@ public class Projections {
}
private static class StrictProjection extends BaseProjectionEvaluator {
- private StrictProjection(PartitionSpec spec) {
- super(spec);
+ private StrictProjection(PartitionSpec spec, boolean caseSensitive) {
+ super(spec, caseSensitive);
}
@Override
diff --git a/api/src/main/java/com/netflix/iceberg/expressions/ResidualEvaluator.java b/api/src/main/java/com/netflix/iceberg/expressions/ResidualEvaluator.java
index 610bdc5..dff24e7 100644
--- a/api/src/main/java/com/netflix/iceberg/expressions/ResidualEvaluator.java
+++ b/api/src/main/java/com/netflix/iceberg/expressions/ResidualEvaluator.java
@@ -49,6 +49,7 @@ import java.util.Comparator;
public class ResidualEvaluator implements Serializable {
private final PartitionSpec spec;
private final Expression expr;
+ private final boolean caseSensitive;
private transient ThreadLocal<ResidualVisitor> visitors = null;
private ResidualVisitor visitor() {
@@ -58,9 +59,10 @@ public class ResidualEvaluator implements Serializable {
return visitors.get();
}
- public ResidualEvaluator(PartitionSpec spec, Expression expr) {
+ public ResidualEvaluator(PartitionSpec spec, Expression expr, boolean caseSensitive) {
this.spec = spec;
this.expr = expr;
+ this.caseSensitive = caseSensitive;
}
/**
@@ -170,7 +172,7 @@ public class ResidualEvaluator implements Serializable {
.projectStrict(part.name(), pred);
if (strictProjection != null) {
- Expression bound = strictProjection.bind(spec.partitionType(), true);
+ Expression bound = strictProjection.bind(spec.partitionType(), caseSensitive);
if (bound instanceof BoundPredicate) {
// the predicate methods will evaluate and return alwaysTrue or alwaysFalse
return super.predicate((BoundPredicate<?>) bound);
@@ -184,7 +186,7 @@ public class ResidualEvaluator implements Serializable {
@Override
public <T> Expression predicate(UnboundPredicate<T> pred) {
- Expression bound = pred.bind(spec.schema().asStruct(), true);
+ Expression bound = pred.bind(spec.schema().asStruct(), caseSensitive);
if (bound instanceof BoundPredicate) {
Expression boundResidual = predicate((BoundPredicate<?>) bound);
diff --git a/api/src/main/java/com/netflix/iceberg/types/TypeUtil.java b/api/src/main/java/com/netflix/iceberg/types/TypeUtil.java
index 6f2ca78..81bb422 100644
--- a/api/src/main/java/com/netflix/iceberg/types/TypeUtil.java
+++ b/api/src/main/java/com/netflix/iceberg/types/TypeUtil.java
@@ -24,9 +24,11 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import com.netflix.iceberg.Schema;
import java.util.LinkedList;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.function.Predicate;
@@ -79,6 +81,13 @@ public class TypeUtil {
return visit(struct, new IndexByName());
}
+ public static Map<String, Integer> indexByLowerCaseName(Types.StructType struct) {
+ Map<String, Integer> indexByLowerCaseName = Maps.newHashMap();
+ indexByName(struct).forEach( (name, integer) ->
+ indexByLowerCaseName.put(name.toLowerCase(Locale.ROOT), integer));
+ return indexByLowerCaseName;
+ }
+
public static Map<Integer, Types.NestedField> indexById(Types.StructType struct) {
return visit(struct, new IndexById());
}
diff --git a/api/src/test/java/com/netflix/iceberg/expressions/TestEvaluatior.java b/api/src/test/java/com/netflix/iceberg/expressions/TestEvaluatior.java
index 4723f83..0c08903 100644
--- a/api/src/test/java/com/netflix/iceberg/expressions/TestEvaluatior.java
+++ b/api/src/test/java/com/netflix/iceberg/expressions/TestEvaluatior.java
@@ -20,6 +20,7 @@
package com.netflix.iceberg.expressions;
import com.netflix.iceberg.TestHelpers;
+import com.netflix.iceberg.exceptions.ValidationException;
import com.netflix.iceberg.types.Types;
import com.netflix.iceberg.types.Types.StructType;
import org.apache.avro.util.Utf8;
@@ -146,6 +147,23 @@ public class TestEvaluatior {
}
@Test
+ public void testCaseInsensitiveNot() {
+ Evaluator evaluator = new Evaluator(STRUCT, not(equal("X", 7)), false);
+ Assert.assertFalse("not(7 == 7) => false", evaluator.eval(TestHelpers.Row.of(7)));
+ Assert.assertTrue("not(8 == 7) => false", evaluator.eval(TestHelpers.Row.of(8)));
+ }
+
+ @Test
+ public void testCaseSensitiveNot() {
+ TestHelpers.assertThrows(
+ "X != x when case sensitivity is on",
+ ValidationException.class,
+ "Cannot find field 'X' in struct",
+ () -> { new Evaluator(STRUCT, not(equal("X", 7)), true); }
+ );
+ }
+
+ @Test
public void testCharSeqValue() {
StructType struct = StructType.of(required(34, "s", Types.StringType.get()));
Evaluator evaluator = new Evaluator(struct, equal("s", "abc"));
diff --git a/api/src/test/java/com/netflix/iceberg/expressions/TestInclusiveManifestEvaluator.java b/api/src/test/java/com/netflix/iceberg/expressions/TestInclusiveManifestEvaluator.java
index f92f700..b65a61c 100644
--- a/api/src/test/java/com/netflix/iceberg/expressions/TestInclusiveManifestEvaluator.java
+++ b/api/src/test/java/com/netflix/iceberg/expressions/TestInclusiveManifestEvaluator.java
@@ -292,4 +292,35 @@ public class TestInclusiveManifestEvaluator {
shouldRead = new InclusiveManifestEvaluator(SPEC, not(equal("id", 85))).eval(FILE);
Assert.assertTrue("Should read: id above upper bound", shouldRead);
}
+
+ @Test
+ public void testCaseInsensitiveIntegerNotEqRewritten() {
+ boolean shouldRead = new InclusiveManifestEvaluator(SPEC, not(equal("ID", 5)), false).eval(FILE);
+ Assert.assertTrue("Should read: id below lower bound", shouldRead);
+
+ shouldRead = new InclusiveManifestEvaluator(SPEC, not(equal("ID", 29)), false).eval(FILE);
+ Assert.assertTrue("Should read: id below lower bound", shouldRead);
+
+ shouldRead = new InclusiveManifestEvaluator(SPEC, not(equal("ID", 30)), false).eval(FILE);
+ Assert.assertTrue("Should read: id equal to lower bound", shouldRead);
+
+ shouldRead = new InclusiveManifestEvaluator(SPEC, not(equal("ID", 75)), false).eval(FILE);
+ Assert.assertTrue("Should read: id between lower and upper bounds", shouldRead);
+
+ shouldRead = new InclusiveManifestEvaluator(SPEC, not(equal("ID", 79)), false).eval(FILE);
+ Assert.assertTrue("Should read: id equal to upper bound", shouldRead);
+
+ shouldRead = new InclusiveManifestEvaluator(SPEC, not(equal("ID", 80)), false).eval(FILE);
+ Assert.assertTrue("Should read: id above upper bound", shouldRead);
+
+ shouldRead = new InclusiveManifestEvaluator(SPEC, not(equal("ID", 85)), false).eval(FILE);
+ Assert.assertTrue("Should read: id above upper bound", shouldRead);
+ }
+
+ @Test
+ public void testCaseSensitiveIntegerNotEqRewritten() {
+ TestHelpers.assertThrows("Should complain about missing column in expression",
+ ValidationException.class, "Cannot find field 'ID'",
+ () -> new InclusiveManifestEvaluator(SPEC, not(equal("ID", 5)), true).eval(FILE));
+ }
}
diff --git a/api/src/test/java/com/netflix/iceberg/expressions/TestInclusiveMetricsEvaluator.java b/api/src/test/java/com/netflix/iceberg/expressions/TestInclusiveMetricsEvaluator.java
index bbc96ce..12b8d0d 100644
--- a/api/src/test/java/com/netflix/iceberg/expressions/TestInclusiveMetricsEvaluator.java
+++ b/api/src/test/java/com/netflix/iceberg/expressions/TestInclusiveMetricsEvaluator.java
@@ -312,4 +312,33 @@ public class TestInclusiveMetricsEvaluator {
shouldRead = new InclusiveMetricsEvaluator(SCHEMA, not(equal("id", 85))).eval(FILE);
Assert.assertTrue("Should read: id above upper bound", shouldRead);
}
+
+ @Test
+ public void testCaseInsensitiveIntegerNotEqRewritten() {
+ boolean shouldRead = new InclusiveMetricsEvaluator(SCHEMA, not(equal("ID", 5)), false).eval(FILE);
+ Assert.assertTrue("Should read: id below lower bound", shouldRead);
+
+ shouldRead = new InclusiveMetricsEvaluator(SCHEMA, not(equal("ID", 29)), false).eval(FILE);
+ Assert.assertTrue("Should read: id below lower bound", shouldRead);
+
+ shouldRead = new InclusiveMetricsEvaluator(SCHEMA, not(equal("ID", 30)), false).eval(FILE);
+ Assert.assertTrue("Should read: id equal to lower bound", shouldRead);
+
+ shouldRead = new InclusiveMetricsEvaluator(SCHEMA, not(equal("ID", 75)), false).eval(FILE);
+ Assert.assertTrue("Should read: id between lower and upper bounds", shouldRead);
+
+ shouldRead = new InclusiveMetricsEvaluator(SCHEMA, not(equal("ID", 79)), false).eval(FILE);
+ Assert.assertTrue("Should read: id equal to upper bound", shouldRead);
+
+ shouldRead = new InclusiveMetricsEvaluator(SCHEMA, not(equal("ID", 80)), false).eval(FILE);
+ Assert.assertTrue("Should read: id above upper bound", shouldRead);
+
+ shouldRead = new InclusiveMetricsEvaluator(SCHEMA, not(equal("ID", 85)), false).eval(FILE);
+ Assert.assertTrue("Should read: id above upper bound", shouldRead);
+ }
+
+ @Test(expected = ValidationException.class)
+ public void testCaseSensitiveIntegerNotEqRewritten() {
+ boolean shouldRead = new InclusiveMetricsEvaluator(SCHEMA, not(equal("ID", 5)), true).eval(FILE);
+ }
}
diff --git a/api/src/test/java/com/netflix/iceberg/transforms/TestProjection.java b/api/src/test/java/com/netflix/iceberg/transforms/TestProjection.java
index f08fe3f..546cb85 100644
--- a/api/src/test/java/com/netflix/iceberg/transforms/TestProjection.java
+++ b/api/src/test/java/com/netflix/iceberg/transforms/TestProjection.java
@@ -21,6 +21,7 @@ package com.netflix.iceberg.transforms;
import com.google.common.collect.Lists;
import com.netflix.iceberg.Schema;
+import com.netflix.iceberg.exceptions.ValidationException;
import com.netflix.iceberg.expressions.BoundPredicate;
import com.netflix.iceberg.expressions.Expression;
import com.netflix.iceberg.expressions.Expressions;
@@ -35,6 +36,7 @@ import java.util.List;
import static com.netflix.iceberg.TestHelpers.assertAndUnwrap;
import static com.netflix.iceberg.TestHelpers.assertAndUnwrapUnbound;
+import static com.netflix.iceberg.TestHelpers.assertThrows;
import static com.netflix.iceberg.expressions.Expressions.and;
import static com.netflix.iceberg.expressions.Expressions.equal;
import static com.netflix.iceberg.expressions.Expressions.greaterThanOrEqual;
@@ -87,6 +89,56 @@ public class TestProjection {
}
@Test
+ public void testCaseInsensitiveIdentityProjection() {
+ List<UnboundPredicate<?>> predicates = Lists.newArrayList(
+ Expressions.notNull("ID"),
+ Expressions.isNull("ID"),
+ Expressions.lessThan("ID", 100),
+ Expressions.lessThanOrEqual("ID", 101),
+ Expressions.greaterThan("ID", 102),
+ Expressions.greaterThanOrEqual("ID", 103),
+ Expressions.equal("ID", 104),
+ Expressions.notEqual("ID", 105)
+ );
+
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .identity("id")
+ .build();
+
+ for (UnboundPredicate<?> predicate : predicates) {
+ // get the projected predicate
+ Expression expr = Projections.inclusive(spec, false).project(predicate);
+ UnboundPredicate<?> projected = assertAndUnwrapUnbound(expr);
+
+ // check inclusive the bound predicate to ensure the types are correct
+ BoundPredicate<?> bound = assertAndUnwrap(predicate.bind(spec.schema().asStruct(), false));
+
+ Assert.assertEquals("Field name should match partition struct field",
+ "id", projected.ref().name());
+ Assert.assertEquals("Operation should match", bound.op(), projected.op());
+
+ if (bound.literal() != null) {
+ Assert.assertEquals("Literal should be equal",
+ bound.literal().value(), projected.literal().value());
+ } else {
+ Assert.assertNull("Literal should be null", projected.literal());
+ }
+ }
+ }
+
+ @Test
+ public void testCaseSensitiveIdentityProjection() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .identity("id")
+ .build();
+
+ assertThrows("X != x when case sensitivity is on",
+ ValidationException.class,
+ "Cannot find field 'ID' in struct",
+ () -> { Projections.inclusive(spec, true).project(Expressions.notNull("ID")); });
+ }
+
+ @Test
public void testStrictIdentityProjection() {
List<UnboundPredicate<?>> predicates = Lists.newArrayList(
Expressions.notNull("id"),
@@ -125,6 +177,57 @@ public class TestProjection {
}
@Test
+ public void testCaseInsensitiveStrictIdentityProjection() {
+ List<UnboundPredicate<?>> predicates = Lists.newArrayList(
+ Expressions.notNull("ID"),
+ Expressions.isNull("ID"),
+ Expressions.lessThan("ID", 100),
+ Expressions.lessThanOrEqual("ID", 101),
+ Expressions.greaterThan("ID", 102),
+ Expressions.greaterThanOrEqual("ID", 103),
+ Expressions.equal("ID", 104),
+ Expressions.notEqual("ID", 105)
+ );
+
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .identity("id")
+ .build();
+
+ for (UnboundPredicate<?> predicate : predicates) {
+ // get the projected predicate
+ Expression expr = Projections.strict(spec, false).project(predicate);
+ UnboundPredicate<?> projected = assertAndUnwrapUnbound(expr);
+
+ // check inclusive the bound predicate to ensure the types are correct
+ BoundPredicate<?> bound = assertAndUnwrap(predicate.bind(spec.schema().asStruct(), false));
+
+ Assert.assertEquals("Field name should match partition struct field",
+ "id", projected.ref().name());
+ Assert.assertEquals("Operation should match", bound.op(), projected.op());
+
+ if (bound.literal() != null) {
+ Assert.assertEquals("Literal should be equal",
+ bound.literal().value(), projected.literal().value());
+ } else {
+ Assert.assertNull("Literal should be null", projected.literal());
+ }
+ }
+ }
+
+ @Test
+ public void testCaseSensitiveStrictIdentityProjection() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA)
+ .identity("id")
+ .build();
+
+ assertThrows(
+ "X != x when case sensitivity is on",
+ ValidationException.class,
+ "Cannot find field 'ID' in struct",
+ () -> { Projections.strict(spec, true).project(Expressions.notNull("ID")); });
+ }
+
+ @Test
public void testBadSparkPartitionFilter() {
// this tests a case that results in a full table scan in Spark with Hive tables. because the
// hour field is not a partition, mixing it with partition columns in the filter expression
diff --git a/api/src/test/java/com/netflix/iceberg/transforms/TestResiduals.java b/api/src/test/java/com/netflix/iceberg/transforms/TestResiduals.java
index 27b1c11..ec0cbe7 100644
--- a/api/src/test/java/com/netflix/iceberg/transforms/TestResiduals.java
+++ b/api/src/test/java/com/netflix/iceberg/transforms/TestResiduals.java
@@ -21,6 +21,7 @@ package com.netflix.iceberg.transforms;
import com.netflix.iceberg.Schema;
import com.netflix.iceberg.TestHelpers.Row;
+import com.netflix.iceberg.exceptions.ValidationException;
import com.netflix.iceberg.expressions.Expression;
import com.netflix.iceberg.expressions.ResidualEvaluator;
import com.netflix.iceberg.expressions.UnboundPredicate;
@@ -55,7 +56,8 @@ public class TestResiduals {
ResidualEvaluator resEval = new ResidualEvaluator(spec, or(or(
and(lessThan("dateint", 20170815), greaterThan("dateint", 20170801)),
and(equal("dateint", 20170815), lessThan("hour", 12))),
- and(equal("dateint", 20170801), greaterThan("hour", 11)))
+ and(equal("dateint", 20170801), greaterThan("hour", 11))),
+ true
);
// equal to the upper date bound
@@ -80,4 +82,59 @@ public class TestResiduals {
residual = resEval.residualFor(Row.of(20170817));
Assert.assertEquals("Residual should be alwaysFalse", alwaysFalse(), residual);
}
+
+ @Test
+ public void testCaseInsensitiveIdentityTransformResiduals() {
+ Schema schema = new Schema(
+ Types.NestedField.optional(50, "dateint", Types.IntegerType.get()),
+ Types.NestedField.optional(51, "hour", Types.IntegerType.get()));
+
+ PartitionSpec spec = PartitionSpec.builderFor(schema)
+ .identity("dateint")
+ .build();
+
+ ResidualEvaluator resEval = new ResidualEvaluator(spec, or(or(
+ and(lessThan("DATEINT", 20170815), greaterThan("dateint", 20170801)),
+ and(equal("dateint", 20170815), lessThan("HOUR", 12))),
+ and(equal("DateInt", 20170801), greaterThan("hOUr", 11))),
+ false);
+
+ // equal to the upper date bound
+ Expression residual = resEval.residualFor(Row.of(20170815));
+ UnboundPredicate<?> unbound = assertAndUnwrapUnbound(residual);
+ Assert.assertEquals("Residual should be hour < 12", LT, unbound.op());
+ Assert.assertEquals("Residual should be hour < 12", "HOUR", unbound.ref().name());
+ Assert.assertEquals("Residual should be hour < 12", 12, unbound.literal().value());
+
+ // equal to the lower date bound
+ residual = resEval.residualFor(Row.of(20170801));
+ unbound = assertAndUnwrapUnbound(residual);
+ Assert.assertEquals("Residual should be hour > 11", GT, unbound.op());
+ Assert.assertEquals("Residual should be hour > 11", "hOUr", unbound.ref().name());
+ Assert.assertEquals("Residual should be hour > 11", 11, unbound.literal().value());
+
+ // inside the date range
+ residual = resEval.residualFor(Row.of(20170812));
+ Assert.assertEquals("Residual should be alwaysTrue", alwaysTrue(), residual);
+
+ // outside the date range
+ residual = resEval.residualFor(Row.of(20170817));
+ Assert.assertEquals("Residual should be alwaysFalse", alwaysFalse(), residual);
+ }
+
+ @Test(expected = ValidationException.class)
+ public void testCaseSensitiveIdentityTransformResiduals() {
+ Schema schema = new Schema(
+ Types.NestedField.optional(50, "dateint", Types.IntegerType.get()),
+ Types.NestedField.optional(51, "hour", Types.IntegerType.get())
+ );
+
+ PartitionSpec spec = PartitionSpec.builderFor(schema)
+ .identity("dateint")
+ .build();
+
+ ResidualEvaluator resEval = new ResidualEvaluator(spec, lessThan("DATEINT", 20170815), true);
+
+ resEval.residualFor(Row.of(20170815));
+ }
}
diff --git a/api/src/test/java/com/netflix/iceberg/types/TestReadabilityChecks.java b/api/src/test/java/com/netflix/iceberg/types/TestReadabilityChecks.java
index 6b74338..90ccbed 100644
--- a/api/src/test/java/com/netflix/iceberg/types/TestReadabilityChecks.java
+++ b/api/src/test/java/com/netflix/iceberg/types/TestReadabilityChecks.java
@@ -373,4 +373,23 @@ public class TestReadabilityChecks {
List<String> errors = CheckCompatibility.readCompatibilityErrors(read, write);
Assert.assertEquals("Should produce no error messages", 0, errors.size());
}
+
+ @Test
+ public void testCaseInsensitiveSchemaProjection() {
+ Schema schema = new Schema(
+ Types.NestedField.required(0, "id", Types.LongType.get()),
+ Types.NestedField.optional(5, "locations", Types.MapType.ofOptional(6, 7,
+ Types.StringType.get(),
+ Types.StructType.of(
+ Types.NestedField.required(1, "lat", Types.FloatType.get()),
+ Types.NestedField.required(2, "long", Types.FloatType.get())
+ )
+ ))
+ );
+
+ Assert.assertNotNull(schema.caseInsensitiveSelect("ID").findField(0));
+ Assert.assertNotNull(schema.caseInsensitiveSelect("loCATIONs").findField(5));
+ Assert.assertNotNull(schema.caseInsensitiveSelect("LoCaTiOnS.LaT").findField(1));
+ Assert.assertNotNull(schema.caseInsensitiveSelect("locations.LONG").findField(2));
+ }
}
diff --git a/core/src/main/java/com/netflix/iceberg/BaseTableScan.java b/core/src/main/java/com/netflix/iceberg/BaseTableScan.java
index f8c4215..4a39409 100644
--- a/core/src/main/java/com/netflix/iceberg/BaseTableScan.java
+++ b/core/src/main/java/com/netflix/iceberg/BaseTableScan.java
@@ -74,17 +74,22 @@ class BaseTableScan implements TableScan {
private final Long snapshotId;
private final Schema schema;
private final Expression rowFilter;
+ private final boolean caseSensitive;
+ private final Collection<String> selectedColumns;
BaseTableScan(TableOperations ops, Table table) {
- this(ops, table, null, table.schema(), Expressions.alwaysTrue());
+ this(ops, table, null, table.schema(), Expressions.alwaysTrue(), true, null);
}
- private BaseTableScan(TableOperations ops, Table table, Long snapshotId, Schema schema, Expression rowFilter) {
+ private BaseTableScan(TableOperations ops, Table table, Long snapshotId, Schema schema,
+ Expression rowFilter, boolean caseSensitive, Collection<String> selectedColumns) {
this.ops = ops;
this.table = table;
this.snapshotId = snapshotId;
this.schema = schema;
this.rowFilter = rowFilter;
+ this.caseSensitive = caseSensitive;
+ this.selectedColumns = selectedColumns;
}
@Override
@@ -98,7 +103,7 @@ class BaseTableScan implements TableScan {
"Cannot override snapshot, already set to id=%s", snapshotId);
Preconditions.checkArgument(ops.current().snapshot(snapshotId) != null,
"Cannot find snapshot with ID %s", snapshotId);
- return new BaseTableScan(ops, table, snapshotId, schema, rowFilter);
+ return new BaseTableScan(ops, table, snapshotId, schema, rowFilter, caseSensitive, selectedColumns);
}
@Override
@@ -122,28 +127,23 @@ class BaseTableScan implements TableScan {
}
public TableScan project(Schema schema) {
- return new BaseTableScan(ops, table, snapshotId, schema, rowFilter);
+ return new BaseTableScan(ops, table, snapshotId, schema, rowFilter, caseSensitive, selectedColumns);
}
@Override
- public TableScan select(Collection<String> columns) {
- Set<Integer> requiredFieldIds = Sets.newHashSet();
-
- // all of the filter columns are required
- requiredFieldIds.addAll(
- Binder.boundReferences(table.schema().asStruct(), Collections.singletonList(rowFilter), true));
-
- // all of the projection columns are required
- requiredFieldIds.addAll(TypeUtil.getProjectedIds(table.schema().select(columns)));
-
- Schema projection = TypeUtil.select(table.schema(), requiredFieldIds);
+ public TableScan caseSensitive(boolean caseSensitive) {
+ return new BaseTableScan(ops, table, snapshotId, schema, rowFilter, caseSensitive, selectedColumns);
+ }
- return new BaseTableScan(ops, table, snapshotId, projection, rowFilter);
+ @Override
+ public TableScan select(Collection<String> columns) {
+ return new BaseTableScan(ops, table, snapshotId, schema, rowFilter, caseSensitive, columns);
}
@Override
public TableScan filter(Expression expr) {
- return new BaseTableScan(ops, table, snapshotId, schema, Expressions.and(rowFilter, expr));
+ return new BaseTableScan(ops, table, snapshotId, schema, Expressions.and(rowFilter, expr),
+ caseSensitive, selectedColumns);
}
private final LoadingCache<Integer, InclusiveManifestEvaluator> EVAL_CACHE = CacheBuilder
@@ -152,7 +152,7 @@ class BaseTableScan implements TableScan {
@Override
public InclusiveManifestEvaluator load(Integer specId) {
PartitionSpec spec = ops.current().spec(specId);
- return new InclusiveManifestEvaluator(spec, rowFilter);
+ return new InclusiveManifestEvaluator(spec, rowFilter, caseSensitive);
}
});
@@ -168,7 +168,7 @@ class BaseTableScan implements TableScan {
rowFilter);
Listeners.notifyAll(
- new ScanEvent(table.toString(), snapshot.snapshotId(), rowFilter, schema));
+ new ScanEvent(table.toString(), snapshot.snapshotId(), rowFilter, schema()));
Iterable<ManifestFile> matchingManifests = Iterables.filter(snapshot.manifests(),
manifest -> EVAL_CACHE.getUnchecked(manifest.partitionSpecId()).eval(manifest));
@@ -177,11 +177,13 @@ class BaseTableScan implements TableScan {
Iterable<Iterable<FileScanTask>> readers = Iterables.transform(
matchingManifests,
manifest -> {
- ManifestReader reader = ManifestReader.read(ops.io().newInputFile(manifest.path()));
+ ManifestReader reader = ManifestReader
+ .read(ops.io().newInputFile(manifest.path()))
+ .caseSensitive(caseSensitive);
toClose.add(reader);
String schemaString = SchemaParser.toJson(reader.spec().schema());
String specString = PartitionSpecParser.toJson(reader.spec());
- ResidualEvaluator residuals = new ResidualEvaluator(reader.spec(), rowFilter);
+ ResidualEvaluator residuals = new ResidualEvaluator(reader.spec(), rowFilter, caseSensitive);
return Iterables.transform(
reader.filterRows(rowFilter).select(SNAPSHOT_COLUMNS),
file -> new BaseFileScanTask(file, schemaString, specString, residuals)
@@ -221,7 +223,7 @@ class BaseTableScan implements TableScan {
@Override
public Schema schema() {
- return schema;
+ return lazyColumnProjection();
}
@Override
@@ -230,11 +232,17 @@ class BaseTableScan implements TableScan {
}
@Override
+ public boolean isCaseSensitive() {
+ return caseSensitive;
+ }
+
+ @Override
public String toString() {
return Objects.toStringHelper(this)
.add("table", table)
- .add("projection", schema.asStruct())
+ .add("projection", schema().asStruct())
.add("filter", rowFilter)
+ .add("caseSensitive", caseSensitive)
.toString();
}
@@ -246,4 +254,33 @@ class BaseTableScan implements TableScan {
// Capture manifests which can be closed after scan planning
return CloseableIterable.combine(splitTasks, ImmutableList.of(fileScanTasks));
}
+
+ /**
+ * To be able to make refinements {@link #select(Collection)} and {@link #caseSensitive(boolean)} in any order,
+ * we resolve the schema to be projected lazily here.
+ *
+ * @return the Schema to project
+ */
+ private Schema lazyColumnProjection() {
+ if (selectedColumns != null ) {
+ Set<Integer> requiredFieldIds = Sets.newHashSet();
+
+ // all of the filter columns are required
+ requiredFieldIds.addAll(
+ Binder.boundReferences(table.schema().asStruct(), Collections.singletonList(rowFilter), caseSensitive));
+
+ // all of the projection columns are required
+ Set<Integer> selectedIds;
+ if (caseSensitive) {
+ selectedIds = TypeUtil.getProjectedIds(table.schema().select(selectedColumns));
+ } else {
+ selectedIds = TypeUtil.getProjectedIds(table.schema().caseInsensitiveSelect(selectedColumns));
+ }
+ requiredFieldIds.addAll(selectedIds);
+
+ return TypeUtil.select(table.schema(), requiredFieldIds);
+ }
+
+ return schema;
+ }
}
diff --git a/core/src/main/java/com/netflix/iceberg/FilteredManifest.java b/core/src/main/java/com/netflix/iceberg/FilteredManifest.java
index bea656c..10c68cb 100644
--- a/core/src/main/java/com/netflix/iceberg/FilteredManifest.java
+++ b/core/src/main/java/com/netflix/iceberg/FilteredManifest.java
@@ -36,23 +36,25 @@ public class FilteredManifest implements Filterable<FilteredManifest> {
private final Expression partFilter;
private final Expression rowFilter;
private final Collection<String> columns;
+ private final boolean caseSensitive;
// lazy state
private Evaluator lazyEvaluator = null;
private InclusiveMetricsEvaluator lazyMetricsEvaluator = null;
FilteredManifest(ManifestReader reader, Expression partFilter, Expression rowFilter,
- Collection<String> columns) {
+ Collection<String> columns, boolean caseSensitive) {
Preconditions.checkNotNull(reader, "ManifestReader cannot be null");
this.reader = reader;
this.partFilter = partFilter;
this.rowFilter = rowFilter;
this.columns = columns;
+ this.caseSensitive = caseSensitive;
}
@Override
public FilteredManifest select(Collection<String> columns) {
- return new FilteredManifest(reader, partFilter, rowFilter, columns);
+ return new FilteredManifest(reader, partFilter, rowFilter, columns, caseSensitive);
}
@Override
@@ -60,16 +62,18 @@ public class FilteredManifest implements Filterable<FilteredManifest> {
return new FilteredManifest(reader,
Expressions.and(partFilter, expr),
rowFilter,
- columns);
+ columns,
+ caseSensitive);
}
@Override
public FilteredManifest filterRows(Expression expr) {
- Expression projected = Projections.inclusive(reader.spec()).project(expr);
+ Expression projected = Projections.inclusive(reader.spec(), caseSensitive).project(expr);
return new FilteredManifest(reader,
Expressions.and(partFilter, projected),
Expressions.and(rowFilter, expr),
- columns);
+ columns,
+ caseSensitive);
}
Iterable<ManifestEntry> allEntries() {
@@ -128,9 +132,9 @@ public class FilteredManifest implements Filterable<FilteredManifest> {
private Evaluator evaluator() {
if (lazyEvaluator == null) {
if (partFilter != null) {
- this.lazyEvaluator = new Evaluator(reader.spec().partitionType(), partFilter);
+ this.lazyEvaluator = new Evaluator(reader.spec().partitionType(), partFilter, caseSensitive);
} else {
- this.lazyEvaluator = new Evaluator(reader.spec().partitionType(), Expressions.alwaysTrue());
+ this.lazyEvaluator = new Evaluator(reader.spec().partitionType(), Expressions.alwaysTrue(), caseSensitive);
}
}
return lazyEvaluator;
@@ -140,10 +144,10 @@ public class FilteredManifest implements Filterable<FilteredManifest> {
if (lazyMetricsEvaluator == null) {
if (rowFilter != null) {
this.lazyMetricsEvaluator = new InclusiveMetricsEvaluator(
- reader.spec().schema(), rowFilter);
+ reader.spec().schema(), rowFilter, caseSensitive);
} else {
this.lazyMetricsEvaluator = new InclusiveMetricsEvaluator(
- reader.spec().schema(), Expressions.alwaysTrue());
+ reader.spec().schema(), Expressions.alwaysTrue(), caseSensitive);
}
}
return lazyMetricsEvaluator;
diff --git a/core/src/main/java/com/netflix/iceberg/ManifestReader.java b/core/src/main/java/com/netflix/iceberg/ManifestReader.java
index f1128fc..9baaf0c 100644
--- a/core/src/main/java/com/netflix/iceberg/ManifestReader.java
+++ b/core/src/main/java/com/netflix/iceberg/ManifestReader.java
@@ -61,20 +61,33 @@ public class ManifestReader extends CloseableGroup implements Filterable<Filtere
* @return a manifest reader
*/
public static ManifestReader read(InputFile file) {
- return new ManifestReader(file);
+ return new ManifestReader(file, true);
+ }
+
+ /**
+ * Returns a new {@link ManifestReader} for an {@link InputFile}.
+ *
+ * @param file an InputFile
+ * @param caseSensitive whether column name matching should have case sensitivity
+ * @return a manifest reader
+ */
+ public static ManifestReader read(InputFile file, boolean caseSensitive) {
+ return new ManifestReader(file, caseSensitive);
}
private final InputFile file;
private final Map<String, String> metadata;
private final PartitionSpec spec;
private final Schema schema;
+ private final boolean caseSensitive;
// lazily initialized
private List<ManifestEntry> adds = null;
private List<ManifestEntry> deletes = null;
- private ManifestReader(InputFile file) {
+ private ManifestReader(InputFile file, boolean caseSensitive) {
this.file = file;
+ this.caseSensitive = caseSensitive;
try {
try (AvroIterable<ManifestEntry> headerReader = Avro.read(file)
@@ -94,6 +107,27 @@ public class ManifestReader extends CloseableGroup implements Filterable<Filtere
this.spec = PartitionSpecParser.fromJsonFields(schema, specId, metadata.get("partition-spec"));
}
+ private ManifestReader(InputFile file, Map<String, String> metadata,
+ PartitionSpec spec, Schema schema, boolean caseSensitive) {
+ this.file = file;
+ this.metadata = metadata;
+ this.spec = spec;
+ this.schema = schema;
+ this.caseSensitive = caseSensitive;
+ }
+
+ /**
+ * Returns a new {@link ManifestReader} that, if filtered via {@link #select(java.util.Collection)},
+ * {@link #filterPartitions(Expression)} or {@link #filterRows(Expression)}, will apply the specified
+ * case sensitivity for column name matching.
+ *
+ * @param caseSensitive whether column name matching should have case sensitivity
+ * @return a manifest reader with case sensitivity as stated
+ */
+ public ManifestReader caseSensitive(boolean caseSensitive) {
+ return new ManifestReader(file, metadata, spec, schema, caseSensitive);
+ }
+
public InputFile file() {
return file;
}
@@ -113,17 +147,21 @@ public class ManifestReader extends CloseableGroup implements Filterable<Filtere
@Override
public FilteredManifest select(Collection<String> columns) {
- return new FilteredManifest(this, alwaysTrue(), alwaysTrue(), Lists.newArrayList(columns));
+ return new FilteredManifest(this, alwaysTrue(), alwaysTrue(), Lists.newArrayList(columns), caseSensitive);
}
@Override
public FilteredManifest filterPartitions(Expression expr) {
- return new FilteredManifest(this, expr, alwaysTrue(), ALL_COLUMNS);
+ return new FilteredManifest(this, expr, alwaysTrue(), ALL_COLUMNS, caseSensitive);
}
@Override
public FilteredManifest filterRows(Expression expr) {
- return new FilteredManifest(this, Projections.inclusive(spec).project(expr), expr, ALL_COLUMNS);
+ return new FilteredManifest(this,
+ Projections.inclusive(spec, caseSensitive).project(expr),
+ expr,
+ ALL_COLUMNS,
+ caseSensitive);
}
public List<ManifestEntry> addedFiles() {
diff --git a/core/src/test/java/com/netflix/iceberg/TestFilterFiles.java b/core/src/test/java/com/netflix/iceberg/TestBaseTableScan.java
similarity index 54%
copy from core/src/test/java/com/netflix/iceberg/TestFilterFiles.java
copy to core/src/test/java/com/netflix/iceberg/TestBaseTableScan.java
index cafff00..ae0a841 100644
--- a/core/src/test/java/com/netflix/iceberg/TestFilterFiles.java
+++ b/core/src/test/java/com/netflix/iceberg/TestBaseTableScan.java
@@ -19,16 +19,6 @@
package com.netflix.iceberg;
-import java.io.File;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.Map;
-
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Maps;
-import com.netflix.iceberg.expressions.Expressions;
-import com.netflix.iceberg.types.Conversions;
import com.netflix.iceberg.types.Types;
import org.junit.After;
import org.junit.Before;
@@ -36,10 +26,13 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import java.io.File;
+import java.io.IOException;
+
import static com.netflix.iceberg.types.Types.NestedField.required;
import static org.junit.Assert.assertEquals;
-public class TestFilterFiles {
+public class TestBaseTableScan {
@Rule
public TemporaryFolder temp = new TemporaryFolder();
@@ -60,42 +53,41 @@ public class TestFilterFiles {
}
@Test
- public void testFilterFilesUnpartitionedTable() {
+ public void testTableScanHonorsSelect() {
PartitionSpec spec = PartitionSpec.unpartitioned();
Table table = TestTables.create(tableDir, "test", schema, spec);
- testFilterFiles(table);
- }
- @Test
- public void testFilterFilesPartitionedTable() {
- PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("data", 16).build();
- Table table = TestTables.create(tableDir, "test", schema, spec);
- testFilterFiles(table);
- }
+ TableScan scan = table.newScan().select("id");
- private void testFilterFiles(Table table) {
- Map<Integer, ByteBuffer> lowerBounds = new HashMap<>();
- Map<Integer, ByteBuffer> upperBounds = new HashMap<>();
- lowerBounds.put(1, Conversions.toByteBuffer(Types.IntegerType.get(), 1));
- upperBounds.put(1, Conversions.toByteBuffer(Types.IntegerType.get(), 2));
+ Schema expectedSchema = new Schema(
+ required(1, "id", Types.IntegerType.get())
+ );
- Metrics metrics = new Metrics(2L, Maps.newHashMap(), Maps.newHashMap(),
- Maps.newHashMap(), lowerBounds, upperBounds);
+ assertEquals("A tableScan.select() should prune the schema",
+ expectedSchema.asStruct(),
+ scan.schema().asStruct());
+ }
- DataFile file = DataFiles.builder(table.spec())
- .withPath("/path/to/file.parquet")
- .withFileSizeInBytes(0)
- .withMetrics(metrics)
- .build();
+ @Test
+ public void testTableScanHonorsSelectWithoutCaseSensitivity() {
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = TestTables.create(tableDir, "test", schema, spec);
- table.newAppend().appendFile(file).commit();
+ TableScan scan1 = table.newScan().caseSensitive(false).select("ID");
+ // order of refinements shouldn't matter
+ TableScan scan2 = table.newScan().select("ID").caseSensitive(false);
- table.refresh();
+ Schema expectedSchema = new Schema(
+ required(1, "id", Types.IntegerType.get())
+ );
- TableScan emptyScan = table.newScan().filter(Expressions.equal("id", 5));
- assertEquals(0, Iterables.size(emptyScan.planFiles()));
+ assertEquals("A tableScan.select() should prune the schema without case sensitivity",
+ expectedSchema.asStruct(),
+ scan1.schema().asStruct());
- TableScan nonEmptyScan = table.newScan().filter(Expressions.equal("id", 1));
- assertEquals(1, Iterables.size(nonEmptyScan.planFiles()));
+ assertEquals("A tableScan.select() should prune the schema regardless of scan refinement order",
+ expectedSchema.asStruct(),
+ scan2.schema().asStruct());
}
+
}
diff --git a/core/src/test/java/com/netflix/iceberg/TestFilterFiles.java b/core/src/test/java/com/netflix/iceberg/TestFilterFiles.java
index cafff00..22b0ae5 100644
--- a/core/src/test/java/com/netflix/iceberg/TestFilterFiles.java
+++ b/core/src/test/java/com/netflix/iceberg/TestFilterFiles.java
@@ -67,12 +67,26 @@ public class TestFilterFiles {
}
@Test
+ public void testCaseInsensitiveFilterFilesUnpartitionedTable() {
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = TestTables.create(tableDir, "test", schema, spec);
+ testCaseInsensitiveFilterFiles(table);
+ }
+
+ @Test
public void testFilterFilesPartitionedTable() {
PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("data", 16).build();
Table table = TestTables.create(tableDir, "test", schema, spec);
testFilterFiles(table);
}
+ @Test
+ public void testCaseInsensitiveFilterFilesPartitionedTable() {
+ PartitionSpec spec = PartitionSpec.builderFor(schema).bucket("data", 16).build();
+ Table table = TestTables.create(tableDir, "test", schema, spec);
+ testCaseInsensitiveFilterFiles(table);
+ }
+
private void testFilterFiles(Table table) {
Map<Integer, ByteBuffer> lowerBounds = new HashMap<>();
Map<Integer, ByteBuffer> upperBounds = new HashMap<>();
@@ -98,4 +112,30 @@ public class TestFilterFiles {
TableScan nonEmptyScan = table.newScan().filter(Expressions.equal("id", 1));
assertEquals(1, Iterables.size(nonEmptyScan.planFiles()));
}
+
+ private void testCaseInsensitiveFilterFiles(Table table) {
+ Map<Integer, ByteBuffer> lowerBounds = new HashMap<>();
+ Map<Integer, ByteBuffer> upperBounds = new HashMap<>();
+ lowerBounds.put(1, Conversions.toByteBuffer(Types.IntegerType.get(), 1));
+ upperBounds.put(1, Conversions.toByteBuffer(Types.IntegerType.get(), 2));
+
+ Metrics metrics = new Metrics(2L, Maps.newHashMap(), Maps.newHashMap(),
+ Maps.newHashMap(), lowerBounds, upperBounds);
+
+ DataFile file = DataFiles.builder(table.spec())
+ .withPath("/path/to/file.parquet")
+ .withFileSizeInBytes(0)
+ .withMetrics(metrics)
+ .build();
+
+ table.newAppend().appendFile(file).commit();
+
+ table.refresh();
+
+ TableScan emptyScan = table.newScan().caseSensitive(false).filter(Expressions.equal("ID", 5));
+ assertEquals(0, Iterables.size(emptyScan.planFiles()));
+
+ TableScan nonEmptyScan = table.newScan().caseSensitive(false).filter(Expressions.equal("ID", 1));
+ assertEquals(1, Iterables.size(nonEmptyScan.planFiles()));
+ }
}
diff --git a/data/src/main/java/com/netflix/iceberg/data/IcebergGenerics.java b/data/src/main/java/com/netflix/iceberg/data/IcebergGenerics.java
index 587ed00..941da26 100644
--- a/data/src/main/java/com/netflix/iceberg/data/IcebergGenerics.java
+++ b/data/src/main/java/com/netflix/iceberg/data/IcebergGenerics.java
@@ -44,6 +44,7 @@ public class IcebergGenerics {
private Expression where = Expressions.alwaysTrue();
private List<String> columns = ImmutableList.of("*");
private boolean reuseContainers = false;
+ private boolean caseSensitive = true;
public ScanBuilder(Table table) {
this.table = table;
@@ -59,13 +60,25 @@ public class IcebergGenerics {
return this;
}
+ public ScanBuilder caseInsensitive() {
+ this.caseSensitive = false;
+ return this;
+ }
+
public ScanBuilder select(String... columns) {
this.columns = ImmutableList.copyOf(columns);
return this;
}
public Iterable<Record> build() {
- return new TableScanIterable(table.newScan().filter(where).select(columns), reuseContainers);
+ return new TableScanIterable(
+ table
+ .newScan()
+ .filter(where)
+ .caseSensitive(caseSensitive)
+ .select(columns),
+ reuseContainers
+ );
}
}
}
diff --git a/data/src/main/java/com/netflix/iceberg/data/TableScanIterable.java b/data/src/main/java/com/netflix/iceberg/data/TableScanIterable.java
index 4f59556..b48b057 100644
--- a/data/src/main/java/com/netflix/iceberg/data/TableScanIterable.java
+++ b/data/src/main/java/com/netflix/iceberg/data/TableScanIterable.java
@@ -58,6 +58,7 @@ class TableScanIterable extends CloseableGroup implements CloseableIterable<Reco
private final TableOperations ops;
private final Schema projection;
private final boolean reuseContainers;
+ private final boolean caseSensitive;
private final CloseableIterable<CombinedScanTask> tasks;
TableScanIterable(TableScan scan, boolean reuseContainers) {
@@ -66,6 +67,7 @@ class TableScanIterable extends CloseableGroup implements CloseableIterable<Reco
this.ops = ((HasTableOperations) scan.table()).operations();
this.projection = scan.schema();
this.reuseContainers = reuseContainers;
+ this.caseSensitive = scan.isCaseSensitive();
// start planning tasks in the background
this.tasks = scan.planTasks();
@@ -73,7 +75,7 @@ class TableScanIterable extends CloseableGroup implements CloseableIterable<Reco
@Override
public Iterator<Record> iterator() {
- ScanIterator iter = new ScanIterator(tasks);
+ ScanIterator iter = new ScanIterator(tasks, caseSensitive);
addCloseable(iter);
return iter;
}
@@ -121,11 +123,13 @@ class TableScanIterable extends CloseableGroup implements CloseableIterable<Reco
private class ScanIterator implements Iterator<Record>, Closeable {
private final Iterator<FileScanTask> tasks;
+ private final boolean caseSensitive;
private Closeable currentCloseable = null;
private Iterator<Record> currentIterator = emptyIterator();
- private ScanIterator(Iterable<CombinedScanTask> tasks) {
+ private ScanIterator(Iterable<CombinedScanTask> tasks, boolean caseSensitive) {
this.tasks = Lists.newArrayList(concat(transform(tasks, CombinedScanTask::files))).iterator();
+ this.caseSensitive = caseSensitive;
}
@Override
@@ -148,7 +152,7 @@ class TableScanIterable extends CloseableGroup implements CloseableIterable<Reco
this.currentCloseable = reader;
if (task.residual() != null && task.residual() != Expressions.alwaysTrue()) {
- Evaluator filter = new Evaluator(projection.asStruct(), task.residual());
+ Evaluator filter = new Evaluator(projection.asStruct(), task.residual(), caseSensitive);
this.currentIterator = filter(reader, filter::eval).iterator();
} else {
this.currentIterator = reader.iterator();
diff --git a/spark/src/main/java/com/netflix/iceberg/spark/SparkSchemaUtil.java b/spark/src/main/java/com/netflix/iceberg/spark/SparkSchemaUtil.java
index bfe9390..6b193af 100644
--- a/spark/src/main/java/com/netflix/iceberg/spark/SparkSchemaUtil.java
+++ b/spark/src/main/java/com/netflix/iceberg/spark/SparkSchemaUtil.java
@@ -224,8 +224,10 @@ public class SparkSchemaUtil {
* @return a Schema corresponding to the Spark projection
* @throws IllegalArgumentException if the Spark type does not match the Schema
*/
- public static Schema prune(Schema schema, StructType requestedType, Expression filter) {
- Set<Integer> filterRefs = Binder.boundReferences(schema.asStruct(), Collections.singletonList(filter), true);
+ public static Schema prune(Schema schema, StructType requestedType, Expression filter, boolean caseSensitive) {
+ Set<Integer> filterRefs =
+ Binder.boundReferences(schema.asStruct(), Collections.singletonList(filter), caseSensitive);
+
return new Schema(visit(schema, new PruneColumnsWithoutReordering(requestedType, filterRefs))
.asNestedType()
.asStructType()
diff --git a/spark/src/main/java/com/netflix/iceberg/spark/source/IcebergSource.java b/spark/src/main/java/com/netflix/iceberg/spark/source/IcebergSource.java
index 1991d29..bacd173 100644
--- a/spark/src/main/java/com/netflix/iceberg/spark/source/IcebergSource.java
+++ b/spark/src/main/java/com/netflix/iceberg/spark/source/IcebergSource.java
@@ -20,6 +20,7 @@
package com.netflix.iceberg.spark.source;
import com.google.common.base.Preconditions;
+import com.netflix.iceberg.ConfigProperties;
import com.netflix.iceberg.FileFormat;
import com.netflix.iceberg.Schema;
import com.netflix.iceberg.Table;
@@ -59,7 +60,9 @@ public class IcebergSource implements DataSourceV2, ReadSupport, WriteSupport, D
public DataSourceReader createReader(DataSourceOptions options) {
Configuration conf = new Configuration(lazyBaseConf());
Table table = getTableAndResolveHadoopConfiguration(options, conf);
- return new Reader(table);
+ String caseSensitive = lazySparkSession().conf().get("spark.sql.caseSensitive", "true");
+
+ return new Reader(table, Boolean.valueOf(caseSensitive));
}
@Override
diff --git a/spark/src/main/java/com/netflix/iceberg/spark/source/Reader.java b/spark/src/main/java/com/netflix/iceberg/spark/source/Reader.java
index 85628de..3f7cb43 100644
--- a/spark/src/main/java/com/netflix/iceberg/spark/source/Reader.java
+++ b/spark/src/main/java/com/netflix/iceberg/spark/source/Reader.java
@@ -98,6 +98,7 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
private final Table table;
private final FileIO fileIo;
private final EncryptionManager encryptionManager;
+ private final boolean caseSensitive;
private StructType requestedSchema = null;
private List<Expression> filterExpressions = null;
private Filter[] pushedFilters = NO_FILTERS;
@@ -107,11 +108,12 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
private StructType type = null; // cached because Spark accesses it multiple times
private List<CombinedScanTask> tasks = null; // lazy cache of tasks
- Reader(Table table) {
+ Reader(Table table, boolean caseSensitive) {
this.table = table;
this.schema = table.schema();
this.fileIo = table.io();
this.encryptionManager = table.encryption();
+ this.caseSensitive = caseSensitive;
}
private Schema lazySchema() {
@@ -144,7 +146,8 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
List<InputPartition<InternalRow>> readTasks = Lists.newArrayList();
for (CombinedScanTask task : tasks()) {
- readTasks.add(new ReadTask(task, tableSchemaString, expectedSchemaString, fileIo, encryptionManager));
+ readTasks.add(
+ new ReadTask(task, tableSchemaString, expectedSchemaString, fileIo, encryptionManager, caseSensitive));
}
return readTasks;
@@ -208,7 +211,10 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
private List<CombinedScanTask> tasks() {
if (tasks == null) {
- TableScan scan = table.newScan().project(lazySchema());
+ TableScan scan = table
+ .newScan()
+ .caseSensitive(caseSensitive)
+ .project(lazySchema());
if (filterExpressions != null) {
for (Expression filter : filterExpressions) {
@@ -229,8 +235,8 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
@Override
public String toString() {
return String.format(
- "IcebergScan(table=%s, type=%s, filters=%s)",
- table, lazySchema().asStruct(), filterExpressions);
+ "IcebergScan(table=%s, type=%s, filters=%s, caseSensitive=%s)",
+ table, lazySchema().asStruct(), filterExpressions, caseSensitive);
}
private static class ReadTask implements InputPartition<InternalRow>, Serializable {
@@ -239,23 +245,26 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
private final String expectedSchemaString;
private final FileIO fileIo;
private final EncryptionManager encryptionManager;
+ private final boolean caseSensitive;
private transient Schema tableSchema = null;
private transient Schema expectedSchema = null;
private ReadTask(
CombinedScanTask task, String tableSchemaString, String expectedSchemaString, FileIO fileIo,
- EncryptionManager encryptionManager) {
+ EncryptionManager encryptionManager, boolean caseSensitive) {
this.task = task;
this.tableSchemaString = tableSchemaString;
this.expectedSchemaString = expectedSchemaString;
this.fileIo = fileIo;
this.encryptionManager = encryptionManager;
+ this.caseSensitive = caseSensitive;
}
@Override
public InputPartitionReader<InternalRow> createPartitionReader() {
- return new TaskDataReader(task, lazyTableSchema(), lazyExpectedSchema(), fileIo, encryptionManager);
+ return new TaskDataReader(task, lazyTableSchema(), lazyExpectedSchema(), fileIo,
+ encryptionManager, caseSensitive);
}
private Schema lazyTableSchema() {
@@ -284,13 +293,14 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
private final Schema expectedSchema;
private final FileIO fileIo;
private final Map<String, InputFile> inputFiles;
+ private final boolean caseSensitive;
private Iterator<InternalRow> currentIterator = null;
private Closeable currentCloseable = null;
private InternalRow current = null;
public TaskDataReader(CombinedScanTask task, Schema tableSchema, Schema expectedSchema, FileIO fileIo,
- EncryptionManager encryptionManager) {
+ EncryptionManager encryptionManager, boolean caseSensitive) {
this.fileIo = fileIo;
this.tasks = task.files().iterator();
this.tableSchema = tableSchema;
@@ -305,6 +315,7 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
this.inputFiles = inputFileBuilder.build();
// open last because the schemas and fileIo must be set
this.currentIterator = open(tasks.next());
+ this.caseSensitive = caseSensitive;
}
@Override
@@ -349,7 +360,7 @@ class Reader implements DataSourceReader, SupportsPushDownFilters, SupportsPushD
Set<Integer> idColumns = spec.identitySourceIds();
// schema needed for the projection and filtering
- Schema requiredSchema = prune(tableSchema, convert(finalSchema), task.residual());
+ Schema requiredSchema = prune(tableSchema, convert(finalSchema), task.residual(), caseSensitive);
boolean hasJoinedPartitionColumns = !idColumns.isEmpty();
boolean hasExtraFilterColumns = requiredSchema.columns().size() != finalSchema.columns().size();
diff --git a/spark/src/test/java/com/netflix/iceberg/spark/source/TestFilteredScan.java b/spark/src/test/java/com/netflix/iceberg/spark/source/TestFilteredScan.java
index 2340cd6..6c2eafc 100644
--- a/spark/src/test/java/com/netflix/iceberg/spark/source/TestFilteredScan.java
+++ b/spark/src/test/java/com/netflix/iceberg/spark/source/TestFilteredScan.java
@@ -219,6 +219,37 @@ public class TestFilteredScan {
}
@Test
+ public void testUnpartitionedCaseInsensitiveIDFilters() {
+ DataSourceOptions options = new DataSourceOptions(ImmutableMap.of(
+ "path", unpartitioned.toString())
+ );
+
+ // set spark.sql.caseSensitive to false
+ String caseSensitivityBeforeTest = TestFilteredScan.spark.conf().get("spark.sql.caseSensitive");
+ TestFilteredScan.spark.conf().set("spark.sql.caseSensitive", "false");
+
+ try {
+ IcebergSource source = new IcebergSource();
+
+ for (int i = 0; i < 10; i += 1) {
+ DataSourceReader reader = source.createReader(options);
+
+ pushFilters(reader, EqualTo.apply("ID", i)); // note lower(ID) == lower(id), so there must be a match
+
+ List<InputPartition<InternalRow>> tasks = reader.planInputPartitions();
+ Assert.assertEquals("Should only create one task for a small file", 1, tasks.size());
+
+ // validate row filtering
+ assertEqualsSafe(SCHEMA.asStruct(), expected(i),
+ read(unpartitioned.toString(), "id = " + i));
+ }
+ } finally {
+ // return global conf to previous state
+ TestFilteredScan.spark.conf().set("spark.sql.caseSensitive", caseSensitivityBeforeTest);
+ }
+ }
+
+ @Test
public void testUnpartitionedTimestampFilter() {
DataSourceOptions options = new DataSourceOptions(ImmutableMap.of(
"path", unpartitioned.toString())