SkardiLabs
diff --git a/‎README.md‎
Lines changed: 1 addition & 0 deletions b/‎README.md‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎crates/cli/src/main.rs‎
Lines changed: 64 additions & 37 deletions b/‎crates/cli/src/main.rs‎
Lines changed: 64 additions & 37 deletions
diff --git a/‎crates/server/src/config.rs‎
Lines changed: 2 additions & 0 deletions b/‎crates/server/src/config.rs‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎crates/server/src/optimizer_registry.rs‎
Lines changed: 59 additions & 0 deletions b/‎crates/server/src/optimizer_registry.rs‎
Lines changed: 59 additions & 0 deletions
@@ -330,6 +330,7 @@ For end-to-end walkthroughs — RAG, recommendations, an agent-native wiki, a si
 | S3 / GCS / Azure | Read | No | CSV, Parquet, Lance from object stores | [docs/S3_USAGE.md](docs/S3_USAGE.md) |
 | Apache Iceberg | Read | No | Schema evolution, partition pruning | [docs/iceberg/](docs/iceberg/) |
 | InfluxDB 3 | Read | No | Time-series measurements over Arrow Flight SQL | [docs/influxdb/](docs/influxdb/) |
+| Open Connector | Read | Yes | SaaS resources as stable SQL tables via a self-hosted [Open Connector](https://github.com/oomol-lab/open-connector) gateway; `open_connector_query` / `open_connector_scan` UDTFs, filter + limit pushdown, bounded TTL cache (foundation + synthetic pack today; provider packs rolling out) | [docs/open-connector.md](docs/open-connector.md) |
 | Documents | Read | No | PDF/Office/ODF/image -> per-page markdown, tables, images (local directories; `documents` feature) | [docs/documents.md](docs/documents.md) |
 
 ---
 
@@ -52,7 +52,10 @@ use skardi::sources::providers::{
     lance::register_lance_table,
     mongo::register_mongo_tables,
     mysql::register_mysql_tables,
-    open_connector::{OpenConnectorConfig, register_open_connector_tables},
+    open_connector::{
+        OpenConnectorConfig, OpenConnectorGateways, register_open_connector_tables,
+        register_open_connector_udtfs,
+    },
     sqlite::{
         register_sqlite_fts_udtf, register_sqlite_knn_udtf, register_sqlite_tables,
         register_vec_to_binary_udf,
@@ -479,9 +482,10 @@ impl UrlTableFactory for SkardiUrlTableFactory {
 }
 
 /// Create a new SessionContext with custom URL table support (built-in files + Lance)
-/// and the `lance_knn` / `pg_knn` UDTFs registered.
-fn new_session_context() -> (SessionContext, DatasetRegistry) {
+/// and the `lance_knn` / `pg_knn` / Open Connector UDTFs registered.
+fn new_session_context() -> (SessionContext, DatasetRegistry, OpenConnectorGateways) {
     let dataset_registry: DatasetRegistry = Arc::new(RwLock::new(HashMap::new()));
+    let open_connector_gateways = OpenConnectorGateways::default();
     let session_store = SessionStore::new();
     let factory = Arc::new(SkardiUrlTableFactory::new(
         session_store,
@@ -516,6 +520,9 @@ fn new_session_context() -> (SessionContext, DatasetRegistry) {
     register_sqlite_knn_udtf(&ctx, Arc::clone(&dataset_registry));
     register_sqlite_fts_udtf(&ctx, Arc::clone(&dataset_registry));
     register_vec_to_binary_udf(&mut ctx);
+    // Open Connector UDTFs plan against the gateway state that
+    // register_open_connector_tables fills in during ctx registration.
+    register_open_connector_udtfs(&ctx, Arc::clone(&open_connector_gateways));
 
     // Embedding UDFs (gated by feature flags, lazy model loading on first call).
     #[cfg(feature = "onnx")]
@@ -544,7 +551,7 @@ fn new_session_context() -> (SessionContext, DatasetRegistry) {
         registry.register_chunk_udf(&mut ctx);
     }
 
-    (ctx, dataset_registry)
+    (ctx, dataset_registry, open_connector_gateways)
 }
 
 /// Resolve a path string: if relative (and not remote), resolve against cwd.
@@ -733,13 +740,19 @@ async fn load_and_register_all(
     ctx_path: &Path,
     session_ctx: &mut SessionContext,
     dataset_registry: &DatasetRegistry,
+    open_connector_gateways: &OpenConnectorGateways,
 ) -> Result<LocalContextConfig> {
     let config = read_context_file(ctx_path)?;
 
     for source in &config.data_sources {
-        register_source(session_ctx, source, dataset_registry)
-            .await
-            .with_context(|| format!("Failed to register data source '{}'", source.name))?;
+        register_source(
+            session_ctx,
+            source,
+            dataset_registry,
+            open_connector_gateways,
+        )
+        .await
+        .with_context(|| format!("Failed to register data source '{}'", source.name))?;
     }
 
     Ok(config)
@@ -774,6 +787,7 @@ async fn register_source(
     session_ctx: &mut SessionContext,
     source: &LocalDataSource,
     dataset_registry: &DatasetRegistry,
+    open_connector_gateways: &OpenConnectorGateways,
 ) -> Result<()> {
     let source_type = source.source_type.to_lowercase();
 
@@ -957,6 +971,7 @@ async fn register_source(
                 source.open_connector.as_ref(),
                 source.is_read_write(),
                 source.hierarchy_level,
+                Some(open_connector_gateways),
             )
             .await
             .with_context(|| format!("Failed to register Open Connector '{}'", source.name))?;
@@ -1222,8 +1237,9 @@ async fn show_schema(
     table_filter: Option<&str>,
     out: &mut dyn Write,
 ) -> Result<()> {
-    let (mut session_ctx, dataset_registry) = new_session_context();
-    let config = load_and_register_all(ctx_path, &mut session_ctx, &dataset_registry).await?;
+    let (mut session_ctx, dataset_registry, oc_gateways) = new_session_context();
+    let config =
+        load_and_register_all(ctx_path, &mut session_ctx, &dataset_registry, &oc_gateways).await?;
 
     // Build the semantics registry from the same inputs the server uses:
     //   - the ctx-inline `description` field on each data source (fallback), and
@@ -1368,12 +1384,13 @@ fn source_name_for<'a>(
 /// data sources first. If no context file is found, run the query in a bare session with
 /// URL table support (allowing direct file/lance paths in SQL).
 async fn run_query(ctx_override: Option<PathBuf>, sql: &str) -> Result<()> {
-    let (mut session_ctx, dataset_registry) = new_session_context();
+    let (mut session_ctx, dataset_registry, oc_gateways) = new_session_context();
 
     // Try to load context file, but don't fail if not found when no explicit --ctx was given
     match resolve_ctx_path(ctx_override.as_deref()) {
         Some(ctx_path) if ctx_path.exists() => {
-            load_and_register_all(&ctx_path, &mut session_ctx, &dataset_registry).await?;
+            load_and_register_all(&ctx_path, &mut session_ctx, &dataset_registry, &oc_gateways)
+                .await?;
         }
         Some(ctx_path) if ctx_override.is_some() => {
             anyhow::bail!("Context file not found: {}", ctx_path.display());
@@ -1525,9 +1542,9 @@ async fn run_pipeline_with_params(
         .with_context(|| format!("Failed to render SQL for pipeline '{}'", pipeline_name))?;
 
     // 3. Build a SessionContext with ctx data sources registered, then execute.
-    let (mut session_ctx, dataset_registry) = new_session_context();
+    let (mut session_ctx, dataset_registry, oc_gateways) = new_session_context();
     if let Some(p) = &ctx_path_for_load {
-        load_and_register_all(p, &mut session_ctx, &dataset_registry).await?;
+        load_and_register_all(p, &mut session_ctx, &dataset_registry, &oc_gateways).await?;
     }
 
     auto_register_object_stores_from_sql(&session_ctx, &sql)?;
@@ -2879,10 +2896,15 @@ spec:
 
         #[tokio::test]
         async fn errors_without_connection_string() {
-            let (mut session_ctx, registry) = new_session_context();
-            let err = register_source(&mut session_ctx, &dynamodb_source(None), &registry)
-                .await
-                .unwrap_err();
+            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let err = register_source(
+                &mut session_ctx,
+                &dynamodb_source(None),
+                &registry,
+                &oc_gateways,
+            )
+            .await
+            .unwrap_err();
             let msg = format!("{err:?}");
             assert!(
                 msg.contains("connection_string (endpoint URL) required"),
@@ -2892,9 +2914,9 @@ spec:
 
         #[tokio::test]
         async fn errors_without_options() {
-            let (mut session_ctx, registry) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) = new_session_context();
             let source = dynamodb_source(Some("http://localhost:8000"));
-            let err = register_source(&mut session_ctx, &source, &registry)
+            let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
                 .await
                 .unwrap_err();
             let msg = format!("{err:?}");
@@ -2927,10 +2949,15 @@ spec:
 
         #[tokio::test]
         async fn errors_without_connection_string() {
-            let (mut session_ctx, registry) = new_session_context();
-            let err = register_source(&mut session_ctx, &clickhouse_source(None), &registry)
-                .await
-                .unwrap_err();
+            let (mut session_ctx, registry, oc_gateways) = new_session_context();
+            let err = register_source(
+                &mut session_ctx,
+                &clickhouse_source(None),
+                &registry,
+                &oc_gateways,
+            )
+            .await
+            .unwrap_err();
             let msg = format!("{err:?}");
             assert!(
                 msg.contains("connection_string required"),
@@ -2943,10 +2970,10 @@ spec:
             // The provider is the single enforcement point for the read-only
             // invariant — the CLI must reject read_write exactly like the
             // server's UnsupportedWriteMode.
-            let (mut session_ctx, registry) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) = new_session_context();
             let mut source = clickhouse_source(Some("http://127.0.0.1:1"));
             source.access_mode = Some("read_write".to_string());
-            let err = register_source(&mut session_ctx, &source, &registry)
+            let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
                 .await
                 .unwrap_err();
             let msg = format!("{err:?}");
@@ -2988,9 +3015,9 @@ bindings:
 
         #[tokio::test]
         async fn errors_without_connection_string() {
-            let (mut session_ctx, registry) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) = new_session_context();
             let source = open_connector_source(None, Some(VALID_CONFIG));
-            let err = register_source(&mut session_ctx, &source, &registry)
+            let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
                 .await
                 .unwrap_err();
             let msg = format!("{err:?}");
@@ -3004,11 +3031,11 @@ bindings:
         async fn errors_with_table_hierarchy() {
             // hierarchy_level defaults to Table; the CLI must reject it with
             // a clear message, not the provider's wrapped error.
-            let (mut session_ctx, registry) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) = new_session_context();
             let mut source =
                 open_connector_source(Some("http://localhost:3000"), Some(VALID_CONFIG));
             source.hierarchy_level = HierarchyLevel::Table;
-            let err = register_source(&mut session_ctx, &source, &registry)
+            let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
                 .await
                 .unwrap_err();
             let msg = format!("{err:?}");
@@ -3020,9 +3047,9 @@ bindings:
 
         #[tokio::test]
         async fn errors_without_typed_config() {
-            let (mut session_ctx, registry) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) = new_session_context();
             let source = open_connector_source(Some("http://localhost:3000"), None);
-            let err = register_source(&mut session_ctx, &source, &registry)
+            let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
                 .await
                 .unwrap_err();
             let msg = format!("{err:?}");
@@ -3038,11 +3065,11 @@ bindings:
             // The provider is the single enforcement point for the
             // read-only invariant — the CLI must reject read_write exactly
             // like the server's UnsupportedWriteMode.
-            let (mut session_ctx, registry) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) = new_session_context();
             let mut source =
                 open_connector_source(Some("http://localhost:3000"), Some(VALID_CONFIG));
             source.access_mode = Some("read_write".to_string());
-            let err = register_source(&mut session_ctx, &source, &registry)
+            let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
                 .await
                 .unwrap_err();
             let msg = format!("{err:?}");
@@ -3051,11 +3078,11 @@ bindings:
 
         #[tokio::test]
         async fn errors_when_typed_config_on_wrong_type() {
-            let (mut session_ctx, registry) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) = new_session_context();
             let mut source =
                 open_connector_source(Some("http://localhost:3000"), Some(VALID_CONFIG));
             source.source_type = "csv".to_string();
-            let err = register_source(&mut session_ctx, &source, &registry)
+            let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
                 .await
                 .unwrap_err();
             let msg = format!("{err:?}");
@@ -3069,11 +3096,11 @@ bindings:
         async fn errors_when_token_env_missing() {
             // With the config valid, the next failure is the unset runtime
             // token — before any network call to the (unroutable) gateway.
-            let (mut session_ctx, registry) = new_session_context();
+            let (mut session_ctx, registry, oc_gateways) = new_session_context();
             let config =
                 VALID_CONFIG.replace("OPEN_CONNECTOR_TOKEN", "SKARDI_CLI_TEST_OC_TOKEN_UNSET");
             let source = open_connector_source(Some("http://127.0.0.1:1"), Some(config.as_str()));
-            let err = register_source(&mut session_ctx, &source, &registry)
+            let err = register_source(&mut session_ctx, &source, &registry, &oc_gateways)
                 .await
                 .unwrap_err();
             let msg = format!("{err:?}");
 
@@ -1421,13 +1421,15 @@ async fn register_data_source(
             // `validate_data_sources` already guarantees the config block is
             // present and access is read-only; the provider re-checks both so
             // the CLI path (which has no such validation layer) is covered.
+            let oc_gateways = optimizer_registry.map(|r| r.open_connector_gateways());
             register_open_connector_tables(
                 session_ctx,
                 &source.name,
                 connection_string,
                 source.open_connector.as_ref(),
                 source.access_mode.is_read_write(),
                 source.hierarchy_level,
+                oc_gateways.as_ref(),
             )
             .await
             .map_err(|e| {
 
@@ -13,6 +13,9 @@ use datafusion::prelude::SessionContext;
 use lance::dataset::Dataset;
 use skardi::sources::providers::lance::{register_lance_fts_udtf, register_lance_knn_udtf};
 use skardi::sources::providers::mongo::fts_table_function::register_mongo_fts_udtf;
+use skardi::sources::providers::open_connector::{
+    OpenConnectorGateways, register_open_connector_udtfs,
+};
 use skardi::sources::providers::seekdb::{register_seekdb_fts_udtf, register_seekdb_knn_udtf};
 use skardi::sources::providers::sqlite::{
     register_sqlite_fts_udtf, register_sqlite_knn_udtf, register_vec_to_binary_udf,
@@ -35,13 +38,18 @@ pub struct OptimizerRegistry {
     /// Stores both Lance datasets and Postgres entries, used by
     /// lance_knn, lance_fts, pg_knn, and pg_fts table functions.
     dataset_registry: DatasetRegistry,
+    /// Open Connector gateway state indexed by gateway (data source) name,
+    /// filled during data-source registration and used by the
+    /// open_connector_query / open_connector_scan table functions.
+    open_connector_gateways: OpenConnectorGateways,
 }
 
 impl OptimizerRegistry {
     /// Create a new registry
     pub fn new() -> Self {
         Self {
             dataset_registry: Arc::new(RwLock::new(HashMap::new())),
+            open_connector_gateways: OpenConnectorGateways::default(),
         }
     }
 
@@ -109,6 +117,13 @@ impl OptimizerRegistry {
             self.register_seekdb_functions(ctx)?;
         }
 
+        // Register Open Connector table functions
+        if source_types.contains(&DataSourceType::OpenConnector) {
+            tracing::info!("Registering Open Connector table functions");
+            register_open_connector_udtfs(ctx, self.open_connector_gateways());
+            tracing::info!("✓ Registered open_connector_query and open_connector_scan");
+        }
+
         Ok(())
     }
 
@@ -204,6 +219,12 @@ impl OptimizerRegistry {
         Arc::clone(&self.dataset_registry)
     }
 
+    /// Get a clone of the Open Connector gateway map to share with
+    /// `register_open_connector_tables` and the Open Connector UDTFs.
+    pub fn open_connector_gateways(&self) -> OpenConnectorGateways {
+        Arc::clone(&self.open_connector_gateways)
+    }
+
     /// Alias for `datasets()` — used when passing to `register_postgres_tables`.
     pub fn pg_knn_pools(&self) -> DatasetRegistry {
         self.datasets()
@@ -256,6 +277,44 @@ mod tests {
         assert!(result.is_ok());
     }
 
+    #[tokio::test]
+    async fn test_register_udfs_with_open_connector_source() {
+        use crate::config::DataSource;
+        use std::path::PathBuf;
+
+        let registry = OptimizerRegistry::new();
+        let mut ctx = SessionContext::new();
+        let data_sources = vec![DataSource {
+            name: "saas".to_string(),
+            source_type: DataSourceType::OpenConnector,
+            path: PathBuf::new(),
+            connection_string: Some("http://open-connector:3000".to_string()),
+            schema: None,
+            options: None,
+            access_mode: crate::config::AccessMode::default(),
+            enable_cache: false,
+            hierarchy_level: Default::default(),
+            description: None,
+            open_connector: None,
+        }];
+
+        registry
+            .register_udfs(&mut ctx, &data_sources)
+            .expect("register open connector UDTFs");
+
+        // The functions are registered; with no gateway handle published
+        // (this test never ran data-source registration) planning fails with
+        // the targeted gateway error rather than "function not found".
+        let err = ctx
+            .sql("SELECT * FROM open_connector_query('saas', 'mock.items', '{}')")
+            .await
+            .expect_err("no gateway state registered");
+        assert!(
+            err.to_string().contains("gateway 'saas' is not registered"),
+            "unexpected error: {err}"
+        );
+    }
+
     #[tokio::test]
     async fn test_register_udfs_with_csv_only() {
         use crate::config::DataSource;