Error "entered unreachable code: NamedStructField should be rewritten in OperatorToFunction" after upgrade to 37

[alamb]

NOTE -- Here is an example of how to make Expr::NamedStructField work in 37.1.0: #10183

Is your feature request related to a problem or challenge?

In 37.0.0 many of the built in functions have been migrated to UDFs as described on #8045 . The migration is completed in 38.0.0

One part of this change is that now certain Exprs must be rewritten into the appropriate functions. Most notably get_field that extracts a field from a Struct

Among other things this allows people to customize how Expr behaves: #7845 (comment) or in slack to return NULLs for rows that don't pass in maps

The rewrite happens automatically as part of the logical planner (in the Analyzer pass)

However if you bypass those passes it will not happen

Yeah you need to use the FunctionRewriter here (with the relevant rewriter registered) https://github.com/apache/arrow-datafusion/blob/0573f78c7e7a4d94c3204cee464b3860479e0afb/datafusion/optimizer/src/analyzer/function_rewrite.rs#L33

Example

An example from discord: link is:

  let schema = Schema::new(vec![
        Field::new("id", DataType::Utf8, true),
        Field::new(
            "props",
            DataType::Struct(Fields::from(vec![Field::new("a", DataType::Utf8, true)])),
            true,
        ),
    ]);

    println!("schema {:?}", schema);

    let df_schema = DFSchema::try_from(schema.clone()).unwrap();

    let plan = table_scan(Some("props_test"), &schema, None)?
        .filter(col("props").field("a").eq(lit("2021-02-02")))?
        .build()?;
    println!("logical plan {:?}", plan);
    let phys = DefaultPhysicalPlanner::default().create_physical_expr(&plan.expressions()[0], &df_schema, &SessionContext::new().state())?;
    println!("phys {:?}", phys);
    Ok(())

This returns an error "NamedStructField should be rewritten in OperatorToFunction"

Describe the solution you'd like

No response

Describe alternatives you've considered

One potential workaround is to call get_field directly rather than Expr::field

So instead of

    let plan = table_scan(Some("props_test"), &schema, None)?
        .filter(col("props").field("a").eq(lit("2021-02-02")))?
        .build()?;

call like

  let plan = table_scan(Some("props_test"), &schema, None)?
        .filter(get_field(col("props", "a")).eq(lit("2021-02-02")))?
        .build()?;

Additional context

@ion-elgreco is seeing the same issue in Delta-rs: #9904 (comment)

I tried it with 37.1.0 in delta-rs, but we still get this error: internal error: entered unreachable code: NamedStructField should be rewritten in OperatorToFunction, wasn't this regression fixed?

@westonpace brings it up in discord link

Another report in discord: link

[alamb]

Example from @ion-elgreco

@alamb this is the code:

  let (table, _metrics) = DeltaOps(table)
            .delete()
            .with_predicate("props['a'] = '2021-02-02'")
            .await
            .unwrap();

Which comes from here: https://github.com/delta-io/delta-rs/blob/main/crates%2Fcore%2Fsrc%2Foperations%2Fdelete.rs#L770-L774

[alamb]

I see a few options:

