From 8a576b74165878bf406d21ab0f2fa4bdc01c5141 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 15 Jul 2026 16:55:14 -0700 Subject: [PATCH 1/4] [adapters] Support JSON functions in ad-hoc queries Register the datafusion-functions-json UDF family (json_get_str, json_get_int, json_contains, ..., and the ->> operator) in every DataFusion session context Feldera creates, so ad-hoc queries can take apart VARIANT columns, which reach DataFusion as JSON-encoded strings. Casting json_get is rewritten to the matching typed getter. The -> and ? operators from the crate do not parse in the default (generic) SQL dialect and stay unavailable. Fixes #6644 Signed-off-by: Leonid Ryzhyk --- Cargo.toml | 3 + crates/adapterlib/Cargo.toml | 1 + crates/adapterlib/src/utils/datafusion.rs | 45 +++++++- crates/adapters/src/adhoc.rs | 133 ++++++++++++++++++++++ 4 files changed, 181 insertions(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 861b544588d..c8bc0707973 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -109,6 +109,9 @@ csv-core = "0.1.10" dashmap = "6.1.0" datafusion = "53.1" dbsp = { path = "crates/dbsp", version = "0.323.0" } +# JSON functions (json_get_str, json_contains, ...) for ad-hoc queries over +# VARIANT columns, which reach DataFusion as JSON-encoded 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 index 8b06684aba8..031a23575c0 100644 --- 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 index ab3643d5546..56fd6251584 100644 --- a/crates/adapterlib/src/utils/datafusion.rs +++ b/crates/adapterlib/src/utils/datafusion.rs @@ -219,11 +219,18 @@ 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(); + // DataFusion has no VARIANT type; VARIANT columns reach it as + // JSON-encoded strings. The `datafusion-functions-json` family + // (`json_get_str`, `json_contains`, ...) lets queries take such + // strings apart. Of the crate's operators only `->>` is reachable: + // `->` and `?` do not parse in the default (generic) SQL dialect. + datafusion_functions_json::register_all(&mut state) + .expect("registering JSON functions on a fresh session state cannot fail"); SessionContext::from(state) } @@ -733,6 +740,42 @@ mod tests { assert_eq!(ctx.copied_config().target_partitions(), 99); } + /// The JSON function family must be registered on every context built + /// here: ad-hoc queries rely on it to take apart VARIANT columns, which + /// DataFusion sees as JSON-encoded strings (issue #6644). + #[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 index 44ead9b7d1c..2331bf44f96 100644 --- a/crates/adapters/src/adhoc.rs +++ b/crates/adapters/src/adhoc.rs @@ -705,6 +705,139 @@ mod tests { assert_eq!(total_rows, 1); } + /// Ad-hoc queries can take apart VARIANT columns with the JSON function + /// family from `datafusion-functions-json` (issue #6644). 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::() + .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::() + .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, as in the queries + // from issue #6644. + 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::() + .expect("int64 column"); + assert_eq!(ids.len(), 1, "query: {query}"); + assert_eq!(ids.value(0), expected_id, "query: {query}"); + } + + // `->>` is shorthand for `json_as_text`, and casting `json_get` is + // rewritten to the matching typed getter. + 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::() + .expect("string column"); + assert_eq!(names.value(0), "Bob"); + let scores = batches[0] + .column(1) + .as_any() + .downcast_ref::() + .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 From 8da78f774ae16d676f96464970b56dff1900c678 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 15 Jul 2026 16:55:25 -0700 Subject: [PATCH 2/4] [python] Test ad-hoc queries over VARIANT columns with JSON functions Feed JSON documents into a materialized table with a VARIANT column and exercise the datafusion-functions-json surface end to end: typed getters, nested paths with 0-based array indexes, JSON predicates, the ->> operator, and the cast-to-typed-getter rewrite (issue #6644). Signed-off-by: Leonid Ryzhyk --- python/tests/platform/test_shared_pipeline.py | 82 +++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/python/tests/platform/test_shared_pipeline.py b/python/tests/platform/test_shared_pipeline.py index 9ebf255cdc6..9949fb85812 100644 --- a/python/tests/platform/test_shared_pipeline.py +++ b/python/tests/platform/test_shared_pipeline.py @@ -208,6 +208,88 @@ def test_adhoc_query_arrow(self): table_pipeline = pa.Table.from_batches(batches_pipeline) assert table_pipeline.column("id").to_pylist() == expected_ids + def test_adhoc_json_functions(self): + """ + -- Ad-hoc queries read VARIANT columns as JSON-encoded strings and + -- take them apart with datafusion-functions-json (issue #6644). + CREATE TABLE json_docs (id INT NOT NULL, doc VARIANT) WITH ('materialized' = 'true'); + """ + self.pipeline.start() + self.pipeline.input_json( + "json_docs", + [ + {"id": 1, "doc": {"name": "Bob", "scores": [8, 10], "active": True}}, + { + "id": 2, + "doc": { + "name": "Ann", + "scores": [3], + "address": {"city": "Berlin"}, + }, + }, + {"id": 3, "doc": None}, + ], + ) + + # A bare VARIANT column comes back as a JSON-encoded string. + got = list(self.pipeline.query("SELECT doc FROM json_docs WHERE id = 1")) + assert json.loads(got[0]["doc"]) == { + "name": "Bob", + "scores": [8, 10], + "active": True, + } + + # Typed getters pull concrete values out of the VARIANT column. + # A missing key ('scores' of row 2 has one element; indexes are + # 0-based) and a NULL document both yield NULL. + got = list( + self.pipeline.query( + "SELECT id, json_get_str(doc, 'name') AS name," + " json_get_int(doc, 'scores', 1) AS second_score" + " FROM json_docs ORDER BY id" + ) + ) + assert got == [ + {"id": 1, "name": "Bob", "second_score": 10}, + {"id": 2, "name": "Ann", "second_score": None}, + {"id": 3, "name": None, "second_score": None}, + ] + + # Nested paths mix object keys and array indexes. + got = list( + self.pipeline.query( + "SELECT json_get_str(doc, 'address', 'city') AS city" + " FROM json_docs WHERE id = 2" + ) + ) + assert got == [{"city": "Berlin"}] + + # JSON predicates filter on document contents. + got = list( + self.pipeline.query( + "SELECT id FROM json_docs WHERE json_contains(doc, 'address')" + ) + ) + assert got == [{"id": 2}] + got = list( + self.pipeline.query( + "SELECT id FROM json_docs WHERE json_get_bool(doc, 'active') = TRUE" + ) + ) + assert got == [{"id": 1}] + + # `->>` is shorthand for json_as_text; casting json_get is rewritten + # to the matching typed getter (json_get_int here). + got = list( + self.pipeline.query( + "SELECT doc->>'name' AS name," + " CAST(json_get(doc, 'scores', 0) AS BIGINT) AS first_score" + " FROM json_docs WHERE id = 1" + ) + ) + assert got == [{"name": "Bob", "first_score": 8}] + self.pipeline.stop(force=True) + def test_local(self): """ CREATE TABLE students ( From a1ba6ee18f2b450fc1815e7afe2518337bd7ed2c Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 15 Jul 2026 16:55:25 -0700 Subject: [PATCH 3/4] [DOC] Document JSON functions in ad-hoc queries Add a section on querying VARIANT columns with datafusion-functions-json to the ad-hoc query page: the function table, path semantics (0-based array indexes, unlike 1-based VARIANT subscripts in Feldera SQL), and the limits (json_get's union return type, unavailable -> and ? operators). Signed-off-by: Leonid Ryzhyk --- docs.feldera.com/docs/sql/ad-hoc.md | 57 +++++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/docs.feldera.com/docs/sql/ad-hoc.md b/docs.feldera.com/docs/sql/ad-hoc.md index d4346a8b05c..787c27f9a9b 100644 --- a/docs.feldera.com/docs/sql/ad-hoc.md +++ b/docs.feldera.com/docs/sql/ad-hoc.md @@ -38,6 +38,10 @@ ad-hoc queries that need to be taken into account: (`SELECT 1729595568::TIMESTAMP;` will yield `2024-10-22T11:12:48` in ad-hoc queries and `1970-01-21 00:26:35` in Feldera SQL). - Ad-hoc SQL cannot perform as-of joins. +- Feldera SQL reads fields of `VARIANT` values with subscripts (`doc['scores'][1]`, array indexes + are 1-based); ad-hoc queries use + [JSON functions](#querying-variant-columns-with-json-functions) instead + (`json_get(doc, 'scores', 0)`, array indexes are 0-based). We will continue to improve the consistency between the two engines in future releases. @@ -155,6 +159,59 @@ An ad-hoc query to insert data into the `complex_types` table would look like th insert into complex_types values ([1,2,3], struct(2, 'b'), '{"field": 3}', MAP(['answer'], [42]), struct(2, 3)); ``` +### Querying VARIANT Columns with JSON Functions + +Datafusion has no equivalent of the Feldera SQL [`VARIANT`](/sql/json) type. An ad-hoc query +receives a `VARIANT` column as a string that holds the JSON-encoded value, so the Feldera SQL +subscript syntax (`doc['name']`) does not work in ad-hoc queries. Instead, ad-hoc queries +take JSON strings apart with the +[datafusion-functions-json](https://github.com/datafusion-contrib/datafusion-functions-json) +function family. These functions also apply to string columns that hold JSON text. + +| Function | Return type | Description | +|-----------------------------------|-----------------|------------------------------------------------------------------------| +| `json_get(json, path...)` | union | Value at `path`; cast the result to select a concrete type (see below) | +| `json_get_str(json, path...)` | `VARCHAR` | String value at `path` | +| `json_get_int(json, path...)` | `BIGINT` | Integer value at `path` | +| `json_get_float(json, path...)` | `DOUBLE` | Float value at `path` | +| `json_get_bool(json, path...)` | `BOOLEAN` | Boolean value at `path` | +| `json_get_json(json, path...)` | `VARCHAR` | Raw JSON text of the value at `path` | +| `json_get_array(json, path...)` | array | Array at `path`, each element as raw JSON text | +| `json_as_text(json, path...)` | `VARCHAR` | Value at `path` as text; also available as the `->>` operator | +| `json_contains(json, path...)` | `BOOLEAN` | Whether a value exists at `path` | +| `json_length(json, path...)` | `BIGINT` | Length of the object or array at `path` | +| `json_object_keys(json, path...)` | `VARCHAR` array | Keys of the object at `path` | + +A path is a sequence of object keys and 0-based array indexes: +`json_get_int(doc, 'scores', 0)` reads element 0 of the array under the `scores` key. +A missing key, a mismatched type, or a `NULL` document yields `NULL`. + +Given + +```sql +CREATE TABLE json_docs (id INT, doc VARIANT) WITH ('materialized' = 'true'); +``` + +where `doc` holds documents such as `{"name": "Bob", "scores": [8, 10], "active": true}`, +the following ad-hoc query returns the names of active users whose first score exceeds 5: + +```sql +SELECT id, json_get_str(doc, 'name') AS name +FROM json_docs +WHERE json_get_bool(doc, 'active') = TRUE + AND json_get_int(doc, 'scores', 0) > 5; +``` + +Notes: + +- `json_get` returns a value of a union type that only the `text` and `arrow_ipc` output + formats can render. To use `json_get` results in the `json` or `parquet` output format, + cast them to a concrete type; the cast is rewritten to the matching typed getter, so + `json_get(doc, 'scores', 0)::BIGINT` and `CAST(json_get(doc, 'scores', 0) AS BIGINT)` + both execute as `json_get_int(doc, 'scores', 0)`. +- Of the operators provided by the crate, only `->>` (alias of `json_as_text`) is + available; `->` and `?` do not parse in the SQL dialect used by ad-hoc queries. + ### Parameterized Queries with PREPARE / EXECUTE Ad-hoc requests accept a `PREPARE` statement followed by a single From 4e193b3dd41bd3d44edbad9ba4a9d425e2d3153d Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 15 Jul 2026 20:46:31 -0700 Subject: [PATCH 4/4] Trim comments and simplify ad-hoc JSON docs per review Shorten dependency and test comments, drop issue references, and remove output-format details and the mention of unavailable operators from the ad-hoc JSON functions documentation (PR #6645 review). --- Cargo.lock | 106 +++++++++++++++++- Cargo.toml | 3 +- crates/adapterlib/src/utils/datafusion.rs | 12 +- crates/adapters/src/adhoc.rs | 12 +- docs.feldera.com/docs/sql/ad-hoc.md | 17 +-- python/tests/platform/test_shared_pipeline.py | 7 -- 6 files changed, 116 insertions(+), 41 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index af55ff52639..8490ca046ce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1906,7 +1906,7 @@ dependencies = [ "bitflags 2.10.0", "cexpr", "clang-sys", - "itertools 0.11.0", + "itertools 0.13.0", "proc-macro2", "quote", "regex", @@ -1924,7 +1924,7 @@ dependencies = [ "bitflags 2.10.0", "cexpr", "clang-sys", - "itertools 0.11.0", + "itertools 0.13.0", "proc-macro2", "quote", "regex", @@ -1942,7 +1942,7 @@ dependencies = [ "bitflags 2.10.0", "cexpr", "clang-sys", - "itertools 0.11.0", + "itertools 0.13.0", "log", "prettyplease", "proc-macro2", @@ -3678,6 +3678,18 @@ dependencies = [ "datafusion-physical-expr-common", ] +[[package]] +name = "datafusion-functions-json" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13ff70cb2c1960f03ba647aa2813fb1efba4c33bc221d973dd4a462a6376359a" +dependencies = [ + "datafusion", + "jiter", + "log", + "paste", +] + [[package]] name = "datafusion-functions-nested" version = "53.1.0" @@ -5256,6 +5268,7 @@ dependencies = [ "bytemuck", "chrono", "datafusion", + "datafusion-functions-json", "dbsp", "dyn-clone", "erased-serde 0.3.31", @@ -6567,7 +6580,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core 0.57.0", + "windows-core 0.61.2", ] [[package]] @@ -7084,6 +7097,21 @@ dependencies = [ "jiff-tzdb", ] +[[package]] +name = "jiter" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "020ba671987d7444d251d3ee5340be1bf4606cd6c0b53e6f4066b5a1ee376b22" +dependencies = [ + "ahash 0.8.12", + "bitvec", + "lexical-parse-float", + "num-bigint", + "num-traits", + "pyo3", + "smallvec", +] + [[package]] name = "jni" version = "0.22.4" @@ -9590,7 +9618,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.11.0", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.117", @@ -9603,7 +9631,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" dependencies = [ "anyhow", - "itertools 0.11.0", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.117", @@ -9667,6 +9695,66 @@ dependencies = [ "syn 1.0.109", ] +[[package]] +name = "pyo3" +version = "0.28.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91fd8e38a3b50ed1167fb981cd6fd60147e091784c427b8f7183a7ee32c31c12" +dependencies = [ + "libc", + "num-bigint", + "num-traits", + "once_cell", + "portable-atomic", + "pyo3-build-config", + "pyo3-ffi", + "pyo3-macros", +] + +[[package]] +name = "pyo3-build-config" +version = "0.28.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e368e7ddfdeb98c9bca7f8383be1648fd84ab466bf2bc015e94008db6d35611e" +dependencies = [ + "target-lexicon", +] + +[[package]] +name = "pyo3-ffi" +version = "0.28.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f29e10af80b1f7ccaf7f69eace800a03ecd13e883acfacc1e5d0988605f651e" +dependencies = [ + "libc", + "pyo3-build-config", +] + +[[package]] +name = "pyo3-macros" +version = "0.28.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df6e520eff47c45997d2fc7dd8214b25dd1310918bbb2642156ef66a67f29813" +dependencies = [ + "proc-macro2", + "pyo3-macros-backend", + "quote", + "syn 2.0.117", +] + +[[package]] +name = "pyo3-macros-backend" +version = "0.28.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4cdc218d835738f81c2338f822078af45b4afdf8b2e33cbb5916f108b813acb" +dependencies = [ + "heck 0.5.0", + "proc-macro2", + "pyo3-build-config", + "quote", + "syn 2.0.117", +] + [[package]] name = "quad-rand" version = "0.2.3" @@ -12451,6 +12539,12 @@ dependencies = [ "xattr", ] +[[package]] +name = "target-lexicon" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "adb6935a6f5c20170eeceb1a3835a49e12e19d792f6dd344ccc76a985ca5a6ca" + [[package]] name = "target-triple" version = "0.1.4" diff --git a/Cargo.toml b/Cargo.toml index c8bc0707973..1ac57b24b1d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -109,8 +109,7 @@ csv-core = "0.1.10" dashmap = "6.1.0" datafusion = "53.1" dbsp = { path = "crates/dbsp", version = "0.323.0" } -# JSON functions (json_get_str, json_contains, ...) for ad-hoc queries over -# VARIANT columns, which reach DataFusion as JSON-encoded strings. +# JSON functions applied to strings. datafusion-functions-json = "0.53.1" dbsp_nexmark = { path = "crates/nexmark" } deadpool-postgres = "0.14.1" diff --git a/crates/adapterlib/src/utils/datafusion.rs b/crates/adapterlib/src/utils/datafusion.rs index 56fd6251584..70fb60f14ca 100644 --- a/crates/adapterlib/src/utils/datafusion.rs +++ b/crates/adapterlib/src/utils/datafusion.rs @@ -224,11 +224,9 @@ where .with_runtime_env(runtime_env) .with_default_features() .build(); - // DataFusion has no VARIANT type; VARIANT columns reach it as - // JSON-encoded strings. The `datafusion-functions-json` family - // (`json_get_str`, `json_contains`, ...) lets queries take such - // strings apart. Of the crate's operators only `->>` is reachable: - // `->` and `?` do not parse in the default (generic) SQL dialect. + // 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) @@ -740,9 +738,7 @@ mod tests { assert_eq!(ctx.copied_config().target_partitions(), 99); } - /// The JSON function family must be registered on every context built - /// here: ad-hoc queries rely on it to take apart VARIANT columns, which - /// DataFusion sees as JSON-encoded strings (issue #6644). + /// 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"); diff --git a/crates/adapters/src/adhoc.rs b/crates/adapters/src/adhoc.rs index 2331bf44f96..893677715da 100644 --- a/crates/adapters/src/adhoc.rs +++ b/crates/adapters/src/adhoc.rs @@ -706,9 +706,9 @@ mod tests { } /// Ad-hoc queries can take apart VARIANT columns with the JSON function - /// family from `datafusion-functions-json` (issue #6644). VARIANT values - /// reach DataFusion as JSON-encoded Utf8 strings; the in-memory table - /// below stands in for a materialized table with a VARIANT column. + /// 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 @@ -796,8 +796,7 @@ mod tests { assert!(scores.is_null(1)); assert!(scores.is_null(2)); - // JSON predicates filter on document contents, as in the queries - // from issue #6644. + // JSON predicates filter on document contents. for (query, expected_id) in [ ("SELECT id FROM t WHERE json_contains(val, 'extra')", 2), ( @@ -815,8 +814,7 @@ mod tests { assert_eq!(ids.value(0), expected_id, "query: {query}"); } - // `->>` is shorthand for `json_as_text`, and casting `json_get` is - // rewritten to the matching typed getter. + // The `->>` operator and casting `json_get` to a concrete type. let batches = collect_rows( ctx.state(), "SELECT val->>'name' AS name, \ diff --git a/docs.feldera.com/docs/sql/ad-hoc.md b/docs.feldera.com/docs/sql/ad-hoc.md index 787c27f9a9b..725e8eb8a33 100644 --- a/docs.feldera.com/docs/sql/ad-hoc.md +++ b/docs.feldera.com/docs/sql/ad-hoc.md @@ -161,12 +161,13 @@ insert into complex_types values ([1,2,3], struct(2, 'b'), '{"field": 3}', MAP([ ### Querying VARIANT Columns with JSON Functions -Datafusion has no equivalent of the Feldera SQL [`VARIANT`](/sql/json) type. An ad-hoc query +Datafusion currently has no equivalent of the Feldera SQL [`VARIANT`](/sql/json) type. An ad-hoc query receives a `VARIANT` column as a string that holds the JSON-encoded value, so the Feldera SQL subscript syntax (`doc['name']`) does not work in ad-hoc queries. Instead, ad-hoc queries take JSON strings apart with the [datafusion-functions-json](https://github.com/datafusion-contrib/datafusion-functions-json) -function family. These functions also apply to string columns that hold JSON text. +function family. These functions can be used on any string column that can be decoded as +legal JSON. | Function | Return type | Description | |-----------------------------------|-----------------|------------------------------------------------------------------------| @@ -202,15 +203,9 @@ WHERE json_get_bool(doc, 'active') = TRUE AND json_get_int(doc, 'scores', 0) > 5; ``` -Notes: - -- `json_get` returns a value of a union type that only the `text` and `arrow_ipc` output - formats can render. To use `json_get` results in the `json` or `parquet` output format, - cast them to a concrete type; the cast is rewritten to the matching typed getter, so - `json_get(doc, 'scores', 0)::BIGINT` and `CAST(json_get(doc, 'scores', 0) AS BIGINT)` - both execute as `json_get_int(doc, 'scores', 0)`. -- Of the operators provided by the crate, only `->>` (alias of `json_as_text`) is - available; `->` and `?` do not parse in the SQL dialect used by ad-hoc queries. +`json_get` returns a value of a special union type. Prefer the typed getters +(`json_get_int`, ...) or cast the result to a concrete type, e.g. +`json_get(doc, 'scores', 0)::BIGINT`. ### Parameterized Queries with PREPARE / EXECUTE diff --git a/python/tests/platform/test_shared_pipeline.py b/python/tests/platform/test_shared_pipeline.py index 9949fb85812..d51663aef60 100644 --- a/python/tests/platform/test_shared_pipeline.py +++ b/python/tests/platform/test_shared_pipeline.py @@ -210,8 +210,6 @@ def test_adhoc_query_arrow(self): def test_adhoc_json_functions(self): """ - -- Ad-hoc queries read VARIANT columns as JSON-encoded strings and - -- take them apart with datafusion-functions-json (issue #6644). CREATE TABLE json_docs (id INT NOT NULL, doc VARIANT) WITH ('materialized' = 'true'); """ self.pipeline.start() @@ -239,9 +237,6 @@ def test_adhoc_json_functions(self): "active": True, } - # Typed getters pull concrete values out of the VARIANT column. - # A missing key ('scores' of row 2 has one element; indexes are - # 0-based) and a NULL document both yield NULL. got = list( self.pipeline.query( "SELECT id, json_get_str(doc, 'name') AS name," @@ -278,8 +273,6 @@ def test_adhoc_json_functions(self): ) assert got == [{"id": 1}] - # `->>` is shorthand for json_as_text; casting json_get is rewritten - # to the matching typed getter (json_get_int here). got = list( self.pipeline.query( "SELECT doc->>'name' AS name,"