Different parquet reader config on list-typed fields is used to read parquet file generated by clustering

We saw an issue in Hudi 0.14 that `HoodieAvroParquetReader` returns null on the list-typed field on reading a parquet file generated by replacecommit (clustering), see screenshot below.  This caused NullPointerException during indexing with global index using HoodieMergedReadHandle
{code:java}
org.apache.hudi.exception.HoodieException: Unable to instantiate payload class 
    at org.apache.hudi.common.util.HoodieRecordUtils.loadPayload(HoodieRecordUtils.java:101) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.common.util.SpillableMapUtils.convertToHoodieRecordPayload(SpillableMapUtils.java:154) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.common.util.SpillableMapUtils.convertToHoodieRecordPayload(SpillableMapUtils.java:122) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.avro.HoodieAvroUtils.createHoodieRecordFromAvro(HoodieAvroUtils.java:1404) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.common.model.HoodieAvroIndexedRecord.wrapIntoHoodieRecordPayloadWithParams(HoodieAvroIndexedRecord.java:159) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.io.HoodieMergedReadHandle.doMergedRead(HoodieMergedReadHandle.java:164) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.io.HoodieMergedReadHandle.getMergedRecords(HoodieMergedReadHandle.java:91) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.index.HoodieIndexUtils.lambda$getExistingRecords$c7e45d15$1(HoodieIndexUtils.java:246) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.data.HoodieJavaRDD.lambda$flatMap$a6598fcb$1(HoodieJavaRDD.java:137) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.spark.api.java.JavaRDDLike.$anonfun$flatMap$1(JavaRDDLike.scala:125) ~[spark-core_2.12-3.2.3.jar:3.2.3]
    at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:486) ~[scala-library-2.12.19.jar:?]
    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:492) ~[scala-library-2.12.19.jar:?]
    at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) ~[scala-library-2.12.19.jar:?]
    at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:140) ~[spark-core_2.12-3.2.3.jar:3.2.3]
    at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) ~[spark-core_2.12-3.2.3.jar:3.2.3]
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) ~[spark-core_2.12-3.2.3.jar:3.2.3]
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52) ~[spark-core_2.12-3.2.3.jar:3.2.3]
    at org.apache.spark.scheduler.Task.run(Task.scala:131) ~[spark-core_2.12-3.2.3.jar:3.2.3]
    at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506) ~[spark-core_2.12-3.2.3.jar:3.2.3]
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1491) ~[spark-core_2.12-3.2.3.jar:3.2.3]
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509) ~[spark-core_2.12-3.2.3.jar:3.2.3]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_412]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_412]
    at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_412]
Caused by: java.lang.reflect.InvocationTargetException
    at sun.reflect.GeneratedConstructorAccessor53.newInstance(Unknown Source) ~[?:?]
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[?:1.8.0_412]
    at java.lang.reflect.Constructor.newInstance(Constructor.java:423) ~[?:1.8.0_412]
    at org.apache.hudi.common.util.HoodieRecordUtils.loadPayload(HoodieRecordUtils.java:99) ~[hudi-utilities-bundle_2.12.jar]
    ... 23 more
Caused by: java.lang.NullPointerException: null value for (non-nullable) List at triprec.tip_history
    at org.apache.avro.path.TracingNullPointException.summarize(TracingNullPointException.java:88) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.avro.path.TracingNullPointException.summarize(TracingNullPointException.java:30) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:84) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.hudi.avro.HoodieAvroUtils.indexedRecordToBytes(HoodieAvroUtils.java:189) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.avro.HoodieAvroUtils.avroToBytes(HoodieAvroUtils.java:181) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.common.model.BaseAvroPayload.(BaseAvroPayload.java:53) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.common.model.OverwriteWithLatestAvroPayload.(OverwriteWithLatestAvroPayload.java:44) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.common.model.DefaultHoodieRecordPayload.(DefaultHoodieRecordPayload.java:53) ~[hudi-utilities-bundle_2.12.jar]
    at sun.reflect.GeneratedConstructorAccessor53.newInstance(Unknown Source) ~[?:?]
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[?:1.8.0_412]
    at java.lang.reflect.Constructor.newInstance(Constructor.java:423) ~[?:1.8.0_412]
    at org.apache.hudi.common.util.HoodieRecordUtils.loadPayload(HoodieRecordUtils.java:99) ~[hudi-utilities-bundle_2.12.jar]
    ... 23 more
