[core] DataEvolutionFileStoreScan should not filter files by read type when it contains no physical columns.

Author: 喆宇

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

@@ -32,6 +32,7 @@
 import org.apache.paimon.schema.SchemaManager;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.table.SpecialFields;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.RangeHelper;
@@ -127,8 +128,14 @@ private boolean filterManifestByRowIds(ManifestFileMeta manifest) {
 
     @Override
     public FileStoreScan withReadType(RowType readType) {
-        if (readType != null && !readType.getFields().isEmpty()) {
-            this.readType = readType;
+        if (readType != null) {
+            List<DataField> nonSystemFields =
+                    readType.getFields().stream()
+                            .filter(f -> !SpecialFields.isSystemField(f.id()))
+                            .collect(Collectors.toList());
+            if (!nonSystemFields.isEmpty()) {
+                this.readType = readType;
+            }
         }
         return this;
     }

paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java

@@ -109,6 +109,34 @@ public void testBasic() throws Exception {
                     assertThat(r.getString(1).toString()).isEqualTo("a");
                     assertThat(r.getString(2).toString()).isEqualTo("c");
                 });
+
+        // projection with only special fields.
+        readBuilder = getTableDefault().newReadBuilder();
+        reader =
+                readBuilder
+                        .withReadType(RowType.of(SpecialFields.ROW_ID))
+                        .newRead()
+                        .createReader(readBuilder.newScan().plan());
+        AtomicInteger cnt = new AtomicInteger(0);
+        reader.forEachRemaining(
+                r -> {
+                    cnt.incrementAndGet();
+                });
+        assertThat(cnt.get()).isEqualTo(1);
+
+        // projection with an empty read type
+        readBuilder = getTableDefault().newReadBuilder();
+        reader =
+                readBuilder
+                        .withReadType(RowType.of())
+                        .newRead()
+                        .createReader(readBuilder.newScan().plan());
+        AtomicInteger cnt1 = new AtomicInteger(0);
+        reader.forEachRemaining(
+                r -> {
+                    cnt1.incrementAndGet();
+                });
+        assertThat(cnt1.get()).isEqualTo(1);
     }
 
     @Test