[SUPPORT] Schema evolution fails due to NPE with `hoodie.schema.on.read.enable` turned on

[lufzhangzitao]

Tips before filing an issue

• Have you gone through our FAQs?

• Join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org.

• If you have triaged this as a bug, then file an issue directly.

Describe the problem you faced

first, If I use spark ddl to change the field data type or other ddl, and then I use spark df to add a new filed, then will throw a NullPointerException

Environment Description

• Hudi version :0.11.1

• Spark version :3.2.2

• Hive version :3.1.3

• Hadoop version :3.3.2

• Storage (HDFS/S3/GCS..) :hdfs

Additional context

Stacktrace
14608 [Executor task launch worker for task 0.0 in stage 25.0 (TID 32)] ERROR org.apache.spark.executor.Executor - Exception in task 0.0 in stage 25.0 (TID 32)
org.apache.hudi.exception.HoodieUpsertException: Error upserting bucketType UPDATE for partition :0
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:329)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.lambda$mapPartitionsAsRDD$a3ab3c4$1(BaseSparkCommitActionExecutor.java:244)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1(JavaRDDLike.scala:102)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1$adapted(JavaRDDLike.scala:102)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2(RDD.scala:915)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2$adapted(RDD.scala:915)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.$anonfun$getOrCompute$1(RDD.scala:386)
at org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1498)
at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1408)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1472)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1295)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:384)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:335)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:131)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1462)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.NullPointerException
at org.apache.hudi.internal.schema.visitor.NameToIDVisitor.beforeField(NameToIDVisitor.java:42)
at org.apache.hudi.internal.schema.InternalSchemaBuilder.visit(InternalSchemaBuilder.java:87)
at org.apache.hudi.internal.schema.InternalSchemaBuilder.buildNameToId(InternalSchemaBuilder.java:68)
at org.apache.hudi.internal.schema.InternalSchemaBuilder.buildIdToName(InternalSchemaBuilder.java:56)
at org.apache.hudi.internal.schema.InternalSchema.(InternalSchema.java:72)
at org.apache.hudi.internal.schema.InternalSchema.(InternalSchema.java:62)
at org.apache.hudi.internal.schema.utils.AvroSchemaEvolutionUtils.evolveSchemaFromNewAvroSchema(AvroSchemaEvolutionUtils.java:95)
at org.apache.hudi.table.action.commit.HoodieMergeHelper.runMerge(HoodieMergeHelper.java:103)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdateInternal(BaseSparkCommitActionExecutor.java:358)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdate(BaseSparkCommitActionExecutor.java:349)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:322)
... 28 more
14608 [Executor task launch worker for task 1.0 in stage 25.0 (TID 33)] ERROR org.apache.spark.executor.Executor - Exception in task 1.0 in stage 25.0 (TID 33)
org.apache.hudi.exception.HoodieUpsertException: Error upserting bucketType UPDATE for partition :1
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:329)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.lambda$mapPartitionsAsRDD$a3ab3c4$1(BaseSparkCommitActionExecutor.java:244)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1(JavaRDDLike.scala:102)
at org.apache.spark.api.java.JavaRDDLike.$anonfun$mapPartitionsWithIndex$1$adapted(JavaRDDLike.scala:102)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2(RDD.scala:915)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndex$2$adapted(RDD.scala:915)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.$anonfun$getOrCompute$1(RDD.scala:386)
at org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1498)
at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1408)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1472)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1295)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:384)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:335)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:131)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1462)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Caused by: java.lang.NullPointerException
at org.apache.hudi.internal.schema.visitor.NameToIDVisitor.beforeField(NameToIDVisitor.java:42)
at org.apache.hudi.internal.schema.InternalSchemaBuilder.visit(InternalSchemaBuilder.java:87)
at org.apache.hudi.internal.schema.InternalSchemaBuilder.buildNameToId(InternalSchemaBuilder.java:68)
at org.apache.hudi.internal.schema.InternalSchemaBuilder.buildIdToName(InternalSchemaBuilder.java:56)
at org.apache.hudi.internal.schema.InternalSchema.(InternalSchema.java:72)
at org.apache.hudi.internal.schema.InternalSchema.(InternalSchema.java:62)
at org.apache.hudi.internal.schema.utils.AvroSchemaEvolutionUtils.evolveSchemaFromNewAvroSchema(AvroSchemaEvolutionUtils.java:95)
at org.apache.hudi.table.action.commit.HoodieMergeHelper.runMerge(HoodieMergeHelper.java:103)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdateInternal(BaseSparkCommitActionExecutor.java:358)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpdate(BaseSparkCommitActionExecutor.java:349)
at org.apache.hudi.table.action.commit.BaseSparkCommitActionExecutor.handleUpsertPartition(BaseSparkCommitActionExecutor.java:322)
... 28 more

[codope]

@lufzhangzitao To work around this, you can disable schema on read. Adding a field and compatible data type change is supported by Hudi.

@xiarixiaoyao Can you please take a look at the schema node visitor logic? This NPE looks like a bug in the schema on read feature.

[xiarixiaoyao]

@lufzhangzitao
could you pls try hudi 0.12.
in hudi 0.11 schema evolution cannot work with hoodie.datasource.write.reconcile.schema
@lufzhangzitao @codope
#6017 already fix this bug. i have test it with master branch.

[lufzhangzitao]

