[hotfix] Improve some codes in FlinkTableSource (#1083)

Author: swuferhong

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/sink/FlinkTableSink.java

@@ -56,6 +56,8 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import static com.alibaba.fluss.flink.utils.PushdownUtils.extractFieldEquals;
+
 /** A Flink {@link DynamicTableSink}. */
 public class FlinkTableSink
         implements DynamicTableSink,
@@ -263,7 +265,7 @@ public boolean applyDeleteFilters(List<ResolvedExpression> filters) {
         List<ResolvedExpression> remainingFilters = new ArrayList<>();
         Map<Integer, LogicalType> primaryKeyTypes = getPrimaryKeyTypes();
         List<FieldEqual> fieldEquals =
-                PushdownUtils.extractFieldEquals(
+                extractFieldEquals(
                         filters,
                         primaryKeyTypes,
                         acceptedFilters,

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkTableSource.java

@@ -76,6 +76,7 @@
 import java.util.Map;
 
 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. */
@@ -398,7 +399,7 @@ && hasPrimaryKey()
                 && filters.size() == primaryKeyIndexes.length) {
             Map<Integer, LogicalType> primaryKeyTypes = getPrimaryKeyTypes();
             List<FieldEqual> fieldEquals =
-                    PushdownUtils.extractFieldEquals(
+                    extractFieldEquals(
                             filters,
                             primaryKeyTypes,
                             acceptedFilters,
@@ -419,11 +420,10 @@ && hasPrimaryKey()
             return Result.of(acceptedFilters, remainingFilters);
         } else if (isPartitioned()) {
             // dynamic partition pushdown
-            Map<Integer, LogicalType> partitionKeyTypes = getPartitionKeyTypes();
             List<FieldEqual> fieldEquals =
-                    PushdownUtils.extractFieldEquals(
+                    extractFieldEquals(
                             filters,
-                            partitionKeyTypes,
+                            getPartitionKeyTypes(),
                             acceptedFilters,
                             remainingFilters,
                             FLINK_INTERNAL_VALUE);

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/utils/PushdownUtils.java

@@ -83,7 +83,7 @@ public class PushdownUtils {
     /** Extract field equality information from expressions. */
     public static List<FieldEqual> extractFieldEquals(
             List<ResolvedExpression> expressions,
-            Map<Integer, LogicalType> primaryKeyColumn,
+            Map<Integer, LogicalType> fieldIndexToType,
             List<ResolvedExpression> acceptedFiltersResult,
             List<ResolvedExpression> remainingFiltersResult,
             ValueConversion valueConversion) {
@@ -107,7 +107,7 @@ public static List<FieldEqual> extractFieldEquals(
                                 extractFieldEqual(
                                         leftFieldRef,
                                         rightValue,
-                                        primaryKeyColumn,
+                                        fieldIndexToType,
                                         valueConversion);
                     } else if (left instanceof ValueLiteralExpression
                             && right instanceof FieldReferenceExpression) {
@@ -117,7 +117,7 @@ public static List<FieldEqual> extractFieldEquals(
                                 extractFieldEqual(
                                         rightFieldRef,
                                         leftValue,
-                                        primaryKeyColumn,
+                                        fieldIndexToType,
                                         valueConversion);
                     }
 
@@ -141,11 +141,11 @@ public static List<FieldEqual> extractFieldEquals(
     private static FieldEqual extractFieldEqual(
             FieldReferenceExpression fieldsRef,
             ValueLiteralExpression valueLiteral,
-            Map<Integer, LogicalType> primaryKeyColumn,
+            Map<Integer, LogicalType> fieldIndexToType,
             ValueConversion valueConversion) {
         int columnIndex = fieldsRef.getFieldIndex();
-        if (primaryKeyColumn.containsKey(columnIndex)) {
-            LogicalType expectedType = primaryKeyColumn.get(columnIndex);
+        if (fieldIndexToType.containsKey(columnIndex)) {
+            LogicalType expectedType = fieldIndexToType.get(columnIndex);
             if (expectedType.getTypeRoot()
                     != valueLiteral.getOutputDataType().getLogicalType().getTypeRoot()) {
                 return null;