[Spark] Disable implicit casting in Delta streaming sink

[johanl-db]

Description

#3443 introduced implicit casting when writing to a Delta table using a streaming query.

We are disabling this change for now as it regresses behavior when a struct field is missing in the input data. This previously succeeded, filling the missing fields with null but would now fail with:

DELTA_UPDATE_SCHEMA_MISMATCH_EXPRESSION] Cannot cast struct<name:string> to struct<name:string,age:bigint>. All nested columns must match.

Note: batch INSERT fails in this scenario with:

[DELTA_INSERT_COLUMN_ARITY_MISMATCH] Cannot write to '<table>, not enough nested fields in <struct>; target table has 3 column(s) but the inserted data has 2 column(s)

but since streaming write allowed this, we have to preserve that behavior.

How was this patch tested?

Tests added as part of #3443, e.p. with flag disabled.

Does this PR introduce any user-facing changes?

Disabled behavior change that was to be introduced with #3443.

[Kimahriman]

FWIW we noticed something similar recently with inserts on merge, the implicit cast failed if struct fields were missing

[johanl-db]

FWIW we noticed something similar recently with inserts on merge, the implicit cast failed if struct fields were missing

I believe that's expected in that case, unless you enable schema evolution, MERGE / UPDATE rejects writes with missing struct fields. Batch (non-streaming) INSERT also rejects that - although I've recently looked closer at batch insert behavior and the truth is that it's a bit all over the place..

Rejecting missing fields is a reasonable behavior when schema evolution is disabled, as it means we instead fall back to schema enforcement. Streaming writes didn't behave that way though, which I missed

[Kimahriman]

Rejecting missing fields is a reasonable behavior when schema evolution is disabled, as it means we instead fall back to schema enforcement. Streaming writes didn't behave that way though, which I missed

This is missing fields in the source, not the target, so not schema evolution related.

Easy example:

import pyspark.sql.functions as F
from delta import DeltaTable
# Create table of nested struct<id: long, value: long>
spark.range(10).select(F.struct('id', (F.col('id') * 10).alias('value')).alias('nested')).write.format('delta').save('/tmp/merge-test')
# Works without "nested.value"
spark.range(10, 20).select(F.struct('id').alias('nested')).write.format('delta').mode('append').save('/tmp/merge-test')

table = DeltaTable.forPath(spark, "/tmp/merge-test")
# Fails with "Cannot cast struct<id:bigint> to struct<id:bigint,value:bigint>. All nested columns must match
table.alias('target').merge(spark.range(20, 30).select(F.struct('id').alias('nested')).alias('source', 'target.nested.id = source.nested.id').whenNotMatchedInsertAll().execute()

[johanl-db]

Rejecting missing fields is a reasonable behavior when schema evolution is disabled, as it means we instead fall back to schema enforcement. Streaming writes didn't behave that way though, which I missed

This is missing fields in the source, not the target, so not schema evolution related.

Easy example:

import pyspark.sql.functions as F
from delta import DeltaTable
# Create table of nested struct<id: long, value: long>
spark.range(10).select(F.struct('id', (F.col('id') * 10).alias('value')).alias('nested')).write.format('delta').save('/tmp/merge-test')
# Works without "nested.value"
spark.range(10, 20).select(F.struct('id').alias('nested')).write.format('delta').mode('append').save('/tmp/merge-test')

table = DeltaTable.forPath(spark, "/tmp/merge-test")
# Fails with "Cannot cast struct<id:bigint> to struct<id:bigint,value:bigint>. All nested columns must match
table.alias('target').merge(spark.range(20, 30).select(F.struct('id').alias('nested')).alias('source', 'target.nested.id = source.nested.id').whenNotMatchedInsertAll().execute()

Right, save()/saveAsTable() will accept missing fields, but if you try SQL INSERT INTO t (nested) VALUES (..), then I expect that will fail. I think insert by position - df.insertInto(), INSERT INTO t VALUES (..) - will also fail.

We do handle SQL insert by name and SQL/DF insert by position in DeltaAnalysis and apply schema enforcement, but let save()/saveAsTable() get through unhandled and pretty much don't do any schema validation beyond basic compatibility when writing the data.

That's what I meant by insert behavior is a bit all over the place, I looked at it a couple of weeks ago to see if it could be fixed - and allow implicit casting in save()/saveAsTable() - but this would bring a lot of breaking behavior changes

[Kimahriman]

That's what I meant by insert behavior is a bit all over the place, I looked at it a couple of weeks ago to see if it could be fixed - and allow implicit casting in save()/saveAsTable() - but this would bring a lot of breaking behavior changes

Would it make more sense to do a Dataset.to type operation instead of a cast? That seems more like what is trying to be achieved.

[johanl-db]

Would it make more sense to do a Dataset.to type operation instead of a cast? That seems more like what is trying to be achieved.

That seems too limited, Dataset.to also cast types and eventually calls to Cast.canANSIStoreAssign() which for example doesn't allow many casts that we do support today - we allow most or all valid implicit casts, string -> int would be valid - and doesn't respect the value of spark.sql.storeAssignmentPolicy that write operations should respect.

The column reordering part does make sense though, although the issue isn't so much how to apply it as much as making sure we're not breaking existing workloads if we start doing it in more cases