[Spark] Read side changes for V2 Checkpoints #2056

prakharjain09 · 2023-09-15T03:05:56Z

Which Delta project/connector is this regarding?

Description

Read side changes for v2 checkpoints of json/parquet type.
Design doc: https://docs.google.com/document/d/18D4SiI0_t7ak8sgvyQtH9lBUyGJIp8hyZi2-399uByQ/edit#heading=h.dd2cc57oc5wk

They are given higher priority as compared to multi-part/classic checkpoint -- in case multiple checkpoints exist for same version.

How was this patch tested?

Added UTs

Does this PR introduce any user-facing changes?

No

ryan-johnson-databricks · 2023-09-22T12:40:36Z

spark/src/main/scala/org/apache/spark/sql/delta/CheckpointProvider.scala

 import org.apache.hadoop.fs.FileStatus
+import org.apache.hadoop.fs.Path


Suggested change

import org.apache.hadoop.fs.FileStatus

import org.apache.hadoop.fs.Path

import org.apache.hadoop.fs.{FileStatus, Path}

ryan-johnson-databricks · 2023-09-22T13:04:19Z

spark/src/main/scala/org/apache/spark/sql/delta/CheckpointProvider.scala

+      uninitializedCheckpointProvider: UninitializedCheckpointProvider)
+      : CheckpointProvider = uninitializedCheckpointProvider match {
+    // Note: snapshotDescriptor.protocol should be accessed as late as possible inside the futures
+    // as it might need I/O.


Where do we access snapshotDescriptor.protocol? I don't see any direct access, so is it just when we require isV2CheckpointEnabled? Or are there other indirect accesses as well?

We have require(isV2CheckpointEnabled(snapshotDescriptor) few lines below. Converting it to require(isV2CheckpointEnabled(snapshotDescriptor.protocol) to make it more explicit.

ryan-johnson-databricks · 2023-09-22T13:05:48Z

spark/src/main/scala/org/apache/spark/sql/delta/CheckpointProvider.scala

+          spark, provider.logPath, provider.fileStatus, snapshotDescriptor.deltaLog.options)
+      }
+      new LazyCompleteCheckpointProvider(provider) {
+        override def createCheckpointProvider(): CheckpointProvider = {


Am I understanding correctly, that creating the lazy checkpoint provider here kicks off a future to start reading the manifest, but we won't wait on the future unless/until somebody calls a complete checkpoint provider method?

Also adding code comments for this.

ryan-johnson-databricks · 2023-09-22T13:06:12Z

spark/src/main/scala/org/apache/spark/sql/delta/CheckpointProvider.scala

+      new LazyCompleteCheckpointProvider(uninitializedV2CheckpointProvider) {
+        override def createCheckpointProvider(): CheckpointProvider = {
+          val (checkpointMetadata, sidecarFiles) =
+            uninitializedV2CheckpointProvider.finishIOAndGetActions()


We're leveraging an existing future here, right?

Are you asking if uninitializedV2CheckpointProvider has already a future inside it which is doing I/O? Yes.

ryan-johnson-databricks · 2023-09-22T13:08:38Z

spark/src/main/scala/org/apache/spark/sql/delta/CheckpointProvider.scala

+      }
+    case provider: UninitializedV1OrV2ParquetCheckpointProvider
+        if !isV2CheckpointEnabled(checksumOpt).contains(false) =>
+      // Either v2 checkpoints are enabled, or we lack a Protocol to prove otherwise.


Boolean negation is confusing (that took a while to parse). Can we instead do:

case provider: UninitializedV1OrV2ParquetCheckpointProvider if isV2CheckpointEnabled(checksumOpt).contains(false) => // V2 checkpoints are specifically disabled, so it must be V1 ... case provider: UninitializedV1OrV2ParquetCheckpointProvider => // We can't tell immediately whether it's V1 or V2, just by looking at the file name ...

ryan-johnson-databricks · 2023-09-22T17:13:27Z

spark/src/main/scala/org/apache/spark/sql/delta/Checkpoints.scala

@@ -119,28 +166,35 @@ case class CheckpointInstance(
   *    Single part checkpoint.
   * 3. For Multi-part [[CheckpointInstance]]s corresponding to same version, the one with more
   *    parts is greater than the one with less parts.
+   *  4. For V2 Checkpoints corresponding to same version, we use the fileName as tie breaker.


What is the semantics of using file name to break ties? Put another way, when would that make a difference? Reaching step 4 of the comparison means that we're dealing with the same version, same format, and same number of parts. What else could differ in the name?

Update: .json vs. .parquet... tho I'm not sure we actually want to prefer json merely because it lexically precedes parquet?

It is not necessarily json vs parquet.
We could have two v2 checkpoints if someone runs this multiple times: deltaLog.checkpoint(deltaLog.getSnapshotAt(version = 44)).

Both are valid v2 checkpoints and any of them can be preferred one over the other. In long run, we can decide to go ahead with parquet.

oh, uuid-named... right

ryan-johnson-databricks · 2023-09-22T17:15:40Z

spark/src/main/scala/org/apache/spark/sql/delta/Checkpoints.scala

@@ -150,6 +204,12 @@ object CheckpointInstance {
    // * <version>.checkpoint.parquet
    // * <version>.checkpoint.<i>.<n>.parquet


I think the comment is missing the v2 file format description? (it says "three formats" but only shows two)

ryan-johnson-databricks · 2023-09-22T17:17:06Z

spark/src/main/scala/org/apache/spark/sql/delta/Snapshot.scala

+    case cp: CheckpointProvider =>
+      cp
+    case uninitializedProvider: UninitializedCheckpointProvider =>
+      CheckpointProvider(spark, this, checksumOpt, uninitializedProvider)
+    case o =>
+      throw new IllegalStateException(s"Unknown checkpoint provider: ${o.getClass.getName}")


nit: any particular reason to churn the two existing cases?

ryan-johnson-databricks · 2023-09-22T17:19:00Z

spark/src/test/scala/org/apache/spark/sql/delta/CheckpointsSuite.scala

+      case lzProvider : LazyCompleteCheckpointProvider
+          if lzProvider.underlyingCheckpointProvider.isInstanceOf[V2CheckpointProvider] =>
+        lzProvider.underlyingCheckpointProvider.asInstanceOf[V2CheckpointProvider]


cryptic abbreviations don't really help IMO...

Suggested change

case lzProvider : LazyCompleteCheckpointProvider

if lzProvider.underlyingCheckpointProvider.isInstanceOf[V2CheckpointProvider] =>

lzProvider.underlyingCheckpointProvider.asInstanceOf[V2CheckpointProvider]

case provider : LazyCompleteCheckpointProvider

if provider.underlyingCheckpointProvider.isInstanceOf[V2CheckpointProvider] =>

provider.underlyingCheckpointProvider.asInstanceOf[V2CheckpointProvider]

(meanwhile, the cast is annoying, but I'm not sure it's annoying enough to be worth making an extractor for)

ryan-johnson-databricks · 2023-09-22T17:21:40Z

spark/src/test/scala/org/apache/spark/sql/delta/CheckpointsSuite.scala

+          if lzProvider.underlyingCheckpointProvider.isInstanceOf[V2CheckpointProvider] =>
+        lzProvider.underlyingCheckpointProvider.asInstanceOf[V2CheckpointProvider]
+      case EmptyCheckpointProvider =>
+        throw new Exception("underlying snapshot doesn't have a checkpoint")


Throwing a plain Exception seems really strange... IllegalArgumentException or IllegalStateException or even AssertionError would seem better?

prakharjain09 added 2 commits September 22, 2023 00:23

Read side changes for V2 Checkpoints

f2ca7e0

fix

2e638a3

prakharjain09 force-pushed the cm-reads branch from 37de530 to 2e638a3 Compare September 22, 2023 07:27

ryan-johnson-databricks reviewed Sep 22, 2023

View reviewed changes

prakharjain09 requested a review from ryan-johnson-databricks September 22, 2023 23:05

ryan-johnson-databricks mentioned this pull request Sep 25, 2023

[Feature Request] Protocol: clarify immutability guarantees #1975

Open

8 tasks

vkorukanti closed this in 6859c86 Sep 25, 2023

prakharjain09 mentioned this pull request Oct 12, 2023

[Feature Request] Checkpoint V2 Format #1793

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Spark] Read side changes for V2 Checkpoints #2056

[Spark] Read side changes for V2 Checkpoints #2056

prakharjain09 commented Sep 15, 2023 •

edited

Loading

ryan-johnson-databricks Sep 22, 2023

ryan-johnson-databricks Sep 22, 2023

prakharjain09 Sep 22, 2023

ryan-johnson-databricks Sep 22, 2023

prakharjain09 Sep 22, 2023

ryan-johnson-databricks Sep 22, 2023

prakharjain09 Sep 22, 2023

ryan-johnson-databricks Sep 22, 2023

ryan-johnson-databricks Sep 22, 2023

ryan-johnson-databricks Sep 22, 2023

prakharjain09 Sep 22, 2023

ryan-johnson-databricks Sep 23, 2023

ryan-johnson-databricks Sep 22, 2023

ryan-johnson-databricks Sep 22, 2023

prakharjain09 Sep 22, 2023

ryan-johnson-databricks Sep 22, 2023

ryan-johnson-databricks Sep 22, 2023

		import org.apache.hadoop.fs.FileStatus
		import org.apache.hadoop.fs.Path

	import org.apache.hadoop.fs.FileStatus
	import org.apache.hadoop.fs.Path
	import org.apache.hadoop.fs.{FileStatus, Path}

		@@ -150,6 +204,12 @@ object CheckpointInstance {
		// * <version>.checkpoint.parquet
		// * <version>.checkpoint.<i>.<n>.parquet

[Spark] Read side changes for V2 Checkpoints #2056

[Spark] Read side changes for V2 Checkpoints #2056

Conversation

prakharjain09 commented Sep 15, 2023 • edited Loading

Which Delta project/connector is this regarding?

Description

How was this patch tested?

Does this PR introduce any user-facing changes?

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 commented Sep 15, 2023 •

edited

Loading