TheRakeshPurohit
diff --git a/‎crates/runtime/Cargo.toml‎
Lines changed: 4 additions & 0 deletions b/‎crates/runtime/Cargo.toml‎
Lines changed: 4 additions & 0 deletions
diff --git a/‎crates/runtime/benches/flatten_json_properties.rs‎
Lines changed: 98 additions & 0 deletions b/‎crates/runtime/benches/flatten_json_properties.rs‎
Lines changed: 98 additions & 0 deletions
diff --git a/‎crates/runtime/src/datafusion/mod.rs‎
Lines changed: 1 addition & 0 deletions b/‎crates/runtime/src/datafusion/mod.rs‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎crates/runtime/src/datafusion/udf.rs‎
Lines changed: 19 additions & 0 deletions b/‎crates/runtime/src/datafusion/udf.rs‎
Lines changed: 19 additions & 0 deletions
@@ -374,3 +374,7 @@ vortex-datafusion.workspace = true
 [[bench]]
 harness = false
 name = "prepared_statement"
+
+[[bench]]
+harness = false
+name = "flatten_json_properties"
@@ -0,0 +1,98 @@
+#![allow(clippy::expect_used)]
+
+//! Benchmarks for `flatten_json_properties`.
+//!
+//! Exercises the walker in isolation (no `DataFusion` plumbing) so regressions
+//! attributable to the walker itself surface without noise from query planning
+//! or Arrow I/O. `bench_catalog_simulation` approximates the typical
+//! materialization shape — 1k schemas × 50 fields per schema.
+
+use std::hint::black_box;
+
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use runtime::datafusion::udtf::json_properties::{FlattenOptions, flatten_with_options};
+
+fn synthetic_schema(num_fields: usize) -> String {
+    // One flat object with `num_fields` primitive properties. Representative of
+    // a wide data-product schema where most fields are leaves.
+    let mut props = String::from("{");
+    for i in 0..num_fields {
+        if i > 0 {
+            props.push(',');
+        }
+        props.push_str(&format!(
+            r#""field_{i}":{{"type":"string","description":"Field {i}","format":"text"}}"#
+        ));
+    }
+    props.push('}');
+    format!(r#"{{"properties":{props}}}"#)
+}
+
+fn nested_schema(depth: usize) -> String {
+    // Deeply nested single-chain schema. Exercises the recursion path.
+    let mut inner = String::from(r#"{"type":"string"}"#);
+    for _ in 0..depth {
+        inner = format!(r#"{{"type":"object","properties":{{"n":{inner}}}}}"#);
+    }
+    format!(r#"{{"properties":{{"root":{inner}}}}}"#)
+}
+
+fn bench_flat_schemas(c: &mut Criterion) {
+    let opts = FlattenOptions {
+        include_internal: true,
+        ..FlattenOptions::default()
+    };
+    let mut group = c.benchmark_group("flatten_json_properties/flat");
+    for fields in [16usize, 128, 512] {
+        let doc = synthetic_schema(fields);
+        group.throughput(Throughput::Elements(fields as u64));
+        group.bench_with_input(BenchmarkId::new("fields", fields), &doc, |b, doc| {
+            b.iter(|| {
+                let rows = flatten_with_options(black_box(doc), &opts);
+                black_box(rows);
+            });
+        });
+    }
+    group.finish();
+}
+
+fn bench_nested_schemas(c: &mut Criterion) {
+    let opts = FlattenOptions {
+        include_internal: true,
+        max_depth: 32,
+        ..FlattenOptions::default()
+    };
+    let mut group = c.benchmark_group("flatten_json_properties/nested");
+    for depth in [4usize, 8, 16] {
+        let doc = nested_schema(depth);
+        group.throughput(Throughput::Elements(depth as u64));
+        group.bench_with_input(BenchmarkId::new("depth", depth), &doc, |b, doc| {
+            b.iter(|| {
+                let rows = flatten_with_options(black_box(doc), &opts);
+                black_box(rows);
+            });
+        });
+    }
+    group.finish();
+}
+
+fn bench_catalog_simulation(c: &mut Criterion) {
+    let opts = FlattenOptions::default();
+    let doc = synthetic_schema(50);
+    c.bench_function("flatten_json_properties/catalog_1k_schemas", |b| {
+        b.iter(|| {
+            for _ in 0..1000 {
+                let rows = flatten_with_options(black_box(&doc), &opts);
+                black_box(rows);
+            }
+        });
+    });
+}
+
+criterion_group!(
+    benches,
+    bench_flat_schemas,
+    bench_nested_schemas,
+    bench_catalog_simulation
+);
+criterion_main!(benches);
@@ -131,6 +131,7 @@ pub mod secrets_context_extension;
 pub mod sort_columns;
 pub(crate) mod sql_validator;
 pub mod udf;
+pub mod udtf;
 
 pub const SPICE_DEFAULT_CATALOG: &str = "spice";
 pub const SPICE_RUNTIME_SCHEMA: &str = "runtime";
 
@@ -17,6 +17,10 @@ limitations under the License.
 use std::collections::HashSet;
 use std::sync::{Arc, LazyLock};
 
+use crate::datafusion::udtf::json_properties::{
+    FLATTEN_JSON_PROPERTIES_UDTF_NAME, FlattenJsonPropertiesScalar, FlattenJsonPropertiesTableFunc,
+};
+use crate::datafusion::udtf::json_tree::{JSON_TREE_UDTF_NAME, JsonTreeScalar, JsonTreeTableFunc};
 use crate::embeddings::udtf::{VECTOR_SEARCH_UDTF_NAME, VectorSearchTableFunc};
 use crate::search::full_text::udtf::{TEXT_SEARCH_UDTF_NAME, TextSearchTableFunc};
 use crate::search::rrf;
@@ -80,6 +84,17 @@ pub async fn register_udfs(runtime: &crate::Runtime) {
         Arc::new(rrf::ReciprocalRankFusion::from_ctx(ctx)),
     );
 
+    // `flatten_json_properties` / `json_tree` — JSON-Schema and generic JSON
+    // shredders. Registered as both UDTF (FROM-clause, literal input) and
+    // ScalarUDF returning `List<Struct<...>>` (per-row / LATERAL via UNNEST).
+    ctx.register_udtf(
+        FLATTEN_JSON_PROPERTIES_UDTF_NAME,
+        Arc::new(FlattenJsonPropertiesTableFunc::new()),
+    );
+    ctx.register_udf(FlattenJsonPropertiesScalar::new().into());
+    ctx.register_udtf(JSON_TREE_UDTF_NAME, Arc::new(JsonTreeTableFunc::new()));
+    ctx.register_udf(JsonTreeScalar::new().into());
+
     #[cfg(feature = "models")]
     {
         ctx.register_udf(embed::Embed::new(runtime.embeds()).into());
@@ -101,6 +116,8 @@ static DENY_SPICE_SPECIFIC_FUNCTIONS: LazyLock<FunctionSupport> = LazyLock::new(
         #[cfg(feature = "models")]
         AI_UDF_NAME,
         DIGEST_UDF_NAME,
+        FLATTEN_JSON_PROPERTIES_UDTF_NAME,
+        JSON_TREE_UDTF_NAME,
     ];
 
     FunctionSupport::new(
@@ -191,6 +208,8 @@ mod tests {
             spice_udf(Bucket::new()),
             spice_udf(Truncate::new()),
             Arc::new(INSTANCE.clone()),
+            spice_udf(FlattenJsonPropertiesScalar::new()),
+            spice_udf(JsonTreeScalar::new()),
         ];
 
         for udf in spice_udfs {