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/12/23 17:08:26 UTC
[incubator-iceberg] branch vectorized-read updated: Merging master
into vectorized-read (#709)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch vectorized-read
in repository https://gitbox.apache.org/repos/asf/incubator-iceberg.git
The following commit(s) were added to refs/heads/vectorized-read by this push:
new 6f35cd7 Merging master into vectorized-read (#709)
6f35cd7 is described below
commit 6f35cd74c954ac118e9156500947f2503a39b445
Author: Samarth Jain <sa...@apache.org>
AuthorDate: Mon Dec 23 09:08:17 2019 -0800
Merging master into vectorized-read (#709)
---
.../apache/iceberg/catalog/TableIdentifier.java | 8 ++
.../expressions/{Reference.java => Bound.java} | 22 +++-
.../iceberg/expressions/BoundLiteralPredicate.java | 22 ++--
.../apache/iceberg/expressions/BoundPredicate.java | 18 ++-
.../apache/iceberg/expressions/BoundReference.java | 36 ++++--
.../iceberg/expressions/BoundSetPredicate.java | 11 +-
.../expressions/{Reference.java => BoundTerm.java} | 23 +++-
.../{BoundReference.java => BoundTransform.java} | 43 ++++---
.../iceberg/expressions/BoundUnaryPredicate.java | 8 +-
.../org/apache/iceberg/expressions/Evaluator.java | 48 ++++----
.../iceberg/expressions/ExpressionVisitors.java | 130 ++++++++++++++++++--
.../apache/iceberg/expressions/Expressions.java | 103 +++++++++++++++-
.../expressions/InclusiveMetricsEvaluator.java | 7 +-
.../apache/iceberg/expressions/NamedReference.java | 22 +++-
.../org/apache/iceberg/expressions/Predicate.java | 15 ++-
.../org/apache/iceberg/expressions/Reference.java | 4 +-
.../iceberg/expressions/ResidualEvaluator.java | 38 +++---
.../expressions/{Reference.java => Term.java} | 6 +-
.../expressions/{Reference.java => Unbound.java} | 23 +++-
.../iceberg/expressions/UnboundPredicate.java | 105 ++++++++--------
.../{Reference.java => UnboundTerm.java} | 10 +-
.../iceberg/expressions/UnboundTransform.java | 67 +++++++++++
.../java/org/apache/iceberg/transforms/Bucket.java | 15 ++-
.../java/org/apache/iceberg/transforms/Dates.java | 9 ++
.../apache/iceberg/transforms/ProjectionUtil.java | 27 +++++
.../org/apache/iceberg/transforms/Timestamps.java | 9 ++
.../org/apache/iceberg/transforms/Truncate.java | 41 +++++++
.../main/java/org/apache/iceberg/types/Types.java | 32 +++--
.../iceberg/catalog/TestTableIdentifier.java | 13 ++
.../iceberg/expressions/TestExpressionBinding.java | 25 ++--
.../iceberg/expressions/TestExpressionHelpers.java | 44 ++++++-
.../expressions/TestExpressionSerialization.java | 27 ++++-
.../org/apache/iceberg/types/TestTypeUtil.java | 14 +++
.../arrow/vectorized/VectorizedArrowReader.java | 4 +-
baseline.gradle | 40 ++++---
build.gradle | 9 ++
.../iceberg/BaseMetastoreTableOperations.java | 1 -
.../java/org/apache/iceberg/CachingCatalog.java | 28 +++--
.../main/java/org/apache/iceberg/ScanSummary.java | 16 +--
.../java/org/apache/iceberg/TableProperties.java | 3 +
.../org/apache/iceberg/hadoop/HadoopCatalog.java | 6 +-
.../org/apache/iceberg/hadoop/HadoopFileIO.java | 13 +-
.../apache/iceberg/util/SerializableSupplier.java | 11 +-
.../apache/iceberg/hadoop/TestHadoopCatalog.java | 7 +-
.../java/org/apache/iceberg/hive/ClientPool.java | 2 -
.../apache/iceberg/hive/HiveTableOperations.java | 2 +-
jmh.gradle | 4 +-
.../java/org/apache/iceberg/parquet/Parquet.java | 20 ++++
.../org/apache/iceberg/parquet/ParquetFilters.java | 6 +-
.../org/apache/iceberg/parquet/ParquetUtil.java | 7 +-
site/docs/configuration.md | 8 +-
.../IcebergSourceFlatAvroDataReadBenchmark.java | 132 +++++++++++++++++++++
.../IcebergSourceNestedAvroDataReadBenchmark.java | 132 +++++++++++++++++++++
.../iceberg/spark/data/SparkParquetWriters.java | 2 -
.../data/vectorized/IcebergArrowColumnVector.java | 5 +-
.../apache/iceberg/spark/source/IcebergSource.java | 42 ++++++-
.../org/apache/iceberg/spark/source/Reader.java | 32 ++---
.../iceberg/spark/source/StreamingWriter.java | 10 +-
.../org/apache/iceberg/spark/source/Writer.java | 49 ++++----
versions.lock | 5 +-
versions.props | 1 +
61 files changed, 1266 insertions(+), 356 deletions(-)
diff --git a/api/src/main/java/org/apache/iceberg/catalog/TableIdentifier.java b/api/src/main/java/org/apache/iceberg/catalog/TableIdentifier.java
index 0e4d898..87a4048 100644
--- a/api/src/main/java/org/apache/iceberg/catalog/TableIdentifier.java
+++ b/api/src/main/java/org/apache/iceberg/catalog/TableIdentifier.java
@@ -77,6 +77,14 @@ public class TableIdentifier {
return name;
}
+ public TableIdentifier toLowerCase() {
+ String[] newLevels = Arrays.stream(namespace().levels())
+ .map(String::toLowerCase)
+ .toArray(String[]::new);
+ String newName = name().toLowerCase();
+ return TableIdentifier.of(Namespace.of(newLevels), newName);
+ }
+
@Override
public boolean equals(Object other) {
if (this == other) {
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Reference.java b/api/src/main/java/org/apache/iceberg/expressions/Bound.java
similarity index 64%
copy from api/src/main/java/org/apache/iceberg/expressions/Reference.java
copy to api/src/main/java/org/apache/iceberg/expressions/Bound.java
index f08867f..df7abb9 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Reference.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Bound.java
@@ -19,12 +19,24 @@
package org.apache.iceberg.expressions;
-import java.io.Serializable;
+import org.apache.iceberg.StructLike;
/**
- * Represents a variable reference in an {@link Expression expression}.
- * @see BoundReference
- * @see NamedReference
+ * Represents a bound value expression.
+ *
+ * @param <T> the Java type of values produced by this expression
*/
-public interface Reference extends Serializable {
+public interface Bound<T> {
+ /**
+ * @return the underlying reference
+ */
+ BoundReference<?> ref();
+
+ /**
+ * Produce a value from the struct for this expression.
+ *
+ * @param struct a struct of incoming data
+ * @return the value of this expression when evaluated on the incoming struct
+ */
+ T eval(StructLike struct);
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/BoundLiteralPredicate.java b/api/src/main/java/org/apache/iceberg/expressions/BoundLiteralPredicate.java
index 0bf2d6f..65ac03b 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/BoundLiteralPredicate.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/BoundLiteralPredicate.java
@@ -25,8 +25,8 @@ import java.util.Comparator;
public class BoundLiteralPredicate<T> extends BoundPredicate<T> {
private final Literal<T> literal;
- BoundLiteralPredicate(Operation op, BoundReference<T> ref, Literal<T> lit) {
- super(op, ref);
+ BoundLiteralPredicate(Operation op, BoundTerm<T> term, Literal<T> lit) {
+ super(op, term);
Preconditions.checkArgument(op != Operation.IN && op != Operation.NOT_IN,
"Bound literal predicate does not support operation: %s", op);
this.literal = lit;
@@ -73,23 +73,23 @@ public class BoundLiteralPredicate<T> extends BoundPredicate<T> {
public String toString() {
switch (op()) {
case LT:
- return ref() + " < " + literal;
+ return term() + " < " + literal;
case LT_EQ:
- return ref() + " <= " + literal;
+ return term() + " <= " + literal;
case GT:
- return ref() + " > " + literal;
+ return term() + " > " + literal;
case GT_EQ:
- return ref() + " >= " + literal;
+ return term() + " >= " + literal;
case EQ:
- return ref() + " == " + literal;
+ return term() + " == " + literal;
case NOT_EQ:
- return ref() + " != " + literal;
+ return term() + " != " + literal;
case STARTS_WITH:
- return ref() + " startsWith \"" + literal + "\"";
+ return term() + " startsWith \"" + literal + "\"";
case IN:
- return ref() + " in { " + literal + " }";
+ return term() + " in { " + literal + " }";
case NOT_IN:
- return ref() + " not in { " + literal + " }";
+ return term() + " not in { " + literal + " }";
default:
return "Invalid literal predicate: operation = " + op();
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/BoundPredicate.java b/api/src/main/java/org/apache/iceberg/expressions/BoundPredicate.java
index a171bdd..fde4b1e 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/BoundPredicate.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/BoundPredicate.java
@@ -21,17 +21,27 @@ package org.apache.iceberg.expressions;
import org.apache.iceberg.StructLike;
-public abstract class BoundPredicate<T> extends Predicate<BoundReference<T>> {
- protected BoundPredicate(Operation op, BoundReference<T> ref) {
- super(op, ref);
+public abstract class BoundPredicate<T> extends Predicate<T, BoundTerm<T>> implements Bound<Boolean> {
+ protected BoundPredicate(Operation op, BoundTerm<T> term) {
+ super(op, term);
}
public boolean test(StructLike struct) {
- return test(ref().get(struct));
+ return test(term().eval(struct));
}
public abstract boolean test(T value);
+ @Override
+ public Boolean eval(StructLike struct) {
+ return test(term().eval(struct));
+ }
+
+ @Override
+ public BoundReference<?> ref() {
+ return term().ref();
+ }
+
public boolean isUnaryPredicate() {
return false;
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/BoundReference.java b/api/src/main/java/org/apache/iceberg/expressions/BoundReference.java
index f917a8f..3450d12 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/BoundReference.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/BoundReference.java
@@ -22,36 +22,46 @@ package org.apache.iceberg.expressions;
import org.apache.iceberg.Accessor;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
-public class BoundReference<T> implements Reference {
- private final int fieldId;
+public class BoundReference<T> implements BoundTerm<T>, Reference<T> {
+ private final Types.NestedField field;
private final Accessor<StructLike> accessor;
- BoundReference(int fieldId, Accessor<StructLike> accessor) {
- this.fieldId = fieldId;
+ BoundReference(Types.NestedField field, Accessor<StructLike> accessor) {
+ this.field = field;
this.accessor = accessor;
}
+ @Override
+ @SuppressWarnings("unchecked")
+ public T eval(StructLike struct) {
+ return (T) accessor.get(struct);
+ }
+
+ public Types.NestedField field() {
+ return field;
+ }
+
+ @Override
+ public BoundReference<T> ref() {
+ return this;
+ }
+
public Type type() {
- return accessor.type();
+ return field.type();
}
public int fieldId() {
- return fieldId;
+ return field.fieldId();
}
public Accessor<StructLike> accessor() {
return accessor;
}
- @SuppressWarnings("unchecked")
- public T get(StructLike struct) {
- return (T) accessor.get(struct);
- }
-
@Override
public String toString() {
- return String.format("ref(id=%d, accessor-type=%s)", fieldId, accessor.type());
+ return String.format("ref(id=%d, accessor-type=%s)", field.fieldId(), accessor.type());
}
-
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/BoundSetPredicate.java b/api/src/main/java/org/apache/iceberg/expressions/BoundSetPredicate.java
index 5e6eb15..337ccea 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/BoundSetPredicate.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/BoundSetPredicate.java
@@ -27,8 +27,8 @@ public class BoundSetPredicate<T> extends BoundPredicate<T> {
private static final Joiner COMMA = Joiner.on(", ");
private final Set<T> literalSet;
- BoundSetPredicate(Operation op, BoundReference<T> ref, Set<T> lits) {
- super(op, ref);
+ BoundSetPredicate(Operation op, BoundTerm<T> term, Set<T> lits) {
+ super(op, term);
Preconditions.checkArgument(op == Operation.IN || op == Operation.NOT_IN,
"%s predicate does not support a literal set", op);
this.literalSet = lits;
@@ -36,7 +36,7 @@ public class BoundSetPredicate<T> extends BoundPredicate<T> {
@Override
public Expression negate() {
- return new BoundSetPredicate<>(op().negate(), ref(), literalSet);
+ return new BoundSetPredicate<>(op().negate(), term(), literalSet);
}
@Override
@@ -53,6 +53,7 @@ public class BoundSetPredicate<T> extends BoundPredicate<T> {
return literalSet;
}
+ @Override
public boolean test(T value) {
switch (op()) {
case IN:
@@ -68,9 +69,9 @@ public class BoundSetPredicate<T> extends BoundPredicate<T> {
public String toString() {
switch (op()) {
case IN:
- return ref() + " in (" + COMMA.join(literalSet) + ")";
+ return term() + " in (" + COMMA.join(literalSet) + ")";
case NOT_IN:
- return ref() + " not in (" + COMMA.join(literalSet) + ")";
+ return term() + " not in (" + COMMA.join(literalSet) + ")";
default:
return "Invalid unary predicate: operation = " + op();
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Reference.java b/api/src/main/java/org/apache/iceberg/expressions/BoundTerm.java
similarity index 62%
copy from api/src/main/java/org/apache/iceberg/expressions/Reference.java
copy to api/src/main/java/org/apache/iceberg/expressions/BoundTerm.java
index f08867f..a32949d 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Reference.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/BoundTerm.java
@@ -19,12 +19,25 @@
package org.apache.iceberg.expressions;
-import java.io.Serializable;
+import java.util.Comparator;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Type;
/**
- * Represents a variable reference in an {@link Expression expression}.
- * @see BoundReference
- * @see NamedReference
+ * Represents a bound term.
+ *
+ * @param <T> the Java type of values produced by this term
*/
-public interface Reference extends Serializable {
+public interface BoundTerm<T> extends Bound<T>, Term {
+ /**
+ * @return the type produced by this expression
+ */
+ Type type();
+
+ /**
+ * @return a {@link Comparator} for values produced by this term
+ */
+ default Comparator<T> comparator() {
+ return Comparators.forType(type().asPrimitiveType());
+ }
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/BoundReference.java b/api/src/main/java/org/apache/iceberg/expressions/BoundTransform.java
similarity index 56%
copy from api/src/main/java/org/apache/iceberg/expressions/BoundReference.java
copy to api/src/main/java/org/apache/iceberg/expressions/BoundTransform.java
index f917a8f..8e2727c 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/BoundReference.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/BoundTransform.java
@@ -19,39 +19,46 @@
package org.apache.iceberg.expressions;
-import org.apache.iceberg.Accessor;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.transforms.Transform;
import org.apache.iceberg.types.Type;
-public class BoundReference<T> implements Reference {
- private final int fieldId;
- private final Accessor<StructLike> accessor;
+/**
+ * A transform expression.
+ *
+ * @param <S> the Java type of values transformed by this function.
+ * @param <T> the Java type of values returned by the function.
+ */
+public class BoundTransform<S, T> implements BoundTerm<T> {
+ private final BoundReference<S> ref;
+ private final Transform<S, T> transform;
- BoundReference(int fieldId, Accessor<StructLike> accessor) {
- this.fieldId = fieldId;
- this.accessor = accessor;
+ BoundTransform(BoundReference<S> ref, Transform<S, T> transform) {
+ this.ref = ref;
+ this.transform = transform;
}
- public Type type() {
- return accessor.type();
+ @Override
+ public T eval(StructLike struct) {
+ return transform.apply(ref.eval(struct));
}
- public int fieldId() {
- return fieldId;
+ @Override
+ public BoundReference<S> ref() {
+ return ref;
}
- public Accessor<StructLike> accessor() {
- return accessor;
+ public Transform<S, T> transform() {
+ return transform;
}
- @SuppressWarnings("unchecked")
- public T get(StructLike struct) {
- return (T) accessor.get(struct);
+ @Override
+ public Type type() {
+ return transform.getResultType(ref.type());
}
@Override
public String toString() {
- return String.format("ref(id=%d, accessor-type=%s)", fieldId, accessor.type());
+ return transform + "(" + ref + ")";
}
-
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/BoundUnaryPredicate.java b/api/src/main/java/org/apache/iceberg/expressions/BoundUnaryPredicate.java
index 1e5b09b..c5835ba 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/BoundUnaryPredicate.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/BoundUnaryPredicate.java
@@ -20,8 +20,8 @@
package org.apache.iceberg.expressions;
public class BoundUnaryPredicate<T> extends BoundPredicate<T> {
- BoundUnaryPredicate(Operation op, BoundReference<T> ref) {
- super(op, ref);
+ BoundUnaryPredicate(Operation op, BoundTerm<T> term) {
+ super(op, term);
}
@Override
@@ -50,9 +50,9 @@ public class BoundUnaryPredicate<T> extends BoundPredicate<T> {
public String toString() {
switch (op()) {
case IS_NULL:
- return "is_null(" + ref() + ")";
+ return "is_null(" + term() + ")";
case NOT_NULL:
- return "not_null(" + ref() + ")";
+ return "not_null(" + term() + ")";
default:
return "Invalid unary predicate: operation = " + op();
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java b/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
index a12b648..0d3e1ef 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
@@ -23,7 +23,7 @@ import java.io.Serializable;
import java.util.Comparator;
import java.util.Set;
import org.apache.iceberg.StructLike;
-import org.apache.iceberg.expressions.ExpressionVisitors.BoundExpressionVisitor;
+import org.apache.iceberg.expressions.ExpressionVisitors.BoundVisitor;
import org.apache.iceberg.types.Types.StructType;
/**
@@ -56,7 +56,7 @@ public class Evaluator implements Serializable {
return visitor().eval(data);
}
- private class EvalVisitor extends BoundExpressionVisitor<Boolean> {
+ private class EvalVisitor extends BoundVisitor<Boolean> {
private StructLike struct;
private boolean eval(StructLike row) {
@@ -90,63 +90,63 @@ public class Evaluator implements Serializable {
}
@Override
- public <T> Boolean isNull(BoundReference<T> ref) {
- return ref.get(struct) == null;
+ public <T> Boolean isNull(Bound<T> valueExpr) {
+ return valueExpr.eval(struct) == null;
}
@Override
- public <T> Boolean notNull(BoundReference<T> ref) {
- return ref.get(struct) != null;
+ public <T> Boolean notNull(Bound<T> valueExpr) {
+ return valueExpr.eval(struct) != null;
}
@Override
- public <T> Boolean lt(BoundReference<T> ref, Literal<T> lit) {
+ public <T> Boolean lt(Bound<T> valueExpr, Literal<T> lit) {
Comparator<T> cmp = lit.comparator();
- return cmp.compare(ref.get(struct), lit.value()) < 0;
+ return cmp.compare(valueExpr.eval(struct), lit.value()) < 0;
}
@Override
- public <T> Boolean ltEq(BoundReference<T> ref, Literal<T> lit) {
+ public <T> Boolean ltEq(Bound<T> valueExpr, Literal<T> lit) {
Comparator<T> cmp = lit.comparator();
- return cmp.compare(ref.get(struct), lit.value()) <= 0;
+ return cmp.compare(valueExpr.eval(struct), lit.value()) <= 0;
}
@Override
- public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
+ public <T> Boolean gt(Bound<T> valueExpr, Literal<T> lit) {
Comparator<T> cmp = lit.comparator();
- return cmp.compare(ref.get(struct), lit.value()) > 0;
+ return cmp.compare(valueExpr.eval(struct), lit.value()) > 0;
}
@Override
- public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
+ public <T> Boolean gtEq(Bound<T> valueExpr, Literal<T> lit) {
Comparator<T> cmp = lit.comparator();
- return cmp.compare(ref.get(struct), lit.value()) >= 0;
+ return cmp.compare(valueExpr.eval(struct), lit.value()) >= 0;
}
@Override
- public <T> Boolean eq(BoundReference<T> ref, Literal<T> lit) {
+ public <T> Boolean eq(Bound<T> valueExpr, Literal<T> lit) {
Comparator<T> cmp = lit.comparator();
- return cmp.compare(ref.get(struct), lit.value()) == 0;
+ return cmp.compare(valueExpr.eval(struct), lit.value()) == 0;
}
@Override
- public <T> Boolean notEq(BoundReference<T> ref, Literal<T> lit) {
- return !eq(ref, lit);
+ public <T> Boolean notEq(Bound<T> valueExpr, Literal<T> lit) {
+ return !eq(valueExpr, lit);
}
@Override
- public <T> Boolean in(BoundReference<T> ref, Set<T> literalSet) {
- return literalSet.contains(ref.get(struct));
+ public <T> Boolean in(Bound<T> valueExpr, Set<T> literalSet) {
+ return literalSet.contains(valueExpr.eval(struct));
}
@Override
- public <T> Boolean notIn(BoundReference<T> ref, Set<T> literalSet) {
- return !in(ref, literalSet);
+ public <T> Boolean notIn(Bound<T> valueExpr, Set<T> literalSet) {
+ return !in(valueExpr, literalSet);
}
@Override
- public <T> Boolean startsWith(BoundReference<T> ref, Literal<T> lit) {
- return ((String) ref.get(struct)).startsWith((String) lit.value());
+ public <T> Boolean startsWith(Bound<T> valueExpr, Literal<T> lit) {
+ return ((String) valueExpr.eval(struct)).startsWith((String) lit.value());
}
}
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/ExpressionVisitors.java b/api/src/main/java/org/apache/iceberg/expressions/ExpressionVisitors.java
index d57854d..4677953 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/ExpressionVisitors.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/ExpressionVisitors.java
@@ -20,6 +20,7 @@
package org.apache.iceberg.expressions;
import java.util.Set;
+import org.apache.iceberg.exceptions.ValidationException;
/**
* Utils for traversing {@link Expression expressions}.
@@ -58,6 +59,12 @@ public class ExpressionVisitors {
}
}
+ /**
+ * This base class is used by existing visitors that have not been updated to extend BoundExpressionVisitor.
+ *
+ * @deprecated use {@link BoundVisitor} instead
+ */
+ @Deprecated
public abstract static class BoundExpressionVisitor<R> extends ExpressionVisitor<R> {
public <T> R isNull(BoundReference<T> ref) {
return null;
@@ -105,23 +112,124 @@ public class ExpressionVisitors {
@Override
public <T> R predicate(BoundPredicate<T> pred) {
+ ValidationException.check(pred.term() instanceof BoundReference,
+ "Visitor %s does not support expression: %s", this, pred.term());
+
+ if (pred.isLiteralPredicate()) {
+ BoundLiteralPredicate<T> literalPred = pred.asLiteralPredicate();
+ switch (pred.op()) {
+ case LT:
+ return lt((BoundReference<T>) pred.term(), literalPred.literal());
+ case LT_EQ:
+ return ltEq((BoundReference<T>) pred.term(), literalPred.literal());
+ case GT:
+ return gt((BoundReference<T>) pred.term(), literalPred.literal());
+ case GT_EQ:
+ return gtEq((BoundReference<T>) pred.term(), literalPred.literal());
+ case EQ:
+ return eq((BoundReference<T>) pred.term(), literalPred.literal());
+ case NOT_EQ:
+ return notEq((BoundReference<T>) pred.term(), literalPred.literal());
+ case STARTS_WITH:
+ return startsWith((BoundReference<T>) pred.term(), literalPred.literal());
+ default:
+ throw new IllegalStateException("Invalid operation for BoundLiteralPredicate: " + pred.op());
+ }
+
+ } else if (pred.isUnaryPredicate()) {
+ switch (pred.op()) {
+ case IS_NULL:
+ return isNull((BoundReference<T>) pred.term());
+ case NOT_NULL:
+ return notNull((BoundReference<T>) pred.term());
+ default:
+ throw new IllegalStateException("Invalid operation for BoundUnaryPredicate: " + pred.op());
+ }
+
+ } else if (pred.isSetPredicate()) {
+ switch (pred.op()) {
+ case IN:
+ return in((BoundReference<T>) pred.term(), pred.asSetPredicate().literalSet());
+ case NOT_IN:
+ return notIn((BoundReference<T>) pred.term(), pred.asSetPredicate().literalSet());
+ default:
+ throw new IllegalStateException("Invalid operation for BoundSetPredicate: " + pred.op());
+ }
+ }
+
+ throw new IllegalStateException("Unsupported bound predicate: " + pred.getClass().getName());
+ }
+
+ @Override
+ public <T> R predicate(UnboundPredicate<T> pred) {
+ throw new UnsupportedOperationException("Not a bound predicate: " + pred);
+ }
+ }
+
+ public abstract static class BoundVisitor<R> extends ExpressionVisitor<R> {
+ public <T> R isNull(Bound<T> expr) {
+ return null;
+ }
+
+ public <T> R notNull(Bound<T> expr) {
+ return null;
+ }
+
+ public <T> R lt(Bound<T> expr, Literal<T> lit) {
+ return null;
+ }
+
+ public <T> R ltEq(Bound<T> expr, Literal<T> lit) {
+ return null;
+ }
+
+ public <T> R gt(Bound<T> expr, Literal<T> lit) {
+ return null;
+ }
+
+ public <T> R gtEq(Bound<T> expr, Literal<T> lit) {
+ return null;
+ }
+
+ public <T> R eq(Bound<T> expr, Literal<T> lit) {
+ return null;
+ }
+
+ public <T> R notEq(Bound<T> expr, Literal<T> lit) {
+ return null;
+ }
+
+ public <T> R in(Bound<T> expr, Set<T> literalSet) {
+ throw new UnsupportedOperationException("In operation is not supported by the visitor");
+ }
+
+ public <T> R notIn(Bound<T> expr, Set<T> literalSet) {
+ throw new UnsupportedOperationException("notIn operation is not supported by the visitor");
+ }
+
+ public <T> R startsWith(Bound<T> expr, Literal<T> lit) {
+ throw new UnsupportedOperationException("Unsupported operation.");
+ }
+
+ @Override
+ public <T> R predicate(BoundPredicate<T> pred) {
if (pred.isLiteralPredicate()) {
BoundLiteralPredicate<T> literalPred = pred.asLiteralPredicate();
switch (pred.op()) {
case LT:
- return lt(pred.ref(), literalPred.literal());
+ return lt(pred.term(), literalPred.literal());
case LT_EQ:
- return ltEq(pred.ref(), literalPred.literal());
+ return ltEq(pred.term(), literalPred.literal());
case GT:
- return gt(pred.ref(), literalPred.literal());
+ return gt(pred.term(), literalPred.literal());
case GT_EQ:
- return gtEq(pred.ref(), literalPred.literal());
+ return gtEq(pred.term(), literalPred.literal());
case EQ:
- return eq(pred.ref(), literalPred.literal());
+ return eq(pred.term(), literalPred.literal());
case NOT_EQ:
- return notEq(pred.ref(), literalPred.literal());
+ return notEq(pred.term(), literalPred.literal());
case STARTS_WITH:
- return startsWith(pred.ref(), literalPred.literal());
+ return startsWith(pred.term(), literalPred.literal());
default:
throw new IllegalStateException("Invalid operation for BoundLiteralPredicate: " + pred.op());
}
@@ -129,9 +237,9 @@ public class ExpressionVisitors {
} else if (pred.isUnaryPredicate()) {
switch (pred.op()) {
case IS_NULL:
- return isNull(pred.ref());
+ return isNull(pred.term());
case NOT_NULL:
- return notNull(pred.ref());
+ return notNull(pred.term());
default:
throw new IllegalStateException("Invalid operation for BoundUnaryPredicate: " + pred.op());
}
@@ -139,9 +247,9 @@ public class ExpressionVisitors {
} else if (pred.isSetPredicate()) {
switch (pred.op()) {
case IN:
- return in(pred.ref(), pred.asSetPredicate().literalSet());
+ return in(pred.term(), pred.asSetPredicate().literalSet());
case NOT_IN:
- return notIn(pred.ref(), pred.asSetPredicate().literalSet());
+ return notIn(pred.term(), pred.asSetPredicate().literalSet());
default:
throw new IllegalStateException("Invalid operation for BoundSetPredicate: " + pred.op());
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Expressions.java b/api/src/main/java/org/apache/iceberg/expressions/Expressions.java
index d62974b..050f9b4 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Expressions.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Expressions.java
@@ -23,6 +23,9 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import java.util.stream.Stream;
import org.apache.iceberg.expressions.Expression.Operation;
+import org.apache.iceberg.transforms.Transform;
+import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.types.Types;
/**
* Factory methods for creating {@link Expression expressions}.
@@ -46,7 +49,7 @@ public class Expressions {
public static Expression and(Expression left, Expression right, Expression... expressions) {
return Stream.of(expressions)
- .reduce(and(left, right), Expressions::and);
+ .reduce(and(left, right), Expressions::and);
}
public static Expression or(Expression left, Expression right) {
@@ -74,60 +77,144 @@ public class Expressions {
return new Not(child);
}
+ @SuppressWarnings("unchecked")
+ public static <T> UnboundTerm<T> bucket(String name, int numBuckets) {
+ Transform<?, T> transform = (Transform<?, T>) Transforms.bucket(Types.StringType.get(), numBuckets);
+ return new UnboundTransform<>(ref(name), transform);
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> UnboundTerm<T> year(String name) {
+ return new UnboundTransform<>(ref(name), (Transform<?, T>) Transforms.year(Types.TimestampType.withZone()));
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> UnboundTerm<T> month(String name) {
+ return new UnboundTransform<>(ref(name), (Transform<?, T>) Transforms.month(Types.TimestampType.withZone()));
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> UnboundTerm<T> day(String name) {
+ return new UnboundTransform<>(ref(name), (Transform<?, T>) Transforms.day(Types.TimestampType.withZone()));
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> UnboundTerm<T> hour(String name) {
+ return new UnboundTransform<>(ref(name), (Transform<?, T>) Transforms.hour(Types.TimestampType.withZone()));
+ }
+
+ public static <T> UnboundTerm<T> truncate(String name, int width) {
+ return new UnboundTransform<>(ref(name), Transforms.truncate(Types.LongType.get(), width));
+ }
+
public static <T> UnboundPredicate<T> isNull(String name) {
return new UnboundPredicate<>(Expression.Operation.IS_NULL, ref(name));
}
+ public static <T> UnboundPredicate<T> isNull(UnboundTerm<T> expr) {
+ return new UnboundPredicate<>(Expression.Operation.IS_NULL, expr);
+ }
+
public static <T> UnboundPredicate<T> notNull(String name) {
return new UnboundPredicate<>(Expression.Operation.NOT_NULL, ref(name));
}
+ public static <T> UnboundPredicate<T> notNull(UnboundTerm<T> expr) {
+ return new UnboundPredicate<>(Expression.Operation.IS_NULL, expr);
+ }
+
public static <T> UnboundPredicate<T> lessThan(String name, T value) {
return new UnboundPredicate<>(Expression.Operation.LT, ref(name), value);
}
+ public static <T> UnboundPredicate<T> lessThan(UnboundTerm<T> expr, T value) {
+ return new UnboundPredicate<>(Expression.Operation.LT, expr, value);
+ }
+
public static <T> UnboundPredicate<T> lessThanOrEqual(String name, T value) {
return new UnboundPredicate<>(Expression.Operation.LT_EQ, ref(name), value);
}
+ public static <T> UnboundPredicate<T> lessThanOrEqual(UnboundTerm<T> expr, T value) {
+ return new UnboundPredicate<>(Expression.Operation.LT_EQ, expr, value);
+ }
+
public static <T> UnboundPredicate<T> greaterThan(String name, T value) {
return new UnboundPredicate<>(Expression.Operation.GT, ref(name), value);
}
+ public static <T> UnboundPredicate<T> greaterThan(UnboundTerm<T> expr, T value) {
+ return new UnboundPredicate<>(Expression.Operation.GT, expr, value);
+ }
+
public static <T> UnboundPredicate<T> greaterThanOrEqual(String name, T value) {
return new UnboundPredicate<>(Expression.Operation.GT_EQ, ref(name), value);
}
+ public static <T> UnboundPredicate<T> greaterThanOrEqual(UnboundTerm<T> expr, T value) {
+ return new UnboundPredicate<>(Expression.Operation.GT_EQ, expr, value);
+ }
+
public static <T> UnboundPredicate<T> equal(String name, T value) {
return new UnboundPredicate<>(Expression.Operation.EQ, ref(name), value);
}
+ public static <T> UnboundPredicate<T> equal(UnboundTerm<T> expr, T value) {
+ return new UnboundPredicate<>(Expression.Operation.EQ, expr, value);
+ }
+
public static <T> UnboundPredicate<T> notEqual(String name, T value) {
return new UnboundPredicate<>(Expression.Operation.NOT_EQ, ref(name), value);
}
+ public static <T> UnboundPredicate<T> notEqual(UnboundTerm<T> expr, T value) {
+ return new UnboundPredicate<>(Expression.Operation.NOT_EQ, expr, value);
+ }
+
public static UnboundPredicate<String> startsWith(String name, String value) {
return new UnboundPredicate<>(Expression.Operation.STARTS_WITH, ref(name), value);
}
+ public static UnboundPredicate<String> startsWith(UnboundTerm<String> expr, String value) {
+ return new UnboundPredicate<>(Expression.Operation.STARTS_WITH, expr, value);
+ }
+
public static <T> UnboundPredicate<T> in(String name, T... values) {
return predicate(Operation.IN, name, Lists.newArrayList(values));
}
+ public static <T> UnboundPredicate<T> in(UnboundTerm<T> expr, T... values) {
+ return predicate(Operation.IN, expr, Lists.newArrayList(values));
+ }
+
public static <T> UnboundPredicate<T> in(String name, Iterable<T> values) {
Preconditions.checkNotNull(values, "Values cannot be null for IN predicate.");
- return predicate(Operation.IN, name, values);
+ return predicate(Operation.IN, ref(name), values);
+ }
+
+ public static <T> UnboundPredicate<T> in(UnboundTerm<T> expr, Iterable<T> values) {
+ Preconditions.checkNotNull(values, "Values cannot be null for IN predicate.");
+ return predicate(Operation.IN, expr, values);
}
public static <T> UnboundPredicate<T> notIn(String name, T... values) {
return predicate(Operation.NOT_IN, name, Lists.newArrayList(values));
}
+ public static <T> UnboundPredicate<T> notIn(UnboundTerm<T> expr, T... values) {
+ return predicate(Operation.NOT_IN, expr, Lists.newArrayList(values));
+ }
+
public static <T> UnboundPredicate<T> notIn(String name, Iterable<T> values) {
Preconditions.checkNotNull(values, "Values cannot be null for NOT_IN predicate.");
return predicate(Operation.NOT_IN, name, values);
}
+ public static <T> UnboundPredicate<T> notIn(UnboundTerm<T> expr, Iterable<T> values) {
+ Preconditions.checkNotNull(values, "Values cannot be null for NOT_IN predicate.");
+ return predicate(Operation.NOT_IN, expr, values);
+ }
+
public static <T> UnboundPredicate<T> predicate(Operation op, String name, T value) {
return predicate(op, name, Literals.from(value));
}
@@ -135,11 +222,11 @@ public class Expressions {
public static <T> UnboundPredicate<T> predicate(Operation op, String name, Literal<T> lit) {
Preconditions.checkArgument(op != Operation.IS_NULL && op != Operation.NOT_NULL,
"Cannot create %s predicate inclusive a value", op);
- return new UnboundPredicate<>(op, ref(name), lit);
+ return new UnboundPredicate<T>(op, ref(name), lit);
}
public static <T> UnboundPredicate<T> predicate(Operation op, String name, Iterable<T> values) {
- return new UnboundPredicate<>(op, ref(name), values);
+ return predicate(op, ref(name), values);
}
public static <T> UnboundPredicate<T> predicate(Operation op, String name) {
@@ -148,6 +235,10 @@ public class Expressions {
return new UnboundPredicate<>(op, ref(name));
}
+ private static <T> UnboundPredicate<T> predicate(Operation op, UnboundTerm<T> expr, Iterable<T> values) {
+ return new UnboundPredicate<>(op, expr, values);
+ }
+
public static True alwaysTrue() {
return True.INSTANCE;
}
@@ -160,7 +251,7 @@ public class Expressions {
return ExpressionVisitors.visit(expr, RewriteNot.get());
}
- static NamedReference ref(String name) {
- return new NamedReference(name);
+ static <T> NamedReference<T> ref(String name) {
+ return new NamedReference<>(name);
}
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/InclusiveMetricsEvaluator.java b/api/src/main/java/org/apache/iceberg/expressions/InclusiveMetricsEvaluator.java
index 6da73ac..9811f96 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/InclusiveMetricsEvaluator.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/InclusiveMetricsEvaluator.java
@@ -43,10 +43,7 @@ import static org.apache.iceberg.expressions.Expressions.rewriteNot;
* return value of {@code eval} is false.
*/
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() {
@@ -61,9 +58,7 @@ public class InclusiveMetricsEvaluator {
}
public InclusiveMetricsEvaluator(Schema schema, Expression unbound, boolean caseSensitive) {
- this.schema = schema;
- this.struct = schema.asStruct();
- this.caseSensitive = caseSensitive;
+ StructType struct = schema.asStruct();
this.expr = Binder.bind(struct, rewriteNot(unbound), caseSensitive);
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java b/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java
index 8a88ce3..4ed4440 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java
@@ -20,8 +20,11 @@
package org.apache.iceberg.expressions;
import com.google.common.base.Preconditions;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.types.Types;
-public class NamedReference implements Reference {
+public class NamedReference<T> implements UnboundTerm<T>, Reference<T> {
private final String name;
NamedReference(String name) {
@@ -33,6 +36,23 @@ public class NamedReference implements Reference {
return name;
}
+ public BoundReference<T> bind(Types.StructType struct, boolean caseSensitive) {
+ Schema schema = new Schema(struct.fields());
+ Types.NestedField field = caseSensitive ?
+ schema.findField(name) :
+ schema.caseInsensitiveFindField(name);
+
+ ValidationException.check(field != null,
+ "Cannot find field '%s' in struct: %s", name, schema.asStruct());
+
+ return new BoundReference<>(field, schema.accessorForField(field.fieldId()));
+ }
+
+ @Override
+ public NamedReference<T> ref() {
+ return this;
+ }
+
@Override
public String toString() {
return String.format("ref(name=\"%s\")", name);
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Predicate.java b/api/src/main/java/org/apache/iceberg/expressions/Predicate.java
index b169a37..3473048 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Predicate.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Predicate.java
@@ -19,13 +19,16 @@
package org.apache.iceberg.expressions;
-public abstract class Predicate<R extends Reference> implements Expression {
+import com.google.common.base.Preconditions;
+
+public abstract class Predicate<T, C extends Term> implements Expression {
private final Operation op;
- private final R ref;
+ private final C term;
- Predicate(Operation op, R ref) {
+ Predicate(Operation op, C term) {
+ Preconditions.checkNotNull(term, "Term cannot be null");
this.op = op;
- this.ref = ref;
+ this.term = term;
}
@Override
@@ -33,7 +36,7 @@ public abstract class Predicate<R extends Reference> implements Expression {
return op;
}
- public R ref() {
- return ref;
+ public C term() {
+ return term;
}
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Reference.java b/api/src/main/java/org/apache/iceberg/expressions/Reference.java
index f08867f..41c7b0f 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Reference.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Reference.java
@@ -19,12 +19,10 @@
package org.apache.iceberg.expressions;
-import java.io.Serializable;
-
/**
* Represents a variable reference in an {@link Expression expression}.
* @see BoundReference
* @see NamedReference
*/
-public interface Reference extends Serializable {
+public interface Reference<T> extends Term {
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/ResidualEvaluator.java b/api/src/main/java/org/apache/iceberg/expressions/ResidualEvaluator.java
index 64315c8..a6c390d 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/ResidualEvaluator.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/ResidualEvaluator.java
@@ -25,6 +25,7 @@ import java.util.List;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.StructLike;
+import org.apache.iceberg.expressions.ExpressionVisitors.BoundExpressionVisitor;
import org.apache.iceberg.transforms.Transform;
/**
@@ -115,7 +116,7 @@ public class ResidualEvaluator implements Serializable {
return visitor().eval(partitionData);
}
- private class ResidualVisitor extends ExpressionVisitors.BoundExpressionVisitor<Expression> {
+ private class ResidualVisitor extends BoundExpressionVisitor<Expression> {
private StructLike struct;
private Expression eval(StructLike dataStruct) {
@@ -150,68 +151,65 @@ public class ResidualEvaluator implements Serializable {
@Override
public <T> Expression isNull(BoundReference<T> ref) {
- return (ref.get(struct) == null) ? alwaysTrue() : alwaysFalse();
+ return (ref.eval(struct) == null) ? alwaysTrue() : alwaysFalse();
}
@Override
public <T> Expression notNull(BoundReference<T> ref) {
- return (ref.get(struct) != null) ? alwaysTrue() : alwaysFalse();
+ return (ref.eval(struct) != null) ? alwaysTrue() : alwaysFalse();
}
@Override
public <T> Expression lt(BoundReference<T> ref, Literal<T> lit) {
Comparator<T> cmp = lit.comparator();
- return (cmp.compare(ref.get(struct), lit.value()) < 0) ? alwaysTrue() : alwaysFalse();
+ return (cmp.compare(ref.eval(struct), lit.value()) < 0) ? alwaysTrue() : alwaysFalse();
}
@Override
public <T> Expression ltEq(BoundReference<T> ref, Literal<T> lit) {
Comparator<T> cmp = lit.comparator();
- return (cmp.compare(ref.get(struct), lit.value()) <= 0) ? alwaysTrue() : alwaysFalse();
+ return (cmp.compare(ref.eval(struct), lit.value()) <= 0) ? alwaysTrue() : alwaysFalse();
}
@Override
public <T> Expression gt(BoundReference<T> ref, Literal<T> lit) {
Comparator<T> cmp = lit.comparator();
- return (cmp.compare(ref.get(struct), lit.value()) > 0) ? alwaysTrue() : alwaysFalse();
+ return (cmp.compare(ref.eval(struct), lit.value()) > 0) ? alwaysTrue() : alwaysFalse();
}
@Override
public <T> Expression gtEq(BoundReference<T> ref, Literal<T> lit) {
Comparator<T> cmp = lit.comparator();
- return (cmp.compare(ref.get(struct), lit.value()) >= 0) ? alwaysTrue() : alwaysFalse();
+ return (cmp.compare(ref.eval(struct), lit.value()) >= 0) ? alwaysTrue() : alwaysFalse();
}
@Override
public <T> Expression eq(BoundReference<T> ref, Literal<T> lit) {
Comparator<T> cmp = lit.comparator();
- return (cmp.compare(ref.get(struct), lit.value()) == 0) ? alwaysTrue() : alwaysFalse();
+ return (cmp.compare(ref.eval(struct), lit.value()) == 0) ? alwaysTrue() : alwaysFalse();
}
@Override
public <T> Expression notEq(BoundReference<T> ref, Literal<T> lit) {
Comparator<T> cmp = lit.comparator();
- return (cmp.compare(ref.get(struct), lit.value()) != 0) ? alwaysTrue() : alwaysFalse();
+ return (cmp.compare(ref.eval(struct), lit.value()) != 0) ? alwaysTrue() : alwaysFalse();
}
@Override
public <T> Expression startsWith(BoundReference<T> ref, Literal<T> lit) {
- return ((String) ref.get(struct)).startsWith((String) lit.value()) ? alwaysTrue() : alwaysFalse();
+ return ((String) ref.eval(struct)).startsWith((String) lit.value()) ? alwaysTrue() : alwaysFalse();
}
@Override
@SuppressWarnings("unchecked")
public <T> Expression predicate(BoundPredicate<T> pred) {
- /**
- * Get the strict projection and inclusive projection of this predicate in partition data,
- * then use them to determine whether to return the original predicate. The strict projection
- * returns true iff the original predicate would have returned true, so the predicate can be
- * eliminated if the strict projection evaluates to true. Similarly the inclusive projection
- * returns false iff the original predicate would have returned false, so the predicate can
- * also be eliminated if the inclusive projection evaluates to false.
- */
-
- //
+ // Get the strict projection and inclusive projection of this predicate in partition data,
+ // then use them to determine whether to return the original predicate. The strict projection
+ // returns true iff the original predicate would have returned true, so the predicate can be
+ // eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ // returns false iff the original predicate would have returned false, so the predicate can
+ // also be eliminated if the inclusive projection evaluates to false.
+
// If there is no strict projection or if it evaluates to false, then return the predicate.
List<PartitionField> parts = spec.getFieldsBySourceId(pred.ref().fieldId());
if (parts == null) {
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Reference.java b/api/src/main/java/org/apache/iceberg/expressions/Term.java
similarity index 84%
copy from api/src/main/java/org/apache/iceberg/expressions/Reference.java
copy to api/src/main/java/org/apache/iceberg/expressions/Term.java
index f08867f..0d2276b 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Reference.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Term.java
@@ -22,9 +22,7 @@ package org.apache.iceberg.expressions;
import java.io.Serializable;
/**
- * Represents a variable reference in an {@link Expression expression}.
- * @see BoundReference
- * @see NamedReference
+ * An expression that evaluates to a value.
*/
-public interface Reference extends Serializable {
+public interface Term extends Serializable {
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Reference.java b/api/src/main/java/org/apache/iceberg/expressions/Unbound.java
similarity index 56%
copy from api/src/main/java/org/apache/iceberg/expressions/Reference.java
copy to api/src/main/java/org/apache/iceberg/expressions/Unbound.java
index f08867f..f160f93 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Reference.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Unbound.java
@@ -19,12 +19,25 @@
package org.apache.iceberg.expressions;
-import java.io.Serializable;
+import org.apache.iceberg.types.Types;
/**
- * Represents a variable reference in an {@link Expression expression}.
- * @see BoundReference
- * @see NamedReference
+ * Represents an unbound expression node.
+ * @param <T> the Java type of values produced by this node
+ * @param <B> the Java type produced when this node is bound using {@link #bind(Types.StructType, boolean)}
*/
-public interface Reference extends Serializable {
+public interface Unbound<T, B> {
+ /**
+ * Bind this value expression to concrete types.
+ *
+ * @param struct input data types
+ * @param caseSensitive whether binding should match columns using case sensitive resolution
+ * @return a bound value expression
+ */
+ B bind(Types.StructType struct, boolean caseSensitive);
+
+ /**
+ * @return this expression's underlying reference
+ */
+ NamedReference<?> ref();
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java b/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
index 008af7e..1bec4d2 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
@@ -26,45 +26,47 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import java.util.List;
import java.util.Set;
-import org.apache.iceberg.Schema;
import org.apache.iceberg.exceptions.ValidationException;
-import org.apache.iceberg.types.Type;
-import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.StructType;
import org.apache.iceberg.util.CharSequenceSet;
-public class UnboundPredicate<T> extends Predicate<NamedReference> {
+public class UnboundPredicate<T> extends Predicate<T, UnboundTerm<T>> implements Unbound<T, Expression> {
private static final Joiner COMMA = Joiner.on(", ");
private final List<Literal<T>> literals;
- UnboundPredicate(Operation op, NamedReference namedRef, T value) {
- this(op, namedRef, Literals.from(value));
+ UnboundPredicate(Operation op, UnboundTerm<T> term, T value) {
+ this(op, term, Literals.from(value));
}
- UnboundPredicate(Operation op, NamedReference namedRef) {
- super(op, namedRef);
+ UnboundPredicate(Operation op, UnboundTerm<T> term) {
+ super(op, term);
this.literals = null;
}
- UnboundPredicate(Operation op, NamedReference namedRef, Literal<T> lit) {
- super(op, namedRef);
+ UnboundPredicate(Operation op, UnboundTerm<T> term, Literal<T> lit) {
+ super(op, term);
this.literals = Lists.newArrayList(lit);
}
- UnboundPredicate(Operation op, NamedReference namedRef, Iterable<T> values) {
- super(op, namedRef);
+ UnboundPredicate(Operation op, UnboundTerm<T> term, Iterable<T> values) {
+ super(op, term);
this.literals = Lists.newArrayList(Iterables.transform(values, Literals::from));
}
- private UnboundPredicate(Operation op, NamedReference namedRef, List<Literal<T>> literals) {
- super(op, namedRef);
+ private UnboundPredicate(Operation op, UnboundTerm<T> term, List<Literal<T>> literals) {
+ super(op, term);
this.literals = literals;
}
@Override
+ public NamedReference<?> ref() {
+ return term().ref();
+ }
+
+ @Override
public Expression negate() {
- return new UnboundPredicate<>(op().negate(), ref(), literals);
+ return new UnboundPredicate<>(op().negate(), term(), literals);
}
public Literal<T> literal() {
@@ -99,51 +101,43 @@ public class UnboundPredicate<T> extends Predicate<NamedReference> {
* @throws ValidationException if literals do not match bound references, or if comparison on expression is invalid
*/
public Expression bind(StructType struct, boolean caseSensitive) {
- Schema schema = new Schema(struct.fields());
- Types.NestedField field = caseSensitive ?
- schema.findField(ref().name()) :
- schema.caseInsensitiveFindField(ref().name());
-
- ValidationException.check(field != null,
- "Cannot find field '%s' in struct: %s", ref().name(), schema.asStruct());
-
- BoundReference<T> ref = new BoundReference<>(field.fieldId(), schema.accessorForField(field.fieldId()));
+ BoundTerm<T> bound = term().bind(struct, caseSensitive);
if (literals == null) {
- return bindUnaryOperation(ref, field.isRequired());
+ return bindUnaryOperation(bound);
}
if (op() == Operation.IN || op() == Operation.NOT_IN) {
- return bindInOperation(ref, field.type(), op());
+ return bindInOperation(bound);
}
- return bindLiteralOperation(ref, field.type());
+ return bindLiteralOperation(bound);
}
- private Expression bindUnaryOperation(BoundReference<T> ref, boolean isRequired) {
+ private Expression bindUnaryOperation(BoundTerm<T> boundTerm) {
switch (op()) {
case IS_NULL:
- if (isRequired) {
+ if (boundTerm.ref().field().isRequired()) {
return Expressions.alwaysFalse();
}
- return new BoundUnaryPredicate<>(Operation.IS_NULL, ref);
+ return new BoundUnaryPredicate<>(Operation.IS_NULL, boundTerm);
case NOT_NULL:
- if (isRequired) {
+ if (boundTerm.ref().field().isRequired()) {
return Expressions.alwaysTrue();
}
- return new BoundUnaryPredicate<>(Operation.NOT_NULL, ref);
+ return new BoundUnaryPredicate<>(Operation.NOT_NULL, boundTerm);
default:
throw new ValidationException("Operation must be IS_NULL or NOT_NULL");
}
}
- private Expression bindLiteralOperation(BoundReference<T> ref, Type type) {
- Literal<T> lit = literal().to(type);
+ private Expression bindLiteralOperation(BoundTerm<T> boundTerm) {
+ Literal<T> lit = literal().to(boundTerm.type());
if (lit == null) {
throw new ValidationException(String.format(
"Invalid value for conversion to type %s: %s (%s)",
- type, literal().value(), literal().value().getClass().getName()));
+ boundTerm.type(), literal().value(), literal().value().getClass().getName()));
} else if (lit == Literals.aboveMax()) {
switch (op()) {
@@ -169,21 +163,22 @@ public class UnboundPredicate<T> extends Predicate<NamedReference> {
}
}
- return new BoundLiteralPredicate<>(op(), ref, lit);
+ // TODO: translate truncate(col) == value to startsWith(value)
+ return new BoundLiteralPredicate<>(op(), boundTerm, lit);
}
- private Expression bindInOperation(BoundReference<T> ref, Type type, Operation op) {
+ private Expression bindInOperation(BoundTerm<T> boundTerm) {
List<Literal<T>> convertedLiterals = Lists.newArrayList(Iterables.filter(
Lists.transform(literals, lit -> {
- Literal<T> converted = lit.to(type);
+ Literal<T> converted = lit.to(boundTerm.type());
ValidationException.check(converted != null,
- "Invalid value for conversion to type %s: %s (%s)", type, lit, lit.getClass().getName());
+ "Invalid value for conversion to type %s: %s (%s)", boundTerm.type(), lit, lit.getClass().getName());
return converted;
}),
lit -> lit != Literals.aboveMax() && lit != Literals.belowMin()));
if (convertedLiterals.isEmpty()) {
- switch (op) {
+ switch (op()) {
case IN:
return Expressions.alwaysFalse();
case NOT_IN:
@@ -195,44 +190,44 @@ public class UnboundPredicate<T> extends Predicate<NamedReference> {
Set<T> literalSet = setOf(convertedLiterals);
if (literalSet.size() == 1) {
- switch (op) {
+ switch (op()) {
case IN:
- return new BoundLiteralPredicate<>(Operation.EQ, ref, Iterables.get(convertedLiterals, 0));
+ return new BoundLiteralPredicate<>(Operation.EQ, boundTerm, Iterables.get(convertedLiterals, 0));
case NOT_IN:
- return new BoundLiteralPredicate<>(Operation.NOT_EQ, ref, Iterables.get(convertedLiterals, 0));
+ return new BoundLiteralPredicate<>(Operation.NOT_EQ, boundTerm, Iterables.get(convertedLiterals, 0));
default:
throw new ValidationException("Operation must be IN or NOT_IN");
}
}
- return new BoundSetPredicate<>(op, ref, literalSet);
+ return new BoundSetPredicate<>(op(), boundTerm, literalSet);
}
@Override
public String toString() {
switch (op()) {
case IS_NULL:
- return "is_null(" + ref() + ")";
+ return "is_null(" + term() + ")";
case NOT_NULL:
- return "not_null(" + ref() + ")";
+ return "not_null(" + term() + ")";
case LT:
- return ref() + " < " + literal();
+ return term() + " < " + literal();
case LT_EQ:
- return ref() + " <= " + literal();
+ return term() + " <= " + literal();
case GT:
- return ref() + " > " + literal();
+ return term() + " > " + literal();
case GT_EQ:
- return ref() + " >= " + literal();
+ return term() + " >= " + literal();
case EQ:
- return ref() + " == " + literal();
+ return term() + " == " + literal();
case NOT_EQ:
- return ref() + " != " + literal();
+ return term() + " != " + literal();
case STARTS_WITH:
- return ref() + " startsWith \"" + literal() + "\"";
+ return term() + " startsWith \"" + literal() + "\"";
case IN:
- return ref() + " in (" + COMMA.join(literals()) + ")";
+ return term() + " in (" + COMMA.join(literals()) + ")";
case NOT_IN:
- return ref() + " not in (" + COMMA.join(literals()) + ")";
+ return term() + " not in (" + COMMA.join(literals()) + ")";
default:
return "Invalid predicate: operation = " + op();
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Reference.java b/api/src/main/java/org/apache/iceberg/expressions/UnboundTerm.java
similarity index 81%
copy from api/src/main/java/org/apache/iceberg/expressions/Reference.java
copy to api/src/main/java/org/apache/iceberg/expressions/UnboundTerm.java
index f08867f..412c7c0 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Reference.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/UnboundTerm.java
@@ -19,12 +19,10 @@
package org.apache.iceberg.expressions;
-import java.io.Serializable;
-
/**
- * Represents a variable reference in an {@link Expression expression}.
- * @see BoundReference
- * @see NamedReference
+ * Represents an unbound term.
+ *
+ * @param <T> the Java type of values produced by this term
*/
-public interface Reference extends Serializable {
+public interface UnboundTerm<T> extends Unbound<T, BoundTerm<T>>, Term {
}
diff --git a/api/src/main/java/org/apache/iceberg/expressions/UnboundTransform.java b/api/src/main/java/org/apache/iceberg/expressions/UnboundTransform.java
new file mode 100644
index 0000000..541592e
--- /dev/null
+++ b/api/src/main/java/org/apache/iceberg/expressions/UnboundTransform.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.expressions;
+
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.transforms.Transform;
+import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.types.Types;
+
+public class UnboundTransform<S, T> implements UnboundTerm<T>, Term {
+ private final NamedReference<S> ref;
+ private final Transform<S, T> transform;
+
+ UnboundTransform(NamedReference<S> ref, Transform<S, T> transform) {
+ this.ref = ref;
+ this.transform = transform;
+ }
+
+ @Override
+ public NamedReference<S> ref() {
+ return ref;
+ }
+
+ public Transform<S, T> transform() {
+ return transform;
+ }
+
+ @SuppressWarnings("unchecked")
+ public BoundTransform<S, T> bind(Types.StructType struct, boolean caseSensitive) {
+ BoundReference<S> boundRef = ref.bind(struct, caseSensitive);
+
+ Transform<S, T> typeTransform;
+ try {
+ // TODO: Avoid using toString/fromString
+ typeTransform = (Transform<S, T>) Transforms.fromString(boundRef.type(), transform.toString());
+ ValidationException.check(typeTransform.canTransform(boundRef.type()),
+ "Cannot bind: %s cannot transform %s values from '%s'", transform, boundRef.type(), ref.name());
+ } catch (IllegalArgumentException e) {
+ throw new ValidationException(
+ "Cannot bind: %s cannot transform %s values from '%s'", transform, boundRef.type(), ref.name());
+ }
+
+ return new BoundTransform<>(boundRef, typeTransform);
+ }
+
+ @Override
+ public String toString() {
+ return transform + "(" + ref + ")";
+ }
+}
diff --git a/api/src/main/java/org/apache/iceberg/transforms/Bucket.java b/api/src/main/java/org/apache/iceberg/transforms/Bucket.java
index c29ffb0..153e607 100644
--- a/api/src/main/java/org/apache/iceberg/transforms/Bucket.java
+++ b/api/src/main/java/org/apache/iceberg/transforms/Bucket.java
@@ -31,6 +31,7 @@ import java.util.Set;
import java.util.UUID;
import org.apache.iceberg.expressions.BoundLiteralPredicate;
import org.apache.iceberg.expressions.BoundPredicate;
+import org.apache.iceberg.expressions.BoundTransform;
import org.apache.iceberg.expressions.BoundUnaryPredicate;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.UnboundPredicate;
@@ -111,6 +112,10 @@ abstract class Bucket<T> implements Transform<T, Integer> {
@Override
public UnboundPredicate<Integer> project(String name, BoundPredicate<T> predicate) {
+ if (predicate.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, predicate);
+ }
+
if (predicate instanceof BoundUnaryPredicate) {
return Expressions.predicate(predicate.op(), name);
} else if (predicate instanceof BoundLiteralPredicate) {
@@ -135,6 +140,10 @@ abstract class Bucket<T> implements Transform<T, Integer> {
@Override
public UnboundPredicate<Integer> projectStrict(String name, BoundPredicate<T> predicate) {
+ if (predicate.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, predicate);
+ }
+
if (predicate instanceof BoundUnaryPredicate) {
return Expressions.predicate(predicate.op(), name);
} else if (predicate instanceof BoundLiteralPredicate) {
@@ -305,9 +314,9 @@ abstract class Bucket<T> implements Transform<T, Integer> {
@Override
public int hash(UUID value) {
return MURMUR3.newHasher(16)
- .putLong(Long.reverseBytes(value.getMostSignificantBits()))
- .putLong(Long.reverseBytes(value.getLeastSignificantBits()))
- .hash().asInt();
+ .putLong(Long.reverseBytes(value.getMostSignificantBits()))
+ .putLong(Long.reverseBytes(value.getLeastSignificantBits()))
+ .hash().asInt();
}
@Override
diff --git a/api/src/main/java/org/apache/iceberg/transforms/Dates.java b/api/src/main/java/org/apache/iceberg/transforms/Dates.java
index 3b1c2ec..80d3257 100644
--- a/api/src/main/java/org/apache/iceberg/transforms/Dates.java
+++ b/api/src/main/java/org/apache/iceberg/transforms/Dates.java
@@ -24,6 +24,7 @@ import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import org.apache.iceberg.expressions.BoundPredicate;
+import org.apache.iceberg.expressions.BoundTransform;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.UnboundPredicate;
import org.apache.iceberg.types.Type;
@@ -71,6 +72,10 @@ enum Dates implements Transform<Integer, Integer> {
@Override
public UnboundPredicate<Integer> project(String fieldName, BoundPredicate<Integer> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
if (pred.isUnaryPredicate()) {
return Expressions.predicate(pred.op(), fieldName);
} else if (pred.isLiteralPredicate()) {
@@ -81,6 +86,10 @@ enum Dates implements Transform<Integer, Integer> {
@Override
public UnboundPredicate<Integer> projectStrict(String fieldName, BoundPredicate<Integer> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
if (pred.isUnaryPredicate()) {
return Expressions.predicate(pred.op(), fieldName);
} else if (pred.isLiteralPredicate()) {
diff --git a/api/src/main/java/org/apache/iceberg/transforms/ProjectionUtil.java b/api/src/main/java/org/apache/iceberg/transforms/ProjectionUtil.java
index 9611f56..635b037 100644
--- a/api/src/main/java/org/apache/iceberg/transforms/ProjectionUtil.java
+++ b/api/src/main/java/org/apache/iceberg/transforms/ProjectionUtil.java
@@ -22,7 +22,10 @@ package org.apache.iceberg.transforms;
import java.math.BigDecimal;
import java.math.BigInteger;
import org.apache.iceberg.expressions.BoundLiteralPredicate;
+import org.apache.iceberg.expressions.BoundPredicate;
+import org.apache.iceberg.expressions.BoundTransform;
import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.UnboundPredicate;
import static org.apache.iceberg.expressions.Expressions.predicate;
@@ -217,4 +220,28 @@ class ProjectionUtil {
return null;
}
}
+
+ /**
+ * If the predicate has a transformed child that matches the given transform, return a predicate.
+ */
+ @SuppressWarnings("unchecked")
+ static <T> UnboundPredicate<T> projectTransformPredicate(Transform<?, T> transform,
+ String partitionName, BoundPredicate<?> pred) {
+ if (pred.term() instanceof BoundTransform && transform.equals(((BoundTransform<?, ?>) pred.term()).transform())) {
+ // the bound value must be a T because the transform matches
+ return (UnboundPredicate<T>) removeTransform(partitionName, pred);
+ }
+ return null;
+ }
+
+ private static <T> UnboundPredicate<T> removeTransform(String partitionName, BoundPredicate<T> pred) {
+ if (pred.isUnaryPredicate()) {
+ return Expressions.predicate(pred.op(), partitionName);
+ } else if (pred.isLiteralPredicate()) {
+ return Expressions.predicate(pred.op(), partitionName, pred.asLiteralPredicate().literal());
+ } else if (pred.isSetPredicate()) {
+ return Expressions.predicate(pred.op(), partitionName, pred.asSetPredicate().literalSet());
+ }
+ throw new UnsupportedOperationException("Cannot replace transform in unknown predicate: " + pred);
+ }
}
diff --git a/api/src/main/java/org/apache/iceberg/transforms/Timestamps.java b/api/src/main/java/org/apache/iceberg/transforms/Timestamps.java
index 11a8db2..596306b 100644
--- a/api/src/main/java/org/apache/iceberg/transforms/Timestamps.java
+++ b/api/src/main/java/org/apache/iceberg/transforms/Timestamps.java
@@ -24,6 +24,7 @@ import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import org.apache.iceberg.expressions.BoundPredicate;
+import org.apache.iceberg.expressions.BoundTransform;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.UnboundPredicate;
import org.apache.iceberg.types.Type;
@@ -73,6 +74,10 @@ enum Timestamps implements Transform<Long, Integer> {
@Override
public UnboundPredicate<Integer> project(String fieldName, BoundPredicate<Long> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
if (pred.isUnaryPredicate()) {
return Expressions.predicate(pred.op(), fieldName);
} else if (pred.isLiteralPredicate()) {
@@ -83,6 +88,10 @@ enum Timestamps implements Transform<Long, Integer> {
@Override
public UnboundPredicate<Integer> projectStrict(String fieldName, BoundPredicate<Long> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
if (pred.isUnaryPredicate()) {
return Expressions.predicate(pred.op(), fieldName);
} else if (pred.isLiteralPredicate()) {
diff --git a/api/src/main/java/org/apache/iceberg/transforms/Truncate.java b/api/src/main/java/org/apache/iceberg/transforms/Truncate.java
index f9fd60b..f3c8086 100644
--- a/api/src/main/java/org/apache/iceberg/transforms/Truncate.java
+++ b/api/src/main/java/org/apache/iceberg/transforms/Truncate.java
@@ -25,6 +25,7 @@ import java.math.BigInteger;
import java.nio.ByteBuffer;
import org.apache.iceberg.expressions.BoundLiteralPredicate;
import org.apache.iceberg.expressions.BoundPredicate;
+import org.apache.iceberg.expressions.BoundTransform;
import org.apache.iceberg.expressions.BoundUnaryPredicate;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
@@ -90,6 +91,10 @@ abstract class Truncate<T> implements Transform<T, T> {
@Override
public UnboundPredicate<Integer> project(String name, BoundPredicate<Integer> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
if (pred.isUnaryPredicate()) {
return Expressions.predicate(pred.op(), name);
} else if (pred.isLiteralPredicate()) {
@@ -100,6 +105,10 @@ abstract class Truncate<T> implements Transform<T, T> {
@Override
public UnboundPredicate<Integer> projectStrict(String name, BoundPredicate<Integer> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
// TODO: for integers, can this return the original predicate?
// No. the predicate needs to be in terms of the applied value. For all x, apply(x) <= x.
// Therefore, the lower bound can be transformed outside of a greater-than bound.
@@ -162,6 +171,10 @@ abstract class Truncate<T> implements Transform<T, T> {
@Override
public UnboundPredicate<Long> project(String name, BoundPredicate<Long> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
if (pred.isUnaryPredicate()) {
return Expressions.predicate(pred.op(), name);
} else if (pred.isLiteralPredicate()) {
@@ -172,6 +185,10 @@ abstract class Truncate<T> implements Transform<T, T> {
@Override
public UnboundPredicate<Long> projectStrict(String name, BoundPredicate<Long> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
if (pred.isUnaryPredicate()) {
return Expressions.predicate(pred.op(), name);
} else if (pred.isLiteralPredicate()) {
@@ -232,6 +249,10 @@ abstract class Truncate<T> implements Transform<T, T> {
@Override
public UnboundPredicate<CharSequence> project(String name,
BoundPredicate<CharSequence> predicate) {
+ if (predicate.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, predicate);
+ }
+
if (predicate.isUnaryPredicate()) {
return Expressions.predicate(predicate.op(), name);
} else if (predicate.isLiteralPredicate()) {
@@ -243,6 +264,10 @@ abstract class Truncate<T> implements Transform<T, T> {
@Override
public UnboundPredicate<CharSequence> projectStrict(String name,
BoundPredicate<CharSequence> predicate) {
+ if (predicate.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, predicate);
+ }
+
if (predicate instanceof BoundUnaryPredicate) {
return Expressions.predicate(predicate.op(), name);
} else if (predicate instanceof BoundLiteralPredicate) {
@@ -314,6 +339,10 @@ abstract class Truncate<T> implements Transform<T, T> {
@Override
public UnboundPredicate<ByteBuffer> project(String name,
BoundPredicate<ByteBuffer> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
if (pred.isUnaryPredicate()) {
return Expressions.predicate(pred.op(), name);
} else if (pred.isLiteralPredicate()) {
@@ -325,6 +354,10 @@ abstract class Truncate<T> implements Transform<T, T> {
@Override
public UnboundPredicate<ByteBuffer> projectStrict(String name,
BoundPredicate<ByteBuffer> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
if (pred.isUnaryPredicate()) {
return Expressions.predicate(pred.op(), name);
} else if (pred.isLiteralPredicate()) {
@@ -397,6 +430,10 @@ abstract class Truncate<T> implements Transform<T, T> {
@Override
public UnboundPredicate<BigDecimal> project(String name,
BoundPredicate<BigDecimal> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
if (pred.isUnaryPredicate()) {
return Expressions.predicate(pred.op(), name);
} else if (pred.isLiteralPredicate()) {
@@ -408,6 +445,10 @@ abstract class Truncate<T> implements Transform<T, T> {
@Override
public UnboundPredicate<BigDecimal> projectStrict(String name,
BoundPredicate<BigDecimal> pred) {
+ if (pred.term() instanceof BoundTransform) {
+ return ProjectionUtil.projectTransformPredicate(this, name, pred);
+ }
+
if (pred.isUnaryPredicate()) {
return Expressions.predicate(pred.op(), name);
} else if (pred.isLiteralPredicate()) {
diff --git a/api/src/main/java/org/apache/iceberg/types/Types.java b/api/src/main/java/org/apache/iceberg/types/Types.java
index f7e91a0..48e7c96 100644
--- a/api/src/main/java/org/apache/iceberg/types/Types.java
+++ b/api/src/main/java/org/apache/iceberg/types/Types.java
@@ -611,38 +611,36 @@ public class Types {
private Map<String, NestedField> lazyFieldsByName() {
if (fieldsByName == null) {
- indexFields();
+ ImmutableMap.Builder<String, NestedField> byNameBuilder = ImmutableMap.builder();
+ for (NestedField field : fields) {
+ byNameBuilder.put(field.name(), field);
+ }
+ fieldsByName = byNameBuilder.build();
}
return fieldsByName;
}
private Map<String, NestedField> lazyFieldsByLowerCaseName() {
if (fieldsByLowerCaseName == null) {
- indexFields();
+ ImmutableMap.Builder<String, NestedField> byLowerCaseNameBuilder = ImmutableMap.builder();
+ for (NestedField field : fields) {
+ byLowerCaseNameBuilder.put(field.name().toLowerCase(Locale.ROOT), field);
+ }
+ fieldsByLowerCaseName = byLowerCaseNameBuilder.build();
}
return fieldsByLowerCaseName;
}
private Map<Integer, NestedField> lazyFieldsById() {
if (fieldsById == null) {
- indexFields();
+ ImmutableMap.Builder<Integer, NestedField> byIdBuilder = ImmutableMap.builder();
+ for (NestedField field : fields) {
+ byIdBuilder.put(field.fieldId(), field);
+ }
+ this.fieldsById = byIdBuilder.build();
}
return fieldsById;
}
-
- private void indexFields() {
- ImmutableMap.Builder<String, NestedField> byNameBuilder = ImmutableMap.builder();
- ImmutableMap.Builder<String, NestedField> byLowerCaseNameBuilder = ImmutableMap.builder();
- ImmutableMap.Builder<Integer, NestedField> byIdBuilder = ImmutableMap.builder();
- for (NestedField field : fields) {
- byNameBuilder.put(field.name(), field);
- byLowerCaseNameBuilder.put(field.name().toLowerCase(Locale.ROOT), field);
- byIdBuilder.put(field.fieldId(), field);
- }
- this.fieldsByName = byNameBuilder.build();
- this.fieldsByLowerCaseName = byLowerCaseNameBuilder.build();
- this.fieldsById = byIdBuilder.build();
- }
}
public static class ListType extends NestedType {
diff --git a/api/src/test/java/org/apache/iceberg/catalog/TestTableIdentifier.java b/api/src/test/java/org/apache/iceberg/catalog/TestTableIdentifier.java
index 1216403..b67bdcd 100644
--- a/api/src/test/java/org/apache/iceberg/catalog/TestTableIdentifier.java
+++ b/api/src/test/java/org/apache/iceberg/catalog/TestTableIdentifier.java
@@ -41,4 +41,17 @@ public class TestTableIdentifier {
Assert.assertEquals("userdb", threeLevelIdentifier.namespace().levels()[1]);
Assert.assertEquals("tbl", threeLevelIdentifier.name());
}
+
+ @Test
+ public void testToLowerCase() {
+ Assert.assertEquals(
+ TableIdentifier.of("Tbl").toLowerCase(),
+ TableIdentifier.of("tbl"));
+ Assert.assertEquals(
+ TableIdentifier.of("dB", "TBL").toLowerCase(),
+ TableIdentifier.of("db", "tbl"));
+ Assert.assertEquals(
+ TableIdentifier.of("Catalog", "dB", "TBL").toLowerCase(),
+ TableIdentifier.of("catalog", "db", "tbl"));
+ }
}
diff --git a/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java b/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java
index fc4fc39..ea24ed7 100644
--- a/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java
+++ b/api/src/test/java/org/apache/iceberg/expressions/TestExpressionBinding.java
@@ -29,6 +29,7 @@ import org.junit.Test;
import static org.apache.iceberg.expressions.Expressions.alwaysFalse;
import static org.apache.iceberg.expressions.Expressions.alwaysTrue;
import static org.apache.iceberg.expressions.Expressions.and;
+import static org.apache.iceberg.expressions.Expressions.bucket;
import static org.apache.iceberg.expressions.Expressions.equal;
import static org.apache.iceberg.expressions.Expressions.greaterThan;
import static org.apache.iceberg.expressions.Expressions.lessThan;
@@ -97,9 +98,9 @@ public class TestExpressionBinding {
// make sure the refs are for the right fields
BoundPredicate<?> left = TestHelpers.assertAndUnwrap(and.left());
- Assert.assertEquals("Should bind x correctly", 0, left.ref().fieldId());
+ Assert.assertEquals("Should bind x correctly", 0, left.term().ref().fieldId());
BoundPredicate<?> right = TestHelpers.assertAndUnwrap(and.right());
- Assert.assertEquals("Should bind y correctly", 1, right.ref().fieldId());
+ Assert.assertEquals("Should bind y correctly", 1, right.term().ref().fieldId());
}
@Test
@@ -113,9 +114,9 @@ public class TestExpressionBinding {
// make sure the refs are for the right fields
BoundPredicate<?> left = TestHelpers.assertAndUnwrap(or.left());
- Assert.assertEquals("Should bind z correctly", 2, left.ref().fieldId());
+ Assert.assertEquals("Should bind z correctly", 2, left.term().ref().fieldId());
BoundPredicate<?> right = TestHelpers.assertAndUnwrap(or.right());
- Assert.assertEquals("Should bind y correctly", 1, right.ref().fieldId());
+ Assert.assertEquals("Should bind y correctly", 1, right.term().ref().fieldId());
}
@Test
@@ -129,7 +130,7 @@ public class TestExpressionBinding {
// make sure the refs are for the right fields
BoundPredicate<?> child = TestHelpers.assertAndUnwrap(not.child());
- Assert.assertEquals("Should bind x correctly", 0, child.ref().fieldId());
+ Assert.assertEquals("Should bind x correctly", 0, child.term().ref().fieldId());
}
@Test
@@ -141,7 +142,7 @@ public class TestExpressionBinding {
// make sure the expression is a StartsWith
BoundPredicate<?> pred = TestHelpers.assertAndUnwrap(boundExpr, BoundPredicate.class);
Assert.assertEquals("Should be right operation", Expression.Operation.STARTS_WITH, pred.op());
- Assert.assertEquals("Should bind s correctly", 0, pred.ref().fieldId());
+ Assert.assertEquals("Should bind s correctly", 0, pred.term().ref().fieldId());
}
@Test
@@ -173,6 +174,16 @@ public class TestExpressionBinding {
Expression bound = Binder.bind(STRUCT, not(not(lessThan("y", 100))));
BoundPredicate<?> pred = TestHelpers.assertAndUnwrap(bound);
- Assert.assertEquals("Should have the correct bound field", 1, pred.ref().fieldId());
+ Assert.assertEquals("Should have the correct bound field", 1, pred.term().ref().fieldId());
+ }
+
+ @Test
+ public void testTransformExpressionBinding() {
+ Expression bound = Binder.bind(STRUCT, equal(bucket("x", 16), 10));
+ TestHelpers.assertAllReferencesBound("BoundTransform", bound);
+ BoundPredicate<?> pred = TestHelpers.assertAndUnwrap(bound);
+ Assert.assertTrue("Should use a BoundTransform child", pred.term() instanceof BoundTransform);
+ BoundTransform<?, ?> transformExpr = (BoundTransform<?, ?>) pred.term();
+ Assert.assertEquals("Should use a bucket[16] transform", "bucket[16]", transformExpr.transform().toString());
}
}
diff --git a/api/src/test/java/org/apache/iceberg/expressions/TestExpressionHelpers.java b/api/src/test/java/org/apache/iceberg/expressions/TestExpressionHelpers.java
index 450c9c0..74edb15 100644
--- a/api/src/test/java/org/apache/iceberg/expressions/TestExpressionHelpers.java
+++ b/api/src/test/java/org/apache/iceberg/expressions/TestExpressionHelpers.java
@@ -26,10 +26,16 @@ import org.junit.Test;
import static org.apache.iceberg.expressions.Expressions.alwaysFalse;
import static org.apache.iceberg.expressions.Expressions.alwaysTrue;
import static org.apache.iceberg.expressions.Expressions.and;
+import static org.apache.iceberg.expressions.Expressions.bucket;
+import static org.apache.iceberg.expressions.Expressions.day;
import static org.apache.iceberg.expressions.Expressions.equal;
+import static org.apache.iceberg.expressions.Expressions.hour;
import static org.apache.iceberg.expressions.Expressions.lessThan;
+import static org.apache.iceberg.expressions.Expressions.month;
import static org.apache.iceberg.expressions.Expressions.not;
import static org.apache.iceberg.expressions.Expressions.or;
+import static org.apache.iceberg.expressions.Expressions.truncate;
+import static org.apache.iceberg.expressions.Expressions.year;
public class TestExpressionHelpers {
private final UnboundPredicate<?> pred = lessThan("x", 7);
@@ -72,17 +78,49 @@ public class TestExpressionHelpers {
}
@Test
+ public void testTransformExpressions() {
+ Assert.assertEquals("Should produce the correct expression string",
+ "year(ref(name=\"ts\")) == \"2019\"",
+ equal(year("ts"), "2019").toString());
+ Assert.assertEquals("Should produce the correct expression string",
+ "month(ref(name=\"ts\")) == 1234",
+ equal(month("ts"), 1234).toString());
+ Assert.assertEquals("Should produce the correct expression string",
+ "day(ref(name=\"ts\")) == \"2019-12-04\"",
+ equal(day("ts"), "2019-12-04").toString());
+ Assert.assertEquals("Should produce the correct expression string",
+ "hour(ref(name=\"ts\")) == \"2019-12-04-10\"",
+ equal(hour("ts"), "2019-12-04-10").toString());
+ Assert.assertEquals("Should produce the correct expression string",
+ "truncate[6](ref(name=\"str\")) == \"abcdef\"",
+ equal(truncate("str", 6), "abcdef").toString());
+ Assert.assertEquals("Should produce the correct expression string",
+ "truncate[5](ref(name=\"i\")) == 10",
+ equal(truncate("i", 5), 10).toString());
+ Assert.assertEquals("Should produce the correct expression string",
+ "bucket[16](ref(name=\"id\")) == 12",
+ equal(bucket("id", 16), 12).toString());
+ }
+
+ @Test
public void testNullName() {
AssertHelpers.assertThrows("Should catch null column names when creating expressions",
- NullPointerException.class, "Name cannot be null", () -> equal(null, 5));
+ NullPointerException.class, "Name cannot be null", () -> equal((String) null, 5));
+ }
+
+ @Test
+ public void testNullValueExpr() {
+ AssertHelpers.assertThrows("Should catch null value expressions",
+ NullPointerException.class, "Term cannot be null",
+ () -> equal((UnboundTerm<Integer>) null, 5));
}
@Test
public void testMultiAnd() {
Expression expected = and(
and(
- equal("a", 1),
- equal("b", 2)),
+ equal("a", 1),
+ equal("b", 2)),
equal("c", 3));
Expression actual = and(
diff --git a/api/src/test/java/org/apache/iceberg/expressions/TestExpressionSerialization.java b/api/src/test/java/org/apache/iceberg/expressions/TestExpressionSerialization.java
index 6997636..0f10881 100644
--- a/api/src/test/java/org/apache/iceberg/expressions/TestExpressionSerialization.java
+++ b/api/src/test/java/org/apache/iceberg/expressions/TestExpressionSerialization.java
@@ -99,13 +99,36 @@ public class TestExpressionSerialization {
}
}
+ private static boolean equals(Term left, Term right) {
+ if (left instanceof Reference && right instanceof Reference) {
+ return equals((Reference<?>) left, (Reference<?>) right);
+ } else if (left instanceof UnboundTransform && right instanceof UnboundTransform) {
+ UnboundTransform<?, ?> unboundLeft = (UnboundTransform<?, ?>) left;
+ UnboundTransform<?, ?> unboundRight = (UnboundTransform<?, ?>) right;
+ if (equals(unboundLeft.ref(), unboundRight.ref()) &&
+ unboundLeft.transform().toString().equals(unboundRight.transform().toString())) {
+ return true;
+ }
+
+ } else if (left instanceof BoundTransform && right instanceof BoundTransform) {
+ BoundTransform<?, ?> boundLeft = (BoundTransform<?, ?>) left;
+ BoundTransform<?, ?> boundRight = (BoundTransform<?, ?>) right;
+ if (equals(boundLeft.ref(), boundRight.ref()) &&
+ boundLeft.transform().toString().equals(boundRight.transform().toString())) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
@SuppressWarnings({"unchecked", "checkstyle:CyclomaticComplexity"})
private static boolean equals(Predicate left, Predicate right) {
if (left.op() != right.op()) {
return false;
}
- if (!equals(left.ref(), right.ref())) {
+ if (!equals(left.term(), right.term())) {
return false;
}
@@ -151,7 +174,7 @@ public class TestExpressionSerialization {
return left.containsAll(right);
}
- private static boolean equals(Reference left, Reference right) {
+ private static boolean equals(Reference<?> left, Reference<?> right) {
if (left instanceof NamedReference) {
if (!(right instanceof NamedReference)) {
return false;
diff --git a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
index 8d3d4b4..400354e 100644
--- a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
+++ b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
@@ -21,12 +21,26 @@
package org.apache.iceberg.types;
import org.apache.iceberg.Schema;
+import org.junit.Assert;
import org.junit.Test;
import static org.apache.iceberg.types.Types.NestedField.required;
public class TestTypeUtil {
+ @Test
+ public void testReassignIdsDuplicateColumns() {
+ Schema schema = new Schema(
+ required(0, "a", Types.IntegerType.get()),
+ required(1, "A", Types.IntegerType.get())
+ );
+ Schema sourceSchema = new Schema(
+ required(1, "a", Types.IntegerType.get()),
+ required(2, "A", Types.IntegerType.get())
+ );
+ final Schema actualSchema = TypeUtil.reassignIds(schema, sourceSchema);
+ Assert.assertEquals(sourceSchema.asStruct(), actualSchema.asStruct());
+ }
@Test(expected = IllegalArgumentException.class)
public void testReassignIdsIllegalArgumentException() {
diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java
index ab8e5b8..7fbf917 100644
--- a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java
+++ b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java
@@ -187,8 +187,8 @@ public class VectorizedArrowReader implements VectorizedReader<VectorHolder> {
case UTF8:
case BSON:
this.vec = new IcebergVarcharArrowVector(icebergField.name(), rootAlloc);
+ //TODO: Possibly use the uncompressed page size info to set the initial capacity
vec.setInitialCapacity(batchSize * 10);
- //TODO: samarth use the uncompressed page size info here
vec.allocateNewSafe();
return UNKNOWN_WIDTH;
case INT_8:
@@ -242,8 +242,8 @@ public class VectorizedArrowReader implements VectorizedReader<VectorHolder> {
return len;
case BINARY:
this.vec = new IcebergVarBinaryArrowVector(icebergField.name(), rootAlloc);
+ //TODO: Possibly use the uncompressed page size info to set the initial capacity
vec.setInitialCapacity(batchSize * 10);
- //TODO: samarth use the uncompressed page size info here
vec.allocateNewSafe();
return UNKNOWN_WIDTH;
case INT32:
diff --git a/baseline.gradle b/baseline.gradle
index d0c122a..8547679 100644
--- a/baseline.gradle
+++ b/baseline.gradle
@@ -32,8 +32,10 @@ subprojects {
// Thus we concede to applying all of the Baseline plugins individually on all the projects we are
// ready to enforce linting on.
apply plugin: 'org.inferred.processors'
- apply plugin: 'com.palantir.baseline-checkstyle'
- apply plugin: 'com.palantir.baseline-error-prone'
+ if (!project.hasProperty('quick')) {
+ apply plugin: 'com.palantir.baseline-checkstyle'
+ apply plugin: 'com.palantir.baseline-error-prone'
+ }
apply plugin: 'com.palantir.baseline-scalastyle'
apply plugin: 'com.palantir.baseline-class-uniqueness'
apply plugin: 'com.palantir.baseline-reproducibility'
@@ -58,21 +60,23 @@ subprojects {
}
}
- tasks.withType(JavaCompile).configureEach {
- options.errorprone.errorproneArgs += [
- // error-prone is slow, don't run on tests
- '-XepExcludedPaths:.*/test/.*',
- // specific to Palantir
- '-Xep:PreferSafeLoggingPreconditions:OFF',
- '-Xep:PreferSafeLoggableExceptions:OFF',
- '-Xep:Slf4jLogsafeArgs:OFF',
- // subclasses are not equal
- '-Xep:EqualsGetClass:OFF',
- // patterns that are allowed
- '-Xep:SwitchStatementDefaultCase:OFF',
- '-Xep:MissingCasesInEnumSwitch:OFF',
- '-Xep:TypeParameterShadowing:OFF',
- '-Xep:TypeParameterUnusedInFormals:OFF',
+ pluginManager.withPlugin('com.palantir.baseline-error-prone') {
+ tasks.withType(JavaCompile).configureEach {
+ options.errorprone.errorproneArgs += [
+ // error-prone is slow, don't run on tests
+ '-XepExcludedPaths:.*/test/.*',
+ // specific to Palantir
+ '-Xep:PreferSafeLoggingPreconditions:OFF',
+ '-Xep:PreferSafeLoggableExceptions:OFF',
+ '-Xep:Slf4jLogsafeArgs:OFF',
+ // subclasses are not equal
+ '-Xep:EqualsGetClass:OFF',
+ // patterns that are allowed
+ '-Xep:SwitchStatementDefaultCase:OFF',
+ '-Xep:MissingCasesInEnumSwitch:OFF',
+ '-Xep:TypeParameterShadowing:OFF',
+ '-Xep:TypeParameterUnusedInFormals:OFF',
]
+ }
}
-}
+}
\ No newline at end of file
diff --git a/build.gradle b/build.gradle
index 2e1b031..30bd078 100644
--- a/build.gradle
+++ b/build.gradle
@@ -129,6 +129,7 @@ project(':iceberg-core') {
exclude group: 'org.apache.avro', module: 'avro'
exclude group: 'org.slf4j', module: 'slf4j-log4j12'
}
+
testCompile project(path: ':iceberg-api', configuration: 'testArtifacts')
}
}
@@ -263,6 +264,14 @@ project(':iceberg-spark') {
}
testCompile project(path: ':iceberg-hive', configuration: 'testArtifacts')
testCompile project(path: ':iceberg-api', configuration: 'testArtifacts')
+
+ // spark-avro is required only for JMH tests, so ideally we would add this
+ // dependency only to the jmh configuration, however gradle-consistent-versions
+ // plugin does not respect this configuration and does not seem to have a way
+ // to add custom configurations in its lockable configuration detection
+ compileOnly("org.apache.spark:spark-avro_2.11") {
+ exclude group: 'org.apache.avro', module: 'avro'
+ }
}
}
diff --git a/core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java b/core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java
index 1fbbd33..3b6ba09 100644
--- a/core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java
+++ b/core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java
@@ -45,7 +45,6 @@ public abstract class BaseMetastoreTableOperations implements TableOperations {
public static final String PREVIOUS_METADATA_LOCATION_PROP = "previous_metadata_location";
private static final String METADATA_FOLDER_NAME = "metadata";
- private static final String DATA_FOLDER_NAME = "data";
private TableMetadata currentMetadata = null;
private String currentMetadataLocation = null;
diff --git a/core/src/main/java/org/apache/iceberg/CachingCatalog.java b/core/src/main/java/org/apache/iceberg/CachingCatalog.java
index 4812f63..ac0fb4d 100644
--- a/core/src/main/java/org/apache/iceberg/CachingCatalog.java
+++ b/core/src/main/java/org/apache/iceberg/CachingCatalog.java
@@ -33,7 +33,11 @@ import org.apache.iceberg.exceptions.AlreadyExistsException;
public class CachingCatalog implements Catalog {
public static Catalog wrap(Catalog catalog) {
- return new CachingCatalog(catalog);
+ return wrap(catalog, true);
+ }
+
+ public static Catalog wrap(Catalog catalog, boolean caseSensitive) {
+ return new CachingCatalog(catalog, caseSensitive);
}
private final Cache<TableIdentifier, Table> tableCache = Caffeine.newBuilder()
@@ -41,9 +45,19 @@ public class CachingCatalog implements Catalog {
.expireAfterAccess(1, TimeUnit.MINUTES)
.build();
private final Catalog catalog;
+ private final boolean caseSensitive;
- private CachingCatalog(Catalog catalog) {
+ private CachingCatalog(Catalog catalog, boolean caseSensitive) {
this.catalog = catalog;
+ this.caseSensitive = caseSensitive;
+ }
+
+ private TableIdentifier canonicalizeIdentifier(TableIdentifier tableIdentifier) {
+ if (caseSensitive) {
+ return tableIdentifier;
+ } else {
+ return tableIdentifier.toLowerCase();
+ }
}
@Override
@@ -53,14 +67,14 @@ public class CachingCatalog implements Catalog {
@Override
public Table loadTable(TableIdentifier ident) {
- return tableCache.get(ident, catalog::loadTable);
+ return tableCache.get(canonicalizeIdentifier(ident), catalog::loadTable);
}
@Override
public Table createTable(TableIdentifier ident, Schema schema, PartitionSpec spec, String location,
Map<String, String> properties) {
AtomicBoolean created = new AtomicBoolean(false);
- Table table = tableCache.get(ident, identifier -> {
+ Table table = tableCache.get(canonicalizeIdentifier(ident), identifier -> {
created.set(true);
return catalog.createTable(identifier, schema, spec, location, properties);
});
@@ -88,20 +102,20 @@ public class CachingCatalog implements Catalog {
// when the transaction commits, invalidate the table in the cache if it is present.
return CommitCallbackTransaction.addCallback(
catalog.newReplaceTableTransaction(ident, schema, spec, location, properties, orCreate),
- () -> tableCache.invalidate(ident));
+ () -> tableCache.invalidate(canonicalizeIdentifier(ident)));
}
@Override
public boolean dropTable(TableIdentifier ident, boolean purge) {
boolean dropped = catalog.dropTable(ident, false);
- tableCache.invalidate(ident);
+ tableCache.invalidate(canonicalizeIdentifier(ident));
return dropped;
}
@Override
public void renameTable(TableIdentifier from, TableIdentifier to) {
catalog.renameTable(from, to);
- tableCache.invalidate(from);
+ tableCache.invalidate(canonicalizeIdentifier(from));
}
}
diff --git a/core/src/main/java/org/apache/iceberg/ScanSummary.java b/core/src/main/java/org/apache/iceberg/ScanSummary.java
index 46b4fe8..1cdef1f 100644
--- a/core/src/main/java/org/apache/iceberg/ScanSummary.java
+++ b/core/src/main/java/org/apache/iceberg/ScanSummary.java
@@ -130,13 +130,15 @@ public class ScanSummary {
} else if (expression instanceof UnboundPredicate) {
UnboundPredicate pred = (UnboundPredicate) expression;
- NamedReference ref = (NamedReference) pred.ref();
- Literal<?> lit = pred.literal();
- if (TIMESTAMP_NAMES.contains(ref.name())) {
- Literal<Long> tsLiteral = lit.to(Types.TimestampType.withoutZone());
- long millis = toMillis(tsLiteral.value());
- addTimestampFilter(Expressions.predicate(pred.op(), "timestamp_ms", millis));
- return;
+ if (pred.term() instanceof NamedReference) {
+ NamedReference<?> ref = (NamedReference<?>) pred.term();
+ Literal<?> lit = pred.literal();
+ if (TIMESTAMP_NAMES.contains(ref.name())) {
+ Literal<Long> tsLiteral = lit.to(Types.TimestampType.withoutZone());
+ long millis = toMillis(tsLiteral.value());
+ addTimestampFilter(Expressions.predicate(pred.op(), "timestamp_ms", millis));
+ return;
+ }
}
}
diff --git a/core/src/main/java/org/apache/iceberg/TableProperties.java b/core/src/main/java/org/apache/iceberg/TableProperties.java
index 857d3ab..4db587a 100644
--- a/core/src/main/java/org/apache/iceberg/TableProperties.java
+++ b/core/src/main/java/org/apache/iceberg/TableProperties.java
@@ -59,6 +59,9 @@ public class TableProperties {
public static final String PARQUET_COMPRESSION = "write.parquet.compression-codec";
public static final String PARQUET_COMPRESSION_DEFAULT = "gzip";
+ public static final String PARQUET_COMPRESSION_LEVEL = "write.parquet.compression-level";
+ public static final String PARQUET_COMPRESSION_LEVEL_DEFAULT = null;
+
public static final String AVRO_COMPRESSION = "write.avro.compression-codec";
public static final String AVRO_COMPRESSION_DEFAULT = "gzip";
diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java
index 03108f5..644a627 100644
--- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java
+++ b/core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java
@@ -32,6 +32,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
import org.apache.iceberg.BaseMetastoreCatalog;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -59,6 +60,8 @@ import org.apache.iceberg.exceptions.RuntimeIOException;
*/
public class HadoopCatalog extends BaseMetastoreCatalog implements Closeable {
private static final String ICEBERG_HADOOP_WAREHOUSE_BASE = "iceberg/warehouse";
+ private static final String TABLE_METADATA_FILE_EXTENSION = ".metadata.json";
+ private static final PathFilter TABLE_FILTER = path -> path.getName().endsWith(TABLE_METADATA_FILE_EXTENSION);
private final Configuration conf;
private String warehouseLocation;
@@ -116,7 +119,8 @@ public class HadoopCatalog extends BaseMetastoreCatalog implements Closeable {
}
Path metadataPath = new Path(path, "metadata");
- if (fs.exists(metadataPath) && fs.isDirectory(metadataPath)) {
+ if (fs.exists(metadataPath) && fs.isDirectory(metadataPath) &&
+ (fs.listStatus(metadataPath, TABLE_FILTER).length >= 1)) {
// Only the path which contains metadata is the path for table, otherwise it could be
// still a namespace.
TableIdentifier tblIdent = TableIdentifier.of(namespace, path.getName());
diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java
index a9a9fdb..fbaa75d 100644
--- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java
+++ b/core/src/main/java/org/apache/iceberg/hadoop/HadoopFileIO.java
@@ -27,13 +27,22 @@ import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.util.SerializableSupplier;
public class HadoopFileIO implements FileIO {
- private final SerializableConfiguration hadoopConf;
+ private final SerializableSupplier<Configuration> hadoopConf;
public HadoopFileIO(Configuration hadoopConf) {
- this.hadoopConf = new SerializableConfiguration(hadoopConf);
+ this(new SerializableConfiguration(hadoopConf)::get);
+ }
+
+ public HadoopFileIO(SerializableSupplier<Configuration> hadoopConf) {
+ this.hadoopConf = hadoopConf;
+ }
+
+ public Configuration conf() {
+ return hadoopConf.get();
}
@Override
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Reference.java b/core/src/main/java/org/apache/iceberg/util/SerializableSupplier.java
similarity index 79%
copy from api/src/main/java/org/apache/iceberg/expressions/Reference.java
copy to core/src/main/java/org/apache/iceberg/util/SerializableSupplier.java
index f08867f..9ff3fa3 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Reference.java
+++ b/core/src/main/java/org/apache/iceberg/util/SerializableSupplier.java
@@ -17,14 +17,9 @@
* under the License.
*/
-package org.apache.iceberg.expressions;
+package org.apache.iceberg.util;
import java.io.Serializable;
+import java.util.function.Supplier;
-/**
- * Represents a variable reference in an {@link Expression expression}.
- * @see BoundReference
- * @see NamedReference
- */
-public interface Reference extends Serializable {
-}
+public interface SerializableSupplier<T> extends Supplier<T>, Serializable {}
diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java b/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
index 12bdfbd..f89d06a 100644
--- a/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
@@ -90,8 +90,9 @@ public class TestHadoopCatalog extends HadoopTableTestBase {
TableIdentifier tbl1 = TableIdentifier.of("db", "tbl1");
TableIdentifier tbl2 = TableIdentifier.of("db", "tbl2");
TableIdentifier tbl3 = TableIdentifier.of("db", "ns1", "tbl3");
+ TableIdentifier tbl4 = TableIdentifier.of("db", "metadata", "metadata");
- Lists.newArrayList(tbl1, tbl2, tbl3).forEach(t ->
+ Lists.newArrayList(tbl1, tbl2, tbl3, tbl4).forEach(t ->
catalog.createTable(t, SCHEMA, PartitionSpec.unpartitioned())
);
@@ -107,7 +108,7 @@ public class TestHadoopCatalog extends HadoopTableTestBase {
AssertHelpers.assertThrows("should throw exception", NotFoundException.class,
"Unknown namespace", () -> {
- catalog.listTables(Namespace.of("db", "ns1", "ns2"));
- });
+ catalog.listTables(Namespace.of("db", "ns1", "ns2"));
+ });
}
}
diff --git a/hive/src/main/java/org/apache/iceberg/hive/ClientPool.java b/hive/src/main/java/org/apache/iceberg/hive/ClientPool.java
index bbb7991..c66df74 100644
--- a/hive/src/main/java/org/apache/iceberg/hive/ClientPool.java
+++ b/hive/src/main/java/org/apache/iceberg/hive/ClientPool.java
@@ -35,7 +35,6 @@ public abstract class ClientPool<C, E extends Exception> implements Closeable {
private final Object signal = new Object();
private volatile int currentSize;
private boolean closed;
- private int runs = 0;
ClientPool(int poolSize, Class<? extends E> reconnectExc) {
this.poolSize = poolSize;
@@ -50,7 +49,6 @@ public abstract class ClientPool<C, E extends Exception> implements Closeable {
}
public <R> R run(Action<R, C, E> action) throws E, InterruptedException {
- runs += 1;
C client = get();
try {
return action.run(client);
diff --git a/hive/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java b/hive/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
index 9871203..0eeafc7 100644
--- a/hive/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
+++ b/hive/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java
@@ -177,7 +177,7 @@ public class HiveTableOperations extends BaseMetastoreTableOperations {
throw new AlreadyExistsException("Table already exists: %s.%s", database, tableName);
} catch (TException | UnknownHostException e) {
- if (e.getMessage().contains("Table/View 'HIVE_LOCKS' does not exist")) {
+ if (e.getMessage() != null && e.getMessage().contains("Table/View 'HIVE_LOCKS' does not exist")) {
throw new RuntimeException("Failed to acquire locks from metastore because 'HIVE_LOCKS' doesn't " +
"exist, this probably happened when using embedded metastore or doesn't create a " +
"transactional meta table. To fix this, use an alternative metastore", e);
diff --git a/jmh.gradle b/jmh.gradle
index 732edd2..2d3d933 100644
--- a/jmh.gradle
+++ b/jmh.gradle
@@ -32,7 +32,9 @@ configure(jmhProjects) {
}
jmhCompileGeneratedClasses {
- options.errorprone.enabled = false
+ pluginManager.withPlugin('com.palantir.baseline-error-prone') {
+ options.errorprone.enabled = false
+ }
}
}
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java b/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
index 9ea7713..5d66daa 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
@@ -60,6 +60,8 @@ import org.apache.parquet.schema.MessageType;
import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION;
import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION_DEFAULT;
+import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION_LEVEL;
+import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION_LEVEL_DEFAULT;
import static org.apache.iceberg.TableProperties.PARQUET_DICT_SIZE_BYTES;
import static org.apache.iceberg.TableProperties.PARQUET_DICT_SIZE_BYTES_DEFAULT;
import static org.apache.iceberg.TableProperties.PARQUET_PAGE_SIZE_BYTES;
@@ -184,6 +186,24 @@ public class Parquet {
PARQUET_PAGE_SIZE_BYTES, PARQUET_PAGE_SIZE_BYTES_DEFAULT));
int dictionaryPageSize = Integer.parseInt(config.getOrDefault(
PARQUET_DICT_SIZE_BYTES, PARQUET_DICT_SIZE_BYTES_DEFAULT));
+ String compressionLevel = config.getOrDefault(
+ PARQUET_COMPRESSION_LEVEL, PARQUET_COMPRESSION_LEVEL_DEFAULT);
+
+ if (compressionLevel != null) {
+ switch (codec()) {
+ case GZIP:
+ config.put("zlib.compress.level", compressionLevel);
+ break;
+ case BROTLI:
+ config.put("compression.brotli.quality", compressionLevel);
+ break;
+ case ZSTD:
+ config.put("io.compression.codec.zstd.level", compressionLevel);
+ break;
+ default:
+ // compression level is not supported; ignore it
+ }
+ }
WriterVersion writerVersion = WriterVersion.PARQUET_1_0;
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetFilters.java b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetFilters.java
index ae40da7..7a2f03b 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetFilters.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetFilters.java
@@ -111,8 +111,12 @@ class ParquetFilters {
@Override
public <T> FilterPredicate predicate(BoundPredicate<T> pred) {
+ if (!(pred.term() instanceof BoundReference)) {
+ throw new UnsupportedOperationException("Cannot convert non-reference to Parquet filter: " + pred.term());
+ }
+
Operation op = pred.op();
- BoundReference<T> ref = pred.ref();
+ BoundReference<T> ref = (BoundReference<T>) pred.term();
String path = schema.idToAlias(ref.fieldId());
Literal<T> lit;
if (pred.isUnaryPredicate()) {
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetUtil.java b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetUtil.java
index 2621f5f..f8a21d6 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetUtil.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetUtil.java
@@ -339,11 +339,8 @@ public class ParquetUtil {
encodings.remove(Encoding.RLE);
encodings.remove(Encoding.BIT_PACKED);
- if (encodings.isEmpty()) {
- return false; // no encodings other than dictionary or rep/def levels
- }
-
- return true;
+ // when empty, no encodings other than dictionary or rep/def levels
+ return !encodings.isEmpty();
} else {
// if PLAIN_DICTIONARY wasn't present, then either the column is not
// dictionary-encoded, or the 2.0 encoding, RLE_DICTIONARY, was used.
diff --git a/site/docs/configuration.md b/site/docs/configuration.md
index abec8ea..b0322f4 100644
--- a/site/docs/configuration.md
+++ b/site/docs/configuration.md
@@ -38,11 +38,15 @@ Iceberg tables support table properties to configure table behavior, like the de
| write.parquet.page-size-bytes | 1048576 (1 MB) | Parquet page size |
| write.parquet.dict-size-bytes | 2097152 (2 MB) | Parquet dictionary page size |
| write.parquet.compression-codec | gzip | Parquet compression codec |
+| write.parquet.compression-level | null | Parquet compression level |
| write.avro.compression-codec | gzip | Avro compression codec |
| write.metadata.compression-codec | none | Metadata compression codec; none or gzip |
| write.metadata.metrics.default | truncate(16) | Default metrics mode for all columns in the table; none, counts, truncate(length), or full |
| write.metadata.metrics.column.col1 | (not set) | Metrics mode for column 'col1' to allow per-column tuning; none, counts, truncate(length), or full |
-| write.target-file-size-bytes | Long.MAX_VALUE | Controls the size of files generated to target about this many bytes. |
+| write.target-file-size-bytes | Long.MAX_VALUE | Controls the size of files generated to target about this many bytes |
+| write.wap.enabled | false | Enables write-audit-publish writes |
+| write.metadata.delete-after-commit.enabled | false | Controls whether to delete the oldest version metadata files after commit |
+| write.metadata.previous-versions-max | 100 | The max number of previous version metadata files to keep before deleting after commit |
### Table behavior properties
@@ -54,6 +58,7 @@ Iceberg tables support table properties to configure table behavior, like the de
| commit.retry.total-timeout-ms | 1800000 (30 min) | Maximum time in milliseconds to wait before retrying a commit |
| commit.manifest.target-size-bytes | 8388608 (8 MB) | Target size when merging manifest files |
| commit.manifest.min-count-to-merge | 100 | Minimum number of manifests to accumulate before merging |
+| commit.manifest-merge.enabled | true | Controls whether to automatically merge manifests on writes |
## Spark options
@@ -95,4 +100,3 @@ df.write
| write-format | Table write.format.default | File format to use for this write operation; parquet or avro |
| target-file-size-bytes | As per table property | Overrides this table's write.target-file-size-bytes |
| check-nullability | true | Sets the nullable check on fields |
-
diff --git a/spark/src/jmh/java/org/apache/iceberg/spark/source/avro/IcebergSourceFlatAvroDataReadBenchmark.java b/spark/src/jmh/java/org/apache/iceberg/spark/source/avro/IcebergSourceFlatAvroDataReadBenchmark.java
new file mode 100644
index 0000000..cbf744c
--- /dev/null
+++ b/spark/src/jmh/java/org/apache/iceberg/spark/source/avro/IcebergSourceFlatAvroDataReadBenchmark.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.spark.source.avro;
+
+import com.google.common.collect.Maps;
+import java.io.IOException;
+import java.util.Map;
+import org.apache.iceberg.spark.source.IcebergSourceFlatDataBenchmark;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.internal.SQLConf;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.SPLIT_OPEN_FILE_COST;
+import static org.apache.spark.sql.functions.current_date;
+import static org.apache.spark.sql.functions.date_add;
+import static org.apache.spark.sql.functions.expr;
+
+/**
+ * A benchmark that evaluates the performance of reading Avro data with a flat schema
+ * using Iceberg and the built-in file source in Spark.
+ *
+ * To run this benchmark:
+ * <code>
+ * ./gradlew :iceberg-spark:jmh
+ * -PjmhIncludeRegex=IcebergSourceFlatAvroDataReadBenchmark
+ * -PjmhOutputPath=benchmark/iceberg-source-flat-avro-data-read-benchmark-result.txt
+ * </code>
+ */
+public class IcebergSourceFlatAvroDataReadBenchmark extends IcebergSourceFlatDataBenchmark {
+
+ private static final int NUM_FILES = 10;
+ private static final int NUM_ROWS = 1000000;
+
+ @Setup
+ public void setupBenchmark() {
+ setupSpark();
+ appendData();
+ }
+
+ @TearDown
+ public void tearDownBenchmark() throws IOException {
+ tearDownSpark();
+ cleanupFiles();
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void readIceberg() {
+ Map<String, String> tableProperties = Maps.newHashMap();
+ tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ withTableProperties(tableProperties, () -> {
+ String tableLocation = table().location();
+ Dataset<Row> df = spark().read().format("iceberg").load(tableLocation);
+ materialize(df);
+ });
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void readFileSource() {
+ Map<String, String> conf = Maps.newHashMap();
+ conf.put(SQLConf.FILES_OPEN_COST_IN_BYTES().key(), Integer.toString(128 * 1024 * 1024));
+ withSQLConf(conf, () -> {
+ Dataset<Row> df = spark().read().format("avro").load(dataLocation());
+ materialize(df);
+ });
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void readWithProjectionIceberg() {
+ Map<String, String> tableProperties = Maps.newHashMap();
+ tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ withTableProperties(tableProperties, () -> {
+ String tableLocation = table().location();
+ Dataset<Row> df = spark().read().format("iceberg").load(tableLocation).select("longCol");
+ materialize(df);
+ });
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void readWithProjectionFileSource() {
+ Map<String, String> conf = Maps.newHashMap();
+ conf.put(SQLConf.FILES_OPEN_COST_IN_BYTES().key(), Integer.toString(128 * 1024 * 1024));
+ withSQLConf(conf, () -> {
+ Dataset<Row> df = spark().read().format("avro").load(dataLocation()).select("longCol");
+ materialize(df);
+ });
+ }
+
+ private void appendData() {
+ Map<String, String> tableProperties = Maps.newHashMap();
+ tableProperties.put(DEFAULT_FILE_FORMAT, "avro");
+ withTableProperties(tableProperties, () -> {
+ for (int fileNum = 1; fileNum <= NUM_FILES; fileNum++) {
+ Dataset<Row> df = spark().range(NUM_ROWS)
+ .withColumnRenamed("id", "longCol")
+ .withColumn("intCol", expr("CAST(longCol AS INT)"))
+ .withColumn("floatCol", expr("CAST(longCol AS FLOAT)"))
+ .withColumn("doubleCol", expr("CAST(longCol AS DOUBLE)"))
+ .withColumn("decimalCol", expr("CAST(longCol AS DECIMAL(20, 5))"))
+ .withColumn("dateCol", date_add(current_date(), fileNum))
+ .withColumn("timestampCol", expr("TO_TIMESTAMP(dateCol)"))
+ .withColumn("stringCol", expr("CAST(dateCol AS STRING)"));
+ appendAsFile(df);
+ }
+ });
+ }
+}
diff --git a/spark/src/jmh/java/org/apache/iceberg/spark/source/avro/IcebergSourceNestedAvroDataReadBenchmark.java b/spark/src/jmh/java/org/apache/iceberg/spark/source/avro/IcebergSourceNestedAvroDataReadBenchmark.java
new file mode 100644
index 0000000..0abfac3
--- /dev/null
+++ b/spark/src/jmh/java/org/apache/iceberg/spark/source/avro/IcebergSourceNestedAvroDataReadBenchmark.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.spark.source.avro;
+
+import com.google.common.collect.Maps;
+import java.io.IOException;
+import java.util.Map;
+import org.apache.iceberg.spark.source.IcebergSourceNestedDataBenchmark;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.internal.SQLConf;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.SPLIT_OPEN_FILE_COST;
+import static org.apache.spark.sql.functions.expr;
+import static org.apache.spark.sql.functions.lit;
+import static org.apache.spark.sql.functions.struct;
+
+
+/**
+ * A benchmark that evaluates the performance of reading Avro data with a flat schema
+ * using Iceberg and the built-in file source in Spark.
+ *
+ * To run this benchmark:
+ * <code>
+ * ./gradlew :iceberg-spark:jmh
+ * -PjmhIncludeRegex=IcebergSourceNestedAvroDataReadBenchmark
+ * -PjmhOutputPath=benchmark/iceberg-source-nested-avro-data-read-benchmark-result.txt
+ * </code>
+ */
+public class IcebergSourceNestedAvroDataReadBenchmark extends IcebergSourceNestedDataBenchmark {
+
+ private static final int NUM_FILES = 10;
+ private static final int NUM_ROWS = 1000000;
+
+ @Setup
+ public void setupBenchmark() {
+ setupSpark();
+ appendData();
+ }
+
+ @TearDown
+ public void tearDownBenchmark() throws IOException {
+ tearDownSpark();
+ cleanupFiles();
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void readIceberg() {
+ Map<String, String> tableProperties = Maps.newHashMap();
+ tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ withTableProperties(tableProperties, () -> {
+ String tableLocation = table().location();
+ Dataset<Row> df = spark().read().format("iceberg").load(tableLocation);
+ materialize(df);
+ });
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void readFileSource() {
+ Map<String, String> conf = Maps.newHashMap();
+ conf.put(SQLConf.FILES_OPEN_COST_IN_BYTES().key(), Integer.toString(128 * 1024 * 1024));
+ withSQLConf(conf, () -> {
+ Dataset<Row> df = spark().read().format("avro").load(dataLocation());
+ materialize(df);
+ });
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void readWithProjectionIceberg() {
+ Map<String, String> tableProperties = Maps.newHashMap();
+ tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+ withTableProperties(tableProperties, () -> {
+ String tableLocation = table().location();
+ Dataset<Row> df = spark().read().format("iceberg").load(tableLocation).select("nested.col3");
+ materialize(df);
+ });
+ }
+
+ @Benchmark
+ @Threads(1)
+ public void readWithProjectionFileSource() {
+ Map<String, String> conf = Maps.newHashMap();
+ conf.put(SQLConf.FILES_OPEN_COST_IN_BYTES().key(), Integer.toString(128 * 1024 * 1024));
+ withSQLConf(conf, () -> {
+ Dataset<Row> df = spark().read().format("avro").load(dataLocation()).select("nested.col3");
+ materialize(df);
+ });
+ }
+
+ private void appendData() {
+ Map<String, String> tableProperties = Maps.newHashMap();
+ tableProperties.put(DEFAULT_FILE_FORMAT, "avro");
+ withTableProperties(tableProperties, () -> {
+ for (int fileNum = 1; fileNum <= NUM_FILES; fileNum++) {
+ Dataset<Row> df = spark().range(NUM_ROWS)
+ .withColumn(
+ "nested",
+ struct(
+ expr("CAST(id AS string) AS col1"),
+ expr("CAST(id AS double) AS col2"),
+ lit(fileNum).cast("long").as("col3")
+ ));
+ appendAsFile(df);
+ }
+ });
+ }
+}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/data/SparkParquetWriters.java b/spark/src/main/java/org/apache/iceberg/spark/data/SparkParquetWriters.java
index d4068c3..24b4e8e 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/data/SparkParquetWriters.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/data/SparkParquetWriters.java
@@ -413,11 +413,9 @@ public class SparkParquetWriters {
private final ArrayData keys;
private final ArrayData values;
private final ReusableEntry<K, V> entry;
- private final MapData map;
private int index;
private EntryIterator(MapData map) {
- this.map = map;
size = map.numElements();
keys = map.keyArray();
values = map.valueArray();
diff --git a/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/IcebergArrowColumnVector.java b/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/IcebergArrowColumnVector.java
index ae6e44c..09583d3 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/IcebergArrowColumnVector.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/IcebergArrowColumnVector.java
@@ -732,8 +732,7 @@ public class IcebergArrowColumnVector extends ColumnVector {
this.vector = vector;
}
- //TODO: samarth not sure this is efficient or correct.
- //TODO: samarth refer to decodeDictionaryIds in VectorizedColumnReader
+ //TODO: still need to evaluate if this is the most efficient way
@Override
final Decimal getDecimal(int rowId, int precision, int scale) {
if (isNullAt(rowId)) {
@@ -753,7 +752,7 @@ public class IcebergArrowColumnVector extends ColumnVector {
this.vector = vector;
}
- //TODO: samarth not sure this is efficient or correct
+ //TODO: still need to evaluate if this is the most efficient way
@Override
final Decimal getDecimal(int rowId, int precision, int scale) {
if (isNullAt(rowId)) {
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java
index f97de81..4c95aca 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java
@@ -30,13 +30,18 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.hadoop.HadoopFileIO;
import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.hive.HiveCatalog;
import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.spark.SparkSchemaUtil;
import org.apache.iceberg.transforms.Transform;
import org.apache.iceberg.transforms.UnknownTransform;
import org.apache.iceberg.types.CheckCompatibility;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.execution.streaming.StreamExecution;
@@ -51,10 +56,12 @@ import org.apache.spark.sql.sources.v2.writer.DataSourceWriter;
import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter;
import org.apache.spark.sql.streaming.OutputMode;
import org.apache.spark.sql.types.StructType;
+import org.apache.spark.util.SerializableConfiguration;
public class IcebergSource implements DataSourceV2, ReadSupport, WriteSupport, DataSourceRegister, StreamWriteSupport {
private SparkSession lazySpark = null;
+ private JavaSparkContext lazySparkContext = null;
private Configuration lazyConf = null;
@Override
@@ -73,7 +80,10 @@ public class IcebergSource implements DataSourceV2, ReadSupport, WriteSupport, D
Table table = getTableAndResolveHadoopConfiguration(options, conf);
String caseSensitive = lazySparkSession().conf().get("spark.sql.caseSensitive");
- Reader reader = new Reader(table, Boolean.parseBoolean(caseSensitive), options);
+ Broadcast<FileIO> io = lazySparkContext().broadcast(fileIO(table));
+ Broadcast<EncryptionManager> encryptionManager = lazySparkContext().broadcast(table.encryption());
+
+ Reader reader = new Reader(table, io, encryptionManager, Boolean.parseBoolean(caseSensitive), options);
if (readSchema != null) {
// convert() will fail if readSchema contains fields not in table.schema()
SparkSchemaUtil.convert(table.schema(), readSchema);
@@ -95,7 +105,12 @@ public class IcebergSource implements DataSourceV2, ReadSupport, WriteSupport, D
validatePartitionTransforms(table.spec());
String appId = lazySparkSession().sparkContext().applicationId();
String wapId = lazySparkSession().conf().get("spark.wap.id", null);
- return Optional.of(new Writer(table, options, mode == SaveMode.Overwrite, appId, wapId, dsSchema));
+ boolean replacePartitions = mode == SaveMode.Overwrite;
+
+ Broadcast<FileIO> io = lazySparkContext().broadcast(fileIO(table));
+ Broadcast<EncryptionManager> encryptionManager = lazySparkContext().broadcast(table.encryption());
+
+ return Optional.of(new Writer(table, io, encryptionManager, options, replacePartitions, appId, wapId, dsSchema));
}
@Override
@@ -113,7 +128,11 @@ public class IcebergSource implements DataSourceV2, ReadSupport, WriteSupport, D
// so we fetch it directly from sparkContext to make writes idempotent
String queryId = lazySparkSession().sparkContext().getLocalProperty(StreamExecution.QUERY_ID_KEY());
String appId = lazySparkSession().sparkContext().applicationId();
- return new StreamingWriter(table, options, queryId, mode, appId, dsSchema);
+
+ Broadcast<FileIO> io = lazySparkContext().broadcast(fileIO(table));
+ Broadcast<EncryptionManager> encryptionManager = lazySparkContext().broadcast(table.encryption());
+
+ return new StreamingWriter(table, io, encryptionManager, options, queryId, mode, appId, dsSchema);
}
protected Table findTable(DataSourceOptions options, Configuration conf) {
@@ -137,6 +156,13 @@ public class IcebergSource implements DataSourceV2, ReadSupport, WriteSupport, D
return lazySpark;
}
+ private JavaSparkContext lazySparkContext() {
+ if (lazySparkContext == null) {
+ this.lazySparkContext = new JavaSparkContext(lazySparkSession().sparkContext());
+ }
+ return lazySparkContext;
+ }
+
private Configuration lazyBaseConf() {
if (lazyConf == null) {
this.lazyConf = lazySparkSession().sessionState().newHadoopConf();
@@ -201,4 +227,14 @@ public class IcebergSource implements DataSourceV2, ReadSupport, WriteSupport, D
boolean dataFrameCheckNullability = options.getBoolean("check-nullability", true);
return sparkCheckNullability && dataFrameCheckNullability;
}
+
+ private FileIO fileIO(Table table) {
+ if (table.io() instanceof HadoopFileIO) {
+ // we need to use Spark's SerializableConfiguration to avoid issues with Kryo serialization
+ SerializableConfiguration conf = new SerializableConfiguration(((HadoopFileIO) table.io()).conf());
+ return new HadoopFileIO(conf::value);
+ } else {
+ return table.io();
+ }
+ }
}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/Reader.java b/spark/src/main/java/org/apache/iceberg/spark/source/Reader.java
index 49bfd2a..018b1ac 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/Reader.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/Reader.java
@@ -67,6 +67,7 @@ import org.apache.iceberg.spark.data.SparkParquetReaders;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.ByteBuffers;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.catalyst.expressions.Attribute;
import org.apache.spark.sql.catalyst.expressions.AttributeReference;
@@ -111,8 +112,8 @@ class Reader implements DataSourceReader,
private final Long splitSize;
private final Integer splitLookback;
private final Long splitOpenFileCost;
- private final FileIO fileIo;
- private final EncryptionManager encryptionManager;
+ private final Broadcast<FileIO> io;
+ private final Broadcast<EncryptionManager> encryptionManager;
private final boolean caseSensitive;
private final int numRecordsPerBatch;
private StructType requestedSchema = null;
@@ -125,7 +126,8 @@ class Reader implements DataSourceReader,
private List<CombinedScanTask> tasks = null; // lazy cache of tasks
private Boolean enableBatchedReads = null; // cache variable for enabling batched reads
- Reader(Table table, boolean caseSensitive, DataSourceOptions options) {
+ Reader(Table table, Broadcast<FileIO> io, Broadcast<EncryptionManager> encryptionManager,
+ boolean caseSensitive, DataSourceOptions options) {
this.table = table;
this.snapshotId = options.get("snapshot-id").map(Long::parseLong).orElse(null);
this.asOfTimestamp = options.get("as-of-timestamp").map(Long::parseLong).orElse(null);
@@ -156,8 +158,8 @@ class Reader implements DataSourceReader,
this.splitOpenFileCost = options.get("file-open-cost").map(Long::parseLong).orElse(null);
this.schema = table.schema();
- this.fileIo = table.io();
- this.encryptionManager = table.encryption();
+ this.io = io;
+ this.encryptionManager = encryptionManager;
this.caseSensitive = caseSensitive;
}
@@ -197,7 +199,7 @@ class Reader implements DataSourceReader,
for (CombinedScanTask task : tasks()) {
readTasks.add(
new VectorizedReading.ReadTask(task, tableSchemaString, expectedSchemaString,
- fileIo, encryptionManager, caseSensitive, numRecordsPerBatch));
+ io.value(), encryptionManager.value(), caseSensitive, numRecordsPerBatch));
}
LOG.info("=> Batching input partitions with {} tasks.", readTasks.size());
@@ -216,7 +218,7 @@ class Reader implements DataSourceReader,
List<InputPartition<InternalRow>> readTasks = Lists.newArrayList();
for (CombinedScanTask task : tasks()) {
readTasks.add(
- new ReadTask(task, tableSchemaString, expectedSchemaString, fileIo, encryptionManager, caseSensitive));
+ new ReadTask(task, tableSchemaString, expectedSchemaString, io, encryptionManager, caseSensitive));
}
return readTasks;
@@ -372,28 +374,28 @@ class Reader implements DataSourceReader,
private final CombinedScanTask task;
private final String tableSchemaString;
private final String expectedSchemaString;
- private final FileIO fileIo;
- private final EncryptionManager encryptionManager;
+ private final Broadcast<FileIO> io;
+ private final Broadcast<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, boolean caseSensitive) {
+ private ReadTask(CombinedScanTask task, String tableSchemaString, String expectedSchemaString,
+ Broadcast<FileIO> io, Broadcast<EncryptionManager> encryptionManager,
+ boolean caseSensitive) {
this.task = task;
this.tableSchemaString = tableSchemaString;
this.expectedSchemaString = expectedSchemaString;
- this.fileIo = fileIo;
+ this.io = io;
this.encryptionManager = encryptionManager;
this.caseSensitive = caseSensitive;
}
@Override
public InputPartitionReader<InternalRow> createPartitionReader() {
- return new TaskDataReader(task, lazyTableSchema(), lazyExpectedSchema(), fileIo,
- encryptionManager, caseSensitive);
+ return new TaskDataReader(task, lazyTableSchema(), lazyExpectedSchema(), io.value(),
+ encryptionManager.value(), caseSensitive);
}
private Schema lazyTableSchema() {
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/StreamingWriter.java b/spark/src/main/java/org/apache/iceberg/spark/source/StreamingWriter.java
index fa3cbd0..9a3fd63 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/StreamingWriter.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/StreamingWriter.java
@@ -27,7 +27,10 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotUpdate;
import org.apache.iceberg.Table;
+import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.FileIO;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.sources.v2.DataSourceOptions;
import org.apache.spark.sql.sources.v2.writer.WriterCommitMessage;
import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter;
@@ -44,9 +47,10 @@ public class StreamingWriter extends Writer implements StreamWriter {
private final String queryId;
private final OutputMode mode;
- StreamingWriter(Table table, DataSourceOptions options, String queryId, OutputMode mode, String applicationId,
- Schema dsSchema) {
- super(table, options, false, applicationId, dsSchema);
+ StreamingWriter(Table table, Broadcast<FileIO> io, Broadcast<EncryptionManager> encryptionManager,
+ DataSourceOptions options, String queryId, OutputMode mode, String applicationId,
+ Schema dsSchema) {
+ super(table, io, encryptionManager, options, false, applicationId, dsSchema);
this.queryId = queryId;
this.mode = mode;
}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/source/Writer.java b/spark/src/main/java/org/apache/iceberg/spark/source/Writer.java
index f20bffc..d067e50 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/source/Writer.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/source/Writer.java
@@ -56,6 +56,7 @@ import org.apache.iceberg.spark.data.SparkAvroWriter;
import org.apache.iceberg.spark.data.SparkParquetWriters;
import org.apache.iceberg.util.PropertyUtil;
import org.apache.iceberg.util.Tasks;
+import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.sources.v2.DataSourceOptions;
import org.apache.spark.sql.sources.v2.writer.DataSourceWriter;
@@ -84,24 +85,26 @@ class Writer implements DataSourceWriter {
private final Table table;
private final FileFormat format;
- private final FileIO fileIo;
- private final EncryptionManager encryptionManager;
+ private final Broadcast<FileIO> io;
+ private final Broadcast<EncryptionManager> encryptionManager;
private final boolean replacePartitions;
private final String applicationId;
private final String wapId;
private final long targetFileSize;
private final Schema dsSchema;
- Writer(Table table, DataSourceOptions options, boolean replacePartitions, String applicationId, Schema dsSchema) {
- this(table, options, replacePartitions, applicationId, null, dsSchema);
+ Writer(Table table, Broadcast<FileIO> io, Broadcast<EncryptionManager> encryptionManager,
+ DataSourceOptions options, boolean replacePartitions, String applicationId, Schema dsSchema) {
+ this(table, io, encryptionManager, options, replacePartitions, applicationId, null, dsSchema);
}
- Writer(Table table, DataSourceOptions options, boolean replacePartitions, String applicationId, String wapId,
- Schema dsSchema) {
+ Writer(Table table, Broadcast<FileIO> io, Broadcast<EncryptionManager> encryptionManager,
+ DataSourceOptions options, boolean replacePartitions, String applicationId, String wapId,
+ Schema dsSchema) {
this.table = table;
this.format = getFileFormat(table.properties(), options);
- this.fileIo = table.io();
- this.encryptionManager = table.encryption();
+ this.io = io;
+ this.encryptionManager = encryptionManager;
this.replacePartitions = replacePartitions;
this.applicationId = applicationId;
this.wapId = wapId;
@@ -127,7 +130,7 @@ class Writer implements DataSourceWriter {
@Override
public DataWriterFactory<InternalRow> createWriterFactory() {
return new WriterFactory(
- table.spec(), format, table.locationProvider(), table.properties(), fileIo, encryptionManager, targetFileSize,
+ table.spec(), format, table.locationProvider(), table.properties(), io, encryptionManager, targetFileSize,
dsSchema);
}
@@ -194,7 +197,7 @@ class Writer implements DataSourceWriter {
2.0 /* exponential */)
.throwFailureWhenFinished()
.run(file -> {
- fileIo.deleteFile(file.path().toString());
+ io.value().deleteFile(file.path().toString());
});
}
@@ -251,19 +254,20 @@ class Writer implements DataSourceWriter {
private final FileFormat format;
private final LocationProvider locations;
private final Map<String, String> properties;
- private final FileIO fileIo;
- private final EncryptionManager encryptionManager;
+ private final Broadcast<FileIO> io;
+ private final Broadcast<EncryptionManager> encryptionManager;
private final long targetFileSize;
private final Schema dsSchema;
WriterFactory(PartitionSpec spec, FileFormat format, LocationProvider locations,
- Map<String, String> properties, FileIO fileIo, EncryptionManager encryptionManager,
- long targetFileSize, Schema dsSchema) {
+ Map<String, String> properties, Broadcast<FileIO> io,
+ Broadcast<EncryptionManager> encryptionManager, long targetFileSize,
+ Schema dsSchema) {
this.spec = spec;
this.format = format;
this.locations = locations;
this.properties = properties;
- this.fileIo = fileIo;
+ this.io = io;
this.encryptionManager = encryptionManager;
this.targetFileSize = targetFileSize;
this.dsSchema = dsSchema;
@@ -275,9 +279,9 @@ class Writer implements DataSourceWriter {
AppenderFactory<InternalRow> appenderFactory = new SparkAppenderFactory();
if (spec.fields().isEmpty()) {
- return new UnpartitionedWriter(spec, format, appenderFactory, fileFactory, fileIo, targetFileSize);
+ return new UnpartitionedWriter(spec, format, appenderFactory, fileFactory, io.value(), targetFileSize);
} else {
- return new PartitionedWriter(spec, format, appenderFactory, fileFactory, fileIo, targetFileSize);
+ return new PartitionedWriter(spec, format, appenderFactory, fileFactory, io.value(), targetFileSize);
}
}
@@ -338,16 +342,17 @@ class Writer implements DataSourceWriter {
* Generates EncryptedOutputFile for UnpartitionedWriter.
*/
public EncryptedOutputFile newOutputFile() {
- OutputFile file = fileIo.newOutputFile(locations.newDataLocation(generateFilename()));
- return encryptionManager.encrypt(file);
+ OutputFile file = io.value().newOutputFile(locations.newDataLocation(generateFilename()));
+ return encryptionManager.value().encrypt(file);
}
/**
* Generates EncryptedOutputFile for PartitionedWriter.
*/
public EncryptedOutputFile newOutputFile(PartitionKey key) {
- OutputFile rawOutputFile = fileIo.newOutputFile(locations.newDataLocation(spec, key, generateFilename()));
- return encryptionManager.encrypt(rawOutputFile);
+ String newDataLocation = locations.newDataLocation(spec, key, generateFilename());
+ OutputFile rawOutputFile = io.value().newOutputFile(newDataLocation);
+ return encryptionManager.value().encrypt(rawOutputFile);
}
}
}
@@ -458,8 +463,6 @@ class Writer implements DataSourceWriter {
}
private static class UnpartitionedWriter extends BaseWriter {
- private static final int ROWS_DIVISOR = 1000;
-
UnpartitionedWriter(
PartitionSpec spec,
FileFormat format,
diff --git a/versions.lock b/versions.lock
index 8f04955..62ba95c 100644
--- a/versions.lock
+++ b/versions.lock
@@ -159,6 +159,7 @@ org.apache.parquet:parquet-format:2.4.0 (3 constraints: e72a97ca)
org.apache.parquet:parquet-hadoop:1.10.1 (2 constraints: de1ac5b3)
org.apache.parquet:parquet-jackson:1.10.1 (1 constraints: b70ee763)
org.apache.pig:pig:0.14.0 (1 constraints: 37052f3b)
+org.apache.spark:spark-avro_2.11:2.4.4 (1 constraints: 0c050536)
org.apache.spark:spark-catalyst_2.11:2.4.4 (1 constraints: c20cc327)
org.apache.spark:spark-core_2.11:2.4.4 (3 constraints: b528ed99)
org.apache.spark:spark-hive_2.11:2.4.4 (1 constraints: 0c050536)
@@ -168,7 +169,7 @@ org.apache.spark:spark-network-common_2.11:2.4.4 (2 constraints: b01eeee2)
org.apache.spark:spark-network-shuffle_2.11:2.4.4 (1 constraints: 1b0dcc36)
org.apache.spark:spark-sketch_2.11:2.4.4 (2 constraints: 981bd4f5)
org.apache.spark:spark-sql_2.11:2.4.4 (1 constraints: 1e0d0037)
-org.apache.spark:spark-tags_2.11:2.4.4 (7 constraints: da61cceb)
+org.apache.spark:spark-tags_2.11:2.4.4 (8 constraints: 036fa69d)
org.apache.spark:spark-unsafe_2.11:2.4.4 (2 constraints: f11bc213)
org.apache.thrift:libfb303:0.9.3 (3 constraints: 6725fac0)
org.apache.thrift:libthrift:0.9.3 (5 constraints: 71415452)
@@ -229,7 +230,7 @@ org.slf4j:slf4j-api:1.7.25 (47 constraints: 26a3ad5d)
org.sonatype.sisu.inject:cglib:2.2.1-v20090111 (1 constraints: aa0cfd36)
org.spark-project.hive:hive-exec:1.2.1.spark2 (1 constraints: 990fa09c)
org.spark-project.hive:hive-metastore:1.2.1.spark2 (1 constraints: 990fa09c)
-org.spark-project.spark:unused:1.0.0 (11 constraints: 7a9e9eab)
+org.spark-project.spark:unused:1.0.0 (12 constraints: 9aab75cf)
org.tukaani:xz:1.5 (1 constraints: f008458a)
org.xerial.snappy:snappy-java:1.1.7.3 (3 constraints: 1126203f)
oro:oro:2.0.8 (3 constraints: 3b229337)
diff --git a/versions.props b/versions.props
index ce7e60c..6f8723c 100644
--- a/versions.props
+++ b/versions.props
@@ -7,6 +7,7 @@ org.apache.hive:hive-standalone-metastore = 1.2.1
org.apache.orc:orc-core = 1.5.6
org.apache.parquet:parquet-avro = 1.10.1
org.apache.spark:spark-hive_2.11 = 2.4.4
+org.apache.spark:spark-avro_2.11 = 2.4.4
org.apache.pig:pig = 0.14.0
org.apache.commons:commons-lang3 = 3.9
com.fasterxml.jackson.*:* = 2.10.0