Wenning Ding created HUDI-1489:
----------------------------------
Summary: Not able to read after updating bootstrap table with
written table
Key: HUDI-1489
URL: https://issues.apache.org/jira/browse/HUDI-1489
Project: Apache Hudi
Issue Type: Bug
Reporter: Wenning Ding
After updating Hudi table with the written bootstrap table, it would fail to
read the latest bootstrap table.
h3. Reproduction steps
{code:java}
import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.common.model.HoodieTableType
import org.apache.hudi.config.HoodieBootstrapConfig
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.SparkSession
val bucketName = "wenningd-emr-dev"
val tableName = "hudi_bootstrap_test_cow_5c1a5147_888e_4b638bef8"
val recordKeyName = "event_id"
val partitionKeyName = "event_type"
val precombineKeyName = "event_time"
val verificationRecordKey = "4"
val verificationColumn = "event_name"
val originalVerificationValue = "event_d"
val updatedVerificationValue = "event_test"
// val sourceTableWithoutHiveStylePartition =
"s3://wenningd-emr-dev/hudi/test-data/source_table/"
// new parameters
val sourceTableLocation =
"s3://wenningd-emr-dev/hudi/test-data/source_table/"
val tableType = HoodieTableType.COPY_ON_WRITE.name()
val verificationSqlQuery = "select " + verificationColumn + " from " +
tableName + " where " + recordKeyName + " = '" + verificationRecordKey + "'"
val tablePath = "s3://" + bucketName + "/hudi/tables/" + tableName
val loadTablePath = tablePath + "/*/*"
// Create table and sync with hive
val df = spark.emptyDataFrame
val tableType = HoodieTableType.COPY_ON_WRITE.name
// val tableType = HoodieTableType.MERGE_ON_READ.name
df.write
.format("hudi")
.option(HoodieWriteConfig.TABLE_NAME, tableName)
.option(HoodieBootstrapConfig.BOOTSTRAP_BASE_PATH_PROP,
sourceTableLocation)
.option(HoodieBootstrapConfig.BOOTSTRAP_KEYGEN_CLASS,
"org.apache.hudi.keygen.SimpleKeyGenerator")
.option(DataSourceWriteOptions.OPERATION_OPT_KEY,
DataSourceWriteOptions.BOOTSTRAP_OPERATION_OPT_VAL)
.option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, tableType)
.option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, recordKeyName)
.option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY, "true")
.option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY, tableName)
.option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY,
partitionKeyName)
.option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY,
"org.apache.hudi.hive.MultiPartKeysValueExtractor")
.mode(SaveMode.Overwrite)
.save(tablePath)
// Verify create with spark sql query
val result0 = spark.sql(verificationSqlQuery)
if (!(result0.count == 1) ||
!result0.collect.mkString.contains(originalVerificationValue)) {
throw new TestFailureException("Create table verification failed!")
}
val df3 = spark.read.format("org.apache.hudi").load(loadTablePath)
val df4 = df3.filter(col(recordKeyName) === verificationRecordKey)
val df5 = df4.withColumn(verificationColumn, lit(updatedVerificationValue))
df5.write.format("org.apache.hudi")
.option(DataSourceWriteOptions.STORAGE_TYPE_OPT_KEY, tableType)
.option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, recordKeyName)
.option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, partitionKeyName)
.option(HoodieWriteConfig.TABLE_NAME, tableName)
.option(DataSourceWriteOptions.OPERATION_OPT_KEY,
DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL)
.option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, precombineKeyName)
// .option(DataSourceWriteOptions.HIVE_STYLE_PARTITIONING_OPT_KEY, "true")
.option(DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY, "true")
.option(DataSourceWriteOptions.HIVE_TABLE_OPT_KEY, tableName)
.option(DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY,
partitionKeyName)
.option(DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY,
"org.apache.hudi.hive.MultiPartKeysValueExtractor")
.mode(SaveMode.Append)
.save(tablePath)
val result1 = spark.sql(verificationSqlQuery)
val df6 = spark.read.format("org.apache.hudi").load(loadTablePath)
df6.show
{code}
df6.show would return:
{code:java}
Driver stacktrace:
at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:2043)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2031)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:2030)
at
scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2030)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:967)
at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:967)
at scala.Option.foreach(Option.scala:257)
at
org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:967)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2264)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2213)
at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2202)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:778)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2061)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2082)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:2101)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:407)
at
org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:38)
at
org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collectFromPlan(Dataset.scala:3395)
at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2552)
at org.apache.spark.sql.Dataset$$anonfun$head$1.apply(Dataset.scala:2552)
at org.apache.spark.sql.Dataset$$anonfun$52.apply(Dataset.scala:3370)
at
org.apache.spark.sql.execution.SQLExecution$.org$apache$spark$sql$execution$SQLExecution$$executeQuery$1(SQLExecution.scala:83)
at
org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1$$anonfun$apply$1.apply(SQLExecution.scala:94)
at
org.apache.spark.sql.execution.QueryExecutionMetrics$.withMetrics(QueryExecutionMetrics.scala:141)
at
org.apache.spark.sql.execution.SQLExecution$.org$apache$spark$sql$execution$SQLExecution$$withMetrics(SQLExecution.scala:178)
at
org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:93)
at
org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:200)
at
org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:92)
at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3369)
at org.apache.spark.sql.Dataset.head(Dataset.scala:2552)
at org.apache.spark.sql.Dataset.take(Dataset.scala:2766)
at org.apache.spark.sql.Dataset.getRows(Dataset.scala:255)
at org.apache.spark.sql.Dataset.showString(Dataset.scala:292)
at org.apache.spark.sql.Dataset.show(Dataset.scala:753)
at org.apache.spark.sql.Dataset.show(Dataset.scala:712)
at org.apache.spark.sql.Dataset.show(Dataset.scala:721)
... 49 elided
Caused by: java.lang.NullPointerException
at
org.apache.spark.sql.execution.vectorized.WritableColumnVector.arrayData(WritableColumnVector.java:637)
at
org.apache.spark.sql.execution.vectorized.WritableColumnVector.getUTF8String(WritableColumnVector.java:378)
at
org.apache.spark.sql.execution.vectorized.MutableColumnarRow.getUTF8String(MutableColumnarRow.java:135)
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.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:585)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:297)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:289)
at
org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:858)
at
org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:858)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:310)
{code}
The root cause is:
the {{requiredColumns}} in the
[buildScan()|https://github.com/apache/hudi/blob/release-0.6.0/hudi-spark/src/main/scala/org/apache/hudi/HoodieBootstrapRelation.scala#L72]
does not follow the same order as the schema file.
For example, when I selected all the columns, I printed the {{requiredColumns}}:
{{20/10/13 22:57:59 WARN HoodieBootstrapRelation: wenningd = > required
columns: _hoodie_commit_time _hoodie_record_key _hoodie_partition_path
event_type event_id event_guests event_time _hoodie_commit_seqno
_hoodie_file_name event_name}}
You can see not all the metadata columns are in the front. So the problem here
is when we try to use
[regularReadFunction|https://github.com/apache/hudi/blob/release-0.6.0/hudi-spark/src/main/scala/org/apache/hudi/HoodieBootstrapRelation.scala#L127],
we use this as the schema: {{requiredSkeletonSchema.fields ++
requiredDataSchema.fields}}. But since the required columns do not follow the
same order as schema file, there's a schema mismatch between
[requiredSchema|https://github.com/apache/hudi/blob/release-0.6.0/hudi-spark/src/main/scala/org/apache/hudi/HoodieBootstrapRelation.scala#L132]
and {{requiredColumns}}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
