You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2023/02/05 02:17:46 UTC
[hudi] branch release-0.13.0 updated (4286d798237 -> 2b17dd58243)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a change to branch release-0.13.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
from 4286d798237 [HUDI-5496] Avoid unnecessary file system parsing to initialize metadata table for a new data table (#7841)
new 003d6e79999 [HUDI-5689] Make CDC config working for DeltaStreamer path (#7830)
new 989d5403666 [MINOR] Validate configs for OCC early conflict detection (#7848)
new 2d2d751f553 Make most of the Spark SQL DML operations configs overridable (#7850)
new eb6bceefa5e [HUDI-5700] Annotate config classes for docs generation (#7853)
new f5cfec43270 [HUDI-5701] Remove meta fields from cdc new record in CDCLogger (#7852)
new 2b17dd58243 [HUDI-5702] Avoid to write useless change-data when compaction (#7849)
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/hudi/config/HoodieWriteConfig.java | 15 +++--
.../java/org/apache/hudi/io/HoodieCDCLogger.java | 2 +-
.../apache/hudi/io/HoodieMergeHandleFactory.java | 18 ++----
.../apache/hudi/common/config/ConfigGroups.java | 1 +
.../hudi/gcp/bigquery/BigQuerySyncConfig.java | 8 +++
.../spark/sql/hudi/ProvidesHoodieConfig.scala | 69 +++++++++++-----------
.../apache/spark/sql/hudi/TestCDCForSparkSQL.scala | 20 +++++--
.../sync/datahub/config/DataHubSyncConfig.java | 24 +++++---
.../java/org/apache/hudi/hive/HiveSyncConfig.java | 8 +++
.../hive/replication/GlobalHiveSyncConfig.java | 8 +++
.../apache/hudi/sync/common/HoodieSyncConfig.java | 8 +++
.../hudi/utilities/deltastreamer/DeltaSync.java | 4 ++
12 files changed, 118 insertions(+), 67 deletions(-)
[hudi] 01/06: [HUDI-5689] Make CDC config working for DeltaStreamer path (#7830)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch release-0.13.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 003d6e79999de9f2d700d3f76d7eaddabddfd14f
Author: Shiyan Xu <27...@users.noreply.github.com>
AuthorDate: Sat Feb 4 01:10:38 2023 -0600
[HUDI-5689] Make CDC config working for DeltaStreamer path (#7830)
---
.../main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
index 77079fe0f4f..8c010e9484d 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
@@ -342,6 +342,10 @@ public class DeltaSync implements Serializable, Closeable {
.setPreCombineField(cfg.sourceOrderingField)
.setPartitionMetafileUseBaseFormat(props.getBoolean(HoodieTableConfig.PARTITION_METAFILE_USE_BASE_FORMAT.key(),
HoodieTableConfig.PARTITION_METAFILE_USE_BASE_FORMAT.defaultValue()))
+ .setCDCEnabled(props.getBoolean(HoodieTableConfig.CDC_ENABLED.key(),
+ HoodieTableConfig.CDC_ENABLED.defaultValue()))
+ .setCDCSupplementalLoggingMode(props.getString(HoodieTableConfig.CDC_SUPPLEMENTAL_LOGGING_MODE.key(),
+ HoodieTableConfig.CDC_SUPPLEMENTAL_LOGGING_MODE.defaultValue()))
.setShouldDropPartitionColumns(isDropPartitionColumns())
.initTable(new Configuration(jssc.hadoopConfiguration()),
cfg.targetBasePath);
[hudi] 05/06: [HUDI-5701] Remove meta fields from cdc new record in CDCLogger (#7852)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch release-0.13.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit f5cfec43270a98d9a10512e9a651fb35d4e63e76
Author: Yann Byron <bi...@gmail.com>
AuthorDate: Sun Feb 5 08:05:13 2023 +0800
[HUDI-5701] Remove meta fields from cdc new record in CDCLogger (#7852)
---
.../java/org/apache/hudi/io/HoodieCDCLogger.java | 2 +-
.../apache/spark/sql/hudi/TestCDCForSparkSQL.scala | 20 ++++++++++++++------
2 files changed, 15 insertions(+), 7 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCDCLogger.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCDCLogger.java
index fd2dc60b58b..096bf475667 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCDCLogger.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCDCLogger.java
@@ -245,7 +245,7 @@ public class HoodieCDCLogger implements Closeable {
private CDCTransformer getTransformer() {
if (cdcSupplementalLoggingMode == data_before_after) {
return (operation, recordKey, oldRecord, newRecord) ->
- HoodieCDCUtils.cdcRecord(cdcSchema, operation.getValue(), commitTime, removeCommitMetadata(oldRecord), newRecord);
+ HoodieCDCUtils.cdcRecord(cdcSchema, operation.getValue(), commitTime, removeCommitMetadata(oldRecord), removeCommitMetadata(newRecord));
} else if (cdcSupplementalLoggingMode == data_before) {
return (operation, recordKey, oldRecord, newRecord) ->
HoodieCDCUtils.cdcRecord(cdcSchema, operation.getValue(), recordKey, removeCommitMetadata(oldRecord));
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCDCForSparkSQL.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCDCForSparkSQL.scala
index bec2230e5ab..a27da08866a 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCDCForSparkSQL.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestCDCForSparkSQL.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.hudi
import org.apache.hudi.DataSourceReadOptions._
import org.apache.hudi.common.table.HoodieTableMetaClient
-import org.apache.hudi.common.table.cdc.HoodieCDCSupplementalLoggingMode.{data_before, op_key_only}
+import org.apache.hudi.common.table.cdc.HoodieCDCSupplementalLoggingMode.{data_before, data_before_after, op_key_only}
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._
import org.junit.jupiter.api.Assertions.assertEquals
@@ -53,10 +53,15 @@ class TestCDCForSparkSQL extends HoodieSparkSqlTestBase {
spark.sql(s"use $databaseName")
Seq("cow", "mor").foreach { tableType =>
- Seq(op_key_only, data_before).foreach { loggingMode =>
+ Seq(op_key_only, data_before, data_before_after).foreach { loggingMode =>
withTempDir { tmp =>
val tableName = generateTableName
val basePath = s"${tmp.getCanonicalPath}/$tableName"
+ val otherTableProperties = if (tableType == "mor") {
+ "'hoodie.compact.inline'='true', 'hoodie.compact.inline.max.delta.commits'='2',"
+ } else {
+ ""
+ }
spark.sql(
s"""
| create table $tableName (
@@ -70,6 +75,7 @@ class TestCDCForSparkSQL extends HoodieSparkSqlTestBase {
| 'preCombineField' = 'ts',
| 'hoodie.table.cdc.enabled' = 'true',
| 'hoodie.table.cdc.supplemental.logging.mode' = '${loggingMode.name()}',
+ | $otherTableProperties
| type = '$tableType'
| )
| location '$basePath'
@@ -88,7 +94,9 @@ class TestCDCForSparkSQL extends HoodieSparkSqlTestBase {
spark.sql(s"insert into $tableName values (1, 'a1_v2', 11, 1100)")
val commitTime2 = metaClient.reloadActiveTimeline.lastInstant().get().getTimestamp
- val cdcDataOnly2 = cdcDataFrame(basePath, commitTime2.toLong - 1)
+ // here we use `commitTime1` to query the change data in commit 2.
+ // because `commitTime2` is maybe the ts of the compaction operation, not the write operation.
+ val cdcDataOnly2 = cdcDataFrame(basePath, commitTime1.toLong)
cdcDataOnly2.show(false)
assertCDCOpCnt(cdcDataOnly2, 0, 1, 0)
@@ -110,13 +118,13 @@ class TestCDCForSparkSQL extends HoodieSparkSqlTestBase {
spark.sql(s"update $tableName set name = 'a2_v2', ts = 1200 where id = 2")
val commitTime3 = metaClient.reloadActiveTimeline.lastInstant().get().getTimestamp
- val cdcDataOnly3 = cdcDataFrame(basePath, commitTime3.toLong - 1)
+ val cdcDataOnly3 = cdcDataFrame(basePath, commitTime2.toLong)
cdcDataOnly3.show(false)
assertCDCOpCnt(cdcDataOnly3, 0, 1, 0)
spark.sql(s"delete from $tableName where id = 3")
val commitTime4 = metaClient.reloadActiveTimeline.lastInstant().get().getTimestamp
- val cdcDataOnly4 = cdcDataFrame(basePath, commitTime4.toLong - 1)
+ val cdcDataOnly4 = cdcDataFrame(basePath, commitTime3.toLong)
cdcDataOnly4.show(false)
assertCDCOpCnt(cdcDataOnly4, 0, 0, 1)
@@ -135,7 +143,7 @@ class TestCDCForSparkSQL extends HoodieSparkSqlTestBase {
| when not matched then insert *
""".stripMargin)
val commitTime5 = metaClient.reloadActiveTimeline.lastInstant().get().getTimestamp
- val cdcDataOnly5 = cdcDataFrame(basePath, commitTime5.toLong - 1)
+ val cdcDataOnly5 = cdcDataFrame(basePath, commitTime4.toLong)
cdcDataOnly5.show(false)
assertCDCOpCnt(cdcDataOnly5, 1, 1, 0)
[hudi] 03/06: Make most of the Spark SQL DML operations configs overridable (#7850)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch release-0.13.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 2d2d751f55333187316eaa74095b5584cb63dc98
Author: Alexey Kudinkin <al...@gmail.com>
AuthorDate: Sat Feb 4 10:54:17 2023 -0800
Make most of the Spark SQL DML operations configs overridable (#7850)
This PR makes most of the Spark SQL operations configs are overridable, leaving only a few ones as those that should not and couldn't be overridden by the user
---
.../spark/sql/hudi/ProvidesHoodieConfig.scala | 69 +++++++++++-----------
1 file changed, 36 insertions(+), 33 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala
index 0c766f5135b..ffc4079824f 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala
@@ -61,20 +61,10 @@ trait ProvidesHoodieConfig extends Logging {
val hiveSyncConfig = buildHiveSyncConfig(sparkSession, hoodieCatalogTable, tableConfig)
val defaultOpts = Map[String, String](
+ OPERATION.key -> UPSERT_OPERATION_OPT_VAL,
KEYGENERATOR_CLASS_NAME.key -> classOf[SqlKeyGenerator].getCanonicalName,
- OPERATION.key -> UPSERT_OPERATION_OPT_VAL
- )
-
- val overridingOpts = Map[String, String](
- "path" -> hoodieCatalogTable.tableLocation,
- RECORDKEY_FIELD.key -> hoodieCatalogTable.primaryKeys.mkString(","),
- TBL_NAME.key -> hoodieCatalogTable.tableName,
- PRECOMBINE_FIELD.key -> preCombineField,
- HIVE_STYLE_PARTITIONING.key -> tableConfig.getHiveStylePartitioningEnable,
- URL_ENCODE_PARTITIONING.key -> tableConfig.getUrlEncodePartitioning,
SqlKeyGenerator.ORIGINAL_KEYGEN_CLASS_NAME -> tableConfig.getKeyGeneratorClassName,
SqlKeyGenerator.PARTITION_SCHEMA -> hoodieCatalogTable.partitionSchema.toDDL,
- PARTITIONPATH_FIELD.key -> tableConfig.getPartitionFieldProp,
HoodieSyncConfig.META_SYNC_ENABLED.key -> hiveSyncConfig.getString(HoodieSyncConfig.META_SYNC_ENABLED.key),
HiveSyncConfigHolder.HIVE_SYNC_ENABLED.key -> hiveSyncConfig.getString(HiveSyncConfigHolder.HIVE_SYNC_ENABLED.key),
HiveSyncConfigHolder.HIVE_SYNC_MODE.key -> hiveSyncConfig.getStringOrDefault(HiveSyncConfigHolder.HIVE_SYNC_MODE, HiveSyncMode.HMS.name()),
@@ -85,6 +75,16 @@ trait ProvidesHoodieConfig extends Logging {
HiveSyncConfigHolder.HIVE_SUPPORT_TIMESTAMP_TYPE.key -> hiveSyncConfig.getBoolean(HiveSyncConfigHolder.HIVE_SUPPORT_TIMESTAMP_TYPE).toString
)
+ val overridingOpts = Map[String, String](
+ "path" -> hoodieCatalogTable.tableLocation,
+ RECORDKEY_FIELD.key -> hoodieCatalogTable.primaryKeys.mkString(","),
+ TBL_NAME.key -> hoodieCatalogTable.tableName,
+ PRECOMBINE_FIELD.key -> preCombineField,
+ HIVE_STYLE_PARTITIONING.key -> tableConfig.getHiveStylePartitioningEnable,
+ URL_ENCODE_PARTITIONING.key -> tableConfig.getUrlEncodePartitioning,
+ PARTITIONPATH_FIELD.key -> tableConfig.getPartitionFieldProp
+ )
+
combineOptions(hoodieCatalogTable, tableConfig, sparkSession.sqlContext.conf,
defaultOpts = defaultOpts, overridingOpts = overridingOpts)
}
@@ -183,22 +183,10 @@ trait ProvidesHoodieConfig extends Logging {
PAYLOAD_CLASS_NAME.key -> payloadClassName,
// NOTE: By default insert would try to do deduplication in case that pre-combine column is specified
// for the table
- HoodieWriteConfig.COMBINE_BEFORE_INSERT.key -> String.valueOf(hasPrecombineColumn)
- )
-
- val overridingOpts = extraOptions ++ Map(
- "path" -> path,
- TABLE_TYPE.key -> tableType,
- TBL_NAME.key -> hoodieCatalogTable.tableName,
- OPERATION.key -> operation,
- HIVE_STYLE_PARTITIONING.key -> hiveStylePartitioningEnable,
- URL_ENCODE_PARTITIONING.key -> urlEncodePartitioning,
+ HoodieWriteConfig.COMBINE_BEFORE_INSERT.key -> String.valueOf(hasPrecombineColumn),
KEYGENERATOR_CLASS_NAME.key -> classOf[SqlKeyGenerator].getCanonicalName,
SqlKeyGenerator.ORIGINAL_KEYGEN_CLASS_NAME -> keyGeneratorClassName,
SqlKeyGenerator.PARTITION_SCHEMA -> hoodieCatalogTable.partitionSchema.toDDL,
- RECORDKEY_FIELD.key -> hoodieCatalogTable.primaryKeys.mkString(","),
- PRECOMBINE_FIELD.key -> preCombineField,
- PARTITIONPATH_FIELD.key -> partitionFieldsStr,
HoodieSyncConfig.META_SYNC_PARTITION_FIELDS.key -> partitionFieldsStr,
HoodieSyncConfig.META_SYNC_ENABLED.key -> hiveSyncConfig.getString(HoodieSyncConfig.META_SYNC_ENABLED.key),
HiveSyncConfigHolder.HIVE_SYNC_ENABLED.key -> hiveSyncConfig.getString(HiveSyncConfigHolder.HIVE_SYNC_ENABLED.key),
@@ -209,6 +197,18 @@ trait ProvidesHoodieConfig extends Logging {
HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS.key -> hiveSyncConfig.getStringOrDefault(HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS)
)
+ val overridingOpts = extraOptions ++ Map(
+ "path" -> path,
+ TABLE_TYPE.key -> tableType,
+ TBL_NAME.key -> hoodieCatalogTable.tableName,
+ OPERATION.key -> operation,
+ HIVE_STYLE_PARTITIONING.key -> hiveStylePartitioningEnable,
+ URL_ENCODE_PARTITIONING.key -> urlEncodePartitioning,
+ RECORDKEY_FIELD.key -> hoodieCatalogTable.primaryKeys.mkString(","),
+ PRECOMBINE_FIELD.key -> preCombineField,
+ PARTITIONPATH_FIELD.key -> partitionFieldsStr
+ )
+
combineOptions(hoodieCatalogTable, tableConfig, sparkSession.sqlContext.conf,
defaultOpts = defaultOpts, overridingOpts = overridingOpts)
}
@@ -257,17 +257,10 @@ trait ProvidesHoodieConfig extends Logging {
val hiveSyncConfig = buildHiveSyncConfig(sparkSession, hoodieCatalogTable, tableConfig)
- val overridingOpts = Map(
- "path" -> path,
- RECORDKEY_FIELD.key -> hoodieCatalogTable.primaryKeys.mkString(","),
- TBL_NAME.key -> tableConfig.getTableName,
- HIVE_STYLE_PARTITIONING.key -> tableConfig.getHiveStylePartitioningEnable,
- URL_ENCODE_PARTITIONING.key -> tableConfig.getUrlEncodePartitioning,
+ val defaultOpts = Map(
KEYGENERATOR_CLASS_NAME.key -> classOf[SqlKeyGenerator].getCanonicalName,
SqlKeyGenerator.ORIGINAL_KEYGEN_CLASS_NAME -> tableConfig.getKeyGeneratorClassName,
SqlKeyGenerator.PARTITION_SCHEMA -> partitionSchema.toDDL,
- OPERATION.key -> DataSourceWriteOptions.DELETE_OPERATION_OPT_VAL,
- PARTITIONPATH_FIELD.key -> tableConfig.getPartitionFieldProp,
HoodieSyncConfig.META_SYNC_ENABLED.key -> hiveSyncConfig.getString(HoodieSyncConfig.META_SYNC_ENABLED.key),
HiveSyncConfigHolder.HIVE_SYNC_ENABLED.key -> hiveSyncConfig.getString(HiveSyncConfigHolder.HIVE_SYNC_ENABLED.key),
HiveSyncConfigHolder.HIVE_SYNC_MODE.key -> hiveSyncConfig.getStringOrDefault(HiveSyncConfigHolder.HIVE_SYNC_MODE, HiveSyncMode.HMS.name()),
@@ -278,8 +271,18 @@ trait ProvidesHoodieConfig extends Logging {
HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS.key -> hiveSyncConfig.getStringOrDefault(HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS)
)
+ val overridingOpts = Map(
+ "path" -> path,
+ RECORDKEY_FIELD.key -> hoodieCatalogTable.primaryKeys.mkString(","),
+ TBL_NAME.key -> tableConfig.getTableName,
+ HIVE_STYLE_PARTITIONING.key -> tableConfig.getHiveStylePartitioningEnable,
+ URL_ENCODE_PARTITIONING.key -> tableConfig.getUrlEncodePartitioning,
+ OPERATION.key -> DataSourceWriteOptions.DELETE_OPERATION_OPT_VAL,
+ PARTITIONPATH_FIELD.key -> tableConfig.getPartitionFieldProp
+ )
+
combineOptions(hoodieCatalogTable, tableConfig, sparkSession.sqlContext.conf,
- defaultOpts = Map.empty, overridingOpts = overridingOpts)
+ defaultOpts = defaultOpts, overridingOpts = overridingOpts)
}
def buildHiveSyncConfig(sparkSession: SparkSession,
[hudi] 06/06: [HUDI-5702] Avoid to write useless change-data when compaction (#7849)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch release-0.13.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 2b17dd58243e6b688a2e1927748479809437a1a4
Author: Yann Byron <bi...@gmail.com>
AuthorDate: Sun Feb 5 08:06:05 2023 +0800
[HUDI-5702] Avoid to write useless change-data when compaction (#7849)
Co-authored-by: Raymond Xu <27...@users.noreply.github.com>
---
.../org/apache/hudi/io/HoodieMergeHandleFactory.java | 18 ++++--------------
1 file changed, 4 insertions(+), 14 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
index b110c2c0817..8aadd637f0e 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
@@ -80,21 +80,11 @@ public class HoodieMergeHandleFactory {
TaskContextSupplier taskContextSupplier,
Option<BaseKeyGenerator> keyGeneratorOpt) {
if (table.requireSortedRecords()) {
- if (table.getMetaClient().getTableConfig().isCDCEnabled()) {
- return new HoodieSortedMergeHandleWithChangeLog<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
- dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
- } else {
- return new HoodieSortedMergeHandle<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
- dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
- }
+ return new HoodieSortedMergeHandle<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
+ dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
} else {
- if (table.getMetaClient().getTableConfig().isCDCEnabled()) {
- return new HoodieMergeHandleWithChangeLog<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
- dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
- } else {
- return new HoodieMergeHandle<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
- dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
- }
+ return new HoodieMergeHandle<>(writeConfig, instantTime, table, keyToNewRecords, partitionPath, fileId,
+ dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
}
}
}
[hudi] 04/06: [HUDI-5700] Annotate config classes for docs generation (#7853)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch release-0.13.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit eb6bceefa5e6b8e7be77e5785dd0d009805cb37c
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Sat Feb 4 13:49:07 2023 -0800
[HUDI-5700] Annotate config classes for docs generation (#7853)
This PR annotates the missing config classes so that the configs in these classes are properly processed and populated on our "All Configurations" page.
---
.../apache/hudi/common/config/ConfigGroups.java | 1 +
.../hudi/gcp/bigquery/BigQuerySyncConfig.java | 8 ++++++++
.../sync/datahub/config/DataHubSyncConfig.java | 24 ++++++++++++++--------
.../java/org/apache/hudi/hive/HiveSyncConfig.java | 8 ++++++++
.../hive/replication/GlobalHiveSyncConfig.java | 8 ++++++++
.../apache/hudi/sync/common/HoodieSyncConfig.java | 8 ++++++++
6 files changed, 49 insertions(+), 8 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/ConfigGroups.java b/hudi-common/src/main/java/org/apache/hudi/common/config/ConfigGroups.java
index 0488401f10a..6fe8e5002a8 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/ConfigGroups.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/ConfigGroups.java
@@ -32,6 +32,7 @@ public class ConfigGroups {
SPARK_DATASOURCE("Spark Datasource Configs"),
FLINK_SQL("Flink Sql Configs"),
WRITE_CLIENT("Write Client Configs"),
+ META_SYNC("Metastore and Catalog Sync Configs"),
METRICS("Metrics Configs"),
RECORD_PAYLOAD("Record Payload Config"),
KAFKA_CONNECT("Kafka Connect Configs"),
diff --git a/hudi-gcp/src/main/java/org/apache/hudi/gcp/bigquery/BigQuerySyncConfig.java b/hudi-gcp/src/main/java/org/apache/hudi/gcp/bigquery/BigQuerySyncConfig.java
index 52b3d3b74e5..8c04c0fa136 100644
--- a/hudi-gcp/src/main/java/org/apache/hudi/gcp/bigquery/BigQuerySyncConfig.java
+++ b/hudi-gcp/src/main/java/org/apache/hudi/gcp/bigquery/BigQuerySyncConfig.java
@@ -19,6 +19,8 @@
package org.apache.hudi.gcp.bigquery;
+import org.apache.hudi.common.config.ConfigClassProperty;
+import org.apache.hudi.common.config.ConfigGroups;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.sync.common.HoodieSyncConfig;
@@ -26,12 +28,18 @@ import org.apache.hudi.sync.common.HoodieSyncConfig;
import com.beust.jcommander.Parameter;
import com.beust.jcommander.ParametersDelegate;
+import javax.annotation.concurrent.Immutable;
+
import java.io.Serializable;
import java.util.Properties;
/**
* Configs needed to sync data into BigQuery.
*/
+@Immutable
+@ConfigClassProperty(name = "BigQuery Sync Configs",
+ groupName = ConfigGroups.Names.META_SYNC,
+ description = "Configurations used by the Hudi to sync metadata to Google BigQuery.")
public class BigQuerySyncConfig extends HoodieSyncConfig implements Serializable {
public static final ConfigProperty<String> BIGQUERY_SYNC_PROJECT_ID = ConfigProperty
diff --git a/hudi-sync/hudi-datahub-sync/src/main/java/org/apache/hudi/sync/datahub/config/DataHubSyncConfig.java b/hudi-sync/hudi-datahub-sync/src/main/java/org/apache/hudi/sync/datahub/config/DataHubSyncConfig.java
index e4fa695ecef..d9a3f00e46b 100644
--- a/hudi-sync/hudi-datahub-sync/src/main/java/org/apache/hudi/sync/datahub/config/DataHubSyncConfig.java
+++ b/hudi-sync/hudi-datahub-sync/src/main/java/org/apache/hudi/sync/datahub/config/DataHubSyncConfig.java
@@ -19,6 +19,8 @@
package org.apache.hudi.sync.datahub.config;
+import org.apache.hudi.common.config.ConfigClassProperty;
+import org.apache.hudi.common.config.ConfigGroups;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.ReflectionUtils;
@@ -28,11 +30,17 @@ import com.beust.jcommander.Parameter;
import com.beust.jcommander.ParametersDelegate;
import datahub.client.rest.RestEmitter;
+import javax.annotation.concurrent.Immutable;
+
import java.util.Properties;
import static org.apache.hudi.sync.datahub.config.HoodieDataHubDatasetIdentifier.DEFAULT_DATAHUB_ENV;
import static org.apache.hudi.sync.datahub.config.HoodieDataHubDatasetIdentifier.DEFAULT_HOODIE_DATAHUB_PLATFORM_NAME;
+@Immutable
+@ConfigClassProperty(name = "DataHub Sync Configs",
+ groupName = ConfigGroups.Names.META_SYNC,
+ description = "Configurations used by the Hudi to sync metadata to DataHub.")
public class DataHubSyncConfig extends HoodieSyncConfig {
public static final ConfigProperty<String> META_SYNC_DATAHUB_DATASET_IDENTIFIER_CLASS = ConfigProperty
@@ -56,15 +64,15 @@ public class DataHubSyncConfig extends HoodieSyncConfig {
.withDocumentation("Pluggable class to supply a DataHub REST emitter to connect to the DataHub instance. This overwrites other emitter configs.");
public static final ConfigProperty<String> META_SYNC_DATAHUB_DATAPLATFORM_NAME = ConfigProperty
- .key("hoodie.meta.sync.datahub.dataplatform.name")
- .defaultValue(DEFAULT_HOODIE_DATAHUB_PLATFORM_NAME)
- .withDocumentation("String used to represent Hudi when creating its corresponding DataPlatform entity "
- + "within Datahub");
+ .key("hoodie.meta.sync.datahub.dataplatform.name")
+ .defaultValue(DEFAULT_HOODIE_DATAHUB_PLATFORM_NAME)
+ .withDocumentation("String used to represent Hudi when creating its corresponding DataPlatform entity "
+ + "within Datahub");
public static final ConfigProperty<String> META_SYNC_DATAHUB_DATASET_ENV = ConfigProperty
- .key("hoodie.meta.sync.datahub.dataset.env")
- .defaultValue(DEFAULT_DATAHUB_ENV.name())
- .withDocumentation("Environment to use when pushing entities to Datahub");
+ .key("hoodie.meta.sync.datahub.dataset.env")
+ .defaultValue(DEFAULT_DATAHUB_ENV.name())
+ .withDocumentation("Environment to use when pushing entities to Datahub");
public final HoodieDataHubDatasetIdentifier datasetIdentifier;
@@ -102,7 +110,7 @@ public class DataHubSyncConfig extends HoodieSyncConfig {
public String emitterSupplierClass;
@Parameter(names = {"--data-platform-name"}, description = "String used to represent Hudi when creating its "
- + "corresponding DataPlatform entity within Datahub")
+ + "corresponding DataPlatform entity within Datahub")
public String dataPlatformName;
@Parameter(names = {"--dataset-env"}, description = "Which Datahub Environment to use when pushing entities")
diff --git a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfig.java b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfig.java
index db0fe820942..9d23b88cc1b 100644
--- a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfig.java
+++ b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncConfig.java
@@ -18,6 +18,8 @@
package org.apache.hudi.hive;
+import org.apache.hudi.common.config.ConfigClassProperty;
+import org.apache.hudi.common.config.ConfigGroups;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.ValidationUtils;
@@ -28,11 +30,17 @@ import com.beust.jcommander.ParametersDelegate;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
+import javax.annotation.concurrent.Immutable;
+
import java.util.Properties;
/**
* Configs needed to sync data into the Hive Metastore.
*/
+@Immutable
+@ConfigClassProperty(name = "Hive Sync Configs",
+ groupName = ConfigGroups.Names.META_SYNC,
+ description = "Configurations used by the Hudi to sync metadata to Hive Metastore.")
public class HiveSyncConfig extends HoodieSyncConfig {
/*
diff --git a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/replication/GlobalHiveSyncConfig.java b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/replication/GlobalHiveSyncConfig.java
index dc4355c57c3..0f0a99bbbb1 100644
--- a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/replication/GlobalHiveSyncConfig.java
+++ b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/replication/GlobalHiveSyncConfig.java
@@ -18,6 +18,8 @@
package org.apache.hudi.hive.replication;
+import org.apache.hudi.common.config.ConfigClassProperty;
+import org.apache.hudi.common.config.ConfigGroups;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.hive.HiveSyncConfig;
@@ -26,8 +28,14 @@ import com.beust.jcommander.Parameter;
import com.beust.jcommander.ParametersDelegate;
import org.apache.hadoop.conf.Configuration;
+import javax.annotation.concurrent.Immutable;
+
import java.util.Properties;
+@Immutable
+@ConfigClassProperty(name = "Global Hive Sync Configs",
+ groupName = ConfigGroups.Names.META_SYNC,
+ description = "Global replication configurations used by the Hudi to sync metadata to Hive Metastore.")
public class GlobalHiveSyncConfig extends HiveSyncConfig {
public static final ConfigProperty<String> META_SYNC_GLOBAL_REPLICATE_TIMESTAMP = ConfigProperty
diff --git a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncConfig.java b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncConfig.java
index c0d1075c779..eb9affe3085 100644
--- a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncConfig.java
+++ b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieSyncConfig.java
@@ -18,6 +18,8 @@
package org.apache.hudi.sync.common;
+import org.apache.hudi.common.config.ConfigClassProperty;
+import org.apache.hudi.common.config.ConfigGroups;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.config.HoodieMetadataConfig;
@@ -35,6 +37,8 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
+import javax.annotation.concurrent.Immutable;
+
import java.util.Comparator;
import java.util.List;
import java.util.Properties;
@@ -51,6 +55,10 @@ import static org.apache.hudi.common.table.HoodieTableConfig.URL_ENCODE_PARTITIO
/**
* Configs needed to sync data into external meta stores, catalogs, etc.
*/
+@Immutable
+@ConfigClassProperty(name = "Metadata Sync Configs",
+ groupName = ConfigGroups.Names.META_SYNC,
+ description = "Configurations used by the Hudi to sync metadata to external metastores and catalogs.")
public class HoodieSyncConfig extends HoodieConfig {
private static final Logger LOG = LogManager.getLogger(HoodieSyncConfig.class);
[hudi] 02/06: [MINOR] Validate configs for OCC early conflict detection (#7848)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch release-0.13.0
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 989d5403666e71f5a3080875d44f8446e454e5b0
Author: Shiyan Xu <27...@users.noreply.github.com>
AuthorDate: Sat Feb 4 03:39:58 2023 -0600
[MINOR] Validate configs for OCC early conflict detection (#7848)
* [MINOR] Validate configs for OCC early conflict detection
* fix style
---
.../java/org/apache/hudi/config/HoodieWriteConfig.java | 15 ++++++++++-----
1 file changed, 10 insertions(+), 5 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
index 7e29c7dcbae..707364b4b21 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
@@ -53,7 +53,6 @@ import org.apache.hudi.common.util.HoodieRecordUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.common.util.StringUtils;
-import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.common.util.queue.DisruptorWaitStrategyType;
import org.apache.hudi.common.util.queue.ExecutorType;
@@ -99,6 +98,7 @@ import java.util.Properties;
import java.util.function.Supplier;
import java.util.stream.Collectors;
+import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
import static org.apache.hudi.common.util.queue.ExecutorType.SIMPLE;
import static org.apache.hudi.config.HoodieCleanConfig.CLEANER_POLICY;
import static org.apache.hudi.table.marker.ConflictDetectionUtils.getDefaultEarlyConflictDetectionStrategy;
@@ -2956,9 +2956,14 @@ public class HoodieWriteConfig extends HoodieConfig {
// Ensure Layout Version is good
new TimelineLayoutVersion(Integer.parseInt(layoutVersion));
Objects.requireNonNull(writeConfig.getString(BASE_PATH));
+ if (writeConfig.isEarlyConflictDetectionEnable()) {
+ checkArgument(writeConfig.getString(WRITE_CONCURRENCY_MODE)
+ .equalsIgnoreCase(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.value()),
+ "To use early conflict detection, set hoodie.write.concurrency.mode=OPTIMISTIC_CONCURRENCY_CONTROL");
+ }
if (writeConfig.getString(WRITE_CONCURRENCY_MODE)
.equalsIgnoreCase(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL.value())) {
- ValidationUtils.checkArgument(!writeConfig.getString(HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY)
+ checkArgument(!writeConfig.getString(HoodieCleanConfig.FAILED_WRITES_CLEANER_POLICY)
.equals(HoodieFailedWritesCleaningPolicy.EAGER.name()), "To enable optimistic concurrency control, set hoodie.cleaner.policy.failed.writes=LAZY");
}
@@ -2969,12 +2974,12 @@ public class HoodieWriteConfig extends HoodieConfig {
int maxInstantsToKeep = Integer.parseInt(writeConfig.getStringOrDefault(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP));
int cleanerCommitsRetained =
Integer.parseInt(writeConfig.getStringOrDefault(HoodieCleanConfig.CLEANER_COMMITS_RETAINED));
- ValidationUtils.checkArgument(maxInstantsToKeep > minInstantsToKeep,
+ checkArgument(maxInstantsToKeep > minInstantsToKeep,
String.format(
"Increase %s=%d to be greater than %s=%d.",
HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(), maxInstantsToKeep,
HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(), minInstantsToKeep));
- ValidationUtils.checkArgument(minInstantsToKeep > cleanerCommitsRetained,
+ checkArgument(minInstantsToKeep > cleanerCommitsRetained,
String.format(
"Increase %s=%d to be greater than %s=%d. Otherwise, there is risk of incremental pull "
+ "missing data from few instants.",
@@ -2983,7 +2988,7 @@ public class HoodieWriteConfig extends HoodieConfig {
boolean inlineCompact = writeConfig.getBoolean(HoodieCompactionConfig.INLINE_COMPACT);
boolean inlineCompactSchedule = writeConfig.getBoolean(HoodieCompactionConfig.SCHEDULE_INLINE_COMPACT);
- ValidationUtils.checkArgument(!(inlineCompact && inlineCompactSchedule), String.format("Either of inline compaction (%s) or "
+ checkArgument(!(inlineCompact && inlineCompactSchedule), String.format("Either of inline compaction (%s) or "
+ "schedule inline compaction (%s) can be enabled. Both can't be set to true at the same time. %s, %s", HoodieCompactionConfig.INLINE_COMPACT.key(),
HoodieCompactionConfig.SCHEDULE_INLINE_COMPACT.key(), inlineCompact, inlineCompactSchedule));
}