This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 4ce7ef834e2dd50e9ac2cf8cad8a74f6394dfe6c Author: Prashant Wason <[email protected]> AuthorDate: Sat Jun 27 13:09:27 2026 -0700 [MINOR] Forward spark.hoodie.* SparkConf to write path (parity with read path) (#18650) [MINOR] Forward spark.hoodie.* SparkConf to writes (parity with reads) The read side of DefaultSource collects hoodie.* and spark.hoodie.* from SparkConf and strips the spark. prefix before handing to the write path. The write side dropped them entirely — configs set via --conf spark.hoodie.X=Y had no effect on writes (e.g. hoodie.datasource.hive_sync.use_spark_catalog). Extract collectHoodieAndSparkHoodieConfs + normalizeSparkHoodiePrefix in DataSourceOptionsHelper. Wire both into DefaultSource.createRelation (write) and into parametersWithWriteDefaults so callers bypassing the DefaultSource entry point (SQL ALTER TABLE, HoodieCLIUtils) get the same parity. Explicit .option(...) calls still win over SparkConf. Closes #18649 Co-Authored-By: Claude Opus 4.8 <[email protected]> --------- Co-authored-by: Claude Opus 4.7 <[email protected]> Co-authored-by: sivabalan <[email protected]> Co-authored-by: Rahil Chertara <[email protected]> (cherry picked from commit 01f1496ff565381f95fbeb2096d0fa3c208c4fdc) --- .../scala/org/apache/hudi/DataSourceOptions.scala | 88 ++++++++++++-- .../main/scala/org/apache/hudi/DefaultSource.scala | 22 ++-- .../scala/org/apache/hudi/HoodieWriterUtils.scala | 8 +- .../org/apache/hudi/TestDataSourceOptions.scala | 131 ++++++++++++++++++++- 4 files changed, 231 insertions(+), 18 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala index 445f052ee4f6..b2548029e621 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DataSourceOptions.scala @@ -35,6 +35,7 @@ import org.apache.hudi.keygen.factory.HoodieSparkKeyGeneratorFactory.{getKeyGene import org.apache.hudi.sync.common.HoodieSyncConfig import org.apache.hudi.util.{JFunction, SparkConfigUtils} +import org.apache.spark.sql.SQLContext import org.apache.spark.sql.execution.datasources.{DataSourceUtils => SparkDataSourceUtils} import org.apache.spark.sql.hudi.HoodieSqlCommonUtils import org.slf4j.LoggerFactory @@ -1033,9 +1034,87 @@ object DataSourceOptionsHelper { private val log = LoggerFactory.getLogger(DataSourceOptionsHelper.getClass) // Prefix constants for config normalization + private val HOODIE_PREFIX = "hoodie." private val SPARK_HOODIE_PREFIX = "spark.hoodie." private val SPARK_PREFIX = "spark." + /** + * Collects `hoodie.*` and `spark.hoodie.*` configs from the SparkConf, normalizes the + * `spark.hoodie.*` keys to canonical `hoodie.*`, and merges with explicit DataFrame + * options. Explicit options win over SparkConf. + * + * This is the read-path entry point: reads have always picked up session-level `hoodie.*` + * confs (e.g. `hoodie.datasource.query.type`), so both prefixes are forwarded here. + * Do NOT use this for writes — see `collectSparkHoodieConfs` for why ambient `hoodie.*` + * confs must not be forwarded to the write path. + * + * Example (SparkConf has both prefixes set; explicit options override): + * {{{ + * SparkConf: spark.hoodie.X = "a", hoodie.Y = "b" + * optParams: hoodie.X = "c" + * result: hoodie.X = "c" // explicit wins over both prefixes + * hoodie.Y = "b" + * }}} + */ + def collectHoodieAndSparkHoodieConfs(sqlContext: SQLContext, + optParams: Map[String, String]): Map[String, String] = + collectConfsByPrefix(sqlContext, optParams, includeHoodiePrefix = true) + + /** + * Collects only `spark.hoodie.*` configs from the SparkConf, normalizes them to canonical + * `hoodie.*`, and merges with explicit DataFrame options. Explicit options win over SparkConf. + * + * This is the write-path entry point. It deliberately does NOT forward bare `hoodie.*` + * session confs: unlike reads, the DataFrame write path historically honored only the + * explicit `.option(...)` map, so injecting ambient `hoodie.*` session state (e.g. a + * session-level `hoodie.datasource.write.operation` or `hoodie.logfile.data.block.format`) + * would silently change every `df.write`. The bug this addresses (HUDI-#18649) is about + * `--conf spark.hoodie.X=Y` being dropped on writes, which only requires forwarding the + * `spark.hoodie.*` form. + * + * Example: + * {{{ + * SparkConf: spark.hoodie.X = "a", hoodie.Y = "b" // bare hoodie.Y is NOT forwarded + * optParams: hoodie.Z = "c" + * result: hoodie.X = "a" + * hoodie.Z = "c" + * }}} + */ + def collectSparkHoodieConfs(sqlContext: SQLContext, + optParams: Map[String, String]): Map[String, String] = + collectConfsByPrefix(sqlContext, optParams, includeHoodiePrefix = false) + + private def collectConfsByPrefix(sqlContext: SQLContext, + optParams: Map[String, String], + includeHoodiePrefix: Boolean): Map[String, String] = { + val sparkConfs = sqlContext.getAllConfs.filter { + case (key, _) => + key.startsWith(SPARK_HOODIE_PREFIX) || (includeHoodiePrefix && key.startsWith(HOODIE_PREFIX)) + } + normalizeSparkHoodiePrefix(sparkConfs) ++ optParams + } + + /** + * Strips the `spark.` prefix from `spark.hoodie.*` keys so downstream code only sees + * canonical `hoodie.*` keys. If both `spark.hoodie.X` and `hoodie.X` are present, the + * latter wins (explicit options/configs override the SparkConf-prefixed form). + * + * The function is idempotent: running it on an already-normalized map is a no-op. + * Both `collectHoodieAndSparkHoodieConfs` (the entry-point helper) and + * `parametersWithReadDefaults` / `parametersWithWriteDefaults` (the per-path defaulting + * helpers) call it; this defense-in-depth ensures callers that bypass + * `collectHoodieAndSparkHoodieConfs` (e.g., SQL `ALTER TABLE` paths) still get + * normalized configs. + */ + def normalizeSparkHoodiePrefix(parameters: Map[String, String]): Map[String, String] = { + val rekeyedSparkHoodie = parameters.collect { + case (key, value) if key.startsWith(SPARK_HOODIE_PREFIX) => + (key.stripPrefix(SPARK_PREFIX), value) + } + val nonSparkHoodie = parameters.filterNot(_._1.startsWith(SPARK_HOODIE_PREFIX)) + rekeyedSparkHoodie ++ nonSparkHoodie + } + // put all the configs with alternatives here private val allConfigsWithAlternatives = List( DataSourceReadOptions.QUERY_TYPE, @@ -1132,13 +1211,8 @@ object DataSourceOptionsHelper { // 2) spark.hoodie.* (normalized to hoodie.*) // 3) hoodie.* / explicit data source options // NOTE: If both spark.hoodie.X and hoodie.X are set, hoodie.X wins. - val normalizedSparkHoodieConfigs = parameters.collect { - case (key, value) if key.startsWith(SPARK_HOODIE_PREFIX) => (key.stripPrefix(SPARK_PREFIX), value) - } - val paramsWithoutSparkHoodie = parameters.filterNot(_._1.startsWith(SPARK_HOODIE_PREFIX)) - val paramsWithGlobalProps = DFSPropertiesConfiguration.getGlobalProps.asScala.toMap ++ - normalizedSparkHoodieConfigs ++ - paramsWithoutSparkHoodie + val normalized = normalizeSparkHoodiePrefix(parameters) + val paramsWithGlobalProps = DFSPropertiesConfiguration.getGlobalProps.asScala.toMap ++ normalized val queryType = paramsWithGlobalProps.get(IS_QUERY_AS_RO_TABLE) .map(is => if (is.toBoolean) QUERY_TYPE_READ_OPTIMIZED_OPT_VAL else QUERY_TYPE_SNAPSHOT_OPT_VAL) .getOrElse(paramsWithGlobalProps.getOrElse(QUERY_TYPE.key, QUERY_TYPE.defaultValue())) diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala index 2682de70f1fa..48fb989a2c9f 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/DefaultSource.scala @@ -107,17 +107,14 @@ class DefaultSource extends RelationProvider throw new HoodieException("Glob paths are not supported for read paths as of Hudi 1.2.0") } - val hoodieAndSparkHoodieSqlConfs = sqlContext.getAllConfs.filter { - case (key, _) => key.startsWith("hoodie.") || key.startsWith("spark.hoodie.") - } // Add default options for unspecified read options keys. // Effective precedence (low -> high): // 1) global DFS props - // 2) spark.hoodie.* SQL confs (normalized in parametersWithReadDefaults) + // 2) spark.hoodie.* SQL confs (normalized to hoodie.* in collectHoodieAndSparkHoodieConfs) // 3) hoodie.* SQL confs // 4) explicit DataFrame/DataSource options val parameters = DataSourceOptionsHelper.parametersWithReadDefaults( - hoodieAndSparkHoodieSqlConfs ++ optParams) + DataSourceOptionsHelper.collectHoodieAndSparkHoodieConfs(sqlContext, optParams)) // Get the table base path val tablePath = DataSourceUtils.getTablePath(storage, Seq(new StoragePath(path.get)).asJava) @@ -173,11 +170,20 @@ class DefaultSource extends RelationProvider mode: SaveMode, optParams: Map[String, String], df: DataFrame): BaseRelation = { + // Pull `spark.hoodie.*` from SparkConf, normalize to canonical `hoodie.*`, and merge + // with explicit options (explicit options win), so configs like + // `--conf spark.hoodie.datasource.hive_sync.use_spark_catalog=true` are honored on + // writes too. Unlike the read path, we deliberately do NOT forward bare `hoodie.*` + // session confs here: the DataFrame write path historically honored only the explicit + // `.option(...)` map, and injecting ambient session `hoodie.*` state would silently + // change every write. `HoodieSparkSqlWriter` and downstream callers see only canonical keys. + val effectiveOpts = + DataSourceOptionsHelper.collectSparkHoodieConfs(sqlContext, optParams) try { - if (optParams.get(OPERATION.key).contains(BOOTSTRAP_OPERATION_OPT_VAL)) { - HoodieSparkSqlWriter.bootstrap(sqlContext, mode, optParams, df) + if (effectiveOpts.get(OPERATION.key).contains(BOOTSTRAP_OPERATION_OPT_VAL)) { + HoodieSparkSqlWriter.bootstrap(sqlContext, mode, effectiveOpts, df) } else { - val (success, _, _, _, _, _) = HoodieSparkSqlWriter.write(sqlContext, mode, optParams, df) + val (success, _, _, _, _, _) = HoodieSparkSqlWriter.write(sqlContext, mode, effectiveOpts, df) if (!success) { throw new HoodieException("Failed to write to Hudi") } diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala index e76af4a2733d..54459ff61e00 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieWriterUtils.scala @@ -53,8 +53,12 @@ object HoodieWriterUtils { * Add default options for unspecified write options keys. */ def parametersWithWriteDefaults(parameters: Map[String, String]): Map[String, String] = { + // Strip the `spark.` prefix from `spark.hoodie.*` keys so write/hive_sync configs + // forwarded from SparkConf reach Hudi under their canonical names. Mirrors what + // parametersWithReadDefaults does for the read path. + val normalizedParams = DataSourceOptionsHelper.normalizeSparkHoodiePrefix(parameters) val globalProps = DFSPropertiesConfiguration.getGlobalProps.asScala - val props = TypedProperties.fromMap(parameters.asJava) + val props = TypedProperties.fromMap(normalizedParams.asJava) val hoodieConfig: HoodieConfig = new HoodieConfig(props) hoodieConfig.setDefaultValue(OPERATION) hoodieConfig.setDefaultValue(TABLE_TYPE) @@ -87,7 +91,7 @@ object HoodieWriterUtils { hoodieConfig.setDefaultValue(RECONCILE_SCHEMA) hoodieConfig.setDefaultValue(DROP_PARTITION_COLUMNS) hoodieConfig.setDefaultValue(KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED) - Map() ++ hoodieConfig.getProps.asScala ++ globalProps ++ DataSourceOptionsHelper.translateConfigurations(parameters) + Map() ++ hoodieConfig.getProps.asScala ++ globalProps ++ DataSourceOptionsHelper.translateConfigurations(normalizedParams) } /** diff --git a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestDataSourceOptions.scala b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestDataSourceOptions.scala index 20d61973f213..7dfce2faf571 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestDataSourceOptions.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/hudi/TestDataSourceOptions.scala @@ -22,9 +22,11 @@ package org.apache.hudi import org.apache.hudi.common.config.{DFSPropertiesConfiguration, HoodieCommonConfig} import org.apache.hudi.common.table.HoodieTableConfig +import org.apache.spark.sql.SQLContext import org.junit.jupiter.api.AfterEach -import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue} +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue} import org.junit.jupiter.api.Test +import org.mockito.Mockito.{mock, when} class TestDataSourceOptions { @Test @@ -91,6 +93,133 @@ class TestDataSourceOptions { assertEquals(DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL, params3(DataSourceReadOptions.QUERY_TYPE.key)) } + @Test + def testNormalizeSparkHoodiePrefixStripsSparkPrefix(): Unit = { + val result = DataSourceOptionsHelper.normalizeSparkHoodiePrefix(Map( + "spark.hoodie.datasource.query.type" -> "snapshot", + "spark.hoodie.datasource.hive_sync.use_spark_catalog" -> "true", + "non.hoodie.key" -> "ignored" + )) + + assertEquals("snapshot", result("hoodie.datasource.query.type")) + assertEquals("true", result("hoodie.datasource.hive_sync.use_spark_catalog")) + assertEquals("ignored", result("non.hoodie.key")) + assertFalse(result.contains("spark.hoodie.datasource.query.type")) + assertFalse(result.contains("spark.hoodie.datasource.hive_sync.use_spark_catalog")) + } + + @Test + def testNormalizeSparkHoodiePrefixPrefersHoodieOverSparkHoodie(): Unit = { + val result = DataSourceOptionsHelper.normalizeSparkHoodiePrefix(Map( + "spark.hoodie.datasource.query.type" -> DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL, + "hoodie.datasource.query.type" -> DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL + )) + + assertEquals(DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL, + result("hoodie.datasource.query.type")) + assertFalse(result.contains("spark.hoodie.datasource.query.type")) + } + + @Test + def testNormalizeSparkHoodiePrefixIsIdempotent(): Unit = { + val once = DataSourceOptionsHelper.normalizeSparkHoodiePrefix(Map( + "spark.hoodie.datasource.query.type" -> "snapshot", + "hoodie.other.key" -> "v" + )) + val twice = DataSourceOptionsHelper.normalizeSparkHoodiePrefix(once) + + assertEquals(once, twice) + } + + @Test + def testCollectHoodieAndSparkHoodieConfsReturnsCanonicalKeys(): Unit = { + val sqlContext = mock(classOf[SQLContext]) + when(sqlContext.getAllConfs).thenReturn(Map( + "spark.hoodie.datasource.query.type" -> DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL, + "hoodie.datasource.write.operation" -> "upsert", + "spark.sql.shuffle.partitions" -> "200" // non-hoodie, must be filtered out + )) + + val result = DataSourceOptionsHelper.collectHoodieAndSparkHoodieConfs(sqlContext, Map.empty) + + assertEquals(DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL, + result("hoodie.datasource.query.type")) + assertEquals("upsert", result("hoodie.datasource.write.operation")) + assertFalse(result.contains("spark.hoodie.datasource.query.type")) + assertFalse(result.contains("spark.sql.shuffle.partitions")) + } + + @Test + def testCollectHoodieAndSparkHoodieConfsExplicitOptionsWin(): Unit = { + val sqlContext = mock(classOf[SQLContext]) + when(sqlContext.getAllConfs).thenReturn(Map( + "spark.hoodie.datasource.write.operation" -> "insert", + "hoodie.datasource.write.precombine.field" -> "ts_from_conf" + )) + + val result = DataSourceOptionsHelper.collectHoodieAndSparkHoodieConfs(sqlContext, Map( + "hoodie.datasource.write.operation" -> "upsert", // explicit overrides spark.hoodie.* + "hoodie.datasource.write.precombine.field" -> "ts_from_options" // explicit overrides hoodie.* + )) + + assertEquals("upsert", result("hoodie.datasource.write.operation")) + assertEquals("ts_from_options", result("hoodie.datasource.write.precombine.field")) + } + + @Test + def testCollectSparkHoodieConfsForwardsOnlySparkPrefix(): Unit = { + val sqlContext = mock(classOf[SQLContext]) + when(sqlContext.getAllConfs).thenReturn(Map( + "spark.hoodie.datasource.hive_sync.use_spark_catalog" -> "true", // forwarded (the --conf use case) + "hoodie.logfile.data.block.format" -> "parquet", // bare hoodie.* must NOT leak into writes + "hoodie.datasource.write.operation" -> "bulk_insert", // bare hoodie.* must NOT leak into writes + "spark.sql.shuffle.partitions" -> "200" // non-hoodie, filtered out + )) + + val result = DataSourceOptionsHelper.collectSparkHoodieConfs(sqlContext, Map.empty) + + assertEquals("true", result("hoodie.datasource.hive_sync.use_spark_catalog")) + assertFalse(result.contains("hoodie.logfile.data.block.format")) + assertFalse(result.contains("hoodie.datasource.write.operation")) + assertFalse(result.contains("spark.sql.shuffle.partitions")) + } + + @Test + def testCollectSparkHoodieConfsExplicitOptionsWin(): Unit = { + val sqlContext = mock(classOf[SQLContext]) + when(sqlContext.getAllConfs).thenReturn(Map( + "spark.hoodie.datasource.write.operation" -> "insert" + )) + + val result = DataSourceOptionsHelper.collectSparkHoodieConfs(sqlContext, Map( + "hoodie.datasource.write.operation" -> "upsert" // explicit overrides spark.hoodie.* + )) + + assertEquals("upsert", result("hoodie.datasource.write.operation")) + assertFalse(result.contains("spark.hoodie.datasource.write.operation")) + } + + @Test + def testWriteDefaultsSupportSparkHoodieConfigs(): Unit = { + val params = HoodieWriterUtils.parametersWithWriteDefaults(Map( + "spark.hoodie.datasource.write.operation" -> "upsert" + )) + + assertEquals("upsert", params(DataSourceWriteOptions.OPERATION.key)) + assertFalse(params.contains("spark.hoodie.datasource.write.operation")) + } + + @Test + def testWriteDefaultsPreferHoodieOverSparkHoodieWhenBothSet(): Unit = { + val params = HoodieWriterUtils.parametersWithWriteDefaults(Map( + "spark.hoodie.datasource.write.operation" -> "insert", + "hoodie.datasource.write.operation" -> "upsert" + )) + + assertEquals("upsert", params(DataSourceWriteOptions.OPERATION.key)) + assertFalse(params.contains("spark.hoodie.datasource.write.operation")) + } + @AfterEach def cleanup(): Unit = { DFSPropertiesConfiguration.clearGlobalProps()
