[
https://issues.apache.org/jira/browse/SPARK-44517?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18096568#comment-18096568
]
Josh Rosen commented on SPARK-44517:
------------------------------------
I hit this same issue and confirmed it still reproduces on current master
(5.0.0-SNAPSHOT, commit 84bb88952f3).
I'd like to share two updated reproductions and a suggestion on the fix
approach.
*The failure now occurs inside stock Spark with default configs.* When this
ticket was filed, the nullability change was only surfaced by external writers
that enforce nullability (Iceberg, Delta). The root defect is unchanged – the
{{ReplaceDeduplicateWithAggregate}} rule rewrites every non-key column as
{{{}first(col){}}}, whose nullability is hardcoded to {{{}true{}}}, so the
optimizer changes the plan's output schema – but Spark itself now detects it
when a v2 write sits on top of the {{{}Deduplicate{}}}. {{AppendData.resolved}}
includes {{{}outputResolved{}}}, which requires each query column's nullability
to be no wider than the corresponding table column's.
With a {{NOT NULL}} table column that predicate holds at analysis time (the
input column is non-nullable) but flips to false when the rule widens the
column mid-optimization, so the previously-resolved {{AppendData}} becomes
unresolved.
{{{}spark.sql.lightweightPlanChangeValidation{}}}, enabled by default since
Spark 4.0, rejects exactly this (a rule turning a resolved plan unresolved) and
fails the query during optimization.
*Minimal demonstration of the schema change:*
{code:java}
val df = spark.range(3).withColumn("v", lit(1)).dropDuplicates("id")
df.queryExecution.analyzed.schema("v").nullable // false
df.queryExecution.optimizedPlan.schema("v").nullable // true -- optimizer
changed the plan schema
{code}
*Self-contained end-to-end failure, pure PySpark (no extra jars, default
configs):*
{code:python}
from pyspark.sql import SparkSession
from pyspark.sql.datasource import DataSource
from pyspark.sql.functions import lit
spark = SparkSession.builder.getOrCreate()
class NullSink(DataSource):
def schema(self):
return "id bigint, v int not null"
spark.dataSource.register(NullSink)
spark.sql("CREATE TABLE tbl USING NullSink")
df = spark.range(3).withColumn("v", lit(1)).dropDuplicates(["id"])
df.write.insertInto("tbl")
{code}
(The data source never needs a writer: the failure happens during optimization,
before execution. The Python data source is just a small stock way to get a
DSv2 table with a declared NOT NULL column).
which fails with:
{noformat}
org.apache.spark.SparkException: [PLAN_VALIDATION_FAILED_RULE_IN_BATCH] Rule
org.apache.spark.sql.catalyst.optimizer.ReplaceDeduplicateWithAggregate in batch
Replace Operators generated an invalid plan: The plan was previously resolved
and
now became unresolved. SQLSTATE: XXKD0
{noformat}
The same failure occurs with any DSv2 catalog whose tables declare {{NOT NULL}}
columns (e.g. {{InMemoryTableCatalog}} in Spark's own tests).
*On the fix approach:* the stale PR for this ticket
([https://github.com/apache/spark/pull/42117]) changed {{First.nullable}}
globally to {{{}if (ignoreNulls) false else child.nullable{}}}. I don't think
that is sound in general:
* in a global (non-grouped) aggregate over zero input rows, {{first(col)}}
returns null regardless of the child's nullability;
* with {{{}ignoreNulls = true{}}}, {{first(col, true)}} returns null for a
group whose values are all null, so {{false}} is wrong there too.
However, the specific usage in {{ReplaceDeduplicateWithAggregate}} should be
safe to fix: the rewrite always produces a grouping aggregate with a non-empty
grouping key list (it substitutes {{Literal(1)}} when the key list is empty),
so every group contains at least one row and {{first(col, false)}} on a
non-nullable input can never return null. So I'd propose scoping the fix to the
rule itself, e.g. wrapping the
{{{}ReplaceDeduplicateWithAggregate{}}}-generated aggregate expression in
{{KnownNotNull}} when the input attribute is non-nullable, restoring the
original nullability without touching {{{}First{}}}'s general semantics.
> first operator should respect the nullability of child expression as well as
> ignoreNulls option
> -----------------------------------------------------------------------------------------------
>
> Key: SPARK-44517
> URL: https://issues.apache.org/jira/browse/SPARK-44517
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 3.2.0, 3.2.1, 3.3.0, 3.2.2, 3.3.1, 3.2.3, 3.2.4, 3.3.2,
> 3.4.0, 3.4.1
> Reporter: Nan Zhu
> Priority: Major
> Labels: pull-request-available
>
> I found the following problem when using Spark recently:
>
> {code:java}
> // code placeholder
> import spark.implicits._
> val s = Seq((1.2, "s", 2.2)).toDF("v1", "v2", "v3")
> val schema = StructType(Seq(StructField("v1", DoubleType, nullable =
> false),StructField("v2", StringType, nullable = true),StructField("v3",
> DoubleType, nullable = false)))
> val df = spark.createDataFrame(s.rdd, schema)val inputDF =
> val inputDF = df.dropDuplicates("v3")
> spark.sql("CREATE TABLE local.db.table (\n v1 DOUBLE NOT NULL,\n v2 STRING,
> v3 DOUBLE NOT NULL)")
> inputDF.write.mode("overwrite").format("iceberg").save("local.db.table")
> {code}
>
>
> when I use the above code to write to iceberg (i guess Delta Lake will have
> the same problem) , I got very confusing exception
> {code:java}
> Exception in thread "main" java.lang.IllegalArgumentException: Cannot write
> incompatible dataset to table with schema:
> table
> { 1: v1: required double 2: v2: optional string 3: v3: required double}
> Provided schema:
> table { 1: v1: optional double 2: v2: optional string 3: v3: required
> double} {code}
> basically it complains that we have v1 as the nullable column in our
> `inputDF` above which is not allowed since we created table with the v1 as
> not nullable. The confusion comes from that, if we check the schema with
> printSchema() of inputDF, v1 is not nullable
> {noformat}
> root
> |-- v1: double (nullable = false)
> |-- v2: string (nullable = true)
> |-- v3: double (nullable = false){noformat}
> Clearly, something changed the v1's nullability unexpectedly!
>
> After some debugging I found that the key is that dropDuplicates("v3"). In
> optimization phase, we have ReplaceDeduplicateWithAggregate to replace the
> Deduplicate with aggregate on v3 and run first() over all other columns.
> However, first() operator has hard coded nullable as always "true" which is
> the source of changed nullability of v1
>
> this is a very confusing behavior of Spark, and probably no one really
> noticed as we do not care too much without the new table formats like delta
> lake and iceberg which can make nullability check correctly. Nowadays, we
> users adopt them more and more, this is surfaced up
>
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
