Substrait: Support expression serialization

[westonpace]

Is your feature request related to a problem or challenge?

The goal is to allow expressions (not plans and, to start with, scalar expressions) to be passed between different libraries (e.g. pyarrow and datafusion). As a concrete use case we have users in lance who pass in pyarrow.compute filter expressions and we want to use datafusion to satisfy the filter. Currently we do this by going from pyarrow expression to sql-like string and then use the SQL parser but this doesn't work in all cases and substrait is a more natural fit.

Describe the solution you'd like

About 8 months ago Substrait added support for a new top-level message "ExtendedExpression". This consists of a collection of named expressions and an input schema. Pyarrow is adding support for this in the next release. I've prototyped a solution using datafusion that looks something like this:

import pyarrow as pa
import pyarrow.compute as pc
import datafusion.substrait

schema = pa.schema([pa.field("x", pa.int32())])
expr = pc.field("x") + 3
expr_bytes = expr.to_substrait(schema)
expr_sub = datafusion.substrait.substrait.serde.deserialize_expr_bytes(expr_bytes)
expr_df = datafusion.substrait.substrait.consumer.from_substrait_expr(expr_sub)

print("Pyarrow")
print(expr)
print("Datafusion")
print(expr_df)

expr_bytes = datafusion.substrait.substrait.serde.serialize_dfexpr_bytes(expr_df, schema)
expr = pc.Expression.from_substrait(expr_bytes)

print("Pyarrow (return)")
print(expr)

Which outputs:

Pyarrow
add_checked(x, 3)
Datafusion
Expr(x + Int32(3))
Pyarrow (return)
add_checked(x, 3)

Describe alternatives you've considered

Within datafusion there is also pyarrow_filter_expression which has a similar goal but requires a pyarrow dependency and is limited to pyarrow. Eventually I'd like for there to be support with other languages and I'd also eventually like to support interop with more libraries that have expressions (e.g. polars, ibis)

Additional context

I plan on implementing this soon. I have a working prototype but it needs cleaned up and a lot of additional testing. If anyone would like to provide some early feedback I can share:

arrow-datafusion changes: https:/apache/arrow-datafusion/compare/main...westonpace:arrow-datafusion:experiment/substrait-extended-expression?expand=1

arrow-datafusion-python changes: https:/apache/arrow-datafusion-python/compare/main...westonpace:arrow-datafusion-python:experiment/substrait-extended-expr?expand=1