[Kernel] Add kernel support for v2 checkpoints #2826

chirag-s-db · 2024-03-29T22:44:58Z

Which Delta project/connector is this regarding?

Description

Add support for V2 checkpoints. When reconstructing the LogSegment of a table at a given version, check if the checkpoint file to be read is a checkpoint manifest. If it is, include the sidecar files referenced by that manifest in the LogSegment checkpoint files. See #2232

How was this patch tested?

See changes to LogReplaySuite, SnapshotManagementSuite, CheckpointerSuite, FileNamesSuite, and CheckpointInstanceSuite.

Does this PR introduce any user-facing changes?

No.

chirag-s-db · 2024-04-01T15:54:40Z

kernel/kernel-api/src/test/scala/io/delta/kernel/CheckpointInstanceSuite.scala


    // invalid checkpoints
    intercept[RuntimeException] {
      new CheckpointInstance(
        new Path(FAKE_DELTA_LOG_PATH,
-          "00000000000000000010.checkpoint.0000000002.parquet").toString)


Note: Had to change to make in valid - previously was valid V2 checkpoint path

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/DeltaHistoryManagerSuite.scala

chirag-s-db · 2024-04-01T17:17:52Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/snapshot/SnapshotManager.java

@@ -598,6 +629,22 @@ protected Optional<LogSegment> getLogSegmentForVersion(
                );
                throw new IllegalStateException(msg);
            }
+
+            // Reading sidecars only applies for v2 checkpoints.
+            if (newCheckpoint.format == CheckpointInstance.CheckpointFormat.V2) {


Note: For a V2 checkpoint instance, do not include the compatibility file in the checkpoints in the LogSegment. Only include checkpoint manifest + sidecar files.

This can potentially lead to incorrect results being served to the user:

Table A only has one checkpoint which happens to be a compatibility V2 checkpoint. Because of https://github.com/delta-io/delta/pull/2826/files#diff-12397663510cec22feeea64e8e3f0ddee114de86025e00495b02ab89cc3d2d01R69, newCheckpoint.format will be single_checkpoint.

Since we only read sidecars if format == v2, no sidecars will be read for this checkpoint and only the non-file actions will be read.
This could result in the construction of a snapshot which looks legal but is missing references to many AddFiles that are present in those sidecars.

This should be fixed - now, we read the highest version checkpoint, preferring V2 over V1 checkpoints if multiple exist at the same version.

…nto kernel-v2-checkpoint

dhruvarya-db · 2024-04-01T18:10:53Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/CheckpointInstance.java


        if (pathParts.length == 3 && pathParts[1].equals("checkpoint") &&
                pathParts[2].equals("parquet")) {
            // Classic checkpoint 00000000000000000010.checkpoint.parquet
            this.version = Long.parseLong(pathParts[0]);
            this.numParts = Optional.empty();
+            this.format = CheckpointFormat.SINGLE_PART;


With V2 Checkpoints, a checkpoint that appears to be a single_part checkpoint can actually be a V2 checkpoint. The only way to be sure about whether the checkpoint is V2 or classic is to read the checkpoint and look for the CheckpointManifest ~~option~~ action.

Got it, fixed.

vkorukanti

Thanks for working on this!

few questions:

How does the ActionsIterator.java handle the v2 checkpoint (of json format) or sidecar files when reading the actions?
Can we add some integration tests that test see end-2-end? You can look at DeltaTableReadsSuite for examples.
Do we need to read the sidecar files list within a checkpoint file while creating the LogSegment? Can this be delayed until we start reading the delta actions for table state construction?
Small doc explaining the design and any decision will be helpful in understanding.

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/Checkpointer.java

vkorukanti · 2024-04-01T20:14:21Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/Checkpointer.java

+            }
+            return Optional.of(files);
+        } catch (Exception e) {
+            logger.warn("Failed to load sidecars from file {}", checkpointPath, e);


use the logger.warn(new ParametrizedMessage(..., checkpointPath), e)

kernel/kernel-api/src/main/java/io/delta/kernel/internal/snapshot/SnapshotManager.java

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/CheckpointInstance.java

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/Checkpointer.java

vkorukanti · 2024-04-01T20:51:50Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/Checkpointer.java

+                // Either V2 Parquet manifest or V1 Parquet single checkpoint.
+                actionItr = tableClient.getParquetHandler().readParquetFiles(
+                        singletonCloseableIterator(checkpointFile),
+                        SidecarFile.READ_SCHEMA,


qn: is the read schema SidecardFile.READ_SCHEMA valid? From my understanding looking at the spec, the checkpoint parquet file will have top-level columns add, remove, sidecar etc. The read schema should be sidecard: struct (path: String, sizeInBytes: Long, modTime: Long)

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/CheckpointInstance.java

vkorukanti · 2024-04-01T21:01:48Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/snapshot/SnapshotManager.java

+            if (newCheckpoint.format.usesSidecars()) {
+                final List<FileStatus> sidecarFiles = listSidecars(tableClient);
+                // Only read the checkpoint for sidecars if any sidecars exist in the table.
+                if (!sidecarFiles.isEmpty()) {
+                    // Safe to call .get() here, as we know the CheckpointInstance was initialized
+                    // with a valid filepath.
+                    final Set<Path> referencedSidecars = new HashSet<>(
+                            newCheckpoint.getReferencedSidecars(tableClient, logPath).get());
+                    newCheckpointFileList.addAll(sidecarFiles.stream()
+                            .filter(f -> referencedSidecars.contains(new Path(f.getPath())))
+                            .collect(Collectors.toList()));
+                }
+            }


can we delay the loading of the sidecard files until we start reading?

I think so - will update.

…nto kernel-v2-checkpoint

vkorukanti

LGTM. Few minor changes.

@prakharjain09 @dhruvarya-db Please take a look at it.

kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/FileNames.java

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/CheckpointInstance.java

vkorukanti · 2024-04-24T16:46:56Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/CheckpointInstance.java

+        if (format == CheckpointFormat.V2) {
+            return Objects.hash(version, numParts, format, filePath);
+        }
+        return Objects.hash(version, numParts, format, "");


Please see my other comment.

prakharjain09 · 2024-04-24T17:10:52Z

kernel/kernel-api/src/test/scala/io/delta/kernel/CheckpointInstanceSuite.scala

    ).foreach(ci => assert(ci.compareTo(CheckpointInstance.MAX_VALUE) < 0))
  }

