Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 34 additions & 8 deletions datafusion/substrait/src/logical_plan/consumer/rel/read_rel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -148,19 +148,45 @@ pub async fn from_read_rel(
let values = if !vt.expressions.is_empty() {
let mut exprs = vec![];
for row in &vt.expressions {
if row.fields.len() != substrait_schema.fields().len() {
return substrait_err!(
"Field count mismatch: expected {} fields but found {} in virtual table row",
substrait_schema.fields().len(),
row.fields.len()
);
}

let mut row_exprs = vec![];
let mut name_idx = 0;
for expression in &row.fields {
let expr = consumer
.consume_expression(expression, &substrait_schema)
.await?;
let expr = match expression.rex_type.as_ref() {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Both branches increment name_idx by 1. I think we can extract that out out before the match and expand on the comment with something like

// Top-level names are provided through schema
// Each expression consumes at least one name, and Literals may consume additional names.
name_idx += 1;

to provide a bit more context about what is happening.

Some(substrait::proto::expression::RexType::Literal(lit)) => {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The first questions that readers might have is:

Why does the Literal expression type needs special handling? Adding a comment with some context about why this special handling is needed can save some investigation time from readers.

name_idx += 1; // top-level names are provided through schema
Expr::Literal(
from_substrait_literal(
consumer,
lit,
&named_struct.names,
&mut name_idx,
)?,
None,
)
Comment thread
eliot1480 marked this conversation as resolved.
}
_ => {
name_idx += 1; // top-level names are provided through schema
consumer
.consume_expression(expression, &substrait_schema)
.await?
}
};
row_exprs.push(expr);
}
// For expressions, validate against top-level schema fields, not nested names
if row_exprs.len() != substrait_schema.fields().len() {

if name_idx != named_struct.names.len() {
return substrait_err!(
"Field count mismatch: expected {} fields but found {} in virtual table row",
substrait_schema.fields().len(),
row_exprs.len()
"Names list must match exactly to nested schema, but found {} uses for {} names",
name_idx,
named_struct.names.len()
);
}
exprs.push(row_exprs);
Expand Down
78 changes: 9 additions & 69 deletions datafusion/substrait/src/logical_plan/producer/rel/read_rel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,55 +15,19 @@
// specific language governing permissions and limitations
// under the License.

use crate::logical_plan::producer::{
SubstraitProducer, to_substrait_literal, to_substrait_named_struct,
};
use crate::logical_plan::producer::{SubstraitProducer, to_substrait_named_struct};
use datafusion::common::{DFSchema, ToDFSchema, substrait_datafusion_err};
use datafusion::logical_expr::utils::conjunction;
use datafusion::logical_expr::{EmptyRelation, Expr, TableScan, Values};
use datafusion::scalar::ScalarValue;
use std::sync::Arc;
use substrait::proto::expression::MaskExpression;
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::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
Expand Down Expand Up @@ -163,6 +127,7 @@ pub fn from_empty_relation(
let base_schema = to_substrait_named_struct(producer, &e.schema)?;

let read_type = if e.produce_one_row {
let empty_schema = Arc::new(DFSchema::empty());
// Create one row with default scalar values for each field in the schema.
// For example, an Int32 field gets Int32(NULL), a Utf8 field gets Utf8(NULL), etc.
// This represents the "phantom row" that provides a context for evaluating
Expand All @@ -173,25 +138,16 @@ pub fn from_empty_relation(
.iter()
.map(|f| {
let scalar = ScalarValue::try_from(f.data_type())?;
to_substrait_literal(producer, &scalar)
producer.handle_expr(&Expr::Literal(scalar, None), &empty_schema)
})
.collect::<datafusion::common::Result<_>>()?;

ReadType::VirtualTable(VirtualTable {
// Use deprecated 'values' field instead of 'expressions' because the consumer's
// nested expression support (RexType::Nested) is not yet implemented.
// The 'values' field uses literal::Struct which the consumer can properly
// deserialize with field name preservation.
#[expect(deprecated)]
values: vec![LiteralStruct { fields }],
expressions: vec![],
Comment thread
eliot1480 marked this conversation as resolved.
expressions: vec![NestedStruct { fields }],
..Default::default()
})
} else {
ReadType::VirtualTable(VirtualTable {
#[expect(deprecated)]
values: vec![],
expressions: vec![],
})
ReadType::VirtualTable(VirtualTable::default())
};
Ok(Box::new(Rel {
rel_type: Some(RelType::Read(Box::new(ReadRel {
Expand All @@ -212,23 +168,8 @@ pub fn from_values(
) -> datafusion::common::Result<Box<Rel>> {
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,
})
});

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)
};
let expressions =
convert_expression_rows(producer, &v.values, schema_len, &empty_schema)?;
Ok(Box::new(Rel {
rel_type: Some(RelType::Read(Box::new(ReadRel {
common: None,
Expand All @@ -237,10 +178,9 @@ pub fn from_values(
best_effort_filter: None,
projection: None,
advanced_extension: None,
#[expect(deprecated)]
read_type: Some(ReadType::VirtualTable(VirtualTable {
values,
expressions,
..Default::default()
})),
}))),
}))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -100,29 +100,52 @@
}
},
"virtualTable": {
"values": [{
"fields": [{
"string": "aaa",
"nullable": true
}, {
"string": "host-a",
"nullable": true
}, {
"i64": "128",
"nullable": true
}]
}, {
"fields": [{
"string": "bbb",
"nullable": true
}, {
"string": "host-b",
"nullable": true
}, {
"i64": "256",
"nullable": true
}]
}]
"expressions": [
{
"fields": [
{
"literal": {
"string": "aaa",
"nullable": true
}
},
{
"literal": {
"string": "host-a",
"nullable": true
}
},
{
"literal": {
"i64": "128",
"nullable": true
}
}
]
},
{
"fields": [
{
"literal": {
"string": "bbb",
"nullable": true
}
},
{
"literal": {
"string": "host-b",
"nullable": true
}
},
{
"literal": {
"i64": "256",
"nullable": true
}
}
]
}
]
}
}
},
Expand Down Expand Up @@ -293,23 +316,40 @@
}
},
"virtualTable": {
"values": [{
"fields": [{
"string": "host-a",
"nullable": true
}, {
"i64": "107",
"nullable": true
}]
}, {
"fields": [{
"string": "host-b",
"nullable": true
}, {
"i64": "214",
"nullable": true
}]
}]
"expressions": [
{
"fields": [
{
"literal": {
"string": "host-a",
"nullable": true
}
},
{
"literal": {
"i64": "107",
"nullable": true
}
}
]
},
{
"fields": [
{
"literal": {
"string": "host-b",
"nullable": true
}
},
{
"literal": {
"i64": "214",
"nullable": true
}
}
]
}
]
}
}
},
Expand Down Expand Up @@ -365,29 +405,52 @@
}
},
"virtualTable": {
"values": [{
"fields": [{
"string": "aaa",
"nullable": true
}, {
"string": "host-a",
"nullable": true
}, {
"i64": "128",
"nullable": true
}]
}, {
"fields": [{
"string": "bbb",
"nullable": true
}, {
"string": "host-b",
"nullable": true
}, {
"i64": "256",
"nullable": true
}]
}]
"expressions": [
{
"fields": [
{
"literal": {
"string": "aaa",
"nullable": true
}
},
{
"literal": {
"string": "host-a",
"nullable": true
}
},
{
"literal": {
"i64": "128",
"nullable": true
}
}
]
},
{
"fields": [
{
"literal": {
"string": "bbb",
"nullable": true
}
},
{
"literal": {
"string": "host-b",
"nullable": true
}
},
{
"literal": {
"i64": "256",
"nullable": true
}
}
]
}
]
}
}
},
Expand Down
Loading
Loading