[SPARK-35912][SQL] Fix cast struct contains null value to string/struct

[cfmcgrady]

What changes were proposed in this pull request?

This PR fixes an issue that cast the struct which contains null value to other type has a difference result when we enable/disable codegen.

Here is an example:

valschema=StructType(Seq(StructField("value",
StructType(Seq(
StructField("x", IntegerType, nullable =false),
StructField("y", IntegerType, nullable =false)
))
)))
schema.printTreeString()
// root// |-- value: struct (nullable = true)// | |-- x: integer (nullable = false)// | |-- y: integer (nullable = false)valtestDS=Seq("""{"value":{"x":1}}""").toDS
valjsonDS= spark.read.schema(schema).json(testDS)
jsonDS.show()
// incorrect result// +---------+// | value|// +---------+// |{1, null}|// +---------+
spark.sql("set spark.sql.codegen.wholeStage=false")
jsonDS.show()
// the result we expected due to y.nullable = false.// +------+// | value|// +------+// |{1, 0}|// +------+

Actually, the result should be depending on the field nullable setting. this bug also happens when we cast struct to struct.

Does this PR introduce any user-facing change?

No, only bug fix.

How was this patch tested?

New test.

[AmplabJenkins]

Can one of the admins verify this patch?

[cfmcgrady]

When the actual value is null, for primitive type field, row.isNullAt(i) return ture, but row.getXXX return a default value.

For exmaple:

valr=new org.apache.spark.sql.catalyst.expressions.GenericInternalRow(Array(1, null))
println(r.getInt(0)) // 1
println(r.getInt(1)) // 0
println(r.isNullAt(1)) // true

so we cann't only check row.isNullAt(i) here, we need to do the same logical like BoundReference.doGenCode(), add nullable check.

[HyukjinKwon]

how is the cache issue related to the cast?

[cfmcgrady]

how is the cache issue related to the cast?

HI, @HyukjinKwon
The root cause is cast struct to other types, updated the pr description.

[cfmcgrady]

cc @cloud-fan@viirya@maropu

[cloud-fan]

if fields(0).nullable is false, how can row.isNullAt(0) be true?

[maropu]

(I have the same question)

[cfmcgrady]

If user create dataframe from spark.internalCreateDataFrame(), the row.isNullAt() may be true even though the schema nullable is false.
For instance:

valschema=StructType(Seq(
StructField("x",
StructType(Seq(
StructField("y", IntegerType, true),
StructField("z", IntegerType, false)
)))))
valrdd= spark.sparkContext.parallelize(Seq(InternalRow(InternalRow(1, null))))
valdf= spark.internalCreateDataFrame(rdd, schema)
df.show
// current master branch output// +---------+// | x|// +---------+// |{1, null}|// +---------+

Although the spark.internalCreateDataFrame() is sql package private API, but spark.read.json() and spark.read.csv() call it without null value handled.(the example show in pr description)

[cloud-fan]

Then we need to fix the nullability. There are so many places in the Spark codebase that relies on nullability to do optimizations. It's not possible to change all of them to not trust the nullability anymore.

Can we fix spark.read.json() to set the nullability correctly?

[cfmcgrady]

Ok, let me try.

[HyukjinKwon]

Hey mind explaining why cast path issue is related to being cached?

[maropu]

nit:

 Seq(true, false).foreach { nullable =>

[maropu]

nit: Seq(true, false).foreach { nullable =>

[maropu]

nit: please remove this blank.

[maropu]

(I have the same question)

[cfmcgrady]

Hey mind explaining why cast path issue is related to being cached?

Actually, the cached result is what we want. The issue is that spark.read.json is not an expected result when the input schema has a nullable setting to false. From the log, we found it's a cast operation.

=== Result of Batch Preparations ===
CollectLimit 21 CollectLimit 21
!+- Project [cast(value#5 as string) AS value#18] +- *(1) Project [cast(value#5 as string) AS value#18]
! +- Scan ExistingRDD[value#5] +- *(1) Scan ExistingRDD[value#5]

[HyukjinKwon]

Shouldn't it fail instead of setting it as 0 or null? I feel like the handling should be done somewhere in JacksonParser.

[cfmcgrady]

Shouldn't it fail instead of setting it as 0 or null? I feel like the handling should be done somewhere in JacksonParser.

Thanks for your suggestion, I'll try.

[cfmcgrady]

Create a new PR

BTW, Shall we merge this PR? The cast issue may occur when the user create dataframe from API spark.internalCreateDataFrame(). cc @cloud-fan@maropu

[cloud-fan]

spark.internalCreateDataFrame() is a private API and users should be responsible if they do something wrong. We can't merge this PR, because we can't sacrifice the performance in the happy path for invalid usage of private APIs from users.