feat(query): add api to list stream

src/query/service/src/servers/http/v1/catalog/list_database_streams.rs

@@ -0,0 +1,148 @@
+// Copyright 2021 Datafuse 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 std::collections::HashMap;
+use std::collections::HashSet;
+
+use chrono::DateTime;
+use chrono::Utc;
+use databend_common_catalog::catalog::CatalogManager;
+use databend_common_exception::ErrorCode;
+use databend_common_exception::Result;
+use databend_common_storages_stream::stream_table::StreamTable;
+use databend_common_users::Object;
+use poem::error::InternalServerError;
+use poem::error::NotFound;
+use poem::error::Result as PoemResult;
+use poem::web::Json;
+use poem::web::Path;
+use poem::IntoResponse;
+use serde::Deserialize;
+use serde::Serialize;
+
+use crate::servers::http::v1::HttpQueryContext;
+
+#[derive(Serialize, Deserialize, Eq, PartialEq, Debug, Default)]
+pub struct ListDatabaseStreamsResponse {
+    pub streams: Vec<StreamInfo>,
+    pub warnings: Vec<String>,
+}
+
+#[derive(Serialize, Deserialize, Eq, PartialEq, Debug, Default)]
+pub struct StreamInfo {
+    pub name: String,
+    pub database: String,
+    pub catalog: String,
+    pub stream_id: u64,
+    pub created_on: DateTime<Utc>,
+    pub updated_on: DateTime<Utc>,
+    pub mode: String,
+    pub comment: String,
+    pub table_name: Option<String>,
+    pub table_id: Option<u64>,
+    pub table_version: Option<u64>,
+    pub snapshot_location: Option<String>,
+}
+
+#[async_backtrace::framed]
+async fn handle(ctx: &HttpQueryContext, database: String) -> Result<ListDatabaseStreamsResponse> {
+    let tenant = ctx.session.get_current_tenant();
+    let visibility_checker = ctx
+        .session
+        .get_visibility_checker(false, Object::All)
+        .await?;
+
+    let catalog = CatalogManager::instance().get_default_catalog(Default::default())?;
+    let db = catalog.get_database(&tenant, &database).await?;
+
+    if !visibility_checker.check_database_visibility(
+        catalog.name().as_str(),
+        db.name(),
+        db.get_db_info().database_id.db_id,
+    ) {
+        return Err(ErrorCode::UnknownDatabase(format!(
+            "[HTTP-CATALOG] Unknown database: '{}'",
+            database
+        )));
+    }
+
+    let warnings = vec![];
+    let mut streams = vec![];
+    let tables = db.list_tables().await?;
+    let mut source_table_id_set = HashSet::new();
+    for table in tables {
+        if !table.is_stream() {
+            continue;
+        }
+        let info = table.get_table_info();
+        if !visibility_checker.check_table_visibility(
+            catalog.name().as_str(),
+            db.name(),
+            table.name(),
+            db.get_db_info().database_id.db_id,
+            info.ident.table_id,
+        ) {
+            continue;
+        }
+        let stream = StreamTable::try_from_table(table.as_ref())?;
+        let source_table_id = stream.source_table_id()?;
+        source_table_id_set.insert(source_table_id);
+        streams.push(StreamInfo {
+            name: table.name().to_string(),
+            database: db.name().to_string(),
+            catalog: catalog.name().clone(),
+            stream_id: info.ident.table_id,
+            created_on: info.meta.created_on,
+            updated_on: info.meta.updated_on,
+            mode: stream.mode().to_string(),
+            comment: info.meta.comment.clone(),
+            table_name: None,
+            table_id: Some(source_table_id),
+            table_version: stream.offset().ok(),
+            snapshot_location: stream.snapshot_loc(),
+        });
+    }
+
+    let source_table_ids = source_table_id_set.into_iter().collect::<Vec<u64>>();
+    let source_table_names = catalog
+        .mget_table_names_by_ids(&tenant, &source_table_ids, false)
+        .await?;
+    let source_table_map = source_table_ids
+        .into_iter()
+        .zip(source_table_names.into_iter())
+        .filter(|(_, tb_name)| tb_name.is_some())
+        .map(|(tb_id, tb_name)| (tb_id, tb_name.unwrap()))
+        .collect::<HashMap<_, _>>();
+
+    streams.iter_mut().for_each(|stream| {
+        stream.table_name = stream
+            .table_id
+            .and_then(|id| source_table_map.get(&id).cloned());
+    });
+
+    Ok(ListDatabaseStreamsResponse { streams, warnings })
+}
+
+#[poem::handler]
+#[async_backtrace::framed]
+pub async fn list_database_streams_handler(
+    ctx: &HttpQueryContext,
+    Path(database): Path<String>,
+) -> PoemResult<impl IntoResponse> {
+    let resp = handle(ctx, database).await.map_err(|e| match e.code() {
+        ErrorCode::UNKNOWN_DATABASE => NotFound(e),
+        _ => InternalServerError(e),
+    })?;
+    Ok(Json(resp))
+}

