You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2021/04/04 09:49:34 UTC

[GitHub] [hudi] aditiwari01 opened a new pull request #2765: [HUDI-1716]: Resolvig default values for schema from dataframe

aditiwari01 opened a new pull request #2765:
URL: https://github.com/apache/hudi/pull/2765


   ## What is the purpose of the pull request
   
   For Union schema types, avro expects "null" to be first entry for null to be considered as default value. But since Hudi leverages SchemaConverters.toAvroType(...) from org.apache.spark:spark-avro_* library, structType to avro results in "null" being 2nd entry for UNION type schemas. Also, there is no default value set in this avro schema thus generated. This patch fixes this issue.
   
   This also fixes simple schema evolution w/ MOR tables. adding new columns was failing if not for this patch.
   
   For eg:
   if incoming structType is
   StructType(StructField(rowId,StringType,true), StructField(partitionId,StringType,true), StructField(preComb,LongType,true), StructField(name,StringType,true))
   
   Generated avro scheme if not for this patch:
   {"type":"record","name":"hudi_trips_cow_record","namespace":"hoodie.hudi_trips_cow","fields":[{"name":"rowId","type":["string","null"]},{"name":"partitionId","type":["string","null"]},{"name":"preComb","type":["string","null"]},{"name":"name","type":["string","null"]}]}
   
   Note that "null" is 2nd entry in UNIONs and there is no default set.
   
   Generated avro schema with this patch:
   {"type":"record","name":"hudi_trips_cow_record","namespace":"hoodie.hudi_trips_cow","fields":[{"name":"rowId","type":["null","string"],"default":null},{"name":"partitionId","type":["null","string"],"default":null},{"name":"preComb","type":["null","long"],"default":null},{"name":"name","type":["null","string"],"default":null},{"name":"newField","type":["null","string"],"default":null}]}
   
   Note that default value is null and not null in string format. So, this should work for other data types as well(not just strings). Default value of null is allowed only if null is the first entry in UNION for a given schema field.
   
   ## Brief change log
   
   Fixing struct type to avro schema conversion to fix null as first entry in UNION schema types and adds default values for the same.
   
   ## Verify this pull request
   
   Added unit test for convertor function and functional test for schema evolution in HoodieSparkSqlWriterSuite.
   
   ## Committer checklist
   
    - [ *] Has a corresponding JIRA in PR title & commit
    
    - [ *] Commit message is descriptive of the change
    
    - [ *] CI is green
   
    - [ ] Necessary doc changes done or have another open PR (Not required)
          
    - [ ] For large changes, please consider breaking it into sub-tasks under an umbrella JIRA. (Not required)


-- 
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:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on a change in pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
nsivabalan commented on a change in pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#discussion_r615879081



##########
File path: hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/AvroConversionUtils.scala
##########
@@ -46,10 +48,67 @@ object AvroConversionUtils {
     }
   }
 