Caused by: java.lang.NullPointerException
    at org.apache.avro.generic.GenericDatumWriter.getArraySize(GenericDatumWriter.java:315) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.avro.generic.GenericDatumWriter.writeArray(GenericDatumWriter.java:281) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:151) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:95) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.avro.generic.GenericDatumWriter.writeField(GenericDatumWriter.java:245) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.avro.generic.GenericDatumWriter.writeRecord(GenericDatumWriter.java:234) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:145) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:95) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:82) ~[avro-1.11.3.jar:1.11.3]
    at org.apache.hudi.avro.HoodieAvroUtils.indexedRecordToBytes(HoodieAvroUtils.java:189) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.avro.HoodieAvroUtils.avroToBytes(HoodieAvroUtils.java:181) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.common.model.BaseAvroPayload.(BaseAvroPayload.java:53) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.common.model.OverwriteWithLatestAvroPayload.(OverwriteWithLatestAvroPayload.java:44) ~[hudi-utilities-bundle_2.12.jar]
    at org.apache.hudi.common.model.DefaultHoodieRecordPayload.(DefaultHoodieRecordPayload.java:53) ~[hudi-utilities-bundle_2.12.jar]
    at sun.reflect.GeneratedConstructorAccessor53.newInstance(Unknown Source) ~[?:?]
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[?:1.8.0_412]
    at java.lang.reflect.Constructor.newInstance(Constructor.java:423) ~[?:1.8.0_412]
    at org.apache.hudi.common.util.HoodieRecordUtils.loadPayload(HoodieRecordUtils.java:99) ~[hudi-utilities-bundle_2.12.jar]
    ... 23 more
{code}
However, the parquet file itself has non-null values for the field (see "tip_history" field):
{code:java}
FROM replacecommit:
parquet.avro.schema: {"type":"record","name":"triprec","fields":[{"name":"_hoodie_commit_time","type":["null","string"],"doc":"","default":null},{"name":"_hoodie_commit_seqno","type":["null","string"],"doc":"","default":null},{"name":"_hoodie_record_key","type":["null","string"],"doc":"","default":null},{"name":"_hoodie_partition_path","type":["null","string"],"doc":"","default":null},{"name":"_hoodie_file_name","type":["null","string"],"doc":"","default":null},{"name":"timestamp","type":"long"},{"name":"_row_key","type":"string"},{"name":"partition_path","type":["null","string"],"default":null},{"name":"trip_type","type":{"type":"enum","name":"TripType","symbols":["UNKNOWN","UBERX","BLACK"],"default":"UNKNOWN"}},{"name":"rider","type":"string"},{"name":"driver","type":"string"},{"name":"begin_lat","type":"double"},{"name":"begin_lon","type":"double"},{"name":"end_lat","type":"double"},{"name":"end_lon","type":"double"},{"name":"distance_in_meters","type":"int"},{"name":"seconds_since_epoch","type":"long"},{"name":"weight","type":"float"},{"name":"nation","type":"bytes"},{"name":"current_date","type":{"type":"int","logicalType":"date"}},{"name":"current_ts","type":"long"},{"name":"height","type":{"type":"fixed","name":"abc","size":5,"logicalType":"decimal","precision":10,"scale":6}},{"name":"city_to_state","type":{"type":"map","values":"string"}},{"name":"fare","type":{"type":"record","name":"fare","fields":[{"name":"amount","type":"double"},{"name":"currency","type":"string"}]}},{"name":"tip_history","type":{"type":"array","items":{"type":"record","name":"tip_history","fields":[{"name":"amount","type":"double"},{"name":"currency","type":"string"}],"default":null},"default":[]},"default":[]},{"name":"_hoodie_is_deleted","type":"boolean","default":false}]}
writer.model.name: avroSchema:
message triprec {
  ..
  required group tip_history (LIST) = 25 {
    repeated group array = 31 {
      required double amount = 32;
      required binary currency (STRING) = 33;
    }
  }
}Row group 0:  count: 123  101.62 B records  start: 4  total(compressed): 12.206 kB total(uncompressed):19.198 kB 
--------------------------------------------------------------------------------
                               type      encodings count     avg size   nulls   min / max
tip_history.array.amount       DOUBLE    G   _     123       8.44 B     0       "1.5205662454434887" / "99.8588537812955"
tip_history.array.currency     BINARY    G _ R     123       0.82 B     0       "USD" / "USD" {code}
 

