Core: Predicate pushdown for all_data_files table

core/src/main/java/org/apache/iceberg/AllDataFilesTable.java

@@ -20,18 +20,10 @@
 package org.apache.iceberg;
 
 import java.io.IOException;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import org.apache.iceberg.BaseFilesTable.ManifestReadTask;
 import org.apache.iceberg.exceptions.RuntimeIOException;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.expressions.ResidualEvaluator;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
-import org.apache.iceberg.types.TypeUtil;
-import org.apache.iceberg.types.Types.StructType;
 import org.apache.iceberg.util.ParallelIterable;
 
 /**
@@ -41,7 +33,7 @@
  * <p>
  * This table may return duplicate rows.
  */
-public class AllDataFilesTable extends BaseMetadataTable {
+public class AllDataFilesTable extends BaseFilesTable {
 
   AllDataFilesTable(TableOperations ops, Table table) {
     this(ops, table, table.name() + ".all_data_files");
@@ -56,45 +48,25 @@ public TableScan newScan() {
     return new AllDataFilesTableScan(operations(), table(), schema());
   }
 
-  @Override
-  public Schema schema() {
-    StructType partitionType = Partitioning.partitionType(table());
-    Schema schema = new Schema(DataFile.getType(partitionType).fields());
-    if (partitionType.fields().size() < 1) {
-      // avoid returning an empty struct, which is not always supported. instead, drop the partition field (id 102)
-      return TypeUtil.selectNot(schema, Sets.newHashSet(102));
-    } else {
-      return schema;
-    }
-  }
-
   @Override
   MetadataTableType metadataTableType() {
     return MetadataTableType.ALL_DATA_FILES;
   }
 
-  public static class AllDataFilesTableScan extends BaseAllMetadataTableScan {
-    private final Schema fileSchema;
+  public static class AllDataFilesTableScan extends BaseFilesTableScan {
 
     AllDataFilesTableScan(TableOperations ops, Table table, Schema fileSchema) {
-      super(ops, table, fileSchema);
-      this.fileSchema = fileSchema;
+      super(ops, table, fileSchema, MetadataTableType.ALL_DATA_FILES);
     }
 
     private AllDataFilesTableScan(TableOperations ops, Table table, Schema schema, Schema fileSchema,
                                   TableScanContext context) {
-      super(ops, table, schema, context);
-      this.fileSchema = fileSchema;
-    }
-
-    @Override
-    protected String tableType() {
-      return MetadataTableType.ALL_DATA_FILES.name();
+      super(ops, table, schema, fileSchema, context, MetadataTableType.ALL_DATA_FILES);
     }
 
     @Override
     protected TableScan newRefinedScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
-      return new AllDataFilesTableScan(ops, table, schema, fileSchema, context);
+      return new AllDataFilesTableScan(ops, table, schema, fileSchema(), context);
     }
 
     @Override
@@ -108,30 +80,20 @@ public TableScan asOfTime(long timestampMillis) {
     }
 
     @Override
-    protected CloseableIterable<FileScanTask> planFiles(
-        TableOperations ops, Snapshot snapshot, Expression rowFilter,
-        boolean ignoreResiduals, boolean caseSensitive, boolean colStats) {
-      CloseableIterable<ManifestFile> manifests = allDataManifestFiles(
-          ops.current().snapshots(), context().planExecutor());
-      String schemaString = SchemaParser.toJson(schema());
-      String specString = PartitionSpecParser.toJson(PartitionSpec.unpartitioned());
-      Expression filter = ignoreResiduals ? Expressions.alwaysTrue() : rowFilter;
-      ResidualEvaluator residuals = ResidualEvaluator.unpartitioned(filter);
-
-      return CloseableIterable.transform(manifests, manifest ->
-          new ManifestReadTask(ops.io(), ops.current().specsById(), manifest, schema(),
-              schemaString, specString, residuals));
+    public CloseableIterable<FileScanTask> planFiles() {
+      return super.planFilesAllSnapshots();
     }
-  }
 
-  private static CloseableIterable<ManifestFile> allDataManifestFiles(
-      List<Snapshot> snapshots, ExecutorService workerPool) {
-    try (CloseableIterable<ManifestFile> iterable = new ParallelIterable<>(
-        Iterables.transform(snapshots, snapshot -> (Iterable<ManifestFile>) () -> snapshot.dataManifests().iterator()),
-        workerPool)) {
-      return CloseableIterable.withNoopClose(Sets.newHashSet(iterable));
-    } catch (IOException e) {
-      throw new RuntimeIOException(e, "Failed to close parallel iterable");
+    @Override
+    protected CloseableIterable<ManifestFile> manifests() {
+      try (CloseableIterable<ManifestFile> iterable = new ParallelIterable<>(
+          Iterables.transform(table().snapshots(),
+              snapshot -> (Iterable<ManifestFile>) () -> snapshot.dataManifests().iterator()),
+          context().planExecutor())) {
+        return CloseableIterable.withNoopClose(Sets.newHashSet(iterable));
+      } catch (IOException e) {
+        throw new RuntimeIOException(e, "Failed to close parallel iterable");
+      }
     }
   }
 }

core/src/main/java/org/apache/iceberg/BaseAllMetadataTableScan.java

@@ -19,14 +19,9 @@
 
 package org.apache.iceberg;
 
-import org.apache.iceberg.events.Listeners;
-import org.apache.iceberg.events.ScanEvent;
 import org.apache.iceberg.io.CloseableIterable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 abstract class BaseAllMetadataTableScan extends BaseMetadataTableScan {
-  private static final Logger LOG = LoggerFactory.getLogger(BaseAllMetadataTableScan.class);
 
   BaseAllMetadataTableScan(TableOperations ops, Table table, Schema fileSchema) {
     super(ops, table, fileSchema);
@@ -58,9 +53,6 @@ public TableScan appendsAfter(long fromSnapshotId) {
 
   @Override
   public CloseableIterable<FileScanTask> planFiles() {
-    LOG.info("Scanning metadata table {} with filter {}.", table(), filter());
-    Listeners.notifyAll(new ScanEvent(table().name(), 0L, filter(), schema()));
-
-    return planFiles(tableOps(), snapshot(), filter(), shouldIgnoreResiduals(), isCaseSensitive(), colStats());
+    return super.planFilesAllSnapshots();
   }
 }

core/src/main/java/org/apache/iceberg/BaseFilesTable.java

@@ -19,7 +19,6 @@
 
 package org.apache.iceberg;
 
-import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
@@ -56,6 +55,7 @@ public Schema schema() {
   }
 
   abstract static class BaseFilesTableScan extends BaseMetadataTableScan {
+
     private final Schema fileSchema;
     private final MetadataTableType type;
 
@@ -90,7 +90,8 @@ public TableScan appendsAfter(long fromSnapshotId) {
 
     @Override
     protected CloseableIterable<FileScanTask> planFiles(TableOperations ops, Snapshot snapshot, Expression rowFilter,
-        boolean ignoreResiduals, boolean caseSensitive, boolean colStats) {
+                                                        boolean ignoreResiduals, boolean caseSensitive,
+                                                        boolean colStats) {
       CloseableIterable<ManifestFile> filtered = filterManifests(manifests(), rowFilter, caseSensitive);
 
       String schemaString = SchemaParser.toJson(schema());
@@ -108,23 +109,21 @@ protected CloseableIterable<FileScanTask> planFiles(TableOperations ops, Snapsho
     }
 
     /**
-     * @return list of manifest files to explore for this files metadata table scan
+     * Returns an iterable of manifest files to explore for this Files metadata table scan
      */
-    protected abstract List<ManifestFile> manifests();
+    protected abstract CloseableIterable<ManifestFile> manifests();
 
-    private CloseableIterable<ManifestFile> filterManifests(List<ManifestFile> manifests,
+    private CloseableIterable<ManifestFile> filterManifests(CloseableIterable<ManifestFile> manifests,
                                                             Expression rowFilter,
                                                             boolean caseSensitive) {
-      CloseableIterable<ManifestFile> manifestIterable = CloseableIterable.withNoopClose(manifests);
-
       // use an inclusive projection to remove the partition name prefix and filter out any non-partition expressions
       PartitionSpec spec = transformSpec(fileSchema, table().spec(), PARTITION_FIELD_PREFIX);
       Expression partitionFilter = Projections.inclusive(spec, caseSensitive).project(rowFilter);
 
       ManifestEvaluator manifestEval = ManifestEvaluator.forPartitionFilter(
           partitionFilter, table().spec(), caseSensitive);
 
-      return CloseableIterable.filter(manifestIterable, manifestEval::eval);
+      return CloseableIterable.filter(manifests, manifestEval::eval);
     }
   }
 

core/src/main/java/org/apache/iceberg/BaseMetadataTableScan.java

@@ -19,9 +19,15 @@
 
 package org.apache.iceberg;
 
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.events.ScanEvent;
+import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.util.PropertyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 abstract class BaseMetadataTableScan extends BaseTableScan {
+  private static final Logger LOG = LoggerFactory.getLogger(BaseMetadataTableScan.class);
 
   protected BaseMetadataTableScan(TableOperations ops, Table table, Schema schema) {
     super(ops, table, schema);
@@ -38,4 +44,14 @@ public long targetSplitSize() {
         TableProperties.METADATA_SPLIT_SIZE_DEFAULT);
     return PropertyUtil.propertyAsLong(options(), TableProperties.SPLIT_SIZE, tableValue);
   }
+
+  /**
+   * Alternative to {@link #planFiles()}, allows exploring old snapshots even for an empty table.
+   */
+  protected CloseableIterable<FileScanTask> planFilesAllSnapshots() {
+    LOG.info("Scanning metadata table {} with filter {}.", table(), filter());
+    Listeners.notifyAll(new ScanEvent(table().name(), 0L, filter(), schema()));
+
+    return planFiles(tableOps(), snapshot(), filter(), shouldIgnoreResiduals(), isCaseSensitive(), colStats());
+  }
 }

core/src/main/java/org/apache/iceberg/DataFilesTable.java

@@ -19,7 +19,7 @@
 
 package org.apache.iceberg;
 
-import java.util.List;
+import org.apache.iceberg.io.CloseableIterable;
 
 /**
  * A {@link Table} implementation that exposes a table's data files as rows.
@@ -60,8 +60,8 @@ protected TableScan newRefinedScan(TableOperations ops, Table table, Schema sche
     }
 
     @Override
-    protected List<ManifestFile> manifests() {
-      return snapshot().dataManifests();
+    protected CloseableIterable<ManifestFile> manifests() {
+      return CloseableIterable.withNoopClose(snapshot().dataManifests());
     }
   }
 }

core/src/main/java/org/apache/iceberg/DeleteFilesTable.java

@@ -19,7 +19,7 @@
 
 package org.apache.iceberg;
 
-import java.util.List;
+import org.apache.iceberg.io.CloseableIterable;
 
 /**
  * A {@link Table} implementation that exposes a table's delete files as rows.
@@ -60,8 +60,8 @@ protected TableScan newRefinedScan(TableOperations ops, Table table, Schema sche
     }
 
     @Override
-    protected List<ManifestFile> manifests() {
-      return snapshot().deleteManifests();
+    protected CloseableIterable<ManifestFile> manifests() {
+      return CloseableIterable.withNoopClose(snapshot().deleteManifests());
     }
   }
 }

core/src/main/java/org/apache/iceberg/FilesTable.java

@@ -19,7 +19,7 @@
 
 package org.apache.iceberg;
 
-import java.util.List;
+import org.apache.iceberg.io.CloseableIterable;
 
 /**
  * A {@link Table} implementation that exposes a table's files as rows.
@@ -60,8 +60,8 @@ protected TableScan newRefinedScan(TableOperations ops, Table table, Schema sche
     }
 
     @Override
-    protected List<ManifestFile> manifests() {
-      return snapshot().allManifests();
+    protected CloseableIterable<ManifestFile> manifests() {
+      return CloseableIterable.withNoopClose(snapshot().allManifests());
     }
   }
 }

core/src/test/java/org/apache/iceberg/TestMetadataTableFilters.java

@@ -48,7 +48,9 @@ public static Object[][] parameters() {
         { MetadataTableType.DATA_FILES, 2 },
         { MetadataTableType.DELETE_FILES, 2 },
         { MetadataTableType.FILES, 1 },
-        { MetadataTableType.FILES, 2 }
+        { MetadataTableType.FILES, 2 },
+        { MetadataTableType.ALL_DATA_FILES, 1 },
+        { MetadataTableType.ALL_DATA_FILES, 2 }
     };
   }
 
@@ -88,6 +90,14 @@ public void setupTable() throws Exception {
           .addDeletes(FILE_D2_DELETES)
           .commit();
     }
+
+    if (type.equals(MetadataTableType.ALL_DATA_FILES)) {
+      // Clear all files from current snapshot to test whether 'all' Files tables scans previous files
+      table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit();  // Moves file entries to DELETED state
+      table.newDelete().deleteFromRowFilter(Expressions.alwaysTrue()).commit();  // Removes all entries
+      Assert.assertEquals("Current snapshot should be made empty",
+          0, table.currentSnapshot().allManifests().size());
+    }
   }
 
   private Table createMetadataTable() {
@@ -98,6 +108,8 @@ private Table createMetadataTable() {
         return new DataFilesTable(table.ops(), table);
       case DELETE_FILES:
         return new DeleteFilesTable(table.ops(), table);
+      case ALL_DATA_FILES:
+        return new AllDataFilesTable(table.ops(), table);
       default:
         throw new IllegalArgumentException("Unsupported metadata table type:" + type);
     }
@@ -114,6 +126,8 @@ private int expectedScanTaskCount(int partitions) {
       case DATA_FILES:
       case DELETE_FILES:
         return partitions;
+      case ALL_DATA_FILES:
+        return partitions * 2; // ScanTask for Data Manifest in DELETED and ADDED states
       default:
         throw new IllegalArgumentException("Unsupported metadata table type:" + type);
     }