feat: refreshable source, SourceExec part #22620

xxchan · 2025-07-16T02:29:20Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Second part of #22690. See the tracking issue for the whole picture of the user journey.

Source part.

Key changes include:

Added a new BatchSourceSplit interface in connector, and use it in SourceExec.
Added a new batch_posix_fs connector for testing purporse
Added a mechanism to trigger LoadFinish barrier: Similar to MV creation progress reporting, SourceExec will report load finish to barrier worker, and reported to meta as part of BarrierCompleteResponse. When handling this, meta will trigger a LoadFinish barrier.

Checklist

I have written necessary rustdoc comments.
I have added necessary unit tests and integration tests.
I have added test labels as necessary.
I have added fuzzing tests or opened an issue to track them.
My PR contains breaking changes.
My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
I have checked the Release Timeline and Currently Supported Versions to determine which release branches I need to cherry-pick this PR into.

Documentation

My PR needs documentation updates.

xxchan · 2025-07-16T02:29:35Z

feat: refreshable source, SourceExec part #22620 : 2 dependent PRs (#22621 , #22733 ) 👈 (View in Graphite)
feat(frontend): add refreshable property & refresh command #22619
feat: minor improve risedev d in cursorrule #22689
main

This stack of pull requests is managed by Graphite. Learn more about stacking.

cursor

Bug: StreamChunk Filtering Blocks Batch Source Completion

The SourceExecutor incorrectly filters out empty chunks (cardinality = 0). This prevents the StreamChunk::default() signal, which batch sources use to indicate completion, from propagating downstream, thereby preventing proper reporting of load completion for batch sources.

src/stream/src/executor/source/source_executor.rs#L810-L814

risingwave/src/stream/src/executor/source/source_executor.rs

Lines 810 to 814 in 3707c1b

    
               prune_additional_cols(&chunk, split_idx, offset_idx, &source_desc.columns); 
        
           if card > 0 { 
        
               yield Message::Chunk(chunk); 
        
           } 
        
           self.try_flush_data().await?;

Fix in Cursor • Fix in Web

Bugbot free trial expires on July 29, 2025
Learn more in the Cursor dashboard.

Was this report helpful? Give feedback by reacting with 👍 or 👎

Copilot

Pull Request Overview

This PR implements the SourceExec part of the refreshable source feature, which allows batch sources to be refreshed with new data. The implementation focuses on adding mechanisms to track and report when batch sources have finished loading data.

Added BatchSourceSplit trait and implementation for batch posix fs connector as a testing framework
Introduced load finish reporting mechanism through barrier system to notify when batch sources complete data loading
Added refresh functionality to trigger re-enumeration of splits when refresh commands are received

Reviewed Changes

Copilot reviewed 32 out of 32 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
src/stream/src/task/barrier_worker/mod.rs	Added load_finished_source_ids field to BarrierCompleteResult for tracking completed batch sources
src/stream/src/task/barrier_worker/managed_state.rs	Enhanced barrier state management to handle source load finished events and track them per epoch
src/stream/src/task/barrier_manager/mod.rs	Added ReportSourceLoadFinished event and methods for reporting load completion
src/stream/src/executor/source/source_executor.rs	Implemented batch source detection, refresh logic, and load finish reporting in SourceExecutor
src/stream/src/executor/source/reader_stream.rs	Added logic to mark batch splits as finished when stream consumption completes
src/stream/src/executor/source/executor_core.rs	Added batch source detection and batch split retrieval methods to StreamSourceCore
src/meta/src/barrier/worker.rs	Updated GlobalBarrierWorker to include barrier scheduler for handling load finish commands
src/meta/src/barrier/context/context_impl.rs	Implemented handling of load finished source IDs by scheduling LoadFinish commands
src/meta/src/barrier/complete_task.rs	Added load_finished_source_ids field to CompleteBarrierTask for async handling
src/connector/src/source/filesystem/opendal_source/batch_posix_fs_source.rs	New batch posix fs connector implementation for testing refreshable sources
src/connector/src/source/batch.rs	Added BatchSourceSplit trait and BatchSourceSplitImpl enum for batch source abstractions

src/connector/with_options_source.yaml

src/frontend/src/optimizer/mod.rs

src/connector/src/with_options.rs

src/connector/src/source/filesystem/opendal_source/batch_posix_fs_source.rs

src/stream/src/executor/source/source_executor.rs

src/stream/src/task/barrier_worker/managed_state.rs

chenzl25 · 2025-07-25T09:51:45Z

src/connector/src/with_options.rs

+        // TODO: enable this when implementation done
+        // self.get(UPSTREAM_SOURCE_KEY)
+        //     .map(|s| s.eq_ignore_ascii_case(BATCH_POSIX_FS_CONNECTOR))
+        //     .unwrap_or(false)


Is it done?

We enable this after the whole feature is done, i.e., in #22695 (which has e2e test)

chenzl25 · 2025-07-25T10:54:04Z

src/meta/src/barrier/context/context_impl.rs

