Fix config for S3 multi-cluster mode

- problem 1: there was a mismatch between the config params in the documentation (`io.delta.storage.S3DynamoDBLogStore.ddb.tableName`) and the config params actually used in the code + in the integration test (`io.delta.storage.ddb.tableName`).
- solution 1: include the `S3DynamoDBLogStore` string in the config in the code.
- problem 2: the `io.delta.storage` prefix didn't work when specified using `--conf`. they DID work using `spark.conf.set()` or `SparkSession.builder.config()` but not with `--conf`.
- solution 2: we now allow 2 prefixes.
  - `spark.io.delta.storage.S3DynamoDBLogStore.$key.....` this will work all contexts (`--conf`, `spark.conf.set()`, etc).
  - `io.delta.storage.S3DynamoDBLogStore.$key`. this is the original prefix. this will be able to be used by delta-standalone and flink and hive. they just use hadoopConfs and don't need to have prefix starting with `spark`

- resolves #1094
- update config for S3 multi-cluster mode (i.e. S3DynamoDBLogStore) to match the public documentation. the configs were missing the string literal `S3DynamoDBLogStore` from the conf prefix
- now supports 2 confs. `spark.io.delta.storage.S3DynamoDBLogStore.$key` and `io.delta.storage.S3DynamoDBLogStore.$key`.

- added unit test for the 2 new confs
- ran integration test using existing + new s3 table, and using existing + new ddb table
- ran manual tests using locally published jars on pyspark + spark-shell + spark-submit (spark-submit via integration test)
- tested using `--conf` as well as `spark.conf.set` as well as `SparkSession.builder.config()`

Not really, just fixes a just-released, experimental LogStore config key

Closes #1095

Signed-off-by: Scott Sandre <[email protected]>
GitOrigin-RevId: e5db1e6b0dfe958e3234644462891f269313ca33

Author: scottsand-db

core/src/test/scala/org/apache/spark/sql/delta/LogStoreSuite.scala

@@ -497,45 +497,6 @@ class FailingRenameAbstractFileSystem(uri: URI, conf: org.apache.hadoop.conf.Con
   }
 }
 
-///////////////////////////////////////////////////////////////////////////////
-// Fake LogStore class & test suite to check that hadoopConf is set properly //
-///////////////////////////////////////////////////////////////////////////////
-
-class FakePublicLogStore(initHadoopConf: Configuration)
-  extends io.delta.storage.HDFSLogStore(initHadoopConf) {
-
-  assert(initHadoopConf.get("spark.delta.storage.custom.key") == "foo")
-  assert(initHadoopConf.get("this.is.a.non.spark.prefix.key") == "bar")
-}
-
-/**
- * We want to ensure that, to set configuration values for the Java LogStore implementations,
- * users can simply use `--conf $key=$value`, instead of `--conf spark.hadoop.$key=$value`.
- *
- * We also want to test that users can use a non-Spark prefix, so that our public, Java LogStore
- * implementations are not coupled to Spark-related conf keys.
- */
-class CorrectHadoopConfLogStoreSuite
-  extends SparkFunSuite
-  with LocalSparkSession
-  with LogStoreProvider {
-
-  test("java LogStore is instantiated with hadoopConf with SQLConf values") {
-    val sparkConf = new SparkConf()
-      .setMaster("local")
-      // equivalent to --conf spark.delta.storage.custom.key=foo
-      .set("spark.delta.storage.custom.key", "foo")
-      .set("spark.delta.logStore.class", classOf[FakePublicLogStore].getName)
-      .set("this.is.a.non.spark.prefix.key", "bar")
-
-    withSparkSession(SparkSession.builder.config(sparkConf).getOrCreate()) { spark =>
-      // this will instantiate the FakePublicLogStore above. If its assertion fails,
-      // then this test will fail
-      createLogStore(spark)
-    }
-  }
-}
-
 ////////////////////////////////////////////////////////////////////
 // Public LogStore (Java) suite tests from delta-storage artifact //
 ////////////////////////////////////////////////////////////////////

