lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-4204] Fixing NPE with row writer path and with OCC (#5850)

Browse Source
This commit is contained in:
Sivabalan Narayanan
2022-07-21 18:57:34 -04:00
committed by GitHub
parent 50cdb867c7
commit 2e0dd29714
4 changed files with 17 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
View File

@@ -491,7 +491,7 @@ public abstract class BaseHoodieWriteClient<T extends HoodieRecordPayload, I, K,
* @param writeOperationType * @param writeOperationType
* @param metaClient * @param metaClient
*/ */
protected void preWrite(String instantTime, WriteOperationType writeOperationType, public void preWrite(String instantTime, WriteOperationType writeOperationType,
HoodieTableMetaClient metaClient) { HoodieTableMetaClient metaClient) {
setOperationType(writeOperationType); setOperationType(writeOperationType);
this.lastCompletedTxnAndMetadata = TransactionUtils.getLastCompletedTxnInstantAndMetadata(metaClient); this.lastCompletedTxnAndMetadata = TransactionUtils.getLastCompletedTxnInstantAndMetadata(metaClient);

2
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
View File

@@ -254,7 +254,7 @@ public class HoodieFlinkWriteClient<T extends HoodieRecordPayload> extends
} }
@Override @Override
protected void preWrite(String instantTime, WriteOperationType writeOperationType, HoodieTableMetaClient metaClient) { public void preWrite(String instantTime, WriteOperationType writeOperationType, HoodieTableMetaClient metaClient) {
setOperationType(writeOperationType); setOperationType(writeOperationType);
// Note: the code to read the commit metadata is not thread safe for JSON deserialization, // Note: the code to read the commit metadata is not thread safe for JSON deserialization,
// remove the table metadata sync // remove the table metadata sync

1
hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/internal/DataSourceInternalWriterHelper.java
View File

@@ -68,6 +68,7 @@ public class DataSourceInternalWriterHelper {
this.metaClient = HoodieTableMetaClient.builder().setConf(configuration).setBasePath(writeConfig.getBasePath()).build(); this.metaClient = HoodieTableMetaClient.builder().setConf(configuration).setBasePath(writeConfig.getBasePath()).build();
this.metaClient.validateTableProperties(writeConfig.getProps()); this.metaClient.validateTableProperties(writeConfig.getProps());
this.hoodieTable = HoodieSparkTable.create(writeConfig, new HoodieSparkEngineContext(new JavaSparkContext(sparkSession.sparkContext())), metaClient); this.hoodieTable = HoodieSparkTable.create(writeConfig, new HoodieSparkEngineContext(new JavaSparkContext(sparkSession.sparkContext())), metaClient);
writeClient.preWrite(instantTime, WriteOperationType.BULK_INSERT, metaClient);
} }
public boolean useCommitCoordinator() { public boolean useCommitCoordinator() {

16
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieSparkSqlWriter.scala
View File

@@ -167,14 +167,21 @@ class TestHoodieSparkSqlWriter {
* @param sortMode Bulk insert sort mode * @param sortMode Bulk insert sort mode
* @param populateMetaFields Flag for populating meta fields * @param populateMetaFields Flag for populating meta fields
*/ */
def testBulkInsertWithSortMode(sortMode: BulkInsertSortMode, populateMetaFields: Boolean = true): Unit = { def testBulkInsertWithSortMode(sortMode: BulkInsertSortMode, populateMetaFields: Boolean = true, enableOCCConfigs: Boolean = false): Unit = {
//create a new table //create a new table
val fooTableModifier = commonTableModifier.updated("hoodie.bulkinsert.shuffle.parallelism", "4") var fooTableModifier = commonTableModifier.updated("hoodie.bulkinsert.shuffle.parallelism", "4")
.updated(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL) .updated(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL)
.updated(DataSourceWriteOptions.ENABLE_ROW_WRITER.key, "true") .updated(DataSourceWriteOptions.ENABLE_ROW_WRITER.key, "true")
.updated(HoodieTableConfig.POPULATE_META_FIELDS.key(), String.valueOf(populateMetaFields)) .updated(HoodieTableConfig.POPULATE_META_FIELDS.key(), String.valueOf(populateMetaFields))
.updated(HoodieWriteConfig.BULK_INSERT_SORT_MODE.key(), sortMode.name()) .updated(HoodieWriteConfig.BULK_INSERT_SORT_MODE.key(), sortMode.name())
if (enableOCCConfigs) {
fooTableModifier = fooTableModifier
.updated("hoodie.write.concurrency.mode","optimistic_concurrency_control")
.updated("hoodie.cleaner.policy.failed.writes","LAZY")
.updated("hoodie.write.lock.provider","org.apache.hudi.client.transaction.lock.InProcessLockProvider")
}
// generate the inserts // generate the inserts
val schema = DataSourceTestUtils.getStructTypeExampleSchema val schema = DataSourceTestUtils.getStructTypeExampleSchema
val structType = AvroConversionUtils.convertAvroSchemaToStructType(schema) val structType = AvroConversionUtils.convertAvroSchemaToStructType(schema)
@@ -306,6 +313,11 @@ class TestHoodieSparkSqlWriter {
testBulkInsertWithSortMode(sortMode, populateMetaFields = true) testBulkInsertWithSortMode(sortMode, populateMetaFields = true)
} }
@Test
def testBulkInsertForSortModeWithOCC(): Unit = {
testBulkInsertWithSortMode(BulkInsertSortMode.GLOBAL_SORT, populateMetaFields = true, true)
}
/** /**
* Test case for Bulk insert with populating meta fields or * Test case for Bulk insert with populating meta fields or
* without populating meta fields. * without populating meta fields.