Stream large transaction log jsons instead of storing in-memory #24491

raunaqmorarka · 2024-12-16T18:55:54Z

Description

Operations fetching metadata and protocol entries can skip reading the rest of the json file after those entries are found

Additional context and related issues

On a example transaction log json of 1.5GB, the time taken for simple operations
like register table, DESCRIBE and SELECTs which don't use table statistics (or any read with set session delta.statistics_enabled=false) reduces from 18s to under 1s on local machine.
Such large transaction log jsons were observed to have been produced by CLONE operation from Apache Spark.

Release notes

( ) This is not user-visible or is docs only, and no release notes are required.
( ) Release notes are required. Please propose a release note for me.
(x) Release notes are required, with the following suggested text:

## Delta Lake
* Improve performance of queries on tables with large transaction log jsons. ({issue}`24491`)

Pluies · 2024-12-16T19:38:56Z

@raunaqmorarka very interesting! I've skimmed the code, would you consider this an alternative approach to the metadata/protocol caching we were trying to push in #20437, or more as a separate improvement? I assume they'll impact the same sort of time-consuming operations.

(Asking as I'm keen on getting out fork closer to upstream; it doesn't really matter which approach ends up being used as long as performance improves 👍 )

raunaqmorarka · 2024-12-17T16:38:14Z

@raunaqmorarka very interesting! I've skimmed the code, would you consider this an alternative approach to the metadata/protocol caching we were trying to push in #20437, or more as a separate improvement? I assume they'll impact the same sort of time-consuming operations.

(Asking as I'm keen on getting out fork closer to upstream; it doesn't really matter which approach ends up being used as long as performance improves 👍 )

Thanks for pointing out that PR, I hadn't looked at it before. For me the priority was to deal gracefully with transaction log jsons that are GBs in size. I've tweaked the PR a bit to be better about caching metadata/protocol entries. Feel free to try this out on your workloads or add review comments.

findinpath

Looks promissing to me.

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

...src/main/java/io/trino/plugin/deltalake/transactionlog/checkpoint/TransactionLogEntries.java

findinpath · 2024-12-18T18:44:44Z

...src/main/java/io/trino/plugin/deltalake/transactionlog/checkpoint/TransactionLogEntries.java

+     * Returns a stream of DeltaLakeTransactionLogEntry
+     * Caller has the responsibility to close this stream as it potentially holds an open file
+     */
+    public Stream<DeltaLakeTransactionLogEntry> getEntries(TrinoFileSystem fileSystem)


Consider passing inputFile from the constructor for eventually creating the stream instead of specifying again the TrinoFileSystem.

this would likely make the interface of DeltaLakeCommitSummary cleaner.

TransactionLogEntries ends up being a part of TableSnapshot cache which is used across queries. So I don't think it would be a good idea to hold on to inputFile created from one query's context to other queries.

findinpath · 2024-12-18T19:27:05Z

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

