Skip to content

feat: removed all instances of deprecated virtualtable values field #23672

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
eliot1480 wants to merge 1 commit into
base: main
Choose a base branch
from
+1,107 −426
Open

feat: removed all instances of deprecated virtualtable values field #23672

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
142 changes: 62 additions & 80 deletions datafusion/substrait/src/logical_plan/consumer/rel/read_rel.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
use crate::logical_plan::consumer::SubstraitConsumer;
use crate::logical_plan::consumer::from_substrait_literal;
use crate::logical_plan::consumer::from_substrait_named_struct;
use crate::logical_plan::consumer::utils::ensure_schema_compatibility;
use crate::logical_plan::consumer::utils::{ensure_schema_compatibility, rename_field};
use datafusion::common::{
DFSchema, DFSchemaRef, TableReference, not_impl_err, plan_err,
substrait_datafusion_err, substrait_err,
Expand All @@ -35,7 +35,6 @@ use substrait::proto::read_rel::local_files::file_or_files::PathType::UriFile;
use substrait::proto::{Expression, ReadRel};
use url::Url;

#[expect(deprecated)]
pub async fn from_read_rel(
consumer: &impl SubstraitConsumer,
read: &ReadRel,
Expand Down Expand Up @@ -114,29 +113,17 @@ pub async fn from_read_rel(
.await
}
Some(ReadType::VirtualTable(vt)) => {
if vt.values.is_empty() && vt.expressions.is_empty() {
return Ok(LogicalPlan::EmptyRelation(EmptyRelation {
produce_one_row: false,
schema: DFSchemaRef::new(substrait_schema),
}));
}

// Check for produce_one_row pattern in both old (values) and new (expressions) formats.
// Check for produce_one_row pattern.
// A VirtualTable with exactly one row containing only empty/default fields represents
// an EmptyRelation with produce_one_row=true. This pattern is used for queries without
// a FROM clause (e.g., "SELECT 1 AS one") where a single phantom row is needed to
// provide a context for evaluating scalar expressions. This is conceptually similar to
// the SQL "DUAL" table (see: https://en.wikipedia.org/wiki/DUAL_table) which some
// databases provide as a single-row source for selecting constant expressions when no
// real table is present.
let is_produce_one_row = (vt.values.len() == 1
&& vt.expressions.is_empty()
let is_produce_one_row = vt.expressions.len() == 1
&& substrait_schema.fields().is_empty()
&& vt.values[0].fields.is_empty())
|| (vt.expressions.len() == 1
&& vt.values.is_empty()
&& substrait_schema.fields().is_empty()
&& vt.expressions[0].fields.is_empty());
&& vt.expressions[0].fields.is_empty();

if is_produce_one_row {
return Ok(LogicalPlan::EmptyRelation(EmptyRelation {
Expand All @@ -145,30 +132,65 @@ pub async fn from_read_rel(
}));
}

let values = if !vt.expressions.is_empty() {
let mut exprs = vec![];
for row in &vt.expressions {
let mut row_exprs = vec![];
for expression in &row.fields {
let expr = 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() {
return substrait_err!(
"Field count mismatch: expected {} fields but found {} in virtual table row",
substrait_schema.fields().len(),
row_exprs.len()
);
}
exprs.push(row_exprs);
if vt.expressions.is_empty() {
return Ok(LogicalPlan::EmptyRelation(EmptyRelation {
produce_one_row: false,
schema: DFSchemaRef::new(substrait_schema),
}));
}

let mut values = 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()
);
}
exprs
} else {
convert_literal_rows(consumer, vt, named_struct)?
};

let mut row_exprs = vec![];
let mut name_idx = 0;
for (field_idx, expression) in row.fields.iter().enumerate() {
let expr = match expression.rex_type.as_ref() {
Some(substrait::proto::expression::RexType::Literal(lit)) => {
if !named_struct.names.is_empty() {
name_idx += 1; // top-level names are provided through schema
}
Expr::Literal(
from_substrait_literal(
consumer,
lit,
&named_struct.names,
&mut name_idx,
)?,
None,
)
}
_ => {
rename_field(
substrait_schema.field(field_idx).as_ref(),
&named_struct.names,
field_idx,
&mut name_idx,
)?;
consumer
.consume_expression(expression, &substrait_schema)
.await?
}
};
row_exprs.push(expr);
}

if name_idx != named_struct.names.len() {
return substrait_err!(
"Names list must match exactly to nested schema, but found {} uses for {} names",
name_idx,
named_struct.names.len()
);
}
values.push(row_exprs);
}

Ok(LogicalPlan::Values(Values {
schema: DFSchemaRef::new(substrait_schema),
Expand Down Expand Up @@ -222,46 +244,6 @@ pub async fn from_read_rel(
}
}

/// Converts Substrait literal rows from a VirtualTable into DataFusion expressions.
///
/// This function processes the deprecated `values` field of VirtualTable, converting
/// each literal value into a `Expr::Literal` while tracking and validating the name
/// indices against the provided named struct schema.
fn convert_literal_rows(
consumer: &impl SubstraitConsumer,
vt: &substrait::proto::read_rel::VirtualTable,
named_struct: &substrait::proto::NamedStruct,
) -> datafusion::common::Result<Vec<Vec<Expr>>> {
#[expect(deprecated)]
vt.values
.iter()
.map(|row| {
let mut name_idx = 0;
let lits = row
.fields
.iter()
.map(|lit| {
name_idx += 1; // top-level names are provided through schema
Ok(Expr::Literal(from_substrait_literal(
consumer,
lit,
&named_struct.names,
&mut name_idx,
)?, None))
})
.collect::<datafusion::common::Result<_>>()?;
if name_idx != named_struct.names.len() {
return substrait_err!(
"Names list must match exactly to nested schema, but found {} uses for {} names",
name_idx,
named_struct.names.len()
);
}
Ok(lits)
})
.collect::<datafusion::common::Result<_>>()
}

pub fn apply_masking(
schema: DFSchema,
mask_expression: &::core::option::Option<MaskExpression>,
Expand Down
78 changes: 9 additions & 69 deletions datafusion/substrait/src/logical_plan/producer/rel/read_rel.rs
View file
Open in desktop
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![],
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
Loading
Loading