1. We can put get_field back into the core crate (and move it out of datafusion-functions) - might be the least surprising but would not allow people to customize how field access worked (e.g. for implementing JSON support)
2. We can make a better API / more examples how to get the core rewrites working in

   datafusion/datafusion-examples/examples/expr_api.rs

   Lines 84 to 110 in 0780438

   	fn evaluate_demo() -> Result<()> {
   	// For example, let's say you have some integers in an array
   	let batch = RecordBatch::try_from_iter([(
   	"a",
   	Arc::new(Int32Array::from(vec![4, 5, 6, 7, 8, 7, 4])) as _,
   	)])?;
   	// If you want to find all rows where the expression `a < 5 OR a = 8` is true
   	let expr = col("a").lt(lit(5)).or(col("a").eq(lit(8)));
   	// First, you make a "physical expression" from the logical `Expr`
   	let physical_expr = physical_expr(&batch.schema(), expr)?;
   	// Now, you can evaluate the expression against the RecordBatch
   	let result = physical_expr.evaluate(&batch)?;
   	// The result contain an array that is true only for where `a < 5 OR a = 8`
   	let expected_result = Arc::new(BooleanArray::from(vec![
   	true, false, false, false, true, false, true,
   	])) as _;
   	assert!(
   	matches!(&result, ColumnarValue::Array(r) if r == &expected_result),
   	"result: {:?}",
   	result
   	);
   	Ok(())

I would be happy to work on 2 if someone could point me at how people are creating Physical exprs today

[westonpace]

I think I'd be happy with 2. The example you linked is how we are using datafusion. Here is an updated example that fails with the error:

    // For example, let's say you have some integers in an array
    let b = Arc::new(Int32Array::from(vec![4, 5, 6, 7, 8, 7, 4]));
    let a = Arc::new(StructArray::new(
        vec![Field::new("b", DataType::Int32, false)].into(),
        vec![b],
        None,
    ));
    let batch = RecordBatch::try_from_iter([("a", a as _)])?;

    // If you want to find all rows where the expression `a < 5 OR a = 8` is true
    let expr = col("a")
        .field("b")
        .lt(lit(5))
        .or(col("a").field("b").eq(lit(8)));

    // First, you make a "physical expression" from the logical `Expr`
    let physical_expr = physical_expr(&batch.schema(), expr)?;

[ion-elgreco]

@alamb this is how we create expressions:

/// Parse a string predicate into an `Expr`
pub(crate) fn parse_predicate_expression(
    schema: &DFSchema,
    expr: impl AsRef<str>,
    df_state: &SessionState,
) -> DeltaResult<Expr> {
    let dialect = &GenericDialect {};
    let mut tokenizer = Tokenizer::new(dialect, expr.as_ref());
    let tokens = tokenizer
        .tokenize()
        .map_err(|err| DeltaTableError::GenericError {
            source: Box::new(err),
        })?;
    let sql = Parser::new(dialect)
        .with_tokens(tokens)
        .parse_expr()
        .map_err(|err| DeltaTableError::GenericError {
            source: Box::new(err),
        })?;

    let context_provider = DeltaContextProvider { state: df_state };
    let sql_to_rel =
        SqlToRel::new_with_options(&context_provider, DeltaParserOptions::default().into());

    Ok(sql_to_rel.sql_to_expr(sql, schema, &mut Default::default())?)
}

[alamb]

I'll work on creating an example shortly

[westonpace]

This also leads to a sort of "variant problem" for any code we write that handles Expr. For example, we have code that walks through an Expr and calculates which (potentially nested) columns are referenced. The "Expr walking" code now has to be aware of both the GetStructField and the ScalarUDF variants of field access.

It would be nicer I think if there was a single canonical way to represent a nested field access in Expr. For example, maybe the translation from GetStructField to ScalarUDF happens as part of the translation from Expr to PhysicalExpr? This way Expr only has GetStructField but there is still the flexibility to customize field access?

[alamb]

Here is an example of how to make Expr::struct work in 37.1.0: #10183

I think we need a better API to do this for real (in 38.0.0 and going forward). I will think about this -- maybe @jayzhan211 has some thoughts

[alamb]

The "Expr walking" code now has to be aware of both the GetStructField and the ScalarUDF variants of field access.

I think this can be controlled by the consumer -- for example if you are walking Exprs in lancedb, you can control when you transform Expr::GetStructField into ScalarUDF and depending on where you do your analysis you only have to check for one

It would be nicer I think if there was a single canonical way to represent a nested field access in Expr.

I think the idea is people might want to override Expr::GetStructFields semantics and they way they would do so is to rewrite it into a different function. I think this is especially compelling for supporting JSON/JSONB for example

[westonpace]

I think this can be controlled by the consumer -- for example if you are walking Exprs in lancedb, you can control when you transform Expr::GetStructField into ScalarUDF and depending on where you do your analysis you only have to check for one

I agree (and thanks for the example). This works for us. All of our expr from the user start as &str. However, if we were receiving Expr directly then it wouldn't work because we wouldn't know which approach the user used. This is not a problem for us, we aren't expecting the user to provide direct DF structs in any of our roadmap features, I'm just thinking through possibilities.

I think the idea is people might want to override Expr::GetStructFields semantics and they way they would do so is to rewrite it into a different function. I think this is especially compelling for supporting JSON/JSONB for example

What's the motivation for doing this at the logical level instead of doing this as part of the conversion from logical to physical?

[alamb]

What's the motivation for doing this at the logical level instead of doing this as part of the conversion from logical to physical?

I don't think there is any particular motivation (or any reason that the conversion needs to be done at either spot) 🤔