[
https://issues.apache.org/jira/browse/SPARK-59682?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18123728#comment-18123728
]
Pratham Manja commented on SPARK-59682:
---------------------------------------
{{{{I found that _spark.sql.optimizer.excludedRules_}} allows excluding
{{{}RemoveRedundantAliases{}}}, but doing so breaks typed {{Dataset}}
execution. {{EliminateSerialization}} wraps a collapsed
{{{}DeserializeToObject{}}}/{{{}SerializeFromObject{}}} pair in an
exprId-preserving {{Alias}} typed as {{{}ObjectType{}}};
{{RemoveRedundantAliases}} normally strips that alias before execution. With it
excluded, the {{ObjectType}} alias leaks into the physical plan, and neither
codegen nor the interpreted projection fallback supports projecting an
{{ObjectType}} field, causing a {{CompileException}} on any typed {{Dataset}}
chain that triggers {{EliminateSerialization}} (e.g.
{{{}groupByKey().mapGroups().mapPartitions(){}}}).}}
As a fix i added {{RemoveRedundantAliases}} to {{nonExcludableRules}} in the
optimizer, consistent with the other rules already in that list that exist to
prevent {{excludedRules}} from leaking unsupported plan nodes into execution.
corresponding PR https://github.com/apache/spark/pull/59256
> [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
> Priority: Minor
> Labels: pull-request-available
>
> 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]