src/query/service/src/servers/http/v1/catalog/mod.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 pub mod get_database_table;
+pub mod list_database_streams;
 pub mod list_database_table_fields;
 pub mod list_database_tables;
 pub mod list_databases;
@@ -21,6 +22,7 @@ pub mod search_tables;
 pub mod stats;
 
 pub use get_database_table::get_database_table_handler;
+pub use list_database_streams::list_database_streams_handler;
 pub use list_database_table_fields::list_database_table_fields_handler;
 pub use list_database_tables::list_database_tables_handler;
 pub use list_databases::list_databases_handler;

src/query/service/src/servers/http/v1/http_query_handlers.rs

@@ -71,6 +71,7 @@ use crate::servers::http::middleware::HTTPSessionMiddleware;
 use crate::servers::http::middleware::MetricsMiddleware;
 use crate::servers::http::v1::catalog::catalog_stats_handler;
 use crate::servers::http::v1::catalog::get_database_table_handler;
+use crate::servers::http::v1::catalog::list_database_streams_handler;
 use crate::servers::http::v1::catalog::list_database_table_fields_handler;
 use crate::servers::http::v1::catalog::list_database_tables_handler;
 use crate::servers::http::v1::catalog::list_databases_handler;
@@ -792,6 +793,11 @@ pub fn query_route() -> Route {
             get(list_database_table_fields_handler),
             EndpointKind::Catalog,
         ),