+  test("checkpoint instance equality") {
+    val single = new CheckpointInstance("01.checkpoint.parquet")
+    val multipartPart1 = new CheckpointInstance("01.checkpoint.01.02.parquet")


The delta spec mandates the versions to be 20 length padded. So these are invalid names.

I think delta-spark also doesn't enforce this right now and this should be fixed in both places. Should not block this PR though.

prakharjain09 · 2024-04-24T17:13:20Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/CheckpointV2ReadSuite.scala

+    }
+  }
+
+  test("v2 checkpoint support with multiple sidecars") {


Could we also test empty sidecars?

prakharjain09 · 2024-04-24T17:14:07Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/CheckpointV2ReadSuite.scala

+    }
+  }
+
+  test("compatibility checkpoint with sidecar files") {


Can we also have a test case where both checkpoints - v2 and compat are present and assert that v2 is picked.

This case is covered in SnapshotManagerSuite

Lets add here also. From the SnapshotImpl you can get the LogSegment and verify it.

@chirag-s-db Is this handled?

prakharjain09 · 2024-04-24T17:16:03Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/CheckpointV2ReadSuite.scala

+
+          // Validate snapshot and data.
+          validateSnapshot(path.toString, DeltaLog.forTable(spark, path.toString).update())
+          checkTable(


Can we also assert that the right checkpoint is being used for the snapshot.

This should also be covered in SnapshotManagerSuite - this suite is end-to-end just to cover the overall read process.

Yes its already covered in UTs. It will be good to assert that the tests are exactly testing the scenario that they were supposed to. Helps with avoiding and configuration related issues e.g. this one.

Got it, this is now in the test as well.

vkorukanti · 2024-04-24T17:21:25Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/CheckpointV2ReadSuite.scala

+          spark.sql("INSERT INTO tbl VALUES (1, 'a'), (2, 'b')")
+          spark.sql("INSERT INTO tbl VALUES (3, 'c'), (4, 'd')")
+          spark.sql("INSERT INTO tbl VALUES (5, 'e'), (6, 'f')")


can we insert bit more data, may be 100 rows using Seq.range. Set the parquet/json handler batch size to less than 100, so that we can multiple batches from the ParquetHandler when reading checkpoint manifest or side car files. This exercises the iterator changes.

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/CheckpointV2ReadSuite.scala

vkorukanti · 2024-04-24T20:19:34Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/CheckpointV2ReadSuite.scala

+    }
+  }
+
+  test("compatibility checkpoint with sidecar files") {


@chirag-s-db Is this handled?

kernel/kernel-api/src/main/java/io/delta/kernel/internal/SnapshotImpl.java

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/CheckpointInstance.java

vkorukanti · 2024-04-24T22:51:58Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/CheckpointInstance.java

+        // For V2 checkpoints, the filepath is included in the hash of the instance (as we consider
+        // different UUID checkpoints to be different checkpoint instances. Otherwise, ignore
+        // the filepath (which is empty) when hashing.


is this comment still needed?

Yes, has been updated for only V2 containing filepath.

vkorukanti · 2024-04-24T22:52:25Z

kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/SidecarFile.java

+    public static StructType READ_SCHEMA = new StructType()
+            .add("path", StringType.STRING, false /* nullable */)
+            .add("sizeInBytes", LongType.LONG, false /* nullable */)
+            .add("modificationTime", LongType.LONG, false /* nullable */);


move the public static fields to the top.

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

vkorukanti · 2024-04-24T23:13:30Z

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/checkpoints/CheckpointerSuite.scala

@@ -224,7 +221,7 @@ class CheckpointerSuite extends AnyFunSuite
  }
 }

