feat(datafusion): support lateral collection generators

crates/integrations/datafusion/src/catalog.rs

@@ -1426,7 +1426,8 @@ async fn validate_view_dependencies(
 }
 
 fn view_relation_identifiers(view: &View, catalog_name: &str) -> DFResult<Vec<Identifier>> {
-    let parseable = crate::regex_operators::parseable_sql(view.query_for("datafusion"))?;
+    let regex_sql = crate::regex_operators::parseable_sql(view.query_for("datafusion"))?;
+    let parseable = crate::lateral_views::parseable_sql(regex_sql.as_ref())?;
     let statements =
         Parser::parse_sql(&GenericDialect {}, parseable.as_ref()).map_err(|error| {
             plan_datafusion_err!(

crates/integrations/datafusion/src/lib.rs

@@ -56,6 +56,7 @@ mod golden_warehouse;
 mod hybrid_search;
 mod json_functions;
 mod lateral_vector_search;
+mod lateral_views;
 mod memory;
 mod merge_condition;
 mod merge_into;

crates/integrations/datafusion/src/sql_context.rs

@@ -158,9 +158,10 @@ impl SQLContextBuilder {
             .with_config(crate::lateral_vector_search::session_config())
             .with_default_features()
             .with_type_planner(Arc::new(crate::binary_cast::BinaryTypePlanner))
-            .with_relation_planners(vec![Arc::new(
-                crate::relation_planner::PaimonRelationPlanner::new(),
-            )])
+            .with_relation_planners(vec![
+                Arc::new(crate::lateral_views::ExplodeRelationPlanner),
+                Arc::new(crate::relation_planner::PaimonRelationPlanner::new()),
+            ])
             .with_optimizer_rules(crate::lateral_vector_search::optimizer_rules())
             .with_query_planner(Arc::new(
                 crate::lateral_vector_search::PaimonQueryPlanner::new(),
@@ -172,6 +173,7 @@ impl SQLContextBuilder {
         let ctx = SessionContext::new_with_state(state);
         crate::compat_functions::register_compat_functions(&ctx);
         crate::binary_cast::register(&ctx);
+        crate::lateral_views::register(&ctx);
         crate::scalar_compat::register(&ctx);
         crate::blob_descriptor_functions::register_blob_descriptor_functions(&ctx);
         crate::variant_functions::register_variant_functions(&ctx);
@@ -501,7 +503,8 @@ impl SQLContext {
         let changed = !persistent_definition && {
             let binary_changed = crate::binary_cast::rewrite(&mut statements[0])?;
             let regex_changed = crate::regex_operators::rewrite(&mut statements[0]);
-            binary_changed || regex_changed
+            let lateral_changed = crate::lateral_views::rewrite(&mut statements[0])?;
+            binary_changed || regex_changed || lateral_changed
         };
         let sql = if changed {
             rewritten = statements[0].to_string();
@@ -2692,7 +2695,8 @@ fn validate_persistent_create_function(create_function: &CreateFunction) -> DFRe
 }
 
 fn parse_sql_statements(sql: &str) -> DFResult<Vec<Statement>> {
-    let parseable = crate::regex_operators::parseable_sql(sql)?;
+    let regex_sql = crate::regex_operators::parseable_sql(sql)?;
+    let parseable = crate::lateral_views::parseable_sql(regex_sql.as_ref())?;
     let sql = parseable.as_ref();
     let dialect = GenericDialect {};
     let mut tokens = Tokenizer::new(&dialect, sql)

crates/integrations/datafusion/src/sql_function.rs

@@ -61,7 +61,8 @@ async fn expand_sql_internal(
     current_database: &str,
     candidate: Option<&Function>,
 ) -> DFResult<String> {
-    let parseable = crate::regex_operators::parseable_sql(sql)?;
+    let regex_sql = crate::regex_operators::parseable_sql(sql)?;
+    let parseable = crate::lateral_views::parseable_sql(regex_sql.as_ref())?;
     let mut statements = Parser::parse_sql(&GenericDialect {}, parseable.as_ref())
         .map_err(|error| DataFusionError::Plan(format!("Invalid REST SQL: {error}")))?;
     if statements.len() != 1 {
@@ -149,6 +150,7 @@ async fn expand_statement_internal(
     max_expanded_calls: usize,
     candidate: Option<&Function>,
 ) -> DFResult<Statement> {
+    crate::lateral_views::rewrite(&mut statement)?;
     let candidate_reference =
         candidate.map(|function| (current_catalog.to_string(), function.identifier().clone()));
     let mut functions = HashMap::new();

crates/integrations/datafusion/tests/feature_combination.rs

@@ -0,0 +1,65 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+use datafusion::common::ScalarValue;
+use paimon::{catalog::Identifier, Catalog};
+#[tokio::test]
+async fn six_sql_extensions_work_together() {
+    let (_tmp, catalog) = common::create_test_env();
+    let ctx = common::create_sql_context(catalog.clone()).await;
+    ctx.sql("CREATE TABLE default.events (id INT, txt STRING, b VARBINARY) COMMENT 'events'")
+        .await
+        .unwrap();
+    ctx.sql("INSERT INTO default.events VALUES (1,'a,b',CAST('xy' AS BINARY(4))), (2,NULL,CAST(NULL AS BINARY))").await.unwrap().collect().await.unwrap();
+    ctx.sql("ALTER TABLE default.events ALTER COLUMN txt COMMENT 'tokens'")
+        .await
+        .unwrap();
+    ctx.sql("CALL sys.create_tag(table => concat('default.','events'), tag => concat('v','1'), snapshot_id => CAST(1+0 AS BIGINT))").await.unwrap();
+    let table = catalog
+        .get_table(&Identifier::new("default", "events"))
+        .await
+        .unwrap();
+    assert_eq!(table.schema().comment(), Some("events"));
+    assert_eq!(table.schema().fields()[1].description(), Some("tokens"));
+    assert_eq!(
+        table.tag_manager().get("v1").await.unwrap().unwrap().id(),
+        1
+    );
+    let batches=ctx.sql("SELECT id,p.pos,p.val,hex(b),typeof(p.val),size(split(txt,',')),hour(TIME '12:00:00') FROM default.events LATERAL VIEW posexplode_outer(split(txt,',')) p AS pos,val WHERE p.val RLIKE '^a$' OR p.val IS NULL ORDER BY id,p.pos").await.unwrap().collect().await.unwrap();
+    let rows = batches
+        .iter()
+        .flat_map(|b| {
+            (0..b.num_rows()).map(|r| {
+                (0..b.num_columns())
+                    .map(|c| {
+                        ScalarValue::try_from_array(b.column(c), r)
+                            .unwrap()
+                            .to_string()
+                    })
+                    .collect::<Vec<_>>()
+            })
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(
+        rows,
+        vec![
+            vec!["1", "0", "a", "78790000", "string", "2", "12"],
+            vec!["2", "NULL", "NULL", "NULL", "string", "NULL", "12"]
+        ]
+    );
+}

crates/integrations/datafusion/tests/lateral_views.rs

@@ -0,0 +1,169 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use datafusion::common::ScalarValue;
+use paimon_datafusion::SQLContext;
+#[tokio::test]
+async fn lateral_posexplode_outer_preserves_parent_rows_and_zero_based_positions() {
+    let ctx = SQLContext::new();
+    for (suffix, expected) in [
+        ("LIMIT ALL", vec![vec!["1"], vec!["2"]]),
+        ("LIMIT (1)", vec![vec!["1"]]),
+        ("OFFSET (1)", vec![vec!["2"]]),
+    ] {
+        assert_eq!(rows(&ctx,&format!("SELECT lv.col FROM (VALUES (ARRAY[1,2])) t(xs) LATERAL VIEW explode(xs) lv {suffix}")).await,expected);
+    }
+    assert_eq!(rows(&ctx,"SELECT lv.col, row_number() OVER (ORDER BY lv.col) rn FROM (VALUES (ARRAY[1,2])) t(xs) LATERAL VIEW explode(xs) lv QUALIFY rn=1").await,vec![vec!["1","1"]]);
+    let batches=ctx.sql("SELECT id, lv.pos, lv.val FROM (VALUES (1, ARRAY[10,20]), (2, ARRAY[]), (3, CAST(NULL AS INT[])), (4, ARRAY[NULL])) t(id,xs) LATERAL VIEW posexplode_outer(xs) lv AS pos, val ORDER BY id,pos").await.unwrap().collect().await.unwrap();
+    let rows: Vec<_> = batches
+        .iter()
+        .flat_map(|b| {
+            (0..b.num_rows()).map(|r| {
+                (0..b.num_columns())
+                    .map(|c| {
+                        ScalarValue::try_from_array(b.column(c), r)
+                            .unwrap()
+                            .to_string()
+                    })
+                    .collect::<Vec<_>>()
+            })
+        })
+        .collect();
+    assert_eq!(
+        rows,
+        vec![
+            vec!["1", "0", "10"],
+            vec!["1", "1", "20"],
+            vec!["2", "NULL", "NULL"],
+            vec!["3", "NULL", "NULL"],
+            vec!["4", "0", "NULL"]
+        ]
+    );
+}
+
+async fn rows(ctx: &SQLContext, sql: &str) -> Vec<Vec<String>> {
+    ctx.sql(sql)
+        .await
+        .unwrap_or_else(|e| panic!("{sql}: {e}"))
+        .collect()
+        .await
+        .unwrap_or_else(|e| panic!("{sql}: {e}"))
+        .iter()
+        .flat_map(|b| {
+            (0..b.num_rows()).map(|r| {
+                (0..b.num_columns())
+                    .map(|c| {
+                        ScalarValue::try_from_array(b.column(c), r)
+                            .unwrap()
+                            .to_string()
+                    })
+                    .collect()
+            })
+        })
+        .collect()
+}
+#[tokio::test]
+async fn lateral_generators_cover_maps_chaining_filters_and_set_operations() {
+    let ctx = SQLContext::new();
+    assert_eq!(rows(&ctx,"SELECT lv.pos,lv.key,lv.value FROM (SELECT map(ARRAY['a','b'], ARRAY[10,20]) AS m) t LATERAL VIEW posexplode(m) lv ORDER BY lv.pos").await,vec![vec!["0","a","10"],vec!["1","b","20"]]);
+    assert_eq!(rows(&ctx,"SELECT lv.key,lv.value FROM (SELECT map(ARRAY['a'],ARRAY[10]) AS m) t LATERAL VIEW explode(m) lv").await,vec![vec!["a","10"]]);
+    assert_eq!(rows(&ctx,"SELECT a.x,b.y FROM (VALUES (ARRAY[1,2],ARRAY[3,4])) t(xs,ys) LATERAL VIEW explode(xs) a AS x LATERAL VIEW explode(ys) b AS y WHERE a.x=2 ORDER BY b.y").await,vec![vec!["2","3"],vec!["2","4"]]);
+    assert_eq!(rows(&ctx,"SELECT lv.x FROM (VALUES (ARRAY[1,2])) t(xs) LATERAL VIEW explode(xs) lv AS x UNION ALL SELECT 3 ORDER BY x").await,vec![vec!["1"],vec!["2"],vec!["3"]]);
+    assert_eq!(
+        rows(
+            &ctx,
+            "SELECT lv.pos,lv.col FROM (VALUES (ARRAY[])) t(xs) LATERAL VIEW posexplode(xs) lv"
+        )
+        .await,
+        Vec::<Vec<String>>::new()
+    );
+    assert_eq!(
+        rows(
+            &ctx,
+            "SELECT lv.col FROM (VALUES (ARRAY[])) t(xs) LATERAL VIEW OUTER explode(xs) lv"
+        )
+        .await,
+        vec![vec!["NULL"]]
+    );
+    assert_eq!(rows(&ctx,"SELECT \"L V\".\"Val X\" FROM (VALUES (ARRAY[7])) t(xs) LATERAL VIEW explode(xs) \"L V\" AS \"Val X\"").await,vec![vec!["7"]]);
+}
+#[tokio::test]
+async fn lateral_explode_decodes_dictionary_collections_and_value_nulls() {
+    use datafusion::arrow::{
+        array::{Array, ArrayRef, DictionaryArray, Int32Array, Int8Array, ListArray, RecordBatch},
+        datatypes::{Field, Int32Type, Int8Type, Schema},
+    };
+    use std::sync::Arc;
+    let values = ListArray::from_iter_primitive::<Int32Type, _, _>([
+        Some(vec![Some(10), Some(20)]),
+        None,
+        Some(vec![]),
+    ]);
+    let dict = Arc::new(
+        DictionaryArray::<Int8Type>::try_new(
+            Int8Array::from(vec![Some(0), Some(1), None, Some(2)]),
+            Arc::new(values),
+        )
+        .unwrap(),
+    );
+    let batch = RecordBatch::try_new(
+        Arc::new(Schema::new(vec![
+            Field::new("id", datafusion::arrow::datatypes::DataType::Int32, false),
+            Field::new("xs", dict.data_type().clone(), true),
+        ])),
+        vec![
+            Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as ArrayRef,
+            dict,
+        ],
+    )
+    .unwrap();
+    let ctx = SQLContext::new();
+    ctx.ctx().register_batch("dict", batch).unwrap();
+    assert_eq!(rows(&ctx,"SELECT id,lv.pos,lv.col FROM dict LATERAL VIEW posexplode_outer(xs) lv ORDER BY id,lv.pos").await,vec![vec!["1","0","10"],vec!["1","1","20"],vec!["2","NULL","NULL"],vec!["3","NULL","NULL"],vec!["4","NULL","NULL"]]);
+}
+
+#[tokio::test]
+async fn lateral_expansion_preserves_empty_parents_joins_and_dependent_generators() {
+    let ctx = SQLContext::new();
+    assert_eq!(rows(&ctx,"SELECT lv.col FROM (VALUES (ARRAY[1,2])) t(xs) LATERAL VIEW explode(xs) 'lv' LIMIT (1)").await,vec![vec!["1"]]);
+    assert_eq!(rows(&ctx,"SELECT lv.col, row_number() OVER () AS rn FROM (VALUES (ARRAY[1])) t(xs) LATERAL VIEW explode(xs) lv QUALIFY E'a' = 'a' AND rn=1").await,vec![vec!["1","1"]]);
+    let mut union_rows = rows(&ctx,"SELECT a.\"limit\" AS x FROM (VALUES (ARRAY[1])) t(xs) LATERAL VIEW explode(xs) a limit UNION ALL SELECT b.col FROM (VALUES (ARRAY[2])) t(xs) LATERAL VIEW explode(xs) b LIMIT (2)").await;
+    union_rows.sort();
+    assert_eq!(union_rows, vec![vec!["1"], vec!["2"]]);
+    for (table_alias, column_alias) in [("limit", "col"), ("order", "by"), ("fetch", "first")] {
+        let sql = format!(
+            r#"SELECT "{table_alias}"."{column_alias}" AS x FROM (VALUES (ARRAY[1])) t(xs) LATERAL VIEW explode(xs) {table_alias} {column_alias} UNION ALL SELECT b.col FROM (VALUES (ARRAY[2])) t(xs) LATERAL VIEW explode(xs) b LIMIT (2)"#
+        );
+        let mut actual = rows(&ctx, &sql).await;
+        actual.sort();
+        assert_eq!(actual, vec![vec!["1"], vec!["2"]], "{sql}");
+    }
+    assert_eq!(rows(&ctx,"SELECT COUNT(*) FROM (SELECT ARRAY[1] xs WHERE FALSE) t LATERAL VIEW explode_outer(xs) lv").await,vec![vec!["0"]]);
+    assert_eq!(rows(&ctx,"SELECT a.id,b.id,lv.col FROM (VALUES (1,ARRAY[7,7])) a(id,xs) JOIN (VALUES (1)) b(id) ON a.id=b.id LATERAL VIEW explode(a.xs) lv").await,vec![vec!["1","1","7"],vec!["1","1","7"]]);
+    assert_eq!(rows(&ctx,"SELECT b.p,b.v FROM (SELECT ARRAY[ARRAY[1,2],ARRAY[3]] items) t LATERAL VIEW explode(items) a AS xs LATERAL VIEW posexplode(xs) b AS p,v").await,vec![vec!["0","1"],vec!["1","2"],vec!["0","3"]]);
+    for sql in [
+        "SELECT lv.pos,lv.key,lv.value FROM (SELECT CASE WHEN FALSE THEN map(ARRAY['a'],ARRAY[1]) ELSE NULL END m) t LATERAL VIEW posexplode_outer(m) lv",
+        "SELECT lv.pos,lv.key,lv.value FROM (SELECT map(CAST(ARRAY[] AS VARCHAR[]),CAST(ARRAY[] AS INT[])) m) t LATERAL VIEW posexplode_outer(m) lv",
+    ] {assert_eq!(rows(&ctx,sql).await,vec![vec!["NULL","NULL","NULL"]]);}
+    for sql in [
+        "SELECT * FROM (SELECT ARRAY[1] xs) t LATERAL VIEW posexplode(xs) lv AS wrong",
+        "SELECT * FROM (SELECT 1 xs) t LATERAL VIEW explode(xs) lv",
+        "SELECT * FROM (SELECT ARRAY[1] xs) t LATERAL VIEW unsupported_generator(xs) lv",
+    ] {
+        assert!(ctx.sql(sql).await.is_err(), "{sql}");
+    }
+}

docs/src/sql.md

@@ -2005,6 +2005,31 @@ Only the required columns are read, reducing I/O:
 SELECT name FROM paimon.my_db.users;
 ```
 
+### Lateral collection expansion
+
+SQLContext supports `LATERAL VIEW explode`, `posexplode`,
+`explode_outer`, and `posexplode_outer` for arrays and maps. The
+`LATERAL VIEW OUTER` spelling also preserves empty/NULL collections.
+
+```sql
+SELECT t.id, expanded.pos, expanded.value
+FROM events t
+LATERAL VIEW posexplode_outer(t.items) expanded AS pos, value;
+```
+
+Positions are zero-based INT values. Array defaults are `col`, or
+`pos, col`; map defaults are `key, value`, or `pos, key, value`.
+An explicit AS list must have the matching number of column names. A NULL array
+element remains an element with its position. Outer expansion of an empty or
+NULL collection emits one row with NULL positions and values; ordinary
+expansion emits no rows. Empty parent inputs emit no rows in either mode.
+
+Multiple lateral views can reference columns produced by earlier views.
+Qualified/quoted aliases, joins, filters, subqueries and set operations retain
+their normal SQL behavior. Use one FROM relation with explicit joins instead
+of comma-separated FROM items. Unsupported generators and modifiers report
+planning errors.
+
 ### Filter Pushdown
 
 The following filter predicates are pushed down to the Paimon storage layer: