feat(structure): search and paginate source table picker for copy columns

crates/dbx-core/src/db/clickhouse_driver.rs

@@ -616,20 +616,101 @@ pub async fn list_databases(client: &ChClient) -> Result<Vec<DatabaseInfo>, Stri
 }
 
 pub async fn list_tables(client: &ChClient, database: &str) -> Result<Vec<TableInfo>, String> {
-    let database_lit = clickhouse_literal(database);
-    let sql_with_comment =
-        format!("SELECT name, engine, comment FROM system.tables WHERE database = '{database_lit}' ORDER BY name");
+    list_tables_by_kind(client, database, None, None, None, false).await
+}
+
+/// Lists physical tables with server-side name filtering and pagination.
+pub async fn list_table_objects_filtered(
+    client: &ChClient,
+    database: &str,
+    filter: Option<&str>,
+    limit: Option<usize>,
+    offset: Option<usize>,
+) -> Result<Vec<TableInfo>, String> {
+    list_tables_by_kind(client, database, filter, limit, offset, true).await
+}
+
+async fn list_tables_by_kind(
+    client: &ChClient,
+    database: &str,
+    filter: Option<&str>,
+    limit: Option<usize>,
+    offset: Option<usize>,
+    table_objects_only: bool,
+) -> Result<Vec<TableInfo>, String> {
+    let sql_with_comment = clickhouse_tables_sql(database, filter, limit, offset, table_objects_only, true);
     let result = match ch_query(client, &sql_with_comment, Some(database)).await {
         Ok(result) => result,
         Err(error) => {
             log::debug!("Falling back to ClickHouse table list without comments: {error}");
-            let sql = format!("SELECT name, engine FROM system.tables WHERE database = '{database_lit}' ORDER BY name");
+            let sql = clickhouse_tables_sql(database, filter, limit, offset, table_objects_only, false);
             ch_query(client, &sql, Some(database)).await?
         }
     };
     Ok(result.data.iter().map(|row| clickhouse_table_info_from_row(row)).collect())
 }
 
