feat: refreshable table, source part

Signed-off-by: xxchan <[email protected]>

Author: xxchan

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/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 into_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),
+        }
+    }
+}