feat(frontend): add refreshable property & refresh command

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

Author: xxchan

proto/catalog.proto

@@ -498,6 +498,10 @@ message Table {
   // NOTICE: when it is "None", the watermark column should be the first column in the pk
   optional int32 clean_watermark_index_in_pk = 44;
 
+  // Whether the table supports manual refresh operation:
+  // reload data from external source and emit messages based on the diff with current data.
+  bool refreshable = 45;
+
   // Per-table catalog version, used by schema change. `None` for internal
   // tables and tests. Not to be confused with the global catalog version for
   // notification service.

proto/meta.proto

@@ -178,6 +178,15 @@ message ResumeRequest {}
 
 message ResumeResponse {}
 
+message RefreshRequest {
+  uint32 table_id = 1;
+  uint32 associated_source_id = 2;
+}
+
+message RefreshResponse {
+  common.Status status = 1;
+}
+
 message CancelCreatingJobsRequest {
   message CreatingJobInfo {
     uint32 database_id = 1;
@@ -366,6 +375,7 @@ service StreamManagerService {
   rpc Flush(FlushRequest) returns (FlushResponse);
   rpc Pause(PauseRequest) returns (PauseResponse);
   rpc Resume(ResumeRequest) returns (ResumeResponse);
+  rpc Refresh(RefreshRequest) returns (RefreshResponse);
   rpc CancelCreatingJobs(CancelCreatingJobsRequest) returns (CancelCreatingJobsResponse);
   rpc ListTableFragments(ListTableFragmentsRequest) returns (ListTableFragmentsResponse);
   rpc ListStreamingJobStates(ListStreamingJobStatesRequest) returns (ListStreamingJobStatesResponse);

proto/stream_plan.proto

@@ -123,6 +123,17 @@ message StartFragmentBackfillMutation {
   repeated uint32 fragment_ids = 1;
 }
 
+message RefreshStartMutation {
+  // Table ID to start refresh operation.
+  uint32 table_id = 1;
+  // Associated source ID for this refresh operation.
+  uint32 associated_source_id = 2;
+}
+message LoadFinishMutation {
+  // Associated source ID for this load operation.
+  uint32 associated_source_id = 1;
+}
+
 message BarrierMutation {
   oneof mutation {
     // Add new dispatchers to some actors, used for creating materialized views.
@@ -150,6 +161,10 @@ message BarrierMutation {
     // If we use rate limit to pause / resume backfill fragments, if user manually
     // resumes some fragments, this will overwrite the backfill order configuration.
     StartFragmentBackfillMutation start_fragment_backfill = 14;
+    // Start refresh signal for refreshing tables
+    RefreshStartMutation refresh_start = 15;
+    // Load finish signal for refreshing tables
+    LoadFinishMutation load_finish = 16;
     // Combined mutation.
     // Currently, it can only be Add & Update, which is for sink into table.
     CombinedMutation combined = 100;

src/connector/src/with_options.rs

@@ -201,6 +201,10 @@ pub trait WithPropertiesExt: Get + GetKeyIter + Sized {
             })
             .unwrap_or(false)
     }
+
+    fn is_refreshable_connector(&self) -> bool {
+        false
+    }
 }
 
 impl<T: Get + GetKeyIter> WithPropertiesExt for T {}

src/frontend/src/catalog/system_catalog/rw_catalog/rw_tables.rs

@@ -28,6 +28,7 @@ struct RwTable {
     owner: i32,
     definition: String,
     append_only: bool,
+    refreshable: bool,
     acl: Vec<String>,
     initialized_at: Option<Timestamptz>,
     created_at: Option<Timestamptz>,
@@ -57,6 +58,7 @@ fn read_rw_table_info(reader: &SysCatalogReaderImpl) -> Result<Vec<RwTable>> {
                     owner: table.owner as i32,
                     definition: table.create_sql_purified(),
                     append_only: table.append_only,
+                    refreshable: table.refreshable,
                     acl: get_acl_items(
                         &GrantObject::TableId(table.id.table_id),
                         true,

src/frontend/src/catalog/table_catalog.rs

@@ -197,6 +197,9 @@ pub struct TableCatalog {
     pub engine: Engine,
 
     pub clean_watermark_index_in_pk: Option<usize>,
+
+    /// Whether the table supports manual refresh operations
+    pub refreshable: bool,
 }
 
 pub const ICEBERG_SOURCE_PREFIX: &str = "__iceberg_source_";
@@ -582,6 +585,7 @@ impl TableCatalog {
             job_id: self.job_id.map(|id| id.table_id),
             engine: Some(self.engine.to_protobuf().into()),
             clean_watermark_index_in_pk: self.clean_watermark_index_in_pk.map(|x| x as i32),
+            refreshable: self.refreshable,
         }
     }
 
@@ -784,6 +788,7 @@ impl From<PbTable> for TableCatalog {
             job_id: tb.job_id.map(TableId::from),
             engine,
             clean_watermark_index_in_pk: tb.clean_watermark_index_in_pk.map(|x| x as usize),
+            refreshable: tb.refreshable,
         }
     }
 }

src/frontend/src/handler/mod.rs

@@ -108,6 +108,7 @@ mod prepared_statement;
 pub mod privilege;
 pub mod query;
 mod recover;
+mod refresh;
 pub mod show;
 mod transaction;
 mod use_db;
@@ -1283,6 +1284,9 @@ pub async fn handle(
             prepared_statement::handle_deallocate(name, prepare).await
         }
         Statement::Vacuum { object_name } => vacuum::handle_vacuum(handler_args, object_name).await,
+        Statement::Refresh { table_name } => {
+            refresh::handle_refresh(handler_args, table_name).await
+        }
         _ => bail_not_implemented!("Unhandled statement: {}", stmt),
     }
 }

src/frontend/src/handler/refresh.rs

@@ -0,0 +1,104 @@
+// 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 pgwire::pg_response::{PgResponse, StatementType};
+use risingwave_pb::meta::RefreshRequest;
+use risingwave_sqlparser::ast::ObjectName;
+
+use crate::catalog::table_catalog::TableType;
+use crate::error::{ErrorCode, Result};
+use crate::handler::util::get_table_catalog_by_table_name;
+use crate::handler::{HandlerArgs, RwPgResponse};
+
+/// Handle REFRESH statement
+///
+/// This function processes the REFRESH statement by:
+/// 1. Validating the table exists and is refreshable
+/// 2. Sending a refresh command to the meta service
+/// 3. Returning appropriate response to the client
+pub async fn handle_refresh(
+    handler_args: HandlerArgs,
+    table_name: ObjectName,
+) -> Result<RwPgResponse> {
+    let session = handler_args.session;
+
+    // Get table catalog to validate table exists
+    let (table_catalog, schema_name) =
+        get_table_catalog_by_table_name(session.as_ref(), &table_name)?;
+
+    // Check if table supports refresh operations
+    if !table_catalog.refreshable {
+        return Err(ErrorCode::InvalidInputSyntax(format!(
+            "Table '{}.{}' is not refreshable. Only tables created with REFRESHABLE flag support manual refresh.",
+            schema_name, table_name
+        )).into());
+    }
+
+    // Only allow refresh on tables, not views or materialized views
+    match table_catalog.table_type() {
+        TableType::Table => {
+            // This is valid
+        }
+        t @ (TableType::MaterializedView | TableType::Index | TableType::Internal) => {
+            return Err(ErrorCode::InvalidInputSyntax(format!(
+                "REFRESH is only supported for tables, got {:?}.",
+                t
+            ))
+            .into());
+        }
+    }
+
+    let table_id = table_catalog.id();
+
+    // Create refresh request
+    let refresh_request = RefreshRequest {
+        table_id: table_id.table_id(),
+        associated_source_id: table_catalog.associated_source_id().unwrap().table_id(),
+    };
+
+    // Send refresh command to meta service via stream manager
+    let meta_client = session.env().meta_client();
+    match meta_client.refresh(refresh_request).await {
+        Ok(_) => {
+            // Refresh command sent successfully
+            tracing::info!(
+                table_id = %table_id,
+                table_name = %table_name,
+                "Manual refresh initiated"
+            );
+
+            // Return success response
+            Ok(PgResponse::builder(StatementType::OTHER)
+                .notice(format!(
+                    "REFRESH initiated for table '{}.{}'",
+                    schema_name, table_name
+                ))
+                .into())
+        }
+        Err(e) => {
+            tracing::error!(
+                error = %e,
+                table_id = %table_id,
+                table_name = %table_name,
+                "Failed to initiate refresh"
+            );
+
+            Err(ErrorCode::InternalError(format!(
+                "Failed to refresh table '{}.{}': {}",
+                schema_name, table_name, e
+            ))
+            .into())
+        }
+    }
+}

src/frontend/src/meta_client.rs

@@ -36,7 +36,10 @@ use risingwave_pb::meta::list_object_dependencies_response::PbObjectDependencies
 use risingwave_pb::meta::list_rate_limits_response::RateLimitInfo;
 use risingwave_pb::meta::list_streaming_job_states_response::StreamingJobState;
 use risingwave_pb::meta::list_table_fragments_response::TableFragmentInfo;
-use risingwave_pb::meta::{EventLog, FragmentDistribution, PbThrottleTarget, RecoveryStatus};
+use risingwave_pb::meta::{
+    EventLog, FragmentDistribution, PbThrottleTarget, RecoveryStatus, RefreshRequest,
+    RefreshResponse,
+};
 use risingwave_pb::secret::PbSecretRef;
 use risingwave_rpc_client::error::Result;
 use risingwave_rpc_client::{HummockMetaClient, MetaClient};
@@ -161,6 +164,8 @@ pub trait FrontendMetaClient: Send + Sync {
     async fn set_sync_log_store_aligned(&self, job_id: u32, aligned: bool) -> Result<()>;
 
     async fn compact_iceberg_table(&self, sink_id: SinkId) -> Result<u64>;
+
+    async fn refresh(&self, request: RefreshRequest) -> Result<RefreshResponse>;
 }
 
 pub struct FrontendMetaClientImpl(pub MetaClient);
@@ -402,4 +407,8 @@ impl FrontendMetaClient for FrontendMetaClientImpl {
     async fn compact_iceberg_table(&self, sink_id: SinkId) -> Result<u64> {
         self.0.compact_iceberg_table(sink_id).await
     }
+
+    async fn refresh(&self, request: RefreshRequest) -> Result<RefreshResponse> {
+        self.0.refresh(request).await
+    }
 }

src/frontend/src/optimizer/mod.rs

@@ -58,6 +58,7 @@ use risingwave_common::catalog::{ColumnCatalog, ColumnDesc, ConflictBehavior, Fi
 use risingwave_common::types::DataType;
 use risingwave_common::util::column_index_mapping::ColIndexMapping;
 use risingwave_common::util::iter_util::ZipEqDebug;
+use risingwave_connector::WithPropertiesExt;
 use risingwave_connector::sink::catalog::SinkFormatDesc;
 use risingwave_pb::stream_plan::StreamScanType;
 
@@ -943,6 +944,12 @@ impl LogicalPlanRoot {
                 StreamFilter::filter_out_any_null_rows(stream_plan.clone(), &not_null_idxs);
         }
 
+        // Determine if the table should be refreshable based on the connector type
+        let refreshable = source_catalog
+            .as_ref()
+            .map(|catalog| catalog.with_properties.is_refreshable_connector())
+            .unwrap_or(false);
+
         StreamMaterialize::create_for_table(
             stream_plan,
             table_name,
@@ -960,6 +967,7 @@ impl LogicalPlanRoot {
             retention_seconds,
             webhook_info,
             engine,
+            refreshable,
         )
     }