+fn clickhouse_tables_sql(
+    database: &str,
+    filter: Option<&str>,
+    limit: Option<usize>,
+    offset: Option<usize>,
+    table_objects_only: bool,
+    include_comment: bool,
+) -> String {
+    let columns = if include_comment { "name, engine, comment" } else { "name, engine" };
+    let database_lit = clickhouse_literal(database);
+    let table_kind_filter = if table_objects_only { " AND position(engine, 'View') = 0" } else { "" };
+    let name_filter = clickhouse_table_name_filter_sql(filter, include_comment);
+    let pagination = limit.map(|limit| format!(" LIMIT {limit} OFFSET {}", offset.unwrap_or(0))).unwrap_or_default();
+
+    format!(
+        "SELECT {columns} FROM system.tables WHERE database = '{database_lit}'{table_kind_filter}{name_filter} ORDER BY name{pagination}"
+    )
+}
+
+fn clickhouse_table_name_filter_sql(filter: Option<&str>, include_comment: bool) -> String {
+    let filter = filter.unwrap_or("").trim();
+    if filter.is_empty() {
+        return String::new();
+    }
+
+    let mut patterns = vec![clickhouse_like_contains_pattern(filter)];
+    if crate::sql::fuzzy_filter_enabled(filter) {
+        patterns.push(crate::sql::fuzzy_like_pattern_with_escape(filter, clickhouse_escape_like_literal));
+    }
+    let predicates = patterns
+        .into_iter()
+        .map(|pattern| {
+            // Keep comment-based search parity with the name predicate; the comment
+            // column is absent from the no-comment fallback query, so the extra
+            // predicate must be gated on include_comment.
+            let name_predicate = format!("lowerUTF8(name) LIKE '{}'", clickhouse_literal(&pattern));
+            if include_comment {
+                format!("{name_predicate} OR lowerUTF8(coalesce(comment, '')) LIKE '{}'", clickhouse_literal(&pattern))
+            } else {
+                name_predicate
+            }
+        })
+        .collect::<Vec<_>>();
+    format!(" AND ({})", predicates.join(" OR "))
+}
+
+fn clickhouse_like_contains_pattern(value: &str) -> String {
+    format!("%{}%", clickhouse_escape_like_literal(value))
+}
+
+fn clickhouse_escape_like_literal(value: &str) -> String {
+    let mut escaped = String::with_capacity(value.len() + 1);
+    for ch in value.chars().flat_map(char::to_lowercase) {
+        if ch == '\\' || ch == '%' || ch == '_' {
+            escaped.push('\\');
+        }
+        escaped.push(ch);
+    }
+    escaped
+}
+
 pub async fn list_object_statistics(client: &ChClient, database: &str) -> Result<Vec<ObjectStatistics>, String> {
     let database_lit = clickhouse_literal(database);
     let sql = format!(
@@ -1128,4 +1209,36 @@ mod tests {
         assert_eq!(table.table_type, "VIEW");
         assert_eq!(table.comment, None);
     }
+
+    #[test]
+    fn table_object_query_filters_before_pagination() {
+        let sql = clickhouse_tables_sql("analytics", Some("orders"), Some(102), Some(100), true, true);
+
+        assert!(sql.contains("SELECT name, engine, comment FROM system.tables"));
+        assert!(sql.contains("database = 'analytics'"));
+        assert!(sql.contains("position(engine, 'View') = 0"));
+        assert!(sql.contains("lowerUTF8(name) LIKE '%orders%'"));
+        assert!(sql.contains("lowerUTF8(coalesce(comment, '')) LIKE '%orders%'"));
+        assert!(sql.contains("ORDER BY name LIMIT 102 OFFSET 100"));
+        assert!(sql.find("position(engine, 'View') = 0").unwrap() < sql.find("ORDER BY name").unwrap());
+        assert!(sql.find("lowerUTF8(name) LIKE '%orders%'").unwrap() < sql.find("ORDER BY name").unwrap());
+    }
+
+    #[test]
+    fn table_object_query_supports_fuzzy_name_filtering() {
+        let sql = clickhouse_tables_sql("analytics", Some("ord"), Some(100), None, true, false);
+
+        assert!(sql.contains("lowerUTF8(name) LIKE '%ord%'"));
+        assert!(sql.contains("lowerUTF8(name) LIKE '%o%r%d%'"));
+        assert!(!sql.contains("comment"));
+        assert!(sql.ends_with("ORDER BY name LIMIT 100 OFFSET 0"));
+    }
+
+    #[test]
+    fn table_object_query_matches_fuzzy_filter_against_comments() {
+        let sql = clickhouse_tables_sql("analytics", Some("ord"), Some(100), None, true, true);
+
+        assert!(sql.contains("lowerUTF8(coalesce(comment, '')) LIKE '%ord%'"));
+        assert!(sql.contains("lowerUTF8(coalesce(comment, '')) LIKE '%o%r%d%'"));
+    }
 }

crates/dbx-core/src/db/postgres.rs

@@ -2831,6 +2831,29 @@ pub async fn list_tables_filtered(
     filter: Option<&str>,
     limit: Option<usize>,
     offset: Option<usize>,
+) -> Result<Vec<TableInfo>, String> {
+    list_tables_filtered_by_kind(pool, schema, filter, limit, offset, false).await
+}
+
+/// Lists only table-like relations, excluding views and materialized views while
+/// retaining server-side filtering and pagination.
+pub async fn list_table_objects_filtered(
+    pool: &Pool,
+    schema: &str,
+    filter: Option<&str>,
+    limit: Option<usize>,
+    offset: Option<usize>,
+) -> Result<Vec<TableInfo>, String> {
+    list_tables_filtered_by_kind(pool, schema, filter, limit, offset, true).await
+}
+
+async fn list_tables_filtered_by_kind(
+    pool: &Pool,
+    schema: &str,
+    filter: Option<&str>,
+    limit: Option<usize>,
+    offset: Option<usize>,
+    table_objects_only: bool,
 ) -> Result<Vec<TableInfo>, String> {
     let schema = if schema.is_empty() { "public" } else { schema };
     let filter = filter.unwrap_or("").trim();
@@ -2840,7 +2863,11 @@ pub async fn list_tables_filtered(
     let limit_param = limit.and_then(|value| i64::try_from(value).ok());
     let offset_param = offset.and_then(|value| i64::try_from(value).ok()).unwrap_or(0);
     let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
-    let sql = postgres_tables_sql(limit_param, offset_param);
+    let sql = if table_objects_only {
+        postgres_table_objects_sql(limit_param, offset_param)
+    } else {
+        postgres_tables_sql(limit_param, offset_param)
+    };
     let params: &[(&(dyn tokio_postgres::types::ToSql + Sync), Type)] =
         &[(&schema, Type::VARCHAR), (&filter_pattern, Type::VARCHAR), (&fuzzy_filter_pattern, Type::VARCHAR)];
     // The pagination literals make this SQL vary by page. Use an unnamed typed
@@ -4030,6 +4057,14 @@ fn postgres_table_comment_sql() -> &'static str {
 }
 
 fn postgres_tables_sql(limit: Option<i64>, offset: i64) -> String {
+    postgres_tables_sql_with_kind(limit, offset, false)
+}
+
+fn postgres_table_objects_sql(limit: Option<i64>, offset: i64) -> String {
+    postgres_tables_sql_with_kind(limit, offset, true)
+}
+
+fn postgres_tables_sql_with_kind(limit: Option<i64>, offset: i64, table_objects_only: bool) -> String {
     // PostgreSQL-compatible servers do not agree on the inferred wire types
     // or accepted expression grammar for LIMIT/OFFSET parameters. These values
     // originate as usize and are converted to non-negative i64 literals.
@@ -4039,6 +4074,7 @@ fn postgres_tables_sql(limit: Option<i64>, offset: i64) -> String {
         Some(limit) => format!("LIMIT {limit} OFFSET {offset}"),
         None => format!("OFFSET {offset}"),
     };
+    let relation_kinds = if table_objects_only { "'r','f','p'" } else { "'r','v','m','f','p'" };
     format!(
         "SELECT c.relname AS table_name, \
          CASE c.relkind WHEN 'r' THEN 'BASE TABLE' WHEN 'v' THEN 'VIEW' \
@@ -4052,7 +4088,7 @@ fn postgres_tables_sql(limit: Option<i64>, offset: i64) -> String {
          LEFT JOIN pg_catalog.pg_inherits i ON i.inhrelid = c.oid \
          LEFT JOIN pg_catalog.pg_class pc ON pc.oid = i.inhparent \
          LEFT JOIN pg_catalog.pg_namespace pn ON pn.oid = pc.relnamespace \
-         WHERE n.nspname = $1 AND c.relkind IN ('r','v','m','f','p') \
+         WHERE n.nspname = $1 AND c.relkind IN ({relation_kinds}) \
            AND ($2 = '%%' OR c.relname ILIKE $2 OR ($3 <> '' AND c.relname ILIKE $3)) \
          ORDER BY CASE WHEN pc.relkind = 'p' THEN 1 ELSE 0 END, c.relname \
          {pagination}"
@@ -9831,6 +9867,14 @@ mod tests {
         assert!(sql.contains("FOREIGN TABLE"));
     }
 
+    #[test]
+    fn postgres_table_objects_sql_excludes_views_before_pagination() {
+        let sql = postgres_table_objects_sql(Some(101), 100);
+
+        assert!(sql.contains("c.relkind IN ('r','f','p')"));
+        assert!(sql.contains("LIMIT 101 OFFSET 100"));
+    }
+
     #[test]
     fn postgres_table_comment_sql_targets_single_table() {
         let sql = postgres_table_comment_sql();

crates/dbx-core/src/db/sqlserver.rs

@@ -2066,7 +2066,33 @@ pub async fn list_tables(
     limit: Option<usize>,
     offset: Option<usize>,
 ) -> Result<Vec<TableInfo>, String> {
-    let sql = sqlserver_list_tables_sql(schema, filter, limit, offset);
+    list_tables_by_kind(client, schema, filter, limit, offset, false).await
+}
+
+/// Lists user tables only, preserving server-side filtering and pagination.
+pub async fn list_table_objects(
+    client: &mut SqlServerClient,
+    schema: &str,
+    filter: Option<&str>,
+    limit: Option<usize>,
+    offset: Option<usize>,
+) -> Result<Vec<TableInfo>, String> {
+    list_tables_by_kind(client, schema, filter, limit, offset, true).await
+}
+
+async fn list_tables_by_kind(
+    client: &mut SqlServerClient,
+    schema: &str,
+    filter: Option<&str>,
+    limit: Option<usize>,
+    offset: Option<usize>,
+    table_objects_only: bool,
+) -> Result<Vec<TableInfo>, String> {
+    let sql = if table_objects_only {
+        sqlserver_table_objects_sql(schema, filter, limit, offset)
+    } else {
+        sqlserver_list_tables_sql(schema, filter, limit, offset)
+    };
     let stream = client.query(&*sql, &[]).await.map_err(|e| e.to_string())?;
     let rows = stream.into_first_result().await.map_err(|e| e.to_string())?;
     Ok(rows
@@ -2284,6 +2310,25 @@ fn sqlserver_list_tables_sql(
     filter: Option<&str>,
     limit: Option<usize>,
     offset: Option<usize>,
+) -> String {
+    sqlserver_list_tables_sql_with_kind(schema, filter, limit, offset, false)
+}
+
+fn sqlserver_table_objects_sql(
+    schema: &str,
+    filter: Option<&str>,
+    limit: Option<usize>,
+    offset: Option<usize>,
+) -> String {
+    sqlserver_list_tables_sql_with_kind(schema, filter, limit, offset, true)
+}
+
+fn sqlserver_list_tables_sql_with_kind(
+    schema: &str,
+    filter: Option<&str>,
+    limit: Option<usize>,
+    offset: Option<usize>,
+    table_objects_only: bool,
 ) -> String {
     let filter_clause = filter
         .filter(|value| !value.trim().is_empty())
@@ -2307,7 +2352,9 @@ fn sqlserver_list_tables_sql(
            WHERE ep.major_id = o.object_id AND ep.minor_id = 0 AND ep.name = N'MS_Description') ep";
     let object_visibility = sqlserver_visible_object_predicate();
     let schema_filter = sqlserver_schema_name_predicate(schema, "s.name");
-    let base_where = format!("WHERE {schema_filter} AND o.type IN ('U','V') AND {object_visibility} {filter_clause}");
+    let object_type_predicate = if table_objects_only { "o.type = 'U'" } else { "o.type IN ('U','V')" };
+    let base_where =
+        format!("WHERE {schema_filter} AND {object_type_predicate} AND {object_visibility} {filter_clause}");
     let order_by = "ORDER BY o.name";
 
     // Use SELECT TOP for broad SQL Server version compatibility.
@@ -3317,10 +3364,10 @@ mod tests {
         sqlserver_legacy_wildcard_metadata_query, sqlserver_list_objects_sql, sqlserver_list_schemas_sql,
         sqlserver_list_tables_sql, sqlserver_probe_explicit_alias, sqlserver_query_messages,
         sqlserver_schema_name_predicate, sqlserver_spatial_marker, sqlserver_supports_session_database_switch,
-        sqlserver_table_comment_sql, sqlserver_triggers_sql, sqlserver_visible_object_predicate,
-        strip_dbx_sqlserver_row_number_column, SqlServerDescribedColumn, SqlServerProbeOutputNameOverride,
-        SqlServerRestoredColumn, SqlServerResultSet, SqlServerSpatialColumn, SqlServerTdsEvent,
-        SQLSERVER_COMPLETION_CONTEXT_SQL, SQLSERVER_RESULT_TYPE_PROBE_SQL,
+        sqlserver_table_comment_sql, sqlserver_table_objects_sql, sqlserver_triggers_sql,
+        sqlserver_visible_object_predicate, strip_dbx_sqlserver_row_number_column, SqlServerDescribedColumn,
+        SqlServerProbeOutputNameOverride, SqlServerRestoredColumn, SqlServerResultSet, SqlServerSpatialColumn,
+        SqlServerTdsEvent, SQLSERVER_COMPLETION_CONTEXT_SQL, SQLSERVER_RESULT_TYPE_PROBE_SQL,
     };
     use crate::types::{
         CompletionAssistantMatchMode, CompletionAssistantObjectKind, CompletionAssistantRequest, QueryResult,
@@ -4175,6 +4222,16 @@ mod tests {
         assert!(sql.contains("SELECT TOP (200)"));
     }
 
+    #[test]
+    fn sqlserver_table_objects_sql_excludes_views_before_pagination() {
+        let sql = sqlserver_table_objects_sql("dbo", Some("orders"), Some(101), Some(100));
+
+        assert!(sql.contains("o.type = 'U'"));
+        assert!(!sql.contains("o.type IN ('U','V')"));
+        assert!(sql.contains("ROW_NUMBER() OVER (ORDER BY o.name)"));
+        assert!(sql.contains("__dbx_rn > 100 AND __dbx_rn <= 201"));
+    }
+
     #[test]
     fn sqlserver_list_tables_filter_escapes_like_literals() {
         let sql = sqlserver_list_tables_sql("dbo", Some("Temp_Table[%]"), Some(200), None);

crates/dbx-core/src/schema.rs

@@ -2186,7 +2186,8 @@ async fn list_tables_once(
 
     {
         let connections = state.connections.read().await;
-        if let Some(PoolKind::ExternalDriver { config, session, .. }) = connections.get(&pool_key) {
+        if let Some(PoolKind::ExternalDriver { driver_id, config, session }) = connections.get(&pool_key) {
+            let driver_id = driver_id.clone();
             let config = config.clone();
             let session = session.clone();
             drop(connections);
@@ -2212,6 +2213,12 @@ async fn list_tables_once(
             if let Some(object_types) = object_types {
                 params["object_types"] = serde_json::json!(object_types);
             }
+            if let Some(limit) = limit {
+                params["limit"] = serde_json::json!(limit);
+            }
+            if let Some(offset) = offset {
+                params["offset"] = serde_json::json!(offset);
+            }
             return session
                 .invoke_with_timeout::<Vec<db::TableInfo>>(
                     "listTables",
@@ -2219,7 +2226,14 @@ async fn list_tables_once(
                     agent_metadata_timeout(Some(config.as_ref())),
                 )
                 .await
-                .map(|tables| filter_table_infos(tables, filter, limit, offset, object_types, table_name_filter));
+                .map(|tables| {
+                    let final_offset = if external_driver_paging_likely_applied(&driver_id, limit, tables.len()) {
+                        Some(0)
+                    } else {
+                        offset
+                    };
+                    filter_table_infos(tables, filter, limit, final_offset, object_types, table_name_filter)
+                });
         }
         #[cfg(feature = "duckdb-sidecar")]
         if let Some(client) = extract_pool!(&connections, &pool_key, DuckDbWorker) {
@@ -2233,6 +2247,17 @@ async fn list_tables_once(
         }
         if let Some(client) = extract_pool!(&connections, &pool_key, ClickHouse) {
             drop(connections);
+            if requests_table_objects_only(object_types) && table_name_filter.is_none_or(TableNameFilter::is_empty) {
+                return db::clickhouse_driver::list_table_objects_filtered(
+                    &client,
+                    clickhouse_metadata_database(database, schema),
+                    filter,
+                    limit,
+                    offset,
+                )
+                .await
+                .map(|tables| filter_table_infos(tables, None, None, None, object_types, None));
+            }
             return db::clickhouse_driver::list_tables(&client, clickhouse_metadata_database(database, schema))
                 .await
                 .map(|tables| filter_table_infos(tables, filter, limit, offset, object_types, table_name_filter));
@@ -2266,6 +2291,13 @@ async fn list_tables_once(
                 .map(|tables| filter_table_infos(tables, filter, limit, offset, object_types, table_name_filter));
             }
         }
+        if requests_table_objects_only(object_types) && table_name_filter.is_none_or(TableNameFilter::is_empty) {
+            if let Some(client) = extract_pool!(&connections, &pool_key, SqlServer) {
+                drop(connections);
+                let mut client = lock_sqlserver_metadata_client(&client).await?;
+                return db::sqlserver::list_table_objects(&mut client, schema, filter, limit, offset).await;
+            }
+        }
         if object_types.is_some() || table_name_filter.is_some_and(|filter| !filter.is_empty()) {
             if let Some(client) = extract_pool!(&connections, &pool_key, SqlServer) {
                 drop(connections);
@@ -2513,7 +2545,9 @@ async fn list_tables_once(
             }
         }
         PoolKind::Postgres(p) => {
-            if object_types.is_some() || table_name_filter.is_some_and(|filter| !filter.is_empty()) {
+            if requests_table_objects_only(object_types) && table_name_filter.is_none_or(TableNameFilter::is_empty) {
+                db::postgres::list_table_objects_filtered(p, schema, filter, limit, offset).await
+            } else if object_types.is_some() || table_name_filter.is_some_and(|filter| !filter.is_empty()) {
                 db::postgres::list_tables_filtered(p, schema, filter, None, None)
                     .await
                     .map(|tables| filter_table_infos(tables, filter, limit, offset, object_types, table_name_filter))
@@ -2694,6 +2728,10 @@ fn normalize_table_info_object_type(value: &str) -> String {
     "TABLE".to_string()
 }
 
+fn requests_table_objects_only(object_types: Option<&[String]>) -> bool {
+    object_types.is_some_and(|types| types.len() == 1 && normalize_table_info_object_type(&types[0]) == "TABLE")
+}
+
 fn uses_presto_like_information_schema_tables(db_type: &DatabaseType) -> bool {
     matches!(db_type, DatabaseType::PrestoSql | DatabaseType::Trino)
 }
@@ -3993,6 +4031,15 @@ for line in sys.stdin:
         assert!(!super::agent_paging_likely_applied(true, None, 120));
     }
 
+    #[test]
+    fn external_driver_paging_detection_avoids_double_offset_for_jdbc_only() {
+        assert!(super::external_driver_paging_likely_applied("jdbc", Some(500), 500));
+        assert!(super::external_driver_paging_likely_applied("jdbc", Some(500), 120));
+        assert!(!super::external_driver_paging_likely_applied("jdbc", Some(500), 501));
+        assert!(!super::external_driver_paging_likely_applied("jdbc", None, 120));
+        assert!(!super::external_driver_paging_likely_applied("other", Some(500), 120));
+    }
+
     #[test]
     fn filter_visible_schema_names_preserves_database_order() {
         let schemas = vec!["APP".to_string(), "SYS".to_string(), "REPORTING".to_string()];
@@ -7592,6 +7639,10 @@ fn agent_paging_likely_applied(enabled: bool, limit: Option<usize>, returned_len
     enabled && limit.is_some_and(|limit| returned_len <= limit)
 }
 
+fn external_driver_paging_likely_applied(driver_id: &str, limit: Option<usize>, returned_len: usize) -> bool {
+    driver_id == "jdbc" && limit.is_some_and(|limit| returned_len <= limit)
+}
+
 fn mysql_show_metadata_database_for_config<'a>(config: Option<&ConnectionConfig>, database: &'a str) -> &'a str {
     if config.is_some_and(db::manticoresearch::is_config) {
         ""