You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "naivedogger (via GitHub)" <gi...@apache.org> on 2026/08/10 07:54:31 UTC
[PR] [flink] Support batch scan for virtual log tables [fluss]
naivedogger opened a new pull request, #3925:
URL: https://github.com/apache/fluss/pull/3925
### Purpose
Linked issue: close #3723
Support bounded batch scans for Fluss `$changelog` and `$binlog` virtual tables.
This enables historical changelog replay, incremental export, audit queries, data backfill, and log archival in Flink batch runtime.
### Brief change log
- Remove the streaming-only runtime validation for `$changelog` and `$binlog` virtual tables.
- Allow `ChangelogFlinkTableSource` and `BinlogFlinkTableSource` to be created in batch mode.
- Reuse the existing bounded Fluss log source path in batch runtime to:
- resolve the configured starting offset;
- capture the latest offset of each bucket as its stopping offset;
- terminate after all bounded log splits are consumed.
- Replace the previous batch fail-fast tests with batch query coverage for:
- full changelog and binlog replay;
- `LIMIT`;
- primary-key and nested-field filters;
- virtual-table metadata filters;
- timestamp startup mode;
- timestamp-range queries;
- partitioned changelog tables.
This PR does not change `FlinkSourceSplitReader` or introduce a separate batch reader. Filters and `LIMIT` continue to be planned and executed by Flink. No virtual-table filter or projection pushdown is introduced.
### Tests
Added or updated:
- `FlinkTableFactoryTest#testVirtualLogTableSourceSupportsBatchMode`
- `ChangelogVirtualTableITCase#testBatchReadChangelogTable`
- `BinlogVirtualTableITCase#testBatchReadBinlogTable`
Verified with:
- `Flink120TableFactoryTest`: 5 tests passed
- `Flink120ChangelogVirtualTableITCase`: 9 tests passed
- `Flink120BinlogVirtualTableITCase`: 8 tests passed
- Spotless check
- Checkstyle check
- `git diff --check`
### API and Format
No public API or storage format changes.
### Documentation
This introduces batch runtime support for the existing `$changelog` and `$binlog` virtual-table syntax. No new SQL syntax or connector options are introduced.
### Generative AI disclosure
- [ ] No generative AI tools used
- [x] Yes, Codex was used to assist with implementation and testing.
<!--
Generated-by: Codex following https://github.com/apache/fluss/blob/main/AGENTS.md
-->
--
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] [flink] Support batch scan for virtual log tables [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #3925:
URL: https://github.com/apache/fluss/pull/3925
--
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] [flink] Support batch scan for virtual log tables [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on PR #3925:
URL: https://github.com/apache/fluss/pull/3925#issuecomment-5239580675
Thanks @leonardBang for the review! I've updated the documentation accordingly. PTAL.
--
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] [flink] Support batch scan for virtual log tables [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3925:
URL: https://github.com/apache/fluss/pull/3925#discussion_r3758028178
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java:
##########
@@ -236,19 +237,128 @@ public void testBinlogUnsupportedForLogTable() throws Exception {
}
@Test
- public void testBatchReadBinlogTableFailsFast() throws Exception {
+ public void testBatchReadBinlogTable() throws Exception {
tEnv.executeSql(
"CREATE TABLE batch_binlog_test ("
+ " id INT NOT NULL,"
+ " name STRING,"
+ " PRIMARY KEY (id) NOT ENFORCED"
+ ") WITH ('bucket.num' = '1')");
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "batch_binlog_test");
+ CLOCK.advanceTime(Duration.ofMillis(1000));
+ writeRows(conn, tablePath, Arrays.asList(row(1, "Alice"), row(2, "Bob")), false);
+ CLOCK.advanceTime(Duration.ofMillis(1000));
+ writeRows(conn, tablePath, Arrays.asList(row(1, "Alice-updated")), false);
+ CLOCK.advanceTime(Duration.ofMillis(1000));
+ deleteRows(conn, tablePath, Arrays.asList(row(2, "Bob")));
+
+ tEnv = initBatchTableEnvironment();
+
+ // The batch source resolves the configured starting offsets and captures the latest log
+ // offsets as stopping offsets, so historical binlog replay completes as a bounded query.
+ List<String> allChanges =
+ collectBatchRows(
+ tEnv.executeSql(
+ "SELECT _change_type, _log_offset, "
+ + "before.id, before.name, after.id, after.name "
+ + "FROM batch_binlog_test$binlog "
+ + "ORDER BY _log_offset")
+ .collect());
Review Comment:
`collectBatchRows(...)` does not apply any timeout while waiting for results. If the bounded execution blocks, this test will hang rather than failing fast. Since the expected number of rows is known for this query, prefer `collectRowsWithTimeout(...)` (or another timeout-aware collector).
This issue also appears in the following locations of the same file:
- line 275
- line 284
- line 301
- line 311
- line 352
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java:
##########
@@ -249,19 +249,159 @@ public void testDescribeChangelogTable() throws Exception {
}
@Test
- public void testBatchReadChangelogTableFailsFast() throws Exception {
+ public void testBatchReadChangelogTable() throws Exception {
tEnv.executeSql(
"CREATE TABLE batch_changelog_test ("
+ " id INT NOT NULL,"
+ " name STRING,"
+ " PRIMARY KEY (id) NOT ENFORCED"
+ ") WITH ('bucket.num' = '1')");
+ TablePath tablePath = TablePath.of(DEFAULT_DB, "batch_changelog_test");
+ CLOCK.advanceTime(Duration.ofMillis(1000));
+ writeRows(conn, tablePath, Arrays.asList(row(1, "Alice"), row(2, "Bob")), false);
+ CLOCK.advanceTime(Duration.ofMillis(1000));
+ writeRows(conn, tablePath, Arrays.asList(row(1, "Alice-updated")), false);
+ CLOCK.advanceTime(Duration.ofMillis(1000));
+ deleteRows(conn, tablePath, Arrays.asList(row(2, "Bob")));
+
tEnv = initBatchTableEnvironment();
- assertThatThrownBy(() -> tEnv.explainSql("SELECT * FROM batch_changelog_test$changelog"))
- .hasRootCauseInstanceOf(UnsupportedOperationException.class)
- .hasRootCauseMessage("$changelog virtual tables only support streaming mode.");
+ // The default FULL startup mode maps to the earliest log offset for virtual tables. The
+ // batch source captures the latest offsets as stopping offsets and terminates after
+ // replaying the bounded changelog.
+ List<String> allChanges =
+ collectBatchRows(
+ tEnv.executeSql(
+ "SELECT _change_type, _log_offset, id, name "
+ + "FROM batch_changelog_test$changelog "
+ + "ORDER BY _log_offset")
+ .collect());
Review Comment:
`collectBatchRows(...)` iterates without any timeout, so if the bounded job stalls/regresses the test can hang indefinitely. Prefer `collectRowsWithTimeout(...)` (or another timeout-aware collector) for batch queries as well; the expected row count is known here, so a bounded wait is possible.
This issue also appears in the following locations of the same file:
- line 288
- line 302
- line 312
- line 340
- line 382
- ...and 1 more
--
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] [flink] Support batch scan for virtual log tables [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #3925:
URL: https://github.com/apache/fluss/pull/3925#issuecomment-5248134477
The failed test should be addressed before merge this PR
```
[ERROR] Flink118ChangelogVirtualTableITCase.testBatchReadChangelogTable:21->ChangelogVirtualTableITCase.testBatchReadChangelogTable:309
Expecting actual:
[]
to contain exactly (and in same order):
["+I[update_before, 2, 1, Alice]", "+I[update_after, 3, 1, Alice-updated]"]
but could not find the following elements:
["+I[update_before, 2, 1, Alice]", "+I[update_after, 3, 1, Alice-updated]"]
[INFO]
[ERROR] Tests run: 256, Failures: 2, Errors: 0, Skipped: 0
```
--
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] [flink] Support batch scan for virtual log tables [fluss]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #3925:
URL: https://github.com/apache/fluss/pull/3925#issuecomment-5248952664
Thanks for the update. I rechecked the PR against the exact merge commit used by CI (`29f785ff`, which merges this PR with the latest `main` changes), and I was able to narrow down the Flink 1.18 failure.
The timestamp assertion failure does not appear to be caused by missing log records or by the bounded source finishing too early:
- The raw `_commit_timestamp` values are correct (`1000`, `2000`, and `3000` ms).
- The exact CI merge commit reproduces both failures locally with Flink 1.18 in a non-UTC timezone.
- Both tests pass with `-Duser.timezone=UTC`.
- They also pass in a non-UTC timezone if the virtual table's `SupportsFilterPushDown` capability introduced by #3759 is temporarily removed.
- Flink 1.20 is not affected.
This suggests that Flink 1.18 reconstructs or folds the remaining `TIMESTAMP_LTZ` filter with different timezone semantics after `applyFilters()`. For example, the optimized plan represents `TO_TIMESTAMP_LTZ(2000, 3)` as `1969-12-31 16:00:02` on a UTC+8 machine, and the query returns no rows even though the raw source rows contain `1970-01-01T00:00:02Z`.
Would it make sense to address the Flink 1.18 compatibility path rather than only forcing the test timezone or changing the expected values? The new documentation also uses a `_commit_timestamp < TO_TIMESTAMP_LTZ(...)` predicate, so this can affect the documented user-facing query in non-UTC environments.
I also noticed two bounded-reader edge cases in `FlinkSourceSplitReader#forLogRecords` that become reachable after enabling batch virtual-table scans:
1. The reader forwards the entire polled record list and only uses the last record to detect completion. If records are appended after the stopping offset is captured, the same poll may contain records with `logOffset >= stoppingOffset`, and those records can be emitted even though the documented batch range is `[startingOffset, stoppingOffset)`. Could we filter records at or beyond the exclusive stopping offset before emitting them? A regression test that appends records after split initialization would be helpful.
2. Split completion currently depends on a materialized record reaching `stoppingOffset - 1`. With the `first_row` merge engine, duplicate upserts can consume WAL offsets while producing an empty record batch. In that case, `ScanRecords.consumedUpToOffset(bucket)` reaches the stopping offset, but `bucketScanRecords` is empty and the bounded split keeps polling indefinitely. The tiering reader handles the same case by completing when the exclusive `consumedUpToOffset >= stoppingOffset`. Could we use the same progress condition here and add a `first_row` duplicate-upsert regression test?
The failing core test (`UnknownTableOrBucketException` in `FlussAuthorizationITCase`) looks unrelated to this PR: the test starts while the table metadata cache on one tablet server is still being updated. That one should be safe to treat as a separate metadata-propagation test flake.
--
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