Merge origin/main into add_RSS_Feed

Brings in the Open Connector GitHub source pack (#168), UDTF string-arg
extractor refactor (#169), POST /query endpoint (#158), and the squash-merged
spec snapshot from #163 — the branch's newer spec revision is kept as-is.

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

Author: gracexmatin

Cargo.lock

@@ -65,7 +65,6 @@ rstest = "0.25.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 serde_yaml = "0.9"
-sqlparser = "0.55.0"
 sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio", "tls-rustls", "postgres"] }
 tempfile = "3.23.0"
 tokio = { version = "1.44.2", features = ["macros", "rt", "sync"] }

Cargo.toml

@@ -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; GitHub pack (repos, issues, PRs, reviews, commits, workflow runs, releases — [guide](docs/open-connector-github.md)), `open_connector_query` / `open_connector_scan` UDTFs, filter + limit pushdown, bounded TTL cache (more provider packs rolling out) | [docs/open-connector.md](docs/open-connector.md), [demo](docs/open-connector/) |
 | Documents | Read | No | PDF/Office/ODF/image -> per-page markdown, tables, images (local directories; `documents` feature) | [docs/documents.md](docs/documents.md) |
 
 ---

README.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:?}");

crates/cli/src/main.rs

@@ -178,8 +178,15 @@ impl TableProvider for AuthSessionsTable {
 // Registration helper
 // ---------------------------------------------------------------------------
 
+/// Schema the auth virtual tables (`users`, `sessions`) register under.
+///
+/// Single source of truth for the reserved schema name: the ad-hoc `/query`
+/// policy denies exactly this schema (see `config::adhoc_policy_from_sources`),
+/// so the two must never drift apart.
+pub const AUTH_SCHEMA: &str = "auth";
+
 /// Register `auth.users` and `auth.sessions` virtual tables into the given
-/// DataFusion `SessionContext` under a dedicated `auth` schema.
+/// DataFusion `SessionContext` under the [`AUTH_SCHEMA`] schema.
 pub fn register_auth_tables(
     ctx: &mut SessionContext,
     auth: Arc<BetterAuth<DieselSqliteAdapter>>,
@@ -204,12 +211,14 @@ pub fn register_auth_tables(
         .catalog("datafusion")
         .ok_or_else(|| anyhow::anyhow!("Default catalog 'datafusion' not found"))?;
 
-    if catalog.schema("auth").is_some() {
-        return Err(anyhow::anyhow!("Auth schema 'auth' is already registered"));
+    if catalog.schema(AUTH_SCHEMA).is_some() {
+        return Err(anyhow::anyhow!(
+            "Auth schema '{AUTH_SCHEMA}' is already registered"
+        ));
     }
 
     catalog
-        .register_schema("auth", schema)
+        .register_schema(AUTH_SCHEMA, schema)
         .map_err(|e| anyhow::anyhow!("Failed to register auth schema: {}", e))?;
 
     tracing::info!("Registered DataFusion tables: auth.users, auth.sessions");