[
https://issues.apache.org/jira/browse/SPARK-59995?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-59995:
-------------------------------
Description:
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.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled=true}}
** {{spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled=true}} (the
default on master)
**
{{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.
was:
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.
> 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: 4.2.0
> Reporter: Peter Toth
> Priority: Major
>
> 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.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled=true}}
> ** {{spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled=true}} (the
> default on master)
> **
> {{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]
