[HUDI-1748] Read operation will possibly fail on mor table rt view when a write operations is concurrency running

[li36909]

Tips

• Thank you very much for contributing to Apache Hudi.
• Please review https://hudi.apache.org/contributing.html before opening a pull request.

What is the purpose of the pull request

Solve read write concurrency bug on mor table rt view

Brief change log

(for example:)

• Modify AnnotationLocation checkstyle rule in checkstyle.xml

Verify this pull request

Testing concurrency probabilistic problems in UT is difficult. I add sleep stability to the getRealtimeSplits method to reproduce the problem and verify the fix.

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

• For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.

[li36909]

cc @nsivabalan could you help to take a look, thank you

[codecov-io]

Codecov Report

Merging #2751 (2de7140) into master (fe16d0d) will increase coverage by 17.69%.
The diff coverage is n/a.

@@              Coverage Diff              @@
##             master    #2751       +/-   ##
=============================================
+ Coverage     52.04%   69.73%   +17.69%     
+ Complexity     3625      371     -3254     
=============================================
  Files           479       54      -425     
  Lines         22804     1989    -20815     
  Branches       2415      236     -2179     
=============================================
- Hits          11868     1387    -10481     
+ Misses         9911      471     -9440     
+ Partials       1025      131      -894     

Flag	Coverage Δ	Complexity Δ
hudicli	?	?
hudiclient	?	?
hudicommon	?	?
hudiflink	?	?
hudihadoopmr	?	?
hudisparkdatasource	?	?
hudisync	?	?
huditimelineservice	?	?
hudiutilities	69.73% <ø> (-0.06%)	0.00 <ø> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ	Complexity Δ
...apache/hudi/utilities/deltastreamer/DeltaSync.java	71.37% <0.00%> (-0.35%)	55.00% <0.00%> (-1.00%)
...sioning/clean/CleanMetadataV2MigrationHandler.java
...org/apache/hudi/common/table/log/AppendResult.java
...oning/compaction/CompactionV2MigrationHandler.java
...on/table/timeline/versioning/MetadataMigrator.java
...e/hudi/common/table/timeline/dto/FileGroupDTO.java
...di/sink/partitioner/delta/DeltaBucketAssigner.java
...che/hudi/metadata/TimelineMergedTableMetadata.java
...pache/hudi/io/storage/HoodieFileReaderFactory.java
...oning/compaction/CompactionV1MigrationHandler.java
... and 415 more

[nsivabalan]

@li36909 : can you clarify what do you mean by concurrent operations. Ignoring any table services, you mean to say that just a batch of write and rt snapshot view happening concurrently is running into issues? which hudi version are you running.
@n3nash : you may want to check it out.

[li36909]

@nsivabalan I run the test at hudi 0.7. yes, you are right, I start a spark-shell for upserting, and query the same table by spark datasouce api, then the problem arises. The cause of the problem is clear, during the query, hudi get partitions at MergeOnReadSnapshotRelation, and build a new fsview at HoodieRealtimeInputFormatUtils.groupLogsByBaseFile, when a write operation is happening, HoodieRealtimeInputFormatUtils.groupLogsByBaseFile will find some new base files.
we can reproduce this issue by add a Thread.sleep(60000) at HoodieRealtimeInputFormatUtils.groupLogsByBaseFile, then run this test:
step 1: write first batch into a hudi table:
spark-shell

import org.apache.hudi.QuickstartUtils.
import scala.collection.JavaConversions.
import org.apache.spark.sql.SaveMode.
import org.apache.hudi.DataSourceReadOptions.
import org.apache.hudi.DataSourceWriteOptions.
import org.apache.hudi.config.HoodieWriteConfig.

