feldera · mihaibudiu · Jul 16, 2026 · Jul 15, 2026 · Jul 15, 2026 · Jul 15, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -109,6 +109,8 @@ csv-core = "0.1.10"
 dashmap = "6.1.0"
 datafusion = "53.1"
 dbsp = { path = "crates/dbsp", version = "0.323.0" }
+# JSON functions applied to strings.
+datafusion-functions-json = "0.53.1"
 dbsp_nexmark = { path = "crates/nexmark" }
 deadpool-postgres = "0.14.1"
 # Feldera fork of delta-rs: upstream tag `rust-v0.32.3` + 3 patches, on branch

diff --git a/crates/adapterlib/Cargo.toml b/crates/adapterlib/Cargo.toml
@@ -42,6 +42,7 @@ num-derive = { workspace = true }
 tokio = { workspace = true, features = ["sync"] }
 xxhash-rust = { workspace = true, features = ["xxh3"] }
 datafusion = { workspace = true }
+datafusion-functions-json = { workspace = true }
 thiserror = { workspace = true }
 serde_json_path_to_error = { workspace = true }
 chrono = { workspace = true }

diff --git a/crates/adapterlib/src/utils/datafusion.rs b/crates/adapterlib/src/utils/datafusion.rs
@@ -219,11 +219,16 @@ where
         );
     let session_config = customize_config(session_config);
 
-    let state = SessionStateBuilder::new()
+    let mut state = SessionStateBuilder::new()
         .with_config(session_config)
         .with_runtime_env(runtime_env)
         .with_default_features()
         .build();
+    // JSON functions for querying VARIANT columns, which reach DataFusion
+    // as JSON-encoded strings. Note: the crate's `->` and `?` operators do
+    // not parse in the default (generic) SQL dialect; `->>` works.
+    datafusion_functions_json::register_all(&mut state)
+        .expect("registering JSON functions on a fresh session state cannot fail");
     SessionContext::from(state)
 }
 
@@ -733,6 +738,40 @@ mod tests {
         assert_eq!(ctx.copied_config().target_partitions(), 99);
     }
 
+    /// Every context built here must provide the JSON function family.
+    #[test]
+    fn create_session_context_registers_json_functions() {
+        let storage = TempStorage::new("feldera-datafusion-create-session-context-json-test");
+        let cfg = pipeline_config(
+            RuntimeConfig {
+                workers: 1,
+                ..Default::default()
+            },
+            Some(storage.path()),
+        );
+        let env = create_runtime_env(&cfg).unwrap();
+        let ctx = create_session_context(&cfg, env);
+        let state = ctx.state();
+        for function in [
+            "json_get",
+            "json_get_str",
+            "json_get_int",
+            "json_get_float",
+            "json_get_bool",
+            "json_get_json",
+            "json_get_array",
+            "json_as_text",
+            "json_contains",
+            "json_length",
+            "json_object_keys",
+        ] {
+            assert!(
+                state.scalar_functions().contains_key(function),
+                "JSON function '{function}' is not registered"
+            );
+        }
+    }
+
     /// Tripwire: `clean_stale_scratch_entries` refuses to walk a directory
     /// whose final component isn't `DATAFUSION_TEMP_DIR`, so a misuse can't
     /// recursively wipe an arbitrary path.

diff --git a/crates/adapters/src/adhoc.rs b/crates/adapters/src/adhoc.rs
@@ -705,6 +705,137 @@ mod tests {
         assert_eq!(total_rows, 1);
     }
 
