Do not plan unnecessary Sort

[martint]

Avoid planning unnecessary sort operations. This fixes #759 and is an alternative to #810.

[martint]

Examples:

ORDER BY at top-level

EXPLAIN
WITH t(x) AS (VALUES 1,2,3) 
SELECT * FROM t 
ORDER BY x

 - Output[x]
     - RemoteMerge[field ASC_NULLS_LAST]
         - LocalMerge[field ASC_NULLS_LAST]
             - PartialSort[field ASC_NULLS_LAST]
                 - RemoteExchange[REPARTITION]
                     - Values

Unnecessary ORDER BY in subquery query

EXPLAIN 
WITH t(x) AS (VALUES 1,2,3) 
SELECT * FROM (
    SELECT * FROM t 
    ORDER BY x
)

- Output[x]
     - Values

Necessary ORDER BY in subquery

EXPLAIN 
WITH t(x) AS (VALUES 1,2,3) 
SELECT * FROM (
    SELECT * FROM t 
    ORDER BY x 
    LIMIT 1
)

- Output[x]
     - TopN[1 by (field ASC_NULLS_LAST)]
         - LocalExchange[SINGLE] ()
             - TopNPartial[1 by (field ASC_NULLS_LAST)]
                 - LocalExchange[ROUND_ROBIN] ()
                     - Values

Unnecessary ORDER BY in scalar subquery

EXPLAIN 
WITH t(x) AS (VALUES 1,2,3) 
SELECT (SELECT x FROM t ORDER BY x) 
FROM (VALUES 10)

 - Output[_col0]
     - EnforceSingleRow
         - Values

Necessary ORDER BY in scalar subquery

EXPLAIN 
WITH t(x) AS (VALUES 1,2,3) 
SELECT (SELECT x FROM t ORDER BY x LIMIT 1) 
FROM (VALUES 10)

 - Output[_col0]
     - EnforceSingleRow
         - TopN[1 by (field_1 ASC_NULLS_LAST)]
             - LocalExchange[SINGLE] ()
                 - TopNPartial[1 by (field_1 ASC_NULLS_LAST)]
                     - LocalExchange[ROUND_ROBIN] ()
                         - Values

Unnecessary ORDER BY in aggregation query

EXPLAIN 
WITH t(x) AS (VALUES 1,2,3) 
SELECT array_agg(x) 
FROM (
    SELECT * FROM t 
    ORDER BY x
)

 - Output[_col0]
     - Aggregate
         - Values

Necessary ORDER BY in UNION ALL query

EXPLAIN
VALUES 1 
UNION ALL 
VALUES 2 
ORDER BY 1

 - Output[_col0]
     - RemoteMerge[field_1 ASC_NULLS_LAST]
         - LocalMerge[field_1 ASC_NULLS_LAST]
             - PartialSort[field_1 ASC_NULLS_LAST]
                 - RemoteExchange[REPARTITION]
                     - LocalExchange[ROUND_ROBIN] ()
                         - Values
                         - Values

Unnecessary ORDER BY in UNION ALL query

EXPLAIN
SELECT * FROM (
    VALUES 1 
    UNION ALL 
    VALUES 2 
    ORDER BY 1
)

 - Output[_col0]
     - LocalExchange[ROUND_ROBIN] ()
         - Values
         - Values

[martint]

With warnings:

[findepi]

as above

[findepi]

You added getter/setter last, but this isn't the last field. This class is messy.

[findepi]

Let's think again about example that you also mentioned in your blog:

SELECT *, row_number() OVER ()
FROM (SELECT * FROM nation ORDER BY name DESC)

the proper way of writing this is:

SELECT *, row_number() OVER (ORDER BY name DESC)
FROM nation

Is the proper way a drop-in replacement?
Before the change, the old way would use distributed sort.
Does the standard-compliant version use distributed sort too?

[findepi]

If the proper way is not equally performant yet, we shouldn't have the feature on by default just yet.

[martint]

I believe we added distributed sort specifically for the unpartitioned window function case, so I think that should be the case. I’ll double check

[martint]

You added getter/setter last, but this isn't the last field.

I added it right below "work processor pipelines" in both cases. The other ones are out of order.

This class is messy.

100% agree

[martint]

Is the proper way a drop-in replacement?
Before the change, the old way would use distributed sort.

Regardless, a query like this is just broken, semantically speaking. Almost every case I've seen is people relying on ORDER BY for aggregation queries, since 1) we didn't use to support ordered aggregates until sometime last year, 2) window functions have always supported ORDER BY.

For anyone relying on that behavior, there's the fallback of disabling the config flag (for now).

[martint]

It turns out we currently don't run distributed sorts for unpartitioned window functions with an ORDER BY clause. Here's a PR to add it: #821

[sopel39]

I believe we added distributed sort specifically for the unpartitioned window function case, so I think that should be the case. I’ll double check

Distributed sort utilizes entire cluster memory, while WindowOperator does not (it collects partition rows).

Keep in mind that unordered row_number window function was replaced with RowNumberNode operator (See: io.prestosql.sql.planner.optimizations.WindowFilterPushDown.Rewriter#canReplaceWithRowNumber). This means that such hacky:

SELECT *, row_number() OVER ()
FROM (SELECT * FROM nation ORDER BY name DESC)

query utilizes distributed sort and skips WindowOperator completely. However, for:

SELECT *, row_number() OVER (ORDER BY name DESC)
FROM nation

It will still execute WindowOperator which will collect all partition data (thus mitigating benefits of distributed sort).

Maybe we should improve WindowFilterPushDown rule too

[findepi]

@sopel39 good point. So in #821 when insert Sort below WindowNode, we could convert the WindowNode to RowNumberNode (in row_number() case of course). (I think we should move continue this discussion under #821 though).

[martint]

BTW, for this shape:

SELECT *, row_number() OVER ()
FROM (SELECT * FROM nation ORDER BY name DESC)

even if the ORDER BY weren't removed, there's no guarantee that the row_number() function (or any other window function, for that matter) will not load all the data in memory and shuffle it in some way before producing its results. So it could break at any time for other reasons unrelated to this optimization, so I'm not too concerned about it, especially since we have an escape hatch.

[martint]

Updated.

[findepi]

LGTM except for "Clean up analysis of ORDER BY" commit which I didn't review.