rename source

src/common/src/session_config/mod.rs

@@ -55,7 +55,7 @@ type SessionConfigResult<T> = std::result::Result<T, SessionConfigError>;
 // NOTE(kwannoel): We declare it separately as a constant,
 // otherwise seems like it can't infer the type of -1 when written inline.
 const DISABLE_BACKFILL_RATE_LIMIT: i32 = -1;
-const DISABLE_STREAMING_RATE_LIMIT: i32 = -1;
+const DISABLE_SOURCE_RATE_LIMIT: i32 = -1;
 
 #[serde_as]
 /// This is the Session Config of RisingWave.
@@ -263,8 +263,8 @@ pub struct SessionConfig {
     /// Set streaming rate limit (rows per second) for each parallelism for mv / source backfilling, source reads.
     /// If set to -1, disable rate limit.
     /// If set to 0, this pauses the snapshot read / source read.
-    #[parameter(default = DISABLE_STREAMING_RATE_LIMIT)]
-    streaming_rate_limit: i32,
+    #[parameter(default = DISABLE_SOURCE_RATE_LIMIT)]
+    source_rate_limit: i32,
 
     /// Cache policy for partition cache in streaming over window.
     /// Can be "full", "recent", "`recent_first_n`" or "`recent_last_n`".

src/frontend/src/handler/alter_streaming_rate_limit.rs

@@ -68,7 +68,7 @@ pub async fn handle_alter_streaming_rate_limit(
             let source_id = if let Some(id) = table.associated_source_id {
                 id.table_id()
             } else {
-                bail!("ALTER STREAMING_RATE_LIMIT is not for table without source")
+                bail!("ALTER SOURCE_RATE_LIMIT is not for table without source")
             };
             (StatementType::ALTER_SOURCE, source_id)
         }

src/frontend/src/handler/mod.rs

@@ -697,7 +697,7 @@ pub async fn handle(
         } => alter_table_with_sr::handle_refresh_schema(handler_args, name).await,
         Statement::AlterTable {
             name,
-            operation: AlterTableOperation::SetStreamingRateLimit { rate_limit },
+            operation: AlterTableOperation::SetSourceRateLimit { rate_limit },
         } => {
             alter_streaming_rate_limit::handle_alter_streaming_rate_limit(
                 handler_args,

src/frontend/src/optimizer/plan_node/stream_fs_fetch.rs

@@ -119,7 +119,7 @@ impl StreamNode for StreamFsFetch {
                     .map(|c| c.to_protobuf())
                     .collect_vec(),
                 with_properties,
-                rate_limit: self.base.ctx().overwrite_options().streaming_rate_limit,
+                rate_limit: self.base.ctx().overwrite_options().source_rate_limit,
                 secret_refs,
             }
         });

src/frontend/src/optimizer/plan_node/stream_source.rs

@@ -112,7 +112,7 @@ impl StreamNode for StreamSource {
                     .map(|c| c.to_protobuf())
                     .collect_vec(),
                 with_properties,
-                rate_limit: self.base.ctx().overwrite_options().streaming_rate_limit,
+                rate_limit: self.base.ctx().overwrite_options().source_rate_limit,
                 secret_refs,
             }
         });

src/frontend/src/utils/overwrite_options.rs

@@ -16,21 +16,21 @@ use crate::handler::HandlerArgs;
 
 #[derive(Debug, Clone, Default)]
 pub struct OverwriteOptions {
-    pub streaming_rate_limit: Option<u32>,
+    pub source_rate_limit: Option<u32>,
     pub backfill_rate_limit: Option<u32>,
 }
 
 impl OverwriteOptions {
     pub(crate) const BACKFILL_RATE_LIMIT_KEY: &'static str = "backfill_rate_limit";
-    pub(crate) const STREAMING_RATE_LIMIT_KEY: &'static str = "streaming_rate_limit";
+    pub(crate) const SOURCE_RATE_LIMIT_KEY: &'static str = "source_rate_limit";
 
     pub fn new(args: &mut HandlerArgs) -> Self {
-        let streaming_rate_limit = {
-            if let Some(x) = args.with_options.remove(Self::STREAMING_RATE_LIMIT_KEY) {
+        let source_rate_limit = {
+            if let Some(x) = args.with_options.remove(Self::SOURCE_RATE_LIMIT_KEY) {
                 // FIXME(tabVersion): validate the value
                 Some(x.parse::<u32>().unwrap())
             } else {
-                let rate_limit = args.session.config().streaming_rate_limit();
+                let rate_limit = args.session.config().source_rate_limit();
                 if rate_limit < 0 {
                     None
                 } else {
@@ -52,7 +52,7 @@ impl OverwriteOptions {
             }
         };
         Self {
-            streaming_rate_limit,
+            source_rate_limit,
             backfill_rate_limit,
         }
     }

src/frontend/src/utils/with_options.rs

@@ -89,8 +89,7 @@ impl WithOptions {
             .inner
             .into_iter()
             .filter(|(key, _)| {
-                key != OverwriteOptions::STREAMING_RATE_LIMIT_KEY
-                    && key != options::RETENTION_SECONDS
+                key != OverwriteOptions::SOURCE_RATE_LIMIT_KEY && key != options::RETENTION_SECONDS
             })
             .collect();
 

src/sqlparser/src/ast/ddl.rs

@@ -102,8 +102,8 @@ pub enum AlterTableOperation {
         deferred: bool,
     },
     RefreshSchema,
-    /// `SET STREAMING_RATE_LIMIT TO <rate_limit>`
-    SetStreamingRateLimit {
+    /// `SET SOURCE_RATE_LIMIT TO <rate_limit>`
+    SetSourceRateLimit {
         rate_limit: i32,
     },
 }
@@ -290,8 +290,8 @@ impl fmt::Display for AlterTableOperation {
             AlterTableOperation::RefreshSchema => {
                 write!(f, "REFRESH SCHEMA")
             }
-            AlterTableOperation::SetStreamingRateLimit { rate_limit } => {
-                write!(f, "SET STREAMING_RATE_LIMIT TO {}", rate_limit)
+            AlterTableOperation::SetSourceRateLimit { rate_limit } => {
+                write!(f, "SET SOURCE_RATE_LIMIT TO {}", rate_limit)
             }
         }
     }

src/sqlparser/src/parser.rs

@@ -3089,7 +3089,7 @@ impl Parser<'_> {
                     deferred,
                 }
             } else if let Some(rate_limit) = self.parse_alter_source_rate_limit(true)? {
-                AlterTableOperation::SetStreamingRateLimit { rate_limit }
+                AlterTableOperation::SetSourceRateLimit { rate_limit }
             } else {
                 return self.expected("SCHEMA/PARALLELISM/SOURCE_RATE_LIMIT after SET");
             }
@@ -3402,7 +3402,7 @@ impl Parser<'_> {
             AlterSourceOperation::RefreshSchema
         } else {
             return self.expected(
-                "RENAME, ADD COLUMN, OWNER TO, SET or STREAMING_RATE_LIMIT after ALTER SOURCE",
+                "RENAME, ADD COLUMN, OWNER TO, SET or SOURCE_RATE_LIMIT after ALTER SOURCE",
             );
         };