feat(source): recoverable bounded sources and generic hybrid source composition

Motivation

Introduce connector-independent support for bounded sources and hybrid sources. A pipeline should be able to consume finite input and then continue with a streaming source while retaining the same downstream tables and materialized views.

Bounded input can be the result of a PostgreSQL/MySQL query, a database/table snapshot, a finite export, an object/file collection, or a bounded range of a log. Streaming input can be a message system, a CDC/change feed, or another continuous connector. Iceberg/S3 followed by Kafka is one motivating example. Connectors should participate through explicit capabilities and compatibility checks.

This requires two related capabilities: bounded ingestion on its own, and ordered composition of compatible sources. Users should not need to coordinate separate import and streaming jobs or rebuild downstream state at cutover. The streaming system may keep producing changes while the bounded stage runs.

Recovery during bounded ingestion is a core requirement, including failures partway through a PostgreSQL/MySQL query. It must work for standalone bounded input and for the bounded stage of a hybrid pipeline.

Current RisingWave foundations

Reviewed against main at fab4919c.

• We already support finite snapshot/file reads through refreshable tables. Iceberg supports continuous append-only ingestion and CREATE TABLE ... refresh_mode = 'FULL_RELOAD' with manual or scheduled refreshes. This provides a starting point for bounded ingestion. (Iceberg documentation, frontend restrictions)
• The Iceberg and OpenDAL batch source executors distinguish listing completion from fetching completion and report completion through checkpoint barriers. Meta aggregates progress across the expected actors. These mechanisms are currently tied to the refresh lifecycle. (Iceberg fetch, OpenDAL fetch, refresh tracking)
• Kafka already has finite offset-range enumeration for batch reads; source backfill also tracks progress toward upstream offsets. These illustrate reusable bounded-read and catch-up primitives. A generic cross-connector handoff still needs an explicit contract. (Kafka enumeration, source backfill background)
• PostgreSQL/MySQL CDC readers already perform key-ordered table scans, and CDC backfill checkpoints scan progress together with CDC offsets. These are foundations for investigation; generic bounded SQL queries need their own result/replay contract. (PostgreSQL scan, MySQL scan, CDC checkpoint state)
• The shared connector interfaces expose split enumeration, data, and split progress, but no general contract for boundedness, globally completed input, or switching to another connector. (Connector interfaces, reader events)

The gap is a reusable bounded-source lifecycle and a durable cross-source handoff, building on these existing capabilities. The paths above are implementation foundations; the shared abstractions should accommodate other connectors and source kinds.

Flink reference

Flink's DataStream HybridSource composes heterogeneous sources sequentially as one logical input. Every stage except the last must be bounded. The next source can use a predefined start position or be constructed from the preceding enumerator's end-position metadata. This still requires connector-specific boundary information. (Hybrid Source documentation)

Its coordinator waits for readers to finish a stage, and recovery accounts for the source index and splits from previous stages. This is useful precedent for coordination and recovery; consistency between two representations of the same logical dataset still needs a source-specific boundary contract. (FLIP-150, enumerator implementation)

Proposed behavior

1. Standalone bounded input: consume a finite dataset/range once and expose durable completion. Boundedness is a property of the configured read: a connector may offer both bounded and continuous modes. It is separate from append-only/upsert semantics. Completion must not terminate unrelated streaming work; resulting tables/MVs remain queryable.
2. Hybrid input: compose compatible sources as ordered stages feeding the same logical input and downstream state. All non-final stages must be bounded. The final stage may be bounded or continuous, determining the composite input's boundedness. A later stage starts emitting only after its predecessor completes; downstream state survives the transition.
3. Connector capabilities: declare boundedness, finite-input completion, checkpoint/resume support, row semantics, and supported handoff metadata. The coordinator handles stage order and recovery; connector adapters interpret their own positions and boundaries. Adding a connector should not require a new switching protocol for each connector pair.
4. Operational visibility: expose the current stage, input identity/bounds, discovered and completed work, completion checkpoint, and connector-specific starting/progress positions. Pause, cancellation, backpressure, and recovery must work across stages.

Conceptually:

bounded source A -> [bounded source B -> ...] -> final source (bounded or streaming)
                  durable transitions; shared downstream state

Correctness contract

