update

Author: luoluoyuyu

Cargo.lock

@@ -14,7 +14,6 @@ arrow-schema = "52"
 comfy-table = "7"
 protocol = { path = "../../protocol" }
 clap = { version = "4.5", features = ["derive"] }
-thiserror = "2"
 tokio = { version = "1.0", features = ["full", "signal"] }
 tonic = { version = "0.12", features = ["default"] }
 rustyline = { version = "14.0", features = ["with-dirs"] }

cli/cli/Cargo.toml

@@ -20,26 +20,62 @@ use comfy_table::{Attribute, Cell, Color, ContentArrangement, Table, TableCompon
 use protocol::cli::{function_stream_service_client::FunctionStreamServiceClient, SqlRequest};
 use rustyline::error::ReadlineError;
 use rustyline::{Config, DefaultEditor, EditMode};
+use std::fmt;
 use std::io::{self, Cursor, Write};
 use std::sync::Arc;
 use tokio::sync::Mutex;
 use tonic::Request;
 
-#[derive(Debug, thiserror::Error)]
+/// CLI errors.
+///
+/// **Important:** [`tonic::Status`] must not be formatted with `{}` — its [`fmt::Display`] dumps
+/// `details` / `metadata` (e.g. HTTP headers). Only [`tonic::Status::message`] is stored in
+/// [`ReplError::Rpc`].
+#[derive(Debug)]
 pub enum ReplError {
-    #[error("RPC error: {0}")]
-    Rpc(Box<tonic::Status>),
-    #[error("Connection failed: {0}")]
+    Rpc(String),
     Connection(String),
-    #[error("Internal error: {0}")]
     Internal(String),
-    #[error("IO error: {0}")]
-    Io(#[from] io::Error),
+    Io(io::Error),
+}
+
+impl fmt::Display for ReplError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            ReplError::Rpc(s) => f.write_str(s),
+            ReplError::Connection(s) => f.write_str(s),
+            ReplError::Internal(s) => write!(f, "Internal error: {s}"),
+            ReplError::Io(e) => write!(f, "IO error: {e}"),
+        }
+    }
+}
+
+impl std::error::Error for ReplError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            ReplError::Io(e) => Some(e),
+            _ => None,
+        }
+    }
+}
+
+impl From<io::Error> for ReplError {
+    fn from(e: io::Error) -> Self {
+        ReplError::Io(e)
+    }
 }
 
 impl From<tonic::Status> for ReplError {
     fn from(s: tonic::Status) -> Self {
-        ReplError::Rpc(Box::new(s))
+        let msg = s.message();
+        if msg.is_empty() {
+            ReplError::Rpc(format!(
+                "gRPC {} (server returned no message)",
+                s.code()
+            ))
+        } else {
+            ReplError::Rpc(msg.to_string())
+        }
     }
 }
 

cli/cli/src/repl.rs

@@ -117,3 +117,10 @@ task_storage:
 
     # Maximum bytes for level base in bytes (optional)
     max_bytes_for_level_base: 268435456
+
+# Stream table catalog (SQL: CREATE TABLE connector sources, SHOW TABLES, SHOW CREATE TABLE).
+# When persist is true (default), metadata is stored under RocksDB at db_path (default: data/stream_catalog)
+# and reloaded after process restart. Set persist: false only for tests/ephemeral nodes.
+stream_catalog:
+  persist: true
+  # db_path: data/stream_catalog

conf/config.yaml

@@ -27,6 +27,8 @@ message StreamSource {
   bytes arrow_schema_ipc = 1;
   optional string event_time_field = 2;
   optional string watermark_field = 3;
+  // Original CREATE TABLE ... WITH ('k'='v', ...) pairs (best-effort; keys sorted in DDL).
+  map<string, string> with_options = 4;
 }
 
 message StreamSink {

protocol/proto/storage.proto

@@ -40,6 +40,8 @@ pub struct GlobalConfig {
     pub task_storage: crate::config::storage::TaskStorageConfig,
     #[serde(default)]
     pub streaming: StreamingConfig,
+    #[serde(default)]
+    pub stream_catalog: crate::config::storage::StreamCatalogConfig,
 }
 
 impl GlobalConfig {

src/config/global_config.rs

@@ -118,3 +118,27 @@ impl Default for TaskStorageConfig {
         }
     }
 }
+
+/// Stream table catalog (`CREATE TABLE` / `SHOW TABLES`) storage.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct StreamCatalogConfig {
+    /// When `false`, the catalog is in-memory only and is **lost on process restart**.
+    #[serde(default = "default_stream_catalog_persist")]
+    pub persist: bool,
+    /// RocksDB directory for persisted catalog. Default: `{data_dir}/stream_catalog`.
+    #[serde(default)]
+    pub db_path: Option<String>,
+}
+
+fn default_stream_catalog_persist() -> bool {
+    true
+}
+
+impl Default for StreamCatalogConfig {
+    fn default() -> Self {
+        Self {
+            persist: default_stream_catalog_persist(),
+            db_path: None,
+        }
+    }
+}

