[flink] Support general predicates for partition pushdown in Flink Connector (apache#420)

Author: Alibaba-HZY

LICENSE

@@ -361,6 +361,7 @@ Apache Kafka
 ./fluss-server/src/main/java/org/apache/fluss/server/utils/timer/TimingWheel.java
 
 Apache Paimon
+./fluss-common/src/main/java/org/apache/fluss/flink/utils/PredicateConverter.java
 ./fluss-common/src/main/java/org/apache/fluss/predicate/And.java
 ./fluss-common/src/main/java/org/apache/fluss/predicate/CompareUtils.java
 ./fluss-common/src/main/java/org/apache/fluss/predicate/CompoundPredicate.java

fluss-common/src/main/java/org/apache/fluss/predicate/LeafPredicate.java

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DataTypes;
 import org.apache.fluss.types.DecimalType;
 import org.apache.fluss.types.LocalZonedTimestampType;
 import org.apache.fluss.types.TimestampType;
@@ -84,6 +85,10 @@ public LeafPredicate copyWithNewIndex(int fieldIndex) {
         return new LeafPredicate(function, type, fieldIndex, fieldName, literals);
     }
 
+    public LeafPredicate copyWithNewLiterals(List<Object> literals) {
+        return new LeafPredicate(function, DataTypes.STRING(), fieldIndex, fieldName, literals);
+    }
+
     @Override
     public boolean test(InternalRow row) {
         return function.test(type, get(row, fieldIndex, type), literals);

fluss-common/src/main/java/org/apache/fluss/predicate/SimpleColStats.java

@@ -0,0 +1,87 @@
+/*
+ * 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.fluss.predicate;
+
+import javax.annotation.Nullable;
+
+import java.util.Objects;
+
+/**
+ * A simple column statistics, supports the following stats.
+ *
+ * <ul>
+ *   <li>min: the minimum value of the column
+ *   <li>max: the maximum value of the column
+ *   <li>nullCount: the number of nulls
+ * </ul>
+ */
+public class SimpleColStats {
+
+    public static final SimpleColStats NONE = new SimpleColStats(null, null, null);
+
+    @Nullable private final Object min;
+    @Nullable private final Object max;
+    private final Long nullCount;
+
+    public SimpleColStats(@Nullable Object min, @Nullable Object max, @Nullable Long nullCount) {
+        this.min = min;
+        this.max = max;
+        this.nullCount = nullCount;
+    }
+
+    @Nullable
+    public Object min() {
+        return min;
+    }
+
+    @Nullable
+    public Object max() {
+        return max;
+    }
+
+    @Nullable
+    public Long nullCount() {
+        return nullCount;
+    }
+
+    public boolean isNone() {
+        return min == null && max == null && nullCount == null;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof SimpleColStats)) {
+            return false;
+        }
+        SimpleColStats that = (SimpleColStats) o;
+        return Objects.equals(min, that.min)
+                && Objects.equals(max, that.max)
+                && Objects.equals(nullCount, that.nullCount);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(min, max, nullCount);
+    }
+
+    @Override
+    public String toString() {
+        return String.format("{%s, %s, %d}", min, max, nullCount);
+    }
+}

fluss-common/src/main/java/org/apache/fluss/utils/TypeUtils.java

@@ -21,10 +21,12 @@
 import org.apache.fluss.row.Decimal;
 import org.apache.fluss.types.DataType;
 import org.apache.fluss.types.DecimalType;
+import org.apache.fluss.types.LocalZonedTimestampType;
 import org.apache.fluss.types.TimestampType;
 
 import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
