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 2024/05/14 22:09:00 UTC
(hudi) 24/28: [HUDI-6317] Streaming read should skip compaction and clustering instants to avoid duplicates (#8884)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d8cccb2ee12ab3d0367769de525f41b5f23ac232
Author: Nicholas Jiang <pr...@163.com>
AuthorDate: Thu Mar 28 10:44:45 2024 +0800
[HUDI-6317] Streaming read should skip compaction and clustering instants to avoid duplicates (#8884)
---
.../java/org/apache/hudi/configuration/FlinkOptions.java | 4 ++--
.../java/org/apache/hudi/table/ITTestHoodieDataSource.java | 14 +++++++++++---
.../java/org/apache/hudi/table/ITTestSchemaEvolution.java | 6 ++++--
3 files changed, 17 insertions(+), 7 deletions(-)
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
index 6f0f6db7c28..0f934b609f6 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
@@ -316,7 +316,7 @@ public class FlinkOptions extends HoodieConfig {
public static final ConfigOption<Boolean> READ_STREAMING_SKIP_COMPACT = ConfigOptions
.key("read.streaming.skip_compaction")
.booleanType()
- .defaultValue(false)// default read as batch
+ .defaultValue(true)
.withDescription("Whether to skip compaction instants and avoid reading compacted base files for streaming read to improve read performance.\n"
+ "This option can be used to avoid reading duplicates when changelog mode is enabled, it is a solution to keep data integrity\n");
@@ -325,7 +325,7 @@ public class FlinkOptions extends HoodieConfig {
public static final ConfigOption<Boolean> READ_STREAMING_SKIP_CLUSTERING = ConfigOptions
.key("read.streaming.skip_clustering")
.booleanType()
- .defaultValue(false)
+ .defaultValue(true)
.withDescription("Whether to skip clustering instants to avoid reading base files of clustering operations for streaming read "
+ "to improve read performance.");
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
index de80a219989..9be2090f5bc 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
@@ -133,6 +133,7 @@ public class ITTestHoodieDataSource {
hoodieTableDDL = sql("t1")
.option(FlinkOptions.PATH, tempFile.getAbsolutePath())
.option(FlinkOptions.READ_AS_STREAMING, true)
+ .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, false)
.option(FlinkOptions.TABLE_TYPE, tableType)
.option(FlinkOptions.READ_START_COMMIT, firstCommit)
.end();
@@ -165,6 +166,7 @@ public class ITTestHoodieDataSource {
String hoodieTableDDL = sql("t1")
.option(FlinkOptions.PATH, tempFile.getAbsolutePath())
.option(FlinkOptions.READ_AS_STREAMING, true)
+ .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, false)
.option(FlinkOptions.CDC_ENABLED, true)
.option(FlinkOptions.SUPPLEMENTAL_LOGGING_MODE, mode.name())
.end();
@@ -198,6 +200,7 @@ public class ITTestHoodieDataSource {
String hoodieTableDDL = sql("t1")
.option(FlinkOptions.PATH, tempFile.getAbsolutePath())
.option(FlinkOptions.READ_AS_STREAMING, true)
+ .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, false)
.option(FlinkOptions.TABLE_TYPE, tableType)
.end();
streamTableEnv.executeSql(hoodieTableDDL);
@@ -241,6 +244,7 @@ public class ITTestHoodieDataSource {
String createHoodieTable2 = sql("t2")
.option(FlinkOptions.PATH, tempFile.getAbsolutePath())
.option(FlinkOptions.READ_AS_STREAMING, true)
+ .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, false)
.option(FlinkOptions.TABLE_TYPE, tableType)
.option(FlinkOptions.READ_START_COMMIT, specifiedCommit)
.end();
@@ -334,7 +338,6 @@ public class ITTestHoodieDataSource {
.option(FlinkOptions.PATH, tempFile.getAbsolutePath())
.option(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_MERGE_ON_READ)
.option(FlinkOptions.READ_AS_STREAMING, true)
- .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, true)
.option(FlinkOptions.COMPACTION_DELTA_COMMITS, 1)
.option(FlinkOptions.COMPACTION_TASKS, 1)
.end();
@@ -361,7 +364,6 @@ public class ITTestHoodieDataSource {
.option(FlinkOptions.PATH, tempFile.getAbsolutePath())
.option(FlinkOptions.OPERATION, "insert")
.option(FlinkOptions.READ_AS_STREAMING, true)
- .option(FlinkOptions.READ_STREAMING_SKIP_CLUSTERING, true)
.option(FlinkOptions.CLUSTERING_SCHEDULE_ENABLED,true)
.option(FlinkOptions.CLUSTERING_ASYNC_ENABLED, true)
.option(FlinkOptions.CLUSTERING_DELTA_COMMITS,1)
@@ -492,6 +494,7 @@ public class ITTestHoodieDataSource {
.option(FlinkOptions.TABLE_TYPE, tableType)
.option(FlinkOptions.READ_AS_STREAMING, true)
.option(FlinkOptions.READ_STREAMING_CHECK_INTERVAL, 2)
+ .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, false)
.option(FlinkOptions.HIVE_STYLE_PARTITIONING, hiveStylePartitioning)
.end();
streamTableEnv.executeSql(hoodieTableDDL);
@@ -677,7 +680,8 @@ public class ITTestHoodieDataSource {
+ "with (\n"
+ " 'connector' = 'hudi',\n"
+ " 'path' = '" + tempFile.getAbsolutePath() + "',\n"
- + " 'read.streaming.enabled' = '" + streaming + "'\n"
+ + " 'read.streaming.enabled' = '" + streaming + "',\n"
+ + " 'read.streaming.skip_compaction' = 'false'\n"
+ ")";
streamTableEnv.executeSql(hoodieTableDDL);
String insertInto = "insert into t1 values\n"
@@ -723,6 +727,7 @@ public class ITTestHoodieDataSource {
.noPartition()
.option(FlinkOptions.PATH, tempFile.getAbsolutePath())
.option(FlinkOptions.READ_AS_STREAMING, streaming)
+ .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, false)
.end();
streamTableEnv.executeSql(hoodieTableDDL);
String insertInto = "insert into t1 values\n"
@@ -826,6 +831,7 @@ public class ITTestHoodieDataSource {
String hoodieTableDDL = sql("t1")
.option(FlinkOptions.PATH, tempFile.getAbsolutePath())
.option(FlinkOptions.READ_AS_STREAMING, true)
+ .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, false)
.option(FlinkOptions.TABLE_TYPE, tableType)
.option(FlinkOptions.READ_START_COMMIT, "earliest")
.option(FlinkOptions.WRITE_BATCH_SIZE, 0.00001)
@@ -1078,6 +1084,7 @@ public class ITTestHoodieDataSource {
.pkField("id")
.option(FlinkOptions.PATH, tempFile.getAbsolutePath())
.option(FlinkOptions.READ_AS_STREAMING, execMode == ExecMode.STREAM)
+ .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, false)
.option(FlinkOptions.PRE_COMBINE, true)
.noPartition()
.end();
@@ -2020,6 +2027,7 @@ public class ITTestHoodieDataSource {
.option(FlinkOptions.TABLE_TYPE, tableType)
.option(FlinkOptions.READ_AS_STREAMING, true)
.option(FlinkOptions.READ_STREAMING_CHECK_INTERVAL, 2)
+ .option(FlinkOptions.READ_STREAMING_SKIP_COMPACT, false)
.option(FlinkOptions.HIVE_STYLE_PARTITIONING, hiveStylePartitioning)
.end();
streamTableEnv.executeSql(hoodieTableDDL);
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestSchemaEvolution.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestSchemaEvolution.java
index 0417285815a..46f51df741f 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestSchemaEvolution.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestSchemaEvolution.java
@@ -90,7 +90,8 @@ public class ITTestSchemaEvolution {
public void testMergeOnReadInputFormatBaseFileOnlyIterator() throws Exception {
TableOptions tableOptions = defaultTableOptions(tempFile.getAbsolutePath())
.withOption(FlinkOptions.READ_AS_STREAMING.key(), true)
- .withOption(FlinkOptions.READ_START_COMMIT.key(), FlinkOptions.START_COMMIT_EARLIEST);
+ .withOption(FlinkOptions.READ_START_COMMIT.key(), FlinkOptions.START_COMMIT_EARLIEST)
+ .withOption(FlinkOptions.READ_STREAMING_SKIP_COMPACT.key(), false);
testSchemaEvolution(tableOptions);
}
@@ -98,7 +99,8 @@ public class ITTestSchemaEvolution {
public void testMergeOnReadInputFormatBaseFileOnlyFilteringIterator() throws Exception {
TableOptions tableOptions = defaultTableOptions(tempFile.getAbsolutePath())
.withOption(FlinkOptions.READ_AS_STREAMING.key(), true)
- .withOption(FlinkOptions.READ_START_COMMIT.key(), 1);
+ .withOption(FlinkOptions.READ_START_COMMIT.key(), 1)
+ .withOption(FlinkOptions.READ_STREAMING_SKIP_COMPACT.key(), false);
testSchemaEvolution(tableOptions);
}