[Spark] Managed Commits: add a DynamoDB-based commit owner #3107

dhruvarya-db · 2024-05-16T21:35:38Z

Which Delta project/connector is this regarding?

Description

Taking inspiration from #339, this PR adds a Commit Owner Client which uses DynamoDB as the backend. Each Delta table managed by a DynamoDB instance will have one corresponding entry in a DynamoDB table. The table schema is as follows:

tableId: String --- The unique identifier for the entry. This is a UUID.
path: String --- The fully qualified path of the table in the file system. e.g. s3://bucket/path.
acceptingCommits: Boolean --- Whether the commit owner is accepting new commits. This will only
be set to false when the table is converted from managed commits to file system commits.
tableVersion: Number --- The version of the latest commit.
tableTimestamp: Number --- The inCommitTimestamp of the latest commit.
schemaVersion: Number --- The version of the schema used to store the data.
commits: --- The list of unbackfilled commits.
- version: Number --- The version of the commit.
- inCommitTimestamp: Number --- The inCommitTimestamp of the commit.
- fsName: String --- The name of the unbackfilled file.
- fsLength: Number --- The length of the unbackfilled file.
- fsTimestamp: Number --- The modification time of the unbackfilled file.

For a table to be managed by DynamoDB, registerTable must be called for that Delta table. This will create a new entry in the db for this Delta table. Every commit invocation appends the UUID delta file status to the commits list in the table entry. commit is performed through a conditional write in DynamoDB.

How was this patch tested?

Added a new suite called DynamoDBCommitOwnerClient5BackfillSuite which uses a mock DynamoDB client. + plus manual testing against a DynamoDB instance.

Does this PR introduce any user-facing changes?

prakharjain09 · 2024-05-18T01:30:18Z

...src/test/scala/org/apache/spark/sql/delta/managedcommit/CommitOwnerClientImplSuiteBase.scala

-      assert(tableCommitOwnerClient.getCommits() == GetCommitsResponse(Seq.empty, -1))
+      var expectedVersion = -1
+      if (tableCommitOwnerClient.commitOwnerClient.isInstanceOf[DynamoDBCommitOwnerClient]) {
+        // DynamoDBCommitOwnerClient stores attemptVersion as the current table version when


DynamoDBCommitOwnerClient.getCommits starts advertising attemptVersion as the current commit version even without knowledge around whether the commit has gone through or not. Practically this won't cause any issue because until the commit is done, no client will invoke DynamoDBCommitOwnerClient.getCommits since the table was a FS table. But just for the sake of semantics, we should fix it in a followup PR.

Yes, for DynamoDB, we treat registerTable as a valid commit and store the attemptVersion instead of the current Delta table version. This is not risky because the DynamoDB table entry is only treated as valid if the corresponding filesystem commit goes through (which means that the attemptVersion is in fact the Delta table version). In case of a filesystem conflict, it is expected that registerTable will be called again. However, for consistency, we can use another bool to track whether any commits have gone through DynamoDB and return -1.

prakharjain09 · 2024-05-18T01:31:38Z

spark/src/main/java/io/delta/dynamodbcommitstore/DynamoDBCommitOwnerClient.java

+            5 /* readCapacityUnits */,
+            5 /* writeCapacityUnits */,
+            false /* skipPathCheck */);
+        tryEnsureTableExists();


We are calling the other constructor here so it will take care of invoking tryEnsureTableExists ?

You are right, will fix

prakharjain09 · 2024-05-18T01:31:50Z

spark/src/main/java/io/delta/dynamodbcommitstore/DynamoDBCommitOwnerClient.java

+        return tableConf;
+    }
+
+    // Copied from DynamoDbLogStore. TODO: add the logging back.


What logging needs to be added back?

That the table exists/does not exist

Table creation attempt

Table creation failure
I will add these and logs in other parts of this class in a follow up PR.

prakharjain09 · 2024-05-18T01:32:06Z

spark/src/main/java/io/delta/dynamodbcommitstore/DynamoDBCommitOwnerClientBuilder.java

