[Spark] Allow missing fields with implicit casting during streaming write

[johanl-db]

Description

Follow-up on #3443 that introduced implicit casting during streaming write to delta tables.

The feature was shipped disabled due to a regression found in testing where writing data with missing struct fields start being rejected. Streaming writes are one of the few inserts that allows missing struct fields.

This change allows configuring the casting behavior used in MERGE, UPDATE and streaming writes wrt to missing struct fields.

How was this patch tested?

Extensive tests were added in #3762 in preparation for this changes, covering for all inserts (SQL, dataframe, append/overwrite, ..):

• Missing top-level columns and nested struct fields.
• Extra top-level columns and nested struct fields with schema evolution.
• Position vs. name based resolution for top-level columns and nested struct fields.
  with e.p. the goal of ensuring that enabling implicit casting in stream writes here doesn't cause any other unwanted behavior change.

This PR introduces the following user-facing changes

From the initial PR: #3443

Previously, writing to a Delta sink using a type that doesn't match the column type in the Delta table failed with DELTA_FAILED_TO_MERGE_FIELDS:

spark.readStream
    .table("delta_source")
    # Column 'a' has type INT in 'delta_sink'.
    .select(col("a").cast("long").alias("a"))
    .writeStream
    .format("delta")
    .option("checkpointLocation", "<location>")
    .toTable("delta_sink")

DeltaAnalysisException: [DELTA_FAILED_TO_MERGE_FIELDS] Failed to merge fields 'a' and 'a'

With this change, writing to the sink now succeeds and data is cast from LONG to INT. If any value overflows, the stream fails with (assuming default storeAssignmentPolicy=ANSI):

SparkArithmeticException: [CAST_OVERFLOW_IN_TABLE_INSERT] Fail to assign a value of 'LONG' type to the 'INT' type column or variable 'a' due to an overflow. Use `try_cast` on the input value to tolerate overflow and return NULL instead."

[tomvanbussel]

LGTM, but left a few nits.

The casting and missing columns behavior is getting really complex overall, as it depends so strongly on the operation performed. It would be good if we could clean this up, but this will likely require some breaking changes.

[tomvanbussel]

Nit: This is mixing policy flags (isMergeOrUpdate) with mechanism flags (allowMissingStructField, resolveStructsByName).

[johanl-db]

I pulled isMergeOrUpdate out, this is now a trait that gets mixed in instead of a flag

[tomvanbussel]

It seems like this config is completely ignored when resolveStructsByName is false. This may lead to some surprising behavior in the future.

[johanl-db]

I'm now using the type system to enforce a constraint that by position cannot specify allowMissingStructField.

[tomvanbussel]

Nit: This is getting a little bit complex. It's hard for me to understand for me which cases are actually covered here. It's okay to have some duplication in tests.

[johanl-db]

Minor improvement: I introduced a variable for "inserts that don't support implicit casting" to make it a bit easier to reason about.

[tomvanbussel]

Why not just:

Suggested change

	includeInserts = inserts -- (insertsByName.intersect(insertsDataframe) - StreamingInsert)
	includeInserts = inserts -- insertsByName.intersect(insertsDataframe) + StreamingInsert

[johanl-db]

inserts is one of:

• insertsAppend - StreamingInsert
• insertsOverwrite - SQLInsertOverwritePartitionByPosition
• Set(StreamingInsert)
• Set(SQLInsertOverwritePartitionByPosition)

It doesn't always contain StreamingInsert, we don't want to add it then otherwise we'll get duplicate tests