[SPARK-16472][SQL] Force user specified schema to the nullable one

[HyukjinKwon]

What changes were proposed in this pull request?

This PR proposes to force the user-specified schema to the nullable one, as Spark SQL can't validate it.

How was this patch tested?

Unit tests adde in FileStreamSourceSuite.scala and DataFrameReaderWriterSuite.scala.

[HyukjinKwon]

Hi @gatorsmile and @marmbrus, I saw the discussion and found you are related with this one. Could you please review this?

[HyukjinKwon]

This one is forcing the schema as nullable but it has no tests. So the tests were added.

[SparkQA]

Test build #62049 has finished for PR 14124 at commit a917678.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #62053 has finished for PR 14124 at commit adae8de.

• This patch fails PySpark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #62054 has finished for PR 14124 at commit 3980681.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[gatorsmile]

valrdd= spark.sparkContext.makeRDD(Seq("{\"a\" : 1}", "{\"a\" : null}"))
valschema=StructType(StructField("a", IntegerType, nullable =false) ::Nil)
valdf= spark.read.schema(schema).json(rdd)
df.show()

When user-specified schemas are not nullable and the data contains null, the null value in the result becomes 0. This looks like a bug, right?

[HyukjinKwon]

Ah, yes it seems a bug to me. I thought it throws an exception in that case or works fine after thid PR. I will test this before/after this PR. Thanks!

[HyukjinKwon]

Oh, I see, before this patch

+---+
| a|
+---+
| 1|
| 0|
+---+

after this patch

+----+
| a|
+----+
| 1|
|null|
+----+

FYI, currently (before this patch) the code with StringType below

valrdd= spark.sparkContext.makeRDD(Seq("{\"a\" : 1}", "{\"a\" : null}"))
valschema=StructType(StructField("a", StringType, nullable =false) ::Nil)
valdf= spark.read.schema(schema).json(rdd)
df.show()

is being failed with the exception below:

Error while decoding: java.lang.NullPointerException
createexternalrow(input[0, string, false].toString, StructField(a,StringType,false))
+- input[0, string, false].toString
+- input[0, string, false]
java.lang.RuntimeException: Error while decoding: java.lang.NullPointerException
createexternalrow(input[0, string, false].toString, StructField(a,StringType,false))
+- input[0, string, false].toString
+- input[0, string, false]
at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.fromRow(ExpressionEncoder.scala:292)
at org.apache.spark.sql.Dataset$$anonfun$org$apache$spark$sql$Dataset$$execute$1$1$$anonfun$apply$15.apply(Dataset.scala:2218)
at org.apache.spark.sql.Dataset$$anonfun$org$apache$spark$sql$Dataset$$execute$1$1$$anonfun$apply$15.apply(Dataset.scala:2218)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
...

It seems an unexpected behaviour anyway (I think at least we should fix the error message). I will submit a patch if this one is decided not worth being added. Thanks @gatorsmile again!

[gatorsmile]

@HyukjinKwon No matter whether this PR is merged or not, I still think we should fix the above issue. Silent conversion does not look good to me.

[HyukjinKwon]

@gatorsmile I am a bit confused if we are allowed to read JSON (via json(jsonRDD: RDD[String]) API) with schema having fields set false in nullable.
If it is meant to be not allowed, this PR will prevents the case above.

But, yea, I think I agree that it is a potential problem anyway (even if the case above is not allowed.)

[viirya]

@HyukjinKwon Your patch solves this inconsistency by forcing schema as nullable at all. However, looks like the parquet case is for compatibility, is this the same for json? If no, why we want to do this?

[HyukjinKwon]

@viirya Thanks for your comment! Actually, that's what I want to have some feedback about from @marmbrus .

It seems forcing to a nullable schema all is already happening, (see here) for all data sources implementing FileFormat when you read data via read.format(...).save(...) API (but not for structured streaming and another API for json, which this PR deals with).

So, actually, the purpose of this PR is, to make all read APIs consistent. The reason to make them consistent in a way that the schema is forced as nullable, is what he said in the mailing list.

Sure, but a traditional RDBMS has the opportunity to do validation before
loading data in. Thats not really an option when you are reading random
files from S3. This is why Hive and many other systems in this space treat
all columns as nullable.

