Spark-compatible CAST operation

[Blizzara]

Is your feature request related to a problem or challenge?

We're looking to use DataFusion as a replacement for Spark for some workflows, through Spark -> Substrait -> DataFusion conversions. Lot of the functionality already works, and it's been nice to see DataFusion and Spark agree on most behavior we've tested so far. However, one place where we're seeing more differences is the CAST expression, for example on casting complex types into strings (, or casting strings into numbers (where Spark is more lenient).

One option I've considered is to use Comet's cast, which I'd expect to be closer aligned with Spark (or at least on the way there). However, is there a way for me to replace/redirect the inbuilt Cast expression into using the Comet implementation?

Or would there be any other alternatives I could try?

Describe the solution you'd like

No response

Describe alternatives you've considered

No response

Additional context

No response

[alamb]

cc @viirya and @andygrove and @Omega359

You could always create a user defined function that does the cast 🤔 bit I am not sure about how much in DataFusion special cases the CAST

[Omega359]

I'm surprised cast is the thing that is catching you .. for me it was the differences in functions and their behaviour between the two systems. I wouldn't have a clue atm how to make cast in datafusion replaceable - I suspect it wouldn't be an easy task.

[Blizzara]

There's definitely difference in some functions as well, but those have been mostly minor so far (I haven't gotten to regexps or date/time stuff yet), like returning null vs throwing on overflow.

I did confirm today I can use Comet's scalar functions by overriding the existing ScalarUDFs in DF, that's quite nice!

For Expr::Cast that doesn't work as it's part of the Expr enum rather than being a ScalarUDF - is that by choice, or just pending migration? But I plan to try if I can use FunctionRewrite to change the Expr::Cast into a ScalarUDF that uses the Comet cast.

[alamb]

For Expr::Cast that doesn't work as it's part of the Expr enum rather than being a ScalarUDF - is that by choice, or just pending migration? But I plan to try if I can use FunctionRewrite to change the Expr::Cast into a ScalarUDF that uses the Comet cast.

I think it is largely historical, but realistically there is enough code that handles Cast specially in the core that switching it to use a UDF is likely a susbtantial project (we could certainly do it, but I wanted to offer my opinion on the scope)

[alamb]

Now that we have the datafusion-spark crate, maybe we have a home for this functionality:

• [EPIC] Complete datafusion-spark Spark Compatible Functions #15914

[logan-keede]

do we just need to port cast here from comet?

[alamb]

do we just need to port cast here from comet?

I think that is probably a good place to start (it would also be good to look at what Sail does)

Then we probably need to think about "what happens if someone installs the spark compatible functions library" -- if they want all casts to have the spark semantics, we probably need to figure out how to update / change the Expr::Cast and Expr::TryCast to know to call the spark version 🤔

[logan-keede]

The comet implementation already has a PhysicalExpr for cast. I was thinking if we could make it datafusion compatible(perhaps it already is) and while making physical expression we replace CastExpr with SparkCastExpr when some spark-compatibility parameter is true. WDYT?

[alamb]

The key will be to find some way to control what PhysicalExpr is used during the physical planner

There might already be an API to do this (provide a custom planner for Expr --> PhysicalExpr) but I don't know off the top of my head

[Blizzara]

Just fwiw, we already use Comet's Cast in our use of DF, through a rewrite rule:

impl FunctionRewrite for OurCastRewrite {
    fn name(&self) -> &str {
        "use_comet_cast"
    }

    fn rewrite(
        &self,
        expr: Expr,
        _schema: &DFSchema,
        _config: &ConfigOptions,
    ) -> datafusion::common::Result<Transformed<Expr>> {
        match expr {
            Expr::Cast(cast) => {
                let new_expr = ScalarUDF::from(OurCometSparkCastUdf::new(&cast.data_type))
                    .call(vec![*cast.expr]);
                Ok(Transformed::yes(new_expr))
            }
            _ => Ok(Transformed::no(expr)),
        }
    }
}

This has seemed to work so far, but likely isn't ideal solution. We also have to run the analyser twice, since the first run may create more casts that then still need to be rewritten..