A self-join on a rank offset is a neighbour lookup: answering it with lagInFrame/leadInFrame is 66% faster on TPC-DS query 47 and query 57

### Describe the situation

TPC-DS query_47 and query_57 read a ranked CTE three times and join it to itself twice, only to reach
the previous and the next row of each ranking partition:

```sql
WITH v1 AS
(
SELECT i_category, i_brand, s_store_name, s_company_name, d_year, d_moy,
sum(ss_sales_price) AS sum_sales,
rank() OVER (PARTITION BY i_category, i_brand, s_store_name, s_company_name
ORDER BY d_year, d_moy) AS rn
FROM item, store_sales, date_dim, store
WHERE ...
GROUP BY i_category, i_brand, s_store_name, s_company_name, d_year, d_moy
)
SELECT v1.sum_sales, v1_lag.sum_sales AS psum, v1_lead.sum_sales AS nsum, ...
FROM v1, v1 AS v1_lag, v1 AS v1_lead
WHERE (v1.i_category = v1_lag.i_category) AND (v1.i_category = v1_lead.i_category)
AND ...
AND (v1.rn = v1_lag.rn + 1) AND (v1.rn = v1_lead.rn - 1)
```

Both rows of every matched pair belong to the same partition of the same ranking, so one pass over
`v1` produces the same rows:

```sql
SELECT sum_sales, lagInFrame(sum_sales, 1) OVER w AS psum, leadInFrame(sum_sales, 1) OVER w AS nsum, ...
FROM v1
WINDOW w AS (PARTITION BY i_category, i_brand, s_store_name, s_company_name ORDER BY rn
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)
QUALIFY (lagInFrame(toNullable(rn), 1, NULL) OVER w IS NOT NULL)
AND (leadInFrame(toNullable(rn), 1, NULL) OVER w IS NOT NULL)
```

The patch below adds an analyzer pass, `optimize_rank_selfjoin_to_lag_lead` (default off), that does
this rewrite. On TPC-DS SF10 it removes two of the three reads of the CTE and both joins: query_47
3606 -> 1238 ms (-65.7%) and query_57 1756 -> 596 ms (-66.1%), with byte-identical answers.

Three conditions decide whether the rewrite is valid, and the pass refuses unless it can establish
all of them.

1. `rn = other.rn + k` means "the k-th row back in the partition" only when the ranking numbers each
partition without repetitions and without gaps. The pass accepts `row_number`, `rank` and
`dense_rank`, and requires the ranking's `ORDER BY` to be unique inside the partition. It proves
that from the subquery's `GROUP BY`: every grouping key must be one of the partitioning or
ordering expressions, so two rows of one partition that agreed on the ordering values would agree
on every grouping key and would be the same row. query_47's `v1` groups by the four partition
columns plus `d_year` and `d_moy`, which is exactly what its `rank()` partitions and orders by.
2. The inner join drops a row whose neighbour does not exist - the first and the last rank of each
partition. The `QUALIFY` probe reproduces that by reading the rank column, which is never NULL.
A probe on the read value itself would also have dropped a row whose neighbour exists but holds
NULL.
3. The join equality never matches a NULL partition key, while `PARTITION BY` keeps NULL as a group
of its own, so rows with a NULL partition key are filtered explicitly. All four partition columns
of query_47 are `Nullable`. Removing them before the window is safe because a NULL key puts the
whole group out of the join's reach, so no surviving row loses a neighbour.

The equated columns must also be exactly the ranking's partition columns: fewer would let the join
pair rows the window keeps apart, more would let the window pair rows the join keeps apart.

Measured on one binary with the setting as the only difference (TPC-DS SF10, warm, `max_threads=4`,
one warm-up then 11 interleaved pairs per query, `query_duration_ms` from `system.query_log`):

| query | off (mean of 11) | on (mean of 11) | change | 95% interval | rows read | peak memory |
|---|---|---|---|---|---|---|
| query_47 | 3606.0 ms | 1238.5 ms | **-65.7%** | -66.9% .. -64.4% | 86.9M -> 29.0M | 304 -> 199 MiB |
| query_57 | 1756.2 ms | 595.5 ms | **-66.1%** | -66.5% .. -65.7% | 43.7M -> 14.6M | 332 -> 214 MiB |

All 11 pairs are negative for both queries. An A/A null on the same arm gives +1.1% (6 pairs, sd
2.3, interval -1.3% .. +3.6%). Four concurrent copies of query_47 take 15.4-15.7 s of wall time with
the setting off and 5.27-5.29 s with it on, with peak query memory 306 -> 203 MiB.

The whole 99-query suite was run with the setting off and on on this binary. 96 queries return
byte-identical results. query_14 fails identically in both arms with a pre-existing code 206 under
`--union_default_mode=DISTINCT`. query_65 and query_71 print different bytes, and both do so with
byte-identical query trees in the two arms: query_65 produces three different digests in three runs
of the off arm alone, and query_71's two outputs are the same 9669 rows in a different order, so
their `ORDER BY` does not determine the result. An `EXPLAIN QUERY TREE` comparison of all 99 queries
shows the pass changes exactly two plans, query_47 and query_57. Two-rep suite minima also suggested
movement on query_21 (+120%), query_06 (+17%), query_82 (+12%) and query_66 (-19%); their query trees
are byte-identical in both arms, and a paired 6-repetition re-measurement puts all four within 1.1%
at the minimum (means -1.6% to -0.5%), so that was host noise.

Correctness: query_47 and query_57 answers are byte-identical with the setting off and on. A 19-case
matrix over a small table returns identical answers in both arms, and the pass fires in exactly the
10 cases it should: three-way lag+lead, lag only, lead at offset 2, the offset written with the
constant on either side, both sides shifted, `INNER JOIN ... ON`, `row_number`, `dense_rank`, an
expression over the neighbour's column, and a neighbour holding NULL. It refuses the other 9: a
ranking whose `ORDER BY` is not unique inside the partition, a condition the rewrite does not account
for, equating fewer columns than the ranking partitions by, a zero offset, `LEFT JOIN`, two different
relations, an aggregate over the joined rows, a kept side that does not produce the column read from
the neighbour, and a read of the neighbour from inside a lambda body, where a window function of this
query cannot be evaluated.