-object CheckpointerSuite {
+object CheckpointerSuite extends MockTableClientUtils with VectorTestUtils {


why extend MockTableClientUtils? VectorTestUtils makes sense as it contains the stringVector etc.

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/SnapshotManagerSuite.scala

vkorukanti · 2024-04-24T23:27:06Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/CheckpointV2ReadSuite.scala

+            "delta.kernel.default.parquet.reader.batch-size" -> "10",
+            "delta.kernel.default.json.reader.batch-size" -> "10",


Setting configs like this works for Kernel?

I thought we need to create a custom table client with Configuration containing these. And pass that custom table client to checkTable.

vkorukanti · 2024-04-24T23:31:03Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/CheckpointV2ReadSuite.scala

+    val expectedV2CkptToRead =
+      ckptVersionExpected.getOrElse(snapshotFromSpark.version - (snapshotFromSpark.version % 2))


what is this check? snapshotFromSpark.version % 2?

Checks that it is the most recent even version.

vkorukanti · 2024-04-24T23:32:49Z

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/CheckpointV2ReadSuite.scala

+    assert(snapshotImpl.getLogSegment.checkpoints.asScala.map(
+      f => new Path(f.getPath).getName.contains("-")).contains(expectV2CheckpointFormat))


can you use the regex added in FileNames to identify the v2 checkpoint conclusively?

We don't have a check in FileNames to identify conclusively, used CheckpointInstance.

prakharjain09

LGTM

chirag-s-db and others added 8 commits March 29, 2024 11:36

more changes

266d2df

changes

9863dc9

resolve merge

5a34ee4

changes

4362fac

Merge branch 'master' into kernel-v2-checkpoint

29eac32

cleanup

84074f8

more cleanup

21b071e

fix

f4d5db2

chirag-s-db changed the title ~~[WIP][Kernel] Add kernel support for v2 checkpoints.~~ [WIP][Kernel] Add kernel support for v2 checkpoints Mar 29, 2024

chirag-s-db added 4 commits March 29, 2024 16:17

fix style

f4a93db

fix

3ba99a4

remove

9f55aa9

fix compatibility file selection

bbe2a13

chirag-s-db commented Apr 1, 2024

View reviewed changes

some minor fixes

26dc16f

chirag-s-db changed the title ~~[WIP][Kernel] Add kernel support for v2 checkpoints~~ [Kernel] Add kernel support for v2 checkpoints Apr 1, 2024

fix failing test

8480492

chirag-s-db commented Apr 1, 2024

View reviewed changes

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/DeltaHistoryManagerSuite.scala Show resolved Hide resolved

chirag-s-db commented Apr 1, 2024

View reviewed changes

chirag-s-db and others added 4 commits April 1, 2024 11:00

Update MockFileSystemClientUtils.scala

158395e

fix imports

8ad2925

Merge branch 'kernel-v2-checkpoint' of github.com:chirag-s-db/delta i…

51526e3

…nto kernel-v2-checkpoint

fix imports

5ecc91e

dhruvarya-db reviewed Apr 1, 2024

View reviewed changes

chirag-s-db added 4 commits April 1, 2024 11:43

add compatibility file tests

0f477e3

fix format

4915656

fix format

8f0ba2c

remove listing

c15e335

vkorukanti requested changes Apr 1, 2024

View reviewed changes

some changes

017741b

Merge branch 'kernel-v2-checkpoint' of github.com:chirag-s-db/delta i…

011889a

…nto kernel-v2-checkpoint

chirag-s-db requested a review from prakharjain09 April 23, 2024 16:28

refactoring

6eaeca4

vkorukanti approved these changes Apr 24, 2024

View reviewed changes

fixes

f3555a6

prakharjain09 reviewed Apr 24, 2024

View reviewed changes

vkorukanti reviewed Apr 24, 2024

View reviewed changes

chirag-s-db added 5 commits April 24, 2024 11:17

fix

76d7fd4

fix

e9fa308

fix

9d9bf6f

fix

84197b7

fixes

0bbe467

vkorukanti reviewed Apr 24, 2024

View reviewed changes

chirag-s-db and others added 4 commits April 24, 2024 13:39

fix test

6da40e1

fix

b45ee5a

Merge branch 'master' into kernel-v2-checkpoint

7a77194

fix

9f5204a

vkorukanti requested changes Apr 24, 2024

View reviewed changes

chirag-s-db and others added 5 commits April 24, 2024 16:50

fix

c7de684

Update JsonUtilsSuite.scala

7885495

Update ParquetFileWriterSuite.scala

b9a2175

Update CheckpointerSuite.scala

76e872b

Update MockTableClientUtils.scala

a6e4447

prakharjain09 approved these changes Apr 25, 2024

View reviewed changes

vkorukanti approved these changes Apr 25, 2024

View reviewed changes

vkorukanti merged commit 40205ff into delta-io:master Apr 25, 2024
8 checks passed

vkorukanti mentioned this pull request Apr 25, 2024

[Feature Request][Kernel] Add checkpoint v2 read support in Kernel #2232

Closed

		"delta.kernel.default.parquet.reader.batch-size" -> "10",
		"delta.kernel.default.json.reader.batch-size" -> "10",

		val expectedV2CkptToRead =
		ckptVersionExpected.getOrElse(snapshotFromSpark.version - (snapshotFromSpark.version % 2))

		assert(snapshotImpl.getLogSegment.checkpoints.asScala.map(
		f => new Path(f.getPath).getName.contains("-")).contains(expectV2CheckpointFormat))

[Kernel] Add kernel support for v2 checkpoints #2826

[Kernel] Add kernel support for v2 checkpoints #2826

Conversation

chirag-s-db commented Mar 29, 2024 • edited

Which Delta project/connector is this regarding?

Description

How was this patch tested?

Does this PR introduce any user-facing changes?

chirag-s-db Apr 1, 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

dhruvarya-db Apr 1, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vkorukanti left a comment • 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

vkorukanti 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

chirag-s-db Apr 24, 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

prakharjain09 Apr 25, 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

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

prakharjain09 left a comment

Choose a reason for hiding this comment

chirag-s-db commented Mar 29, 2024 •

edited

chirag-s-db Apr 1, 2024 •

edited

dhruvarya-db Apr 1, 2024 •

edited

vkorukanti left a comment •

edited

chirag-s-db Apr 24, 2024 •

edited

prakharjain09 Apr 25, 2024 •

edited