push_down_filter and common_sub_expression_eliminate fight between them

Describe the bug

With the following table structure:

CREATE OR REPLACE TABLE t1 (
	date DATE,
	timestamp TIMESTAMP_S,
  ids STRUCT(
    id1 VARCHAR,
    extra INT4
  ),
  struct STRUCT(
    var1 VARCHAR,
    extra VARCHAR
  )
);

and the following query

        WITH events AS (
            SELECT
                ids.id1 as device,
                struct.var1 as user,
                timestamp
            FROM t1
            WHERE
                date='2025-01-03'
        )
        SELECT
            *,
            LAG(user, 1) OVER (PARTITION BY device ORDER BY timestamp) AS prev
        FROM events
        WHERE
            device IS NOT NULL AND device != ''
            AND user IS NOT NULL AND user != ''
        LIMIT 100

I get into a situation where the 2 optimization steps given above fight between them:

First, push_down_filter pushes the filter down

    Projection: events.device, events.user, events.timestamp, lag(events.user,Int64(1)) PARTITION BY [events.device] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS prev
      Limit: skip=0, fetch=100
        WindowAggr: windowExpr=[[lag(events.user, Int64(1)) PARTITION BY [events.device] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
          SubqueryAlias: events
            Projection: get_field(t1.ids, Utf8("id1")) AS device, get_field(t1.struct, Utf8("var1")) AS user, t1.timestamp
              Filter: get_field(t1.ids, Utf8("id1")) IS NOT NULL AND get_field(t1.ids, Utf8("id1")) != Utf8("") AND get_field(t1.struct, Utf8("var1")) IS NOT NULL AND get_field(t1.struct, Utf8("var1")) != Utf8("") AND t1._ACP_DATE = Date32("2025-01-03")
                TableScan: t1, partial_filters=[get_field(t1.ids, Utf8("id1")) IS NOT NULL, get_field(t1.ids, Utf8("id1")) != Utf8(""), get_field(t1.struct, Utf8("var1")) IS NOT NULL, get_field(t1.struct, Utf8("var1")) != Utf8(""), t1._ACP_DATE = Date32("2025-01-03")]

Then the common_sub_expression_eliminate acts on the Filter, and of course, adds a Projection below it to contain the aliases:

    Projection: events.DeviceId, events.UserId, events.timestamp, lag(events.UserId,Int64(1)) PARTITION BY [events.DeviceId] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS PreviousUserColName
      Limit: skip=0, fetch=100
        WindowAggr: windowExpr=[[lag(events.UserId, Int64(1)) PARTITION BY [events.DeviceId] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
          SubqueryAlias: events
            Projection: get_field(t1.endUserIDs, Utf8("id1")) AS DeviceId, get_field(t1.struct, Utf8("var1")) AS UserId, t1.timestamp
              Filter: __common_expr_1 IS NOT NULL AND __common_expr_1 != Utf8("") AND get_field(t1.struct, Utf8("var1")) IS NOT NULL AND get_field(t1.struct, Utf8("var1")) != Utf8("") AND t1._ACP_DATE = Date32("2025-01-03")
                Projection: get_field(t1.endUserIDs, Utf8("id1")) AS __common_expr_1, t1._ACP_DATE, t1.timestamp, t1.endUserIDs, t1.struct
                  TableScan: t1 projection=[_ACP_DATE, timestamp, endUserIDs, struct], partial_filters=[get_field(t1.endUserIDs, Utf8("id1")) IS NOT NULL, get_field(t1.endUserIDs, Utf8("id1")) != Utf8(""), get_field(t1.struct, Utf8("var1")) IS NOT NULL, get_field(t1.struct, Utf8("var1")) != Utf8(""), t1._ACP_DATE = Date32("2025-01-03")]

At the next run, push_down_filter sees a Filter on top a Projection and pushes it down, deleting the alias probably because Filters can't have aliases

    Projection: events.DeviceId, events.UserId, events.timestamp, lag(events.UserId,Int64(1)) PARTITION BY [events.DeviceId] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS PreviousUserColName
      Limit: skip=0, fetch=100
        WindowAggr: windowExpr=[[lag(events.UserId, Int64(1)) PARTITION BY [events.DeviceId] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
          SubqueryAlias: events
            Projection: get_field(t1.endUserIDs, Utf8("id1")) AS DeviceId, get_field(t1.struct, Utf8("var1")) AS UserId, t1.timestamp
              Projection: get_field(t1.endUserIDs, Utf8("id1")) AS __common_expr_1, t1._ACP_DATE, t1.timestamp, t1.endUserIDs, t1.struct
                Filter: get_field(t1.endUserIDs, Utf8("id1")) IS NOT NULL AND get_field(t1.endUserIDs, Utf8("id1")) != Utf8("") AND get_field(t1.struct, Utf8("var1")) IS NOT NULL AND get_field(t1.struct, Utf8("var1")) != Utf8("") AND t1._ACP_DATE = Date32("2025-01-03")
                  TableScan: t1 projection=[_ACP_DATE, timestamp, endUserIDs, struct], partial_filters=[get_field(t1.endUserIDs, Utf8("id1")) IS NOT NULL, get_field(t1.endUserIDs, Utf8("id1")) != Utf8(""), get_field(t1.struct, Utf8("var1")) IS NOT NULL, get_field(t1.struct, Utf8("var1")) != Utf8(""), t1._ACP_DATE = Date32("2025-01-03")]

And no progress is made.

This seems like it could happen in other cases. Maybe we could have something like hints, like a flag on the logical plan that can say that the plan was added during a previous optimization step ? The problem seems complicated though.

To Reproduce

No response

Expected behavior

No response

Additional context

No response

@adragomir

Hi @adragomir , thanks for the report, can you provide the reproduce steps with full datafusion sql? So we can look into it more quickly!

reproduction

CREATE OR REPLACE TABLE t1 (
        date DATE,
        timestamp TIMESTAMP,
  ids STRUCT<
    id1 VARCHAR,
    extra INT4
  >,
  structs STRUCT<
    var1 VARCHAR,
    extra VARCHAR
  >
);

explain verbose WITH events AS (
    SELECT
        ids.id1 as device,
        structs.var1 as user,
        timestamp
    FROM t1
    WHERE
        date='2025-01-03'
)
SELECT
    *,
    LAG(user, 1) OVER (PARTITION BY device ORDER BY timestamp) AS prev
FROM events
WHERE
    device IS NOT NULL AND device != ''
    AND user IS NOT NULL AND user != ''
LIMIT 100;

looks like right now its push_down_filter and push_down_leaf_projections fighting with each other

| logical_plan after push_down_filter                        | Projection: events.device, events.user, events.timestamp, lag(events.user,Int64(1)) PARTITION BY [events.device] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS prev |
|                                                            |   Limit: skip=0, fetch=100                                                                                                                                                                                            |
|                                                            |     WindowAggr: windowExpr=[[lag(events.user, Int64(1)) PARTITION BY [events.device] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]                                   |
|                                                            |       SubqueryAlias: events                                                                                                                                                                                           |
|                                                            |         Filter: device IS NOT NULL AND device != Utf8View("") AND user IS NOT NULL AND user != Utf8View("")                                                                                                           |
|                                                            |           Projection: get_field(t1.ids, Utf8("id1")) AS device, get_field(t1.structs, Utf8("var1")) AS user, t1.timestamp                                                                                             |
|                                                            |             Filter: t1.date = Date32("2025-01-03")                                                                                                                                                                    |
|                                                            |               TableScan: t1                                                                                                                                                                                           |
| logical_plan after single_distinct_aggregation_to_group_by | SAME TEXT AS ABOVE                                                                                                                                                                                                    |
...
| logical_plan after extract_leaf_expressions                | SAME TEXT AS ABOVE                                                                                                                                                                                                    |
| logical_plan after push_down_leaf_projections              | Projection: events.device, events.user, events.timestamp, lag(events.user,Int64(1)) PARTITION BY [events.device] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS prev |
|                                                            |   Limit: skip=0, fetch=100                                                                                                                                                                                            |
|                                                            |     WindowAggr: windowExpr=[[lag(events.user, Int64(1)) PARTITION BY [events.device] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]                                   |
|                                                            |       SubqueryAlias: events                                                                                                                                                                                           |
|                                                            |         Filter: device IS NOT NULL AND device != Utf8View("") AND user IS NOT NULL AND user != Utf8View("")                                                                                                           |
|                                                            |           Projection: __datafusion_extracted_1 AS device, __datafusion_extracted_2 AS user, t1.timestamp                                                                                                              |
|                                                            |             Filter: t1.date = Date32("2025-01-03")                                                                                                                                                                    |
|                                                            |               Projection: get_field(t1.ids, Utf8("id1")) AS __datafusion_extracted_1, get_field(t1.structs, Utf8("var1")) AS __datafusion_extracted_2, t1.date, t1.timestamp, t1.ids, t1.structs                      |
|                                                            |                 TableScan: t1                                                                                                                                                                                         |
| logical_plan after optimize_projections                    | Projection: events.device, events.user, events.timestamp, lag(events.user,Int64(1)) PARTITION BY [events.device] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS prev |
|                                                            |   Limit: skip=0, fetch=100                                                                                                                                                                                            |
|                                                            |     WindowAggr: windowExpr=[[lag(events.user, Int64(1)) PARTITION BY [events.device] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]                                   |
|                                                            |       SubqueryAlias: events                                                                                                                                                                                           |
|                                                            |         Filter: device IS NOT NULL AND device != Utf8View("") AND user IS NOT NULL AND user != Utf8View("")                                                                                                           |
|                                                            |           Projection: __datafusion_extracted_1 AS device, __datafusion_extracted_2 AS user, t1.timestamp                                                                                                              |
|                                                            |             Filter: t1.date = Date32("2025-01-03")                                                                                                                                                                    |
|                                                            |               Projection: get_field(t1.ids, Utf8("id1")) AS __datafusion_extracted_1, get_field(t1.structs, Utf8("var1")) AS __datafusion_extracted_2, t1.date, t1.timestamp                                          |
|                                                            |                 TableScan: t1 projection=[date, timestamp, ids, structs]                                                                                                                                              |
| logical_plan after rewrite_set_comparison                  | SAME TEXT AS ABOVE                                                                                                                                                                                                    |
...
| logical_plan after push_down_limit                         | SAME TEXT AS ABOVE                                                                                                                                                                                                    |
| logical_plan after push_down_filter                        | Projection: events.device, events.user, events.timestamp, lag(events.user,Int64(1)) PARTITION BY [events.device] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS prev |
|                                                            |   Limit: skip=0, fetch=100                                                                                                                                                                                            |
|                                                            |     WindowAggr: windowExpr=[[lag(events.user, Int64(1)) PARTITION BY [events.device] ORDER BY [events.timestamp ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]                                   |
|                                                            |       SubqueryAlias: events                                                                                                                                                                                           |
|                                                            |         Projection: __datafusion_extracted_1 AS device, __datafusion_extracted_2 AS user, t1.timestamp                                                                                                                |
|                                                            |           Filter: __datafusion_extracted_1 IS NOT NULL AND __datafusion_extracted_1 != Utf8View("") AND __datafusion_extracted_2 IS NOT NULL AND __datafusion_extracted_2 != Utf8View("")                             |
|                                                            |             Projection: get_field(t1.ids, Utf8("id1")) AS __datafusion_extracted_1, get_field(t1.structs, Utf8("var1")) AS __datafusion_extracted_2, t1.date, t1.timestamp                                            |
|                                                            |               Filter: t1.date = Date32("2025-01-03")                                                                                                                                                                  |
|                                                            |                 TableScan: t1 projection=[date, timestamp, ids, structs]                                                                                                                                              |

@Jefffrey

Ran @Jefffrey's reproduction on released 55.1.0 and traced what the fight actually costs now, since the picture has changed since the original report.

The good news is that the plan no longer fails to converge. The optimizer loop hashes a LogicalPlanSignature after every pass and breaks when it sees one it has already recorded, so this query stops after three passes with a stable, correct plan — the end of pass 2 and the end of pass 3 are byte-identical.

What remains is a cycle inside a pass. In every pass after the first, exactly three rules keep reporting a change and between them return the plan to where it started:

Projection: __datafusion_extracted_1 AS device, ...
  Filter:  __datafusion_extracted_1 IS NOT NULL AND ...
    Projection: __datafusion_extracted_1, __datafusion_extracted_2, t1.timestamp
      Filter: t1.date = Date32("2025-01-03")            <-- (A)
        Projection: get_field(t1.ids, 'id1') AS __datafusion_extracted_1, ...   <-- (B)
          TableScan: t1

push_down_filter swaps (A) below (B), so the date filter runs before get_field is evaluated.
push_down_leaf_projections swaps (B) back below (A), so the extraction runs for every row again.
optimize_projections cleans up the projection the swap left behind, landing back on the plan above.

So the two rules want the opposite relative order of the same adjacent pair, and the one that runs later in the rule list wins. The net cost is one wasted full optimizer pass for any query of this shape: without the fight the signature check would fire one pass earlier.

Which of the two should yield is a plan-quality call I do not think I should make unilaterally. For this query push_down_filter's order looks better — the date predicate does not reference the extracted columns, so filtering first means get_field is evaluated on fewer rows — but push_down_leaf_projections exists precisely to get leaf expressions close enough to the scan to be pushed into it, so preferring the filter unconditionally may cost more than it saves elsewhere.

Is there a preferred direction here? Two that seem plausible: teach push_down_filter not to push a filter below a leaf-extraction projection it will lose to anyway, or teach push_down_leaf_projections to stop at a filter that does not reference the leaves it is extracting. Happy to implement whichever you would rather have.

test: cover the filter and extraction projection precedence

fix: decide the precedence between PushDownFilter and PushDownLeafPro…

fix: `PushDownLeafProjections` yields when the extraction cannot move (…

Navigation Menu

Repository navigation

push_down_filter and common_sub_expression_eliminate fight between them #14540

Describe the bug

To Reproduce

Expected behavior

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

push_down_filter and common_sub_expression_eliminate fight between them #14540

Description

Describe the bug

To Reproduce

Expected behavior

Additional context

Activity

zhuqi-lucas commented on Feb 7, 2025

Jefffrey commented on Jul 10, 2026

kakiuwang-ui commented on Sep 14, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions