You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "ZuebeyirEser (via GitHub)" <gi...@apache.org> on 2026/01/13 21:29:11 UTC
[PR] [lake] Add IT case for union read with evolved schema in Paimon tiering [fluss]
ZuebeyirEser opened a new pull request, #2368:
URL: https://github.com/apache/fluss/pull/2368
### Purpose
Linked issue: close #2363
`PaimonTieringITCase` currently tests that data can be tiered after a schema change, but it doesn't cover the "Union Read" path. This PR adds a test to ensure that Flink SQL correctly merges records from both the lake (old schema) and the log (new schema) after an `ADD COLUMN` operation.
### Brief change log
- Initialized `StreamTableEnvironment` and registered the Fluss catalog in `PaimonTieringITCase`.
- Added `testUnionReadWithAddColumn` to verify that old records in Paimon are projected with NULLs and new records in the log are projected correctly during a union read.
### Tests
- Added `org.apache.fluss.lake.paimon.tiering.PaimonTieringITCase#testUnionReadWithAddColumn`.
- Verified with: `mvn test -pl fluss-lake/fluss-lake-paimon -Dtest=PaimonTieringITCase#testUnionReadWithAddColumn`.
### API and Format
N/A
### Documentation
N/A
--
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] Add IT case for union read with evolved schema in Paimon tiering [fluss]
Posted by "ZuebeyirEser (via GitHub)" <gi...@apache.org>.
ZuebeyirEser commented on code in PR #2368:
URL: https://github.com/apache/fluss/pull/2368#discussion_r2689820406
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java:
##########
@@ -618,6 +634,53 @@ void testTieringWithAddColumn() throws Exception {
}
}
+ @Test
+ void testUnionReadWithAddColumn() throws Exception {
Review Comment:
Done. I moved the `testUnionReadWithAddColumn` to `FlinkUnionReadPrimaryKeyTableITCase` and updated the logic to include a Primary Key UPDATE operation to ensure the union read handles upserts correctly.
--
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] Add IT case for union read with evolved schema in Paimon tiering [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #2368:
URL: https://github.com/apache/fluss/pull/2368#discussion_r2688687485
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringITCase.java:
##########
@@ -618,6 +634,53 @@ void testTieringWithAddColumn() throws Exception {
}
}
+ @Test
+ void testUnionReadWithAddColumn() throws Exception {
Review Comment:
Hi, could you please move it to `org.apache.fluss.lake.paimon.flink.FlinkUnionReadPrimaryKeyTableITCase`, we alway test union read in `FlinkUnionReadPrimaryKeyTableITCase`
Also, I'd like to test primary key table since primary key table cover more code path.
--
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] Add IT case for union read with evolved schema in Paimon tiering [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #2368:
URL: https://github.com/apache/fluss/pull/2368#discussion_r2690167915
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java:
##########
@@ -845,6 +846,74 @@ void testUnionReadPrimaryKeyTableFailover(boolean isPartitioned) throws Exceptio
jobClient.cancel().get();
}
+ @Test
+ void testUnionReadWithAddColumn() throws Exception {
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "unionReadAddColumnPKTable");
+
+ // 1. Create PK Table (Lake Enabled)
+ Schema schema =
+ Schema.newBuilder()
+ .column("c1", DataTypes.INT())
+ .column("c2", DataTypes.STRING())
+ .primaryKey("c1")
+ .build();
+
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(1)
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500))
+ .build();
+
+ long tableId = createTable(tablePath, tableDescriptor);
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+
+ // 2. Write initial data
+ List<InternalRow> initialRows = Arrays.asList(row(1, "v1"), row(2, "v2"));
+ writeRows(tablePath, initialRows, false);
+
+ // 3. Start tiering job
+ JobClient jobClient = buildTieringJob(execEnv);
+
+ try {
+ // 4. Wait for data to snapshot to Paimon
+ assertReplicaStatus(tableBucket, 2);
+
+ // 5. Add Column "c3"
+ List<TableChange> addColumnChanges =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "c3",
+ DataTypes.INT(),
+ "new column",
+ TableChange.ColumnPosition.last()));
+ admin.alterTable(tablePath, addColumnChanges, false).get();
+
+ // 6. Write new data (Update Key 2, Insert Key 3)
+ // Updating key 2 validates that union read correctly merges
+ // the new schema data from log with old schema data from Paimon
+ List<InternalRow> newRows = Arrays.asList(row(2, "v2_updated", 20), row(3, "v3", 30));
+ writeRows(tablePath, newRows, false);
+
+ // 7. Query via Flink SQL
+ CloseableIterator<Row> iterator =
+ batchTEnv.executeSql("SELECT * FROM " + tablePath.getTableName()).collect();
+
+ // 8. Verify union read correctly handles schema evolution with PK updates:
+ // - Key 1: from Paimon snapshot (old schema, c3 should be null)
+ // - Key 2: from Fluss log (updated value, new schema)
+ // - Key 3: from Fluss log (new insert, new schema)
+ List<String> actualRows = collectRowsWithTimeout(iterator, 3, true);
+
+ assertThat(actualRows)
+ .containsExactlyInAnyOrder(
+ "+I[1, v1, null]", "+I[2, v2_updated, 20]", "+I[3, v3, 30]");
Review Comment:
Then, could you please continus to test the case that merge fluss change log with different schema id should also work? We can do like this which I have already verified.
```
// stop tiering to make sure we can merge change log with different schema
jobClient.cancel().get();
addColumnChanges =
Collections.singletonList(
TableChange.addColumn(
"c4",
DataTypes.INT(),
"new column",
TableChange.ColumnPosition.last()));
admin.alterTable(tablePath, addColumnChanges, false).get();
newRows =
Arrays.asList(row(2, "v2_updated_again", 20, 30), row(3, "v3_update", 30, 40));
writeRows(tablePath, newRows, false);
iterator =
batchTEnv.executeSql("SELECT * FROM " + tablePath.getTableName()).collect();
actualRows = collectRowsWithTimeout(iterator, 3, true);
assertThat(actualRows).containsExactlyInAnyOrder(
"+I[1, v1, null, null]",
"+I[2, v2_updated_again, 20, 30]",
"+I[3, v3_update, 30, 40]"
);
```
--
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] Add IT case for union read with evolved schema in Paimon tiering [fluss]
Posted by "ZuebeyirEser (via GitHub)" <gi...@apache.org>.
ZuebeyirEser commented on code in PR #2368:
URL: https://github.com/apache/fluss/pull/2368#discussion_r2690259726
##########
fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadPrimaryKeyTableITCase.java:
##########
@@ -845,6 +846,74 @@ void testUnionReadPrimaryKeyTableFailover(boolean isPartitioned) throws Exceptio
jobClient.cancel().get();
}
+ @Test
+ void testUnionReadWithAddColumn() throws Exception {
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "unionReadAddColumnPKTable");
+
+ // 1. Create PK Table (Lake Enabled)
+ Schema schema =
+ Schema.newBuilder()
+ .column("c1", DataTypes.INT())
+ .column("c2", DataTypes.STRING())
+ .primaryKey("c1")
+ .build();
+
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(1)
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500))
+ .build();
+
+ long tableId = createTable(tablePath, tableDescriptor);
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+
+ // 2. Write initial data
+ List<InternalRow> initialRows = Arrays.asList(row(1, "v1"), row(2, "v2"));
+ writeRows(tablePath, initialRows, false);
+
+ // 3. Start tiering job
+ JobClient jobClient = buildTieringJob(execEnv);
+
+ try {
+ // 4. Wait for data to snapshot to Paimon
+ assertReplicaStatus(tableBucket, 2);
+
+ // 5. Add Column "c3"
+ List<TableChange> addColumnChanges =
+ Collections.singletonList(
+ TableChange.addColumn(
+ "c3",
+ DataTypes.INT(),
+ "new column",
+ TableChange.ColumnPosition.last()));
+ admin.alterTable(tablePath, addColumnChanges, false).get();
+
+ // 6. Write new data (Update Key 2, Insert Key 3)
+ // Updating key 2 validates that union read correctly merges
+ // the new schema data from log with old schema data from Paimon
+ List<InternalRow> newRows = Arrays.asList(row(2, "v2_updated", 20), row(3, "v3", 30));
+ writeRows(tablePath, newRows, false);
+
+ // 7. Query via Flink SQL
+ CloseableIterator<Row> iterator =
+ batchTEnv.executeSql("SELECT * FROM " + tablePath.getTableName()).collect();
+
+ // 8. Verify union read correctly handles schema evolution with PK updates:
+ // - Key 1: from Paimon snapshot (old schema, c3 should be null)
+ // - Key 2: from Fluss log (updated value, new schema)
+ // - Key 3: from Fluss log (new insert, new schema)
+ List<String> actualRows = collectRowsWithTimeout(iterator, 3, true);
+
+ assertThat(actualRows)
+ .containsExactlyInAnyOrder(
+ "+I[1, v1, null]", "+I[2, v2_updated, 20]", "+I[3, v3, 30]");
Review Comment:
Done. I've expanded the test for multiple schema.
--
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] Add IT case for union read with evolved schema in Paimon tiering [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #2368:
URL: https://github.com/apache/fluss/pull/2368
--
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