Core: Refactor metadata tables and scans (#4566)

Author: aokolnychyi

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

@@ -19,12 +19,7 @@
 
 package org.apache.iceberg;
 
-import java.io.IOException;
-import org.apache.iceberg.exceptions.RuntimeIOException;
 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.util.ParallelIterable;
 
 /**
  * A {@link Table} implementation that exposes a table's valid data files as rows.
@@ -53,47 +48,25 @@ MetadataTableType metadataTableType() {
     return MetadataTableType.ALL_DATA_FILES;
   }
 
-  public static class AllDataFilesTableScan extends BaseFilesTableScan {
+  public static class AllDataFilesTableScan extends BaseAllFilesTableScan {
 
-    AllDataFilesTableScan(TableOperations ops, Table table, Schema fileSchema) {
-      super(ops, table, fileSchema, MetadataTableType.ALL_DATA_FILES);
+    AllDataFilesTableScan(TableOperations ops, Table table, Schema schema) {
+      super(ops, table, schema, MetadataTableType.ALL_DATA_FILES);
     }
 
-    private AllDataFilesTableScan(TableOperations ops, Table table, Schema schema, Schema fileSchema,
+    private AllDataFilesTableScan(TableOperations ops, Table table, Schema schema,
                                   TableScanContext context) {
-      super(ops, table, schema, fileSchema, context, MetadataTableType.ALL_DATA_FILES);
+      super(ops, table, schema, MetadataTableType.ALL_DATA_FILES, context);
     }
 
     @Override
     protected TableScan newRefinedScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
-      return new AllDataFilesTableScan(ops, table, schema, fileSchema(), context);
-    }
-
-    @Override
-    public TableScan useSnapshot(long scanSnapshotId) {
-      throw new UnsupportedOperationException("Cannot select snapshot: all_data_files is for all snapshots");
-    }
-
-    @Override
-    public TableScan asOfTime(long timestampMillis) {
-      throw new UnsupportedOperationException("Cannot select snapshot: all_data_files is for all snapshots");
-    }
-
-    @Override
-    public CloseableIterable<FileScanTask> planFiles() {
-      return super.planFilesAllSnapshots();
+      return new AllDataFilesTableScan(ops, table, schema, context);
     }
 
     @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");
-      }
+      return reachableManifests(Snapshot::dataManifests);
     }
   }
 }

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

@@ -19,19 +19,14 @@
 
 package org.apache.iceberg;
 
-import java.io.IOException;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.ManifestEntriesTable.ManifestReadTask;
 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;
 
 /**
  * A {@link Table} implementation that exposes a table's manifest entries as rows, for both delete and data files.
@@ -74,11 +69,11 @@ MetadataTableType metadataTableType() {
   private static class Scan extends BaseAllMetadataTableScan {
 
     Scan(TableOperations ops, Table table, Schema schema) {
-      super(ops, table, schema);
+      super(ops, table, schema, MetadataTableType.ALL_ENTRIES);
     }
 
     private Scan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
-      super(ops, table, schema, context);
+      super(ops, table, schema, MetadataTableType.ALL_ENTRIES, context);
     }
 
     @Override
@@ -88,34 +83,16 @@ protected TableScan newRefinedScan(TableOperations ops, Table table, Schema sche
     }
 
     @Override
-    protected String tableType() {
-      return MetadataTableType.ALL_ENTRIES.name();
-    }
+    protected CloseableIterable<FileScanTask> doPlanFiles() {
+      CloseableIterable<ManifestFile> manifests = reachableManifests(Snapshot::allManifests);
 
-    @Override
-    protected CloseableIterable<FileScanTask> planFiles(
-        TableOperations ops, Snapshot snapshot, Expression rowFilter,
-        boolean ignoreResiduals, boolean caseSensitive, boolean colStats) {
-      CloseableIterable<ManifestFile> manifests = allManifestFiles(
-          ops.current().snapshots(), context().planExecutor());
       String schemaString = SchemaParser.toJson(schema());
       String specString = PartitionSpecParser.toJson(PartitionSpec.unpartitioned());
-      Expression filter = ignoreResiduals ? Expressions.alwaysTrue() : rowFilter;
+      Expression filter = shouldIgnoreResiduals() ? Expressions.alwaysTrue() : filter();
       ResidualEvaluator residuals = ResidualEvaluator.unpartitioned(filter);
 
-      return CloseableIterable.transform(manifests, manifest -> new ManifestEntriesTable.ManifestReadTask(
-          ops.io(), manifest, schema(), schemaString, specString, residuals, ops.current().specsById()));
-    }
-  }
-
-  private static CloseableIterable<ManifestFile> allManifestFiles(
-      List<Snapshot> snapshots, ExecutorService workerPool) {
-    try (CloseableIterable<ManifestFile> iterable = new ParallelIterable<>(
-        Iterables.transform(snapshots, snapshot -> (Iterable<ManifestFile>) () -> snapshot.allManifests().iterator()),
-        workerPool)) {
-      return CloseableIterable.withNoopClose(Sets.newHashSet(iterable));
-    } catch (IOException e) {
-      throw new RuntimeIOException(e, "Failed to close parallel iterable");
+      return CloseableIterable.transform(manifests, manifest ->
+          new ManifestReadTask(table(), manifest, schema(), schemaString, specString, residuals));
     }
   }
 }

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

@@ -85,12 +85,12 @@ MetadataTableType metadataTableType() {
   public static class AllManifestsTableScan extends BaseAllMetadataTableScan {
 
     AllManifestsTableScan(TableOperations ops, Table table, Schema fileSchema) {
-      super(ops, table, fileSchema);
+      super(ops, table, fileSchema, MetadataTableType.ALL_MANIFESTS);
     }
 
     private AllManifestsTableScan(TableOperations ops, Table table, Schema schema,
                                   TableScanContext context) {
-      super(ops, table, schema, context);
+      super(ops, table, schema, MetadataTableType.ALL_MANIFESTS, context);
     }
 
     @Override
@@ -100,43 +100,27 @@ protected TableScan newRefinedScan(TableOperations ops, Table table, Schema sche
     }
 
     @Override
-    public TableScan useSnapshot(long scanSnapshotId) {
-      throw new UnsupportedOperationException("Cannot select snapshot: all_manifests is for all snapshots");
-    }
-
-    @Override
-    public TableScan asOfTime(long timestampMillis) {
-      throw new UnsupportedOperationException("Cannot select snapshot: all_manifests is for all snapshots");
-    }
-
-    @Override
-    protected String tableType() {
-      return MetadataTableType.ALL_MANIFESTS.name();
-    }
-
-    @Override
-    protected CloseableIterable<FileScanTask> planFiles(
-        TableOperations ops, Snapshot snapshot, Expression rowFilter,
-        boolean ignoreResiduals, boolean caseSensitive, boolean colStats) {
+    protected CloseableIterable<FileScanTask> doPlanFiles() {
+      FileIO io = table().io();
       String schemaString = SchemaParser.toJson(schema());
       String specString = PartitionSpecParser.toJson(PartitionSpec.unpartitioned());
       Map<Integer, PartitionSpec> specs = Maps.newHashMap(table().specs());
+      Expression filter = shouldIgnoreResiduals() ? Expressions.alwaysTrue() : filter();
+      ResidualEvaluator residuals = ResidualEvaluator.unpartitioned(filter);
 
-      return CloseableIterable.withNoopClose(Iterables.transform(ops.current().snapshots(), snap -> {
+      return CloseableIterable.withNoopClose(Iterables.transform(table().snapshots(), snap -> {
         if (snap.manifestListLocation() != null) {
-          Expression filter = ignoreResiduals ? Expressions.alwaysTrue() : rowFilter;
-          ResidualEvaluator residuals = ResidualEvaluator.unpartitioned(filter);
           DataFile manifestListAsDataFile = DataFiles.builder(PartitionSpec.unpartitioned())
-              .withInputFile(ops.io().newInputFile(snap.manifestListLocation()))
+              .withInputFile(io.newInputFile(snap.manifestListLocation()))
               .withRecordCount(1)
               .withFormat(FileFormat.AVRO)
               .build();
-          return new ManifestListReadTask(ops.io(), schema(), specs, new BaseFileScanTask(
+          return new ManifestListReadTask(io, schema(), specs, new BaseFileScanTask(
               manifestListAsDataFile, null,
               schemaString, specString, residuals));
         } else {
           return StaticDataTask.of(
-              ops.io().newInputFile(ops.current().metadataFileLocation()),
+              io.newInputFile(tableOps().current().metadataFileLocation()),
               MANIFEST_FILE_SCHEMA, schema(), snap.allManifests(),
               manifest -> ManifestsTable.manifestFileToRow(specs.get(manifest.partitionSpecId()), manifest)
           );

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

@@ -19,40 +19,56 @@
 
 package org.apache.iceberg;
 
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import org.apache.iceberg.events.Listeners;
+import org.apache.iceberg.events.ScanEvent;
 import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.base.Function;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.ParallelIterable;
+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);
+  BaseAllMetadataTableScan(TableOperations ops, Table table, Schema schema, MetadataTableType tableType) {
+    super(ops, table, schema, tableType);
   }
 
-  BaseAllMetadataTableScan(TableOperations ops, Table table, Schema schema, TableScanContext context) {
-    super(ops, table, schema, context);
+  BaseAllMetadataTableScan(TableOperations ops, Table table, Schema schema, MetadataTableType tableType,
+                           TableScanContext context) {
+    super(ops, table, schema, tableType, context);
   }
 
-  /**
-   * Type of scan being performed, such as {@link MetadataTableType#ALL_DATA_FILES} when scanning
-   * a table's {@link org.apache.iceberg.AllDataFilesTable}.
-   * <p>
-   * Used for logging and error messages.
-   */
-  protected abstract String tableType();
-
   @Override
-  public TableScan appendsBetween(long fromSnapshotId, long toSnapshotId) {
-    throw new UnsupportedOperationException(
-        String.format("Cannot incrementally scan table of type %s", tableType()));
+  public TableScan useSnapshot(long scanSnapshotId) {
+    throw new UnsupportedOperationException("Cannot select snapshot in table: " + tableType());
   }
 
   @Override
-  public TableScan appendsAfter(long fromSnapshotId) {
-    throw new UnsupportedOperationException(
-        String.format("Cannot incrementally scan table of type %s", tableType()));
+  public TableScan asOfTime(long timestampMillis) {
+    throw new UnsupportedOperationException("Cannot select snapshot in table: " + tableType());
   }
 
   @Override
   public CloseableIterable<FileScanTask> planFiles() {
-    return super.planFilesAllSnapshots();
+    LOG.info("Scanning metadata table {} with filter {}.", table(), filter());
+    Listeners.notifyAll(new ScanEvent(table().name(), 0L, filter(), schema()));
+
+    return doPlanFiles();
+  }
+
+  protected CloseableIterable<ManifestFile> reachableManifests(Function<Snapshot, Iterable<ManifestFile>> toManifests) {
+    Iterable<Snapshot> snapshots = table().snapshots();
+    Iterable<Iterable<ManifestFile>> manifestIterables = Iterables.transform(snapshots, toManifests);
+
+    try (CloseableIterable<ManifestFile> iterable = new ParallelIterable<>(manifestIterables, planExecutor())) {
+      return CloseableIterable.withNoopClose(Sets.newHashSet(iterable));
+    } catch (IOException e) {
+      throw new UncheckedIOException("Failed to close parallel iterable", e);
+    }
   }
 }