HeartSaVioR commented on code in PR #58238:
URL: https://github.com/apache/spark/pull/58238#discussion_r3850747934
##########
docs/streaming/apis-on-dataframes-and-datasets.md:
##########
@@ -1293,6 +1293,20 @@ joined <- join(
</div>
+For a **left outer** join, the surviving unmatched left rows are emitted when
the left-side state is
Review Comment:
The documentation is user-facing. Shall we make this be concise and not too
much on the detail? Good to skim through the doc and understand the mood of max
complexity we push in the guide doc, and don't go beyond that.
This change "as it is" isn't consumable for most of the users.
##########
docs/streaming/apis-on-dataframes-and-datasets.md:
##########
@@ -1318,6 +1332,15 @@ constraints must be specified for semi join. This is to
evict unmatched input ro
the engine must know when an input row on left side is not going to match with
anything on right
side in future.
+As with a left outer join, the left state must be evicted so that
never-matched left rows do not
Review Comment:
Same here, this is a user facing doc. Apply the same suggestion here, please.
##########
docs/streaming/apis-on-dataframes-and-datasets.md:
##########
@@ -1293,6 +1293,20 @@ joined <- join(
</div>
+For a **left outer** join, the surviving unmatched left rows are emitted when
the left-side state is
+evicted, and an already-emitted `NULL`-extended row can be invalidated by a
right row that arrives
+later. Correct results therefore require the watermark to be placed so that
(1) the left state is
+actually evicted and (2) both sides are late-filtered on the dimension that
bounds matching.
+Concretely: for an equality join on a watermarked event-time key, that key
must be watermarked on
+**both** sides; for a time range condition, the range bound must relate
watermarked event-time
+columns from both sides. The recommended, always-correct configuration --
watermarking both sides on
+the event-time column used by the join, as in the example above -- satisfies
both. Configurations
+that leave the left state un-evicted (a watermark on only the right side of a
range condition) or
+leave either side unfiltered on the eviction key (a watermark on only one
equality join key) are
+rejected at analysis time. To restore the previous, looser behavior, set
Review Comment:
I wouldn't have that config in the guide doc. It's OK to keep it in
migration guide doc only.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3857,6 +3857,22 @@ object SQLConf {
.checkValue(v => Set(1, 2, 3, 4).contains(v), "Valid versions are 1, 2,
3, and 4")
.createWithDefault(2)
+ val STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED =
+ buildConf("spark.sql.streaming.join.stricterWatermarkRequirements.enabled")
+ .doc("When true, the analyzer enforces stricter watermark-placement
requirements for " +
Review Comment:
`.internal()`, please. This config must not be used beyond quick mitigation
during migration.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3857,6 +3857,22 @@ object SQLConf {
.checkValue(v => Set(1, 2, 3, 4).contains(v), "Valid versions are 1, 2,
3, and 4")
.createWithDefault(2)
+ val STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED =
+ buildConf("spark.sql.streaming.join.stricterWatermarkRequirements.enabled")
+ .doc("When true, the analyzer enforces stricter watermark-placement
requirements for " +
+ "stream-stream left semi and left outer joins, so that the left-side
state which their " +
Review Comment:
Can we fix this for left/right/full outer and left semi while we are here?
Now this is pretty much weird to say we only fix the issue in left outer and
left semi.
##########
docs/streaming/ss-migration-guide.md:
##########
@@ -23,6 +23,10 @@ Note that this migration guide describes the items specific
to Structured Stream
Many items of SQL migration can be applied when migrating Structured Streaming
to higher versions.
Please refer [Migration Guide: SQL, Datasets and
DataFrame](../sql-migration-guide.html).
+## Upgrading from Structured Streaming 4.3 to 4.4
+
+- Since Spark 4.4, stream-stream left semi and left outer joins enforce
stricter watermark-placement requirements at analysis time, so that the
left-side state their output (or bounded state size) depends on is actually
evicted, and, for left outer, so that late rows cannot invalidate an
already-emitted unmatched row. Configurations that previously ran but could
silently produce incorrect results or unbounded state -- for example a
range-condition join whose range bound is not between watermarked attributes on
both sides, or a left outer equality join whose eviction key is not watermarked
on both sides -- now fail with an `AnalysisException`. To restore the previous
behavior, set `spark.sql.streaming.join.stricterWatermarkRequirements.enabled`
to `false`. (See
[SPARK-58904](https://issues.apache.org/jira/browse/SPARK-58904) for more
details.)
Review Comment:
This is too detailed as a user-facing doc; I'd describe as following:
> Since Spark 4.4, stream-stream left outer and left semi joins require
watermarks to be defined on the event-time columns used by time-range join
conditions. Left outer equality joins require watermarks on the corresponding
join keys from both inputs. Queries that do not meet these requirements now
fail analysis instead of potentially producing incomplete results or retaining
state indefinitely. To restore the previous behavior, set
`spark.sql.streaming.join.stricterWatermarkRequirements.enabled` to `false`.
(See [SPARK-58904](https://issues.apache.org/jira/browse/SPARK-58904) for more
details.)
##########
docs/streaming/apis-on-dataframes-and-datasets.md:
##########
@@ -1293,6 +1293,20 @@ joined <- join(
</div>
+For a **left outer** join, the surviving unmatched left rows are emitted when
the left-side state is
Review Comment:
Always push LLM to be concise and brief, and get the mood of content in the
doc before writing it. You also need to be clear about whether the content is
user-facing vs internal, to make a judgement on the result of LLM, not just
blindly picking it up.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3857,6 +3857,22 @@ object SQLConf {
.checkValue(v => Set(1, 2, 3, 4).contains(v), "Valid versions are 1, 2,
3, and 4")
.createWithDefault(2)
+ val STREAMING_JOIN_STRICTER_WATERMARK_REQUIREMENTS_ENABLED =
+ buildConf("spark.sql.streaming.join.stricterWatermarkRequirements.enabled")
+ .doc("When true, the analyzer enforces stricter watermark-placement
requirements for " +
+ "stream-stream left semi and left outer joins, so that the left-side
state which their " +
Review Comment:
This is one of LLM's patterns - config doc does not really need to describe
the implementation detail unless it is required to understand the impact of the
config. Maybe similar level of content in migration guide can put here.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
