[predicate]regenerate serialVersionUID

Alibaba-HZY · Alibaba-HZY · commit e2459d88fe08 · 2025-07-25T10:58:08.000+08:00
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/And.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/And.java
@@ -30,7 +30,7 @@
 /** A {@link CompoundPredicate.Function} to eval and. */
 public class And extends CompoundPredicate.Function {
 
-    private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = -2977938814804928712L;
 
     public static final And INSTANCE = new And();
 
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/FieldRef.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/FieldRef.java
@@ -29,7 +29,7 @@
 /** A reference to a field in an input. */
 public class FieldRef implements Serializable {
 
-    private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = 4982103776651292199L;
 
     private final int index;
     private final String name;
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/In.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/In.java
@@ -31,7 +31,7 @@
 /** A {@link LeafFunction} to eval in. */
 public class In extends LeafFunction {
 
-    private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = -9115697441080586485L;
 
     public static final In INSTANCE = new In();
 
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafPredicate.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafPredicate.java
@@ -34,7 +34,7 @@
 /** Leaf node of a {@link Predicate} tree. Compares a field in the row with literals. */
 public class LeafPredicate implements Predicate {
 
-    private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = -9033842253303772188L;
 
     private final LeafFunction function;
     private final DataType type;
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafUnaryFunction.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/LeafUnaryFunction.java
@@ -28,7 +28,7 @@
 /** Function to test a field. */
 public abstract class LeafUnaryFunction extends LeafFunction {
 
-    private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = -155104972966998013L;
 
     public abstract boolean test(DataType type, Object value);
 
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/NotIn.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/NotIn.java
@@ -31,8 +31,7 @@
 /** A {@link LeafFunction} to eval not in. */
 public class NotIn extends LeafFunction {
 
-    private static final long serialVersionUID = 1L;
-
+    private static final long serialVersionUID = 8953845894700582887L;
     public static final NotIn INSTANCE = new NotIn();
 
     private NotIn() {}
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/NullFalseLeafBinaryFunction.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/NullFalseLeafBinaryFunction.java
@@ -28,7 +28,7 @@
 /** Function to test a field with a literal. */
 public abstract class NullFalseLeafBinaryFunction extends LeafFunction {
 
-    private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = 5617091663961558170L;
 
     public abstract boolean test(DataType type, Object field, Object literal);
 
diff --git a/fluss-common/src/main/java/com/alibaba/fluss/predicate/Or.java b/fluss-common/src/main/java/com/alibaba/fluss/predicate/Or.java
@@ -30,7 +30,7 @@
 /** A {@link CompoundPredicate.Function} to eval or. */
 public class Or extends CompoundPredicate.Function {
 
-    private static final long serialVersionUID = 1L;
+    private static final long serialVersionUID = -2110346319473699418L;
 
     public static final Or INSTANCE = new Or();
 
diff --git a/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/row/FlinkAsFlussRow.java b/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/row/FlinkAsFlussRow.java
@@ -17,15 +17,18 @@
 
 package com.alibaba.fluss.flink.row;
 
+import com.alibaba.fluss.flink.utils.FlinkConversions;
 import com.alibaba.fluss.row.BinaryString;
 import com.alibaba.fluss.row.Decimal;
 import com.alibaba.fluss.row.InternalRow;
 import com.alibaba.fluss.row.TimestampLtz;
 import com.alibaba.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)));
+    }
 }
diff --git a/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkSource.java b/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkSource.java
@@ -30,8 +30,8 @@
 import com.alibaba.fluss.flink.source.split.SourceSplitSerializer;
 import com.alibaba.fluss.flink.source.state.FlussSourceEnumeratorStateSerializer;
 import com.alibaba.fluss.flink.source.state.SourceEnumeratorState;
-import com.alibaba.fluss.flink.utils.PushdownUtils.FieldEqual;
 import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.predicate.Predicate;
 import com.alibaba.fluss.types.RowType;
 
 import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -48,10 +48,6 @@
 
 import javax.annotation.Nullable;
 
-import java.util.List;
-
-import static com.alibaba.fluss.utils.Preconditions.checkNotNull;
-
 /** Flink source for Fluss. */
 public class FlinkSource<OUT>
         implements Source<OUT, SourceSplitBase, SourceEnumeratorState>, ResultTypeQueryable {
@@ -68,7 +64,7 @@ public class FlinkSource<OUT>
     private final boolean streaming;
     private final FlussDeserializationSchema<OUT> deserializationSchema;
 
-    private final List<FieldEqual> partitionFilters;
+    private Predicate partitionFilters;
 
     public FlinkSource(
             Configuration flussConf,
@@ -81,7 +77,7 @@ public FlinkSource(
             long scanPartitionDiscoveryIntervalMs,
             FlussDeserializationSchema<OUT> deserializationSchema,
             boolean streaming,
-            List<FieldEqual> partitionFilters) {
+            Predicate partitionFilters) {
         this.flussConf = flussConf;
         this.tablePath = tablePath;
         this.hasPrimaryKey = hasPrimaryKey;
@@ -92,7 +88,7 @@ public FlinkSource(
         this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
         this.deserializationSchema = deserializationSchema;
         this.streaming = streaming;
-        this.partitionFilters = checkNotNull(partitionFilters);
+        this.partitionFilters = partitionFilters;
     }
 
     @Override
diff --git a/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkTableSource.java
@@ -30,6 +30,10 @@
 import com.alibaba.fluss.flink.utils.PushdownUtils.FieldEqual;
 import com.alibaba.fluss.metadata.MergeEngineType;
 import com.alibaba.fluss.metadata.TablePath;
+import com.alibaba.fluss.predicate.PartitionPredicateVisitor;
+import com.alibaba.fluss.predicate.Predicate;
+import com.alibaba.fluss.predicate.PredicateBuilder;
+import com.alibaba.fluss.predicate.PredicateVisitor;
 import com.alibaba.fluss.types.RowType;
 
 import org.apache.flink.annotation.VisibleForTesting;
@@ -64,6 +68,7 @@
 import org.apache.flink.table.functions.LookupFunction;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.VarCharType;
 
 import javax.annotation.Nullable;
 
@@ -75,9 +80,9 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
 
-import static com.alibaba.fluss.flink.utils.PushdownUtils.ValueConversion.FLINK_INTERNAL_VALUE;
-import static com.alibaba.fluss.flink.utils.PushdownUtils.extractFieldEquals;
 import static com.alibaba.fluss.utils.Preconditions.checkNotNull;
 
 /** Flink table source to scan Fluss data. */
@@ -128,7 +133,7 @@ public class FlinkTableSource
 
     private long limit = -1;
 
-    private List<FieldEqual> partitionFilters = Collections.emptyList();
+    @Nullable protected Predicate partitionFilters;
 
     public FlinkTableSource(
             TablePath tablePath,
@@ -386,6 +391,7 @@ public void applyProjection(int[][] projectedFields, DataType producedDataType)
 
     @Override
     public Result applyFilters(List<ResolvedExpression> filters) {
+
         List<ResolvedExpression> acceptedFilters = new ArrayList<>();
         List<ResolvedExpression> remainingFilters = new ArrayList<>();
 
@@ -398,18 +404,19 @@ public Result applyFilters(List<ResolvedExpression> filters) {
                 && startupOptions.startupMode == FlinkConnectorOptions.ScanStartupMode.FULL
                 && hasPrimaryKey()
                 && filters.size() == primaryKeyIndexes.length) {
+
             Map<Integer, LogicalType> primaryKeyTypes = getPrimaryKeyTypes();
-            List<FieldEqual> fieldEquals =
-                    extractFieldEquals(
+            List<PushdownUtils.FieldEqual> fieldEquals =
+                    PushdownUtils.extractFieldEquals(
                             filters,
                             primaryKeyTypes,
                             acceptedFilters,
                             remainingFilters,
-                            FLINK_INTERNAL_VALUE);
+                            PushdownUtils.ValueConversion.FLINK_INTERNAL_VALUE);
             int[] keyRowProjection = getKeyRowProjection();
             HashSet<Integer> visitedPkFields = new HashSet<>();
             GenericRowData lookupRow = new GenericRowData(primaryKeyIndexes.length);
-            for (FieldEqual fieldEqual : fieldEquals) {
+            for (PushdownUtils.FieldEqual fieldEqual : fieldEquals) {
                 lookupRow.setField(keyRowProjection[fieldEqual.fieldIndex], fieldEqual.equalValue);
                 visitedPkFields.add(fieldEqual.fieldIndex);
             }
@@ -419,23 +426,51 @@ && hasPrimaryKey()
             }
             singleRowFilter = lookupRow;
             return Result.of(acceptedFilters, remainingFilters);
-        } else if (isPartitioned()) {
-            // dynamic partition pushdown
-            List<FieldEqual> fieldEquals =
-                    extractFieldEquals(
-                            filters,
-                            getPartitionKeyTypes(),
-                            acceptedFilters,
-                            remainingFilters,
-                            FLINK_INTERNAL_VALUE);
-            // partitions are filtered by string representations, convert the equals to string first
-            fieldEquals = stringifyFieldEquals(fieldEquals);
-
-            this.partitionFilters = fieldEquals;
+        } else if (isPartitioned()
+                && !RowLevelModificationType.UPDATE.equals(modificationScanType)) {
+            // apply partition filter pushdown
+            List<Predicate> converted = new ArrayList<>();
+
+            List<String> fieldNames = tableOutputType.getFieldNames();
+            List<String> partitionKeys =
+                    Arrays.stream(partitionKeyIndexes)
+                            .mapToObj(fieldNames::get)
+                            .collect(Collectors.toList());
+
+            PredicateVisitor<Boolean> partitionPredicateVisitor =
+                    new PartitionPredicateVisitor(partitionKeys);
+
+            // TODO after https://github.com/alibaba/fluss/pull/979
+            //  replace string type with the real type
+            LogicalType[] partitionKeyTypes =
+                    partitionKeys.stream()
+                            .map(key -> VarCharType.STRING_TYPE)
+                            .toArray(LogicalType[]::new);
+            for (ResolvedExpression filter : filters) {
+
+                Optional<Predicate> predicateOptional =
+                        PredicateConverter.convert(
+                                org.apache.flink.table.types.logical.RowType.of(
+                                        partitionKeyTypes, partitionKeys.toArray(new String[0])),
+                                filter);
+
+                if (!predicateOptional.isPresent()) {
+                    remainingFilters.add(filter);
+                } else {
+                    Predicate p = predicateOptional.get();
+                    if (!p.visit(partitionPredicateVisitor)) {
+                        remainingFilters.add(filter);
+                    } else {
+                        acceptedFilters.add(filter);
+                    }
+                    converted.add(p);
+                }
+            }
+            partitionFilters = converted.isEmpty() ? null : PredicateBuilder.and(converted);
             return Result.of(acceptedFilters, remainingFilters);
-        } else {
-            return Result.of(Collections.emptyList(), filters);
         }
+
+        return Result.of(Collections.emptyList(), filters);
     }
 
     @Override
diff --git a/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlussSource.java b/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlussSource.java
@@ -26,8 +26,6 @@
 
 import javax.annotation.Nullable;
 
-import java.util.Collections;
-
 /**
  * A Flink DataStream source implementation for reading data from Fluss tables.
  *
@@ -82,7 +80,7 @@ public class FlussSource<OUT> extends FlinkSource<OUT> {
                 scanPartitionDiscoveryIntervalMs,
                 deserializationSchema,
                 streaming,
-                Collections.emptyList());
+                null);
     }
 
     /**
diff --git a/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/PredicateConverter.java b/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/PredicateConverter.java
diff --git a/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/enumerator/FlinkSourceEnumerator.java b/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/enumerator/FlinkSourceEnumerator.java
diff --git a/fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/FlinkTableSourceITCase.java b/fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/FlinkTableSourceITCase.java
diff --git a/fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
