feat: support refreshable table

Author: xxchan

e2e_test/streaming/refresh_table.slt

@@ -0,0 +1,216 @@
+# Test cases for refreshable table functionality using batch_posix_fs connector
+# This test validates the REFRESH TABLE command and covers insert, update, delete scenarios
+
+control substitution on
+
+# Setup test directory and initial test files
+system ok
+rm -rf /tmp/rw_refresh_test && mkdir -p /tmp/rw_refresh_test
+
+system ok
+echo "1,alice" > /tmp/rw_refresh_test/file1.csv
+
+system ok
+echo "2,bob" > /tmp/rw_refresh_test/file2.csv
+
+# ----
+# Test 1: Create refreshable table and initial load (must have PRIMARY KEY)
+statement ok
+CREATE TABLE refresh_fs_t (id int PRIMARY KEY, name varchar) WITH (
+    connector = 'batch_posix_fs',
+    batch_posix_fs.root = '/tmp/rw_refresh_test',
+    match_pattern = '*.csv'
+) FORMAT PLAIN ENCODE CSV (without_header = 'true', delimiter = ',');
+
+sleep 2s
+
+# Initial load should read existing files
+query II rowsort retry 3 backoff 5s
+SELECT * FROM refresh_fs_t;
+----
+1 alice
+2 bob
+
+# ----
+# Test 2: Add new files and refresh (INSERT scenario)
+system ok
+echo "3,charlie" > /tmp/rw_refresh_test/file3.csv
+
+system ok
+echo "4,david" > /tmp/rw_refresh_test/new_file.csv
+
+sleep 2s
+
+# won't see new data after load finish, before next refresh
+query II rowsort
+SELECT * FROM refresh_fs_t;
+----
+1 alice
+2 bob
+
+statement ok
+REFRESH TABLE refresh_fs_t;
+
+sleep 2s
+
+
+
+
+query II rowsort retry 3 backoff 5s
+SELECT * FROM refresh_fs_t;
+----
+1 alice
+2 bob
+3 charlie
+4 david
+
+# ----
+# Test 3: Modify existing file content (UPDATE scenario)
+system ok
+echo "1,alice_updated" > /tmp/rw_refresh_test/file1.csv
+
+system ok
+echo "2,bob_updated" > /tmp/rw_refresh_test/file2.csv
+
+statement ok
+REFRESH TABLE refresh_fs_t;
+
+sleep 2s
+
+
+
+query II rowsort retry 3 backoff 5s
+SELECT * FROM refresh_fs_t;
+----
+1 alice_updated
+2 bob_updated
+3 charlie
+4 david
+
+# ----
+# Test 4: Remove files (DELETE scenario)
+system ok
+rm /tmp/rw_refresh_test/file3.csv
+
+system ok
+rm /tmp/rw_refresh_test/new_file.csv
+
+statement ok
+REFRESH TABLE refresh_fs_t;
+
+sleep 2s
+
+
+
+query II rowsort retry 3 backoff 5s
+SELECT * FROM refresh_fs_t;
+----
+1 alice_updated
+2 bob_updated
+
+# ----
+# Test 5: Mixed operations - add, update, delete simultaneously
+system ok
+echo "1,alice_final" > /tmp/rw_refresh_test/file1.csv
+
+system ok
+rm /tmp/rw_refresh_test/file2.csv
+
+system ok
+echo "5,eve" > /tmp/rw_refresh_test/file5.csv
+
+system ok
+echo "6,frank" > /tmp/rw_refresh_test/file6.csv
+
+statement ok
+REFRESH TABLE refresh_fs_t;
+
+sleep 2s
+
+
+
+query II rowsort retry 3 backoff 5s
+SELECT * FROM refresh_fs_t;
+----
+1 alice_final
+5 eve
+6 frank
+
+# ----
+# Test 6: Empty directory (all files deleted)
+system ok
+rm /tmp/rw_refresh_test/*.csv
+
+statement ok
+REFRESH TABLE refresh_fs_t;
+
+sleep 2s
+
+
+
+
+query II rowsort retry 3 backoff 5s
+SELECT * FROM refresh_fs_t;
+----
+
+# ----
+# Test 7: Re-populate after empty
+system ok
+echo "7,grace" > /tmp/rw_refresh_test/new1.csv
+
+system ok
+echo "8,henry" > /tmp/rw_refresh_test/new2.csv
+
+statement ok
+REFRESH TABLE refresh_fs_t;
+
+sleep 2s
+
+
+
+
+
+query II rowsort retry 3 backoff 5s
+SELECT * FROM refresh_fs_t;
+----
+7 grace
+8 henry
+
+# ----
+# Test 8: Error cases
+
+# Test refreshing non-existent table
+statement error
+REFRESH TABLE non_existent_table;
+
+# Test refreshing a non-refreshable table
+statement ok
+CREATE TABLE non_refreshable_t (v1 int) WITH (
+    connector = 'datagen',
+    fields.v1.kind = 'sequence',
+    fields.v1.start = '1',
+    fields.v1.end = '10'
+) FORMAT NATIVE ENCODE NATIVE;
+
+statement error Table 'public.non_refreshable_t' is not refreshable
+REFRESH TABLE non_refreshable_t;
+
+# ----
+# Test 9: Error case - create refreshable table without PRIMARY KEY
+statement error
+CREATE TABLE no_pk_refresh_t (id int, name varchar) WITH (
+    connector = 'batch_posix_fs',
+    batch_posix_fs.root = '/tmp/rw_refresh_test',
+    match_pattern = '*.csv'
+) FORMAT PLAIN ENCODE CSV (without_header = 'true', delimiter = ',');
+
+# ----
+# Cleanup
+statement ok
+DROP TABLE refresh_fs_t;
+
+statement ok
+DROP TABLE non_refreshable_t;
+
+system ok
+rm -rf /tmp/rw_refresh_test

proto/stream_plan.proto

@@ -372,6 +372,9 @@ message MaterializeNode {
   repeated common.ColumnOrder column_orders = 2;
   // Used for internal table states.
   catalog.Table table = 3;
+  // For refreshable tables, staging table for collecting new data during refresh.
+  // This table has the same schema as the main table but with a different table_id.
+  optional catalog.Table staging_table = 5;
 }
 
 message AggCallState {

proto/stream_service.proto

@@ -71,6 +71,10 @@ message BarrierCompleteResponse {
   // prev_epoch of barrier
   uint64 epoch = 9;
   uint32 database_id = 10;
+  // Used for refreshable batch source.
+  // SourceExecutor reports the source load is finished, and then
+  // meta will issue a LoadFinished barrier to notify MaterializeExecutor to start diff calculation.
+  repeated uint32 load_finished_source_ids = 11;
 }
 
 message StreamingControlStreamRequest {

src/common/src/util/stream_graph_visitor.rs

@@ -269,12 +269,16 @@ pub fn visit_stream_node_tables_inner<F>(
             }
 
             // Note: add internal tables for new nodes here.
-            NodeBody::Materialize(node) if !internal_tables_only => {
-                // Note: Plan directly generated by frontend may not have the table filled,
-                // as it is filled by the meta service during building fragments.
-                // However, if caller requests to visit this table via `!internal_tables_only`,
-                // we still assert on its existence to avoid undefined behavior.
-                always!(node.table, "Materialize")
+            NodeBody::Materialize(node) => {
+                if !internal_tables_only {
+                    // Note: Plan directly generated by frontend may not have the table filled,
+                    // as it is filled by the meta service during building fragments.
+                    // However, if caller requests to visit this table via `!internal_tables_only`,
+                    // we still assert on its existence to avoid undefined behavior.
+                    always!(node.table, "Materialize");
+                }
+                // Also visit the staging table if it exists (for refreshable tables)
+                optional!(node.staging_table, "MaterializeStaging");
             }
 
             // Global Approx Percentile

src/connector/src/macros.rs

@@ -39,6 +39,7 @@ macro_rules! for_all_classified_sources {
                 { Gcs, $crate::source::filesystem::opendal_source::GcsProperties , $crate::source::filesystem::OpendalFsSplit<$crate::source::filesystem::opendal_source::OpendalGcs> },
                 { OpendalS3, $crate::source::filesystem::opendal_source::OpendalS3Properties, $crate::source::filesystem::OpendalFsSplit<$crate::source::filesystem::opendal_source::OpendalS3> },
                 { PosixFs, $crate::source::filesystem::opendal_source::PosixFsProperties, $crate::source::filesystem::OpendalFsSplit<$crate::source::filesystem::opendal_source::OpendalPosixFs> },
+                { BatchPosixFs, $crate::source::filesystem::opendal_source::BatchPosixFsProperties, $crate::source::filesystem::opendal_source::BatchPosixFsSplit },
                 { Azblob, $crate::source::filesystem::opendal_source::AzblobProperties, $crate::source::filesystem::OpendalFsSplit<$crate::source::filesystem::opendal_source::OpendalAzblob> },
                 { Test, $crate::source::test_source::TestSourceProperties, $crate::source::test_source::TestSourceSplit},
                 { Iceberg, $crate::source::iceberg::IcebergProperties, $crate::source::iceberg::IcebergSplit}

src/connector/src/source/base.rs

@@ -50,6 +50,7 @@ use crate::error::ConnectorResult as Result;
 use crate::parser::ParserConfig;
 use crate::parser::schema_change::SchemaChangeEnvelope;
 use crate::source::SplitImpl::{CitusCdc, MongodbCdc, MysqlCdc, PostgresCdc, SqlServerCdc};
+use crate::source::batch::BatchSourceSplitImpl;
 use crate::source::monitor::EnumeratorMetrics;
 use crate::with_options::WithOptions;
 use crate::{
@@ -450,7 +451,10 @@ pub type BoxSourceMessageStream =
     BoxStream<'static, crate::error::ConnectorResult<Vec<SourceMessage>>>;
 /// Stream of [`StreamChunk`]s parsed from the messages from the external source.
 pub type BoxSourceChunkStream = BoxStream<'static, crate::error::ConnectorResult<StreamChunk>>;
+/// `StreamChunk` with the latest split state.
+/// The state is constructed in `StreamReaderBuilder::into_retry_stream`
 pub type StreamChunkWithState = (StreamChunk, HashMap<SplitId, SplitImpl>);
+/// See [`StreamChunkWithState`].
 pub type BoxSourceChunkWithStateStream =
     BoxStream<'static, crate::error::ConnectorResult<StreamChunkWithState>>;
 
@@ -701,6 +705,15 @@ impl SplitImpl {
             _ => unreachable!("get_cdc_split_offset() is only for cdc split"),
         }
     }
+
+    pub fn as_batch_split(self) -> Option<BatchSourceSplitImpl> {
+        match self {
+            SplitImpl::BatchPosixFs(batch_posix_fs_split) => {
+                Some(BatchSourceSplitImpl::BatchPosixFs(batch_posix_fs_split))
+            }
+            _ => None,
+        }
+    }
 }
 
 impl SplitMetaData for SplitImpl {

src/connector/src/source/batch.rs

@@ -0,0 +1,65 @@
+// Copyright 2025 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::source::filesystem::opendal_source::BatchPosixFsSplit;
+use crate::source::{SplitImpl, SplitMetaData};
+
+/// # Batch Refreshable Source
+///
+/// A batch refreshable source can be refreshed - reload all data from the source, e.g., re-run a `SELECT *` query from the source.
+/// The reloaded data will be handled by `RefreshableMaterialize` to calculate a diff to send to downstream.
+pub trait BatchSourceSplit: SplitMetaData {
+    fn finished(&self) -> bool;
+    /// Mark the source as finished. Called after the source is exhausted.
+    /// A `LoadFinish` signal will be sent by `SourceExecutor`, and the `RefreshableMaterialize` will begin to calculate the diff.
+    fn finish(&mut self);
+    /// Refresh the source to make it ready for re-run.
+    /// A `Refresh` signal will be sent, and the `RefreshableMaterialize` will begin to write the new data into a temporary staging table.
+    fn refresh(&mut self);
+}
+
+pub enum BatchSourceSplitImpl {
+    BatchPosixFs(BatchPosixFsSplit),
+}
+
+/// See [`BatchSourceSplit`] for more details.
+impl BatchSourceSplitImpl {
+    pub fn finished(&self) -> bool {
+        match self {
+            BatchSourceSplitImpl::BatchPosixFs(split) => split.finished(),
+        }
+    }
+
+    pub fn finish(&mut self) {
+        tracing::info!("finishing batch source split");
+        match self {
+            BatchSourceSplitImpl::BatchPosixFs(split) => split.finish(),
+        }
+    }
+
+    pub fn refresh(&mut self) {
+        tracing::info!("refreshing batch source split");
+        match self {
+            BatchSourceSplitImpl::BatchPosixFs(split) => split.refresh(),
+        }
+    }
+}
+
+impl From<BatchSourceSplitImpl> for SplitImpl {
+    fn from(batch_split: BatchSourceSplitImpl) -> Self {
+        match batch_split {
+            BatchSourceSplitImpl::BatchPosixFs(split) => SplitImpl::BatchPosixFs(split),
+        }
+    }
+}