Skip to content

[HUDI-4543] Support natural order when table schema contains a field named 'ts' #6246

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
danny0405 merged 4 commits into from
Aug 10, 2022
Merged

[HUDI-4543] Support natural order when table schema contains a field named 'ts' #6246

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
880a38e
be able to disable precombine field when table schema contains a fiel…
flashJd Jul 29, 2022
0c6cfaa
Merge branch 'master' into deepnova
flashJd Jul 29, 2022
39773f8
make the judgment more elegant
Aug 4, 2022
7b04e73
fix checkStyle conflict
Aug 5, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ private void sanityCheck(Configuration conf, ResolvedSchema schema) {
}
if (preCombineField.equals(FlinkOptions.PRECOMBINE_FIELD.defaultValue())) {
conf.setString(FlinkOptions.PRECOMBINE_FIELD, FlinkOptions.NO_PRE_COMBINE);
Copy link
Contributor

@danny0405 danny0405 Aug 3, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this change valid ? We still set up the option if it is already FlinkOptions.NO_PRE_COMBINE.

Copy link
Contributor Author

@flashJd flashJd Aug 3, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it's valid, suppose a table with a 'ts' column, we want to set PRECOMBINE_FIELD=FlinkOptions.NO_PRE_COMBINE to disable the precombine field, above code will go into line 153, because preCombineField.equals(FlinkOptions.PRECOMBINE_FIELD.defaultValue()) is false, FlinkOptions.PRECOMBINE_FIELD.defaultValue() is 'ts', preCombineField is what we set 'no_precombine', you can also see the test.

Copy link
Contributor Author

@flashJd flashJd Aug 4, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this change valid ? We still set up the option if it is already FlinkOptions.NO_PRE_COMBINE.

@danny0405 looking forward to your reply

Copy link
Contributor

@danny0405 danny0405 Aug 4, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One thing need to note here is that: option FlinkOptions.NO_PRE_COMBINE is only for internal use.

Still curious about your use case, do you mean you want the proc_time/natural order sequence semantics even though you got a ts field in the schema ?

Copy link
Contributor Author

@flashJd flashJd Aug 4, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, i want the proc_time/natural order sequence semantics even though you got a ts field in the schema

Copy link
Contributor

@danny0405 danny0405 Aug 4, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Then we can do such logic:
else if (!preCombineField.equals(FlinkOptions.NO_PRE_COMBINE)) {
throw new HoodieValidationException("Field " + preCombineField + " does not exist in the table schema."
+ "Please check '" + FlinkOptions.PRECOMBINE_FIELD.key() + "' option.");
}

Copy link
Contributor Author

@flashJd flashJd Aug 4, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, it is more elegant. I've modified to it

} else {
} else if (!preCombineField.equals(FlinkOptions.NO_PRE_COMBINE)) {
throw new HoodieValidationException("Field " + preCombineField + " does not exist in the table schema."
+ "Please check '" + FlinkOptions.PRECOMBINE_FIELD.key() + "' option.");
}
Expand Down
24 changes: 24 additions & 0 deletions ...ink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/ITTestHoodieDataSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -461,6 +461,30 @@ void testWriteAndReadWithProctimeSequence(HoodieTableType tableType, boolean hiv
assertRowsEquals(result1, "[+I[id1, Danny, 23, 1970-01-01T00:00:01, par1]]");
}

@ParameterizedTest
@MethodSource("tableTypeAndPartitioningParams")
void testWriteAndReadWithProctimeSequenceWithTsColumnExisting(HoodieTableType tableType, boolean hiveStylePartitioning) {
TableEnvironment tableEnv = batchTableEnv;
String hoodieTableDDL = sql("t1")
.field("uuid varchar(20)")
.field("name varchar(10)")
.field("age int")
.field("ts timestamp(3)") // use the default precombine field 'ts'
.field("`partition` varchar(10)")
.option(FlinkOptions.PATH, tempFile.getAbsolutePath())
.option(FlinkOptions.TABLE_TYPE, tableType)
.option(FlinkOptions.HIVE_STYLE_PARTITIONING, hiveStylePartitioning)
.option(FlinkOptions.PRECOMBINE_FIELD, FlinkOptions.NO_PRE_COMBINE)
.end();
tableEnv.executeSql(hoodieTableDDL);

execInsertSql(tableEnv, TestSQL.INSERT_SAME_KEY_T1);

List<Row> result1 = CollectionUtil.iterableToList(
() -> tableEnv.sqlQuery("select * from t1").execute().collect());
assertRowsEquals(result1, "[+I[id1, Danny, 23, 1970-01-01T00:00:01, par1]]");
}

@ParameterizedTest
@EnumSource(value = HoodieTableType.class)
void testBatchModeUpsertWithoutPartition(HoodieTableType tableType) {
Expand Down
14 changes: 14 additions & 0 deletions ...ink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,6 +152,20 @@ void testRequiredOptionsForSource() {
HoodieTableSink tableSink5 = (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(sourceContext5);
assertThat(tableSource5.getConf().getString(FlinkOptions.PAYLOAD_CLASS_NAME), is(EventTimeAvroPayload.class.getName()));
assertThat(tableSink5.getConf().getString(FlinkOptions.PAYLOAD_CLASS_NAME), is(EventTimeAvroPayload.class.getName()));

// given pk and set pre combine key to no_precombine will be ok
ResolvedSchema schema5 = SchemaBuilder.instance()
.field("f0", DataTypes.INT().notNull())
.field("f1", DataTypes.VARCHAR(20))
.field("f2", DataTypes.TIMESTAMP(3))
.field("ts", DataTypes.TIMESTAMP(3))
.primaryKey("f0")
.build();
this.conf.setString(FlinkOptions.PRECOMBINE_FIELD, FlinkOptions.NO_PRE_COMBINE);
final MockContext sourceContext6 = MockContext.getInstance(this.conf, schema5, "f2");

assertDoesNotThrow(() -> new HoodieTableFactory().createDynamicTableSource(sourceContext6));
assertDoesNotThrow(() -> new HoodieTableFactory().createDynamicTableSink(sourceContext6));
}

@Test
Expand Down