apache / apache/doris-flink-connector

[Feature] Support incremental reads from Doris Source

Open
#682 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
385
Forks
279
Avg merge
2d 2h
Merged PRs (30d)
6

Description

## 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

- 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.scan.mode' = 'initial',
'source.binlog.poll-interval' = '10s'
);

SELECT * FROM orders_source;
```

Contributor guide

Open the contributing guide

Research direction

Start by tracing the Doris Source scan modes and the Arrow Flight SQL path described in the issue. Review how `table@incr`, TSO/LSN/OP columns, checkpoint state, and Flink RowKind conversion fit together. Done means snapshot-only, incremental-only, and snapshot-then-incremental modes work across the documented timestamp ranges and recovery scenarios.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, sql
Domain
data-engineering, databases
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.