@@ -170,7 +171,7 @@ private static List<Transaction> loadNewTailBackward(
        boolean endOfHead = false;

        while (!endOfHead) {
-            Optional<List<DeltaLakeTransactionLogEntry>> results = getEntriesFromJson(entryNumber, transactionLogDir, fileSystem);
+            Optional<TransactionLogEntries> results = getEntriesFromJson(entryNumber, transactionLogDir, fileSystem, DEFAULT_TRANSACTION_LOG_MAX_CACHED_SIZE);


you could maybe retrieve through TransactionLogAccess the transactionLogMaxCachedSize instead of using hardcoded values.
I'm refferring to DEFAULT_TRANSACTION_LOG_MAX_CACHED_SIZE

Same for DeltaLakeMetadata

I don't think its useful to cache transaction log in those contexts, so i've just set it to 0 bytes to be clearer

findinpath · 2024-12-18T19:31:13Z

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeConfig.java

@@ -84,6 +85,7 @@ public void testExplicitPropertyMappings()
        Map<String, String> properties = ImmutableMap.<String, String>builder()
                .put("delta.metadata.cache-ttl", "10m")
                .put("delta.metadata.cache-max-retained-size", "1GB")
+                .put("delta.transaction-log.max-cached-file-size", "1MB")


Do we have a test where this value is small enough to actually cause the dynamic retrieval of the transaction log entries ?
I'm looking to have TransactionLogEntries code path tested when cachedEntries is not present.

I've used delta.transaction-log.max-cached-file-size=0B in TestDeltaLakeBasic

findinpath · 2024-12-18T19:36:01Z

...ain/java/io/trino/plugin/deltalake/transactionlog/checkpoint/MetadataAndProtocolEntries.java

+import static io.airlift.slice.SizeOf.instanceSize;
+import static io.airlift.slice.SizeOf.sizeOf;
+
+public record MetadataAndProtocolEntries(Optional<MetadataEntry> metadata, Optional<ProtocolEntry> protocol)


MetadataAndProtocolEntries -> TableDescriptorEntries

I'd rather make it clearer that it's just metadata and protocol entries, a more generic name would imply it could be something more which is not clear to me.

findinpath · 2024-12-18T19:37:40Z

...ain/java/io/trino/plugin/deltalake/transactionlog/checkpoint/MetadataAndProtocolEntries.java

+        this(Optional.ofNullable(metadata), Optional.ofNullable(protocol));
+    }
+
+    public Stream<Object> stream()


Not quite sure whether we need this method - it is used only in a single place.
Probably simple getters for metadata and protocol would be more straightforward.

This method makes the calling code simpler

findinpath · 2024-12-18T19:40:31Z

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

@@ -427,7 +442,7 @@ public static ImmutableList<DeltaLakeColumnMetadata> columnsWithStats(List<Delta
                .collect(toImmutableList());
    }

-    private Stream<AddFileEntry> activeAddEntries(Stream<DeltaLakeTransactionLogEntry> checkpointEntries, List<Transaction> transactions)
+    private Stream<AddFileEntry> activeAddEntries(Stream<DeltaLakeTransactionLogEntry> checkpointEntries, List<Transaction> transactions, TrinoFileSystem fileSystem)


To investigate whether we can do without passing fileSystem given that TransactionLogEntries holds already a pointer to TrinoInputFile

See #24491 (comment)

findinpath · 2024-12-18T19:48:07Z

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

-                fileSystemFactory.create(session),
-                fileFormatDataSourceStats)) {
-            return entries.collect(toImmutableMap(Object::getClass, Function.identity(), (first, second) -> second));
+        if (tableSnapshot.getCachedMetadata().isEmpty() || tableSnapshot.getCachedProtocol().isEmpty()) {


There's room for improvement here.
If we're looking only for metadata & protocol, we should be reading backwards from the highest transaction log towards the checkpoint.

In case you apply this request, please also test your algorithm via TestDeltaLakeFileOperations as well to ensure that e.g., the checkpoint is not read when doing some table changes in the subsequent transaction logs for a SHOW CREATE TABLE statement.

Yes, that would be better, but I'd like to limit scope of this PR. The same improvement could apply to getMetadataEntry and getProtocolEntry as well. You can file a GH issue about it for tracking.

...n/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/TableSnapshot.java

...ain/java/io/trino/plugin/deltalake/transactionlog/checkpoint/MetadataAndProtocolEntries.java

Operations fetching metadata and protocol entries can skip reading the rest of the json file after those entries are found

cla-bot bot added the cla-signed label Dec 16, 2024

github-actions bot added the delta-lake Delta Lake connector label Dec 16, 2024

raunaqmorarka added the performance label Dec 16, 2024

raunaqmorarka requested review from ebyhr, Pluies, jkylling and findinpath December 16, 2024 18:56

raunaqmorarka force-pushed the raunaq/delta-stream-log branch 2 times, most recently from fd27902 to 05a9faa Compare December 17, 2024 11:38

findinpath reviewed Dec 18, 2024

View reviewed changes

raunaqmorarka force-pushed the raunaq/delta-stream-log branch 2 times, most recently from cc4cc3a to d714365 Compare December 23, 2024 08:09

raunaqmorarka requested a review from findinpath December 23, 2024 08:33

ebyhr approved these changes Dec 23, 2024

View reviewed changes

Stream large transaction log jsons instead of storing in-memory

0301c34

Operations fetching metadata and protocol entries can skip reading the rest of the json file after those entries are found

raunaqmorarka force-pushed the raunaq/delta-stream-log branch from d714365 to 0301c34 Compare December 23, 2024 18:14

raunaqmorarka merged commit f888217 into master Dec 24, 2024
57 checks passed

raunaqmorarka deleted the raunaq/delta-stream-log branch December 24, 2024 09:31

github-actions bot added this to the 469 milestone Dec 24, 2024

findinpath mentioned this pull request Dec 24, 2024

Delta: Read the transaction log files backwards for finding the latest metadata & protocol entries #24570

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Stream large transaction log jsons instead of storing in-memory #24491

Stream large transaction log jsons instead of storing in-memory #24491

raunaqmorarka commented Dec 16, 2024 •

edited

Loading

Pluies commented Dec 16, 2024 •

edited

Loading

raunaqmorarka commented Dec 17, 2024

findinpath left a comment

findinpath Dec 18, 2024

findinpath Dec 18, 2024

raunaqmorarka Dec 23, 2024

findinpath Dec 18, 2024

raunaqmorarka Dec 23, 2024

findinpath Dec 18, 2024

raunaqmorarka Dec 23, 2024

findinpath Dec 18, 2024

raunaqmorarka Dec 23, 2024

findinpath Dec 18, 2024

raunaqmorarka Dec 23, 2024

findinpath Dec 18, 2024

raunaqmorarka Dec 23, 2024

findinpath Dec 18, 2024 •

edited

Loading

raunaqmorarka Dec 23, 2024

findinpath Dec 24, 2024

Stream large transaction log jsons instead of storing in-memory #24491

Stream large transaction log jsons instead of storing in-memory #24491

Conversation

raunaqmorarka commented Dec 16, 2024 • edited Loading

Description

Additional context and related issues

Release notes

Pluies commented Dec 16, 2024 • edited Loading

raunaqmorarka commented Dec 17, 2024

findinpath left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

findinpath Dec 18, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

raunaqmorarka commented Dec 16, 2024 •

edited

Loading

Pluies commented Dec 16, 2024 •

edited

Loading

findinpath Dec 18, 2024 •

edited

Loading