[Kernel][LogReplay] Make a single read request for all checkpoint files #2701

vkorukanti · 2024-02-29T17:47:08Z

Description

Currently, the kernel-api reads one file (either checkpoint or commit file) at a time. Once the file is fully read, then the next file is read request is issued. This makes reading large checkpoints split over multiple files slower. Instead kernel-api could issue read requests for all checkpoint files at once (in case of multi-part checkpoints) using the ParquetHandler.readParquetFiles and let the implementations of the ParquetHandler prefetch or using multiple threads to read the checkpoint parts concurrently.

This PR makes the change to kernel-api to issue one read request for all checkpoint files that need to be read for state reconstructions.

Resolves #2668
Resolves #1965

How was this patch tested?

Existing tests and a benchmark with a test only parallel parquet reader. Here are the sample benchmark results with the test only parallel Parquet reader. Score tells the average time to construct the Delta table state. parallelReaderCount indicates the number of parallel Parquet reading threads used.

Benchmark                                     (parallelReaderCount)  Mode  Cnt     Score     Error  Units
BenchmarkParallelCheckpointReading.benchmark                      0  avgt    5  1565.520 ±  20.551  ms/op
BenchmarkParallelCheckpointReading.benchmark                      1  avgt    5  1064.850 ±  19.699  ms/op
BenchmarkParallelCheckpointReading.benchmark                      2  avgt    5   785.918 ± 176.285  ms/op
BenchmarkParallelCheckpointReading.benchmark                      4  avgt    5   729.487 ±  51.470  ms/op
BenchmarkParallelCheckpointReading.benchmark                     10  avgt    5   693.757 ±  41.252  ms/op
BenchmarkParallelCheckpointReading.benchmark                     20  avgt    5   702.656 ±  19.145  ms/op

vkorukanti · 2024-03-11T20:56:18Z

connectors/golden-tables/src/test/scala/io/delta/golden/GoldenTables.scala

