WIP

Author: wuchong

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

@@ -92,6 +92,8 @@
 import java.util.stream.Collectors;
 
 import static org.apache.fluss.flink.utils.LakeSourceUtils.createLakeSource;
+import static org.apache.fluss.flink.utils.PushdownUtils.ValueConversion.FLINK_INTERNAL_VALUE;
+import static org.apache.fluss.flink.utils.PushdownUtils.extractFieldEquals;
 import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
 import static org.apache.fluss.utils.Preconditions.checkState;
@@ -142,11 +144,11 @@ public class FlinkTableSource
     @Nullable private RowLevelModificationType modificationScanType;
 
     // count(*) push down
-    protected boolean selectRowCount = false;
+    private boolean selectRowCount = false;
 
     private long limit = -1;
 
-    @Nullable protected Predicate partitionFilters;
+    @Nullable private Predicate partitionFilters;
 
     private final Map<String, String> tableOptions;
 
@@ -479,17 +481,17 @@ && hasPrimaryKey()
                 && filters.size() == primaryKeyIndexes.length) {
 
             Map<Integer, LogicalType> primaryKeyTypes = getPrimaryKeyTypes();
-            List<PushdownUtils.FieldEqual> fieldEquals =
-                    PushdownUtils.extractFieldEquals(
+            List<FieldEqual> fieldEquals =
+                    extractFieldEquals(
                             filters,
                             primaryKeyTypes,
                             acceptedFilters,
                             remainingFilters,
-                            PushdownUtils.ValueConversion.FLINK_INTERNAL_VALUE);
+                            FLINK_INTERNAL_VALUE);
             int[] keyRowProjection = getKeyRowProjection();
             HashSet<Integer> visitedPkFields = new HashSet<>();
             GenericRowData lookupRow = new GenericRowData(primaryKeyIndexes.length);
-            for (PushdownUtils.FieldEqual fieldEqual : fieldEquals) {
+            for (FieldEqual fieldEqual : fieldEquals) {
                 lookupRow.setField(keyRowProjection[fieldEqual.fieldIndex], fieldEqual.equalValue);
                 visitedPkFields.add(fieldEqual.fieldIndex);
             }
@@ -527,7 +529,7 @@ && hasPrimaryKey()
                                         partitionKeyTypes, partitionKeys.toArray(new String[0])),
                                 filter);
 
-                if (!predicateOptional.isPresent()) {
+                if (predicateOptional.isEmpty()) {
                     remainingFilters.add(filter);
                 } else {
                     Predicate p = predicateOptional.get();
@@ -542,19 +544,11 @@ && hasPrimaryKey()
             partitionFilters = converted.isEmpty() ? null : PredicateBuilder.and(converted);
             // lake source is not null
             if (lakeSource != null) {
-                PredicateVisitor<Boolean> lakePredicateVisitor =
-                        new PartitionPredicateVisitor(tableOutputType.getFieldNames());
-
                 List<Predicate> lakePredicates = new ArrayList<>();
                 for (ResolvedExpression filter : filters) {
-
                     Optional<Predicate> predicateOptional =
                             PredicateConverter.convert(tableOutputType, filter);
-
-                    if (predicateOptional.isPresent()) {
-                        Predicate p = predicateOptional.get();
-                        lakePredicates.add(p);
-                    }
+                    predicateOptional.ifPresent(lakePredicates::add);
                 }
 
                 if (!lakePredicates.isEmpty()) {

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

@@ -406,18 +406,14 @@ private void checkPartitionChanges(Set<PartitionInfo> partitionInfos, Throwable
             return;
         }
 
-        if (LOG.isDebugEnabled()) {
-            LOG.debug(
-                    "Checking partition changes for table {}, found {} partitions",
-                    tablePath,
-                    partitionInfos.size());
-        }
+        LOG.debug(
+                "Checking partition changes for table {}, found {} partitions",
+                tablePath,
+                partitionInfos.size());
 
         final PartitionChange partitionChange = getPartitionChange(partitionInfos);
         if (partitionChange.isEmpty()) {
-            if (LOG.isDebugEnabled()) {
-                LOG.debug("No partition changes detected for table {}", tablePath);
-            }
+            LOG.debug("No partition changes detected for table {}", tablePath);
             return;
         }