This is an automated email from the ASF dual-hosted git repository.
nsivabalan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new b3d45629581d fix(spark)!: forward session settings to streaming sinks
(#20455)
b3d45629581d is described below
commit b3d45629581df9dfbbee5994ee438954e343825f
Author: Danny Chan <[email protected]>
AuthorDate: Thu Oct 29 04:30:41 2026 +0800
fix(spark)!: forward session settings to streaming sinks (#20455)
Streaming sinks now honor the session's spark.hoodie.* settings, as
batch DataFrame writes already do. Settings are captured once when the
sink is created and drive multi-writer validation, checkpoint
deduplication and every micro-batch write. Precedence: explicit writer
options, then session spark.hoodie.*, then defaults. Bare hoodie.*
session keys are not forwarded. Table identity is validated against an
existing table when the sink is created, including on replay.
Also fixes explicit spark.hoodie.* options being overridden by the
session value in collectSparkHoodieConfs.
BREAKING CHANGE: existing writeStream applications pick up ambient
spark.hoodie.* settings (e.g. --conf
spark.hoodie.datasource.write.operation).
Unset them or pass explicit writer options to keep the previous behavior.
Closes #20403.
---
hudi-spark-datasource/README.md | 13 ++
.../scala/org/apache/hudi/DataSourceOptions.scala | 2 +-
.../main/scala/org/apache/hudi/DefaultSource.scala | 13 +-
.../org/apache/hudi/HoodieStreamingSink.scala | 4 +
.../org/apache/hudi/TestDataSourceOptions.scala | 12 ++
.../hudi/functional/TestStreamingSinkOptions.scala | 179 +++++++++++++++++++++
6 files changed, 220 insertions(+), 3 deletions(-)
diff --git a/hudi-spark-datasource/README.md b/hudi-spark-datasource/README.md
index 413eeaf33e0d..e8813576a813 100644
--- a/hudi-spark-datasource/README.md
+++ b/hudi-spark-datasource/README.md
@@ -60,6 +60,19 @@ ZSTD. The GZIP write default on older Spark versions does
not eliminate the risk
files produced by another engine. Explicitly setting
`hoodie.parquet.compression.codec` overrides
the version-specific default.
+## Streaming session configuration compatibility
+
+Starting with 1.3.0, Hudi Structured Streaming writes honor the active
session's `spark.hoodie.*`
+settings (including `--conf spark.hoodie.*` passed at submit time), just like
DataFrame batch writes.
+Previously the streaming sink ignored them. When upgrading an existing
`writeStream` application,
+audit ambient `spark.hoodie.*` settings, or pass explicit writer options to
keep the previous behavior,
+e.g. `.option("hoodie.datasource.write.operation", "upsert")`.
+
+Precedence: explicit writer/storage options, then session `spark.hoodie.*`
settings, then Hudi defaults.
+Settings are captured once when the sink is created; later `SET` statements do
not change a running
+query. Bare `hoodie.*` session settings are not forwarded. Table identity
settings must still agree with
+the existing table. See the Streaming Writes guide on the Hudi website for
details.
+
## Key Features
- **DataSource V1 Support**: Full integration with Spark's DataSource API
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 00de045d9e3b..ffdd58f85c6d 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
@@ -1113,7 +1113,7 @@ object DataSourceOptionsHelper {
case (key, _) =>
key.startsWith(SPARK_HOODIE_PREFIX) || (includeHoodiePrefix &&
key.startsWith(HOODIE_PREFIX))
}
- normalizeSparkHoodiePrefix(sparkConfs) ++ optParams
+ normalizeSparkHoodiePrefix(sparkConfs) ++
normalizeSparkHoodiePrefix(optParams)
}
/**
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 0977f50a074b..201ec3f9f7da 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
@@ -201,10 +201,19 @@ class DefaultSource extends RelationProvider
optParams: Map[String, String],
partitionColumns: Seq[String],
outputMode: OutputMode): Sink = {
- validateMultiWriterConfigs(optParams)
+ // Capture the flow's SQLConf once. The same canonical options must drive
validation,
+ // checkpoint deduplication and every micro-batch write, even if the
session changes later.
+ val sessionOpts =
DataSourceOptionsHelper.collectSparkHoodieConfs(sqlContext, Map.empty)
+ val effectiveOpts =
DataSourceOptionsHelper.collectSparkHoodieConfs(sqlContext, optParams)
+ val overriddenKeys = sessionOpts.keys.filter(key => effectiveOpts.get(key)
!= sessionOpts.get(key)).toSeq.sorted
+ if (overriddenKeys.nonEmpty) {
+ log.warn(s"Explicit streaming writer/storage options override session
spark.hoodie.* settings for: ${overriddenKeys.mkString(", ")}. "
+ + "Check legacy table storage options if a flow's SET does not take
effect.")
+ }
+ validateMultiWriterConfigs(effectiveOpts)
new HoodieStreamingSink(
sqlContext,
- optParams,
+ effectiveOpts,
partitionColumns,
outputMode)
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
index f33c71420f34..7410bbf29011 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
@@ -72,6 +72,10 @@ class HoodieStreamingSink(sqlContext: SQLContext,
Option.empty
}
}
+ // Validate captured flow settings before starting a query, including when
its first batch
+ // is already committed and would otherwise bypass the writer's table
identity validation.
+ metaClient.foreach(client =>
HoodieWriterUtils.validateTableConfig(sqlContext.sparkSession, options,
client.getTableConfig))
+
private val retryCnt = options.getOrElse(STREAMING_RETRY_CNT.key,
STREAMING_RETRY_CNT.defaultValue).toInt
private val retryIntervalMs =
options.getOrElse(STREAMING_RETRY_INTERVAL_MS.key,
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 7dfce2faf571..daa373da4413 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
@@ -199,6 +199,18 @@ class TestDataSourceOptions {
assertFalse(result.contains("spark.hoodie.datasource.write.operation"))
}
+ @Test
+ def testPrefixedExplicitOptionsOverrideSessionConfs(): Unit = {
+ val sqlContext = mock(classOf[SQLContext])
+ when(sqlContext.getAllConfs).thenReturn(Map(
+ "spark.hoodie.datasource.write.operation" -> "insert"
+ ))
+ val result = DataSourceOptionsHelper.collectSparkHoodieConfs(sqlContext,
Map(
+ "spark.hoodie.datasource.write.operation" -> "delete"
+ ))
+ assertEquals(Map("hoodie.datasource.write.operation" -> "delete"), result)
+ }
+
@Test
def testWriteDefaultsSupportSparkHoodieConfigs(): Unit = {
val params = HoodieWriterUtils.parametersWithWriteDefaults(Map(
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSinkOptions.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSinkOptions.scala
new file mode 100644
index 000000000000..03e98999cd2f
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestStreamingSinkOptions.scala
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.functional
+
+import org.apache.hudi.{DefaultSource, HoodieStreamingSink}
+import org.apache.hudi.DataSourceWriteOptions._
+import org.apache.hudi.common.model.WriteConcurrencyMode
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.testutils.HoodieTestUtils
+import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.storage.StoragePath
+import org.apache.hudi.testutils.HoodieSparkClientTestBase
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery, Trigger}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows,
assertTrue}
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.CsvSource
+
+import scala.collection.JavaConverters._
+
+class TestStreamingSinkOptions extends HoodieSparkClientTestBase {
+ private val options = Map(
+ HoodieWriteConfig.TBL_NAME.key -> "streaming_options",
+ RECORDKEY_FIELD.key -> "id",
+ ORDERING_FIELDS.key -> "ts",
+ PARTITIONPATH_FIELD.key -> "",
+ "hoodie.insert.shuffle.parallelism" -> "1",
+ "hoodie.upsert.shuffle.parallelism" -> "1",
+ "hoodie.delete.shuffle.parallelism" -> "1",
+ "hoodie.metadata.enable" -> "false")
+
+ private def rows(session: SparkSession): DataFrame = session.sql("select 1
as id, 1L as ts")
+
+ private def meta(path: String): HoodieTableMetaClient =
HoodieTestUtils.createMetaClient(storage, path)
+
+ private def operations(path: String): Seq[String] = {
+ val timeline =
meta(path).getActiveTimeline.getCommitsTimeline.filterCompletedInstants()
+ timeline.getInstants.asScala.map(instant =>
timeline.readCommitMetadata(instant).getOperationType.name()).toSeq
+ }
+
+ private def start(session: SparkSession, source: String, target: String,
checkpoint: String,
+ extraOptions: Map[String, String] = Map.empty, once:
Boolean = true): StreamingQuery = {
+ session.readStream.schema(rows(session).schema).json(source).writeStream
+ .format("org.apache.hudi").options(options ++ extraOptions)
+ .option("checkpointLocation", checkpoint)
+ .trigger(if (once) Trigger.Once() else Trigger.ProcessingTime(100))
+ .start(target)
+ }
+
+ private def run(session: SparkSession, name: String, target: String,
+ extraOptions: Map[String, String] = Map.empty): Unit = {
+ val source = s"$basePath/source_$name"
+ rows(session).write.json(source)
+ val query = start(session, source, target, s"$basePath/checkpoint_$name",
extraOptions)
+ try {
+ assertTrue(query.awaitTermination(60000), "Streaming query did not
finish within 60 seconds")
+ } finally {
+ query.stop()
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource(Array("none,none,UPSERT", "insert,none,INSERT",
"insert,upsert,UPSERT"))
+ def testSessionOperationAndExplicitPrecedence(sessionOperation: String,
explicitOperation: String,
+ expected: String): Unit = {
+ val session = sqlContext.sparkSession.newSession()
+ // Bare hoodie.* session state must not leak into the writer, even when no
prefixed setting exists.
+ session.conf.set(OPERATION.key, "delete")
+ if (sessionOperation != "none") {
+ session.sql(s"SET spark.${OPERATION.key}=$sessionOperation")
+ }
+ val explicit = if (explicitOperation == "none") Map.empty[String, String]
else Map(OPERATION.key -> explicitOperation)
+ val target = s"$basePath/target"
+ run(session, "first", target, explicit)
+ assertEquals(Seq(expected), operations(target))
+ assertEquals(1L, session.read.format("hudi").load(target).count())
+ // A fresh Spark checkpoint replays batch 0; the sink must still recognize
the committed batch.
+ run(session, "replay", target, explicit)
+ assertEquals(Seq(expected), operations(target))
+ }
+
+ @Test
+ def testPrefixedExplicitDeleteAgreesWithCheckpointDeduplication(): Unit = {
+ val session = sqlContext.sparkSession.newSession()
+ val target = s"$basePath/target"
+ run(session, "insert", target)
+ session.conf.set(s"spark.${OPERATION.key}", "upsert")
+ // The delete stream also starts at batch 0. A noncanonical operation at
canSkipBatch
+ // would treat it as upsert and skip it, even though the writer would
resolve delete.
+ run(session, "delete", target, Map(s"spark.${OPERATION.key}" -> "delete"))
+ assertEquals(Seq("UPSERT", "DELETE"), operations(target))
+ assertEquals(0L, session.read.format("hudi").load(target).count())
+ }
+
+ @Test
+ def testFlowSettingsAreCapturedAtSinkCreation(): Unit = {
+ val session = sqlContext.sparkSession.newSession()
+ val source = s"$basePath/source"
+ storage.createDirectory(new StoragePath(source))
+ session.sql(s"SET spark.${OPERATION.key}=insert")
+ val insertQuery = start(session, source, s"$basePath/inserts",
s"$basePath/insert_checkpoint", once = false)
+ try {
+ session.sql(s"SET spark.${OPERATION.key}=upsert")
+ val upsertQuery = start(session, source, s"$basePath/upserts",
s"$basePath/upsert_checkpoint", once = false)
+ try {
+ // Both queries must retain their captured settings after the shared
session changes again.
+ session.sql(s"SET spark.${OPERATION.key}=delete")
+ rows(session).write.mode("append").json(source)
+ insertQuery.processAllAvailable()
+ upsertQuery.processAllAvailable()
+ rows(session).write.mode("append").json(source)
+ insertQuery.processAllAvailable()
+ upsertQuery.processAllAvailable()
+ assertEquals(Seq("INSERT", "INSERT"), operations(s"$basePath/inserts"))
+ assertEquals(Seq("UPSERT", "UPSERT"), operations(s"$basePath/upserts"))
+ assertEquals(2L,
session.read.format("hudi").load(s"$basePath/inserts").count())
+ assertEquals(1L,
session.read.format("hudi").load(s"$basePath/upserts").count())
+ } finally {
+ upsertQuery.stop()
+ }
+ } finally {
+ insertQuery.stop()
+ }
+ }
+
+ @Test
+ def testMultiWriterValidationUsesCapturedSessionOptions(): Unit = {
+ val session = sqlContext.sparkSession.newSession()
+ val source = new DefaultSource
+ val opts = options + ("path" -> s"$basePath/target")
+ session.conf.set(s"spark.${HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key}",
WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.name())
+ val error = assertThrows(classOf[IllegalStateException], () =>
+ source.createSink(session.sqlContext, opts, Seq.empty,
OutputMode.Append()))
+ assertTrue(error.getMessage.contains(STREAMING_CHECKPOINT_IDENTIFIER.key))
+ session.conf.set(s"spark.${STREAMING_CHECKPOINT_IDENTIFIER.key}", "flow")
+ assertTrue(source.createSink(session.sqlContext, opts, Seq.empty,
OutputMode.Append()).isInstanceOf[HoodieStreamingSink])
+ session.conf.unset(s"spark.${STREAMING_CHECKPOINT_IDENTIFIER.key}")
+ assertTrue(source.createSink(session.sqlContext,
+ opts + (s"spark.${HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key}" ->
WriteConcurrencyMode.SINGLE_WRITER.name()),
+ Seq.empty, OutputMode.Append()).isInstanceOf[HoodieStreamingSink])
+ }
+
+ @Test
+ def testCapturedSettingsCannotChangeExistingTableIdentity(): Unit = {
+ val session = sqlContext.sparkSession.newSession()
+ val target = s"$basePath/target"
+ run(session, "first", target)
+ session.conf.set(s"spark.${RECORDKEY_FIELD.key}", "ts")
+ val source = new DefaultSource
+ val opts = (options - RECORDKEY_FIELD.key) + ("path" -> target)
+ val error = assertThrows(classOf[HoodieException], () =>
+ source.createSink(session.sqlContext, opts, Seq.empty,
OutputMode.Append()))
+ assertTrue(error.getMessage.contains("RecordKey"))
+ // Explicit identity wins over ambient settings, but must still agree with
the persisted table.
+ assertTrue(source.createSink(session.sqlContext, opts +
(RECORDKEY_FIELD.key -> "id"),
+ Seq.empty, OutputMode.Append()).isInstanceOf[HoodieStreamingSink])
+ assertThrows(classOf[HoodieException], () =>
+ source.createSink(session.sqlContext, opts + (RECORDKEY_FIELD.key ->
"ts"), Seq.empty, OutputMode.Append()))
+ assertEquals(Seq("UPSERT"), operations(target))
+ }
+}
