nsivabalan commented on code in PR #20455:
URL: https://github.com/apache/hudi/pull/20455#discussion_r4405647681
##########
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 = sessionOpts ++
DataSourceOptionsHelper.normalizeSparkHoodiePrefix(optParams)
Review Comment:
Nit: with the `collectConfsByPrefix` change in this PR, `sessionOpts ++
normalizeSparkHoodiePrefix(optParams)` is the same as
`collectSparkHoodieConfs(sqlContext, optParams)`. Could we use that for
`effectiveOpts`, so the streaming and batch paths share one merge, and keep
`sessionOpts` only for computing `overriddenKeys`?
##########
hudi-spark-datasource/README.md:
##########
@@ -60,6 +60,49 @@ 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 this change, Hudi Structured Streaming writes honor the active
session's
Review Comment:
"Starting with this change" will not mean much once this is merged; could it
say "Starting with 1.3.0"? Also, this section mostly duplicates the website
guide in #20454. The upgrade guidance would be easier to find in the 1.3.0
release notes / migration guide, with the module README keeping a short pointer.
##########
hudi-spark-datasource/README.md:
##########
@@ -60,6 +60,49 @@ 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 this change, Hudi Structured Streaming writes honor the active
session's
+`spark.hoodie.*` settings, just like DataFrame batch writes. For example,
+`SET spark.hoodie.datasource.write.operation=insert` changes the operation of
subsequently
+started streaming sinks to `insert`. Previously the streaming sink ignored
these session settings.
+Audit ambient `spark.hoodie.*` configuration when upgrading an existing
`writeStream` application.
+To retain its previous behavior, unset those settings before starting the
query or supply explicit
+writer options, such as `.option("hoodie.datasource.write.operation",
"upsert")`.
+
+The sink captures settings once at creation, strips the `spark.` prefix, and
uses the resulting
+options for multi-writer validation, checkpoint deduplication and every
micro-batch write.
+Later `SET` statements do not change an already running sink. Explicit
writer/storage options
+override session settings; session settings override defaults. Explicit
options using either
+`hoodie.*` or `spark.hoodie.*` are supported; when both forms occur in the
explicit options,
+the `hoodie.*` form wins. Bare `hoodie.*` session keys are not forwarded.
+
+Table identity must agree with the persisted Hudi table, even if an explicit
option overrides
+a session value. Conflicting record keys, ordering fields, partition fields,
key generators or
+merge settings fail validation when a sink opens an existing table, including
on checkpoint replay.
+Settings required for multi-writer operation, including a distinct streaming
checkpoint identifier,
+must also be present in the effective options.
+
+For Spark Declarative Pipelines (RFC-112), keep operation and writer tuning in
flow-scoped
+`SET spark.hoodie.*` statements, not shared table properties. SDP captures the
active settings
+at each flow definition. A `SET` remains active for following definitions
until changed.
+New table properties should describe table identity. Legacy storage options
still take precedence
+over flow settings; the sink logs the conflicting keys and identifies the
writer/storage option
+as the winner. Remove the legacy behavior option from the table's storage
options before expecting
+a flow `SET` to control it. Configuration values are not included in this
diagnostic.
+
+The [flow settings integration
scenario](../hudi-integ-test/spark-pipelines/flow_settings.py)
Review Comment:
`flow_settings.py` needs the harness from #20402, which is now up as #20453
and still open, and nothing in CI runs it. Until #20402 lands it is unrunnable
from this tree (hence `--harness-dir` pointing at another checkout). Could we
move the scenario into #20402, or add it in a follow-up once the harness is
merged?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
