Peter Toth created SPARK-59995:
----------------------------------
Summary: GroupPartitionsExec's k-way merge fails when the ordering
has a partition transform
Key: SPARK-59995
URL: https://issues.apache.org/jira/browse/SPARK-59995
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: Peter Toth
With {{spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled}} on,
{{GroupPartitionsExec}} can coalesce partitions with a k-way merge over its
child's ordering (SPARK-55715). The merge compares rows with a code-generated
ordering. When that ordering contains a partition transform, such as
{{years(arrive_time)}}, the task fails:
{code}
[INTERNAL_ERROR] Cannot generate code for expression:
transformexpression(org.apache.spark.sql.connector.catalog.functions.YearsFunction$@...,
input[2, timestamp, true], None) SQLSTATE: XX000
{code}
* {{TransformExpression.eval}} calls the bound {{ScalarFunction}}, but
{{TransformExpression.doGenCode}} always throws.
* The merge's ordering ({{LazyCodeGenOrdering}} in {{GroupPartitionsExec}})
calls {{GenerateOrdering}} directly. Unlike {{RowOrdering.create}}, it has no
interpreted fallback.
The ordering can contain a transform in two ways. Both are measured on master:
* The source reports it via {{SupportsReportOrdering}}. Example: a table
partitioned by {{identity(id)}} reports {{[id, name, years(arrive_time)]}}. A
sort-merge join on {{(id, name)}} fails with:
** {{spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled=true}}
** {{spark.sql.requireAllClusterKeysForCoPartition=false}}
* The scan derives it from its partition keys (SPARK-56241). Example: a table
partitioned by {{(identity(id), years(arrive_time))}} derives {{[id,
years(arrive_time)]}}. A sort-merge join on {{id}} fails with:
** {{spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled=true}}
** {{spark.sql.requireAllClusterKeysForCoPartition=false}}
**
{{spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled=true}}
** {{spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled=true}}
**
{{spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled=false}}
In both examples two partitions with {{id}} = 1 are coalesced. The merge's
ordering satisfies the join's required ordering, so {{EnsureRequirements}}
picks the merge instead of a sort. In the second example,
{{preserveKeyOrderingOnCoalesce}} on keeps the ordering on {{id}} without the
merge, so the query does not fail.
Found while working on SPARK-59981, which derives the ordering in more cases.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