+        String dynamoDBEndpoint = conf.get(DYNAMO_DB_ENDPOINT_KEY).getOrElse(() -> {
+            throw new RuntimeException(DYNAMO_DB_ENDPOINT_KEY + " not found");
+        });
+        String awsCredentialsProviderName = conf.get(AWS_CREDENTIALS_PROVIDER_KEY).getOrElse( () ->


Suggested change

String awsCredentialsProviderName = conf.get(AWS_CREDENTIALS_PROVIDER_KEY).getOrElse( () ->

String awsCredentialsProviderName = conf.get(AWS_CREDENTIALS_PROVIDER_KEY).getOrElse(() ->

prakharjain09 · 2024-05-18T01:32:39Z

spark/src/main/java/io/delta/dynamodbcommitstore/DynamoDBCommitOwnerClientBuilder.java

+        String dynamoDBEndpoint = conf.get(DYNAMO_DB_ENDPOINT_KEY).getOrElse(() -> {
+            throw new RuntimeException(DYNAMO_DB_ENDPOINT_KEY + " not found");
+        });
+        String awsCredentialsProviderName = conf.get(AWS_CREDENTIALS_PROVIDER_KEY).getOrElse( () ->


Should AWS_CREDENTIALS_PROVIDER_KEY come from the commitOwnerConf from DeltaLog? Isn't this more of an authentication config which should be configured at JVM level i.e. via SparkSession.

Yes, once the changes around passing sparkSession to the CommitOwnerBuilder have baked in, I will update this to read it directly from sparkSession. For now, I will hardcode this to the default credentials provider.

prakharjain09 · 2024-05-18T01:32:55Z

spark/src/main/java/io/delta/dynamodbcommitstore/DynamoDBCommitOwnerClientBuilder.java

+
+    private static final String MANAGED_COMMITS_TABLE_NAME_KEY = "managedCommitsTableName";
+    private static final String DYNAMO_DB_ENDPOINT_KEY = "dynamoDBEndpoint";
+    private static final String AWS_CREDENTIALS_PROVIDER_KEY = "awsCredentialsProvider";


Add comments for these properties.

LukasRupprecht · 2024-05-19T22:50:15Z

spark/src/main/java/io/delta/dynamodbcommitstore/DynamoDBCommitOwnerClient.java

+     * signature. This method wraps the write method and declares the exception to ensure that the
+     * caller is aware of the exception.
+     */
+    private void writeActionsToFile(


Should we call this writeActionsToBackfilledFile?

LukasRupprecht · 2024-05-19T22:51:26Z

spark/src/main/java/io/delta/dynamodbcommitstore/DynamoDBCommitOwnerClient.java

+                    "The requested backfill version " + version + " is greater than the latest " +
+                            "version " + resp.getLatestTableVersion() + " for the table.");
+        }
+        // If partial writes a visible in this filesystem, we should not try to overwrite existing


Suggested change

// If partial writes a visible in this filesystem, we should not try to overwrite existing

// If partial writes are visible in this filesystem, we should not try to overwrite existing

LukasRupprecht · 2024-05-19T22:52:54Z

spark/src/main/scala/org/apache/spark/sql/delta/managedcommit/CommitOwnerClient.scala

@@ -43,7 +43,7 @@ case class Commit(
 case class CommitFailedException(
    private val retryable: Boolean,
    private val conflict: Boolean,
-    private val message: String) extends Exception(message) {
+    private val message: String) extends RuntimeException(message) {


Why is this changed needed? Isn't that just making it more restrictive?

This is needed because the CommitOwnerClient trait does not declare that commit can throw CommitFailedException in the signature (Scala doesn't let us do that). The Java code won't compile if I try to throw an exception of type Exception if it has not been declared in the function signature. RuntimeExceptions can be thrown though.

prakharjain09 reviewed May 18, 2024

View reviewed changes

LukasRupprecht reviewed May 19, 2024

View reviewed changes

add dynamodb commit owner

f2150c5

dhruvarya-db force-pushed the mc-dynamodb branch from 99bc634 to f2150c5 Compare May 20, 2024 06:54

dhruvarya-db requested a review from prakharjain09 May 20, 2024 16:06

prakharjain09 approved these changes May 20, 2024

View reviewed changes

vkorukanti merged commit 7b4ee63 into delta-io:master May 20, 2024
9 of 10 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Spark] Managed Commits: add a DynamoDB-based commit owner #3107

[Spark] Managed Commits: add a DynamoDB-based commit owner #3107

dhruvarya-db commented May 16, 2024

prakharjain09 May 18, 2024

dhruvarya-db May 18, 2024

prakharjain09 May 18, 2024

dhruvarya-db May 18, 2024

prakharjain09 May 18, 2024

dhruvarya-db May 18, 2024

prakharjain09 May 18, 2024

prakharjain09 May 18, 2024

dhruvarya-db May 18, 2024

prakharjain09 May 18, 2024

LukasRupprecht May 19, 2024

LukasRupprecht May 19, 2024

LukasRupprecht May 19, 2024

dhruvarya-db May 20, 2024

	String awsCredentialsProviderName = conf.get(AWS_CREDENTIALS_PROVIDER_KEY).getOrElse( () ->
	String awsCredentialsProviderName = conf.get(AWS_CREDENTIALS_PROVIDER_KEY).getOrElse(() ->

	// If partial writes a visible in this filesystem, we should not try to overwrite existing
	// If partial writes are visible in this filesystem, we should not try to overwrite existing

[Spark] Managed Commits: add a DynamoDB-based commit owner #3107

[Spark] Managed Commits: add a DynamoDB-based commit owner #3107

Conversation

dhruvarya-db commented May 16, 2024

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