• Stable bounded input. Persist a reproducible read definition and its bounds: for example a snapshot/version, immutable file manifest, or fixed log range. A mutable query/export needs a consistent-read and recovery contract. Recovery must resume the original input. Honor its logical semantics, such as Iceberg delete files, and report missing/expired input clearly.
• Explicit handoff contract. Use a general descriptor for source identity, the completed input's coverage, and the next source's starting position. Adapters may use snapshot IDs, partition offsets, log positions, or other resumable cursors; the coordinator must not assume Kafka-style offsets. Support predefined boundaries and adapter-derived boundaries where a trustworthy mapping exists.
• State the continuity guarantee. Ordered concatenation of compatible inputs does not itself prove gap-free, non-overlapping coverage of one logical dataset. Snapshot-to-change-feed bootstrap requires an exporter/connector/user contract that establishes this relationship. For example, an export covering a Kafka partition through offset 99 may hand off at offset 100; other connectors use their own position semantics. Starting at the next source's current head after bootstrap can lose intervening changes. Overlap and timestamp-based policies need defined reconciliation or an explicit weaker guarantee.
• Defined row semantics. Event history followed by new events and table state followed by CDC/upserts are different use cases. The latter requires compatible schemas, stable keys, and defined update/delete normalization. Validate the capability/compatibility matrix and reject unsupported combinations; primary-key deduplication alone does not make every downstream computation correct.
• Global, durable completion. Enumeration must be sealed and all assigned work consumed, including empty inputs and idle readers. A poll timeout, one reader reaching EOF, or a completed listing alone is insufficient. Checkpoint source progress, stage identity, handoff metadata, and downstream state consistently so recovery cannot skip data or double-apply it. Fence stale completion reports after recovery/rescaling.
• Replayability and watermarks. The next source's required data must remain available throughout bootstrap and recovery, through retention or durable capture. A non-replayable source needs an explicit capture mechanism or a weaker supported guarantee. Handle expired cursors and changes to source identity/topology explicitly. Intermediate stages must preserve valid watermark progression across the handoff and avoid prematurely closing downstream windows.

Bounded query recovery

For PostgreSQL/MySQL queries, a durable checkpoint must identify the query and parameters, schema, consistent input/result version, split boundaries, consumed positions, and completed splits. These must agree with the downstream state recorded by that checkpoint. Recovery must continue unfinished work without double-applying committed rows; a durably completed query must remain completed after restart.

A saved cursor or last primary key is insufficient if reconnecting executes the query against changed data. PostgreSQL exported snapshots remain importable only while the exporting transaction is open, and MySQL InnoDB repeatable-read snapshots belong to a transaction. These mechanisms alone do not establish durable query recovery after losing that transaction. (PostgreSQL snapshot lifetime, MySQL consistent reads)

The design must choose a recovery strategy for supported query modes: a read version that can be reopened after failure, durable materialization of the query result before downstream consumption, or another mechanism with equivalent guarantees. If materializing, define recovery during materialization and atomically publish the completed result; incomplete attempts must never be mixed. Any new query attempt and its hybrid handoff metadata must refer to the same input generation.

Define deterministic resume/replay for each supported query shape. A keyset scan needs a stable, unique ordering over the same logical input; arbitrary queries, duplicate result rows, and queries without a suitable key may need durable result chunks. Re-running SQL with OFFSET n alone does not provide this contract. Test loss of the reader/connection during the bounded stage, not just failure at the switch to streaming.

Suggested implementation scope

First introduce shared boundedness, completion, and resumability contracts. Then implement a generic hybrid coordinator and connector adapters for input discovery, reading, and handoff metadata. Existing refresh and backfill paths are candidates for reuse. As an initial RW design, consider activating the next stage after the checkpoint containing predecessor completion commits, with enough persisted transition state to recover on either side of that commit.

The first implementation can deliver standalone bounded reads and a two-stage hybrid, while keeping the contracts extensible to additional stages and connectors. Choose representative adapters based on capabilities and demand; Iceberg/S3/Kafka are examples. The public API, catalog model, and coordinator should express generic stages and positions. Define an explicit support matrix rather than assuming every connector combination is valid.

Final SQL syntax, the initial connector/row-semantics matrix, and optional prefetch/capture are design decisions. Existing continuous ingestion and REFRESH TABLE semantics should remain compatible. Manual import plus streaming startup leaves handoff/recovery coordination to users; UNION ALL alone does not provide ordered stage transitions.

Acceptance criteria

• Supported bounded connectors finish durably, including empty inputs; recovery preserves the selected dataset/range despite concurrent external changes.
• PostgreSQL/MySQL bounded queries recover from reader/connection failure partway through execution, both standalone and within a hybrid pipeline. Mutate upstream data during failure/recovery and verify consistent results without omissions or double application; cover completed-query recovery and the chosen materialization/replay strategy.
• End-to-end hybrid pipelines preserve downstream state while the streaming system receives new records during bootstrap. Exercise multiple bounded-input kinds and streaming connector kinds, including a non-Kafka streaming source.
• Adding another compatible connector uses the same lifecycle/coordinator contracts; unsupported capability and row-semantics combinations fail clearly.
• Failure injection around final bounded output, completion reporting, checkpoint commit, and first next-stage output verifies the declared recovery guarantee. For gap-free bootstrap modes, check non-idempotent COUNT/SUM results and supported update/delete cases.
• Cover reader skew, empty work assignments, rescaling, invalid/expired positions, missing input, all-bounded composition, and stage-boundary watermark behavior.
• Document the capability/support matrix, connector-specific boundary and cursor semantics, availability assumptions, consistency guarantees, and observable completion status.