val tableName = "hudi_mor_table"
val basePath = "hdfs://hacluster/tmp/hudi_mor_table"
val dataGen = new DataGenerator
val inserts = convertToStringList(dataGen.generateInserts(10))
val df = spark.read.json(spark.sparkContext.parallelize(inserts, 2))
df.write.format("org.apache.hudi").
option("hoodie.datasource.write.table.type", "MERGE_ON_READ").
option("hoodie.datasource.write.operation", "bulk_insert").
options(getQuickstartWriteConfigs).
option(PRECOMBINE_FIELD_OPT_KEY, "ts").
option(RECORDKEY_FIELD_OPT_KEY, "uuid").
option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
option("hoodie.datasource.write.hive_style_partitioning", "false").
option(TABLE_NAME, tableName).
option("hoodie.datasource.hive_sync.enable", "true").
option("hoodie.datasource.hive_sync.use_jdbc", "false").
option("hoodie.datasource.hive_sync.table", "hudi_mor_test").
option("hoodie.datasource.hive_sync.partition_extractor_class", "org.apache.hudi.hive.MultiPartKeysValueExtractor").
option("hoodie.datasource.hive_sync.partition_fields", "continent,country,city").
option("hoodie.datasource.hive_sync.assume_date_partitioning", "false").
option("hoodie.insert.shuffle.parallelism", "2").
option("hoodie.upsert.shuffle.parallelism","2").
option("hoodie.bulkinsert.shuffle.parallelism", "2").
option("hoodie.delete.shuffle.parallelism","2").
mode(Append).
save(basePath);

step 2: run a query at new spark-shell (when the query hang at Thread.sleep, start to write a new batch at step3)
spark.read.format("hudi").load("hdfs://hacluster/tmp/hudi_mor_table////").count

setp 3: go to the spark-shell at step1, write a new batch:
df.write.format("org.apache.hudi").
option("hoodie.datasource.write.table.type", "MERGE_ON_READ").
option("hoodie.datasource.write.operation", "bulk_insert").
options(getQuickstartWriteConfigs).
option(PRECOMBINE_FIELD_OPT_KEY, "ts").
option(RECORDKEY_FIELD_OPT_KEY, "uuid").
option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
option("hoodie.datasource.write.hive_style_partitioning", "false").
option(TABLE_NAME, tableName).
option("hoodie.datasource.hive_sync.enable", "true").
option("hoodie.datasource.hive_sync.use_jdbc", "false").
option("hoodie.datasource.hive_sync.table", "hudi_mor_test").
option("hoodie.datasource.hive_sync.partition_extractor_class", "org.apache.hudi.hive.MultiPartKeysValueExtractor").
option("hoodie.datasource.hive_sync.partition_fields", "continent,country,city").
option("hoodie.datasource.hive_sync.assume_date_partitioning", "false").
option("hoodie.insert.shuffle.parallelism", "2").
option("hoodie.upsert.shuffle.parallelism","2").
option("hoodie.bulkinsert.shuffle.parallelism", "2").
option("hoodie.delete.shuffle.parallelism","2").
mode(Append).
save(basePath);

we can see the step2 will throw a exception

[n3nash]

@li36909 If I understand this correctly, you are saying reading a MOR table _rt view can fail if a concurrent write is happening ?

[li36909]

@n3nash yes

[n3nash]

Okay, thanks for the confirmation, I will try to reproduce this issue on my end and get back.

[li36909]

cc @n3nash could you help to take a look again, thank you

[nsivabalan]

@n3nash : once you verify the issue, please move it to "ready to review" col in the PR board.

[hudi-bot]

CI report:

• 2de7140 Azure: FAILURE

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build

[nsivabalan]

@alexeykudinkin : this has been a long pending PR. Can you follow up on this please. Guess there are steps to reproduce. If its not a valid issue, we can close the patch and the jira.

[yihua]

@alexeykudinkin Is this already fixed by the new input format classes?

[alexeykudinkin]

@li36909 can you please rebase this on the latest master and validate whether your fix is still valid?

[nsivabalan]

@li36909 : gentle ping. feel free to close if its not an issue anymore.

[bvaradar]

The change would cause skipping file groups with no base parquet file (case when log file can be indexed). Also, File system view should be read only committed files. So, unless the finished commit is reverted ( explicitly which is not snapshot isolation.), this issue should not be seen.

[bvaradar]

Closing this PR due to reasons above.