[SPARK-27340][SS][2.4] Alias on TimeWindow expression may cause watermark metadata lost

[dongjoon-hyun]

This is a backport of #28326 . The authorship is kept. (Credit to @LiangchangZ and @xuanyuanking )

What changes were proposed in this pull request?

Make metadata propagatable between Aliases.

Why are the changes needed?

In Structured Streaming, we added an Alias for TimeWindow by default.

spark/sql/core/src/main/scala/org/apache/spark/sql/functions.scala

Lines 3272 to 3273 in 590b9a0

	TimeWindow(timeColumn.expr, windowDuration, slideDuration, startTime)
	}.as("window")

For some cases like stream join with watermark and window, users need to add an alias for convenience(we also added one in StreamingJoinSuite). The current metadata handling logic for as will lose the watermark metadata

spark/sql/core/src/main/scala/org/apache/spark/sql/Column.scala

Lines 1049 to 1054 in 590b9a0

	def name(alias: String): Column = withExpr {
	normalizedExpr() match {
	case ne: NamedExpression => Alias(expr, alias)(explicitMetadata = Some(ne.metadata))
	case other => Alias(other, alias)()
	}
	}

and finally cause the AnalysisException:

Stream-stream outer join between two streaming DataFrame/Datasets is not supported without a watermark in the join keys, or a watermark on the nullable side and an appropriate range condition

Does this PR introduce any user-facing change?

Bugfix for an alias on time window with watermark.

How was this patch tested?

New UTs added. One for the functionality and one for explaining the common scenario.

[dongjoon-hyun]

cc @cloud-fan , @zsxwing , @LiangchangZ , @xuanyuanking , @HeartSaVioR .

[dongjoon-hyun]

cc @holdenk since she is the release manager of 2.4.6.
Also, cc @dbtsai

Without this, we cannot run long running Structure Streaming with State queries due to the lack of watermark.

[holdenk]

From a backporting perspective it seems like this is maybe a bug given that name doesn't appear to be copying the metadata even without an explicit metadata set. Do I have the right context @dongjoon-hyun ?

[dongjoon-hyun]

Yes. Correct. In name function, the metadata loss occurs. Without the metadata, some structured query fails at Analysis Stage, and another structure queries becomes growing the internal group by state indefinitely.

[gatorsmile]

@dongjoon-hyun @holdenk My understanding is different from what you said above.

It sounds like the metadata is not available for TimeWindow expression when the function name(alias: String) copies the metadata to explicitMetadata. Thus, we do not set it when calling the APIs in Column.

[dongjoon-hyun]

@gatorsmile . Are you assuming NamedExpression only?

[gatorsmile]

Any other Expressions except NamedExpression has the field Metadata?

[dongjoon-hyun]

Got it. You are right.

[HeartSaVioR]

It sounds like the metadata is not available for TimeWindow expression when the function name(alias: String) copies the metadata to explicitMetadata. Thus, we do not set it when calling the APIs in Column.

That's right, but I don't see the reason of the fix only because we encountered an actual problem from there. This has been simply wrong because we set explicit metadata on Alias while it's not asked to overwrite. Javadoc of the Alias clearly describes the usage of explicitMetadata:

@param explicitMetadata Explicit metadata associated with this alias that overwrites child's.

(that's why I couldn't agree with the initial proposal from #28326.)

[HeartSaVioR]

That said, I'd support porting back this to 2.4 as it fixes the wrong code. This change might impact broader audiences (so technically saying the PR/commit title doesn't represent actual change, sorry I should have found this earlier) hence the risk might be bigger than we may imagine, but I couldn't imagine the case which relies on the previous behavior (bug) to make it work. If we can imagine anything then it might be a signal to reconsider.

[SparkQA]

Test build #121930 has finished for PR 28377 at commit bb09254.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #121936 has finished for PR 28377 at commit b6e410b.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[holdenk]

Are we still planning on backporting this or not? I see you deleted the branch so I'm assuming this particular attempt at backporting is abandoned but the root issue does seem important.

[dongjoon-hyun]

We have no viable solution for branch-2.4. You can proceed 2.4.6 release without this.