JNSimba opened a new issue, #682:
URL: https://github.com/apache/doris-flink-connector/issues/682
## Background
Currently, Doris Source mainly supports bounded snapshot reads and cannot
continuously consume inserts, updates, and deletes after the snapshot is
completed.
This feature adds incremental reading support to Doris Source, covering
snapshot-only, incremental-only, and snapshot-then-incremental scenarios.
## How it works
In `initial` mode, the Connector records the current Doris timestamp `T0`
before reading any snapshot split. After all snapshot splits are completed and
the state is included in a successful checkpoint, the Source switches to the
incremental phase.
The incremental phase continuously reads fixed time ranges through Doris
`table@incr`:
```text
(T0, T1] -> (T1, T2] -> (T2, T3]
```
The TSO, LSN, and OP hidden columns returned by Doris are used to preserve
change order and convert records into Flink `INSERT`, `DELETE`,
`UPDATE_BEFORE`, and `UPDATE_AFTER` events.
```mermaid
flowchart LR
A[Start Source] --> B{Scan mode}
B -->|snapshot| C[Read snapshot]
B -->|initial| D[Resolve Doris timestamp T0]
D --> E[Read snapshot]
E --> F[Wait for a successful checkpoint]
B -->|latest| G[Resolve current Doris timestamp]
B -->|from-timestamp| H[Use configured timestamp]
F --> I[Enter incremental phase]
G --> I
H --> I
I --> J[Create a fixed time-range split]
J --> K[Read changes through table@incr]
K --> L[Advance the next start timestamp]
L --> J
C --> M[Finish]
```
## Main changes
- Add `snapshot`, `initial`, `latest`, and `from-timestamp` scan modes.
- Introduce unified coordination and checkpoint state for snapshot and
stream splits.
- Generate continuous incremental time ranges based on the Doris TSO.
- Read Doris row binlog through Arrow Flight SQL and `table@incr`.
- Convert Doris row binlog operations into Flink RowKind.
- Support both Flink 1.x and Flink 2.x Table API.
## Current limitations
- The first version provides at-least-once semantics. Duplicate records may
be produced during failure recovery or the snapshot-to-stream transition.
- Only one stream split can run at a time. Increasing Source parallelism
does not increase incremental read concurrency.
- Incremental modes require Arrow Flight SQL and a Doris table with ROW
binlog enabled.
## Demo
The following Flink SQL reads the snapshot of `sales.orders` first and then
continuously reads incremental changes:
```sql
CREATE TABLE orders_source (
order_id BIGINT,
order_status STRING,
update_time TIMESTAMP(3),
PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
'connector' = 'doris',
'fenodes' = '127.0.0.1:8030',
'username' = 'root',
'password' = '',
'table.identifier' = 'sales.orders',
'source.use-flight-sql' = 'true',
'source.scan.mode' = 'initial',
'source.binlog.increment-type' = 'detail',
'source.binlog.poll-interval' = '10s'
);
SELECT * FROM orders_source;
```
Related implementation: #679
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