storage-s3-dynamodb/integration_tests/dynamodb_logstore.py

@@ -45,7 +45,7 @@
 export DELTA_NUM_ROWS=16
 
 ./run-integration-tests.py --run-storage-s3-dynamodb-integration-tests \
-    --dbb-packages org.apache.hadoop:hadoop-aws:3.3.1,com.amazonaws:aws-java-sdk-bundle:1.12.142 \
+    --dbb-packages org.apache.hadoop:hadoop-aws:3.3.1 \
     --dbb-conf spark.jars.ivySettings=/workspace/ivy.settings \
         spark.driver.extraJavaOptions=-Dlog4j.configuration=file:debug/log4j.properties
 """
@@ -60,6 +60,7 @@
 delta_storage = os.environ.get("DELTA_STORAGE", "io.delta.storage.S3DynamoDBLogStore")
 dynamo_table_name = os.environ.get("DELTA_DYNAMO_TABLE", "delta_log_test")
 dynamo_region = os.environ.get("DELTA_DYNAMO_REGION", "us-west-2")
+# used only by FailingS3DynamoDBLogStore
 dynamo_error_rates = os.environ.get("DELTA_DYNAMO_ERROR_RATES", "")
 
 if delta_table_path is None:
@@ -85,20 +86,26 @@
     .appName("utilities") \
     .master("local[*]") \
     .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
-    .config("spark.delta.logStore.class", delta_storage) \
-    .config("io.delta.storage.ddb.tableName", dynamo_table_name) \
-    .config("io.delta.storage.ddb.region", dynamo_region) \
-    .config("io.delta.storage.errorRates", dynamo_error_rates) \
+    .config("spark.delta.logStore.s3.impl", delta_storage) \
+    .config("spark.delta.logStore.s3a.impl", delta_storage) \
+    .config("spark.delta.logStore.s3n.impl", delta_storage) \
+    .config("spark.io.delta.storage.S3DynamoDBLogStore.ddb.tableName", dynamo_table_name) \
+    .config("spark.io.delta.storage.S3DynamoDBLogStore.ddb.region", dynamo_region) \
+    .config("spark.io.delta.storage.S3DynamoDBLogStore.errorRates", dynamo_error_rates) \
+    .config("spark.io.delta.storage.S3DynamoDBLogStore.provisionedThroughput.rcu", 12) \
+    .config("spark.io.delta.storage.S3DynamoDBLogStore.provisionedThroughput.wcu", 13) \
     .getOrCreate()
 
-# spark.sparkContext.setLogLevel("INFO")
+spark.sparkContext.setLogLevel("INFO")
 
 data = spark.createDataFrame([], "id: int, a: int")
+print("writing:", data.collect())
 data.write.format("delta").mode("overwrite").partitionBy("id").save(delta_table_path)
 
 
 def write_tx(n):
     data = spark.createDataFrame([[n, n]], "id: int, a: int")
+    print("writing:", data.collect())
     data.write.format("delta").mode("append").partitionBy("id").save(delta_table_path)
 
 

storage-s3-dynamodb/src/main/java/io/delta/storage/S3DynamoDBLogStore.java

@@ -71,12 +71,19 @@ public class S3DynamoDBLogStore extends BaseExternalLogStore {
     private static final Logger LOG = LoggerFactory.getLogger(S3DynamoDBLogStore.class);
 
     /**
-     * Configuration keys for the DynamoDB client
+     * Configuration keys for the DynamoDB client.
+     *
+     * Keys are either of the form $SPARK_CONF_PREFIX.$CONF or $BASE_CONF_PREFIX.$CONF,
+     * e.g. spark.io.delta.storage.S3DynamoDBLogStore.ddb.tableName
+     * or io.delta.storage.S3DynamoDBLogStore.ddb.tableName
      */
-    private static final String CONF_PREFIX = "io.delta.storage.";
-    private static final String DBB_CLIENT_TABLE = "ddb.tableName";
-    private static final String DBB_CLIENT_REGION = "ddb.region";
-    private static final String DBB_CLIENT_CREDENTIALS_PROVIDER = "credentials.provider";
+    public static final String SPARK_CONF_PREFIX = "spark.io.delta.storage.S3DynamoDBLogStore";
+    public static final String BASE_CONF_PREFIX = "io.delta.storage.S3DynamoDBLogStore";
+    public static final String DDB_CLIENT_TABLE = "ddb.tableName";
+    public static final String DDB_CLIENT_REGION = "ddb.region";
+    public static final String DDB_CLIENT_CREDENTIALS_PROVIDER = "credentials.provider";
+    public static final String DDB_CREATE_TABLE_RCU = "provisionedThroughput.rcu";
+    public static final String DDB_CREATE_TABLE_WCU = "provisionedThroughput.wcu";
 
     /**
      * DynamoDB table attribute keys
@@ -98,13 +105,13 @@ public class S3DynamoDBLogStore extends BaseExternalLogStore {
     public S3DynamoDBLogStore(Configuration hadoopConf) throws IOException {
         super(hadoopConf);
 
-        tableName = getParam(hadoopConf, DBB_CLIENT_TABLE, "delta_log");
+        tableName = getParam(hadoopConf, DDB_CLIENT_TABLE, "delta_log");
         credentialsProviderName = getParam(
             hadoopConf,
-            DBB_CLIENT_CREDENTIALS_PROVIDER,
+            DDB_CLIENT_CREDENTIALS_PROVIDER,
             "com.amazonaws.auth.DefaultAWSCredentialsProviderChain"
         );
-        regionName = getParam(hadoopConf, DBB_CLIENT_REGION, "us-east-1");
+        regionName = getParam(hadoopConf, DDB_CLIENT_REGION, "us-east-1");
         LOG.info("using tableName {}", tableName);
         LOG.info("using credentialsProviderName {}", credentialsProviderName);
         LOG.info("using regionName {}", regionName);
@@ -221,12 +228,12 @@ private void tryEnsureTableExists(Configuration hadoopConf) throws IOException {
                 TableDescription descr = result.getTable();
                 status = descr.getTableStatus();
             } catch (ResourceNotFoundException e) {
-                final long rcu = Long.parseLong(getParam(hadoopConf, "provisionedThroughput.rcu", "5"));
-                final long wcu = Long.parseLong(getParam(hadoopConf, "provisionedThroughput.wcu", "5"));
+                final long rcu = Long.parseLong(getParam(hadoopConf, DDB_CREATE_TABLE_RCU, "5"));
+                final long wcu = Long.parseLong(getParam(hadoopConf, DDB_CREATE_TABLE_WCU, "5"));
 
                 LOG.info(
-                    "DynamoDB table `{}` in region `{}` does not exist."
-                    + "Creating it now with provisioned throughput of {} RCUs and {} WCUs.",
+                    "DynamoDB table `{}` in region `{}` does not exist. " +
+                    "Creating it now with provisioned throughput of {} RCUs and {} WCUs.",
                     tableName, regionName, rcu, wcu);
                 try {
                     client.createTable(
@@ -289,10 +296,35 @@ private AmazonDynamoDBClient getClient() throws java.io.IOException {
         }
     }
 
-    protected String getParam(Configuration config, String name, String defaultValue) {
-        return config.get(
-            String.format("%s%s", CONF_PREFIX, name),
-            defaultValue
-        );
+    /**
+     * Get the hadoopConf param $name that is prefixed either with $SPARK_CONF_PREFIX or
+     * $BASE_CONF_PREFIX.
+     *
+     * If two parameters exist, one for each prefix, then an IllegalArgumentException is thrown.
+     *
+     * If no parameters exist, then the $defaultValue is returned.
+     */
+    protected static String getParam(Configuration hadoopConf, String name, String defaultValue) {
+        final String sparkPrefixKey = String.format("%s.%s", SPARK_CONF_PREFIX, name);
+        final String basePrefixKey = String.format("%s.%s", BASE_CONF_PREFIX, name);
+
+        final String sparkPrefixVal = hadoopConf.get(sparkPrefixKey);
+        final String basePrefixVal = hadoopConf.get(basePrefixKey);
+
+        if (sparkPrefixVal != null &&
+            basePrefixVal != null &&
+            !sparkPrefixVal.equals(basePrefixVal)) {
+            throw new IllegalArgumentException(
+                String.format(
+                    "Configuration properties `%s=%s` and `%s=%s` have different values. " +
+                    "Please set only one.",
+                    sparkPrefixKey, sparkPrefixVal, basePrefixKey, basePrefixVal
+                )
+            );
+        }
+
+        if (sparkPrefixVal != null) return sparkPrefixVal;
+        if (basePrefixVal != null) return basePrefixVal;
+        return defaultValue;
     }
 }

storage-s3-dynamodb/src/test/scala/io/delta/storage/ExternalLogStoreSuite.scala

@@ -18,7 +18,9 @@ package io.delta.storage
 
 import java.net.URI
 
+import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs._
+import org.scalatest.funsuite.AnyFunSuite
 
 import org.apache.spark.sql.delta.FakeFileSystem
 import org.apache.spark.sql.delta.util.FileNames
@@ -138,6 +140,64 @@ class ExternalLogStoreSuite extends org.apache.spark.sql.delta.PublicLogStoreSui
   protected def shouldUseRenameToWriteCheckpoint: Boolean = false
 }
 
+///////////////////////////////////
+// S3DynamoDBLogStore Test Suite //
+///////////////////////////////////
+
+class S3DynamoDBLogStoreSuite extends AnyFunSuite {
+  test("getParam") {
+    import S3DynamoDBLogStore._
+
+    val sparkPrefixKey = "spark.io.delta.storage.S3DynamoDBLogStore.ddb.tableName"
+    val basePrefixKey = "io.delta.storage.S3DynamoDBLogStore.ddb.tableName"
+
+    // Sanity check
+    require(sparkPrefixKey == SPARK_CONF_PREFIX + "." + DDB_CLIENT_TABLE)
+    require(basePrefixKey == BASE_CONF_PREFIX + "." + DDB_CLIENT_TABLE)
+
+    // Case 1: no parameters exist, should use default
+    assert(getParam(new Configuration(), DDB_CLIENT_TABLE, "default_table") == "default_table")
+
+    // Case 2: spark-prefix param only
+    {
+      val hadoopConf = new Configuration()
+      hadoopConf.set(sparkPrefixKey, "some_other_table_2")
+      assert(getParam(hadoopConf, DDB_CLIENT_TABLE, "default_table") == "some_other_table_2")
+    }
+
+    // Case 3: base-prefix param only
+    {
+      val hadoopConf = new Configuration()
+      hadoopConf.set(basePrefixKey, "some_other_table_3")
+      assert(getParam(hadoopConf, DDB_CLIENT_TABLE, "default_table") == "some_other_table_3")
+    }
+
+    // Case 4: both params set, same value
+    {
+      val hadoopConf = new Configuration()
+      hadoopConf.set(sparkPrefixKey, "some_other_table_4")
+      hadoopConf.set(basePrefixKey, "some_other_table_4")
+      assert(getParam(hadoopConf, DDB_CLIENT_TABLE, "default_table") == "some_other_table_4")
+    }
+
+    // Case 5: both param set, different value
+    {
+      val hadoopConf = new Configuration()
+      hadoopConf.set(sparkPrefixKey, "some_other_table_5a")
+      hadoopConf.set(basePrefixKey, "some_other_table_5b")
+      val e = intercept[IllegalArgumentException] {
+        getParam(hadoopConf, DDB_CLIENT_TABLE, "default_table")
+      }.getMessage
+      assert(e == (s"Configuration properties `$sparkPrefixKey=some_other_table_5a` and " +
+        s"`$basePrefixKey=some_other_table_5b` have different values. Please set only one."))
+    }
+  }
+}
+
+////////////////////////////////
+// File System Helper Classes //
+////////////////////////////////
+
 /**
  * This utility enables failure simulation on file system.
  * Providing a matching suffix results in an exception being