src/config/storage.rs

@@ -14,8 +14,9 @@ use super::Analysis;
 use crate::coordinator::execution_context::ExecutionContext;
 use crate::coordinator::statement::{
     CreateFunction, CreatePythonFunction, CreateTable, DropFunction, DropTableStatement,
-    ShowFunctions, StartFunction, Statement, StatementVisitor, StatementVisitorContext,
-    StatementVisitorResult, StopFunction, StreamingTableStatement,
+    ShowCatalogTables, ShowCreateTable, ShowFunctions, StartFunction, Statement,
+    StatementVisitor, StatementVisitorContext, StatementVisitorResult, StopFunction,
+    StreamingTableStatement,
 };
 use std::fmt;
 
@@ -109,6 +110,22 @@ impl StatementVisitor for Analyzer<'_> {
         StatementVisitorResult::Analyze(Box::new(stmt.clone()))
     }
 
+    fn visit_show_catalog_tables(
+        &self,
+        stmt: &ShowCatalogTables,
+        _context: &StatementVisitorContext,
+    ) -> StatementVisitorResult {
+        StatementVisitorResult::Analyze(Box::new(stmt.clone()))
+    }
+
+    fn visit_show_create_table(
+        &self,
+        stmt: &ShowCreateTable,
+        _context: &StatementVisitorContext,
+    ) -> StatementVisitorResult {
+        StatementVisitorResult::Analyze(Box::new(stmt.clone()))
+    }
+
     fn visit_create_python_function(
         &self,
         stmt: &CreatePythonFunction,

src/coordinator/analyze/analyzer.rs

@@ -12,8 +12,12 @@
 
 mod data_set;
 mod execute_result;
+mod show_catalog_tables_result;
+mod show_create_table_result;
 mod show_functions_result;
 
 pub use data_set::{DataSet, empty_record_batch};
 pub use execute_result::ExecuteResult;
+pub use show_catalog_tables_result::ShowCatalogTablesResult;
+pub use show_create_table_result::ShowCreateTableResult;
 pub use show_functions_result::ShowFunctionsResult;

src/coordinator/dataset/mod.rs

@@ -0,0 +1,92 @@
+// 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 std::sync::Arc;
+
+use arrow_array::{Int32Array, StringArray};
+use arrow_schema::{DataType, Field, Schema};
+
+use super::DataSet;
+use crate::sql::schema::{schema_columns_one_line, stream_table_row_detail, StreamTable};
+
+#[derive(Clone, Debug)]
+pub struct ShowCatalogTablesResult {
+    names: Vec<String>,
+    kinds: Vec<String>,
+    column_counts: Vec<i32>,
+    schema_lines: Vec<String>,
+    details: Vec<String>,
+}
+
+impl ShowCatalogTablesResult {
+    pub fn from_tables(tables: &[Arc<StreamTable>]) -> Self {
+        let mut names = Vec::with_capacity(tables.len());
+        let mut kinds = Vec::with_capacity(tables.len());
+        let mut column_counts = Vec::with_capacity(tables.len());
+        let mut schema_lines = Vec::with_capacity(tables.len());
+        let mut details = Vec::with_capacity(tables.len());
+
+        for t in tables {
+            let schema = t.schema();
+            let ncols = schema.fields().len() as i32;
+            names.push(t.name().to_string());
+            kinds.push(match t.as_ref() {
+                StreamTable::Source { .. } => "SOURCE",
+                StreamTable::Sink { .. } => "SINK",
+            }
+            .to_string());
+            column_counts.push(ncols);
+            schema_lines.push(schema_columns_one_line(&schema));
+            details.push(stream_table_row_detail(t.as_ref()));
+        }
+
+        Self {
+            names,
+            kinds,
+            column_counts,
+            schema_lines,
+            details,
+        }
+    }
+}
+
+impl DataSet for ShowCatalogTablesResult {
+    fn to_record_batch(&self) -> arrow_array::RecordBatch {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("table_name", DataType::Utf8, false),
+            Field::new("kind", DataType::Utf8, false),
+            Field::new("column_count", DataType::Int32, false),
+            Field::new("schema_columns", DataType::Utf8, false),
+            Field::new("details", DataType::Utf8, false),
+        ]));
+
+        arrow_array::RecordBatch::try_new(
+            schema,
+            vec![
+                Arc::new(StringArray::from(
+                    self.names.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
+                )),
+                Arc::new(StringArray::from(
+                    self.kinds.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
+                )),
+                Arc::new(Int32Array::from(self.column_counts.clone())),
+                Arc::new(StringArray::from(
+                    self.schema_lines.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
+                )),
+                Arc::new(StringArray::from(
+                    self.details.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
+                )),
+            ],
+        )
+        .unwrap_or_else(|_| arrow_array::RecordBatch::new_empty(Arc::new(Schema::empty())))
+    }
+}