Two limits are worth knowing. The side that stays must already produce the column read from the
neighbour: the analyzer prunes each use site's projection separately, so `SELECT prev.sx` without the
row's own `sx` is refused rather than rewritten. And the numbers above come from the ClickHouse
TPC-DS benchmark queries at SF10, which is an adapted workload, not an audited TPC-DS result.

### Which ClickHouse versions are affected?

Tested on master at `10bec7b88102`, which reports version 26.9.1. The query shape does not depend on
a version; the patch adds a new optimization rather than fixing a regression.

### How to reproduce

Self-contained, without a TPC-DS fixture:

```sql
CREATE TABLE rank_selfjoin (p UInt32, o UInt32, x UInt32) ENGINE = MergeTree ORDER BY (p, o);
INSERT INTO rank_selfjoin SELECT number % 200000, intDiv(number, 200000), number FROM numbers(2000000);

SELECT count(), sum(sx), sum(psum), sum(nsum)
FROM
(
WITH v AS
(
SELECT p, o, sum(x) AS sx, rank() OVER (PARTITION BY p ORDER BY o) AS rn
FROM rank_selfjoin GROUP BY p, o
)
SELECT v.p AS p, v.sx AS sx, prev.sx AS psum, next.sx AS nsum
FROM v, v AS prev, v AS next
WHERE (v.p = prev.p) AND (v.p = next.p) AND (v.rn = prev.rn + 1) AND (v.rn = next.rn - 1)
)
SETTINGS optimize_rank_selfjoin_to_lag_lead = 0; -- then = 1
```

2,000,000 rows in 200,000 partitions of 10 ranks: 1022.6 -> 415.6 ms (mean of 5 interleaved pairs,
5/5 negative, sd 1.3), peak query memory 401.8 -> 108.0 MiB, and both arms return the same row count
and the same three sums.

TPC-DS: load `tests/benchmarks/tpc-ds` at SF10 and run `queries/query_47.sql` and
`queries/query_57.sql` unchanged. Use one server binary, warm caches and `max_threads=4`; alternate
the two arms (`optimize_rank_selfjoin_to_lag_lead` 0 and 1) inside one process, rotating which arm
runs first, with one warm-up per arm and 11 timed pairs; take `query_duration_ms` per run from
`system.query_log` and pair by repetition. `EXPLAIN QUERY TREE` shows whether the pass fired: the
rewritten query_47 has four `InFrame` window functions and one `QUALIFY` in place of two of its
joins.

### Expected performance

query_47 3606.0 -> 1238.5 ms (-65.7%) and query_57 1756.2 -> 595.5 ms (-66.1%). The rewrite reads the
ranked subquery once instead of three times, which is visible in rows read (86.9M -> 29.0M for
query_47) and in peak query memory (304 -> 199 MiB), and it replaces two hash joins with one window
pass over rows the ranking has already grouped by partition.

### Related issues and pull requests

https://github.com/ClickHouse/ClickHouse/issues/94858 - why `query_47.sql` in the repository
qualifies the outer query's columns with `v1.`; unrelated to the performance of the join itself.

### Additional context

@groeneai could you take a look? If the analysis and patch hold up, feel free to open a PR if it fits.

