Rodrigo Coin Curvo created SPARK-59682:
------------------------------------------
Summary: [SQL] Excluding RemoveRedundantAliases causes typed
Dataset execution to fail with unsupported ObjectType projection
Key: SPARK-59682
URL: https://issues.apache.org/jira/browse/SPARK-59682
Project: Spark
Issue Type: Bug
Components: Optimizer
Affects Versions: 4.2.0
Reporter: Rodrigo Coin Curvo
I had to exclude RemoveRedundantAliases from the optimizer due to having issues
with execution plans. By excluding the rule everything worked so far. But then
I ran into this problem while trying to use VACUUM with Delta Lake.
I asked an LLM to investigate the root cause, and below is the report with a
minimal reproduction code.
h2. Summary
Spark 4.2.0 fails to execute a valid typed Dataset plan when
{{RemoveRedundantAliases}} is excluded through
{{{}spark.sql.optimizer.excludedRules{}}}.
The failure is raised by {{{}InterpretedUnsafeProjection{}}}:
{{org.apache.spark.SparkException: [INTERNAL_ERROR]
The data type 'object' is not supported in generating a writer function for a
struct field, array element, map key or map value.}}
The same plan succeeds with the default optimizer rules. The failure does not
require Delta Lake, ThriftServer, Hive, object storage, or a catalog.
h2. Version
Tested with the current Apache Spark release
[4.2.0|https://spark.apache.org/downloads]:
{{Apache Spark 4.2.0
Scala 2.13.18
Java 17.0.19}}
h2. Minimal reproduction
Save the following as {{repro.scala}} and run it with {{{}spark-shell{}}}:
{code:java}
import org.apache.spark.sql.Encoderscase class Status(path: String, length:
Long, isDir: Boolean, modificationTime: Long)
case class NameAndSize(path: String, length: Long)val input =
spark.createDataset(
spark.sparkContext.parallelize(
Seq(
Status("probe/file-a.parquet", 1L, false, 1L),
Status("probe/file-a.parquet", 2L, false, 2L),
Status("probe/file-b.parquet", 3L, false, 3L)),
numSlices = 2))(
Encoders.product[Status])val keyed = input.groupByKey((f: Status) =>
f.path)(Encoders.STRING)
val grouped = keyed.mapGroups(
(_, values) => values.maxBy(_.modificationTime)
)(Encoders.product[Status])val output = grouped.mapPartitions(
iterator => iterator.map(f => NameAndSize(f.path, f.length))
)(Encoders.product[NameAndSize])println(s"EXCLUDED_RULES=${spark.conf.getOption("spark.sql.optimizer.excludedRules")}")
output.explain(true)
println(s"COUNT=${output.count()}"){code}
Run the failing case in a worker-executing local cluster:
{code:java}
docker run --rm -i \
-v "$PWD/repro.scala:/tmp/repro.scala:ro" \
apache/spark:4.2.0-scala2.13-java17-python3-r-ubuntu \
bash -lc '/opt/spark/bin/spark-shell \
--master local-cluster[2,1,1024] \
--conf
spark.sql.optimizer.excludedRules=org.apache.spark.sql.catalyst.optimizer.RemoveRedundantAliases
\
--conf spark.driver.host=127.0.0.1 \
--conf spark.driver.bindAddress=127.0.0.1 < /tmp/repro.scala' {code}
For the control case, omit the {{spark.sql.optimizer.excludedRules}} option.
h2. Observed results
Control case, default optimizer rules:
{{EXCLUDED_RULES=None}}
{{COUNT=2}}
Failing case, with {{RemoveRedundantAliases}} excluded:
{{EXCLUDED_RULES=Some(org.apache.spark.sql.catalyst.optimizer.RemoveRedundantAliases)}}
{{org.apache.spark.SparkException: [INTERNAL_ERROR]
The data type 'object' is not supported in generating a writer function for a
struct field, array element, map key or map value.}}
The failing optimized/physical plan contains this additional projection:
{noformat}
SerializeFromObject (... NameAndSize fields ...)
+- MapPartitions, obj: NameAndSize
+- Project [obj AS obj]
+- MapGroups, obj: Status
+- Sort
+- Exchange{noformat}
The control plan removes {{Project [obj AS obj]}} and succeeds.
h2. Suspected cause
{{RemoveRedundantAliases}} removes the redundant object projection when enabled.
When excluded, the projection remains executable around a typed Dataset
{{{}ObjectType{}}}. {{ProjectExec}} then creates an {{{}UnsafeProjection{}}},
which reaches
the unsupported-type fallback in
{{{}InterpretedUnsafeProjection.generateFieldWriter{}}}.
The relevant Spark source files are:
*
[{{Optimizer.scala}}|https://github.com/apache/spark/blob/v4.2.0/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala]
*
[{{objects.scala}}|https://github.com/apache/spark/blob/v4.2.0/sql/core/src/main/scala/org/apache/spark/sql/execution/objects.scala]
*
[{{InterpretedUnsafeProjection.scala}}|https://github.com/apache/spark/blob/v4.2.0/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/InterpretedUnsafeProjection.scala]
h2. Expected behavior
Excluding an optimizer rule should not cause a valid typed Dataset plan to
fail with an internal unsupported-type exception. The plan should either
execute successfully or Spark should reject the unsupported configuration
explicitly during analysis/optimization.
h2. Workaround
Do not exclude:
{{org.apache.spark.sql.catalyst.optimizer.RemoveRedundantAliases}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
