Read metadata and protocol information from Delta checksum files

Compliant Delta writers may emit optional checksum files alongside
commits containing metadata and protocol information. Instead of
loading the latest checkpoint and replaying intervening commits (which
can be expensive, especially for large v1 checkpoints), Trino can read
the latest commit’s checksum file to obtain this information with a
single listing and small JSON read. Ref.
https://github.com/delta-io/delta/blob/master/PROTOCOL.md#version-checksum-file

If the checksum file is missing or does not contain both metadata and
protocol, we fall back to the existing Delta log scanning approach.

Behavior is gated by session property load_metadata_from_checksum_file
(defaulting to config delta.load_metadata_from_checksum_file, which
defaults to true). Internal testing reduced analysis time for large
v1-checkpoint tables from ~10s to <500ms.

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

Author: 3 people

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

@@ -200,6 +200,11 @@ 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`
+  - Use the Delta checksum metadata file (if available) to retrieve table
+    metadata and protocol entries instead of scanning the transaction log. The
+    equivalent catalog session property is `load_metadata_from_checksum_file`.
+  - `true`
 :::
 
 ### Catalog session properties
@@ -233,6 +238,10 @@ 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`
+  - Use the Delta checksum metadata file (if available) to retrieve table
+    metadata and protocol entries instead of scanning the transaction log.
+  - `true`
 :::
 
 (delta-lake-fte-support)=

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("Use checksum metadata file (if available) for metadata and protocol entry retrieval, rather than scanning the log")
+    public DeltaLakeConfig setLoadMetadataFromChecksumFile(boolean loadMetadataFromChecksumFile)
+    {
+        this.loadMetadataFromChecksumFile = loadMetadataFromChecksumFile;
+        return this;
+    }
 }

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

@@ -69,13 +69,15 @@
 import io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.ColumnMappingMode;
 import io.trino.plugin.deltalake.transactionlog.DeltaLakeSchemaSupport.UnsupportedTypeException;
 import io.trino.plugin.deltalake.transactionlog.DeltaLakeTransactionLogEntry;
+import io.trino.plugin.deltalake.transactionlog.DeltaLakeVersionChecksum;
 import io.trino.plugin.deltalake.transactionlog.MetadataEntry;
 import io.trino.plugin.deltalake.transactionlog.ProtocolEntry;
 import io.trino.plugin.deltalake.transactionlog.RemoveFileEntry;
 import io.trino.plugin.deltalake.transactionlog.TableSnapshot;
 import io.trino.plugin.deltalake.transactionlog.Transaction;
 import io.trino.plugin.deltalake.transactionlog.TransactionLogAccess;
 import io.trino.plugin.deltalake.transactionlog.TransactionLogEntries;
+import io.trino.plugin.deltalake.transactionlog.TransactionLogParser.CommitVersionChecksumFileInfo;
 import io.trino.plugin.deltalake.transactionlog.checkpoint.CheckpointWriterManager;
 import io.trino.plugin.deltalake.transactionlog.checkpoint.LastCheckpoint;
 import io.trino.plugin.deltalake.transactionlog.checkpoint.MetadataAndProtocolEntries;
@@ -162,6 +164,7 @@
 import io.trino.spi.type.VarcharType;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.time.Duration;
@@ -241,6 +244,7 @@
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.getHiveCatalogName;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isCollectExtendedStatisticsColumnStatisticsOnWrite;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isExtendedStatisticsEnabled;
+import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isLoadMetadataFromChecksumFile;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isProjectionPushdownEnabled;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isQueryPartitionFilterRequired;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isStoreTableMetadataInMetastoreEnabled;
@@ -298,8 +302,10 @@
 import static io.trino.plugin.deltalake.transactionlog.MetadataEntry.DELTA_CHANGE_DATA_FEED_ENABLED_PROPERTY;
 import static io.trino.plugin.deltalake.transactionlog.MetadataEntry.configurationForNewTable;
 import static io.trino.plugin.deltalake.transactionlog.TemporalTimeTravelUtil.findLatestVersionUsingTemporal;