+
+        // handle_load_finished_source_ids is called when collecting barrier,
+        // so we need to spawn a new task to yield a LoadFinish barrier to avoid deadlock.
+        tokio::spawn(async move {


@wenym1 Is it ok to inject a command within collect_barrier like this(using a spawn task)?

Since we spawn a task here without awaiting on it, I guess here we only want to schedule a command and don't need to wait for the command to be executed? If so, it will be more straightforward to just enqueue the command to BarrierScheduler without waiting on the notifier. Such operation does not even require async context, because the queue is protected by a sync lock.

We can add a new method in BarrierScheduler with name like run_command_no_wait, which calls self.push with the command and an empty notifier, and then return immediately.

src/connector/src/source/filesystem/opendal_source/batch_posix_fs_source.rs

BugenZhao

6

BugenZhao · 2025-07-28T07:15:00Z

src/stream/src/executor/source/executor_core.rs

    ) -> Self {
+        let is_batch_source = source_desc_builder
+            .with_properties()
+            .is_refreshable_connector();


Can we try unifying the terminology (batch vs refreshable)?

which one do you prefer?

Or just "refreshable batch source"?

Or "batch source" + "refreshable table"?

BugenZhao · 2025-07-28T07:17:28Z

src/stream/src/executor/source/reader_stream.rs

                        break 'consume;
                    }
                }
            }
+            if !is_error {


Is there any other cases where the connector read stream can be finite (that is, finishes without error)?

Actually, for normal connector e.g., datagen, they are finite. Before this change, it will be retried indefinitely.

src/stream/src/executor/source/source_executor.rs

graphite-app · 2025-07-28T08:28:01Z

Looks like this PR extends new SQL syntax or updates existing ones. Make sure that:

Test cases about the new/updated syntax are added in src/sqlparser/tests/testdata. Especially, double check the formatted_sql is still a valid SQL #20713
The meaning of each enum variant is documented in PR description. Additionally, document what it means when each optional clause is omitted.

chenzl25 · 2025-07-29T16:05:06Z

src/connector/src/source/base.rs

+    pub fn into_batch_split(self) -> Option<BatchSourceSplitImpl> {
+        #[expect(clippy::match_single_binding)]
+        match self {
+            // SplitImpl::BatchPosixFs(batch_posix_fs_split) => {
+            //     Some(BatchSourceSplitImpl::BatchPosixFs(batch_posix_fs_split))
+            // }
+            _ => None,
+        }
+    }


Comment here because we separate the batch posix file source PR, right?

chenzl25 · 2025-07-29T16:12:00Z

src/stream/src/executor/source/reader_stream.rs

+                    StreamChunk::empty(
+                        self.source_desc
+                            .columns
+                            .iter()
+                            .map(|c| c.data_type.clone())
+                            .collect_vec()
+                            .as_slice(),
+                    ),
+                    latest_splits_info.clone(),
+                );
+                break 'build_consume_loop;


Why do we need to send an empty chunk?

The signature of the stream is StreamChunkWithState, i.e., (StreamChunk, HashMap<SplitId, SplitImpl>).

What I want to send is the latest_splits_info here (Then in the main loop of SourceExecutor, it can check batch_split.finished()). The least invasive change I can come up with is to pair an empty chunk here.. 🤪

chenzl25

LGTM

xxchan

@BugenZhao would you like to take another look?

Signed-off-by: xxchan <xxchan22f@gmail.com>

tabVersion

Let me confirm the behavior:

upon source receives the RefreshBatchSplits mutation, it converts the current split assignment to batch_split, then rebuild the reader to start with state in batch_split. (but lacks how to define a finite stream)
In the re-snapshot process, the intermediate state is also written to state table, and can resume if recover happens
report to local barrier manager if re-snapshot is done. How we switch back to normal splits?

tabVersion · 2025-07-30T08:21:06Z

src/stream/src/executor/source/executor_core.rs

+        debug_assert_eq!(
+            self.latest_split_info.len(),
+            1,
+            "batch source should have only one split"
+        );


why the statement holds? I dont see how we define refreshable-connector and I don't see each connector parallelism is assigned to exactly one split.

tabVersion · 2025-07-30T08:34:41Z

src/stream/src/executor/source/reader_stream.rs

+                            .collect_vec()
+                            .as_slice(),
+                    ),
+                    latest_splits_info.clone(),


will this state written to state table?

github-actions bot added the type/feature Type: New feature. label Jul 16, 2025

This was referenced Jul 16, 2025

feat(frontend): add refreshable property & refresh command #22619

Merged

feat(stream): support RefreshableMaterialize with staging table #22621

Closed

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from e00eb0f to b3d8058 Compare July 22, 2025 05:37

xxchan force-pushed the xxchan/protective-jackal branch from 75e6f3a to 09dedc4 Compare July 22, 2025 05:37

github-actions bot added ci/run-s3-source-tests ci/run-e2e-single-node-tests ci/main-cron/run-selected ci/run-e2e-test-other-backends labels Jul 22, 2025

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from b3d8058 to 0a9032a Compare July 22, 2025 06:27

