nsivabalan commented on a change in pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#discussion_r608313291
##########
File path:
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/AvroConversionUtils.scala
##########
@@ -49,7 +51,48 @@ object AvroConversionUtils {
def convertStructTypeToAvroSchema(structType: StructType,
structName: String,
recordNamespace: String): Schema = {
- SchemaConverters.toAvroType(structType, nullable = false, structName,
recordNamespace)
+ getAvroSchemaWithDefaults(SchemaConverters.toAvroType(structType, nullable
= false, structName, recordNamespace))
+ }
+
+ def getAvroSchemaWithDefaults(schema: Schema): Schema = {
Review comment:
docs
##########
File path:
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/HoodieSparkSqlWriterSuite.scala
##########
@@ -399,6 +401,90 @@ class HoodieSparkSqlWriterSuite extends FunSuite with
Matchers {
}
})
+ List(DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL,
DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)
+ .foreach(tableType => {
+ test("test schema evolution for " + tableType) {
+ initSparkContext("test_schema_evolution")
+ val path = java.nio.file.Files.createTempDirectory("hoodie_test_path")
+ try {
+ val hoodieFooTableName = "hoodie_foo_tbl"
+ //create a new table
+ val fooTableModifier = Map("path" -> path.toAbsolutePath.toString,
+ HoodieWriteConfig.TABLE_NAME -> hoodieFooTableName,
+ "hoodie.insert.shuffle.parallelism" -> "1",
+ "hoodie.upsert.shuffle.parallelism" -> "1",
+ DataSourceWriteOptions.TABLE_TYPE_OPT_KEY -> tableType,
+ DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY -> "_row_key",
+ DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY -> "partition",
+ DataSourceWriteOptions.KEYGENERATOR_CLASS_OPT_KEY ->
"org.apache.hudi.keygen.SimpleKeyGenerator")
+ val fooTableParams =
HoodieWriterUtils.parametersWithWriteDefaults(fooTableModifier)
+
+ // generate the inserts
+ var schema = DataSourceTestUtils.getStructTypeExampleSchema
+ var structType =
AvroConversionUtils.convertAvroSchemaToStructType(schema)
+ var records = DataSourceTestUtils.generateRandomRows(10)
+ var recordsSeq = convertRowListToSeq(records)
+ var df1 = spark.createDataFrame(sc.parallelize(recordsSeq),
structType)
+ // write to Hudi
+ HoodieSparkSqlWriter.write(sqlContext, SaveMode.Overwrite,
fooTableParams, df1)
+
+ val snapshotDF1 = spark.read.format("org.apache.hudi")
+ .load(path.toAbsolutePath.toString + "/*/*/*/*")
+ assertEquals(10, snapshotDF1.count())
+
+ // remove metadata columns so that expected and actual DFs can be
compared as is
+ val trimmedDf1 =
snapshotDF1.drop(HoodieRecord.HOODIE_META_COLUMNS.get(0)).drop(HoodieRecord.HOODIE_META_COLUMNS.get(1))
+
.drop(HoodieRecord.HOODIE_META_COLUMNS.get(2)).drop(HoodieRecord.HOODIE_META_COLUMNS.get(3))
+ .drop(HoodieRecord.HOODIE_META_COLUMNS.get(4))
+
+ assert(df1.except(trimmedDf1).count() == 0)
+
+ // issue updates so that log files are created for MOR table
+ var updates = DataSourceTestUtils.generateUpdates(records, 5);
+ var updatesSeq = convertRowListToSeq(updates)
+ var updatesDf = spark.createDataFrame(sc.parallelize(updatesSeq),
structType)
+ // write to Hudi
+ HoodieSparkSqlWriter.write(sqlContext, SaveMode.Append,
fooTableParams, updatesDf)
+
+ val snapshotDF2 = spark.read.format("org.apache.hudi")
+ .load(path.toAbsolutePath.toString + "/*/*/*/*")
+ assertEquals(10, snapshotDF2.count())
+
+ // remove metadata columns so that expected and actual DFs can be
compared as is
+ val trimmedDf2 =
snapshotDF1.drop(HoodieRecord.HOODIE_META_COLUMNS.get(0)).drop(HoodieRecord.HOODIE_META_COLUMNS.get(1))
+
.drop(HoodieRecord.HOODIE_META_COLUMNS.get(2)).drop(HoodieRecord.HOODIE_META_COLUMNS.get(3))
+ .drop(HoodieRecord.HOODIE_META_COLUMNS.get(4))
+
+ // ensure 2nd batch of updates matches.
+ assert(updatesDf.intersect(trimmedDf2).except(updatesDf).count() ==
0)
+
+ schema = DataSourceTestUtils.getStructTypeExampleEvolvedSchema
Review comment:
add a comment here that we are ingesting w/ evolved schema
##########
File path:
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestAvroConversionUtils.scala
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi
+
+import org.apache.avro.Schema
+import org.apache.spark.sql.types.StructType
+import org.scalatest.{FunSuite, Matchers}
+
+class TestAvroConversionUtils extends FunSuite with Matchers {
+
+
+ test("test convertStructTypeToAvroSchema") {
Review comment:
can we also add tests for other types. map, array and union.
##########
File path:
hudi-utilities/src/test/resources/delta-streamer-config/source-jdbc.avsc
##########
@@ -26,34 +26,42 @@
},
{
"name": "TIMESTAMP",
- "type": ["double", "null"]
+ "type": ["null", "double"],
Review comment:
why change this. are tests failing?
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