+import static io.trino.plugin.deltalake.transactionlog.TransactionLogParser.findLatestCommitVersionChecksumFileInfo;
 import static io.trino.plugin.deltalake.transactionlog.TransactionLogParser.getMandatoryCurrentVersion;
 import static io.trino.plugin.deltalake.transactionlog.TransactionLogParser.readLastCheckpoint;
+import static io.trino.plugin.deltalake.transactionlog.TransactionLogParser.readVersionChecksumFile;
 import static io.trino.plugin.deltalake.transactionlog.TransactionLogUtil.getTransactionLogDir;
 import static io.trino.plugin.deltalake.transactionlog.TransactionLogUtil.getTransactionLogJsonEntryPath;
 import static io.trino.plugin.deltalake.transactionlog.checkpoint.TransactionLogTail.getEntriesFromJson;
@@ -480,6 +486,15 @@ private record QueriedTable(SchemaTableName schemaTableName, long version)
         }
     }
 
+    private record DeltaLakeTableDescriptor(long version, MetadataEntry metadataEntry, ProtocolEntry protocolEntry)
+    {
+        DeltaLakeTableDescriptor
+        {
+            requireNonNull(metadataEntry, "metadataEntry is null");
+            requireNonNull(protocolEntry, "protocolEntry is null");
+        }
+    }
+
     public DeltaLakeMetadata(
             DeltaLakeMetastore metastore,
             TransactionLogAccess transactionLogAccess,
@@ -717,27 +732,22 @@ public LocatedTableHandle getTableHandle(
 
         String tableLocation = table.location();
         TrinoFileSystem fileSystem = fileSystemFactory.create(session, table);
-        TableSnapshot tableSnapshot = getSnapshot(session, table, endVersion.map(version -> getVersion(session, fileSystem, tableLocation, version, metadataFetchingExecutor)));
 
-        MetadataAndProtocolEntries logEntries;
+        DeltaLakeTableDescriptor descriptor;
         try {
-            logEntries = transactionLogAccess.getMetadataAndProtocolEntry(session, fileSystem, tableSnapshot);
+            descriptor = loadDescriptor(session, tableName, table, fileSystem, tableLocation, startVersion, endVersion);
         }
         catch (TrinoException e) {
             if (e.getErrorCode().equals(DELTA_LAKE_INVALID_SCHEMA.toErrorCode())) {
-                return new CorruptedDeltaLakeTableHandle(tableName, table.catalogOwned(), managed, tableLocation, e);
+                return new CorruptedDeltaLakeTableHandle(tableName, table.catalogOwned(), table.managed(), tableLocation, e);
             }
             throw e;
         }
-        MetadataEntry metadataEntry = logEntries.metadata().orElse(null);
-        if (metadataEntry == null) {
-            return new CorruptedDeltaLakeTableHandle(tableName, table.catalogOwned(), managed, tableLocation, new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Metadata not found in transaction log for " + tableSnapshot.getTable()));
-        }
 
-        ProtocolEntry protocolEntry = logEntries.protocol().orElse(null);
-        if (protocolEntry == null) {
-            return new CorruptedDeltaLakeTableHandle(tableName, table.catalogOwned(), managed, tableLocation, new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Protocol not found in transaction log for " + tableSnapshot.getTable()));
-        }
+        MetadataEntry metadataEntry = descriptor.metadataEntry();
+        ProtocolEntry protocolEntry = descriptor.protocolEntry();
+        long snapshotVersion = descriptor.version();
+
         if (protocolEntry.minReaderVersion() > MAX_READER_VERSION) {
             LOG.debug("Skip %s because the reader version is unsupported: %d", tableName, protocolEntry.minReaderVersion());
             return null;
@@ -750,8 +760,8 @@ public LocatedTableHandle getTableHandle(
         verifySupportedColumnMapping(getColumnMappingMode(metadataEntry, protocolEntry));
         if (metadataScheduler.canStoreTableMetadata(session, metadataEntry.getSchemaString(), Optional.ofNullable(metadataEntry.getDescription())) &&
                 endVersion.isEmpty() &&
-                !isSameTransactionVersion(metastoreTable.get(), tableSnapshot)) {
-            tableUpdateInfos.put(tableName, new TableUpdateInfo(session, tableSnapshot.getVersion(), metadataEntry.getSchemaString(), Optional.ofNullable(metadataEntry.getDescription())));
+                !isSameTransactionVersion(metastoreTable.get(), snapshotVersion)) {
+            tableUpdateInfos.put(tableName, new TableUpdateInfo(session, snapshotVersion, metadataEntry.getSchemaString(), Optional.ofNullable(metadataEntry.getDescription())));
         }
         return new DeltaLakeTableHandle(
                 tableName.getSchemaName(),
@@ -767,10 +777,109 @@ public LocatedTableHandle getTableHandle(
                 Optional.empty(),
                 Optional.empty(),
                 Optional.empty(),
-                tableSnapshot.getVersion(),
+                snapshotVersion,
                 endVersion.isPresent());
     }
 
+    private DeltaLakeTableDescriptor loadDescriptor(ConnectorSession session, SchemaTableName tableName, DeltaMetastoreTable table, TrinoFileSystem fileSystem, String tableLocation, Optional<ConnectorTableVersion> startVersion, Optional<ConnectorTableVersion> endVersion)
+    {
+        Optional<Long> endTableVersion = endVersion.map(version -> getVersion(session, fileSystem, tableLocation, version, metadataFetchingExecutor));
+
+        if (isLoadMetadataFromChecksumFile(session)) {
+            Optional<Long> startTableVersion = startVersion.map(version -> getVersion(session, fileSystem, tableLocation, version, metadataFetchingExecutor));
+
+            Optional<DeltaLakeTableDescriptor> descriptor = loadDescriptorFromChecksum(tableName, fileSystem, tableLocation, startTableVersion, endTableVersion);
+            if (descriptor.isPresent()) {
+                return descriptor.get();
+            }
+        }
+
+        // Fall back to scanning the transaction log if checksum file reading is disabled or if the checksum file is missing
+        // or invalid
+        return loadDescriptorFromTransactionLog(session, table, fileSystem, endTableVersion);
+    }
+
+    private Optional<DeltaLakeTableDescriptor> loadDescriptorFromChecksum(
+            SchemaTableName tableName,
+            TrinoFileSystem fileSystem,
+            String tableLocation,
+            Optional<Long> startTableVersion,
+            Optional<Long> endTableVersion)
+    {
+        long latestEligibleCommit;
+
+        if (endTableVersion.isPresent()) {
+            // Optimization: we already validated the existence of endTableVersion in getVersion, so endTableVersion is
+            // definitionally the latest eligible commit. Attempt to read the latest checksum file directly without an
+            // additional list operation
+            latestEligibleCommit = endTableVersion.orElseThrow();
+        }
+        else {
+            Optional<CommitVersionChecksumFileInfo> checksumFileInfo;
+            try {
+                checksumFileInfo = findLatestCommitVersionChecksumFileInfo(fileSystem, tableLocation, startTableVersion, endTableVersion);
+            }
+            catch (IOException | UncheckedIOException e) {
+                return Optional.empty();
+            }
+
+            if (checksumFileInfo.isEmpty()) {
+                throw new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Metadata not found in transaction log for " + tableName);
+            }
+
+            CommitVersionChecksumFileInfo info = checksumFileInfo.orElseThrow();
+            if (!info.hasVersionChecksumFile()) {
+                return Optional.empty();
+            }
+
+            latestEligibleCommit = info.version();
+        }
+
+        Optional<DeltaLakeVersionChecksum> versionChecksum;
+        try {
+            versionChecksum = readVersionChecksumFile(fileSystem, tableLocation, latestEligibleCommit);
+        }
+        catch (IOException | UncheckedIOException e) {
+            throw new TrinoException(DELTA_LAKE_FILESYSTEM_ERROR, format("Failed to read checksum file for version %d of table %s", latestEligibleCommit, tableName), e);
+        }
+
+        if (versionChecksum.isEmpty()) {
+            return Optional.empty();
+        }
+
+        DeltaLakeVersionChecksum checksum = versionChecksum.orElseThrow();
+
+        MetadataEntry metadataEntry = checksum.getMetadata();
+        ProtocolEntry protocolEntry = checksum.getProtocol();
+        if (metadataEntry == null || protocolEntry == null) {
+            return Optional.empty();
+        }
+
+        return Optional.of(new DeltaLakeTableDescriptor(latestEligibleCommit, metadataEntry, protocolEntry));
+    }
+
+    private DeltaLakeTableDescriptor loadDescriptorFromTransactionLog(
+            ConnectorSession session,
+            DeltaMetastoreTable table,
+            TrinoFileSystem fileSystem,
+            Optional<Long> endTableVersion)
+    {
+        TableSnapshot tableSnapshot = getSnapshot(session, table, endTableVersion);
+        MetadataAndProtocolEntries logEntries = transactionLogAccess.getMetadataAndProtocolEntry(session, fileSystem, tableSnapshot);
+
+        MetadataEntry metadataEntry = logEntries.metadata().orElse(null);
+        if (metadataEntry == null) {
+            throw new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Metadata not found in transaction log for " + tableSnapshot.getTable());
+        }
+
+        ProtocolEntry protocolEntry = logEntries.protocol().orElse(null);
+        if (protocolEntry == null) {
+            throw new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Protocol not found in transaction log for " + tableSnapshot.getTable());
+        }
+
+        return new DeltaLakeTableDescriptor(tableSnapshot.getVersion(), metadataEntry, protocolEntry);
+    }
+
     @Override
     public ConnectorTableProperties getTableProperties(ConnectorSession session, ConnectorTableHandle tableHandle)
     {

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

@@ -75,6 +75,7 @@ public final class DeltaLakeSessionProperties
     private static final String PROJECTION_PUSHDOWN_ENABLED = "projection_pushdown_enabled";
     private static final String QUERY_PARTITION_FILTER_REQUIRED = "query_partition_filter_required";
     private static final String STORE_TABLE_METADATA = "store_table_metadata";
+    private static final String LOAD_METADATA_FROM_CHECKSUM_FILE = "load_metadata_from_checksum_file";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
@@ -226,6 +227,11 @@ public DeltaLakeSessionProperties(
                         "Require filter on partition column",
                         deltaLakeConfig.isQueryPartitionFilterRequired(),
                         false),
+                booleanProperty(
+                        LOAD_METADATA_FROM_CHECKSUM_FILE,
+                        "Use checksum metadata file for metadata and protocol entry retrieval",
+                        deltaLakeConfig.isLoadMetadataFromChecksumFile(),
+                        false),
                 booleanProperty(
                         STORE_TABLE_METADATA,
                         "Store table metadata in metastore",
@@ -344,6 +350,11 @@ public static boolean isQueryPartitionFilterRequired(ConnectorSession session)
         return session.getProperty(QUERY_PARTITION_FILTER_REQUIRED, Boolean.class);
     }
 
+    public static boolean isLoadMetadataFromChecksumFile(ConnectorSession session)
+    {
+        return session.getProperty(LOAD_METADATA_FROM_CHECKSUM_FILE, Boolean.class);
+    }
+
     public static boolean isStoreTableMetadataInMetastoreEnabled(ConnectorSession session)
     {
         return session.getProperty(STORE_TABLE_METADATA, Boolean.class);

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

@@ -213,9 +213,14 @@ public void stop()
     }
 
     public static boolean isSameTransactionVersion(Table table, TableSnapshot snapshot)
+    {
+        return isSameTransactionVersion(table, snapshot.getVersion());
+    }
+
+    public static boolean isSameTransactionVersion(Table table, long snapshotVersion)
     {
         return getLastTransactionVersion(table)
-                .map(version -> version == snapshot.getVersion())
+                .map(version -> version == snapshotVersion)
                 .orElse(false);
     }