[Spark] Make naming of manage commit consistent (#3225)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [X] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->
1. Replaces instances of Managed Commits with Managed Commit (without
the s) in functions, configurations, and docs.
2. Renames the feature name to managedCommit-preview
3. Renames the DynamoDBCommitOwner tableConfig key to dynamoDBTableName

Cherrypicked from #3224
## How was this patch tested?

<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->
Existing tests should cover this refactor.

## Does this PR introduce _any_ user-facing changes?

<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->
The managed commit table feature name has been updated to
managedCommit-preview

Author: dhruvarya-db

protocol_rfcs/managed-commits.md

@@ -1,7 +1,7 @@
 # Managed Commits
 **Associated Github issue for discussions: https://github.com/delta-io/delta/issues/2598**
 
-This RFC proposes a new table feature `managedCommits` which changes the way Delta Lake performs commits.
+This RFC proposes a new table feature `managedCommit` which changes the way Delta Lake performs commits.
 
 Today’s Delta commit protocol relies on the filesystem to provide commit atomicity. This feature request is to allow Delta tables which gets commit atomicity using an external commit-owner and not
 the filesystem (s3, abfs etc). This allows us to deal with various limitations of Delta:
@@ -324,13 +324,13 @@ the seemingly-extra files might think the table metadata was corrupted.
 
 The managed-commit feature is supported and active when:
 - The table must be on Writer Version 7.
-- The table has a `protocol` action with `writerFeatures` containing the feature `managedCommits`.
-- The table has a metadata property `delta.managedCommits.commitOwner` in the [change-metadata](#change-metadata)'s configuration.
-- The table may have a metadata property `delta.managedCommits.commitOwnerConf` in the [change-metadata](#change-metadata)'s configuration. The value of this property is a json-coded string-to-string map.
+- The table has a `protocol` action with `writerFeatures` containing the feature `managedCommit`.
+- The table has a metadata property `delta.managedCommit.commitOwner` in the [change-metadata](#change-metadata)'s configuration.
+- The table may have a metadata property `delta.managedCommit.commitOwnerConf` in the [change-metadata](#change-metadata)'s configuration. The value of this property is a json-coded string-to-string map.
     - A commit-owner can store additional information (e.g. configuration information such as service endpoints) in this field, for use by the commit-owner client (it is opaque to the Delta client).
     - This field should never include secrets such as auth tokens or credentials, because any reader with access to the table's storage location can see them.
 
-Note that a table is in invalid state if the change-metadata contains the `delta.managedCommits.commitOwner` property but the table does not have the `managedCommits` feature in the `protocol` action (or vice versa).
+Note that a table is in invalid state if the change-metadata contains the `delta.managedCommit.commitOwner` property but the table does not have the `managedCommit` feature in the `protocol` action (or vice versa).
 
 E.g.
 ```json
@@ -342,8 +342,8 @@ E.g.
       "partitionColumns":[],
       "configuration":{
          "appendOnly": "true",
-         "delta.managedCommits.commitOwner": "commit-owner-1",
-         "delta.managedCommits.commitOwnerConf":
+         "delta.managedCommit.commitOwner": "commit-owner-1",
+         "delta.managedCommit.commitOwnerConf":
              "{\"endpoint\":\"http://sample-url.com/commit\", \"authenticationMode\":\"oauth2\"}"
       }
    }

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

@@ -63,7 +63,7 @@ public class DynamoDBCommitOwnerClient implements CommitOwnerClient {
     /**
      * The name of the DynamoDB table used to store unbackfilled commits.
      */
-    final String managedCommitsTableName;
+    final String managedCommitTableName;
 
     /**
      * The DynamoDB client used to interact with the DynamoDB table.
@@ -124,12 +124,12 @@ private static class GetCommitsResultInternal {
 
 
     public DynamoDBCommitOwnerClient(
-            String managedCommitsTableName,
+            String managedCommitTableName,
             String endpoint,
             AmazonDynamoDB client,
             long backfillBatchSize) throws IOException {
         this(
-            managedCommitsTableName,
+            managedCommitTableName,
             endpoint,
             client,
             backfillBatchSize,
@@ -139,14 +139,14 @@ public DynamoDBCommitOwnerClient(
     }
 
     public DynamoDBCommitOwnerClient(
-            String managedCommitsTableName,
+            String managedCommitTableName,
             String endpoint,
             AmazonDynamoDB client,
             long backfillBatchSize,
             long readCapacityUnits,
             long writeCapacityUnits,
             boolean skipPathCheck) throws IOException {
-        this.managedCommitsTableName = managedCommitsTableName;
+        this.managedCommitTableName = managedCommitTableName;
         this.endpoint = endpoint;
         this.client = client;
         this.backfillBatchSize = backfillBatchSize;
@@ -169,7 +169,7 @@ private String getTableId(Map<String, String> managedCommitTableConf) {
     private GetItemResult getEntryFromCommitOwner(
             Map<String, String> managedCommitTableConf, String... attributesToGet) {
         GetItemRequest request = new GetItemRequest()
-                .withTableName(managedCommitsTableName)
+                .withTableName(managedCommitTableName)
                 .addKeyEntry(
                         DynamoDBTableEntryConstants.TABLE_ID,
                         new AttributeValue().withS(getTableId(managedCommitTableConf)))
@@ -248,7 +248,7 @@ protected CommitResponse commitToOwner(
                 new AttributeValue().withN(Long.toString(commitFile.getModificationTime())));
 
         UpdateItemRequest request = new UpdateItemRequest()
-                .withTableName(managedCommitsTableName)
+                .withTableName(managedCommitTableName)
                 .addKeyEntry(
                         DynamoDBTableEntryConstants.TABLE_ID,
                         new AttributeValue().withS(getTableId(managedCommitTableConf)))
@@ -579,7 +579,7 @@ public void backfillToVersion(
             }
         }
         UpdateItemRequest request = new UpdateItemRequest()
-                .withTableName(managedCommitsTableName)
+                .withTableName(managedCommitTableName)
                 .addKeyEntry(
                         DynamoDBTableEntryConstants.TABLE_ID,
                         new AttributeValue().withS(getTableId(managedCommitTableConf)))
@@ -660,7 +660,7 @@ public Map<String, String> registerTable(
                 new AttributeValue().withN(Integer.toString(CLIENT_VERSION)));
 
         PutItemRequest request = new PutItemRequest()
-                .withTableName(managedCommitsTableName)
+                .withTableName(managedCommitTableName)
                 .withItem(item)
                 .withConditionExpression(
                         String.format(
@@ -686,14 +686,14 @@ private void tryEnsureTableExists() throws IOException {
         while(retries < 20) {
             String status = "CREATING";
             try {
-                DescribeTableResult result = client.describeTable(managedCommitsTableName);
+                DescribeTableResult result = client.describeTable(managedCommitTableName);
                 TableDescription descr = result.getTable();
                 status = descr.getTableStatus();
             } catch (ResourceNotFoundException e) {
                 LOG.info(
                         "DynamoDB table `{}` for endpoint `{}` does not exist. " +
                         "Creating it now with provisioned throughput of {} RCUs and {} WCUs.",
-                        managedCommitsTableName, endpoint, readCapacityUnits, writeCapacityUnits);
+                        managedCommitTableName, endpoint, readCapacityUnits, writeCapacityUnits);
                 try {
                     client.createTable(
                             // attributeDefinitions
@@ -702,7 +702,7 @@ private void tryEnsureTableExists() throws IOException {
                                             DynamoDBTableEntryConstants.TABLE_ID,
                                             ScalarAttributeType.S)
                             ),
-                            managedCommitsTableName,
+                            managedCommitTableName,
                             // keySchema
                             java.util.Collections.singletonList(
                                     new KeySchemaElement(
@@ -718,23 +718,23 @@ private void tryEnsureTableExists() throws IOException {
             }
             if (status.equals("ACTIVE")) {
                 if (created) {
-                    LOG.info("Successfully created DynamoDB table `{}`", managedCommitsTableName);
+                    LOG.info("Successfully created DynamoDB table `{}`", managedCommitTableName);
                 } else {
-                    LOG.info("Table `{}` already exists", managedCommitsTableName);
+                    LOG.info("Table `{}` already exists", managedCommitTableName);
                 }
                 break;
             } else if (status.equals("CREATING")) {
                 retries += 1;
-                LOG.info("Waiting for `{}` table creation", managedCommitsTableName);
+                LOG.info("Waiting for `{}` table creation", managedCommitTableName);
                 try {
                     Thread.sleep(1000);
                 } catch(InterruptedException e) {
                     throw new InterruptedIOException(e.getMessage());
                 }
             } else {
-                LOG.error("table `{}` status: {}", managedCommitsTableName, status);
+                LOG.error("table `{}` status: {}", managedCommitTableName, status);
                 throw new RuntimeException("DynamoDBCommitOwnerCliet: Unable to create table with " +
-                        "name " + managedCommitsTableName + " for endpoint " + endpoint + ". Ensure " +
+                        "name " + managedCommitTableName + " for endpoint " + endpoint + ". Ensure " +
                         "that the credentials provided have the necessary permissions to create " +
                         "tables in DynamoDB. If the table already exists, ensure that the table " +
                         "is in the ACTIVE state.");
@@ -748,7 +748,7 @@ public boolean semanticEquals(CommitOwnerClient other) {
             return false;
         }
         DynamoDBCommitOwnerClient otherStore = (DynamoDBCommitOwnerClient) other;
-        return this.managedCommitsTableName.equals(otherStore.managedCommitsTableName)
+        return this.managedCommitTableName.equals(otherStore.managedCommitTableName)
                 && this.endpoint.equals(otherStore.endpoint);
     }
 }

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

@@ -42,7 +42,7 @@ public String getName() {
      * commits for this owner. The value of this key is stored in the `conf`
      * which is passed to the `build` method.
      */
-    private static final String MANAGED_COMMITS_TABLE_NAME_KEY = "managedCommitsTableName";
+    private static final String MANAGED_COMMITS_TABLE_NAME_KEY = "dynamoDBTableName";
     /**
      * The endpoint of the DynamoDB service. The value of this key is stored in the
      * `conf` which is passed to the `build` method.
@@ -51,7 +51,7 @@ public String getName() {
 
     @Override
     public CommitOwnerClient build(SparkSession spark, Map<String, String> conf) {
-        String managedCommitsTableName = conf.get(MANAGED_COMMITS_TABLE_NAME_KEY).getOrElse(() -> {
+        String managedCommitTableName = conf.get(MANAGED_COMMITS_TABLE_NAME_KEY).getOrElse(() -> {
             throw new RuntimeException(MANAGED_COMMITS_TABLE_NAME_KEY + " not found");
         });
         String dynamoDBEndpoint = conf.get(DYNAMO_DB_ENDPOINT_KEY).getOrElse(() -> {
@@ -72,7 +72,7 @@ public CommitOwnerClient build(SparkSession spark, Map<String, String> conf) {
                     spark.sessionState().newHadoopConf()
             );
             return getDynamoDBCommitOwnerClient(
-                    managedCommitsTableName,
+                    managedCommitTableName,
                     dynamoDBEndpoint,
                     ddbClient,
                     BACKFILL_BATCH_SIZE,
@@ -86,7 +86,7 @@ public CommitOwnerClient build(SparkSession spark, Map<String, String> conf) {
     }
 
     protected DynamoDBCommitOwnerClient getDynamoDBCommitOwnerClient(
-            String managedCommitsTableName,
+            String managedCommitTableName,
             String dynamoDBEndpoint,
             AmazonDynamoDB ddbClient,
             long backfillBatchSize,
@@ -95,7 +95,7 @@ protected DynamoDBCommitOwnerClient getDynamoDBCommitOwnerClient(
             boolean skipPathCheck
     ) throws IOException {
         return new DynamoDBCommitOwnerClient(
-                managedCommitsTableName,
+                managedCommitTableName,
                 dynamoDBEndpoint,
                 ddbClient,
                 backfillBatchSize,

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

@@ -31,7 +31,7 @@ private ManagedCommitUtils() {}
 
     /** The configuration key for the managed commit owner. */
     private static final String MANAGED_COMMIT_OWNER_CONF_KEY =
-            "delta.managedCommits.commitOwner-preview";
+            "delta.managedCommit.commitOwner-preview";
 
     /**
      * Creates a new unbackfilled delta file path for the given commit version.
@@ -68,10 +68,10 @@ private static String getManagedCommitOwner(AbstractMetadata metadata) {
     public static boolean isManagedCommitToFSConversion(
             Long commitVersion,
             UpdatedActions updatedActions) {
-        boolean oldMetadataHasManagedCommits =
+        boolean oldMetadataHasManagedCommit =
                 !getManagedCommitOwner(updatedActions.getOldMetadata()).isEmpty();
-        boolean newMetadataHasManagedCommits =
+        boolean newMetadataHasManagedCommit =
                 !getManagedCommitOwner(updatedActions.getNewMetadata()).isEmpty();
-        return oldMetadataHasManagedCommits && !newMetadataHasManagedCommits && commitVersion > 0;
+        return oldMetadataHasManagedCommit && !newMetadataHasManagedCommit && commitVersion > 0;
     }
 }

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

@@ -737,7 +737,7 @@ trait DeltaConfigsBase extends DeltaLogging {
     helpMessage = "needs to be a boolean.")
 
   val MANAGED_COMMIT_OWNER_NAME = buildConfig[Option[String]](
-    "managedCommits.commitOwner-preview",
+    "managedCommit.commitOwner-preview",
     null,
     v => Option(v),
     _ => true,
@@ -748,14 +748,14 @@ trait DeltaConfigsBase extends DeltaLogging {
       |""".stripMargin)
 
   val MANAGED_COMMIT_OWNER_CONF = buildConfig[Map[String, String]](
-    "managedCommits.commitOwnerConf-preview",
+    "managedCommit.commitOwnerConf-preview",
     null,
     v => JsonUtils.fromJson[Map[String, String]](Option(v).getOrElse("{}")),
     _ => true,
     "A string-to-string map of configuration properties for the managed commit-owner.")
 
   val MANAGED_COMMIT_TABLE_CONF = buildConfig[Map[String, String]](
-    "managedCommits.tableConf-preview",
+    "managedCommit.tableConf-preview",
     null,
     v => JsonUtils.fromJson[Map[String, String]](Option(v).getOrElse("{}")),
     _ => true,

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

@@ -1246,7 +1246,7 @@ trait OptimisticTransactionImpl extends TransactionalWrite
    */
   protected def updateMetadataWithManagedCommitConfs(): Boolean = {
     validateManagedCommitConfInMetadata(newMetadata)
-    val newManagedCommitTableConfOpt = registerTableForManagedCommitsIfNeeded(metadata, protocol)
+    val newManagedCommitTableConfOpt = registerTableForManagedCommitIfNeeded(metadata, protocol)
     val newManagedCommitTableConf = newManagedCommitTableConfOpt.getOrElse {
       return false
     }
@@ -1516,7 +1516,7 @@ trait OptimisticTransactionImpl extends TransactionalWrite
    *         This metadata should be added to the [[Metadata.configuration]] before doing the
    *         commit.
    */
-  protected def registerTableForManagedCommitsIfNeeded(
+  protected def registerTableForManagedCommitIfNeeded(
       finalMetadata: Metadata,
       finalProtocol: Protocol): Option[Map[String, String]] = {
     val (oldOwnerName, oldOwnerConf) = ManagedCommitUtils.getManagedCommitConfs(snapshot.metadata)

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

@@ -674,7 +674,7 @@ object V2CheckpointTableFeature
 
 /** Table feature to represent tables whose commits are managed by separate commit-owner */
 object ManagedCommitTableFeature
-  extends WriterFeature(name = "managed-commit-preview")
+  extends WriterFeature(name = "managedCommit-preview")
     with FeatureAutomaticallyEnabledByMetadata
     with RemovableFeature {
 

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

@@ -126,11 +126,11 @@ trait AbstractBatchBackfillingCommitOwnerClient extends CommitOwnerClient with L
   private def isManagedCommitToFSConversion(
       commitVersion: Long,
       updatedActions: UpdatedActions): Boolean = {
-    val oldMetadataHasManagedCommits = updatedActions.getOldMetadata.asInstanceOf[Metadata]
+    val oldMetadataHasManagedCommit = updatedActions.getOldMetadata.asInstanceOf[Metadata]
       .managedCommitOwnerName.nonEmpty
-    val newMetadataHasManagedCommits = updatedActions.getNewMetadata.asInstanceOf[Metadata]
+    val newMetadataHasManagedCommit = updatedActions.getNewMetadata.asInstanceOf[Metadata]
       .managedCommitOwnerName.nonEmpty
-    oldMetadataHasManagedCommits && !newMetadataHasManagedCommits && commitVersion > 0
+    oldMetadataHasManagedCommit && !newMetadataHasManagedCommit && commitVersion > 0
   }
 
   protected def generateUUID(): String = UUID.randomUUID().toString

spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala

@@ -539,7 +539,7 @@ trait DeltaSQLConfBase {
       .createWithDefault(20)
 
   val MANAGED_COMMIT_GET_COMMITS_THREAD_POOL_SIZE =
-    buildStaticConf("managedCommits.getCommits.threadPoolSize")
+    buildStaticConf("managedCommit.getCommits.threadPoolSize")
       .internal()
       .doc("The size of the thread pool for listing files from the commit-owner.")
       .intConf
@@ -551,7 +551,7 @@ trait DeltaSQLConfBase {
   /////////////////////////////////////////////
 
   val MANAGED_COMMIT_DDB_AWS_CREDENTIALS_PROVIDER_NAME =
-    buildConf("managedCommits.commitOwner.ddb.awsCredentialsProviderName")
+    buildConf("managedCommit.commitOwner.dynamodb.awsCredentialsProviderName")
       .internal()
       .doc("The fully qualified class name of the AWS credentials provider to use for " +
         "interacting with DynamoDB in the DynamoDB Commit Owner Client. e.g. " +
@@ -560,7 +560,7 @@ trait DeltaSQLConfBase {
       .createWithDefault("com.amazonaws.auth.DefaultAWSCredentialsProviderChain")
 
   val MANAGED_COMMIT_DDB_SKIP_PATH_CHECK =
-    buildConf("managedCommits.commitOwner.ddb.skipPathCheck.enabled")
+    buildConf("managedCommit.commitOwner.dynamodb.skipPathCheckEnabled")
       .internal()
       .doc("When enabled, the DynamoDB Commit Owner will not enforce that the table path of the " +
         "current Delta table matches the stored in the corresponding DynamoDB table. This " +
@@ -572,7 +572,7 @@ trait DeltaSQLConfBase {
       .createWithDefault(false)
 
   val MANAGED_COMMIT_DDB_READ_CAPACITY_UNITS =
-    buildConf("managedCommits.commitOwner.ddb.readCapacityUnits")
+    buildConf("managedCommit.commitOwner.dynamodb.readCapacityUnits")
       .internal()
       .doc("Controls the provisioned read capacity units for the DynamoDB table backing the " +
         "DynamoDB Commit Owner. This configuration is only used when the DynamoDB table is first " +
@@ -581,7 +581,7 @@ trait DeltaSQLConfBase {
       .createWithDefault(5)
 
   val MANAGED_COMMIT_DDB_WRITE_CAPACITY_UNITS =
-    buildConf("managedCommits.commitOwner.ddb.writeCapacityUnits")
+    buildConf("managedCommit.commitOwner.dynamodb.writeCapacityUnits")
       .internal()
       .doc("Controls the provisioned write capacity units for the DynamoDB table backing the " +
         "DynamoDB Commit Owner. This configuration is only used when the DynamoDB table is first " +

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

@@ -499,7 +499,7 @@ class DeltaLogSuite extends QueryTest
           deltaLog.newDeltaHadoopConf())
         .filter(!_.getPath.getName.startsWith("_"))
         .foreach(f => fs.delete(f.getPath, true))
-      if (managedCommitsEnabledInTests) {
+      if (managedCommitEnabledInTests) {
         // For Managed Commit table with a commit that is not backfilled, we can't use
         // 00000000002.json yet. Contact commit store to get uuid file path to malform json file.
         val oc = CommitOwnerProvider.getCommitOwnerClient(
@@ -598,7 +598,7 @@ class DeltaLogSuite extends QueryTest
 
       val log = DeltaLog.forTable(spark, path)
       var commitFilePath = FileNames.unsafeDeltaFile(log.logPath, 1L)
-      if (managedCommitsEnabledInTests) {
+      if (managedCommitEnabledInTests) {
         // For Managed Commit table with a commit that is not backfilled, we can't use
         // 00000000001.json yet. Contact commit store to get uuid file path to malform json file.
         val oc = CommitOwnerProvider.getCommitOwnerClient(