fix(clickhouse): address PR #157 review findings

- Catalog mode is now best-effort (mirrors DynamoDB): a table whose
  schema fetch fails (broken view, permissions gap) is skipped with a
  warning instead of failing server startup. Stream-like engine tables
  (Kafka/RabbitMQ/NATS/FileLog) and materialized-view inner tables
  (.inner* names) are filtered out up front.
- Introspection is one batched system.tables query instead of one
  query per database, so many databases no longer blow the single
  5s retry window. Adds a direct dep on the same `clickhouse` client
  crate the upstream provider already uses.
- URL-embedded credentials are a hard config error instead of a
  warning: the pool ignores them (auth would fail confusingly late)
  and the connection string is logged and exposed by the data-sources
  API. Connection strings are validated before they are logged.
- `database` joins `table`/`schema` in the catalog-mode conflicting
  options list; added ClickHouse config-validation tests (read_write
  rejection, database-in-catalog rejection).
- count(*) empty-projection scans stream the narrowest fixed-width
  column instead of column 0; the no-aggregate-pushdown limitation is
  documented in the README.
- Per-table schema inference is wrapped in retry_with_timeout so a
  hanging endpoint can't stall startup.
- Docs: schema inference is a SELECT ... LIMIT 0 probe, not DESCRIBE
  TABLE; federated_stock_value rounds stock_value to match the README.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: abbccdda

Cargo.lock

@@ -26,6 +26,7 @@ arrow-json = { version = "57.0.1" }
 async-trait = "0.1.88"
 axum = "0.7"
 cookie = "0.18"
+clickhouse = "0.14"
 ctor = "0.4.1"
 datafusion = { version = "52.1.0", default-features = false, features = [
     "nested_expressions",

Cargo.toml

@@ -761,10 +761,11 @@ fn validate_data_sources(data_sources: &[DataSource]) -> Result<()> {
         }
 
         // Catalog mode must not mix with per-table / per-schema options
+        // ("database" is ClickHouse's schema-analog spelling)
         if CATALOG_SUPPORTED_SOURCES.contains(&source.source_type)
             && source.hierarchy_level == HierarchyLevel::Catalog
         {
-            for conflicting in &["table", "schema"] {
+            for conflicting in &["table", "schema", "database"] {
                 if source
                     .options
                     .as_ref()
@@ -2374,6 +2375,72 @@ options:
         );
     }
 
+    fn clickhouse_source(
+        name: &str,
+        options: Option<HashMap<String, String>>,
+        access_mode: AccessMode,
+    ) -> DataSource {
+        DataSource {
+            name: name.to_string(),
+            source_type: DataSourceType::Clickhouse,
+            path: PathBuf::new(),
+            connection_string: Some("http://localhost:8123".to_string()),
+            schema: None,
+            options,
+            hierarchy_level: HierarchyLevel::default(),
+            access_mode,
+            enable_cache: false,
+            description: None,
+        }
+    }
+
+    #[test]
+    fn validate_rejects_clickhouse_read_write() {
+        let mut options = HashMap::new();
+        options.insert("table".to_string(), "events".to_string());
+        let source = clickhouse_source("events", Some(options), AccessMode::ReadWrite);
+        let err = validate_data_sources(&[source]).unwrap_err();
+        let config_err = err.downcast_ref::<ConfigError>().unwrap();
+        assert!(
+            matches!(
+                config_err,
+                ConfigError::UnsupportedWriteMode { name, source_type }
+                    if name == "events" && *source_type == DataSourceType::Clickhouse
+            ),
+            "got {config_err}"
+        );
+    }
+
+    #[test]
+    fn validate_accepts_clickhouse_table_mode_with_database_option() {
+        let mut options = HashMap::new();
+        options.insert("table".to_string(), "events".to_string());
+        options.insert("database".to_string(), "analytics".to_string());
+        let source = clickhouse_source("events", Some(options), AccessMode::ReadOnly);
+        validate_data_sources(&[source]).expect("table mode accepts a database option");
+    }
+
+    #[test]
+    fn validate_rejects_clickhouse_catalog_database_option() {
+        // "database" is ClickHouse's schema-analog option; letting it through
+        // in catalog mode would silently change the pool's default database.
+        let mut options = HashMap::new();
+        options.insert("database".to_string(), "analytics".to_string());
+        let mut source = clickhouse_source("ch", Some(options), AccessMode::ReadOnly);
+        source.hierarchy_level = HierarchyLevel::Catalog;
+
+        let err = validate_data_sources(&[source]).unwrap_err();
+        let config_err = err.downcast_ref::<ConfigError>().unwrap();
+        assert!(
+            matches!(
+                config_err,
+                ConfigError::CatalogModeConflictingOptions { name, option }
+                    if name == "ch" && option == "database"
+            ),
+            "got {config_err}"
+        );
+    }
+
     #[test]
     fn unsupported_write_mode_error_lists_dynamodb() {
         let err = ConfigError::UnsupportedWriteMode {

crates/server/src/config.rs

@@ -59,6 +59,9 @@ text-splitter = { version = "0.30", features = ["markdown"], optional = true }
 # sources
 aws-config = { version = "1", features = ["behavior-version-latest"] }
 aws-sdk-dynamodb = "1"
+# Direct dep on the same client `datafusion-table-providers` uses, so catalog
+# introspection can run one batched `system.tables` query instead of N+1.
+clickhouse = { workspace = true }
 datafusion-federation = { workspace = true }
 datafusion-table-providers = { workspace = true }
 derivative = "2.2.0"