@xiarixiaoyao
thx I tried hudi 0.12.1 schema evolution can work with hoodie.datasource.write.reconcile.schema ，
But it seems the hudi document example cannot work , I try the example add a new string field and change the datatype of a field from int to long in hudi 0.12.1, will throw exception :
org.apache.avro.AvroRuntimeException: cannot support rewrite value for schema type: "int" since the old schema type is: "long"
at org.apache.hudi.avro.HoodieAvroUtils.rewritePrimaryTypeWithDiffSchemaType(HoodieAvroUtils.java:955)
at org.apache.hudi.avro.HoodieAvroUtils.rewritePrimaryType(HoodieAvroUtils.java:869)
at org.apache.hudi.avro.HoodieAvroUtils.rewriteRecordWithNewSchema(HoodieAvroUtils.java:820)
at org.apache.hudi.avro.HoodieAvroUtils.rewriteRecordWithNewSchema(HoodieAvroUtils.java:818)
at org.apache.hudi.avro.HoodieAvroUtils.rewriteRecordWithNewSchema(HoodieAvroUtils.java:772)
at org.apache.hudi.avro.HoodieAvroUtils.rewriteRecordWithNewSchema(HoodieAvroUtils.java:734)
at org.apache.hudi.avro.HoodieAvroUtils.rewriteRecordDeep(HoodieAvroUtils.java:1034)
at org.apache.hudi.HoodieSparkUtils$.$anonfun$createRdd$4(HoodieSparkUtils.scala:109)
at org.apache.hudi.HoodieSparkUtils$.$anonfun$createRdd$5(HoodieSparkUtils.scala:117)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:199)
at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:63)
at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)
at org.apache.spark.scheduler.Task.run(Task.scala:131)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1491)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

[xiarixiaoyao]

@lufzhangzitao
could you share you test code.
int to long is covered by UT, it should worked,
spark.sql(s"alter table ${tableName} alter column userx.age type long")

[lufzhangzitao]

@xiarixiaoyao

I use spark df add a new string field and change the datatype of a field

val tableName: String = "hudi_cow_test"
val tablePath: String = s"D:/data/data/$tableName"

val schema = StructType(Array(
  StructField("rowId", StringType, true),
  StructField("partitionId", StringType, true),
  StructField("preComb", LongType, true),
  StructField("name", StringType, true),
  StructField("versionId", StringType, true),
  StructField("intToLong", IntegerType, true)
))

val data1 = Seq(Row("row_1", "part_0", 0L, "bob", "v_0", 0),
  Row("row_2", "part_0", 0L, "john", "v_0", 0),
  Row("row_3", "part_0", 0L, "tom", "v_0", 0))


var dfFromData1 = spark.createDataFrame(data1, schema)
dfFromData1.write
  .mode(SaveMode.Append)
  .format("hudi")
  .option("hoodie.datasource.write.operation", "upsert")
  .option("hoodie.datasource.write.reconcile.schema", value = true)
  .option("hoodie.schema.on.read.enable", value = true)
  .option(PRECOMBINE_FIELD.key(), "preComb")
  .option(RECORDKEY_FIELD.key(), "rowId")
  .option(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(), "partitionId")
  .option(INDEX_TYPE.key(), HoodieIndex.IndexType.BLOOM.name())
  .option(TBL_NAME.key(), tableName)
  .save(tablePath)

spark.read
  .format("org.apache.hudi").load(tablePath).show()

// In the new schema, we are going to add a String field and
// change the datatype `intToLong` field from  int to long.
val newSchema = StructType(Array(
  StructField("rowId", StringType, true),
  StructField("partitionId", StringType, true),
  StructField("preComb", LongType, true),
  StructField("name", StringType, true),
  StructField("versionId", StringType, true),
  StructField("intToLong", LongType, true),
  StructField("newField", StringType, true)
))

val data2 = Seq(
  Row("row_2", "part_0", 5L, "john", "v_1", 1L, "newField_1"),
  Row("row_5", "part_0", 5L, "maroon", "v_1", 5L, "newField_1"),
  Row("row_9", "part_0", 5L, "michael", "v_1", 3L, "newField_1"))

var dfFromData2 = spark.createDataFrame(data2, newSchema)
dfFromData2.write
  .mode(SaveMode.Append)
  .format("hudi")
  .option("hoodie.datasource.write.operation", "upsert")
  .option("hoodie.datasource.write.reconcile.schema", value = true)
  .option("hoodie.schema.on.read.enable", value = true)
  .option(PRECOMBINE_FIELD.key(), "preComb")
  .option(RECORDKEY_FIELD.key(), "rowId")
  .option(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(), "partitionId")
  .option(INDEX_TYPE.key(), HoodieIndex.IndexType.BLOOM.name())
  .option(TBL_NAME.key(), tableName)
  .save(tablePath)

spark.read
  .format("org.apache.hudi").load(tablePath).show()

spark.stop()

[xiarixiaoyao]

@lufzhangzitao
now reconcile cannot support type change, could you pls raise a jira for this problem, i will fixed it as soon as possible.

 /**
   * Support reconcile from a new avroSchema.
   * 1) incoming data has missing columns that were already defined in the table –> null values will be injected into missing columns
   * 2) incoming data contains new columns not defined yet in the table -> columns will be added to the table schema (incoming dataframe?)
   * 3) incoming data has missing columns that are already defined in the table and new columns not yet defined in the table ->
   *     new columns will be added to the table schema, missing columns will be injected with null values
   * 4) support nested schema change.
   * Notice:
   *    the incoming schema should not have delete/rename semantics.
   *    for example: incoming schema:  int a, int b, int d;   oldTableSchema int a, int b, int c, int d
   *    we must guarantee the column c is missing semantic, instead of delete semantic.
   * @param incomingSchema implicitly evolution of avro when hoodie write operation
   * @param oldTableSchema old internalSchema
   * @return reconcile Schema
   */

you can also use sql not achieve your goals

[xiarixiaoyao]

@lufzhangzitao
close this issue, We already have PR to fix this issue