variant support

Cargo.toml

@@ -14,6 +14,7 @@ members = [
   "crates/core-metastore",
   "crates/core-utils",
   "crates/api-sessions",
+  "crates/visitors",
 ]
 resolver = "2"
 package.license-file = "LICENSE"

crates/core-executor/Cargo.toml

@@ -9,7 +9,7 @@ core-utils = { path = "../core-utils" }
 core-metastore = { path = "../core-metastore" }
 df-builtins = { path = "../df-builtins" }
 df-catalog = { path = "../df-catalog" }
-
+visitors = { path = "../visitors" }
 async-trait = { workspace = true }
 aws-config = { workspace = true }
 aws-credential-types = { workspace = true }

crates/core-executor/src/datafusion/mod.rs

@@ -5,6 +5,5 @@ pub mod error;
 pub mod physical_optimizer;
 pub mod planner;
 pub mod type_planner;
-pub mod visitors;
 
 pub use df_builtins as functions;

crates/core-executor/src/query.rs

@@ -53,12 +53,12 @@ use super::datafusion::planner::ExtendedSqlToRel;
 use super::error::{self as ex_error, ExecutionError, ExecutionResult, RefreshCatalogListSnafu};
 use super::session::UserSession;
 use super::utils::{NormalizedIdent, is_logical_plan_effectively_empty};
-use crate::datafusion::visitors::{copy_into_identifiers, functions_rewriter, json_element};
 use df_catalog::catalog::CachingCatalog;
 use df_catalog::catalogs::slatedb::schema::{
     SLATEDB_CATALOG, SLATEDB_SCHEMA, SlateDBViewSchemaProvider,
 };
 use tracing_attributes::instrument;
+use visitors::{copy_into_identifiers, functions_rewriter, json_element};
 
 #[derive(Default, Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
 pub struct QueryContext {

crates/core-executor/src/session.rs

@@ -29,6 +29,7 @@ use df_catalog::catalog_list::{DEFAULT_CATALOG, EmbucketCatalogList};
 // TODO: We need to fix this after geodatafusion is updated to datafusion 47
 //use geodatafusion::udf::native::register_native as register_geo_native;
 use crate::datafusion::physical_optimizer::physical_optimizer_rules;
+use df_builtins::table::register_udtfs;
 use iceberg_rust::object_store::ObjectStoreBuilder;
 use iceberg_s3tables_catalog::S3TablesCatalog;
 use snafu::ResultExt;
@@ -77,6 +78,7 @@ impl UserSession {
         let mut ctx = SessionContext::new_with_state(state);
         register_udfs(&mut ctx).context(ex_error::RegisterUDFSnafu)?;
         register_udafs(&mut ctx).context(ex_error::RegisterUDAFSnafu)?;
+        register_udtfs(&mut ctx);
         register_json_udfs(&mut ctx).context(ex_error::RegisterUDFSnafu)?;
         //register_geo_native(&ctx);
         //register_geo_udfs(&ctx);

crates/df-builtins/Cargo.toml

@@ -5,6 +5,7 @@ edition = "2024"
 license-file.workspace = true
 
 [dependencies]
+visitors = { path = "../visitors" }
 chrono = { workspace = true }
 datafusion = { workspace = true }
 datafusion-common = { workspace = true }
@@ -15,6 +16,7 @@ datafusion-physical-plan = { workspace = true }
 paste = "1"
 serde = { workspace = true }
 serde_json = { workspace = true }
+jsonpath_lib = "0.3.0"
 ahash = { version = "0.8", default-features = false, features = [
     "runtime-rng",
 ] }

crates/df-builtins/src/aggregate/mod.rs

@@ -24,6 +24,7 @@ pub mod array_unique_agg;
 pub mod booland_agg;
 pub mod boolor_agg;
 pub mod boolxor_agg;
+pub mod object_agg;
 pub mod percentile_cont;
 
 pub fn register_udafs(registry: &mut dyn FunctionRegistry) -> datafusion_common::Result<()> {