Comparing the parquet metadata with the parquet files generated the deltacommit does not reveal any difference in schema:
{code:java}
FROM deltacommit:
parquet.avro.schema: {"type":"record","name":"triprec","fields":[{"name":"_hoodie_commit_time","type":["null","string"],"doc":"","default":null},{"name":"_hoodie_commit_seqno","type":["null","string"],"doc":"","default":null},{"name":"_hoodie_record_key","type":["null","string"],"doc":"","default":null},{"name":"_hoodie_partition_path","type":["null","string"],"doc":"","default":null},{"name":"_hoodie_file_name","type":["null","string"],"doc":"","default":null},{"name":"timestamp","type":"long"},{"name":"_row_key","type":"string"},{"name":"partition_path","type":["null","string"],"default":null},{"name":"trip_type","type":{"type":"enum","name":"TripType","symbols":["UNKNOWN","UBERX","BLACK"],"default":"UNKNOWN"}},{"name":"rider","type":"string"},{"name":"driver","type":"string"},{"name":"begin_lat","type":"double"},{"name":"begin_lon","type":"double"},{"name":"end_lat","type":"double"},{"name":"end_lon","type":"double"},{"name":"distance_in_meters","type":"int"},{"name":"seconds_since_epoch","type":"long"},{"name":"weight","type":"float"},{"name":"nation","type":"bytes"},{"name":"current_date","type":{"type":"int","logicalType":"date"}},{"name":"current_ts","type":"long"},{"name":"height","type":{"type":"fixed","name":"abc","size":5,"logicalType":"decimal","precision":10,"scale":6}},{"name":"city_to_state","type":{"type":"map","values":"string"}},{"name":"fare","type":{"type":"record","name":"fare","fields":[{"name":"amount","type":"double"},{"name":"currency","type":"string"}]}},{"name":"tip_history","type":{"type":"array","items":{"type":"record","name":"tip_history","fields":[{"name":"amount","type":"double"},{"name":"currency","type":"string"}],"default":null},"default":[]},"default":[]},{"name":"_hoodie_is_deleted","type":"boolean","default":false}]}
writer.model.name: avro
Schema:
message triprec {
  ..
  required group tip_history (LIST) {
    repeated group array {
      required double amount;
      required binary currency (STRING);
    }
  }
}
Row group 0:  count: 12  269.67 B records  start: 4  total(compressed): 3.160 kB total(uncompressed):2.990 kB 
--------------------------------------------------------------------------------
                               type      encodings count     avg size   nulls   min / max
tip_history.array.amount       DOUBLE    G   _     12        12.83 B    0       "9.152136419558687" / "98.82597551791528"
tip_history.array.currency     BINARY    G _ R     12        8.33 B     0       "USD" / "USD" {code}
 

When looking into the parquet reader ("org.apache.parquet.hadoop.InternalParquetRecordReader" used under the hood for reading the records), we see that whenever the list-typed field data is lost, the configuration somehow contains "parquet.avro.write-old-list-structure=false" causing AvroSchemaConverter to set "writeOldListStructure" to false (default value is true if the config is not set), which controls how the requestedSchema is generated (requestedSchema is used to read the data).

 
{code:java}
InternalParquetRecordReader:

ReadSupport.ReadContext readContext = this.readSupport.init(new InitContext(conf, toSetMultiMap(fileMetadata), this.fileSchema));
this.columnIOFactory = new ColumnIOFactory(parquetFileMetadata.getCreatedBy());
this.requestedSchema = readContext.getRequestedSchema();
this.columnCount = this.requestedSchema.getPaths().size(); {code}
 
{code:java}
public AvroSchemaConverter(Configuration conf) {
this.assumeRepeatedIsListElement = conf.getBoolean("parquet.avro.add-list-element-records", true);
this.writeOldListStructure = conf.getBoolean("parquet.avro.write-old-list-structure", true);
this.writeParquetUUID = conf.getBoolean("parquet.avro.write-parquet-uuid", false);
this.readInt96AsFixed = conf.getBoolean("parquet.avro.readInt96AsFixed", false);
}

In private org.apache.parquet.schema.Type convertField(String fieldName, Schema schema, org.apache.parquet.schema.Type.Repetition repetition):

if (type.equals(Type.ARRAY)) {
if (this.writeOldListStructure) {
return ConversionPatterns.listType(repetition, fieldName, this.convertField("array", schema.getElementType(), Repetition.REPEATED));
}

return ConversionPatterns.listOfElements(repetition, fieldName, this.convertField("element", schema.getElementType()));
} {code}
 

When "writeOldListStructure" is set to false, the new list structure with 3 levels is used, which is different from the file schema which uses the 2-level old list structure in the parquet schema.
{code:java}
  required group tip_history (LIST) {
    repeated group list {
      required group element {
        required double amount;
        required binary currency (STRING);
      }
    }
  } {code}
 

This means that in other places "writeOldListStructure" is set to true by default, writing old list structure to the parquet files. We need to make this consistent to avoid such data loss on read.

Explicitly setting "parquet.avro.write-old-list-structure=false" in the Hadoop config for all Hudi Spark jobs can get around this issue.

## JIRA info

- Link: https://issues.apache.org/jira/browse/HUDI-8299
- Type: Sub-task
- Parent: https://issues.apache.org/jira/browse/HUDI-9113
- Affects version(s):
- 0.14.1
- Fix version(s):
- 1.1.0
- Attachment(s):
- 03/Oct/24 21:43;yihua;Screenshot 2024-10-03 at 14.31.41.png;https://issues.apache.org/jira/secure/attachment/13071949/Screenshot+2024-10-03+at+14.31.41.png