Core, Spark: V4 flat-tree reads and writes prototype

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

@@ -35,6 +35,7 @@
 import org.apache.iceberg.expressions.Projections;
 import org.apache.iceberg.expressions.ResidualEvaluator;
 import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.metrics.ScanMetricsUtil;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -144,6 +145,10 @@ protected PlanningMode deletePlanningMode() {
 
   @Override
   protected CloseableIterable<ScanTask> doPlanFiles() {
+    if (TableUtil.formatVersion(table()) >= TableMetadata.MIN_FORMAT_VERSION_PARQUET_MANIFESTS) {
+      return doPlanFilesV4();
+    }
+
     Snapshot snapshot = snapshot();
 
     List<ManifestFile> deleteManifests = findMatchingDeleteManifests(snapshot);
@@ -187,6 +192,26 @@ protected CloseableIterable<ScanTask> doPlanFiles() {
     }
   }
 
+  private CloseableIterable<ScanTask> doPlanFilesV4() {
+    Snapshot snapshot = snapshot();
+    InputFile rootManifest = table().io().newInputFile(snapshot.manifestListLocation());
+    ScanTaskPlanner.Builder planner =
+        ScanTaskPlanner.builder(table().io(), rootManifest, specs(), table().location())
+            .filterData(filter())
+            .caseSensitive(isCaseSensitive())
+            .scanMetrics(scanMetrics());
+
+    if (shouldIgnoreResiduals()) {
+      planner = planner.ignoreResiduals();
+    }
+
+    if (shouldPlanWithExecutor()) {
+      planner = planner.planWith(planExecutor());
+    }
+
+    return CloseableIterable.transform(planner.build().planFiles(), task -> (ScanTask) task);
+  }
+
   @Override
   public CloseableIterable<ScanTaskGroup<ScanTask>> planTasks() {
     return TableScanUtil.planTaskGroups(

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

@@ -171,6 +171,14 @@ private void cacheManifests(FileIO fileIO) {
       throw new IllegalArgumentException("Cannot cache changes: FileIO is null");
     }
 
+    if (allManifests == null
+        && manifestListLocation != null
+        && FileFormat.fromFileName(manifestListLocation) == FileFormat.PARQUET) {
+      // A v4 flat-tree root manifest stores data entries inline rather than referencing leaf
+      // manifests, so it exposes no ManifestFiles. Scans plan directly from the root manifest.
+      this.allManifests = ImmutableList.of();
+    }
+
     if (allManifests == null && v1ManifestLocations != null) {
       // if we have a collection of manifest locations, then we need to load them here
       allManifests =

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

@@ -21,6 +21,7 @@
 import java.util.List;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
 public class DataTableScan extends BaseTableScan {
@@ -62,6 +63,34 @@ protected TableScan newRefinedScan(Table table, Schema schema, TableScanContext
 
   @Override
   public CloseableIterable<FileScanTask> doPlanFiles() {
+    if (TableUtil.formatVersion(table()) >= TableMetadata.MIN_FORMAT_VERSION_PARQUET_MANIFESTS) {
+      return doPlanFilesV4();
+    }
+
+    return doPlanFilesV3();
+  }
+
+  private CloseableIterable<FileScanTask> doPlanFilesV4() {
+    Snapshot snapshot = snapshot();
+    InputFile rootManifest = table().io().newInputFile(snapshot.manifestListLocation());
+    ScanTaskPlanner.Builder planner =
+        ScanTaskPlanner.builder(table().io(), rootManifest, specs(), table().location())
+            .filterData(filter())
+            .caseSensitive(isCaseSensitive())
+            .scanMetrics(scanMetrics());
+
+    if (shouldIgnoreResiduals()) {
+      planner = planner.ignoreResiduals();
+    }
+
+    if (shouldPlanWithExecutor()) {
+      planner = planner.planWith(planExecutor());
+    }
+
+    return planner.build().planFiles();
+  }
+
+  private CloseableIterable<FileScanTask> doPlanFilesV3() {
     Snapshot snapshot = snapshot();
 
     FileIO io = table().io();