```diff
diff --git a/src/Analyzer/Passes/RankSelfJoinToLagLeadPass.cpp b/src/Analyzer/Passes/RankSelfJoinToLagLeadPass.cpp
new file mode 100644
index 00000000000..239251c45f9
--- /dev/null
+++ b/src/Analyzer/Passes/RankSelfJoinToLagLeadPass.cpp
@@ -0,0 +1,899 @@
+#include
+
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+
+#include
+
+#include
+#include
+#include
+#include
+
+namespace DB
+{
+
+namespace Setting
+{
+ extern const SettingsBool optimize_rank_selfjoin_to_lag_lead;
+}
+
+namespace
+{
+
+/// The sides of a self-join carry different generated aliases (__table2 vs __table3), so they can
+/// only be recognised as the same relation by comparing them without the aliases.
+constexpr IQueryTreeNode::CompareOptions compare_ignoring_aliases{.compare_aliases = false};
+
+/// A rank offset beyond this is not a neighbour lookup worth rewriting, and keeping the value small
+/// also keeps the offset arithmetic below away from overflow.
+constexpr Int64 max_supported_offset = 1000000;
+
+void collectConjuncts(const QueryTreeNodePtr & node, QueryTreeNodes & conjuncts)
+{
+ const auto * function_node = node->as();
+ if (function_node && function_node->getFunctionName() == "and")
+ {
+ for (const auto & argument : function_node->getArguments().getNodes())
+ collectConjuncts(argument, conjuncts);
+ return;
+ }
+
+ conjuncts.push_back(node);
+}
+
+/// `c`, `c + k`, `k + c` or `c - k`, where c is a column of one of the joined sides.
+struct ShiftedColumn
+{
+ const IQueryTreeNode * source = nullptr;
+ String column_name;
+ Int64 shift = 0;
+};
+
+std::optional tryGetIntegerConstant(const QueryTreeNodePtr & node)
+{
+ const auto * constant_node = node->as();
+ if (!constant_node)
+ return {};
+
+ const auto & value = constant_node->getValue();
+ Int64 result = 0;
+ if (value.getType() == Field::Types::Int64)
+ result = value.safeGet();
+ else if (value.getType() == Field::Types::UInt64)
+ {
+ const UInt64 unsigned_value = value.safeGet();
+ if (unsigned_value > static_cast(max_supported_offset))
+ return {};
+ result = static_cast(unsigned_value);
+ }
+ else
+ return {};
+
+ if (result > max_supported_offset || result < -max_supported_offset)
+ return {};
+
+ return result;
+}
+
+std::optional matchColumn(const QueryTreeNodePtr & node)
+{
+ const auto * column_node = node->as();
+ if (!column_node)
+ return {};
+
+ auto column_source = column_node->getColumnSourceOrNull();
+ if (!column_source)
+ return {};
+
+ return ShiftedColumn{column_source.get(), column_node->getColumnName(), 0};
+}
+
+std::optional matchShiftedColumn(const QueryTreeNodePtr & node)
+{
+ if (auto plain_column = matchColumn(node))
+ return plain_column;
+
+ const auto * function_node = node->as();
+ if (!function_node || !function_node->isOrdinaryFunction())
+ return {};
+
+ const auto & function_name = function_node->getFunctionName();
+ const bool is_plus = function_name == "plus";
+ if (!is_plus && function_name != "minus")
+ return {};
+
+ const auto & arguments = function_node->getArguments().getNodes();
+ if (arguments.size() != 2)
+ return {};
+
+ if (auto constant = tryGetIntegerConstant(arguments[1]))
+ {
+ if (auto column = matchColumn(arguments[0]))
+ {
+ column->shift = is_plus ? *constant : -*constant;
+ return column;
+ }
+ return {};
+ }
+
+ if (is_plus)
+ {
+ if (auto constant = tryGetIntegerConstant(arguments[0]))
+ {
+ if (auto column = matchColumn(arguments[1]))
+ {
+ column->shift = *constant;
+ return column;
+ }
+ }
+ }
+
+ return {};
+}
+
+/// Output name to the expression producing it, and to its name and type.
+struct ProjectionIndex
+{
+ std::unordered_map expressions;
+ std::unordered_map columns;
+};
+
+/// One side of the self-join, read only to reach a row a fixed number of ranks away.
+struct OffsetSide
+{
+ QueryTreeNodePtr table_expression;
+ /// side.rank = base.rank + delta, with delta != 0.
+ Int64 delta = 0;
+ ProjectionIndex projection;
+};
+
+struct RankSelfJoin
+{
+ /// The side that stays, and whose rows the rewritten query reads.
+ QueryTreeNodePtr base;
+ ProjectionIndex base_projection;
+ std::vector sides;
+ /// The columns the sides are equated on, which must be exactly the ranking's partition.
+ NamesAndTypes partition_columns;
+ /// The rank column the offset condition shifts.
+ NameAndTypePair rank_column;
+};
+
+/// Flattens comma joins and inner joins, and collects the conditions they carry.
+bool collectSelfJoinTree(const QueryTreeNodePtr & node, QueryTreeNodes & table_expressions, QueryTreeNodes & conjuncts)
+{
+ if (auto * cross_join_node = node->as())
+ {
+ for (const auto & join_type : cross_join_node->getJoinTypes())
+ {
+ if (join_type.locality != JoinLocality::Unspecified)
+ return false;
+ }
+
+ for (const auto & table_expression : cross_join_node->getTableExpressions())
+ {
+ if (!collectSelfJoinTree(table_expression, table_expressions, conjuncts))
+ return false;
+ }
+
+ return true;
+ }
+
+ if (auto * join_node = node->as())
+ {
+ if (join_node->getKind() != JoinKind::Inner)
+ return false;
+ if (join_node->getStrictness() != JoinStrictness::All && join_node->getStrictness() != JoinStrictness::Unspecified)
+ return false;
+ if (join_node->getLocality() != JoinLocality::Unspecified)
+ return false;
+ /// USING produces one merged column, whose source is the join rather than either side.
+ if (join_node->isUsingJoinExpression())
+ return false;
+
+ if (!collectSelfJoinTree(join_node->getLeftTableExpressionNode(), table_expressions, conjuncts))
+ return false;
+ if (!collectSelfJoinTree(join_node->getRightTableExpressionNode(), table_expressions, conjuncts))
+ return false;
+
+ if (join_node->hasJoinExpression())
+ collectConjuncts(join_node->getJoinExpression(), conjuncts);
+
+ return true;
+ }
+
+ table_expressions.push_back(node);
+ return true;
+}
+
+std::optional indexProjection(const QueryNode & query_node)
+{
+ const auto & projection_nodes = query_node.getProjection().getNodes();
+ const auto & projection_columns = query_node.getProjectionColumns();
+ if (projection_nodes.empty() || projection_nodes.size() != projection_columns.size())
+ return {};
+
+ ProjectionIndex index;
+ for (size_t i = 0; i < projection_nodes.size(); ++i)
+ {
+ /// A repeated output name would make it ambiguous which expression a use site reads.
+ if (!index.expressions.emplace(projection_columns[i].name, projection_nodes[i]).second)
+ return {};
+ index.columns.emplace(projection_columns[i].name, projection_columns[i]);
+ }
+
+ return index;
+}
+
+/// The rewrite replaces "the row k ranks away" by "the k-th row back in the window", so the rank
+/// must number the rows of each partition without repetitions and without gaps. A repetition also
+/// means the join could match several rows where the window reads one.
+///
+/// row_number, rank and dense_rank all produce 1..n when no two rows of a partition compare equal
+/// under the ORDER BY. That is what the subquery's GROUP BY proves here: every grouping key is one
+/// of the partitioning or ordering expressions, so two rows of one partition sharing the ordering
+/// values would share every grouping key and would be the same row.
+bool isGaplessRankOverPartition(const QueryNode & source_query, const QueryTreeNodePtr & rank_expression, QueryTreeNodes & partition_expressions)
+{
+ const auto * function_node = rank_expression->as();
+ if (!function_node || !function_node->isWindowFunction())
+ return false;
+
+ /// dense_rank resolves to its canonical name denseRank.
+ static const std::unordered_set row_numbering_functions = {"row_number", "rank", "dense_rank", "denserank"};
+ if (!row_numbering_functions.contains(Poco::toLower(function_node->getFunctionName())))
+ return false;
+ if (!function_node->getArguments().getNodes().empty())
+ return false;
+
+ const auto * window_node = function_node->getWindowNode()->as();
+ if (!window_node)
+ return false;
+ if (!window_node->hasOrderBy() || !window_node->hasPartitionBy())
+ return false;
+ if (window_node->hasFrameBeginOffset() || window_node->hasFrameEndOffset())
+ return false;
+
+ if (!source_query.hasGroupBy())
+ return false;
+ if (source_query.isGroupByWithTotals() || source_query.isGroupByWithRollup() || source_query.isGroupByWithCube()
+ || source_query.isGroupByWithGroupingSets() || source_query.isGroupByAll())
+ return false;
+ /// Rows removed after the ranking would leave gaps in the ranks that survive.
+ if (source_query.hasLimit() || source_query.hasOffset() || source_query.hasLimitBy() || source_query.isDistinct()
+ || source_query.isLimitWithTies() || source_query.hasInterpolate())
+ return false;
+
+ QueryTreeNodes allowed_group_by_keys;
+ for (const auto & partition_expression : window_node->getPartitionBy().getNodes())
+ allowed_group_by_keys.push_back(partition_expression);
+ for (const auto & sort_node : window_node->getOrderBy().getNodes())
+ {
+ const auto * sort = sort_node->as();
+ if (!sort)
+ return false;
+ allowed_group_by_keys.push_back(sort->getExpression());
+ }
+
+ for (const auto & group_by_key : source_query.getGroupBy().getNodes())
+ {
+ bool found = false;
+ for (const auto & allowed_key : allowed_group_by_keys)
+ {
+ if (group_by_key->isEqual(*allowed_key, compare_ignoring_aliases))
+ {
+ found = true;
+ break;
+ }
+ }
+
+ if (!found)
+ return false;
+ }
+
+ partition_expressions = window_node->getPartitionBy().getNodes();
+ return true;
+}
+
+/// Two sides produce the same rows when they differ only in which of those rows' columns they keep:
+/// the projection has already been pruned per use site by then, so it is compared column by column
+/// for the columns the rewrite actually reads.
+bool isSameRelationIgnoringProjection(const QueryTreeNodePtr & lhs, const QueryTreeNodePtr & rhs)
+{
+ auto lhs_clone = lhs->clone();
+ auto rhs_clone = rhs->clone();
+
+ auto * lhs_query = lhs_clone->as();
+ auto * rhs_query = rhs_clone->as();
+ if (!lhs_query || !rhs_query)
+ return false;
+
+ lhs_query->getProjectionNode() = std::make_shared();
+ rhs_query->getProjectionNode() = std::make_shared();
+ lhs_query->resolveProjectionColumns({});
+ rhs_query->resolveProjectionColumns({});
+
+ return lhs_clone->isEqual(*rhs_clone, compare_ignoring_aliases);
+}
+
+/// True when the expression is evaluated over the joined rows, which the rewrite replaces by the
+/// rows of one side plus a window: the number and order of rows reaching such a function would no
+/// longer be the ones the join produced.
+bool readsRowsAsAGroup(const QueryTreeNodePtr & node)
+{
+ if (!node)
+ return false;
+
+ if (const auto * function_node = node->as())
+ {
+ if (function_node->isAggregateFunction() || function_node->isWindowFunction())
+ return true;
+ }
+
+ switch (node->getNodeType())
+ {
+ /// A nested query is a scope of its own and is judged separately when it is visited.
+ case QueryTreeNodeType::QUERY:
+ case QueryTreeNodeType::UNION:
+ case QueryTreeNodeType::TABLE:
+ case QueryTreeNodeType::TABLE_FUNCTION:
+ return false;
+ default:
+ break;
+ }
+
+ for (const auto & child : node->getChildren())
+ {
+ if (readsRowsAsAGroup(child))
+ return true;
+ }
+
+ return false;
+}
+
+/// SELECT base.*, other.v FROM q AS base, q AS other
+/// WHERE base.p = other.p AND ... AND base.rn = other.rn + k
+///
+/// ORDER BY, LIMIT, LIMIT BY and DISTINCT are evaluated after window functions, so they see the same
+/// rows either way and are left alone. A condition, a grouping or a window function of the query's
+/// own would not: it would read rows the join had already paired or filtered.
+std::optional matchRankSelfJoin(QueryNode & query_node)
+{
+ if (query_node.hasPrewhere() || query_node.hasGroupBy() || query_node.hasHaving() || query_node.hasWindow()
+ || query_node.hasQualify() || query_node.hasSettingsChanges())
+ return {};
+
+ if (query_node.isGroupByAll() || query_node.isGroupByWithTotals() || query_node.isGroupByWithRollup()
+ || query_node.isGroupByWithCube() || query_node.isGroupByWithGroupingSets() || query_node.isRecursiveWith())
+ return {};
+
+ /// An aggregate or a window function of this query would be evaluated over the joined rows.
+ for (const auto & projection_node : query_node.getProjection().getNodes())
+ {
+ if (readsRowsAsAGroup(projection_node))
+ return {};
+ }
+ if (readsRowsAsAGroup(query_node.getOrderByNode()) || readsRowsAsAGroup(query_node.getLimitByNode())
+ || readsRowsAsAGroup(query_node.getInterpolate()))
+ return {};
+
+ QueryTreeNodes table_expressions;
+ QueryTreeNodes conjuncts;
+ if (!collectSelfJoinTree(query_node.getJoinTreeNode(), table_expressions, conjuncts))
+ return {};
+ if (table_expressions.size() < 2)
+ return {};
+
+ if (query_node.hasWhere())
+ collectConjuncts(query_node.getWhere(), conjuncts);
+
+ /// Every side must be a subquery: the ranking to read is defined inside it.
+ for (const auto & table_expression : table_expressions)
+ {
+ if (!table_expression->as())
+ return {};
+ }
+
+ std::unordered_map side_by_address;
+ for (const auto & table_expression : table_expressions)
+ {
+ if (!side_by_address.emplace(table_expression.get(), table_expression).second)
+ return {};
+ }
+
+ /// One offset condition per removed side, all of them against the same kept side.
+ struct OffsetCondition
+ {
+ const IQueryTreeNode * base = nullptr;
+ const IQueryTreeNode * other = nullptr;
+ Int64 delta = 0;
+ String rank_column_name;
+ };
+
+ std::vector offset_conditions;
+ /// Equalities, by the pair of sides they connect.
+ std::unordered_map equated_columns_by_side;
+ std::vector> equality_pairs;
+
+ for (const auto & conjunct : conjuncts)
+ {
+ const auto * function_node = conjunct->as();
+ if (!function_node || function_node->getFunctionName() != "equals")
+ return {};
+
+ const auto & arguments = function_node->getArguments().getNodes();
+ if (arguments.size() != 2)
+ return {};
+
+ auto left = matchShiftedColumn(arguments[0]);
+ auto right = matchShiftedColumn(arguments[1]);
+ if (!left || !right)
+ return {};
+ if (left->column_name != right->column_name)
+ return {};
+ if (!side_by_address.contains(left->source) || !side_by_address.contains(right->source))
+ return {};
+ if (left->source == right->source)
+ return {};
+
+ if (left->shift == 0 && right->shift == 0)
+ {
+ equality_pairs.emplace_back(left->source, right->source);
+ equated_columns_by_side[left->source].insert(left->column_name);
+ equated_columns_by_side[right->source].insert(right->column_name);
+ continue;
+ }
+
+ /// base.rn + left_shift = other.rn + right_shift, so other.rn = base.rn + left_shift - right_shift.
+ const Int64 delta = left->shift - right->shift;
+ if (delta == 0 || delta > max_supported_offset || delta < -max_supported_offset)
+ return {};
+
+ offset_conditions.push_back(OffsetCondition{left->source, right->source, delta, left->column_name});
+ }
+
+ if (offset_conditions.empty() || offset_conditions.size() + 1 != table_expressions.size())
+ return {};
+
+ const auto * base_address = offset_conditions.front().base;
+ const String rank_column_name = offset_conditions.front().rank_column_name;
+ std::unordered_set offset_side_addresses;
+ for (const auto & offset_condition : offset_conditions)
+ {
+ if (offset_condition.base != base_address || offset_condition.rank_column_name != rank_column_name)
+ return {};
+ if (offset_condition.other == base_address)
+ return {};
+ if (!offset_side_addresses.insert(offset_condition.other).second)
+ return {};
+ }
+
+ /// Exactly the kept side plus one removed side per offset condition.
+ for (const auto & side_entry : side_by_address)
+ {
+ if (side_entry.first != base_address && !offset_side_addresses.contains(side_entry.first))
+ return {};
+ }
+
+ /// The partition of the ranking, from the equalities of the first removed side.
+ auto & base_query = side_by_address.at(base_address)->as();
+ auto base_projection = indexProjection(base_query);
+ if (!base_projection)
+ return {};
+ if (!base_projection->expressions.contains(rank_column_name))
+ return {};
+
+ NameSet partition_names = equated_columns_by_side[offset_conditions.front().other];
+ if (partition_names.empty() || partition_names.contains(rank_column_name))
+ return {};
+
+ /// Every equality must connect the kept side to one removed side, on the same columns.
+ for (const auto & [left_address, right_address] : equality_pairs)
+ {
+ if (left_address != base_address && right_address != base_address)
+ return {};
+ }
+ for (const auto * offset_side_address : offset_side_addresses)
+ {
+ if (equated_columns_by_side[offset_side_address] != partition_names)
+ return {};
+ }
+ if (equated_columns_by_side[base_address] != partition_names)
+ return {};
+
+ QueryTreeNodes partition_expressions;
+ if (!isGaplessRankOverPartition(base_query, base_projection->expressions.at(rank_column_name), partition_expressions))
+ return {};
+
+ /// The join matches rows with equal partition columns, so those columns must be exactly what the
+ /// ranking partitions by. Fewer would let the join pair rows the window keeps apart, and more
+ /// would let the window pair rows the join keeps apart.
+ if (partition_expressions.size() != partition_names.size())
+ return {};
+ for (const auto & partition_expression : partition_expressions)
+ {
+ bool found = false;
+ for (const auto & partition_name : partition_names)
+ {
+ auto it = base_projection->expressions.find(partition_name);
+ if (it != base_projection->expressions.end() && it->second->isEqual(*partition_expression, compare_ignoring_aliases))
+ {
+ found = true;
+ break;
+ }
+ }
+
+ if (!found)
+ return {};
+ }
+
+ RankSelfJoin match;
+ match.base = side_by_address.at(base_address);
+ match.base_projection = *base_projection;
+ match.rank_column = base_projection->columns.at(rank_column_name);
+
+ /// Keep the partition columns in the ranking's own order, so the rewritten window is the one the
+ /// ranking already used.
+ for (const auto & partition_expression : partition_expressions)
+ {
+ for (const auto & partition_name : partition_names)
+ {
+ auto it = base_projection->expressions.find(partition_name);
+ if (it != base_projection->expressions.end() && it->second->isEqual(*partition_expression, compare_ignoring_aliases))
+ {
+ match.partition_columns.push_back(base_projection->columns.at(partition_name));
+ break;
+ }
+ }
+ }
+ if (match.partition_columns.size() != partition_names.size())
+ return {};
+
+ /// Each removed side must produce the same rows as the kept one, and the columns read from it
+ /// must mean the same there as in the kept side.
+ for (const auto & offset_condition : offset_conditions)
+ {
+ auto side_node = side_by_address.at(offset_condition.other);
+ if (!isSameRelationIgnoringProjection(match.base, side_node))
+ return {};
+
+ auto side_projection = indexProjection(side_node->as());
+ if (!side_projection)
+ return {};
+
+ NameSet checked_names = partition_names;
+ checked_names.insert(rank_column_name);
+ for (const auto & checked_name : checked_names)
+ {
+ auto side_it = side_projection->expressions.find(checked_name);
+ auto base_it = base_projection->expressions.find(checked_name);
+ if (side_it == side_projection->expressions.end() || base_it == base_projection->expressions.end())
+ return {};
+ if (!side_it->second->isEqual(*base_it->second, compare_ignoring_aliases))
+ return {};
+ }
+
+ match.sides.push_back(OffsetSide{side_node, offset_condition.delta, std::move(*side_projection)});
+ }
+
+ return match;
+}
+
+FunctionNodePtr createResolvedFunction(const String & function_name, QueryTreeNodes arguments, const ContextPtr & context)
+{
+ auto function_node = std::make_shared(function_name);
+ function_node->getArguments().getNodes() = std::move(arguments);
+ function_node->resolveAsFunction(
+ FunctionFactory::instance().get(function_name, context)->build(function_node->getArgumentColumns()));
+ return function_node;
+}
+
+FunctionNodePtr createResolvedWindowFunction(const String & function_name, QueryTreeNodes arguments, QueryTreeNodePtr window_node)
+{
+ auto function_node = std::make_shared(function_name);
+ function_node->getArguments().getNodes() = std::move(arguments);
+ function_node->getWindowNode() = std::move(window_node);
+
+ DataTypes argument_types;
+ for (const auto & argument : function_node->getArguments().getNodes())
+ argument_types.push_back(argument->getResultType());
+
+ AggregateFunctionProperties properties;
+ auto window_function = AggregateFunctionFactory::instance().get(
+ function_name, NullsAction::EMPTY, argument_types, {}, properties, AggregateFunctionStateVariant::Window);
+ function_node->resolveAsWindowFunction(std::move(window_function));
+
+ return function_node;
+}
+
+class RankSelfJoinRewriter
+{
+public:
+ RankSelfJoinRewriter(const RankSelfJoin & match_, const ContextPtr & context_)
+ : match(match_)
+ , context(context_)
+ , base_table_expression(std::static_pointer_cast(match_.base))
+ {
+ }
+
+ /// The whole replacement is built before the query node is touched, because a window function
+ /// over an unsupported type would otherwise leave the query half rewritten.
+ bool run(QueryNode & query_node)
+ {
+ try
+ {
+ buildWindow();
+
+ QueryTreeNodes qualify_conjuncts;
+ for (const auto & side : match.sides)
+ {
+ reads_by_side.emplace(side.table_expression.get(), &side);
+ qualify_conjuncts.push_back(createNeighbourExistsCondition(side));
+ }
+
+ const auto & projection_columns = query_node.getProjectionColumns();
+ const auto & projection_nodes = query_node.getProjection().getNodes();
+ QueryTreeNodes new_projection;
+ for (size_t i = 0; i < projection_nodes.size(); ++i)
+ {
+ auto replaced = projection_nodes[i]->clone();
+ if (!replaceReads(replaced))
+ return false;
+ if (!replaced->getResultType()->equals(*projection_columns[i].type))
+ return false;
+ new_projection.push_back(std::move(replaced));
+ }
+
+ /// ORDER BY and the clauses after it may read the neighbour too, and an alias can make
+ /// them share the very node the projection replaced, so each is rewritten on its own copy.
+ std::vector> post_window_clauses;
+ for (auto * clause : {&query_node.getOrderByNode(), &query_node.getLimitByNode(), &query_node.getInterpolate()})
+ {
+ if (!*clause)
+ continue;
+ auto replaced = (*clause)->clone();
+ if (!replaceReads(replaced))
+ return false;
+ post_window_clauses.emplace_back(clause, std::move(replaced));
+ }
+
+ /// The join equality never matches a NULL key, while PARTITION BY keeps NULL as a group
+ /// of its own. Dropping those rows before the window is safe: a NULL key puts the whole
+ /// group out of reach of the join, so no surviving row loses a neighbour.
+ QueryTreeNodes where_conjuncts;
+ for (const auto & partition_column : match.partition_columns)
+ {
+ if (!partition_column.type->isNullable() && !partition_column.type->isLowCardinalityNullable())
+ continue;
+ where_conjuncts.push_back(createResolvedFunction(
+ "isNotNull", {std::make_shared(partition_column, base_table_expression)}, context));
+ }
+
+ auto qualify = combineWithAnd(std::move(qualify_conjuncts));
+ auto where = combineWithAnd(std::move(where_conjuncts));
+ if (!qualify)
+ return false;
+
+ /// Nothing that stays may still read a side that leaves the join tree.
+ for (const auto & node : new_projection)
+ {
+ if (readsRemovedSide(node))
+ return false;
+ }
+ for (const auto & clause_entry : post_window_clauses)
+ {
+ if (readsRemovedSide(clause_entry.second))
+ return false;
+ }
+
+ query_node.getProjection().getNodes() = std::move(new_projection);
+ for (auto & [clause, replaced] : post_window_clauses)
+ *clause = std::move(replaced);
+ query_node.getJoinTreeNode() = match.base;
+ query_node.getWhere() = std::move(where);
+ query_node.getQualify() = std::move(qualify);
+ return true;
+ }
+ catch (...)
+ {
+ return false;
+ }
+ }
+
+private:
+ const RankSelfJoin & match;
+ const ContextPtr & context;
+ std::shared_ptr base_table_expression;
+ QueryTreeNodePtr window_node;
+ std::unordered_map reads_by_side;
+
+ void buildWindow()
+ {
+ WindowFrame frame;
+ frame.is_default = false;
+ frame.type = WindowFrame::FrameType::ROWS;
+ frame.begin_type = WindowFrame::BoundaryType::Unbounded;
+ frame.begin_preceding = true;
+ frame.end_type = WindowFrame::BoundaryType::Unbounded;
+ frame.end_preceding = false;
+ frame.checkValid();
+
+ auto window = std::make_shared(frame);
+
+ QueryTreeNodes partition_by;
+ for (const auto & partition_column : match.partition_columns)
+ partition_by.push_back(std::make_shared(partition_column, base_table_expression));
+ window->getPartitionByNode() = std::make_shared(std::move(partition_by));
+
+ auto rank_column = std::make_shared(match.rank_column, base_table_expression);
+ window->getOrderByNode()
+ = std::make_shared(QueryTreeNodes{std::make_shared(std::move(rank_column), SortDirection::ASCENDING)});
+
+ window_node = std::move(window);
+ }
+
+ /// The rows are ordered by the rank inside the partition and the rank leaves no gaps, so the row
+ /// the join matched is exactly |delta| rows away in the window.
+ FunctionNodePtr createWindowRead(const OffsetSide & side, QueryTreeNodePtr argument) const
+ {
+ const String function_name = side.delta < 0 ? "lagInFrame" : "leadInFrame";
+ const UInt64 offset = static_cast(side.delta < 0 ? -side.delta : side.delta);
+ QueryTreeNodes arguments{std::move(argument), std::make_shared(offset)};
+ return createResolvedWindowFunction(function_name, std::move(arguments), window_node->clone());
+ }
+
+ /// The inner join drops a row when the neighbour it asks for does not exist. The probe reads the
+ /// rank column, which is never NULL, so it is NULL only outside the partition - unlike a probe
+ /// on the read value itself, which would also drop a row whose neighbour holds a NULL.
+ QueryTreeNodePtr createNeighbourExistsCondition(const OffsetSide & side) const
+ {
+ auto rank_column = std::make_shared(match.rank_column, base_table_expression);
+ auto nullable_rank = createResolvedFunction("toNullable", {std::move(rank_column)}, context);
+ auto nullable_rank_type = nullable_rank->getResultType();
+
+ const String function_name = side.delta < 0 ? "lagInFrame" : "leadInFrame";
+ const UInt64 offset = static_cast(side.delta < 0 ? -side.delta : side.delta);
+ QueryTreeNodes arguments{
+ std::move(nullable_rank),
+ std::make_shared(offset),
+ std::make_shared(Field{}, std::move(nullable_rank_type))};
+
+ auto probe = createResolvedWindowFunction(function_name, std::move(arguments), window_node->clone());
+ return createResolvedFunction("isNotNull", {std::move(probe)}, context);
+ }
+
+ bool replaceReads(QueryTreeNodePtr & node)
+ {
+ /// A window function belongs to this query's scope, so a read from inside a subquery or a
+ /// lambda body cannot be replaced by one. Such a query keeps its join.
+ switch (node->getNodeType())
+ {
+ case QueryTreeNodeType::QUERY:
+ case QueryTreeNodeType::UNION:
+ case QueryTreeNodeType::LAMBDA:
+ return !readsRemovedSide(node);
+ default:
+ break;
+ }
+
+ if (const auto * column_node = node->as())
+ {
+ auto column_source = column_node->getColumnSourceOrNull();
+ if (!column_source)
+ return true;
+
+ auto it = reads_by_side.find(column_source.get());
+ if (it == reads_by_side.end())
+ return true;
+
+ /// The column is read from the removed side, so the kept side must produce it under the
+ /// same name from the same expression - otherwise the two sides only agree on their rows,
+ /// not on what this name means.
+ const auto & column_name = column_node->getColumnName();
+ auto base_expression = match.base_projection.expressions.find(column_name);
+ auto base_column = match.base_projection.columns.find(column_name);
+ auto side_expression = it->second->projection.expressions.find(column_name);
+ if (base_expression == match.base_projection.expressions.end()
+ || side_expression == it->second->projection.expressions.end()
+ || base_column == match.base_projection.columns.end())
+ return false;
+ if (!base_expression->second->isEqual(*side_expression->second, compare_ignoring_aliases))
+ return false;
+ if (!base_column->second.type->equals(*column_node->getResultType()))
+ return false;
+
+ auto replacement = createWindowRead(*it->second, std::make_shared(base_column->second, base_table_expression));
+ if (!replacement->getResultType()->equals(*column_node->getResultType()))
+ return false;
+
+ replacement->setAlias(node->getAlias());
+ node = std::move(replacement);
+ return true;
+ }
+
+ for (auto & child : node->getChildren())
+ {
+ if (child && !replaceReads(child))
+ return false;
+ }
+
+ return true;
+ }
+
+ bool readsRemovedSide(const QueryTreeNodePtr & node) const
+ {
+ if (const auto * column_node = node->as())
+ {
+ auto column_source = column_node->getColumnSourceOrNull();
+ if (column_source && reads_by_side.contains(column_source.get()))
+ return true;
+ }
+
+ for (const auto & child : node->getChildren())
+ {
+ if (child && readsRemovedSide(child))
+ return true;
+ }
+
+ return false;
+ }
+
+ QueryTreeNodePtr combineWithAnd(QueryTreeNodes conjuncts) const
+ {
+ if (conjuncts.empty())
+ return nullptr;
+ if (conjuncts.size() == 1)
+ return std::move(conjuncts.front());
+ return createResolvedFunction("and", std::move(conjuncts), context);
+ }
+};
+
+void collectQueryNodes(const QueryTreeNodePtr & node, QueryTreeNodes & query_nodes)
+{
+ if (!node)
+ return;
+
+ if (node->getNodeType() == QueryTreeNodeType::QUERY)
+ query_nodes.push_back(node);
+
+ for (const auto & child : node->getChildren())
+ collectQueryNodes(child, query_nodes);
+}
+
+}
+
+void RankSelfJoinToLagLeadPass::run(QueryTreeNodePtr & query_tree_node, ContextPtr context)
+{
+ if (!context->getSettingsRef()[Setting::optimize_rank_selfjoin_to_lag_lead])
+ return;
+
+ QueryTreeNodes query_nodes;
+ collectQueryNodes(query_tree_node, query_nodes);
+
+ for (const auto & node : query_nodes)
+ {
+ auto & query_node = node->as();
+ auto match = matchRankSelfJoin(query_node);
+ if (!match)
+ continue;
+
+ /// A column read from a removed side must be reachable in the kept side under the same name.
+ RankSelfJoinRewriter rewriter(*match, context);
+ rewriter.run(query_node);
+ }
+}
+
+}
diff --git a/src/Analyzer/Passes/RankSelfJoinToLagLeadPass.h b/src/Analyzer/Passes/RankSelfJoinToLagLeadPass.h
new file mode 100644
index 00000000000..3d389ae3f51
--- /dev/null
+++ b/src/Analyzer/Passes/RankSelfJoinToLagLeadPass.h
@@ -0,0 +1,45 @@
+#pragma once
+
+#include
+
+namespace DB
+{
+
+/** Read the neighbouring row of a ranked subquery with lagInFrame/leadInFrame instead of joining
+ * the subquery to itself.
+ *
+ * WITH v AS (SELECT p, o, f(x) AS v, row_number() OVER (PARTITION BY p ORDER BY o) AS rn
+ * FROM t GROUP BY p, o)
+ * SELECT v.p, v.v, prev.v FROM v, v AS prev WHERE v.p = prev.p AND v.rn = prev.rn + 1
+ *
+ * reads v twice and joins it to itself only to reach the row one rank back, although both rows
+ * belong to the same partition of the same ranking. One pass with a window function produces the
+ * same rows:
+ *
+ * SELECT p, v, lagInFrame(v, 1) OVER w FROM v
+ * WINDOW w AS (PARTITION BY p ORDER BY rn ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)
+ * QUALIFY lagInFrame(toNullable(rn), 1, NULL) OVER w IS NOT NULL
+ *
+ * `rn = other.rn + k` means "the k-th row back in the partition" only when the rank numbers the
+ * rows of a partition without repetitions or gaps, so the rewrite requires a rank whose ORDER BY
+ * is unique inside the partition, proven from the subquery's GROUP BY. The QUALIFY probe reads the
+ * rank column, which is never NULL, and so drops exactly the rows whose neighbour does not exist -
+ * as the inner join does - while keeping a row whose neighbour holds a NULL value. The join
+ * equality discards NULL partition keys, which PARTITION BY would keep as a group of their own, so
+ * nullable partition keys are filtered explicitly.
+ */
+class RankSelfJoinToLagLeadPass final : public IQueryTreePass
+{
+public:
+ String getName() override { return "RankSelfJoinToLagLead"; }
+
+ String getDescription() override
+ {
+ return "Read the neighbouring row of a ranked subquery with lagInFrame/leadInFrame instead of "
+ "joining the subquery to itself on a rank offset";
+ }
+
+ void run(QueryTreeNodePtr & query_tree_node, ContextPtr context) override;
+};
+
+}
diff --git a/src/Analyzer/QueryTreePassManager.cpp b/src/Analyzer/QueryTreePassManager.cpp
index 9c98c2bc33e..01d63a87007 100644
--- a/src/Analyzer/QueryTreePassManager.cpp
+++ b/src/Analyzer/QueryTreePassManager.cpp
@@ -25,6 +25,7 @@
#include
#include
#include
+#include
#include
#include
#include
@@ -342,6 +343,8 @@ void addQueryTreePasses(QueryTreePassManager & manager, bool only_analyze)
manager.addPass(std::make_unique());
manager.addPass(std::make_unique());

+ manager.addPass(std::make_unique());
+
manager.addPass(std::make_unique());
manager.addPass(std::make_unique());

diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp
index d9b6a23ecca..c05dae69c50 100644
--- a/src/Core/Settings.cpp
+++ b/src/Core/Settings.cpp
@@ -6287,6 +6287,19 @@ Rewrite sumIf() and sum(if()) function countIf() function when logically equival
DECLARE(Bool, optimize_empty_string_comparisons, true, R"(
Convert expressions like col = '' or '' = col into empty(col), and col != '' or '' != col into notEmpty(col),
only when col is of String or FixedString type.
+)", 0) \
+ DECLARE(Bool, optimize_rank_selfjoin_to_lag_lead, false, R"(
+Read the neighbouring row of a ranked subquery with `lagInFrame`/`leadInFrame` instead of joining the subquery to itself.
+
+A self-join of one ranked subquery whose conditions are equalities on the ranking's partition columns
+plus one rank offset, such as `v.rn = prev.rn + 1`, reads that subquery once per side only to reach a
+row a fixed number of ranks away. When the ranking numbers each partition without repetitions or gaps,
+which is proven from the subquery's `GROUP BY`, one pass answers it with `lagInFrame`/`leadInFrame` over
+`PARTITION BY` the equated columns `ORDER BY` the rank, keeping only the rows whose neighbour exists.
+
+:::note
+Supported only with the analyzer (`enable_analyzer = 1`).
+:::
)", 0) \
DECLARE(Bool, optimize_rewrite_aggregate_function_with_if, true, R"(
Rewrite aggregate functions with if expression as argument when logically equivalent.
```