+        (
+            "/catalog/databases/:database/streams",
+            get(list_database_streams_handler),
+            EndpointKind::Catalog,
+        ),
         (
             "/catalog/search/tables",
             post(search_tables_handler),

tests/metactl/metactl_utils.py

@@ -85,9 +85,9 @@ def verify_kv(grpc_addr, key, expected_value=None):
         print(f"Actual value: '{actual_value}', Expected: '{expected_value}'")
 
         if expected_value is not None:
-            assert (
-                actual_value == expected_value
-            ), f"Expected '{expected_value}', got '{actual_value}'"
+            assert actual_value == expected_value, (
+                f"Expected '{expected_value}', got '{actual_value}'"
+            )
 
 
 def metactl_export_from_grpc(addr: str) -> str:

tests/metactl/subcommands/cmd_export_from_grpc.py

@@ -71,9 +71,9 @@ def test_export_from_grpc():
 
     # Compare with expected output by converting all to JSON
     print(f"Got {len(lines)} lines, expected {len(want)} lines")
-    assert len(lines) == len(
-        want
-    ), f"Line count mismatch: got {len(lines)}, expected {len(want)}"
+    assert len(lines) == len(want), (
+        f"Line count mismatch: got {len(lines)}, expected {len(want)}"
+    )
 
     def normalize_json(obj):
         """Remove dynamic fields like time_ms from JSON object for comparison"""
@@ -100,9 +100,9 @@ def normalize_json(obj):
 
         want_json = normalize_json(want_json)
 
-        assert (
-            actual_json == want_json
-        ), f"Line {i} JSON mismatch:\nActual: {actual_json}\nExpected: {want_json}"
+        assert actual_json == want_json, (
+            f"Line {i} JSON mismatch:\nActual: {actual_json}\nExpected: {want_json}"
+        )
 
     print(f"✓ All {len(lines)} JSON lines match expected output")
 

tests/metactl/subcommands/cmd_export_from_raft_dir.py

@@ -75,9 +75,9 @@ def test_export_from_raft_dir():
 
     # Compare with expected output by converting all to JSON
     print(f"Got {len(lines)} lines, expected {len(want)} lines")
-    assert len(lines) == len(
-        want
-    ), f"Line count mismatch: got {len(lines)}, expected {len(want)}"
+    assert len(lines) == len(want), (
+        f"Line count mismatch: got {len(lines)}, expected {len(want)}"
+    )
 
     def normalize_json(obj):
         """Remove dynamic fields like time_ms from JSON object for comparison"""
@@ -105,9 +105,9 @@ def normalize_json(obj):
         except json.JSONDecodeError as e:
             assert False, f"Invalid JSON in expected line {i}: {want_line}, error: {e}"
 
-        assert (
-            actual_json == want_json
-        ), f"Line {i} JSON mismatch:\nActual: {actual_json}\nExpected: {want_json}"
+        assert actual_json == want_json, (
+            f"Line {i} JSON mismatch:\nActual: {actual_json}\nExpected: {want_json}"
+        )
 
     print(f"✓ All {len(lines)} JSON lines match expected output")
 

tests/metactl/subcommands/cmd_import.py

@@ -67,9 +67,9 @@ def test_import_subcommand():
         assert False, f"Import command failed with return code {process.returncode}"
 
     # Verify raft directory was created
-    assert os.path.exists(
-        target_raft_dir
-    ), f"Raft directory should exist: {target_raft_dir}"
+    assert os.path.exists(target_raft_dir), (
+        f"Raft directory should exist: {target_raft_dir}"
+    )
     print(f"✓ Raft directory created: {target_raft_dir}")
 
     # Check for raft log files in correct location
@@ -110,9 +110,9 @@ def test_import_subcommand():
 
     # Parse first line to check version
     header_data = json.loads(lines[0])
-    assert (
-        header_data[1]["DataHeader"]["value"]["version"] == "V004"
-    ), "Should import V004 data"
+    assert header_data[1]["DataHeader"]["value"]["version"] == "V004", (
+        "Should import V004 data"
+    )
     print("✓ Imported data version verification passed")
 
     # Check for required sections

tests/metactl/subcommands/cmd_lua_grpc.py

@@ -71,9 +71,9 @@ def test_lua_grpc_client():
     print("expect:", expected_output)
 
     # Check if entire output matches expected value
-    assert (
-        output == expected_output
-    ), f"Expected:\n{expected_output}\n\nActual:\n{output}"
+    assert output == expected_output, (
+        f"Expected:\n{expected_output}\n\nActual:\n{output}"
+    )
 
     print("✓ Lua gRPC client test passed")
 

tests/metactl/subcommands/cmd_lua_spawn_concurrent.py

@@ -53,9 +53,9 @@ def test_spawn_basic():
     ]
 
     for phrase in expected_phrases:
-        assert (
-            phrase in output
-        ), f"Expected phrase '{phrase}' not found in output:\n{output}"
+        assert phrase in output, (
+            f"Expected phrase '{phrase}' not found in output:\n{output}"
+        )
     print("✓ Basic spawn functionality test passed")
 
 

tests/metactl/subcommands/cmd_lua_spawn_grpc.py

@@ -76,9 +76,9 @@ def test_grpc_cross_task_access():
     ]
 
     for phrase in expected_phrases:
