This is an automated email from the ASF dual-hosted git repository.
uros-b pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 9062e03c0917 [SPARK-58903][SQL][TEST] Add BIN BY tests for resolving
re-output DISTRIBUTE and metadata columns
9062e03c0917 is described below
commit 9062e03c091717d184057ed6dc56782e6ea3aedc
Author: Nikolina Vraneš <[email protected]>
AuthorDate: Thu Aug 20 14:47:14 2026 +0000
[SPARK-58903][SQL][TEST] Add BIN BY tests for resolving re-output
DISTRIBUTE and metadata columns
### What changes were proposed in this pull request?
Test-only. Adds three `BinBySuite` cases covering how a re-output
`DISTRIBUTE UNIFORM` column behaves in `ORDER BY` / `SELECT`. A `DISTRIBUTE
UNIFORM` column is re-output as a produced attribute (SPARK-57858): the scaled
value gets a fresh expression id, drops the input's qualifier and metadata, and
replaces the input column in place.
- a bare `ORDER BY` on a re-output DISTRIBUTE column that is not in the
`SELECT` list resolves;
- a source-qualified reference to a re-output DISTRIBUTE column
(`metrics.value`) fails with `UNRESOLVED_COLUMN.WITH_SUGGESTION`, since the
produced column no longer carries the source qualifier;
- a file-source `_metadata` column stays reachable through `BIN BY` in both
`ORDER BY` and `SELECT`.
No production code change. Sub-task of SPARK-57133.
### Why are the changes needed?
The interaction between the re-output (produced-attribute) DISTRIBUTE
columns and hidden-output column resolution was not covered: a bare `ORDER BY`
on a non-projected re-output column, a source-qualified reference to one, and
metadata-column reachability through the operator. These tests pin the intended
behavior and guard against regressions.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New unit tests. `build/sbt "sql/testOnly org.apache.spark.sql.BinBySuite"`.
### Was this patch authored or co-authored using generative AI tooling?
<complete per the ASF Generative Tooling Guidance before opening the PR>
Closes #58164 from vranes/bin-by-distribute-resolution-tests.
Authored-by: Nikolina Vraneš <[email protected]>
Signed-off-by: Uros Bojanic <[email protected]>
(cherry picked from commit d54b5f384ac384be7769c2fbb9137701759dadc8)
Signed-off-by: Uros Bojanic <[email protected]>
---
.../scala/org/apache/spark/sql/BinBySuite.scala | 74 ++++++++++++++++++++++
1 file changed, 74 insertions(+)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
index 12867a5786a3..9a4001503619 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
@@ -493,4 +493,78 @@ class BinBySuite extends QueryTest with SharedSparkSession
{
Array(ExpectedContext(fragment = "ORDER BY t.value", start = 271,
stop = 286)))
}
}
+
+ test("BIN BY resolves a bare ORDER BY on a re-output DISTRIBUTE column that
is not projected") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ // A bare ORDER BY on the re-output DISTRIBUTE column resolves even when
it is not projected.
+ val df = spark.sql(
+ """SELECT bin_start
+ |FROM VALUES
+ | (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01
00:10:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY value""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(tsAt("2024-01-01 00:00:00")),
+ Row(tsAt("2024-01-01 00:05:00"))))
+ }
+ }
+
+ test("BIN BY rejects a source-qualified reference to a re-output DISTRIBUTE
column") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ // The re-output DISTRIBUTE column drops its source qualifier, so
`metrics.value` fails.
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.sql(
+ """SELECT metrics.value
+ |FROM VALUES
+ | (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01
00:10:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00'
+ | DISTRIBUTE UNIFORM (value))""".stripMargin)
+ },
+ condition = "UNRESOLVED_COLUMN.WITH_SUGGESTION",
+ parameters = Map(
+ "objectName" -> "`metrics`.`value`",
+ "proposal" -> ("`metrics`.`ts_end`, `value`, `bin_start`, " +
+ "`metrics`.`ts_start`, `bin_end`")),
+ queryContext =
+ Array(ExpectedContext(fragment = "metrics.value", start = 7, stop =
19)))
+ }
+ }
+
+ test("BIN BY keeps a file-source metadata column reachable in ORDER BY and
SELECT") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ withTempDir { dir =>
+ val path = new java.io.File(dir, "metrics").getCanonicalPath
+ spark.sql(
+ """SELECT TIMESTAMP '2024-01-01 00:00:00' AS ts_start,
+ | TIMESTAMP '2024-01-01 00:10:00' AS ts_end, 100.0D AS
value""".stripMargin)
+ .write.mode("overwrite").parquet(path)
+ val binBy =
+ s"""parquet.`$path` BIN BY (RANGE ts_start TO ts_end BIN WIDTH
INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))""".stripMargin
+
+ // `_metadata` is a hidden metadata column; BIN BY carries it through
for a bare ORDER BY.
+ checkAnswer(
+ spark.sql(s"SELECT bin_start FROM $binBy ORDER BY
_metadata.file_size"),
+ Seq(Row(tsAt("2024-01-01 00:00:00")), Row(tsAt("2024-01-01
00:05:00"))))
+
+ // Also selectable through BIN BY.
+ checkAnswer(
+ spark.sql(s"SELECT bin_start, _metadata.file_size > 0 FROM $binBy"),
+ Seq(Row(tsAt("2024-01-01 00:00:00"), true), Row(tsAt("2024-01-01
00:05:00"), true)))
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
