You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2018/02/05 14:58:29 UTC
[2/3] flink git commit: [FLINK-8242] [orc] Fix predicate translation
if literal is not Serializable.
[FLINK-8242] [orc] Fix predicate translation if literal is not Serializable.
This closes #5345.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/19fcd5eb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/19fcd5eb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/19fcd5eb
Branch: refs/heads/release-1.4
Commit: 19fcd5ebadabdebc4d56716920937f229f06f5d3
Parents: e5c1261
Author: Fabian Hueske <fh...@apache.org>
Authored: Tue Dec 12 22:38:51 2017 +0100
Committer: Fabian Hueske <fh...@apache.org>
Committed: Mon Feb 5 14:18:16 2018 +0100
----------------------------------------------------------------------
.../org/apache/flink/orc/OrcTableSource.java | 29 +++++++++++++++++++-
.../apache/flink/orc/OrcTableSourceTest.java | 17 +++++++++---
2 files changed, 41 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/19fcd5eb/flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcTableSource.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcTableSource.java b/flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcTableSource.java
index 6be6e49..78f3940 100644
--- a/flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcTableSource.java
+++ b/flink-connectors/flink-orc/src/main/java/org/apache/flink/orc/OrcTableSource.java
@@ -52,6 +52,8 @@ import org.apache.flink.util.Preconditions;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
import org.apache.orc.TypeDescription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.Serializable;
import java.util.ArrayList;
@@ -80,6 +82,8 @@ import java.util.List;
public class OrcTableSource
implements BatchTableSource<Row>, ProjectableTableSource<Row>, FilterableTableSource<Row> {
+ private static final Logger LOG = LoggerFactory.getLogger(OrcTableSource.class);
+
private static final int DEFAULT_BATCH_SIZE = 1000;
// path to read ORC files from
@@ -192,7 +196,10 @@ public class OrcTableSource
for (Expression pred : predicates) {
Predicate orcPred = toOrcPredicate(pred);
if (orcPred != null) {
+ LOG.info("Predicate [{}] converted into OrcPredicate [{}] and pushed into OrcTableSource for path {}.", pred, orcPred, path);
orcPredicates.add(orcPred);
+ } else {
+ LOG.info("Predicate [{}] could not be pushed into OrcTableSource for path {}.", pred, path);
}
}
@@ -241,17 +248,32 @@ public class OrcTableSource
if (!isValid(binComp)) {
// not a valid predicate
+ LOG.debug("Unsupported predicate [{}] cannot be pushed into OrcTableSource.", pred);
return null;
}
PredicateLeaf.Type litType = getLiteralType(binComp);
if (litType == null) {
// unsupported literal type
+ LOG.debug("Unsupported predicate [{}] cannot be pushed into OrcTableSource.", pred);
return null;
}
boolean literalOnRight = literalOnRight(binComp);
String colName = getColumnName(binComp);
- Serializable literal = (Serializable) getLiteral(binComp);
+
+ // fetch literal and ensure it is serializable
+ Object literalObj = getLiteral(binComp);
+ Serializable literal;
+ // validate that literal is serializable
+ if (literalObj instanceof Serializable) {
+ literal = (Serializable) literalObj;
+ } else {
+ LOG.warn("Encountered a non-serializable literal of type {}. " +
+ "Cannot push predicate [{}] into OrcTableSource. " +
+ "This is a bug and should be reported.",
+ literalObj.getClass().getCanonicalName(), pred);
+ return null;
+ }
if (pred instanceof EqualTo) {
return new OrcRowInputFormat.Equals(colName, litType, literal);
@@ -288,6 +310,7 @@ public class OrcTableSource
}
} else {
// unsupported predicate
+ LOG.debug("Unsupported predicate [{}] cannot be pushed into OrcTableSource.", pred);
return null;
}
} else if (pred instanceof UnaryExpression) {
@@ -295,11 +318,13 @@ public class OrcTableSource
UnaryExpression unary = (UnaryExpression) pred;
if (!isValid(unary)) {
// not a valid predicate
+ LOG.debug("Unsupported predicate [{}] cannot be pushed into OrcTableSource.", pred);
return null;
}
PredicateLeaf.Type colType = toOrcType(((UnaryExpression) pred).child().resultType());
if (colType == null) {
// unsupported type
+ LOG.debug("Unsupported predicate [{}] cannot be pushed into OrcTableSource.", pred);
return null;
}
@@ -312,10 +337,12 @@ public class OrcTableSource
new OrcRowInputFormat.IsNull(colName, colType));
} else {
// unsupported predicate
+ LOG.debug("Unsupported predicate [{}] cannot be pushed into OrcTableSource.", pred);
return null;
}
} else {
// unsupported predicate
+ LOG.debug("Unsupported predicate [{}] cannot be pushed into OrcTableSource.", pred);
return null;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/19fcd5eb/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceTest.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceTest.java b/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceTest.java
index 4006110..8413cf4 100644
--- a/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceTest.java
+++ b/flink-connectors/flink-orc/src/test/java/org/apache/flink/orc/OrcTableSourceTest.java
@@ -170,14 +170,15 @@ public class OrcTableSourceTest {
.forOrcSchema(TEST_SCHEMA_NESTED)
.build();
- // expressions for predicates
+ // expressions for supported predicates
Expression pred1 = new GreaterThan(
new ResolvedFieldReference("int1", Types.INT),
new Literal(100, Types.INT));
Expression pred2 = new EqualTo(
new ResolvedFieldReference("string1", Types.STRING),
new Literal("hello", Types.STRING));
- Expression pred3 = new EqualTo(
+ // unsupported predicate
+ Expression unsupportedPred = new EqualTo(
new GetCompositeField(
new ArrayElementAt(
new ResolvedFieldReference(
@@ -188,10 +189,18 @@ public class OrcTableSourceTest {
"int1"),
new Literal(1, Types.INT)
);
+ // invalid predicate
+ Expression invalidPred = new EqualTo(
+ new ResolvedFieldReference("long1", Types.LONG),
+ // some invalid, non-serializable literal (here an object of this test class)
+ new Literal(new OrcTableSourceTest(), Types.LONG)
+ );
+
ArrayList<Expression> preds = new ArrayList<>();
preds.add(pred1);
preds.add(pred2);
- preds.add(pred3);
+ preds.add(unsupportedPred);
+ preds.add(invalidPred);
// apply predicates on TableSource
OrcTableSource projected = (OrcTableSource) orc.applyPredicate(preds);
@@ -207,7 +216,7 @@ public class OrcTableSourceTest {
Types.ROW_NAMED(getNestedFieldNames(), getNestedFieldTypes()),
projected.getReturnType());
- // ensure IF is configured with supported predicates
+ // ensure IF is configured with valid/supported predicates
OrcTableSource spyTS = spy(projected);
OrcRowInputFormat mockIF = mock(OrcRowInputFormat.class);
doReturn(mockIF).when(spyTS).buildOrcInputFormat();