-        assert (
-            phrase in output
-        ), f"Expected phrase '{phrase}' not found in output:\n{output}"
+        assert phrase in output, (
+            f"Expected phrase '{phrase}' not found in output:\n{output}"
+        )
 
     print("✓ Cross-task access test passed")
     kill_databend_meta()

tests/metactl/subcommands/cmd_metrics.py

@@ -43,9 +43,9 @@ def verify_metrics_format(result):
             found_metrics.append(metric)
             print(f"✓ Found expected metric: {metric}")
 
-    assert (
-        len(found_metrics) > 0
-    ), f"Should find at least some expected metrics. Found: {found_metrics}"
+    assert len(found_metrics) > 0, (
+        f"Should find at least some expected metrics. Found: {found_metrics}"
+    )
 
     # Verify at least some lines match prometheus format (metric_name value)
     prometheus_pattern = r"^[a-zA-Z_:][a-zA-Z0-9_:]* \d+(\.\d+)?$"

tests/metactl/subcommands/cmd_status.py

@@ -49,15 +49,15 @@ def verify_status_format(result):
 
     # Check if Node info exists with expected format
     node_pattern = r"Node: id=\d+ raft=.+:\d+"
-    assert re.search(
-        node_pattern, result
-    ), "Node format should match 'id=X raft=host:port'"
+    assert re.search(node_pattern, result), (
+        "Node format should match 'id=X raft=host:port'"
+    )
 
     # Check LastApplied format
     last_applied_pattern = r"LastApplied: T\d+-N\d+\.\d+"
-    assert re.search(
-        last_applied_pattern, result
-    ), "LastApplied should match 'TX-NX.X' format"
+    assert re.search(last_applied_pattern, result), (
+        "LastApplied should match 'TX-NX.X' format"
+    )
 
     print(
         f"✓ Status format verification passed: {len(found_fields)}/{len(expected_fields)} fields found"

tests/metactl/subcommands/cmd_transfer_leader.py

@@ -83,9 +83,9 @@ def test_transfer_leader_subcommand():
         print(
             f"✓ Leadership transferred from node {initial_leader} to node {new_leader}"
         )
-        assert (
-            new_leader != initial_leader
-        ), f"Leader should change from {initial_leader}"
+        assert new_leader != initial_leader, (
+            f"Leader should change from {initial_leader}"
+        )
     else:
         print("✓ No leader change detected (acceptable if cluster is stable)")
 
@@ -149,9 +149,9 @@ def test_transfer_leader_with_target():
             f"✓ Leadership transferred from node {initial_leader} to node {new_leader}"
         )
         # Note: The actual new leader might not be exactly the target due to cluster dynamics
-        assert (
-            new_leader != initial_leader
-        ), f"Leader should change from {initial_leader}"
+        assert new_leader != initial_leader, (
+            f"Leader should change from {initial_leader}"
+        )
     else:
         print("✓ No leader change detected (acceptable if transfer was to same node)")
 

tests/metactl/subcommands/cmd_trigger_snapshot.py

@@ -38,15 +38,15 @@ def test_trigger_snapshot():
     time.sleep(2)
 
     # Verify snapshot file is generated
-    assert os.path.exists(
-        snapshot_dir
-    ), f"Snapshot directory does not exist: {snapshot_dir}"
+    assert os.path.exists(snapshot_dir), (
+        f"Snapshot directory does not exist: {snapshot_dir}"
+    )
 
     current_snapshots = glob.glob(f"{snapshot_dir}/*.snap")
     print("Current_snapshots:", current_snapshots)
-    assert (
-        len(current_snapshots) > len(initial_snapshots)
-    ), f"No new snapshot file created. Before: {len(initial_snapshots)}, After: {len(current_snapshots)}"
+    assert len(current_snapshots) > len(initial_snapshots), (
+        f"No new snapshot file created. Before: {len(initial_snapshots)}, After: {len(current_snapshots)}"
+    )
 
     print(f"✓ Snapshot file created: {len(current_snapshots)} total snapshot(s)")
     print("✓ Trigger snapshot test passed")