You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "xxubai (via GitHub)" <gi...@apache.org> on 2026/09/21 02:57:02 UTC
[PR] [lake/iceberg] Fix temporal identifier keys in tiering writer [fluss]
xxubai opened a new pull request, #4451:
URL: https://github.com/apache/fluss/pull/4451
<!--
Generated-by: OpenAI Codex (GPT-5) following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
Linked issue: close #4444
Fix tiering failures when a temporal column is part of an Iceberg identifier key.
### Brief change log
- Wrap table rows and equality keys with schema-aware `InternalRecordWrapper` instances.
- Cover `DATE`, `TIME`, `TIMESTAMP`, and `TIMESTAMP_LTZ` update/delete comparisons.
### Tests
- `IcebergTieringTest` (14 tests)
### API and Format
No API or storage format changes.
### Documentation
No documentation changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Fix temporal identifier keys in tiering writer [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #4451:
URL: https://github.com/apache/fluss/pull/4451#discussion_r4059236731
##########
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java:
##########
@@ -211,6 +251,94 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitionedTable
}
}
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("temporalIdentifierArgs")
+ void testTieringWriteTemporalIdentifier(
+ String typeName,
+ DataType flussTemporalType,
+ Type icebergTemporalType,
+ Object flussTemporalValue,
+ Object expectedIcebergValue)
+ throws Exception {
+ TablePath tablePath = TablePath.of("iceberg", "test_temporal_identifier_write_" + typeName);
+ TableInfo tableInfo =
+ createTemporalIdentifierTable(tablePath, flussTemporalType, icebergTemporalType);
+
+ GenericRow insertRow = temporalIdentifierRow(flussTemporalValue, "before");
+ GenericRow updateRow = temporalIdentifierRow(flussTemporalValue, "after");
+
+ IcebergWriteResult writeResult;
+ try (LakeWriter<IcebergWriteResult> writer =
+ createLakeWriter(tablePath, 0, null, null, tableInfo)) {
+ writer.write(toRecord(0L, insertRow, INSERT));
+ // The equal id makes StructLikeMap.put continue to compare the temporal identifier
+ // when UPDATE_AFTER replaces the same composite identifier key.
+ // See
+ // https://github.com/apache/iceberg/blob/apache-iceberg-1.10.1/core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java#L153-L162
+ writer.write(toRecord(1L, updateRow, UPDATE_AFTER));
+ writeResult = writer.complete();
+ }
+ try (LakeCommitter<IcebergWriteResult, IcebergCommittable> committer =
+ createLakeCommitter(tablePath, tableInfo)) {
+ committer.commit(
+ committer.toCommittable(Collections.singletonList(writeResult)),
+ Collections.emptyMap());
+ }
+
+ Table table = icebergCatalog.loadTable(toIceberg(tablePath));
+ table.refresh();
+ try (CloseableIterator<Record> rows = IcebergGenerics.read(table).build().iterator()) {
+ assertThat(rows.hasNext()).isTrue();
+ Record actual = rows.next();
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("temporal_identifier")).isEqualTo(expectedIcebergValue);
+ assertThat(actual.getField("payload")).isEqualTo("after");
+ assertThat(rows.hasNext()).isFalse();
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("temporalIdentifierArgs")
+ void testTieringDeleteTemporalIdentifier(
Review Comment:
is it possible to merge the test logic of `testTieringDeleteTemporalIdentifier` into `testTieringWriteTemporalIdentifier`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Fix temporal identifier keys in tiering writer [fluss]
Posted by "xxubai (via GitHub)" <gi...@apache.org>.
xxubai commented on code in PR #4451:
URL: https://github.com/apache/fluss/pull/4451#discussion_r4059494156
##########
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java:
##########
@@ -211,6 +251,94 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitionedTable
}
}
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("temporalIdentifierArgs")
+ void testTieringWriteTemporalIdentifier(
+ String typeName,
+ DataType flussTemporalType,
+ Type icebergTemporalType,
+ Object flussTemporalValue,
+ Object expectedIcebergValue)
+ throws Exception {
+ TablePath tablePath = TablePath.of("iceberg", "test_temporal_identifier_write_" + typeName);
+ TableInfo tableInfo =
+ createTemporalIdentifierTable(tablePath, flussTemporalType, icebergTemporalType);
+
+ GenericRow insertRow = temporalIdentifierRow(flussTemporalValue, "before");
+ GenericRow updateRow = temporalIdentifierRow(flussTemporalValue, "after");
+
+ IcebergWriteResult writeResult;
+ try (LakeWriter<IcebergWriteResult> writer =
+ createLakeWriter(tablePath, 0, null, null, tableInfo)) {
+ writer.write(toRecord(0L, insertRow, INSERT));
+ // The equal id makes StructLikeMap.put continue to compare the temporal identifier
+ // when UPDATE_AFTER replaces the same composite identifier key.
+ // See
+ // https://github.com/apache/iceberg/blob/apache-iceberg-1.10.1/core/src/main/java/org/apache/iceberg/io/BaseTaskWriter.java#L153-L162
+ writer.write(toRecord(1L, updateRow, UPDATE_AFTER));
+ writeResult = writer.complete();
+ }
+ try (LakeCommitter<IcebergWriteResult, IcebergCommittable> committer =
+ createLakeCommitter(tablePath, tableInfo)) {
+ committer.commit(
+ committer.toCommittable(Collections.singletonList(writeResult)),
+ Collections.emptyMap());
+ }
+
+ Table table = icebergCatalog.loadTable(toIceberg(tablePath));
+ table.refresh();
+ try (CloseableIterator<Record> rows = IcebergGenerics.read(table).build().iterator()) {
+ assertThat(rows.hasNext()).isTrue();
+ Record actual = rows.next();
+ assertThat(actual.getField("id")).isEqualTo(1);
+ assertThat(actual.getField("temporal_identifier")).isEqualTo(expectedIcebergValue);
+ assertThat(actual.getField("payload")).isEqualTo("after");
+ assertThat(rows.hasNext()).isFalse();
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("temporalIdentifierArgs")
+ void testTieringDeleteTemporalIdentifier(
Review Comment:
I add a test param to cover different operations in one method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Fix temporal identifier keys in tiering writer [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #4451:
URL: https://github.com/apache/fluss/pull/4451
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org