+  /**
+    *
+    * Returns avro schema from spark StructType.
+    *
+    * @param structType       Dataframe Struct Type.
+    * @param structName       Avro record name.
+    * @param recordNamespace  Avro record namespace.
+    * @return                 Avro schema corresponding to given struct type.
+    */
   def convertStructTypeToAvroSchema(structType: StructType,
                                     structName: String,
                                     recordNamespace: String): Schema = {
-    SchemaConverters.toAvroType(structType, nullable = false, structName, recordNamespace)
+    getAvroSchemaWithDefaults(SchemaConverters.toAvroType(structType, nullable = false, structName, recordNamespace))
+  }
+
+  /**
+    *
+    * Method to add default value of null to nullable fields in given avro schema
+    *
+    * @param schema     input avro schema
+    * @return           Avro schema with null default set to nullable fields
+    */
+  def getAvroSchemaWithDefaults(schema: Schema): Schema = {
+
+    schema.getType match {
+      case Schema.Type.RECORD => {
+
+        val modifiedFields = schema.getFields.map(field => {
+          val newSchema = getAvroSchemaWithDefaults(field.schema())
+          field.schema().getType match {
+            case Schema.Type.UNION => {
+              val innerFields = newSchema.getTypes
+              val containsNullSchema = innerFields.foldLeft(false)((nullFieldEncountered, schema) => nullFieldEncountered | schema.getType == Schema.Type.NULL)
+              if(containsNullSchema) {
+                // Need to re shuffle the fields in list because to set null as default, null schema must be head in union schema
+                val restructuredNewSchema = Schema.createUnion(List(Schema.create(Schema.Type.NULL)) ++ innerFields.filter(innerSchema => !(innerSchema.getType == Schema.Type.NULL)))
+                new Schema.Field(field.name(), restructuredNewSchema, field.doc(), JsonProperties.NULL_VALUE)

Review comment:
       sounds good πŸ‘ 




-- 
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:
users@infra.apache.org

[GitHub] [hudi] aditiwari01 commented on a change in pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
aditiwari01 commented on a change in pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#discussion_r608436934



##########
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 = {
+
+    schema.getType match {
+      case Schema.Type.RECORD => {
+
+        val modifiedFields = schema.getFields.map(field => {
+          val newSchema = getAvroSchemaWithDefaults(field.schema())
+          field.schema().getType match {
+            case Schema.Type.UNION => {
+              val innerFields = newSchema.getTypes
+              val containsNullSchema = innerFields.foldLeft(false)((nullFieldEncountered, schema) => nullFieldEncountered | schema.getType == Schema.Type.NULL)
+              if(containsNullSchema) {
+                // Need to re shuffel the fields in list becase to set null as default, null schema must be head in union schema
+                val restructuredNewSchema = Schema.createUnion(List(Schema.create(Schema.Type.NULL)) ++ innerFields.filter(innerSchema => !(innerSchema.getType == Schema.Type.NULL)))
+                new Schema.Field(field.name(), restructuredNewSchema, field.doc(), JsonProperties.NULL_VALUE)
+              } else {
+                new Schema.Field(field.name(), newSchema, field.doc(), field.defaultVal())
+              }
+            }
+            case _ => new Schema.Field(field.name(), newSchema, field.doc(), field.defaultVal())
+          }
+        }).toList
+        Schema.createRecord(schema.getName, schema.getDoc, schema.getNamespace, schema.isError, modifiedFields)
+      }
+
+      case Schema.Type.UNION => {

Review comment:
       Union is already covered in this test. Will add more tests for Map and Array.




-- 
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:
users@infra.apache.org

[GitHub] [hudi] codecov-io edited a comment on pull request #2765: [HUDI-1716]: Resolvig default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#issuecomment-813008111


   # [Codecov](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=h1) Report
   > Merging [#2765](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=desc) (7761117) into [master](https://codecov.io/gh/apache/hudi/commit/e970e1f48302aec3af7eeca009a2c793757cd501?el=desc) (e970e1f) will **decrease** coverage by `0.00%`.
   > The diff coverage is `n/a`.
   
   [![Impacted file tree graph](https://codecov.io/gh/apache/hudi/pull/2765/graphs/tree.svg?width=650&height=150&src=pr&token=VTTXabwbs2)](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=tree)
   
   ```diff
   @@             Coverage Diff              @@
   ##             master    #2765      +/-   ##
   ============================================
   - Coverage     52.32%   52.31%   -0.01%     
   + Complexity     3689     3688       -1     
   ============================================
     Files           483      483              
     Lines         23095    23095              
     Branches       2460     2460              
   ============================================
   - Hits          12084    12082       -2     
   - Misses         9942     9945       +3     
   + Partials       1069     1068       -1     
   ```
   
   | Flag | Coverage Ξ” | Complexity Ξ” | |
   |---|---|---|---|
   | hudicli | `37.01% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudiclient | `βˆ… <ΓΈ> (βˆ…)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudicommon | `50.80% <ΓΈ> (-0.03%)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudiflink | `56.71% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudihadoopmr | `33.44% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudisparkdatasource | `71.38% <ΓΈ> (+0.05%)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudisync | `45.47% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | huditimelineservice | `64.36% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudiutilities | `69.69% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   | [Impacted Files](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=tree) | Coverage Ξ” | Complexity Ξ” | |
   |---|---|---|---|
   | [...ache/hudi/common/fs/inline/InMemoryFileSystem.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL2ZzL2lubGluZS9Jbk1lbW9yeUZpbGVTeXN0ZW0uamF2YQ==) | `79.31% <0.00%> (-10.35%)` | `15.00% <0.00%> (-1.00%)` | |
   | [...n/scala/org/apache/hudi/HoodieMergeOnReadRDD.scala](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3Bhcmsvc3JjL21haW4vc2NhbGEvb3JnL2FwYWNoZS9odWRpL0hvb2RpZU1lcmdlT25SZWFkUkRELnNjYWxh) | `90.76% <0.00%> (+0.54%)` | `15.00% <0.00%> (ΓΈ%)` | |
   


-- 
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:
users@infra.apache.org

[GitHub] [hudi] aditiwari01 commented on a change in pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
aditiwari01 commented on a change in pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#discussion_r608434413



##########
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:
       Union is already covered in this test. Will add more tests for Map and Array.




-- 
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:
users@infra.apache.org

[GitHub] [hudi] lw309637554 commented on a change in pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
lw309637554 commented on a change in pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#discussion_r608395652



##########
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 = {
+
+    schema.getType match {
+      case Schema.Type.RECORD => {
+
+        val modifiedFields = schema.getFields.map(field => {
+          val newSchema = getAvroSchemaWithDefaults(field.schema())
+          field.schema().getType match {
+            case Schema.Type.UNION => {
+              val innerFields = newSchema.getTypes
+              val containsNullSchema = innerFields.foldLeft(false)((nullFieldEncountered, schema) => nullFieldEncountered | schema.getType == Schema.Type.NULL)
+              if(containsNullSchema) {
+                // Need to re shuffel the fields in list becase to set null as default, null schema must be head in union schema

Review comment:
       β€œNeed to re shuffel the fields in list becase” -> "Need to re shuffle the fields in list because" ?




-- 
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:
users@infra.apache.org

[GitHub] [hudi] lw309637554 commented on pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
lw309637554 commented on pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#issuecomment-815405833


   LGTM


-- 
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:
users@infra.apache.org

[GitHub] [hudi] nsivabalan merged pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
nsivabalan merged pull request #2765:
URL: https://github.com/apache/hudi/pull/2765


   


-- 
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:
users@infra.apache.org

[GitHub] [hudi] codecov-io commented on pull request #2765: [HUDI-1716]: Resolvig default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
codecov-io commented on pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#issuecomment-813008111


   # [Codecov](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=h1) Report
   > Merging [#2765](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=desc) (7761117) into [master](https://codecov.io/gh/apache/hudi/commit/e970e1f48302aec3af7eeca009a2c793757cd501?el=desc) (e970e1f) will **increase** coverage by `17.37%`.
   > The diff coverage is `n/a`.
   
   [![Impacted file tree graph](https://codecov.io/gh/apache/hudi/pull/2765/graphs/tree.svg?width=650&height=150&src=pr&token=VTTXabwbs2)](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=tree)
   
   ```diff
   @@              Coverage Diff              @@
   ##             master    #2765       +/-   ##
   =============================================
   + Coverage     52.32%   69.69%   +17.37%     
   + Complexity     3689      371     -3318     
   =============================================
     Files           483       54      -429     
     Lines         23095     1993    -21102     
     Branches       2460      236     -2224     
   =============================================
   - Hits          12084     1389    -10695     
   + Misses         9942      473     -9469     
   + Partials       1069      131      -938     
   ```
   
   | Flag | Coverage Ξ” | Complexity Ξ” | |
   |---|---|---|---|
   | hudicli | `?` | `?` | |
   | hudiclient | `?` | `?` | |
   | hudicommon | `?` | `?` | |
   | hudiflink | `?` | `?` | |
   | hudihadoopmr | `?` | `?` | |
   | hudisparkdatasource | `?` | `?` | |
   | hudisync | `?` | `?` | |
   | huditimelineservice | `?` | `?` | |
   | hudiutilities | `69.69% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   | [Impacted Files](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=tree) | Coverage Ξ” | Complexity Ξ” | |
   |---|---|---|---|
   | [...va/org/apache/hudi/configuration/FlinkOptions.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS9jb25maWd1cmF0aW9uL0ZsaW5rT3B0aW9ucy5qYXZh) | | | |
   | [...til/jvm/HotSpotMemoryLayoutSpecification32bit.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL3V0aWwvanZtL0hvdFNwb3RNZW1vcnlMYXlvdXRTcGVjaWZpY2F0aW9uMzJiaXQuamF2YQ==) | | | |
   | [...di/common/bootstrap/index/HFileBootstrapIndex.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL2Jvb3RzdHJhcC9pbmRleC9IRmlsZUJvb3RzdHJhcEluZGV4LmphdmE=) | | | |
   | [...apache/hudi/table/format/cow/RunLengthDecoder.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1mbGluay9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaHVkaS90YWJsZS9mb3JtYXQvY293L1J1bkxlbmd0aERlY29kZXIuamF2YQ==) | | | |
   | [...rg/apache/hudi/io/storage/HoodieParquetReader.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvaW8vc3RvcmFnZS9Ib29kaWVQYXJxdWV0UmVhZGVyLmphdmE=) | | | |
   | [.../java/org/apache/hudi/HoodieDataSourceHelpers.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3Bhcmsvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvSG9vZGllRGF0YVNvdXJjZUhlbHBlcnMuamF2YQ==) | | | |
   | [...pache/hudi/common/model/HoodieArchivedLogFile.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL21vZGVsL0hvb2RpZUFyY2hpdmVkTG9nRmlsZS5qYXZh) | | | |
   | [...he/hudi/common/fs/SizeAwareFSDataOutputStream.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL2ZzL1NpemVBd2FyZUZTRGF0YU91dHB1dFN0cmVhbS5qYXZh) | | | |
   | [...g/apache/hudi/hadoop/HoodieParquetInputFormat.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1oYWRvb3AtbXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvaGFkb29wL0hvb2RpZVBhcnF1ZXRJbnB1dEZvcm1hdC5qYXZh) | | | |
   | [...rc/main/java/org/apache/hudi/cli/HoodiePrompt.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1jbGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY2xpL0hvb2RpZVByb21wdC5qYXZh) | | | |
   | ... and [419 more](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree-more) | |
   


-- 
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:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on a change in pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
nsivabalan commented on a change in pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#discussion_r615823969



##########
File path: hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/AvroConversionUtils.scala
##########
@@ -46,10 +48,67 @@ object AvroConversionUtils {
     }
   }
 
+  /**
+    *
+    * Returns avro schema from spark StructType.
+    *
+    * @param structType       Dataframe Struct Type.
+    * @param structName       Avro record name.
+    * @param recordNamespace  Avro record namespace.
+    * @return                 Avro schema corresponding to given struct type.
+    */
   def convertStructTypeToAvroSchema(structType: StructType,
                                     structName: String,
                                     recordNamespace: String): Schema = {
-    SchemaConverters.toAvroType(structType, nullable = false, structName, recordNamespace)
+    getAvroSchemaWithDefaults(SchemaConverters.toAvroType(structType, nullable = false, structName, recordNamespace))
+  }
+
+  /**
+    *
+    * Method to add default value of null to nullable fields in given avro schema
+    *
+    * @param schema     input avro schema
+    * @return           Avro schema with null default set to nullable fields
+    */
+  def getAvroSchemaWithDefaults(schema: Schema): Schema = {
+
+    schema.getType match {
+      case Schema.Type.RECORD => {
+
+        val modifiedFields = schema.getFields.map(field => {
+          val newSchema = getAvroSchemaWithDefaults(field.schema())
+          field.schema().getType match {
+            case Schema.Type.UNION => {
+              val innerFields = newSchema.getTypes
+              val containsNullSchema = innerFields.foldLeft(false)((nullFieldEncountered, schema) => nullFieldEncountered | schema.getType == Schema.Type.NULL)
+              if(containsNullSchema) {
+                // Need to re shuffle the fields in list because to set null as default, null schema must be head in union schema
+                val restructuredNewSchema = Schema.createUnion(List(Schema.create(Schema.Type.NULL)) ++ innerFields.filter(innerSchema => !(innerSchema.getType == Schema.Type.NULL)))
+                new Schema.Field(field.name(), restructuredNewSchema, field.doc(), JsonProperties.NULL_VALUE)

Review comment:
       @aditiwari01 : is it possible for a schema.Field to be nullable and have a default value that is not null? if yes, shouldn't we set the last arg accordingly? as of now, we are always setting null as default value. 




-- 
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:
users@infra.apache.org

[GitHub] [hudi] aditiwari01 closed pull request #2765: [HUDI-1716]: Resolvig default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
aditiwari01 closed pull request #2765:
URL: https://github.com/apache/hudi/pull/2765


   


-- 
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:
users@infra.apache.org

[GitHub] [hudi] aditiwari01 commented on a change in pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
aditiwari01 commented on a change in pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#discussion_r608433997



##########
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:
       Yes. This schema is used as Expected schema in one of the tests. Changed the schema to the evolved schema.




-- 
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:
users@infra.apache.org

[GitHub] [hudi] codecov-io edited a comment on pull request #2765: [HUDI-1716]: Resolvig default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#issuecomment-813008111


   # [Codecov](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=h1) Report
   > Merging [#2765](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=desc) (7761117) into [master](https://codecov.io/gh/apache/hudi/commit/e970e1f48302aec3af7eeca009a2c793757cd501?el=desc) (e970e1f) will **decrease** coverage by `0.01%`.
   > The diff coverage is `n/a`.
   
   [![Impacted file tree graph](https://codecov.io/gh/apache/hudi/pull/2765/graphs/tree.svg?width=650&height=150&src=pr&token=VTTXabwbs2)](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=tree)
   
   ```diff
   @@             Coverage Diff              @@
   ##             master    #2765      +/-   ##
   ============================================
   - Coverage     52.32%   52.30%   -0.02%     
   + Complexity     3689     3688       -1     
   ============================================
     Files           483      483              
     Lines         23095    23095              
     Branches       2460     2460              
   ============================================
   - Hits          12084    12080       -4     
   - Misses         9942     9946       +4     
     Partials       1069     1069              
   ```
   
   | Flag | Coverage Ξ” | Complexity Ξ” | |
   |---|---|---|---|
   | hudicli | `37.01% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudiclient | `βˆ… <ΓΈ> (βˆ…)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudicommon | `50.78% <ΓΈ> (-0.05%)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudiflink | `56.71% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudihadoopmr | `33.44% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudisparkdatasource | `71.38% <ΓΈ> (+0.05%)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudisync | `45.47% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | huditimelineservice | `64.36% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   | hudiutilities | `69.69% <ΓΈ> (ΓΈ)` | `0.00 <ΓΈ> (ΓΈ)` | |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   | [Impacted Files](https://codecov.io/gh/apache/hudi/pull/2765?src=pr&el=tree) | Coverage Ξ” | Complexity Ξ” | |
   |---|---|---|---|
   | [...ache/hudi/common/fs/inline/InMemoryFileSystem.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL2ZzL2lubGluZS9Jbk1lbW9yeUZpbGVTeXN0ZW0uamF2YQ==) | `79.31% <0.00%> (-10.35%)` | `15.00% <0.00%> (-1.00%)` | |
   | [...e/hudi/common/table/log/HoodieLogFormatWriter.java](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL3RhYmxlL2xvZy9Ib29kaWVMb2dGb3JtYXRXcml0ZXIuamF2YQ==) | `78.12% <0.00%> (-1.57%)` | `26.00% <0.00%> (ΓΈ%)` | |
   | [...n/scala/org/apache/hudi/HoodieMergeOnReadRDD.scala](https://codecov.io/gh/apache/hudi/pull/2765/diff?src=pr&el=tree#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3Bhcmsvc3JjL21haW4vc2NhbGEvb3JnL2FwYWNoZS9odWRpL0hvb2RpZU1lcmdlT25SZWFkUkRELnNjYWxh) | `90.76% <0.00%> (+0.54%)` | `15.00% <0.00%> (ΓΈ%)` | |
   


-- 
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:
users@infra.apache.org

[GitHub] [hudi] aditiwari01 commented on a change in pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
aditiwari01 commented on a change in pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#discussion_r615864702



##########
File path: hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/AvroConversionUtils.scala
##########
@@ -46,10 +48,67 @@ object AvroConversionUtils {
     }
   }
 
+  /**
+    *
+    * Returns avro schema from spark StructType.
+    *
+    * @param structType       Dataframe Struct Type.
+    * @param structName       Avro record name.
+    * @param recordNamespace  Avro record namespace.
+    * @return                 Avro schema corresponding to given struct type.
+    */
   def convertStructTypeToAvroSchema(structType: StructType,
                                     structName: String,
                                     recordNamespace: String): Schema = {
-    SchemaConverters.toAvroType(structType, nullable = false, structName, recordNamespace)
+    getAvroSchemaWithDefaults(SchemaConverters.toAvroType(structType, nullable = false, structName, recordNamespace))
+  }
+
+  /**
+    *
+    * Method to add default value of null to nullable fields in given avro schema
+    *
+    * @param schema     input avro schema
+    * @return           Avro schema with null default set to nullable fields
+    */
+  def getAvroSchemaWithDefaults(schema: Schema): Schema = {
+
+    schema.getType match {
+      case Schema.Type.RECORD => {
+
+        val modifiedFields = schema.getFields.map(field => {
+          val newSchema = getAvroSchemaWithDefaults(field.schema())
+          field.schema().getType match {
+            case Schema.Type.UNION => {
+              val innerFields = newSchema.getTypes
+              val containsNullSchema = innerFields.foldLeft(false)((nullFieldEncountered, schema) => nullFieldEncountered | schema.getType == Schema.Type.NULL)
+              if(containsNullSchema) {
+                // Need to re shuffle the fields in list because to set null as default, null schema must be head in union schema
+                val restructuredNewSchema = Schema.createUnion(List(Schema.create(Schema.Type.NULL)) ++ innerFields.filter(innerSchema => !(innerSchema.getType == Schema.Type.NULL)))
+                new Schema.Field(field.name(), restructuredNewSchema, field.doc(), JsonProperties.NULL_VALUE)

Review comment:
       This method is used to get Avro schema from spark dataframe struct type. We can not assign default values in struct. Hence here it does not make sense to have some specific value as default. Either the column would be present in dataframe or not. If it's not present and it is nullable, default must be added as null.
   
   Any default logic that actual users might want must go in their business logic before calling df.write since spark itself doesnot handle default value.




-- 
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:
users@infra.apache.org

[GitHub] [hudi] lw309637554 commented on a change in pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
lw309637554 commented on a change in pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#discussion_r608395924



##########
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 = {
+
+    schema.getType match {
+      case Schema.Type.RECORD => {
+
+        val modifiedFields = schema.getFields.map(field => {
+          val newSchema = getAvroSchemaWithDefaults(field.schema())
+          field.schema().getType match {
+            case Schema.Type.UNION => {
+              val innerFields = newSchema.getTypes
+              val containsNullSchema = innerFields.foldLeft(false)((nullFieldEncountered, schema) => nullFieldEncountered | schema.getType == Schema.Type.NULL)
+              if(containsNullSchema) {
+                // Need to re shuffel the fields in list becase to set null as default, null schema must be head in union schema
+                val restructuredNewSchema = Schema.createUnion(List(Schema.create(Schema.Type.NULL)) ++ innerFields.filter(innerSchema => !(innerSchema.getType == Schema.Type.NULL)))
+                new Schema.Field(field.name(), restructuredNewSchema, field.doc(), JsonProperties.NULL_VALUE)
+              } else {
+                new Schema.Field(field.name(), newSchema, field.doc(), field.defaultVal())
+              }
+            }
+            case _ => new Schema.Field(field.name(), newSchema, field.doc(), field.defaultVal())
+          }
+        }).toList
+        Schema.createRecord(schema.getName, schema.getDoc, schema.getNamespace, schema.isError, modifiedFields)
+      }
+
+      case Schema.Type.UNION => {

Review comment:
       can we add some test case to test this type 




-- 
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:
users@infra.apache.org

[GitHub] [hudi] lw309637554 commented on pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
lw309637554 commented on pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#issuecomment-814636227


   > @lw309637554 : may be you wanna review this.
   
   @nsivabalan thanks , if this pr land, https://github.com/apache/hudi/pull/2776 will be ok , and i add more test for schema evolution .


-- 
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:
users@infra.apache.org

[GitHub] [hudi] codecov-io edited a comment on pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#issuecomment-813008111






-- 
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:
users@infra.apache.org

[GitHub] [hudi] codecov-io edited a comment on pull request #2765: [HUDI-1716]: Resolvig default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#issuecomment-813008111






-- 
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:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
nsivabalan commented on pull request #2765:
URL: https://github.com/apache/hudi/pull/2765#issuecomment-814563007


   @lw309637554 : may be you wanna review this. 


-- 
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:
users@infra.apache.org

[GitHub] [hudi] nsivabalan commented on a change in pull request #2765: [HUDI-1716]: Resolving default values for schema from dataframe

Posted by GitBox <gi...@apache.org>.
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:
users@infra.apache.org