vranes opened a new pull request, #59834: URL: https://github.com/apache/spark/pull/59834
### What changes were proposed in this pull request? JIRA: [SPARK-60582](https://issues.apache.org/jira/browse/SPARK-60582) Resolve DataFrame ASOF keys directly during analysis and pass those keys to the sort-merge operator. This replaces extracting keys from subtraction expressions and shares key validation and comparison-order helpers with SQL ASOF. - Resolve keys in both the fixed-point and single-pass analyzers, including Connect expressions. Preserve explicit casts and decimal values used for nearest matching. - Use text ordering for backward/forward STRING keys and support TIME keys. Reuse SQL ASOF timestamp ordering for mixed LTZ/NTZ keys, including composite keys. - Select the DataFrame execution strategy during analysis and store the decision on the logical node. Build a distance expression only for the correlated-subquery implementation. Remove optimizer reanalysis, configuration replay, and obsolete key-extraction code. - Share tolerance validation so both analyzers report the same error when the tolerance and keys are invalid. ### Why are the changes needed? The existing DataFrame sort-merge path derives its sort keys from subtraction. For STRING keys, subtraction introduces numeric casts, while the matching condition compares the strings as text. For example, text ordering puts `'10'` before `'9'`, but numeric ordering reverses them. Sorting and matching can therefore disagree and select the wrong row. With ANSI off, non-numeric strings become NULL in those numeric sort keys. Deriving keys from arithmetic also ties planning to the shape of the subtraction expression and requires a restriction on mixed LTZ/NTZ timestamps. Resolving keys explicitly lets DataFrame ASOF reuse SQL's timestamp handling and preserve the original values needed for exact nearest gaps. Selecting execution during analysis prevents a later configuration change from sending a plan to an operator for which its keys were not prepared. ### Does this PR introduce _any_ user-facing change? Yes. With `spark.sql.join.sortMergeAsOfJoin.enabled=true`: - Backward/forward STRING matching uses text ordering and the keys' collation; TIME keys are supported. - Mixed LTZ/NTZ keys are accepted, including supported composite keys. - Constant keys and keys with incompatible collations are rejected during analysis. Nearest STRING keys are rejected during analysis with either flag value. The legacy implementation previously attempted implicit numeric conversion; callers that want numeric distance must cast their keys explicitly. The DataFrame flag remains off by default. The legacy implementation retains subtraction-based matching. Changing the flag after a DataFrame is analyzed no longer changes its execution strategy. SQL ASOF continues to always use sort-merge. ### How was this patch tested? The latest run passed 193 tests across: - `RewriteAsOfJoinSuite` - `SortMergeAsOfJoinSuite` - `DataFrameAsOfJoinSuite` - `AsOfJoinSQLSuite` New regressions cover both analyzers, numeric and decimal keys, STRING ordering and collations, TIME keys, explicit casts, mixed timestamps across DST, composite keys, strategy selection, and tolerance error precedence. Existing legacy DataFrame tests were run without adding new legacy-path tests. Additional runs during implementation covered Connect `joinAsOf`, scanner and key-type suites, and SQL ASOF execution/analyzer goldens. Catalyst and SQL scalastyle checks passed. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex CLI 0.164.0 -- 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]
