jonvex commented on code in PR #7974:
URL: https://github.com/apache/hudi/pull/7974#discussion_r1108941277
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkUtils.scala:
##########
@@ -132,81 +136,80 @@ class TestHoodieSparkUtils {
.config("spark.kryo.registrator",
"org.apache.spark.HoodieSparkKryoRegistrar")
.config("spark.sql.extensions",
"org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
.getOrCreate
-
- val innerStruct1 = new
StructType().add("innerKey","string",false).add("innerValue", "long", true)
- val structType1 = new StructType().add("key", "string", false)
-
.add("nonNullableInnerStruct",innerStruct1,false).add("nullableInnerStruct",innerStruct1,true)
- val schema1 =
AvroConversionUtils.convertStructTypeToAvroSchema(structType1,
"test_struct_name", "test_namespace")
- val records1 = Seq(Row("key1", Row("innerKey1_1", 1L), Row("innerKey1_2",
2L)))
-
- val df1 = spark.createDataFrame(spark.sparkContext.parallelize(records1),
structType1)
- val genRecRDD1 = HoodieSparkUtils.createRdd(df1, "test_struct_name",
"test_namespace", true,
- org.apache.hudi.common.util.Option.of(schema1))
- assert(schema1.equals(genRecRDD1.collect()(0).getSchema))
-
- // create schema2 which has one addition column at the root level compared
to schema1
- val structType2 = new StructType().add("key", "string", false)
-
.add("nonNullableInnerStruct",innerStruct1,false).add("nullableInnerStruct",innerStruct1,true)
- .add("nullableInnerStruct2",innerStruct1,true)
- val schema2 =
AvroConversionUtils.convertStructTypeToAvroSchema(structType2,
"test_struct_name", "test_namespace")
- val records2 = Seq(Row("key2", Row("innerKey2_1", 2L), Row("innerKey2_2",
2L), Row("innerKey2_3", 2L)))
- val df2 = spark.createDataFrame(spark.sparkContext.parallelize(records2),
structType2)
- val genRecRDD2 = HoodieSparkUtils.createRdd(df2, "test_struct_name",
"test_namespace", true,
- org.apache.hudi.common.util.Option.of(schema2))
- assert(schema2.equals(genRecRDD2.collect()(0).getSchema))
-
- // send records1 with schema2. should succeed since the new column is
nullable.
- val genRecRDD3 = HoodieSparkUtils.createRdd(df1, "test_struct_name",
"test_namespace", true,
- org.apache.hudi.common.util.Option.of(schema2))
- assert(genRecRDD3.collect()(0).getSchema.equals(schema2))
- genRecRDD3.foreach(entry => assertNull(entry.get("nullableInnerStruct2")))
-
- val innerStruct3 = new
StructType().add("innerKey","string",false).add("innerValue", "long", true)
- .add("new_nested_col","string",true)
-
- // create a schema which has one additional nested column compared to
schema1, which is nullable
- val structType4 = new StructType().add("key", "string", false)
-
.add("nonNullableInnerStruct",innerStruct1,false).add("nullableInnerStruct",innerStruct3,true)
-
- val schema4 =
AvroConversionUtils.convertStructTypeToAvroSchema(structType4,
"test_struct_name", "test_namespace")
- val records4 = Seq(Row("key2", Row("innerKey2_1", 2L), Row("innerKey2_2",
2L, "new_nested_col_val1")))
- val df4 = spark.createDataFrame(spark.sparkContext.parallelize(records4),
structType4)
- val genRecRDD4 = HoodieSparkUtils.createRdd(df4, "test_struct_name",
"test_namespace", true,
- org.apache.hudi.common.util.Option.of(schema4))
- assert(schema4.equals(genRecRDD4.collect()(0).getSchema))
-
- // convert batch 1 with schema4. should succeed.
- val genRecRDD5 = HoodieSparkUtils.createRdd(df1, "test_struct_name",
"test_namespace", true,
- org.apache.hudi.common.util.Option.of(schema4))
- assert(schema4.equals(genRecRDD4.collect()(0).getSchema))
- val genRec = genRecRDD5.collect()(0)
- val nestedRec : GenericRecord =
genRec.get("nullableInnerStruct").asInstanceOf[GenericRecord]
- assertNull(nestedRec.get("new_nested_col"))
- assertNotNull(nestedRec.get("innerKey"))
- assertNotNull(nestedRec.get("innerValue"))
-
- val innerStruct4 = new
StructType().add("innerKey","string",false).add("innerValue", "long", true)
- .add("new_nested_col","string",false)
- // create a schema which has one additional nested column compared to
schema1, which is non nullable
- val structType6 = new StructType().add("key", "string", false)
-
.add("nonNullableInnerStruct",innerStruct1,false).add("nullableInnerStruct",innerStruct4,true)
-
- val schema6 =
AvroConversionUtils.convertStructTypeToAvroSchema(structType6,
"test_struct_name", "test_namespace")
- // convert batch 1 with schema5. should fail since the missed out column
is not nullable.
try {
- val genRecRDD6 = HoodieSparkUtils.createRdd(df1, "test_struct_name",
"test_namespace", true,
- org.apache.hudi.common.util.Option.of(schema6))
- genRecRDD6.collect()
- fail("createRdd should fail, because records don't have a column which
is not nullable in the passed in schema")
- } catch {
- case e: Exception =>
Review Comment:
Actually changed stuff in this Exception case block
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