@@ -378,6 +378,7 @@ class GoldenTables extends QueryTest with SharedSparkSession {

    val commitInfoFile = CommitInfo(
      version = Some(0L),
+      inCommitTimestamp = None,


This is broken on current master.

what does this mean?

there was a commit in delta-spark that updated the argument list. Our current CI setup just runs the delta-spark tests and not its dependencies. We need to revisit our CI trigger job to make sure all run tests for all dependent modules.

I can make a separate PR for this change if you would like.

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java

tdas · 2024-03-11T21:17:32Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java

@@ -149,23 +151,22 @@ private void tryEnsureNextActionsIterIsReady() {
    }

    /**
-     * Get the next file from `filesIter` (.json or .checkpoint.parquet), contextualize it
+     * Get the next file from `filesList` (.json or .checkpoint.parquet), contextualize it


do we still contextualize?

No we don't. Updating the comment.

tdas · 2024-03-11T21:23:25Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java

+                // We can not read multiple JSON files in parallel (like the checkpoint files),
+                // because each one has a different version, and we need to associate the version
+                // with actions read from the JSON file for further optimizations later on.
+
                final CloseableIterator<ColumnarBatch> dataIter =
                    tableClient.getJsonHandler().readJsonFiles(


is this an optimization that we should consider (not in this PR)? that the returned columnar batches has the file status of the file from which it was read? that would allow parallel json reads.

tdas · 2024-03-11T21:29:41Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java

-                // Read that file
+                // We can not read multiple JSON files in parallel (like the checkpoint files),
+                // because each one has a different version, and we need to associate the version
+                // with actions read from the JSON file for further optimizations later on.


nit: what optimizations need the version for each columnar batch?

It is basically the faster metadata and protocol loading with a snapshot hint. The optimizations made for the Flink faster job start. Added in the comments.

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java

tdas · 2024-03-11T21:38:59Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java

+        }
+
+        // Remove the files from the list
+        filesList.removeAll(checkpointFiles);


why do you need removeAll when you are already doing filesList.pop()?

not needed.

tdas · 2024-03-11T22:59:39Z

...ts/src/test/java/io/delta/kernel/defaults/benchmarks/BenchmarkParallelCheckpointReading.java

+ * $ build/sbt
+ * sbt:delta> project kernelDefaults
+ * sbt:delta> set fork in run := true
+ * sbt:delta> test:runMain io.delta.kernel.defaults.benchmarks.BenchmarkParallelCheckpointReading


can you add the results you got here. so that its in the code rather than in the github PR which will be harder to find when browsing code.

add some specs of your machine as well

tdas

just one comment.

allisonport-db

Minor comments LGTM

allisonport-db · 2024-03-11T23:09:47Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java

+            FileStatus checkpointFile,
+            long version) {
+
+        // Filter out all the files that are not part of the same checkpoint


What does this mean?

updated the comment. It should have been: // Find the contiguous parquet files that are part of the same checkpoint

allisonport-db · 2024-03-11T23:11:39Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java

+
+        FileStatus peek = filesList.peek();
+        while (peek != null &&
+                peek.getPath().endsWith(".parquet") &&


Can we use FileNames.isCheckpointFile here instead?

updated here and the existing code.

...ts/src/test/java/io/delta/kernel/defaults/benchmarks/BenchmarkParallelCheckpointReading.java

allisonport-db · 2024-03-11T23:30:23Z

...ts/src/test/java/io/delta/kernel/defaults/benchmarks/BenchmarkParallelCheckpointReading.java

+        ScanBuilder scanBuilder = snapshot.getScanBuilder(tableClient);
+        Scan scan = scanBuilder.build();
+
+        Row row = scan.getScanState(tableClient);


This is fine if it's to mimic the real case but could you add a comment explaining what we're doing in this block/the intention?

This is just to mimic the real case. Added comments.

vkorukanti added the kernel label Feb 29, 2024

vkorukanti added this to the 3.2.0 milestone Mar 4, 2024

vkorukanti force-pushed the multiFileReadRequest branch from f8e92d9 to df6eacc Compare March 6, 2024 18:29

vkorukanti requested review from allisonport-db and scottsand-db March 6, 2024 18:31

vkorukanti force-pushed the multiFileReadRequest branch 2 times, most recently from a0a53cf to 844a719 Compare March 11, 2024 18:40

[Kernel][LogReplay] Make a single read request for all checkpoint files

83c9e69

vkorukanti force-pushed the multiFileReadRequest branch from 844a719 to 83c9e69 Compare March 11, 2024 20:55

vkorukanti commented Mar 11, 2024

View reviewed changes

tdas reviewed Mar 11, 2024

View reviewed changes

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java Show resolved Hide resolved

tdas reviewed Mar 11, 2024

View reviewed changes

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java Show resolved Hide resolved

tdas reviewed Mar 11, 2024

View reviewed changes

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java Outdated Show resolved Hide resolved

tdas reviewed Mar 11, 2024

View reviewed changes

review

e586ef3

vkorukanti force-pushed the multiFileReadRequest branch from b5922b2 to e586ef3 Compare March 11, 2024 22:25

tdas reviewed Mar 11, 2024

View reviewed changes

tdas approved these changes Mar 11, 2024

View reviewed changes

update/fix comments

e7c82aa

allisonport-db approved these changes Mar 11, 2024

View reviewed changes

vkorukanti added 2 commits March 11, 2024 16:53

review

93682e5

Merge branch 'master' into multiFileReadRequest

121a928

vkorukanti merged commit d0477bb into delta-io:master Mar 12, 2024
7 of 8 checks passed

vkorukanti deleted the multiFileReadRequest branch May 9, 2024 02:42

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Kernel][LogReplay] Make a single read request for all checkpoint files #2701

[Kernel][LogReplay] Make a single read request for all checkpoint files #2701

vkorukanti commented Feb 29, 2024 •

edited

vkorukanti Mar 11, 2024

tdas Mar 11, 2024

vkorukanti Mar 11, 2024

tdas Mar 11, 2024

vkorukanti Mar 11, 2024

tdas Mar 11, 2024 •

edited

tdas Mar 11, 2024

vkorukanti Mar 11, 2024

tdas Mar 11, 2024

vkorukanti Mar 11, 2024

tdas Mar 11, 2024

tdas left a comment

allisonport-db left a comment

allisonport-db Mar 11, 2024

vkorukanti Mar 11, 2024

allisonport-db Mar 11, 2024

vkorukanti Mar 11, 2024

allisonport-db Mar 11, 2024

allisonport-db Mar 11, 2024

vkorukanti Mar 11, 2024

[Kernel][LogReplay] Make a single read request for all checkpoint files #2701

[Kernel][LogReplay] Make a single read request for all checkpoint files #2701

Conversation

vkorukanti commented Feb 29, 2024 • edited

Description

How was this patch tested?

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

tdas Mar 11, 2024 • edited

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

tdas left a comment

Choose a reason for hiding this comment

allisonport-db 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

vkorukanti commented Feb 29, 2024 •

edited

tdas Mar 11, 2024 •

edited