refactor(sources): share the UDTF string-argument extractor (#169)

* refactor(sources): share the UDTF string-argument extractor

Nine per-provider Expr-to-string extractors (lance knn/fts, sqlite,
seekdb, pg fts/knn, mongo, open_connector) had drifted in both error
wording ("sqlite: 'x'", "lance_knn: x", "mongo_fts: 'x' ... got {:?}")
and NULL handling — some accepted NULL as the pipeline schema-inference
placeholder, others rejected it, with the difference nowhere stated.

One shared module (providers/udtf_args.rs) now carries the three
deliberate semantics, named at each call site:

- string_arg: NULL is the pipeline {param} placeholder, yields "" (the
  prior behavior of lance_fts / sqlite / seekdb / pg_fts / mongo);
- optional_string_arg: NULL means "argument not provided" (lance_fts's
  or-null variant);
- strict_string_arg: NULL rejected, for arguments that determine the
  planned schema where a placeholder cannot produce a plan (lance_knn,
  pg_knn, and the open_connector UDTFs keep their prior strictness).

Error wording is uniform ("{fn}: '{arg}' must be a string literal");
each provider keeps its exact prior NULL semantics, so this changes no
behavior beyond message text. Addresses the non-blocking cleanup note
from the #165 review. The ColumnarValue-based extractors in model/
(scalar-UDF context, different signature) are out of scope.

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

* refactor(sources): share the UDTF JSON-object argument parser

open_connector_query's resource_json and open_connector_scan's
input_json duplicated the serde_json::from_str + is_object() validation
with already-diverging messages ("of resource inputs" vs "of action
inputs"). One parse_json_object(fn_name, arg, raw) helper keeps the two
diagnostics in lockstep; tests pin the non-object rejection for both
functions. Addresses the second non-blocking cleanup note from the
#165 review.

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

* fix(sources): drop the doc comment orphaned by the extract_string removal

Deleting sqlite's extract_string left its two doc-comment lines
attached to the unrelated quote_sqlite_table below it. The other
providers' section headers were checked and still head real content.

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

* fix(sources): keep the per-argument context in the shared JSON-object error

Sharing parse_json_object flattened "of resource inputs" / "of action
inputs" into a generic "of input fields". Thread the caller's noun
through so the shared implementation and the specific diagnostic
coexist; the non-object tests now pin each function's full wording.

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

* fix(sources): stop pg_knn swallowing malformed inline filters

The 6th argument was extracted with `.ok()` (pre-existing on main), so
any non-string filter was silently dropped and the query returned
unfiltered rows. Use optional_string_arg: NULL still means "no filter"
(the pipeline placeholder), everything else that isn't a string literal
now fails planning with the targeted message — matching lance_knn,
which always propagated. Tests pin both paths without needing a live
Postgres.

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

* style(sources): use the plan_err macro pair throughout parse_json_object

The parse-failure branch built DataFusionError::Plan(format!(...)) by
hand while the non-object branch used plan_err! — two construction
styles in one helper. Use DataFusion's pair as intended:
plan_datafusion_err! (constructs the error, for map_err) alongside
plan_err! (returns it). Note the review's literal suggestion,
.map_err(|e| plan_err!(...))??, does not compile — plan_err! expands to
a full Err(...) Result, which is why the *_datafusion_err! twin exists.

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

* test(sources): pin the optional happy path and typed-null boundaries

Per review: null_semantics_differ_by_variant now asserts
optional_string_arg returns Some for a real string, and a new test
locks Utf8(None)/LargeUtf8(None) — typed NULLs, e.g. from
CAST(NULL AS VARCHAR) — as rejections across all three variants: they
must be neither a valid empty string nor the untyped-Null placeholder,
matching every provider's pre-refactor behavior.

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

* test(sources): cover NULL arguments to open_connector_query at planning

Per review: the argument-rejection test covered non-string literals but
not SQL NULL. Add NULL for gateway, table_id, and resource_json,
asserting the strict_string_arg "must be a string literal, not NULL"
diagnostic — still inside the same test, whose trailing assertion
proves every rejection fires before any HTTP execute.

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

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: bakey

crates/skardi/src/sources/providers/lance/fts_table_function.rs

@@ -39,6 +39,7 @@ use std::sync::Arc;
 
 use super::fts_exec::LanceFtsExec;
 use super::utils::expr_to_lance_sql;
+use crate::sources::providers::udtf_args::{optional_string_arg, string_arg};
 use crate::sources::providers::{DatasetEntry, DatasetRegistry};
 
 /// Table function that creates full-text search on Lance tables
@@ -62,9 +63,9 @@ impl TableFunctionImpl for LanceFtsTableFunction {
             );
         }
 
-        let table_name = extract_string(&exprs[0], "table_name")?;
-        let text_column = extract_string(&exprs[1], "text_column")?;
-        let query = extract_string_or_null(&exprs[2], "query")?;
+        let table_name = string_arg(&exprs[0], "lance_fts", "table_name")?;
+        let text_column = string_arg(&exprs[1], "lance_fts", "text_column")?;
+        let query = optional_string_arg(&exprs[2], "lance_fts", "query")?;
         let limit = extract_int(&exprs[3], "limit")?;
 
         // Get dataset from registry
@@ -309,24 +310,6 @@ impl TableProvider for LanceFtsProvider {
 
 // Helper functions for argument extraction
 
-fn extract_string(expr: &Expr, name: &str) -> DFResult<String> {
-    match expr {
-        Expr::Literal(ScalarValue::Utf8(Some(s)), _) => Ok(s.clone()),
-        Expr::Literal(ScalarValue::LargeUtf8(Some(s)), _) => Ok(s.clone()),
-        // Accept NULL as placeholder during pipeline validation/schema inference.
-        // The inferencer replaces {param} with NULL before plan creation.
-        Expr::Literal(ScalarValue::Null, _) => Ok(String::new()),
-        _ => plan_err!("lance_fts: {} must be a string literal", name),
-    }
-}
-
-fn extract_string_or_null(expr: &Expr, name: &str) -> DFResult<Option<String>> {
-    match expr {
-        Expr::Literal(ScalarValue::Null, _) => Ok(None),
-        other => extract_string(other, name).map(Some),
-    }
-}
-
 fn extract_int(expr: &Expr, name: &str) -> DFResult<usize> {
     match expr {
         Expr::Literal(ScalarValue::Int64(Some(n)), _) => Ok(*n as usize),

crates/skardi/src/sources/providers/lance/knn_table_function.rs

@@ -30,6 +30,7 @@ use std::sync::Arc;
 use super::knn_exec::LanceKnnExec;
 use super::utils::expr_to_lance_sql;
 use crate::sources::providers::knn_utils::extract_k;
+use crate::sources::providers::udtf_args::strict_string_arg;
 use crate::sources::providers::{DatasetEntry, DatasetRegistry};
 
 /// Table function that creates KNN search on Lance tables
@@ -54,11 +55,11 @@ impl TableFunctionImpl for LanceKnnTableFunction {
         }
 
         // Extract string arguments
-        let table_name = extract_string(&exprs[0], "table_name")?;
-        let vector_column = extract_string(&exprs[1], "vector_column")?;
+        let table_name = strict_string_arg(&exprs[0], "lance_knn", "table_name")?;
+        let vector_column = strict_string_arg(&exprs[1], "lance_knn", "vector_column")?;
         let k = extract_k(&exprs[3], "lance_knn")?;
         let filter = if exprs.len() == 5 {
-            Some(extract_string(&exprs[4], "filter")?)
+            Some(strict_string_arg(&exprs[4], "lance_knn", "filter")?)
         } else {
             None
         };
@@ -230,14 +231,6 @@ impl TableProvider for LanceKnnProvider {
 
 // Helper functions for argument extraction
 
-fn extract_string(expr: &Expr, name: &str) -> DFResult<String> {
-    match expr {
-        Expr::Literal(ScalarValue::Utf8(Some(s)), _) => Ok(s.clone()),
-        Expr::Literal(ScalarValue::LargeUtf8(Some(s)), _) => Ok(s.clone()),
-        _ => plan_err!("lance_knn: {} must be a string literal", name),
-    }
-}
-
 fn try_extract_vector(expr: &Expr) -> DFResult<Option<ArrayRef>> {
     match expr {
         Expr::Literal(ScalarValue::List(arr), _) => {

crates/skardi/src/sources/providers/mod.rs

@@ -14,6 +14,7 @@ pub mod redis;
 pub mod seekdb;
 pub mod sqlite;
 pub mod sqlx;
+pub(crate) mod udtf_args;
 
 use ::lance::dataset::Dataset;
 use datafusion::datasource::TableProvider;

crates/skardi/src/sources/providers/mongo/fts_table_function.rs

@@ -29,6 +29,7 @@ use std::sync::Arc;
 
 use super::fts_exec::MongoFtsExec;
 use super::{binary_expr_to_mongo, is_pushable_binary_filter};
+use crate::sources::providers::udtf_args::string_arg;
 use crate::sources::providers::{DatasetEntry, DatasetRegistry};
 
 /// Maximum allowed FTS result limit (matches MAX_KNN_K).
@@ -65,8 +66,8 @@ impl TableFunctionImpl for MongoFtsTableFunction {
             );
         }
 
-        let collection_name = extract_string(&exprs[0], "collection")?;
-        let query = extract_string(&exprs[1], "query")?;
+        let collection_name = string_arg(&exprs[0], "mongo_fts", "collection")?;
+        let query = string_arg(&exprs[1], "mongo_fts", "query")?;
         let limit = extract_int(&exprs[2], "limit")?;
 
         // The inferencer replaces {param} with NULL, yielding empty string for
@@ -236,21 +237,6 @@ fn expr_to_mongo_filter_entry(expr: &Expr, primary_key: &str) -> Option<Document
 
 // ─── Argument extraction helpers ─────────────────────────────────────────────
 
-fn extract_string(expr: &Expr, name: &str) -> DFResult<String> {
-    match expr {
-        Expr::Literal(ScalarValue::Utf8(Some(s)), _)
-        | Expr::Literal(ScalarValue::LargeUtf8(Some(s)), _) => Ok(s.clone()),
-        // Accept NULL as placeholder during pipeline validation/schema inference.
-        // The inferencer replaces {param} with NULL before plan creation.
-        Expr::Literal(ScalarValue::Null, _) => Ok(String::new()),
-        _ => plan_err!(
-            "mongo_fts: '{}' must be a string literal, got {:?}",
-            name,
-            expr
-        ),
-    }
-}
-
 fn extract_int(expr: &Expr, name: &str) -> DFResult<usize> {
     match expr {
         Expr::Literal(ScalarValue::Int64(Some(v)), _) => Ok(*v as usize),

crates/skardi/src/sources/providers/open_connector/table_functions.rs

@@ -37,7 +37,7 @@ use std::time::Duration;
 use arrow::datatypes::SchemaRef;
 use async_trait::async_trait;
 use datafusion::catalog::{Session, TableFunctionImpl, TableProvider};
-use datafusion::common::{ScalarValue, plan_err};
+use datafusion::common::{plan_datafusion_err, plan_err};
 use datafusion::datasource::TableType;
 use datafusion::error::{DataFusionError, Result as DFResult};
 use datafusion::logical_expr::Expr;
@@ -57,6 +57,7 @@ use super::raw_schema::derive_raw_columns;
 use super::row_path::RowPath;
 use super::source_pack::SourcePackRegistry;
 use super::table::OpenConnectorTableProvider;
+use crate::sources::providers::udtf_args::strict_string_arg;
 
 /// Planning-time state of one registered gateway, captured by
 /// `register_open_connector_tables` and shared with both UDTFs.
@@ -143,12 +144,12 @@ impl TableFunctionImpl for OpenConnectorQueryFunction {
                 exprs.len()
             );
         }
-        let gateway = literal_string("open_connector_query", &exprs[0], "gateway")?;
-        let table_id = literal_string("open_connector_query", &exprs[1], "table_id")?;
-        let resource_json = literal_string("open_connector_query", &exprs[2], "resource_json")?;
+        let gateway = strict_string_arg(&exprs[0], "open_connector_query", "gateway")?;
+        let table_id = strict_string_arg(&exprs[1], "open_connector_query", "table_id")?;
+        let resource_json = strict_string_arg(&exprs[2], "open_connector_query", "resource_json")?;
         let alias = exprs
             .get(3)
-            .map(|expr| literal_string("open_connector_query", expr, "connection_alias"))
+            .map(|expr| strict_string_arg(expr, "open_connector_query", "connection_alias"))
             .transpose()?;
 
         let handle = lookup_gateway(&self.gateways, &gateway)?;
@@ -165,17 +166,12 @@ impl TableFunctionImpl for OpenConnectorQueryFunction {
         let pack = self.packs.require(pack_name).map_err(plan_error)?;
         let table = self.packs.table(pack, table_name).map_err(plan_error)?;
 
-        let resource: Value = serde_json::from_str(&resource_json).map_err(|e| {
-            DataFusionError::Plan(format!(
-                "open_connector_query: resource_json is not valid JSON: {e}"
-            ))
-        })?;
-        if !resource.is_object() {
-            return plan_err!(
-                "open_connector_query: resource_json must be a JSON object of resource \
-                 inputs, e.g. '{{\"owner\":\"SkardiLabs\",\"repo\":\"skardi\"}}'"
-            );
-        }
+        let resource = parse_json_object(
+            "open_connector_query",
+            "resource_json",
+            &resource_json,
+            "resource inputs",
+        )?;
         for key in table.required_resources {
             if resource.get(*key).is_none() {
                 return Err(plan_error(OpenConnectorError::MissingResourceInput {
@@ -242,13 +238,13 @@ impl TableFunctionImpl for OpenConnectorScanFunction {
                 exprs.len()
             );
         }
-        let gateway = literal_string("open_connector_scan", &exprs[0], "gateway")?;
-        let action_id = literal_string("open_connector_scan", &exprs[1], "action_id")?;
-        let input_json = literal_string("open_connector_scan", &exprs[2], "input_json")?;
-        let row_path = literal_string("open_connector_scan", &exprs[3], "row_path")?;
+        let gateway = strict_string_arg(&exprs[0], "open_connector_scan", "gateway")?;
+        let action_id = strict_string_arg(&exprs[1], "open_connector_scan", "action_id")?;
+        let input_json = strict_string_arg(&exprs[2], "open_connector_scan", "input_json")?;
+        let row_path = strict_string_arg(&exprs[3], "open_connector_scan", "row_path")?;
         let alias = exprs
             .get(4)
-            .map(|expr| literal_string("open_connector_scan", expr, "connection_alias"))
+            .map(|expr| strict_string_arg(expr, "open_connector_scan", "connection_alias"))
             .transpose()?;
 
         let handle = lookup_gateway(&self.gateways, &gateway)?;
@@ -279,17 +275,12 @@ impl TableFunctionImpl for OpenConnectorScanFunction {
             }
         }
 
-        let input: Value = serde_json::from_str(&input_json).map_err(|e| {
-            DataFusionError::Plan(format!(
-                "open_connector_scan: input_json is not valid JSON: {e}"
-            ))
-        })?;
-        if !input.is_object() {
-            return plan_err!(
-                "open_connector_scan: input_json must be a JSON object of action inputs, \
-                 e.g. '{{\"owner\":\"SkardiLabs\",\"repo\":\"skardi\"}}'"
-            );
-        }
+        let input = parse_json_object(
+            "open_connector_scan",
+            "input_json",
+            &input_json,
+            "action inputs",
+        )?;
 
         let row_path = RowPath::parse(&row_path).map_err(plan_error)?;
         // Deterministic row type or planning error — derived purely from the
@@ -423,13 +414,20 @@ fn plan_error(e: OpenConnectorError) -> DataFusionError {
     DataFusionError::Plan(e.to_string())
 }
 
-/// Extract one string-literal argument.
-fn literal_string(function: &str, expr: &Expr, name: &str) -> DFResult<String> {
-    match expr {
-        Expr::Literal(ScalarValue::Utf8(Some(s)), _)
-        | Expr::Literal(ScalarValue::LargeUtf8(Some(s)), _) => Ok(s.clone()),
-        _ => plan_err!("{function}: {name} must be a string literal"),
+/// Parse a UDTF argument that must carry a JSON object, shared by both
+/// functions so the two diagnostics stay in lockstep. `noun` names what the
+/// object holds in the caller's vocabulary ("resource inputs" / "action
+/// inputs"), so sharing the implementation doesn't flatten the context.
+fn parse_json_object(fn_name: &str, arg: &str, raw: &str, noun: &str) -> DFResult<Value> {
+    let value: Value = serde_json::from_str(raw)
+        .map_err(|e| plan_datafusion_err!("{fn_name}: {arg} is not valid JSON: {e}"))?;
+    if !value.is_object() {
+        return plan_err!(
+            "{fn_name}: {arg} must be a JSON object of {noun}, \
+             e.g. '{{\"owner\":\"SkardiLabs\",\"repo\":\"skardi\"}}'"
+        );
     }
+    Ok(value)
 }
 
 #[cfg(test)]
@@ -746,6 +744,12 @@ raw_action_allowlist:
             "resource_json is not valid JSON",
         )
         .await;
+        expect_plan_error(
+            &ctx,
+            "SELECT * FROM open_connector_query('saas', 'mock.items', '[1, 2]')",
+            "resource_json must be a JSON object of resource inputs",
+        )
+        .await;
         expect_plan_error(
             &ctx,
             "SELECT * FROM open_connector_query('saas', 'mock.items')",
@@ -755,7 +759,27 @@ raw_action_allowlist:
         expect_plan_error(
             &ctx,
             "SELECT * FROM open_connector_query(1, 'mock.items', '{}')",
-            "gateway must be a string literal",
+            "'gateway' must be a string literal",
+        )
+        .await;
+        // NULL is rejected outright for schema-determining arguments — a
+        // placeholder cannot produce a plan (strict_string_arg semantics).
+        expect_plan_error(
+            &ctx,
+            "SELECT * FROM open_connector_query(NULL, 'mock.items', '{}')",
+            "'gateway' must be a string literal, not NULL",
+        )
+        .await;
+        expect_plan_error(
+            &ctx,
+            "SELECT * FROM open_connector_query('saas', NULL, '{}')",
+            "'table_id' must be a string literal, not NULL",
+        )
+        .await;
+        expect_plan_error(
+            &ctx,
+            "SELECT * FROM open_connector_query('saas', 'mock.items', NULL)",
+            "'resource_json' must be a string literal, not NULL",
         )
         .await;
 
@@ -998,6 +1022,13 @@ raw_action_allowlist:
             "must start with '$.'",
         )
         .await;
+        expect_plan_error(
+            &ctx,
+            r#"SELECT * FROM open_connector_scan('saas', 'mock.list_items',
+                                                 '[1, 2]', '$.items')"#,
+            "input_json must be a JSON object of action inputs",
+        )
+        .await;
         assert!(execute_requests(&gateway).is_empty(), "rejected pre-HTTP");
     }
 

crates/skardi/src/sources/providers/seekdb/fts_table_function.rs

@@ -36,8 +36,9 @@ use datafusion_table_providers::sql::db_connection_pool::mysqlpool::MySQLConnect
 use std::any::Any;
 use std::sync::Arc;
 
+use super::expr_to_seekdb_sql;
 use super::fts_exec::SeekDbFtsExec;
-use super::{expr_to_seekdb_sql, extract_string_arg};
+use crate::sources::providers::udtf_args::string_arg;
 use crate::sources::providers::{DatasetEntry, DatasetRegistry};
 
 /// Maximum allowed FTS result limit.
@@ -66,9 +67,9 @@ impl TableFunctionImpl for SeekDbFtsTableFunction {
             );
         }
 
-        let table_name = extract_string_arg(&exprs[0], "seekdb_fts", "table")?;
-        let text_col = extract_string_arg(&exprs[1], "seekdb_fts", "text_col")?;
-        let query = extract_string_arg(&exprs[2], "seekdb_fts", "query")?;
+        let table_name = string_arg(&exprs[0], "seekdb_fts", "table")?;
+        let text_col = string_arg(&exprs[1], "seekdb_fts", "text_col")?;
+        let query = string_arg(&exprs[2], "seekdb_fts", "query")?;
         let limit = match extract_int(&exprs[3], "limit")? {
             None => 1,
             Some(v) if v > MAX_FTS_LIMIT => {

crates/skardi/src/sources/providers/seekdb/knn_table_function.rs

@@ -52,9 +52,10 @@ use datafusion_table_providers::sql::db_connection_pool::mysqlpool::MySQLConnect
 use std::any::Any;
 use std::sync::Arc;
 
+use super::expr_to_seekdb_sql;
 use super::knn_exec::{DistanceMetric, SeekDbKnnExec};
-use super::{expr_to_seekdb_sql, extract_string_arg};
 use crate::sources::providers::knn_utils::{extract_k, extract_literal_vector};
+use crate::sources::providers::udtf_args::string_arg;
 use crate::sources::providers::{DatasetEntry, DatasetRegistry};
 
 /// Entry stored in the registry for each registered SeekDB table.
@@ -92,10 +93,10 @@ impl TableFunctionImpl for SeekDbKnnTableFunction {
             );
         }
 
-        let table_name = extract_string_arg(&exprs[0], "seekdb_knn", "table")?;
-        let vector_col = extract_string_arg(&exprs[1], "seekdb_knn", "vector_col")?;
+        let table_name = string_arg(&exprs[0], "seekdb_knn", "table")?;
+        let vector_col = string_arg(&exprs[1], "seekdb_knn", "vector_col")?;
 
-        let metric_str = extract_string_arg(&exprs[3], "seekdb_knn", "metric")?;
+        let metric_str = string_arg(&exprs[3], "seekdb_knn", "metric")?;
         let metric = if metric_str.is_empty() {
             // NULL placeholder during schema inference — default to L2.
             DistanceMetric::default()

crates/skardi/src/sources/providers/seekdb/mod.rs

@@ -39,7 +39,7 @@ use arrow::array::{RecordBatch, UInt64Array};
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
 use async_trait::async_trait;
 use datafusion::catalog::Session;
-use datafusion::common::{Constraints, ScalarValue, plan_err};
+use datafusion::common::Constraints;
 use datafusion::datasource::{TableProvider, TableType};
 use datafusion::error::{DataFusionError, Result as DataFusionResult};
 use datafusion::execution::{SendableRecordBatchStream, TaskContext};
@@ -793,23 +793,6 @@ pub(crate) fn expr_to_seekdb_sql(expr: &Expr) -> Option<String> {
     unparser.expr_to_sql(expr).ok().map(|ast| ast.to_string())
 }
 
-/// Extract a string-literal argument from a table-function `Expr`. Shared by
-/// `seekdb_fts` and `seekdb_knn` so they can emit consistent error messages
-/// prefixed with the caller's UDTF name (`fn_name`). NULL is accepted as a
-/// placeholder during pipeline schema inference and returns the empty string.
-pub(crate) fn extract_string_arg(
-    expr: &Expr,
-    fn_name: &str,
-    arg: &str,
-) -> DataFusionResult<String> {
-    match expr {
-        Expr::Literal(ScalarValue::Utf8(Some(s)), _)
-        | Expr::Literal(ScalarValue::LargeUtf8(Some(s)), _) => Ok(s.clone()),
-        Expr::Literal(ScalarValue::Null, _) => Ok(String::new()),
-        _ => plan_err!("{fn_name}: '{arg}' must be a string literal"),
-    }
-}
-
 // ─── Execution plan for DELETE / UPDATE results ─────────────────────────────
 
 /// A leaf [`ExecutionPlan`] that executes a pre-built SeekDB DML statement