Read metadata and protocol information from Delta checksum files

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

@@ -201,6 +201,13 @@ values. Typical usage does not require you to configure them.
   - Number of threads used for retrieving checkpoint files of each table. Currently, only 
     retrievals of V2 Checkpoint's sidecar files are parallelized.
   - `4`
+* - `delta.load-metadata-from-checksum-file`
+  - Speed up query planning by reading table metadata and protocol
+    entries from the Delta version checksum file (`<version>.crc`) when
+    available. Falls back to scanning the transaction log if the checksum
+    file is missing, incomplete, or malformed. The equivalent catalog
+    session property is `load_metadata_from_checksum_file`.
+  - `true`
 :::
 
 ### Catalog session properties
@@ -234,6 +241,12 @@ The following table describes {ref}`catalog session properties
   - Read only projected fields from row columns while performing `SELECT`
     queries.
   - `true`
+* - `load_metadata_from_checksum_file`
+  - Speed up query planning by reading table metadata and protocol
+    entries from the Delta version checksum file (`<version>.crc`) when
+    available. Falls back to scanning the transaction log if the checksum
+    file is missing, incomplete, or malformed.
+  - `true`
 :::
 
 (delta-lake-fte-support)=

lib/trino-filesystem-azure/src/test/java/io/trino/filesystem/azure/AbstractTestAzureFileSystem.java

@@ -22,8 +22,11 @@
 import com.azure.storage.file.datalake.DataLakeServiceClientBuilder;
 import com.azure.storage.file.datalake.models.PathItem;
 import com.azure.storage.file.datalake.options.DataLakePathDeleteOptions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.io.Closer;
 import io.opentelemetry.api.OpenTelemetry;
 import io.trino.filesystem.AbstractTestTrinoFileSystem;
+import io.trino.filesystem.FileIterator;
 import io.trino.filesystem.Location;
 import io.trino.filesystem.TrinoFileSystem;
 import io.trino.filesystem.TrinoInput;
@@ -291,4 +294,27 @@ public void testDirectoryExists()
         }
         super.testDirectoryExists();
     }
+
+    @Test
+    void testListFilesStartingFromConsecutiveSlashesInLocation()
+            throws IOException
+    {
+        // ADLS Gen2 hierarchical canonicalizes runs of slashes; verifies
+        // EmulatedListFilesStartingFromIterator falls back to the slash-collapsed prefix.
+        if (!isHierarchical()) {
+            abort("Only ADLS Gen2 hierarchical canonicalizes `//` and routes through EmulatedListFilesStartingFromIterator");
+        }
+        try (Closer closer = Closer.create()) {
+            Location file1 = createBlob(closer, "level0/level1-file1");
+            Location file2 = createBlob(closer, "level0/level1-file2");
+
+            Location doubledSlash = createLocation("level0").appendSuffix("//");
+            ImmutableList.Builder<Location> builder = ImmutableList.builder();
+            FileIterator iterator = getFileSystem().listFilesStartingFrom(doubledSlash, "level1-file1");
+            while (iterator.hasNext()) {
+                builder.add(iterator.next().location());
+            }
+            assertThat(builder.build()).containsExactlyInAnyOrder(file1, file2);
+        }
+    }
 }

lib/trino-filesystem/src/main/java/io/trino/filesystem/EmulatedListFilesStartingFromIterator.java

@@ -15,15 +15,19 @@
 
 import java.io.IOException;
 import java.util.NoSuchElementException;
+import java.util.regex.Pattern;
 