xxchan force-pushed the xxchan/protective-jackal branch from 09dedc4 to 5b518e5 Compare July 22, 2025 06:42

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch 3 times, most recently from 6743e88 to e0f1777 Compare July 22, 2025 08:35

xxchan force-pushed the xxchan/protective-jackal branch from 5b518e5 to a409585 Compare July 22, 2025 08:35

xxchan mentioned this pull request Jul 22, 2025

feat: minor improve risedev d in cursorrule #22689

Merged

8 tasks

xxchan force-pushed the xxchan/protective-jackal branch from a409585 to 6f04b2b Compare July 22, 2025 10:07

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from e0f1777 to de09f3b Compare July 22, 2025 10:07

xxchan mentioned this pull request Jul 22, 2025

feat: refreshable materialize with staging table #22695

Open

10 tasks

xxchan marked this pull request as ready for review July 23, 2025 02:17

xxchan force-pushed the xxchan/protective-jackal branch from 6f04b2b to ca59314 Compare July 23, 2025 02:19

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from de09f3b to 3707c1b Compare July 23, 2025 02:19

cursor bot reviewed Jul 23, 2025

View reviewed changes

xxchan force-pushed the xxchan/protective-jackal branch from ca59314 to e06c1e8 Compare July 23, 2025 02:40

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from 3707c1b to 293dd5e Compare July 23, 2025 02:40

xxchan force-pushed the xxchan/protective-jackal branch from e06c1e8 to 553fc73 Compare July 23, 2025 02:56

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from 293dd5e to 9969797 Compare July 23, 2025 02:56

xxchan force-pushed the xxchan/protective-jackal branch from 553fc73 to 23ca770 Compare July 23, 2025 05:19

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from 9969797 to 462614d Compare July 23, 2025 05:19

xxchan requested review from Copilot and tabVersion July 25, 2025 07:33

Copilot AI reviewed Jul 25, 2025

View reviewed changes

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from 894d2ee to bfd6233 Compare July 25, 2025 08:15

chenzl25 reviewed Jul 25, 2025

View reviewed changes

BugenZhao reviewed Jul 28, 2025

View reviewed changes

src/connector/src/source/filesystem/opendal_source/batch_posix_fs_source.rs Outdated Show resolved Hide resolved

BugenZhao reviewed Jul 28, 2025

View reviewed changes

src/connector/src/source/filesystem/opendal_source/batch_posix_fs_source.rs Outdated Show resolved Hide resolved

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from bfd6233 to 47f55da Compare July 28, 2025 07:11

BugenZhao reviewed Jul 28, 2025

View reviewed changes

Base automatically changed from xxchan/protective-jackal to main July 28, 2025 08:27

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from 47f55da to f1e5606 Compare July 28, 2025 08:41

xxchan mentioned this pull request Jul 28, 2025

feat: batch posix fs connector #22733

Open

8 tasks

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from f1e5606 to 82f268b Compare July 28, 2025 09:01

xxchan requested a review from BugenZhao July 28, 2025 09:01

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch 5 times, most recently from 92a451a to be94be6 Compare July 29, 2025 15:39

xxchan mentioned this pull request Jul 29, 2025

feat: support refreshable bigquery source #22744

Draft

9 tasks

chenzl25 reviewed Jul 29, 2025

View reviewed changes

chenzl25 approved these changes Jul 30, 2025

View reviewed changes

xxchan commented Jul 30, 2025

View reviewed changes

xxchan added 2 commits July 30, 2025 15:27

feat: refreshable table, source part

c8281b0

Signed-off-by: xxchan <xxchan22f@gmail.com>

run_command_no_wait

0266dcb

xxchan force-pushed the 07-14-feat_refreshable_source_sourceexec_part branch from be94be6 to 0266dcb Compare July 30, 2025 07:32

disallow create source with refreshable source

e76e029

tabVersion reviewed Jul 30, 2025

View reviewed changes

	prune_additional_cols(&chunk, split_idx, offset_idx, &source_desc.columns);
	if card > 0 {
	yield Message::Chunk(chunk);
	}
	self.try_flush_data().await?;

feat: refreshable source, SourceExec part #22620

Are you sure you want to change the base?

feat: refreshable source, SourceExec part #22620

Conversation

xxchan commented Jul 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What's changed and what's your intention?

Checklist

Documentation

Uh oh!

xxchan commented Jul 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cursor bot left a comment

Choose a reason for hiding this comment

Bug: StreamChunk Filtering Blocks Batch Source Completion

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chenzl25 Jul 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

BugenZhao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

graphite-app bot commented Jul 28, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chenzl25 left a comment

Choose a reason for hiding this comment

Uh oh!

xxchan left a comment

Choose a reason for hiding this comment

Uh oh!

tabVersion left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

xxchan commented Jul 16, 2025 •

edited

Loading

xxchan commented Jul 16, 2025 •

edited

Loading

chenzl25 Jul 25, 2025 •

edited

Loading