+    /// Ad-hoc queries can take apart VARIANT columns with the JSON function
+    /// family from `datafusion-functions-json`. VARIANT values reach
+    /// DataFusion as JSON-encoded Utf8 strings; the in-memory table below
+    /// stands in for a materialized table with a VARIANT column.
+    ///
+    /// The state must come from `create_session_context`, which registers
+    /// the functions; the plain `test_state()` used elsewhere in this
+    /// module does not have them.
+    #[tokio::test]
+    async fn json_functions_work_in_adhoc_queries() {
+        use datafusion::arrow::array::{Array, Int64Array, StringArray};
+        use datafusion::arrow::datatypes::{DataType, Field, Schema};
+        use datafusion::datasource::MemTable;
+        use feldera_adapterlib::utils::datafusion::{create_runtime_env, create_session_context};
+        use feldera_types::config::{PipelineConfig, RuntimeConfig};
+
+        // Tripwire: stock DataFusion does not know these functions; they
+        // must come from the registration in `create_session_context`. If
+        // this starts failing, DataFusion gained native JSON functions and
+        // the `datafusion-functions-json` dependency may be redundant.
+        let err = execute_sql_with_state(test_state(), "SELECT json_get_str('{}', 'a')")
+            .await
+            .expect_err("stock DataFusion should not provide json_get_str");
+        assert!(
+            format!("{err}").contains("json_get_str"),
+            "unexpected error: {err}"
+        );
+
+        let config = PipelineConfig {
+            global: RuntimeConfig {
+                workers: 1,
+                ..Default::default()
+            },
+            multihost: None,
+            name: None,
+            given_name: None,
+            storage_config: None,
+            secrets_dir: None,
+            inputs: Default::default(),
+            outputs: Default::default(),
+            program_ir: None,
+        };
+        let runtime_env = create_runtime_env(&config).unwrap();
+        let ctx = create_session_context(&config, runtime_env);
+
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int64, false),
+            Field::new("val", DataType::Utf8, true),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(Int64Array::from(vec![1, 2, 3])),
+                Arc::new(StringArray::from(vec![
+                    Some(r#"{"name":"Bob","scores":[8,10],"active":true}"#),
+                    Some(r#"{"name":"Ann","scores":[3],"extra":{"x":5}}"#),
+                    None,
+                ])),
+            ],
+        )
+        .unwrap();
+        let mem = MemTable::try_new(schema, vec![vec![batch]]).unwrap();
+        ctx.register_table("t", Arc::new(mem)).unwrap();
+
+        // Typed getters in the projection. A missing key ('scores' has one
+        // element in row 2; indexes are 0-based) and a NULL document both
+        // yield NULL.
+        let batches = collect_rows(
+            ctx.state(),
+            "SELECT json_get_str(val, 'name') AS name, \
+             json_get_int(val, 'scores', 1) AS second_score \
+             FROM t ORDER BY id",
+        )
+        .await;
+        let names = batches[0]
+            .column(0)
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .expect("string column");
+        assert_eq!(names.value(0), "Bob");
+        assert_eq!(names.value(1), "Ann");
+        assert!(names.is_null(2));
+        let scores = batches[0]
+            .column(1)
+            .as_any()
+            .downcast_ref::<Int64Array>()
+            .expect("int64 column");
+        assert_eq!(scores.value(0), 10);
+        assert!(scores.is_null(1));
+        assert!(scores.is_null(2));
+
+        // JSON predicates filter on document contents.
+        for (query, expected_id) in [
+            ("SELECT id FROM t WHERE json_contains(val, 'extra')", 2),
+            (
+                "SELECT id FROM t WHERE json_get_bool(val, 'active') = TRUE",
+                1,
+            ),
+        ] {
+            let batches = collect_rows(ctx.state(), query).await;
+            let ids = batches[0]
+                .column(0)
+                .as_any()
+                .downcast_ref::<Int64Array>()
+                .expect("int64 column");
+            assert_eq!(ids.len(), 1, "query: {query}");
+            assert_eq!(ids.value(0), expected_id, "query: {query}");
+        }
+
+        // The `->>` operator and casting `json_get` to a concrete type.
+        let batches = collect_rows(
+            ctx.state(),
+            "SELECT val->>'name' AS name, \
+             CAST(json_get(val, 'scores', 0) AS BIGINT) AS first_score \
+             FROM t WHERE json_get_str(val, 'name') = 'Bob'",
+        )
+        .await;
+        let names = batches[0]
+            .column(0)
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .expect("string column");
+        assert_eq!(names.value(0), "Bob");
+        let scores = batches[0]
+            .column(1)
+            .as_any()
+            .downcast_ref::<Int64Array>()
+            .expect("int64 column");
+        assert_eq!(scores.value(0), 8);
+    }
+
     /// Selecting from a non-materialized source fails during *physical
     /// planning* — when the scan node is built — not during scan execution.
     /// `execute_adhoc_stream` must surface that as an `Err`, letting the HTTP