-import static com.google.common.base.Preconditions.checkState;
+import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 
 public final class EmulatedListFilesStartingFromIterator
         implements FileIterator
 {
+    private static final Pattern CONSECUTIVE_SLASHES = Pattern.compile("/+");
+
     private final FileIterator delegate;
     private final String locationPath;
+    private final String collapsedLocationPath;
     private final String startingFrom;
     private FileEntry nextEntry;
 
@@ -33,6 +37,7 @@ public EmulatedListFilesStartingFromIterator(FileIterator delegate, Location loc
 
         String locationPath = location.path();
         this.locationPath = (locationPath.isEmpty() || locationPath.endsWith("/")) ? locationPath : locationPath + "/";
+        this.collapsedLocationPath = CONSECUTIVE_SLASHES.matcher(this.locationPath).replaceAll("/");
 
         this.startingFrom = requireNonNull(startingFrom, "startingFrom is null");
     }
@@ -69,9 +74,22 @@ private void loadNextEntry()
         while (delegate.hasNext()) {
             FileEntry entry = delegate.next();
             String entryPath = entry.location().path();
-            checkState(entryPath.startsWith(locationPath), "Expected listed file to start with directory path '%s': %s", locationPath, entry.location());
 
-            String entryTail = entryPath.substring(locationPath.length());
+            // LocalFileSystem, AlluxioFileSystem, and ADLS Gen2 hierarchical canonicalize runs of
+            // slashes in returned paths. Try the original prefix first to preserve blob-store keys
+            // where `//` is meaningful; fall back to the slash-collapsed form.
+            String prefix;
+            if (entryPath.startsWith(locationPath)) {
+                prefix = locationPath;
+            }
+            else if (entryPath.startsWith(collapsedLocationPath)) {
+                prefix = collapsedLocationPath;
+            }
+            else {
+                throw new IllegalStateException(format("Expected listed file to start with directory path '%s': %s", locationPath, entry.location()));
+            }
+
+            String entryTail = entryPath.substring(prefix.length());
             if (entryTail.compareTo(startingFrom) >= 0) {
                 nextEntry = entry;
                 return;

lib/trino-filesystem/src/test/java/io/trino/filesystem/TestEmulatedListFilesStartingFromIterator.java

@@ -191,6 +191,32 @@ void testListFilesStartingFromDoubleSlashPathComponent()
                         Location.of("file:///double/a"));
     }
 
+    @Test
+    void testListFilesStartingFromHierarchicalLocationNormalization()
+            throws IOException
+    {
+        // FS canonicalizes `//` to `/`; iterator falls back to slash-collapsed prefix.
+        assertThat(listFilesStartingFrom(
+                Location.of("abfs://container@account.dfs.core.windows.net/dir//sub/_delta_log/"),
+                "00000000000000000000",
+                List.of(
+                        entry("abfs://container@account.dfs.core.windows.net/dir/sub/_delta_log/00000000000000000000.json"),
+                        entry("abfs://container@account.dfs.core.windows.net/dir/sub/_delta_log/00000000000000000001.checkpoint.parquet"))))
+                .containsExactly(
+                        Location.of("abfs://container@account.dfs.core.windows.net/dir/sub/_delta_log/00000000000000000000.json"),
+                        Location.of("abfs://container@account.dfs.core.windows.net/dir/sub/_delta_log/00000000000000000001.checkpoint.parquet"));
+
+        // startingFrom filtering still applies to the slash-collapsed remainder.
+        assertThat(listFilesStartingFrom(
+                Location.of("abfs://container@account.dfs.core.windows.net/dir//sub/_delta_log/"),
+                "00000000000000000001",
+                List.of(
+                        entry("abfs://container@account.dfs.core.windows.net/dir/sub/_delta_log/00000000000000000000.json"),
+                        entry("abfs://container@account.dfs.core.windows.net/dir/sub/_delta_log/00000000000000000001.checkpoint.parquet"))))
+                .containsExactly(
+                        Location.of("abfs://container@account.dfs.core.windows.net/dir/sub/_delta_log/00000000000000000001.checkpoint.parquet"));
+    }
+
     @Test
     void testListFilesStartingFromIncludesAllNonAsciiFilenames()
             throws IOException

lib/trino-filesystem/src/test/java/io/trino/filesystem/local/TestLocalFileSystem.java

@@ -13,7 +13,10 @@
  */
 package io.trino.filesystem.local;
 
+import com.google.common.collect.ImmutableList;
+import com.google.common.io.Closer;
 import io.trino.filesystem.AbstractTestTrinoFileSystem;
+import io.trino.filesystem.FileIterator;
 import io.trino.filesystem.Location;
 import io.trino.filesystem.TrinoFileSystem;
 import org.junit.jupiter.api.AfterAll;
@@ -119,6 +122,26 @@ protected void verifyFileSystemIsEmpty()
         }
     }
 
+    @Test
+    void testListFilesStartingFromConsecutiveSlashesInLocation()
+            throws IOException
+    {
+        // LocalFileSystem (Java NIO) canonicalizes runs of slashes; verifies
+        // EmulatedListFilesStartingFromIterator falls back to the slash-collapsed prefix.
+        try (Closer closer = Closer.create()) {
+            Location file1 = createBlob(closer, "level0/level1-file1");
+            Location file2 = createBlob(closer, "level0/level1-file2");
+
+            Location doubledSlash = createLocation("level0").appendSuffix("//");
+            ImmutableList.Builder<Location> builder = ImmutableList.builder();
+            FileIterator iterator = getFileSystem().listFilesStartingFrom(doubledSlash, "level1-file1");
+            while (iterator.hasNext()) {
+                builder.add(iterator.next().location());
+            }
+            assertThat(builder.build()).containsExactlyInAnyOrder(file1, file2);
+        }
+    }
+
     @Test
     void testPathsOutOfBounds()
     {

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

@@ -95,6 +95,7 @@ public class DeltaLakeConfig
     private boolean deltaLogFileSystemCacheDisabled;
     private int metadataParallelism = 8;
     private int checkpointProcessingParallelism = 4;
+    private boolean loadMetadataFromChecksumFile = true;
 
     public Duration getMetadataCacheTtl()
     {
@@ -587,4 +588,17 @@ public DeltaLakeConfig setCheckpointProcessingParallelism(int checkpointProcessi
         this.checkpointProcessingParallelism = checkpointProcessingParallelism;
         return this;
     }
+
+    public boolean isLoadMetadataFromChecksumFile()
+    {
+        return loadMetadataFromChecksumFile;
+    }
+
+    @Config("delta.load-metadata-from-checksum-file")
+    @ConfigDescription("Read table metadata and protocol from the Delta version checksum file when available, falling back to the transaction log")
+    public DeltaLakeConfig setLoadMetadataFromChecksumFile(boolean loadMetadataFromChecksumFile)
+    {
+        this.loadMetadataFromChecksumFile = loadMetadataFromChecksumFile;
+        return this;
+    }
 }