This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new f32ce6e9 [FLINK-30288] Use PredicateVisitor to convert predicate for
orc
f32ce6e9 is described below
commit f32ce6e99f9ced265abbd57ce7af4c2e184c20cf
Author: shammon <zjur...@gmail.com>
AuthorDate: Wed Dec 7 11:32:11 2022 +0800
[FLINK-30288] Use PredicateVisitor to convert predicate for orc
This closes #419
---
.../table/store/format/orc/OrcFileFormat.java | 7 +-
.../table/store/format/orc/OrcFilterConverter.java | 246 ---------------------
.../format/orc/OrcPredicateFunctionVisitor.java | 215 ++++++++++++++++++
.../store/format/orc/OrcFilterConverterTest.java | 4 +-
4 files changed, 220 insertions(+), 252 deletions(-)
diff --git
a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java
b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java
index 663260ab..2c68889f 100644
---
a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java
+++
b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFileFormat.java
@@ -87,10 +87,9 @@ public class OrcFileFormat extends FileFormat {
if (filters != null) {
for (Predicate pred : filters) {
- OrcFilters.Predicate orcPred =
OrcFilterConverter.toOrcPredicate(pred);
- if (orcPred != null) {
- orcPredicates.add(orcPred);
- }
+ Optional<OrcFilters.Predicate> orcPred =
+ pred.visit(OrcPredicateFunctionVisitor.VISITOR);
+ orcPred.ifPresent(orcPredicates::add);
}
}
diff --git
a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFilterConverter.java
b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFilterConverter.java
deleted file mode 100644
index acccd161..00000000
---
a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcFilterConverter.java
+++ /dev/null
@@ -1,246 +0,0 @@
-/*
- * 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.flink.table.store.format.orc;
-
-import org.apache.flink.orc.OrcFilters;
-import org.apache.flink.orc.OrcFilters.Predicate;
-import org.apache.flink.table.data.DecimalData;
-import org.apache.flink.table.data.TimestampData;
-import org.apache.flink.table.store.file.predicate.CompoundPredicate;
-import org.apache.flink.table.store.file.predicate.Equal;
-import org.apache.flink.table.store.file.predicate.GreaterOrEqual;
-import org.apache.flink.table.store.file.predicate.GreaterThan;
-import org.apache.flink.table.store.file.predicate.IsNotNull;
-import org.apache.flink.table.store.file.predicate.IsNull;
-import org.apache.flink.table.store.file.predicate.LeafFunction;
-import org.apache.flink.table.store.file.predicate.LeafPredicate;
-import org.apache.flink.table.store.file.predicate.LessOrEqual;
-import org.apache.flink.table.store.file.predicate.LessThan;
-import org.apache.flink.table.store.file.predicate.NotEqual;
-import org.apache.flink.table.store.file.predicate.Or;
-import org.apache.flink.table.types.logical.LogicalType;
-import org.apache.flink.util.function.TriFunction;
-
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap;
-
-import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
-
-import javax.annotation.Nullable;
-
-import java.io.Serializable;
-import java.sql.Date;
-import java.time.LocalDate;
-import java.util.function.Function;
-
-/** Utility class that provides helper methods to work with Orc Filter
PushDown. */
-public class OrcFilterConverter {
-
- private static final ImmutableMap<
- Class<? extends LeafFunction>, Function<LeafPredicate,
Predicate>>
- FILTERS =
- new ImmutableMap.Builder<
- Class<? extends LeafFunction>,
- Function<LeafPredicate, Predicate>>()
- .put(IsNull.class,
OrcFilterConverter::convertIsNull)
- .put(IsNotNull.class,
OrcFilterConverter::convertIsNotNull)
- .put(
- Equal.class,
- call -> convertBinary(call,
OrcFilterConverter::convertEquals))
- .put(
- NotEqual.class,
- call ->
- convertBinary(
- call,
OrcFilterConverter::convertNotEquals))
- .put(
- GreaterThan.class,
- call ->
- convertBinary(
- call,
OrcFilterConverter::convertGreaterThan))
- .put(
- GreaterOrEqual.class,
- call ->
- convertBinary(
- call,
-
OrcFilterConverter::convertGreaterThanEquals))
- .put(
- LessThan.class,
- call ->
- convertBinary(
- call,
OrcFilterConverter::convertLessThan))
- .put(
- LessOrEqual.class,
- call ->
- convertBinary(
- call,
-
OrcFilterConverter::convertLessThanEquals))
- .build();
-
- private static Predicate convertIsNull(LeafPredicate predicate) {
- PredicateLeaf.Type colType = toOrcType(predicate.type());
- if (colType == null) {
- return null;
- }
-
- return new OrcFilters.IsNull(predicate.fieldName(), colType);
- }
-
- private static Predicate convertIsNotNull(LeafPredicate predicate) {
- Predicate isNull = convertIsNull(predicate);
- if (isNull == null) {
- return null;
- }
- return new OrcFilters.Not(isNull);
- }
-
- private static Predicate convertOr(CompoundPredicate or) {
- if (or.children().size() != 2) {
- throw new RuntimeException("Illegal or children: " +
or.children().size());
- }
-
- Predicate c1 = toOrcPredicate(or.children().get(0));
- if (c1 == null) {
- return null;
- }
- Predicate c2 = toOrcPredicate(or.children().get(1));
- if (c2 == null) {
- return null;
- }
-
- return new OrcFilters.Or(c1, c2);
- }
-
- private static Predicate convertBinary(
- LeafPredicate predicate,
- TriFunction<String, PredicateLeaf.Type, Serializable, Predicate>
func) {
- PredicateLeaf.Type litType = toOrcType(predicate.type());
- if (litType == null) {
- return null;
- }
-
- String colName = predicate.fieldName();
-
- // fetch literal and ensure it is serializable
- Object orcObj = toOrcObject(litType, predicate.literals().get(0));
- Serializable literal;
- // validate that literal is serializable
- if (orcObj instanceof Serializable) {
- literal = (Serializable) orcObj;
- } else {
- return null;
- }
-
- return func.apply(colName, litType, literal);
- }
-
- private static Predicate convertEquals(
- String colName, PredicateLeaf.Type litType, Serializable literal) {
- return new OrcFilters.Equals(colName, litType, literal);
- }
-
- private static Predicate convertNotEquals(
- String colName, PredicateLeaf.Type litType, Serializable literal) {
- return new OrcFilters.Not(convertEquals(colName, litType, literal));
- }
-
- private static Predicate convertGreaterThan(
- String colName, PredicateLeaf.Type litType, Serializable literal) {
- return new OrcFilters.Not(new OrcFilters.LessThanEquals(colName,
litType, literal));
- }
-
- private static Predicate convertGreaterThanEquals(
- String colName, PredicateLeaf.Type litType, Serializable literal) {
- return new OrcFilters.Not(new OrcFilters.LessThan(colName, litType,
literal));
- }
-
- private static Predicate convertLessThan(
- String colName, PredicateLeaf.Type litType, Serializable literal) {
- return new OrcFilters.LessThan(colName, litType, literal);
- }
-
- private static Predicate convertLessThanEquals(
- String colName, PredicateLeaf.Type litType, Serializable literal) {
- return new OrcFilters.LessThanEquals(colName, litType, literal);
- }
-
- @Nullable
- public static Predicate toOrcPredicate(
- org.apache.flink.table.store.file.predicate.Predicate expression) {
- if (expression instanceof CompoundPredicate) {
- CompoundPredicate compound = (CompoundPredicate) expression;
- if (compound.function().equals(Or.INSTANCE)) {
- return convertOr(compound);
- }
- } else if (expression instanceof LeafPredicate) {
- LeafPredicate leaf = (LeafPredicate) expression;
- Function<LeafPredicate, Predicate> function =
FILTERS.get(leaf.function().getClass());
- if (function == null) {
- return null;
- }
- return function.apply(leaf);
- }
- return null;
- }
-
- @Nullable
- private static Object toOrcObject(PredicateLeaf.Type litType, Object
literalObj) {
- if (literalObj == null) {
- return null;
- }
-
- switch (litType) {
- case STRING:
- return literalObj.toString();
- case DECIMAL:
- return ((DecimalData) literalObj).toBigDecimal();
- case DATE:
- return Date.valueOf(LocalDate.ofEpochDay(((Number)
literalObj).longValue()));
- case TIMESTAMP:
- return ((TimestampData) literalObj).toTimestamp();
- default:
- return literalObj;
- }
- }
-
- @Nullable
- private static PredicateLeaf.Type toOrcType(LogicalType type) {
- switch (type.getTypeRoot()) {
- case TINYINT:
- case SMALLINT:
- case INTEGER:
- case BIGINT:
- return PredicateLeaf.Type.LONG;
- case FLOAT:
- case DOUBLE:
- return PredicateLeaf.Type.FLOAT;
- case BOOLEAN:
- return PredicateLeaf.Type.BOOLEAN;
- case CHAR:
- case VARCHAR:
- return PredicateLeaf.Type.STRING;
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- return PredicateLeaf.Type.TIMESTAMP;
- case DATE:
- return PredicateLeaf.Type.DATE;
- case DECIMAL:
- return PredicateLeaf.Type.DECIMAL;
- default:
- return null;
- }
- }
-}
diff --git
a/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcPredicateFunctionVisitor.java
b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcPredicateFunctionVisitor.java
new file mode 100644
index 00000000..8e93a03c
--- /dev/null
+++
b/flink-table-store-format/src/main/java/org/apache/flink/table/store/format/orc/OrcPredicateFunctionVisitor.java
@@ -0,0 +1,215 @@
+/*
+ * 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.flink.table.store.format.orc;
+
+import org.apache.flink.orc.OrcFilters;
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.store.file.predicate.FieldRef;
+import org.apache.flink.table.store.file.predicate.FunctionVisitor;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.util.function.TriFunction;
+
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.sql.Date;
+import java.time.LocalDate;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * Convert {@link org.apache.flink.table.store.file.predicate.Predicate} to
{@link
+ * OrcFilters.Predicate} for orc.
+ */
+public class OrcPredicateFunctionVisitor
+ implements FunctionVisitor<Optional<OrcFilters.Predicate>> {
+ public static final OrcPredicateFunctionVisitor VISITOR = new
OrcPredicateFunctionVisitor();
+
+ private OrcPredicateFunctionVisitor() {}
+
+ @Override
+ public Optional<OrcFilters.Predicate> visitIsNull(FieldRef fieldRef) {
+ PredicateLeaf.Type colType = toOrcType(fieldRef.type());
+ if (colType == null) {
+ return Optional.empty();
+ }
+
+ return Optional.of(new OrcFilters.IsNull(fieldRef.name(), colType));
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate> visitIsNotNull(FieldRef fieldRef) {
+ Optional<OrcFilters.Predicate> isNull = visitIsNull(fieldRef);
+ return isNull.map(OrcFilters.Not::new);
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate> visitStartsWith(FieldRef fieldRef,
Object literal) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate> visitLessThan(FieldRef fieldRef,
Object literal) {
+ return convertBinary(fieldRef, literal, OrcFilters.LessThan::new);
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate> visitGreaterOrEqual(FieldRef
fieldRef, Object literal) {
+ return convertBinary(
+ fieldRef,
+ literal,
+ (colName, litType, serializableLiteral) ->
+ new OrcFilters.Not(
+ new OrcFilters.LessThan(colName, litType,
serializableLiteral)));
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate> visitNotEqual(FieldRef fieldRef,
Object literal) {
+ return convertBinary(
+ fieldRef,
+ literal,
+ (colName, litType, serializableLiteral) ->
+ new OrcFilters.Not(
+ new OrcFilters.Equals(colName, litType,
serializableLiteral)));
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate> visitLessOrEqual(FieldRef fieldRef,
Object literal) {
+ return convertBinary(fieldRef, literal,
OrcFilters.LessThanEquals::new);
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate> visitEqual(FieldRef fieldRef, Object
literal) {
+ return convertBinary(fieldRef, literal, OrcFilters.Equals::new);
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate> visitGreaterThan(FieldRef fieldRef,
Object literal) {
+ return convertBinary(
+ fieldRef,
+ literal,
+ (colName, litType, serializableLiteral) ->
+ new OrcFilters.Not(
+ new OrcFilters.LessThanEquals(
+ colName, litType,
serializableLiteral)));
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate> visitIn(FieldRef fieldRef,
List<Object> literals) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate> visitNotIn(FieldRef fieldRef,
List<Object> literals) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate>
visitAnd(List<Optional<OrcFilters.Predicate>> children) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<OrcFilters.Predicate>
visitOr(List<Optional<OrcFilters.Predicate>> children) {
+ if (children.size() != 2) {
+ throw new RuntimeException("Illegal or children: " +
children.size());
+ }
+
+ Optional<OrcFilters.Predicate> c1 = children.get(0);
+ if (!c1.isPresent()) {
+ return Optional.empty();
+ }
+ Optional<OrcFilters.Predicate> c2 = children.get(1);
+ return c2.map(value -> new OrcFilters.Or(c1.get(), value));
+ }
+
+ private Optional<OrcFilters.Predicate> convertBinary(
+ FieldRef fieldRef,
+ Object literal,
+ TriFunction<String, PredicateLeaf.Type, Serializable,
OrcFilters.Predicate> func) {
+ PredicateLeaf.Type litType = toOrcType(fieldRef.type());
+ if (litType == null) {
+ return Optional.empty();
+ }
+
+ String colName = fieldRef.name();
+
+ // fetch literal and ensure it is serializable
+ Object orcObj = toOrcObject(litType, literal);
+ Serializable serializableLiteral;
+ // validate that literal is serializable
+ if (orcObj instanceof Serializable) {
+ serializableLiteral = (Serializable) orcObj;
+ } else {
+ return Optional.empty();
+ }
+
+ return Optional.of(func.apply(colName, litType, serializableLiteral));
+ }
+
+ @Nullable
+ private static Object toOrcObject(PredicateLeaf.Type litType, Object
literalObj) {
+ if (literalObj == null) {
+ return null;
+ }
+
+ switch (litType) {
+ case STRING:
+ return literalObj.toString();
+ case DECIMAL:
+ return ((DecimalData) literalObj).toBigDecimal();
+ case DATE:
+ return Date.valueOf(LocalDate.ofEpochDay(((Number)
literalObj).longValue()));
+ case TIMESTAMP:
+ return ((TimestampData) literalObj).toTimestamp();
+ default:
+ return literalObj;
+ }
+ }
+
+ @Nullable
+ private static PredicateLeaf.Type toOrcType(LogicalType type) {
+ switch (type.getTypeRoot()) {
+ case TINYINT:
+ case SMALLINT:
+ case INTEGER:
+ case BIGINT:
+ return PredicateLeaf.Type.LONG;
+ case FLOAT:
+ case DOUBLE:
+ return PredicateLeaf.Type.FLOAT;
+ case BOOLEAN:
+ return PredicateLeaf.Type.BOOLEAN;
+ case CHAR:
+ case VARCHAR:
+ return PredicateLeaf.Type.STRING;
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ return PredicateLeaf.Type.TIMESTAMP;
+ case DATE:
+ return PredicateLeaf.Type.DATE;
+ case DECIMAL:
+ return PredicateLeaf.Type.DECIMAL;
+ default:
+ return null;
+ }
+ }
+}
diff --git
a/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFilterConverterTest.java
b/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFilterConverterTest.java
index 1c11af1c..73555294 100644
---
a/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFilterConverterTest.java
+++
b/flink-table-store-format/src/test/java/org/apache/flink/table/store/format/orc/OrcFilterConverterTest.java
@@ -32,7 +32,7 @@ import java.util.Collections;
import static org.assertj.core.api.Assertions.assertThat;
-/** Unit Tests for {@link OrcFilterConverter}. */
+/** Unit Tests for {@link OrcPredicateFunctionVisitor}. */
public class OrcFilterConverterTest {
@Test
@@ -74,7 +74,7 @@ public class OrcFilterConverterTest {
}
private void test(Predicate predicate, OrcFilters.Predicate orcPredicate) {
- assertThat(OrcFilterConverter.toOrcPredicate(predicate))
+ assertThat(predicate.visit(OrcPredicateFunctionVisitor.VISITOR).get())
.hasToString(orcPredicate.toString());
}
}
