Parallelize Parquet checkpoint reads in the Delta Lake connector

Delta uses checkpoints to represent point-in-time snapshots of the
table, e.g. the full set of active files. Most queries and write
operations must load the checkpoint before proceeding. For large tables
checkpoints can be extremely large, so checkpoint loading is important
for overall query performance

This PR introduces full support for parallel reads of checkpoint files
in Delta tables using a new `ParallelUnorderedIterator` that pulls from
several `CheckpointEntryIterator` streams in parallel. New
configuration properties:
- `delta.checkpoint-processing.v{1,2}-parallel-processing.enabled`
- `delta.checkpoint-processing.intra-file-parallel-processing.enabled`
- `delta.checkpoint-processing.intra-file-parallel-processing.split-size`
controlling parallelization for v1/v2 checkpoints, intra-file (Parquet
split) parallelization, and target split size. We reuse the existing
`delta.checkpoint-processing.parallelism` to tune parallelism

Co-authored-by: Eric Hwang <eh@openai.com>

Author: adam-richardson-openai

docs/src/main/sphinx/connector/delta-lake.md

@@ -197,9 +197,23 @@ values. Typical usage does not require you to configure them.
     is parallelized.
   - `8`
 * - `delta.checkpoint-processing.parallelism`
-  - Number of threads used for retrieving checkpoint files of each table. Currently, only 
-    retrievals of V2 Checkpoint's sidecar files are parallelized.
+  - Number of threads used for processing Parquet checkpoint files for each
+    table
   - `4`
+* - `delta.checkpoint-processing.v1-parallel-processing.enabled`
+  - Enable parallel processing for V1 checkpoints
+  - `false`
+* - `delta.checkpoint-processing.v2-parallel-processing.enabled`
+  - Enable fully parallel processing for V2 checkpoint sidecar files (by
+    default, sidecar files are opened in parallel but read serially)
+  - `false`
+* - `delta.checkpoint-processing.intra-file-parallel-processing.enabled`
+  - Enable parallel processing of individual checkpoint Parquet files by
+    subdividing into splits
+  - `false`
+* - `delta.checkpoint-processing.intra-file-parallel-processing.split-size`
+  - Goal split size for parallel processing of individual checkpoint Parquet files
+  - `128MB`
 * - `delta.load-metadata-from-checksum-file`
   - Use the Delta checksum metadata file (if available) to retrieve table
     metadata and protocol entries instead of scanning the transaction log. The

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeConfig.java

@@ -22,6 +22,7 @@
 import io.airlift.units.DataSize;
 import io.airlift.units.Duration;
 import io.airlift.units.MaxDuration;
+import io.airlift.units.MinDataSize;
 import io.airlift.units.MinDuration;
 import io.airlift.units.ThreadCount;
 import io.trino.plugin.hive.HiveCompressionOption;
@@ -95,6 +96,10 @@ public class DeltaLakeConfig
     private boolean deltaLogFileSystemCacheDisabled;
     private int metadataParallelism = 8;
     private int checkpointProcessingParallelism = 4;
+    private boolean checkpointV1ParallelProcessingEnabled;
+    private boolean checkpointV2ParallelProcessingEnabled;
+    private boolean checkpointIntraFileParallelProcessingEnabled;
+    private DataSize checkpointIntraFileParallelProcessingSplitSize = DataSize.of(128, MEGABYTE);
     private boolean loadMetadataFromChecksumFile = true;
 
     public Duration getMetadataCacheTtl()
@@ -581,14 +586,68 @@ public int getCheckpointProcessingParallelism()
         return checkpointProcessingParallelism;
     }
 
-    @ConfigDescription("Limits per table scan checkpoint files processing parallelism")
+    @ConfigDescription("Limits per table scan checkpoint file processing parallelism for checkpoint Parquet files")
     @Config("delta.checkpoint-processing.parallelism")
     public DeltaLakeConfig setCheckpointProcessingParallelism(int checkpointProcessingParallelism)
     {
         this.checkpointProcessingParallelism = checkpointProcessingParallelism;
         return this;
     }
 
+    public boolean isCheckpointV1ParallelProcessingEnabled()
+    {
+        return checkpointV1ParallelProcessingEnabled;
+    }
+
+    @Config("delta.checkpoint-processing.v1-parallel-processing.enabled")
+    @ConfigDescription("Enable parallel processing for v1 checkpoints")
+    public DeltaLakeConfig setCheckpointV1ParallelProcessingEnabled(boolean checkpointV1ParallelProcessingEnabled)
+    {
+        this.checkpointV1ParallelProcessingEnabled = checkpointV1ParallelProcessingEnabled;
+        return this;
+    }
+
+    public boolean isCheckpointV2ParallelProcessingEnabled()
+    {
+        return checkpointV2ParallelProcessingEnabled;
+    }
+
+    @Config("delta.checkpoint-processing.v2-parallel-processing.enabled")
+    @ConfigDescription("Enable fully parallel processing for v2 checkpoint sidecar files")
+    public DeltaLakeConfig setCheckpointV2ParallelProcessingEnabled(boolean checkpointV2ParallelProcessingEnabled)
+    {
+        this.checkpointV2ParallelProcessingEnabled = checkpointV2ParallelProcessingEnabled;
+        return this;
+    }
+
+    public boolean isCheckpointIntraFileParallelProcessingEnabled()
+    {
+        return checkpointIntraFileParallelProcessingEnabled;
+    }
+
+    @Config("delta.checkpoint-processing.intra-file-parallel-processing.enabled")
+    @ConfigDescription("Enable parallel processing of individual checkpoint Parquet files by subdividing them into splits")
+    public DeltaLakeConfig setCheckpointIntraFileParallelProcessingEnabled(boolean checkpointIntraFileParallelProcessingEnabled)
+    {
+        this.checkpointIntraFileParallelProcessingEnabled = checkpointIntraFileParallelProcessingEnabled;
+        return this;
+    }
+
+    @NotNull
+    @MinDataSize("1B")
+    public DataSize getCheckpointIntraFileParallelProcessingSplitSize()
+    {
+        return checkpointIntraFileParallelProcessingSplitSize;
+    }
+
+    @Config("delta.checkpoint-processing.intra-file-parallel-processing.split-size")
+    @ConfigDescription("Target split size for parallel processing of individual checkpoint Parquet files")
+    public DeltaLakeConfig setCheckpointIntraFileParallelProcessingSplitSize(DataSize checkpointIntraFileParallelProcessingSplitSize)
+    {
+        this.checkpointIntraFileParallelProcessingSplitSize = checkpointIntraFileParallelProcessingSplitSize;
+        return this;
+    }
+
     public boolean isLoadMetadataFromChecksumFile()
     {
         return loadMetadataFromChecksumFile;