Actually, apparently, Parquet also reads and writes the schema with nullability correctly if we get rid of asNullable in here (I tested this by a roundtrip in writing and reading before) but it seems that's prevented as a safeguard due to (I assume) the reason above.

@marmbrus Do you mind if I ask to clarify here please?

I think we might have to deal with this as a datasource-specific problem.

[HyukjinKwon]

Could you take a look please @marmbrus ?

[HyukjinKwon]

gentle ping @marmbrus

[marmbrus]

@cloud-fan

[cloud-fan]

What will happen if the given schema is wrong? It seems weird that we allow users to provide schema while reading the data, but without validating it.

[HyukjinKwon]

Thanks for feedback @cloud-fan !

If the user-given schema is wrong, it is handled differently for each datasource.

• For JSON and CSV
  it is kind of permissive generally (for example, compatibility among numeric types).
• For ORC and Parquet
  Generally it is strict to types. So they don't allow the compatibility (except for very few cases, e.g. for parquet, [SPARK-16632][sql] Respect Hive schema when merging parquet schema. #14272 and [SPARK-16632][SQL] Use Spark requested schema to guide vectorized Parquet reader initialization #14278)
• For Text
  it only supports StringType.
• For JDBC
  it does not take user-given schema since it does not implement SchemaRelationProvider.

Should we disallow specifying schemas for these (maybe ORC and Parquet)?

[HyukjinKwon]

BTW, actually, this PR is not only about user-given schema.

Currently, it always reads data into dataframe by datasources based on FileFormat ignoring nullability in schema (for both user-given schema and inferred/read schema).

However, this does not happen when reading for streaming by the datasources (FileFormat) (and another JSON api).

So, this PR tries to make them all ignore the nullability in schema to be consistent.

[HyukjinKwon]

@cloud-fan If nullability should be not ignored, then I can fix this PR to make them consistent to not ignoring it (and of course I will try to identify the related problems). In this case, I will work on what @gatorsmile pointed out in #14124 (comment) about JSON (and will check the other data sources as well).

I will follow your decision.

To cut all the comments above short, (for other reviewers),

• The purpose of this PR is whether it should force all schema to nullable schema or not.
• Forcing this as nullable one is already happening with normal reading and writing for data sources based on FileFormat but not for structured streaming and json(rdd: RDD[String]) API.
• This is for both inferred/read schema and user-given schema.

[SparkQA]

Test build #64066 has finished for PR 14124 at commit 079aae2.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #64631 has finished for PR 14124 at commit ffacb55.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #65360 has finished for PR 14124 at commit f6be52b.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #65747 has finished for PR 14124 at commit 0bc06c6.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #67029 has finished for PR 14124 at commit 3f153a3.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

Sorry for the delay. After thinking it again, I think it doesn't make sense to allow users to specify the nullability when reading a data source, as Spark SQL can't validate it. How about we turn schema to nullable in DataFrameReader.schema?

[HyukjinKwon]

Thanks @cloud-fan, sure, that sounds great.

[HyukjinKwon]

Oh wait, @cloud-fan, it seems, at least, Parquet files could possibly be written with not nullable fields. So, reading it without user-specified schema might also cause the inconsistency between the schema read from structured streaming and the one read from filed sources.

If you are not sure of this, I am fine with turning the schema into nullable in DataFrameReader.schema for now. Let me rebase this one first.

[HyukjinKwon]

Actually, nvm. I think handling this in DataFrameReader.schema will deal with most of general cases.

[SparkQA]

Test build #68474 has finished for PR 14124 at commit 7306937.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #68477 has finished for PR 14124 at commit d240c0d.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #70922 has finished for PR 14124 at commit 1abaf1b.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

cc @liancheng , what do you think about the nullability change?

[kiszk]

#17293 added data validation using schema information for Parquet Reader, as @gatorsmile suggested in https://www.mail-archive.com/user@spark.apache.org/msg39233.html.

[HyukjinKwon]

Let me rather close this for a while. I will reopen if it looks worth. I think the same subject could be discussed in @kiszk 's PR.