[BUG] DataFusion regression in optimizer related to casting

[andygrove]

What happened:

After upgrading to DataFusion 13.0.0-rc1 in #825 we see a regression in the query plan optimization for the filter expression foo between '1998-03-18' and (cast('1998-03-18' as date) + INTERVAL '90 days'), where foo is a string column.

Before

Filter: #foo BETWEEN Utf8("2001-03-09") AND Utf8("2001-06-07")

After

Filter: foo BETWEEN Utf8("2001-03-09") AND CAST(CAST(Utf8("2001-03-09") AS Date32) + IntervalDayTime("386547056640") AS Utf8)

What you expected to happen:

I preferred the original behavior.

Minimal Complete Verifiable Example:

TBD - I am working on a repro for this

# Put your MCVE code here

Anything else we need to know?:

Environment:

• dask-sql version:
• Python version:
• Operating System:
• Install method (conda, pip, source):

[andygrove]

I cannot reproduce this in DataFusion 13.0.0 RC1:

DataFusion CLI v13.0.0
❯ create external table test (a string) stored as csv location '/tmp/dates.csv';
0 rows in set. Query took 0.002 seconds.

❯ explain verbose select * from test where a between '1998-03-18' and
                             (cast('1998-03-18' as date) + INTERVAL '90 days')
;
+------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type                                                  | plan                                                                                                                                                                  |
+------------------------------------------------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| initial_logical_plan                                       | Projection: test.a                                                                                                                                                    |
|                                                            |   Filter: test.a BETWEEN Utf8("1998-03-18") AND CAST(Utf8("1998-03-18") AS Date32) + IntervalDayTime("386547056640")                                                  |
|                                                            |     TableScan: test                                                                                                                                                   |
| logical_plan after type_coercion                           | Projection: test.a                                                                                                                                                    |
|                                                            |   Filter: test.a BETWEEN Utf8("1998-03-18") AND CAST(CAST(Utf8("1998-03-18") AS Date32) + IntervalDayTime("386547056640") AS Utf8)                                    |
|                                                            |     TableScan: test                                                                                                                                                   |
| logical_plan after simplify_expressions                    | Projection: test.a                                                                                                                                                    |
|                                                            |   Filter: test.a >= Utf8("1998-03-18") AND test.a <= Utf8("1998-06-16") AS test.a BETWEEN Utf8("1998-03-18") AND Utf8("1998-03-18") + IntervalDayTime("386547056640") |
|                                                            |     TableScan: test  
| logical_plan after unwrap_cast_in_comparison               | SAME TEXT AS ABOVE                                                                                                                                                    |
| logical_plan after decorrelate_where_exists                | SAME TEXT AS ABOVE                                                                                                                                                    |
| logical_plan after decorrelate_where_in                    | SAME TEXT AS ABOVE                                                                                                                                                    |
| logical_plan after scalar_subquery_to_join                 | SAME TEXT AS ABOVE                                                                                                                                                    |
| logical_plan after subquery_filter_to_join                 | Projection: test.a                                                                                                                                                    |
|                                                            |   Filter: test.a >= Utf8("1998-03-18") AND test.a <= Utf8("1998-06-16")                                                                                               |
|                                                            |     TableScan: test   

[charlesbluca]

I'm not able to reproduce this with a similar repro in dask-sql:

import pandas as pd
from dask_sql import Context

c = Context()

print(
    c.explain(
        """select
                *
            from
                df
            where
                a between '2002-05-08' and (cast('2002-05-08' as date) + interval '5 days')
        """,
        dataframes={
            "df": pd.DataFrame({
                "a": pd.date_range("2002-05-02", periods=20, freq="D").astype(str)
            })
        }
    )
)

Projection: df.a
  Filter: df.a >= Utf8("2002-05-08") AND df.a <= Utf8("2002-05-13")
    TableScan: df projection=[a]

But I do see the regression when attempting this filter in a subquery:

import pandas as pd
from dask_sql import Context

c = Context()

print(
    c.explain(
        """select
                *
            from
                df
            where
                b > (
                    select
                        avg(b)
                    from
                        df
                    where
                        a between '2002-05-08' and (cast('2002-05-08' as date) + interval '5 days')
                )
        """,
        dataframes={
            "df": pd.DataFrame({
                "a": pd.date_range("2002-05-02", periods=20, freq="D").astype(str),
                "b": range(20)
            })
        }
    )
)

Projection: df.a, df.b
  Filter: CAST(df.b AS Float64) > __sq_1.__value
    CrossJoin:
      TableScan: df projection=[a, b]
      Projection: AVG(df.b) AS __value, alias=__sq_1
        Aggregate: groupBy=[[]], aggr=[[AVG(df.b)]]
          Filter: df.a BETWEEN Utf8("2002-05-08") AND CAST(CAST(Utf8("2002-05-08") AS Date32) + IntervalDayTime("21474836480") AS Utf8)
            TableScan: df projection=[a, b]

[andygrove]

Here is the PR to fix this in DataFusion: apache/datafusion#3764

We can fix directly in Dask SQL though by just running the simplify_expression rule again after converting subqueries to join. I will file a PR shortly.

[andygrove]

I applied the fix directly in #825 and added an integration test