voonhous opened a new issue, #20503:
URL: https://github.com/apache/hudi/issues/20503
### Describe the problem
`bulk_insert` (row writer, the default) builds the Parquet schema from the
DataFrame's `StructType`, so a `ShortType` column is written as `INT32 +
INT(16)`, while the table schema records it as `int` (`HoodieSchema` has no
16-bit type, #18974). Every other write path (`insert`, `upsert`, compaction,
clustering) writes plain `INT32`. On master (b0d16f9, Spark 3.5) this gives
three failures:
1. **Reads fail.** Reading the column with Spark's default vectorized
Parquet reader throws a `NullPointerException`, with or without data skipping.
With `spark.sql.parquet.enableVectorizedReader=false` the values read back
correctly, and a table written with `insert` reads correctly. This is the
symptom of #9845 (HUDI-6936), which was closed without a fix.
2. **Column stats are silently skipped.** `HoodieTableMetadataUtil` catches
the failure and logs only `ERROR HoodieTableMetadataUtil: Failed to fetch
column range metadata for: <file>.parquet`, so those files have no column
stats. The cause, from calling the same
`ParquetUtils.readColumnStatsFromMetadata` on one of the files:
`LogicalTypeParquetAdapter` does not handle `INT(8)` / `INT(16)`.
3. **`ByteType` cannot be written** through `bulk_insert`:
`HoodieRowParquetWriteSupport` has no writer for it.
### To Reproduce
Spark 3.5, local mode:
```scala
import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
val schema = StructType(Seq(
StructField("id", IntegerType, false),
StructField("small", ShortType, true),
StructField("ts", LongType, false)))
val df = spark.createDataFrame(spark.sparkContext.parallelize(Seq(Row(1,
(-5).toShort, 1L), Row(2, 300.toShort, 1L))), schema)
df.write.format("hudi")
.option("hoodie.table.name", "t")
.option("hoodie.datasource.write.recordkey.field", "id")
.option("hoodie.datasource.write.precombine.field", "ts")
.option("hoodie.datasource.write.operation", "bulk_insert")
.option("hoodie.metadata.index.column.stats.enable", "true")
.mode("overwrite").save(path) // logs
"Failed to fetch column range metadata"
spark.read.format("hudi").load(path).select("id", "small").collect() //
NullPointerException
```
With `ByteType` instead of `ShortType`, the write itself fails.
### Expected behavior
The column reads back as written, column stats are recorded, and `ByteType`
can be written, on every write path.
### Stack traces
1. Read (executor):
```
java.lang.NullPointerException: Cannot load from int array because
"this.intData" is null
at
org.apache.spark.sql.execution.vectorized.OnHeapColumnVector.getInt(OnHeapColumnVector.java:328)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown
Source)
at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:43)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
at
org.apache.spark.util.random.SamplingUtils$.reservoirSampleAndCount(SamplingUtils.scala:41)
at
org.apache.spark.RangePartitioner$.$anonfun$sketch$1(Partitioner.scala:322)
at
org.apache.spark.RangePartitioner$.$anonfun$sketch$1$adapted(Partitioner.scala:320)
at
org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2(RDD.scala:910)
at
org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2$adapted(RDD.scala:910)
at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:331)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:93)
at
org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166)
at org.apache.spark.scheduler.Task.run(Task.scala:141)
at
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:621)
at
org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
at
org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:624)
at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
```
2. Column stats (`ParquetUtils.readColumnStatsFromMetadata` on one written
file):
```
java.lang.IllegalArgumentException: Unsupported logical type annotation:
INTEGER(16,true)
at
org.apache.hudi.metadata.stats.LogicalTypeParquetAdapter.getValueTypeFromAnnotation(LogicalTypeParquetAdapter.java:68)
at
org.apache.hudi.metadata.stats.ValueType.fromParquetPrimitiveType(ValueType.java:222)
at
org.apache.hudi.metadata.stats.ValueMetadata.getValueMetadata(ValueMetadata.java:259)
at
org.apache.hudi.common.util.ParquetUtils.lambda$readColumnStatsFromMetadata$3(ParquetUtils.java:403)
at
java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
at
java.base/java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:179)
at
java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625)
at
java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
at
java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
at
java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
at
java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
at
java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at
java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596)
at
java.base/java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:276)
at
java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625)
at
java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
at
java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
at
java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921)
at
java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at
java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682)
at
org.apache.hudi.common.util.ParquetUtils.mergeColumnStats(ParquetUtils.java:433)
at
org.apache.hudi.common.util.ParquetUtils.readColumnStatsFromMetadata(ParquetUtils.java:428)
at
org.apache.hudi.common.util.ParquetUtils.readColumnStatsFromMetadata(ParquetUtils.java:384)
```
3. `ByteType` through `bulk_insert`:
<details><summary>Full trace</summary>
```
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in
stage 13.0 failed 1 times, most recent failure: Lost task 0.0 in stage 13.0
(TID 13) (192.168.1.192 executor driver):
org.apache.hudi.exception.HoodieException: Unable to instantiate class
org.apache.hudi.io.storage.row.HoodieRowParquetWriteSupport
at
org.apache.hudi.common.util.ReflectionUtils.loadClass(ReflectionUtils.java:78)
at
org.apache.hudi.io.storage.row.HoodieRowParquetWriteSupport.getHoodieRowParquetWriteSupport(HoodieRowParquetWriteSupport.java:750)
at
org.apache.hudi.io.storage.row.HoodieInternalRowFileWriterFactory.createParquetInternalRowFileWriter(HoodieInternalRowFileWriterFactory.java:131)
at
org.apache.hudi.io.storage.row.HoodieInternalRowFileWriterFactory.newParquetInternalRowFileWriter(HoodieInternalRowFileWriterFactory.java:110)
at
org.apache.hudi.io.storage.row.HoodieInternalRowFileWriterFactory.getInternalRowFileWriter(HoodieInternalRowFileWriterFactory.java:72)
at
org.apache.hudi.io.storage.row.HoodieRowCreateHandle.<init>(HoodieRowCreateHandle.java:147)
at
org.apache.hudi.table.action.commit.BulkInsertDataInternalWriterHelper.createHandle(BulkInsertDataInternalWriterHelper.java:241)
at
org.apache.hudi.table.action.commit.BulkInsertDataInternalWriterHelper.getRowCreateHandle(BulkInsertDataInternalWriterHelper.java:227)
at
org.apache.hudi.table.action.commit.BulkInsertDataInternalWriterHelper.write(BulkInsertDataInternalWriterHelper.java:161)
at
org.apache.hudi.HoodieDatasetBulkInsertHelper$.$anonfun$bulkInsert$2(HoodieDatasetBulkInsertHelper.scala:214)
at
org.apache.hudi.HoodieDatasetBulkInsertHelper$.$anonfun$bulkInsert$2$adapted(HoodieDatasetBulkInsertHelper.scala:214)
at scala.collection.Iterator.foreach(Iterator.scala:943)
at scala.collection.Iterator.foreach$(Iterator.scala:943)
at
org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.foreach(WholeStageCodegenEvaluatorFactory.scala:41)
at
org.apache.hudi.HoodieDatasetBulkInsertHelper$.$anonfun$bulkInsert$1(HoodieDatasetBulkInsertHelper.scala:214)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:858)
at
org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:858)
at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:331)
at
org.apache.spark.sql.execution.SQLConfInjectingRDD.compute(SQLConfInjectingRDD.scala:58)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367)
at org.apache.spark.rdd.RDD.$anonfun$getOrCompute$1(RDD.scala:381)
at
org.apache.spark.storage.BlockManager.$anonfun$getOrElseUpdate$1(BlockManager.scala:1372)
at
org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1614)
at
org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1524)
at
org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1588)
at
org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1389)
at
org.apache.spark.storage.BlockManager.getOrElseUpdateRDDBlock(BlockManager.scala:1343)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:379)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:329)
at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:331)
at
org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:104)
at
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54)
at
org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166)
at org.apache.spark.scheduler.Task.run(Task.scala:141)
at
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:621)
at
org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
at
org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:624)
at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: java.lang.reflect.InvocationTargetException
at
java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native
Method)
at
java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:77)
at
java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at
java.base/java.lang.reflect.Constructor.newInstanceWithCaller(Constructor.java:500)
at
java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:481)
at
org.apache.hudi.common.util.ReflectionUtils.loadClass(ReflectionUtils.java:76)
... 46 more
Caused by: java.lang.UnsupportedOperationException: Unsupported type:
ByteType
at
org.apache.hudi.io.storage.row.HoodieRowParquetWriteSupport.makeWriter(HoodieRowParquetWriteSupport.java:682)
at
org.apache.hudi.io.storage.row.HoodieRowParquetWriteSupport.getFieldWriters(HoodieRowParquetWriteSupport.java:364)
at
org.apache.hudi.io.storage.row.HoodieRowParquetWriteSupport.<init>(HoodieRowParquetWriteSupport.java:180)
... 52 more
```
</details>
Related: #18974 (root cause), #9845.
--
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]
