[core] Refactory evolved and not evolved filters in file store scan

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreScan.java

@@ -44,10 +44,13 @@ public class AppendOnlyFileStoreScan extends AbstractFileStoreScan {
 
     private final boolean fileIndexReadEnabled;
 
-    private Predicate filter;
+    private Predicate inputFilter;
 
-    // just cache.
-    private final Map<Long, Predicate> dataFilterMapping = new ConcurrentHashMap<>();
+    // cache not evolved filter by schema id
+    private final Map<Long, Predicate> notEvolvedFilterMapping = new ConcurrentHashMap<>();
+
+    // cache evolved filter by schema id
+    private final Map<Long, Predicate> evolvedFilterMapping = new ConcurrentHashMap<>();
 
     public AppendOnlyFileStoreScan(
             ManifestsReader manifestsReader,
@@ -72,17 +75,23 @@ public AppendOnlyFileStoreScan(
     }
 
     public AppendOnlyFileStoreScan withFilter(Predicate predicate) {
-        this.filter = predicate;
+        this.inputFilter = predicate;
         this.bucketSelectConverter.convert(predicate).ifPresent(this::withTotalAwareBucketFilter);
         return this;
     }
 
     /** Note: Keep this thread-safe. */
     @Override
     protected boolean filterByStats(ManifestEntry entry) {
-        Predicate safeFilter =
-                simpleStatsEvolutions.toEvolutionSafeStatsFilter(entry.file().schemaId(), filter);
-        if (safeFilter == null) {
+        Predicate notEvolvedFilter =
+                notEvolvedFilterMapping.computeIfAbsent(
+                        entry.file().schemaId(),
+                        id ->
+                                // keepNewFieldFilter to handle add field
+                                // for example, add field 'c', 'c > 3': old files can be filtered
+                                simpleStatsEvolutions.filterUnsafeFilter(
+                                        entry.file().schemaId(), inputFilter, true));
+        if (notEvolvedFilter == null) {
             return true;
         }
 
@@ -93,12 +102,23 @@ protected boolean filterByStats(ManifestEntry entry) {
                         entry.file().rowCount(),
                         entry.file().valueStatsCols());
 
-        return safeFilter.test(
+        // filter by min max
+        boolean result =
+                notEvolvedFilter.test(
                         entry.file().rowCount(),
                         stats.minValues(),
                         stats.maxValues(),
-                        stats.nullCounts())
-                && (!fileIndexReadEnabled || testFileIndex(entry.file().embeddedIndex(), entry));
+                        stats.nullCounts());
+
+        if (!result) {
+            return false;
+        }
+
+        if (!fileIndexReadEnabled) {
+            return true;
+        }
+
+        return testFileIndex(entry.file().embeddedIndex(), entry);
     }
 
     private boolean testFileIndex(@Nullable byte[] embeddedIndexBytes, ManifestEntry entry) {
@@ -109,11 +129,11 @@ private boolean testFileIndex(@Nullable byte[] embeddedIndexBytes, ManifestEntry
         RowType dataRowType = scanTableSchema(entry.file().schemaId()).logicalRowType();
 
         Predicate dataPredicate =
-                dataFilterMapping.computeIfAbsent(
+                evolvedFilterMapping.computeIfAbsent(
                         entry.file().schemaId(),
                         id ->
-                                simpleStatsEvolutions.toEvolutionSafeStatsFilter(
-                                        entry.file().schemaId(), filter));
+                                simpleStatsEvolutions.tryDevolveFilter(
+                                        entry.file().schemaId(), inputFilter));
 
         try (FileIndexPredicate predicate =
                 new FileIndexPredicate(embeddedIndexBytes, dataRowType)) {

paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreScan.java

@@ -61,9 +61,14 @@ public class KeyValueFileStoreScan extends AbstractFileStoreScan {
 
     private Predicate keyFilter;
     private Predicate valueFilter;
-    private final Map<Long, Predicate> schemaId2DataFilter = new ConcurrentHashMap<>();
     private boolean valueFilterForceEnabled = false;
 
+    // cache not evolved filter by schema id
+    private final Map<Long, Predicate> notEvolvedFilterMapping = new ConcurrentHashMap<>();
+
+    // cache evolved filter by schema id
+    private final Map<Long, Predicate> evolvedFilterMapping = new ConcurrentHashMap<>();
+
     public KeyValueFileStoreScan(
             ManifestsReader manifestsReader,
             BucketSelectConverter bucketSelectConverter,
@@ -125,8 +130,7 @@ protected boolean filterByStats(ManifestEntry entry) {
         }
 
         Predicate safeKeyFilter =
-                fieldKeyStatsConverters.toEvolutionSafeStatsFilter(
-                        entry.file().schemaId(), keyFilter);
+                fieldKeyStatsConverters.tryDevolveFilter(entry.file().schemaId(), keyFilter);
         if (safeKeyFilter == null) {
             return true;
         }
@@ -157,10 +161,10 @@ private boolean filterByFileIndex(@Nullable byte[] embeddedIndexBytes, ManifestE
         try (FileIndexPredicate predicate =
                 new FileIndexPredicate(embeddedIndexBytes, dataRowType)) {
             Predicate dataPredicate =
-                    schemaId2DataFilter.computeIfAbsent(
+                    evolvedFilterMapping.computeIfAbsent(
                             entry.file().schemaId(),
                             id ->
-                                    fieldValueStatsConverters.toEvolutionSafeStatsFilter(
+                                    fieldValueStatsConverters.tryDevolveFilter(
                                             entry.file().schemaId(), valueFilter));
             return predicate.evaluate(dataPredicate).remain();
         } catch (IOException e) {
@@ -229,10 +233,15 @@ private boolean filterByValueFilter(ManifestEntry entry) {
             return ((FilteredManifestEntry) entry).selected();
         }
 
-        Predicate safeValueFilter =
-                fieldValueStatsConverters.toEvolutionSafeStatsFilter(
-                        entry.file().schemaId(), valueFilter);
-        if (safeValueFilter == null) {
+        Predicate notEvolvedFilter =
+                notEvolvedFilterMapping.computeIfAbsent(
+                        entry.file().schemaId(),
+                        id ->
+                                // keepNewFieldFilter to handle add field
+                                // for example, add field 'c', 'c > 3': old files can be filtered
+                                fieldValueStatsConverters.filterUnsafeFilter(
+                                        entry.file().schemaId(), valueFilter, true));
+        if (notEvolvedFilter == null) {
             return true;
         }
 
@@ -241,7 +250,7 @@ private boolean filterByValueFilter(ManifestEntry entry) {
                 fieldValueStatsConverters
                         .getOrCreate(file.schemaId())
                         .evolution(file.valueStats(), file.rowCount(), file.valueStatsCols());
-        return safeValueFilter.test(
+        return notEvolvedFilter.test(
                         file.rowCount(),
                         result.minValues(),
                         result.maxValues(),

paimon-core/src/main/java/org/apache/paimon/schema/SchemaEvolutionUtil.java

@@ -42,6 +42,7 @@
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -131,18 +132,17 @@ public CastFieldGetter[] getCastMapping() {
      * @param tableFields the table fields
      * @param dataFields the underlying data fields
      * @param filters the filters
-     * @param forData true if devolve the filters for filtering data file, otherwise, for filtering
-     *     manifest entry
+     * @param keepNewFieldFilter true if keep new field filter, the new field filter needs to be
+     *     properly handled
      * @return the data filters
      */
-    @Nullable
     public static List<Predicate> devolveFilters(
             List<DataField> tableFields,
             List<DataField> dataFields,
             List<Predicate> filters,
-            boolean forData) {
+            boolean keepNewFieldFilter) {
         if (filters == null) {
-            return null;
+            return Collections.emptyList();
         }
 
         Map<String, DataField> nameToTableFields =
@@ -159,36 +159,19 @@ public static List<Predicate> devolveFilters(
                                     String.format("Find no field %s", predicate.fieldName()));
                     DataField dataField = idToDataFields.get(tableField.id());
                     if (dataField == null) {
-                        // For example, add field b and filter b, the filter is safe for old file
-                        // meta without field b because the index mapping array can handle null
-                        return forData ? Optional.empty() : Optional.of(predicate);
+                        return keepNewFieldFilter ? Optional.of(predicate) : Optional.empty();
                     }
 
-                    Optional<List<Object>> castedLiterals =
-                            CastExecutors.castLiteralsWithEvolution(
-                                    predicate.literals(), predicate.type(), dataField.type());
-
-                    // unsafe
-                    if (!castedLiterals.isPresent()) {
-                        return Optional.empty();
-                    }
-
-                    if (forData) {
-                        // For data, the filter will be pushdown to data file, so must use the index
-                        // and literal type of data file
-                        return Optional.of(
-                                new LeafPredicate(
-                                        predicate.function(),
-                                        dataField.type(),
-                                        indexOf(dataField, idToDataFields),
-                                        dataField.name(),
-                                        castedLiterals.get()));
-                    } else {
-                        // For meta, the index mapping array will map the index the cast the
-                        // literals, so just return self
-                        // In other words, return it if it's safe
-                        return Optional.of(predicate);
-                    }
+                    return CastExecutors.castLiteralsWithEvolution(
+                                    predicate.literals(), predicate.type(), dataField.type())
+                            .map(
+                                    literals ->
+                                            new LeafPredicate(
+                                                    predicate.function(),
+                                                    dataField.type(),
+                                                    indexOf(dataField, idToDataFields),
+                                                    dataField.name(),
+                                                    literals));
                 };
 
         for (Predicate predicate : filters) {

paimon-core/src/main/java/org/apache/paimon/stats/SimpleStatsEvolutions.java

@@ -21,20 +21,21 @@
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.predicate.PredicateBuilder;
 import org.apache.paimon.schema.IndexCastMapping;
-import org.apache.paimon.schema.SchemaEvolutionUtil;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.RowType;
 
 import javax.annotation.Nullable;
 
+import java.util.ArrayList;
 import java.util.List;
-import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
+import static java.util.Collections.singletonList;
 import static org.apache.paimon.schema.SchemaEvolutionUtil.createIndexCastMapping;
+import static org.apache.paimon.schema.SchemaEvolutionUtil.devolveFilters;
 
 /** Converters to create col stats array serializer. */
 public class SimpleStatsEvolutions {
@@ -82,7 +83,7 @@ public SimpleStatsEvolution getOrCreate(long dataSchemaId) {
      * filter or null if can't.
      */
     @Nullable
-    public Predicate toEvolutionSafeStatsFilter(long dataSchemaId, @Nullable Predicate filter) {
+    public Predicate tryDevolveFilter(long dataSchemaId, @Nullable Predicate filter) {
         if (filter == null || dataSchemaId == tableSchemaId) {
             return filter;
         }
@@ -91,13 +92,38 @@ public Predicate toEvolutionSafeStatsFilter(long dataSchemaId, @Nullable Predica
         // compute engine to perform p2.
         List<Predicate> filters = PredicateBuilder.splitAnd(filter);
         List<Predicate> devolved =
-                Objects.requireNonNull(
-                        SchemaEvolutionUtil.devolveFilters(
-                                tableDataFields, schemaFields.apply(dataSchemaId), filters, false));
+                devolveFilters(tableDataFields, schemaFields.apply(dataSchemaId), filters, false);
 
         return devolved.isEmpty() ? null : PredicateBuilder.and(devolved);
     }
 
+    /**
+     * Filter unsafe filter, for example, filter is 'a > 9', old type is String, new type is Int, if
+     * records are 9, 10 and 11, the evolved filter is not safe.
+     */
+    @Nullable
+    public Predicate filterUnsafeFilter(
+            long dataSchemaId, @Nullable Predicate filter, boolean keepNewFieldFilter) {
+        if (filter == null || dataSchemaId == tableSchemaId) {
+            return filter;
+        }
+
+        List<Predicate> filters = PredicateBuilder.splitAnd(filter);
+        List<DataField> oldSchema = schemaFields.apply(dataSchemaId);
+        List<Predicate> result = new ArrayList<>();
+        for (Predicate predicate : filters) {
+            if (!devolveFilters(
+                            tableDataFields,
+                            oldSchema,
+                            singletonList(predicate),
+                            keepNewFieldFilter)
+                    .isEmpty()) {
+                result.add(predicate);
+            }
+        }
+        return result.isEmpty() ? null : PredicateBuilder.and(result);
+    }
+
     public List<DataField> tableDataFields() {
         return tableDataFields;
     }

paimon-core/src/main/java/org/apache/paimon/utils/FormatReaderMapping.java

@@ -309,7 +309,7 @@ private List<Predicate> readFilters(
                     tableSchema.id() == dataSchema.id()
                             ? filters
                             : SchemaEvolutionUtil.devolveFilters(
-                                    tableSchema.fields(), dataSchema.fields(), filters, true);
+                                    tableSchema.fields(), dataSchema.fields(), filters, false);
 
             // Skip pushing down partition filters to reader.
             return excludePredicateWithFields(

paimon-core/src/test/java/org/apache/paimon/schema/SchemaEvolutionUtilTest.java

@@ -91,7 +91,7 @@ public void testDevolveDataFilters() {
                         IsNull.INSTANCE, DataTypes.INT(), 7, "a", Collections.emptyList()));
 
         List<Predicate> filters =
-                SchemaEvolutionUtil.devolveFilters(tableFields2, dataFields, predicates, true);
+                SchemaEvolutionUtil.devolveFilters(tableFields2, dataFields, predicates, false);
         assertThat(filters).isNotNull();
         assertThat(filters.size()).isEqualTo(1);