[
https://issues.apache.org/jira/browse/SPARK-60027?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-60027:
-------------------------------
Description:
{{TransformExpression.resolvedFunction}} builds the bound function's call over
the transform's raw children ({{V2ExpressionUtils.resolveScalarFunction}}).
Nothing casts them to the function's {{inputTypes()}}, although the
{{BoundFunction}} Javadoc says Spark will cast input values to the required
data types. Only the write path coerces them ({{TypeCoercionExecutor}} in
{{DistributionAndOrderingUtils}}).
The keyed shuffle evaluates this call to place the other side's rows (the
{{KeyedPartitioning}} case of {{ShuffleExchangeExec}}). So a function whose
input types differ from the column's type fails there.
Example with the test catalog's {{bucket}}, whose {{inputTypes()}} is {{(int,
bigint)}}. Measured on master:
* table {{t}} with an {{int}} column {{id}}, partitioned by {{bucket(4, id)}};
* table {{u}} with an {{int}} column {{id}}, not partitioned;
* {{spark.sql.sources.v2.bucketing.shuffle.enabled=true}};
* {{SELECT /*+ MERGE(t, u) */ t.id, u.data FROM t JOIN u ON t.id = u.id}}.
The task fails:
{code}
java.lang.ClassCastException: class java.lang.Integer cannot be cast to class
java.lang.Long (java.lang.Integer and java.lang.Long are in module java.base of
loader 'bootstrap')
at
org.apache.spark.sql.execution.exchange.ShuffleExchangeExec$.$anonfun$prepareShuffleDependency$14(ShuffleExchangeExec.scala:487)
{code}
The identity reduce (SPARK-56182) evaluates the same call to map the identity
side's keys ({{IdentityReducer}}), and fails the same way. Measured on master:
* table {{t}} as above;
* table {{u}} with an {{int}} column {{id}}, partitioned by {{identity(id)}};
* {{spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled=true}};
* the same query.
was:
{{TransformExpression.resolvedFunction}} builds the bound function's call over
the transform's raw children ({{V2ExpressionUtils.resolveScalarFunction}}).
Nothing casts them to the function's {{inputTypes()}}, although the
{{BoundFunction}} Javadoc says Spark will cast input values to the required
data types. Only the write path coerces them ({{TypeCoercionExecutor}} in
{{DistributionAndOrderingUtils}}).
The keyed shuffle evaluates this call to place the other side's rows (the
{{KeyedPartitioning}} case of {{ShuffleExchangeExec}}). So a function whose
input types differ from the column's type fails there.
Example with the test catalog's {{bucket}}, whose {{inputTypes()}} is {{(int,
bigint)}}. Measured on master:
* table {{t}} with an {{int}} column {{id}}, partitioned by {{bucket(4, id)}};
* table {{u}} with an {{int}} column {{id}}, not partitioned;
* {{spark.sql.sources.v2.bucketing.shuffle.enabled=true}};
* {{SELECT /*+ MERGE(t, u) */ t.id, u.data FROM t JOIN u ON t.id = u.id}}.
The task fails:
{code}
java.lang.ClassCastException: class java.lang.Integer cannot be cast to class
java.lang.Long (java.lang.Integer and java.lang.Long are in module java.base of
loader 'bootstrap')
at
org.apache.spark.sql.execution.exchange.ShuffleExchangeExec$.$anonfun$prepareShuffleDependency$14(ShuffleExchangeExec.scala:487)
{code}
> Partition transforms skip the input casts their bound function's inputTypes()
> asks for
> --------------------------------------------------------------------------------------
>
> Key: SPARK-60027
> URL: https://issues.apache.org/jira/browse/SPARK-60027
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.2.0, 4.1.3, 4.0.4
> Reporter: Peter Toth
> Priority: Major
>
> {{TransformExpression.resolvedFunction}} builds the bound function's call
> over the transform's raw children
> ({{V2ExpressionUtils.resolveScalarFunction}}). Nothing casts them to the
> function's {{inputTypes()}}, although the {{BoundFunction}} Javadoc says
> Spark will cast input values to the required data types. Only the write path
> coerces them ({{TypeCoercionExecutor}} in {{DistributionAndOrderingUtils}}).
> The keyed shuffle evaluates this call to place the other side's rows (the
> {{KeyedPartitioning}} case of {{ShuffleExchangeExec}}). So a function whose
> input types differ from the column's type fails there.
> Example with the test catalog's {{bucket}}, whose {{inputTypes()}} is {{(int,
> bigint)}}. Measured on master:
> * table {{t}} with an {{int}} column {{id}}, partitioned by {{bucket(4, id)}};
> * table {{u}} with an {{int}} column {{id}}, not partitioned;
> * {{spark.sql.sources.v2.bucketing.shuffle.enabled=true}};
> * {{SELECT /*+ MERGE(t, u) */ t.id, u.data FROM t JOIN u ON t.id = u.id}}.
> The task fails:
> {code}
> java.lang.ClassCastException: class java.lang.Integer cannot be cast to class
> java.lang.Long (java.lang.Integer and java.lang.Long are in module java.base
> of loader 'bootstrap')
> at
> org.apache.spark.sql.execution.exchange.ShuffleExchangeExec$.$anonfun$prepareShuffleDependency$14(ShuffleExchangeExec.scala:487)
> {code}
> The identity reduce (SPARK-56182) evaluates the same call to map the identity
> side's keys ({{IdentityReducer}}), and fails the same way. Measured on master:
> * table {{t}} as above;
> * table {{u}} with an {{int}} column {{id}}, partitioned by {{identity(id)}};
> * {{spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled=true}};
> * the same query.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