+import java.util.TimeZone;
 
 /** Type related helper functions. */
 public class TypeUtils {
@@ -62,6 +64,10 @@ public static Object castFromString(String s, DataType type) {
             case TIMESTAMP_WITHOUT_TIME_ZONE:
                 TimestampType timestampType = (TimestampType) type;
                 return BinaryStringUtils.toTimestampNtz(str, timestampType.getPrecision());
+            case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+                LocalZonedTimestampType localZonedTimestampType = (LocalZonedTimestampType) type;
+                return BinaryStringUtils.toTimestampLtz(
+                        str, localZonedTimestampType.getPrecision(), TimeZone.getDefault());
             default:
                 throw new UnsupportedOperationException("Unsupported type " + type);
         }

fluss-common/src/test/java/org/apache/fluss/predicate/SimpleColStatsTestUtils.java

@@ -0,0 +1,38 @@
+/*
+ * 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.fluss.predicate;
+
+import org.apache.fluss.row.GenericRow;
+
+/** Utils for testing with {@link SimpleColStats}. */
+public class SimpleColStatsTestUtils {
+
+    public static boolean test(Predicate predicate, long rowCount, SimpleColStats[] fieldStats) {
+        Object[] min = new Object[fieldStats.length];
+        Object[] max = new Object[fieldStats.length];
+        Long[] nullCounts = new Long[fieldStats.length];
+        for (int i = 0; i < fieldStats.length; i++) {
+            min[i] = fieldStats[i].min();
+            max[i] = fieldStats[i].max();
+            nullCounts[i] = fieldStats[i].nullCount();
+        }
+
+        return predicate.test(rowCount, GenericRow.of(min), GenericRow.of(max), nullCounts);
+    }
+}

fluss-flink/fluss-flink-common/pom.xml

@@ -34,6 +34,7 @@
     <properties>
         <flink.major.version>1.20</flink.major.version>
         <flink.minor.version>1.20.1</flink.minor.version>
+        <scala.binary.version>2.12</scala.binary.version>
     </properties>
 
     <dependencies>
@@ -126,6 +127,21 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+            <version>${flink.minor.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+            <version>${flink.minor.version}</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.fluss</groupId>
             <artifactId>fluss-server</artifactId>

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/row/FlinkAsFlussRow.java

@@ -17,15 +17,18 @@
 
 package org.apache.fluss.flink.row;
 
+import org.apache.fluss.flink.utils.FlinkConversions;
 import org.apache.fluss.row.BinaryString;
 import org.apache.fluss.row.Decimal;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.row.TimestampLtz;
 import org.apache.fluss.row.TimestampNtz;
 
 import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericRowData;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.types.DataType;
 
 /** Wraps a Flink {@link RowData} as a Fluss {@link InternalRow}. */
 public class FlinkAsFlussRow implements InternalRow {
@@ -132,4 +135,12 @@ public byte[] getBinary(int pos, int length) {
     public byte[] getBytes(int pos) {
         return flinkRow.getBinary(pos);
     }
+
+    public static Object fromFlinkObject(Object o, DataType type) {
+        if (o == null) {
+            return null;
+        }
+        return InternalRow.createFieldGetter(FlinkConversions.toFlussType(type), 0)
+                .getFieldOrNull((new FlinkAsFlussRow()).replace(GenericRowData.of(o)));
+    }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java

@@ -30,10 +30,10 @@
 import org.apache.fluss.flink.source.split.SourceSplitSerializer;
 import org.apache.fluss.flink.source.state.FlussSourceEnumeratorStateSerializer;
 import org.apache.fluss.flink.source.state.SourceEnumeratorState;
-import org.apache.fluss.flink.utils.PushdownUtils.FieldEqual;
 import org.apache.fluss.lake.source.LakeSource;
 import org.apache.fluss.lake.source.LakeSplit;
 import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.predicate.Predicate;
 import org.apache.fluss.types.RowType;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -50,10 +50,6 @@
 
 import javax.annotation.Nullable;
 
-import java.util.List;
-
-import static org.apache.fluss.utils.Preconditions.checkNotNull;
-
 /** Flink source for Fluss. */
 public class FlinkSource<OUT>
         implements Source<OUT, SourceSplitBase, SourceEnumeratorState>, ResultTypeQueryable {
@@ -69,10 +65,8 @@ public class FlinkSource<OUT>
     protected final long scanPartitionDiscoveryIntervalMs;
     private final boolean streaming;
     private final FlussDeserializationSchema<OUT> deserializationSchema;
-
-    private final List<FieldEqual> partitionFilters;
-
-    private final @Nullable LakeSource<LakeSplit> lakeSource;
+    @Nullable private final Predicate partitionFilters;
+    @Nullable private final LakeSource<LakeSplit> lakeSource;
 
     public FlinkSource(
             Configuration flussConf,
@@ -85,7 +79,7 @@ public FlinkSource(
             long scanPartitionDiscoveryIntervalMs,
             FlussDeserializationSchema<OUT> deserializationSchema,
             boolean streaming,
-            List<FieldEqual> partitionFilters) {
+            @Nullable Predicate partitionFilters) {
         this(
                 flussConf,
                 tablePath,
@@ -112,8 +106,8 @@ public FlinkSource(
             long scanPartitionDiscoveryIntervalMs,
             FlussDeserializationSchema<OUT> deserializationSchema,
             boolean streaming,
-            List<FieldEqual> partitionFilters,
-            LakeSource<LakeSplit> lakeSource) {
+            @Nullable Predicate partitionFilters,
+            @Nullable LakeSource<LakeSplit> lakeSource) {
         this.flussConf = flussConf;
         this.tablePath = tablePath;
         this.hasPrimaryKey = hasPrimaryKey;
@@ -124,7 +118,7 @@ public FlinkSource(
         this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
         this.deserializationSchema = deserializationSchema;
         this.streaming = streaming;
-        this.partitionFilters = checkNotNull(partitionFilters);
+        this.partitionFilters = partitionFilters;
         this.lakeSource = lakeSource;
     }