Support Non-Literal Expressions in Substrait VirtualTable Values and Improve Round-Trip Robustness (#18866)

## Which issue does this PR close?

* Closes #16276.

## Rationale for this change

This PR addresses failures in Substrait round‑trip mode where Virtual
Tables were limited to accepting only literals or aliased literals.
Several SQLLogicTests—especially those involving nested or scalar
functions such as `map()`—failed because expressions were rejected
prematurely.

Allowing expression-based rows when exporting Virtual Tables is
essential to making Substrait → DataFusion → Substrait round‑trips
stable and representative of real workloads. Fixing this improves test
coverage, unblocks Substrait integration efforts, and eliminates a major
source of round‑trip inconsistencies.

### Before
```

❯ cargo test --test sqllogictests -- --substrait-round-trip map.slt:388
    Finished `test` profile [unoptimized + debuginfo] target(s) in 0.66s
     Running bin/sqllogictests.rs (target/debug/deps/sqllogictests-d29eafa7e841495b)
Completed 1 test files in 0 seconds                                                                  External error: 2 errors in file /Users/kosiew/GitHub/datafusion/datafusion/sqllogictest/test_files/map.slt

1. query failed: DataFusion error: This feature is not implemented: Unsupported plan type: DescribeTable { schema: Schema ...
[SQL] describe data;
at /Users/kosiew/GitHub/datafusion/datafusion/sqllogictest/test_files/map.slt:43


2. query failed: DataFusion error: Substrait error: Only literal types and aliases are supported in Virtual Tables, got: ScalarFunction
[SQL] VALUES (MAP(['a'], [1])), (MAP(['b'], [2])), (MAP(['c', 'a'], [3, 1]))
at /Users/kosiew/GitHub/datafusion/datafusion/sqllogictest/test_files/map.slt:388


```

### After
```
❯ cargo test --test sqllogictests -- --substrait-round-trip map.slt:388
    Finished `test` profile [unoptimized + debuginfo] target(s) in 0.50s
     Running bin/sqllogictests.rs (target/debug/deps/sqllogictests-62df8e0045f6176f)
Completed 1 test files in 0 seconds
``` 


## What changes are included in this PR?

* Introduces `convert_literal_rows` and `convert_expression_rows`
helpers to properly distinguish literal-only rows from expression-based
rows.
* Adds support for generating nested `Struct` expressions (Substrait
`NestedStruct`) for non-literal Virtual Table values.
* Updates `from_values` to dynamically choose between literal structs
and nested structs.
* Ensures schema consistency validation for expression-based rows.
* Adds new round‑trip test cases, including
`roundtrip_values_with_scalar_function`.
* Refactors tests to use a new `substrait_roundtrip` helper to reduce
duplication.
* Adds `LogicalPlan::DescribeTable(_)` handling in the round‑trip
engine.

## Are these changes tested?

Yes. New test coverage includes:

* A dedicated round‑trip test for scalar functions inside `VALUES`
clauses.
* Use of `substrait_roundtrip` across multiple existing tests to ensure
consistent behavior.
* Validation of expression-based row handling, alias stripping, and
schema equivalence.

## Are there any user-facing changes?

No direct API-breaking changes. The behavior of Substrait round‑trip
conversion improves and becomes more permissive regarding expressions in
Virtual Tables. This is internal to Substrait/DataFusion integration and
does not change SQL semantics or user-level APIs.

## LLM-generated code disclosure

This PR includes LLM-generated code and comments. All LLM-generated
content has been manually reviewed and tested.

Author: kosiew

datafusion/sqllogictest/src/engines/datafusion_substrait_roundtrip_engine/runner.rs

@@ -152,6 +152,7 @@ async fn run_query_substrait_round_trip(
         | LogicalPlan::Explain(_)
         | LogicalPlan::Dml(_)
         | LogicalPlan::Copy(_)
+        | LogicalPlan::DescribeTable(_)
         | LogicalPlan::Statement(_) => df.logical_plan().clone(),
         // For any other plan, convert to Substrait
         logical_plan => {

datafusion/substrait/src/logical_plan/producer/rel/read_rel.rs

@@ -22,13 +22,76 @@ use datafusion::common::{not_impl_err, substrait_datafusion_err, DFSchema, ToDFS
 use datafusion::logical_expr::utils::conjunction;
 use datafusion::logical_expr::{EmptyRelation, Expr, TableScan, Values};
 use std::sync::Arc;
-use substrait::proto::expression::literal::Struct;
+use substrait::proto::expression::literal::Struct as LiteralStruct;
 use substrait::proto::expression::mask_expression::{StructItem, StructSelect};
+use substrait::proto::expression::nested::Struct as NestedStruct;
 use substrait::proto::expression::MaskExpression;
 use substrait::proto::read_rel::{NamedTable, ReadType, VirtualTable};
 use substrait::proto::rel::RelType;
 use substrait::proto::{ReadRel, Rel};
 
+/// Converts rows of literal expressions into Substrait literal structs.
+///
+/// Each row is expected to contain only `Expr::Literal` or `Expr::Alias` wrapping literals.
+/// Aliases are unwrapped and the underlying literal is converted.
+fn convert_literal_rows(
+    producer: &mut impl SubstraitProducer,
+    rows: &[Vec<Expr>],
+) -> datafusion::common::Result<Vec<LiteralStruct>> {
+    rows.iter()
+        .map(|row| {
+            let fields = row
+                .iter()
+                .map(|expr| match expr {
+                    Expr::Literal(sv, _) => to_substrait_literal(producer, sv),
+                    Expr::Alias(alias) => match alias.expr.as_ref() {
+                        // The schema gives us the names, so we can skip aliases
+                        Expr::Literal(sv, _) => to_substrait_literal(producer, sv),
+                        _ => Err(substrait_datafusion_err!(
+                            "Only literal types can be aliased in Virtual Tables, got: {}",
+                            alias.expr.variant_name()
+                        )),
+                    },
+                    _ => Err(substrait_datafusion_err!(
+                        "Only literal types and aliases are supported in Virtual Tables, got: {}",
+                        expr.variant_name()
+                    )),
+                })
+                .collect::<datafusion::common::Result<_>>()?;
+            Ok(LiteralStruct { fields })
+        })
+        .collect()
+}
+
+/// Converts rows of arbitrary expressions into Substrait nested structs.
+///
+/// Validates that each row has the expected schema length and converts each expression
+/// using the producer's expression handler.
+fn convert_expression_rows(
+    producer: &mut impl SubstraitProducer,
+    rows: &[Vec<Expr>],
+    schema_len: usize,
+    empty_schema: &Arc<DFSchema>,
+) -> datafusion::common::Result<Vec<NestedStruct>> {
+    rows.iter()
+        .map(|row| {
+            if row.len() != schema_len {
+                return Err(substrait_datafusion_err!(
+                    "Names list must match exactly to nested schema, but found {} uses for {} names",
+                    row.len(),
+                    schema_len
+                ));
+            }
+
+            let fields = row
+                .iter()
+                .map(|expr| producer.handle_expr(expr, empty_schema))
+                .collect::<datafusion::common::Result<_>>()?;
+            Ok(NestedStruct { fields })
+        })
+        .collect()
+}
+
 pub fn from_table_scan(
     producer: &mut impl SubstraitProducer,
     scan: &TableScan,
@@ -111,29 +174,25 @@ pub fn from_values(
     producer: &mut impl SubstraitProducer,
     v: &Values,
 ) -> datafusion::common::Result<Box<Rel>> {
-    let values = v
-        .values
-        .iter()
-        .map(|row| {
-            let fields = row
-                .iter()
-                .map(|v| match v {
-                    Expr::Literal(sv, _) => to_substrait_literal(producer, sv),
-                    Expr::Alias(alias) => match alias.expr.as_ref() {
-                        // The schema gives us the names, so we can skip aliases
-                        Expr::Literal(sv, _) => to_substrait_literal(producer, sv),
-                        _ => Err(substrait_datafusion_err!(
-                                    "Only literal types can be aliased in Virtual Tables, got: {}", alias.expr.variant_name()
-                                )),
-                    },
-                    _ => Err(substrait_datafusion_err!(
-                                "Only literal types and aliases are supported in Virtual Tables, got: {}", v.variant_name()
-                            )),
-                })
-                .collect::<datafusion::common::Result<_>>()?;
-            Ok(Struct { fields })
+    let schema_len = v.schema.fields().len();
+    let empty_schema = Arc::new(DFSchema::empty());
+
+    let use_literals = v.values.iter().all(|row| {
+        row.iter().all(|expr| match expr {
+            Expr::Literal(_, _) => true,
+            Expr::Alias(alias) => matches!(alias.expr.as_ref(), Expr::Literal(_, _)),
+            _ => false,
         })
-        .collect::<datafusion::common::Result<_>>()?;
+    });
+
+    let (values, expressions) = if use_literals {
+        let values = convert_literal_rows(producer, &v.values)?;
+        (values, vec![])
+    } else {
+        let expressions =
+            convert_expression_rows(producer, &v.values, schema_len, &empty_schema)?;
+        (vec![], expressions)
+    };
     #[allow(deprecated)]
     Ok(Box::new(Rel {
         rel_type: Some(RelType::Read(Box::new(ReadRel {
@@ -145,7 +204,7 @@ pub fn from_values(
             advanced_extension: None,
             read_type: Some(ReadType::VirtualTable(VirtualTable {
                 values,
-                expressions: vec![],
+                expressions,
             })),
         }))),
     }))

datafusion/substrait/tests/cases/roundtrip_logical_plan.rs

@@ -17,6 +17,8 @@
 
 use crate::utils::test::read_json;
 use datafusion::arrow::array::ArrayRef;
+use datafusion::functions_nested::map::map;
+use datafusion::logical_expr::LogicalPlanBuilder;
 use datafusion::physical_plan::Accumulator;
 use datafusion::scalar::ScalarValue;
 use datafusion_substrait::logical_plan::{
@@ -26,6 +28,7 @@ use std::cmp::Ordering;
 use std::mem::size_of_val;
 
 use datafusion::arrow::datatypes::{DataType, Field, IntervalUnit, Schema, TimeUnit};
+use datafusion::common::tree_node::Transformed;
 use datafusion::common::{not_impl_err, plan_err, DFSchema, DFSchemaRef};
 use datafusion::error::Result;
 use datafusion::execution::registry::SerializerRegistry;
@@ -1266,6 +1269,34 @@ async fn roundtrip_values_no_columns() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn roundtrip_values_with_scalar_function() -> Result<()> {
+    let ctx = create_context().await?;
+    //  datafusion::functions_nested::map::map;
+    let expr = map(vec![lit("a")], vec![lit(1)]);
+    let plan = LogicalPlanBuilder::values(vec![vec![expr]])?.build()?;
+    let expected = ctx.state().optimize(&plan)?;
+
+    let actual = substrait_roundtrip(&plan, &ctx).await?;
+
+    let strip_aliases_from_values = |plan: &LogicalPlan| -> LogicalPlan {
+        plan.clone()
+            .map_expressions(|expr| Ok(Transformed::yes(expr.unalias())))
+            .map(|t| t.data)
+            .unwrap_or_else(|_| plan.clone())
+    };
+
+    let normalized_expected = strip_aliases_from_values(&expected);
+    let normalized_actual = strip_aliases_from_values(&actual);
+
+    assert_eq!(
+        format!("{normalized_expected}"),
+        format!("{normalized_actual}")
+    );
+    assert_eq!(normalized_expected.schema(), normalized_actual.schema());
+    Ok(())
+}
+
 #[tokio::test]
 async fn roundtrip_values_empty_relation() -> Result<()> {
     roundtrip("SELECT * FROM (VALUES ('a')) LIMIT 0").await
@@ -1449,9 +1480,7 @@ async fn roundtrip_repartition_roundrobin() -> Result<()> {
         partitioning_scheme: Partitioning::RoundRobinBatch(8),
     });
 
-    let proto = to_substrait_plan(&plan, &ctx.state())?;
-    let plan2 = from_substrait_plan(&ctx.state(), &proto).await?;
-    let plan2 = ctx.state().optimize(&plan2)?;
+    let plan2 = substrait_roundtrip(&plan, &ctx).await?;
 
     assert_eq!(format!("{plan}"), format!("{plan2}"));
     Ok(())
@@ -1466,9 +1495,7 @@ async fn roundtrip_repartition_hash() -> Result<()> {
         partitioning_scheme: Partitioning::Hash(vec![col("data.a")], 8),
     });
 
-    let proto = to_substrait_plan(&plan, &ctx.state())?;
-    let plan2 = from_substrait_plan(&ctx.state(), &proto).await?;
-    let plan2 = ctx.state().optimize(&plan2)?;
+    let plan2 = substrait_roundtrip(&plan, &ctx).await?;
 
     assert_eq!(format!("{plan}"), format!("{plan2}"));
     Ok(())
@@ -1650,9 +1677,7 @@ async fn assert_expected_plan(
     let ctx = create_context().await?;
     let df = ctx.sql(sql).await?;
     let plan = df.into_optimized_plan()?;
-    let proto = to_substrait_plan(&plan, &ctx.state())?;
-    let plan2 = from_substrait_plan(&ctx.state(), &proto).await?;
-    let plan2 = ctx.state().optimize(&plan2)?;
+    let plan2 = substrait_roundtrip(&plan, &ctx).await?;
 
     if assert_schema {
         assert_eq!(plan.schema(), plan2.schema());
@@ -1694,9 +1719,7 @@ async fn roundtrip_fill_na(sql: &str) -> Result<()> {
     let ctx = create_context().await?;
     let df = ctx.sql(sql).await?;
     let plan = df.into_optimized_plan()?;
-    let proto = to_substrait_plan(&plan, &ctx.state())?;
-    let plan2 = from_substrait_plan(&ctx.state(), &proto).await?;
-    let plan2 = ctx.state().optimize(&plan2)?;
+    let plan2 = substrait_roundtrip(&plan, &ctx).await?;
 
     // Format plan string and replace all None's with 0
     let plan1str = format!("{plan}").replace("None", "0");
@@ -1708,6 +1731,18 @@ async fn roundtrip_fill_na(sql: &str) -> Result<()> {
     Ok(())
 }
 
+/// Converts a logical plan to Substrait and back, applying optimization.
+/// Returns the roundtripped and optimized logical plan.
+async fn substrait_roundtrip(
+    plan: &LogicalPlan,
+    ctx: &SessionContext,
+) -> Result<LogicalPlan> {
+    let proto = to_substrait_plan(plan, &ctx.state())?;
+    let plan2 = from_substrait_plan(&ctx.state(), &proto).await?;
+    let plan2 = ctx.state().optimize(&plan2)?;
+    Ok(plan2)
+}
+
 async fn test_alias(sql_with_alias: &str, sql_no_alias: &str) -> Result<()> {
     // Since we ignore the SubqueryAlias in the producer, the result should be
     // the same as producing a Substrait plan from the same query without aliases
@@ -1735,8 +1770,7 @@ async fn roundtrip_logical_plan_with_ctx(
     ctx: SessionContext,
 ) -> Result<Box<Plan>> {
     let proto = to_substrait_plan(&plan, &ctx.state())?;
-    let plan2 = from_substrait_plan(&ctx.state(), &proto).await?;
-    let plan2 = ctx.state().optimize(&plan2)?;
+    let plan2 = substrait_roundtrip(&plan, &ctx).await?;
 
     let plan1str = format!("{plan}");
     let plan2str = format!("{plan2}");