Files
discord9 b30765e5e9 fix: preserve distributed topk merge ordering (#8432)
* fix: preserve distributed topk for scalar latest

Signed-off-by: discord9 <discord9@163.com>

* fix: preserve scalar latest ordering across merge scan

Signed-off-by: discord9 <discord9@163.com>

* fix: carry merge scan ordering from planner rewrite

Signed-off-by: discord9 <discord9@163.com>

* test: cover merge scan ordering metadata

Signed-off-by: discord9 <discord9@163.com>

* test: cover merge scan partition ordering gate

Signed-off-by: discord9 <discord9@163.com>

* fix: require adjacent merge sort for ordering metadata

Signed-off-by: discord9 <discord9@163.com>

* test: update ordering sqlness plans

Signed-off-by: discord9 <discord9@163.com>

* fix: preserve distributed topk merge ordering

Signed-off-by: discord9 <discord9@163.com>

* test: cover latest per series queries

Signed-off-by: discord9 <discord9@163.com>

* test: update distributed merge sort sqlness

Signed-off-by: discord9 <discord9@163.com>

* test: cover merge scan ordering partition gate

Signed-off-by: discord9 <discord9@163.com>

* test: cover merge scan over-partition ordering

Signed-off-by: discord9 <discord9@163.com>

* test: cover distributed latest with low parallelism

Signed-off-by: discord9 <discord9@163.com>

* fix: keep distributed merge sort opaque

Signed-off-by: discord9 <discord9@163.com>

* test: assert distributed scalar latest merge sort

Signed-off-by: discord9 <discord9@163.com>

* test: clarify merge scan ordering helper

Signed-off-by: discord9 <discord9@163.com>

* fix: complete merge sort exec delegation

Signed-off-by: discord9 <discord9@163.com>

* docs: explain merge sort limit pushdown

Signed-off-by: discord9 <discord9@163.com>

* test: cover merge sort optimizer opacity

Signed-off-by: discord9 <discord9@163.com>

* docs: explain merge sort optimizer hooks

Signed-off-by: discord9 <discord9@163.com>

* test: cover merge sort optimizer hooks

Signed-off-by: discord9 <discord9@163.com>

* fix: preserve merge sort child topk

Signed-off-by: discord9 <discord9@163.com>

* test: update order by topk plan

Signed-off-by: discord9 <discord9@163.com>

* fix: recognize merge sort global fetch

Signed-off-by: discord9 <discord9@163.com>

---------

Signed-off-by: discord9 <discord9@163.com>
2026-07-14 12:53:07 +00:00

135 lines
4.1 KiB
SQL

CREATE TABLE netflow_raw (
greptime_timestamp TIMESTAMP TIME INDEX,
src STRING,
dst_port INT,
dst_addr STRING,
byte_count DOUBLE,
flow_id INT,
PRIMARY KEY (src, dst_port)
);
INSERT INTO netflow_raw VALUES
(1000, 'source-a', 80, 'target-80-old', 100.0, 1),
(5000, 'source-a', 80, 'target-80-new', 500.0, 2),
(2000, 'source-a', 443, 'target-443-old', 200.0, 3),
(7000, 'source-a', 443, 'target-443-new', 700.0, 4),
(3000, 'source-a', 22, 'target-22-only', 300.0, 5),
(4000, 'source-a', 8080, 'target-8080-only', 400.0, 6),
(8000, 'source-b', 9999, 'target-other-src', 999.0, 7);
-- Latest row per dst_port using a window subquery. The window result must be
-- filtered outside the subquery, not directly in the WHERE clause.
SELECT greptime_timestamp, dst_port, dst_addr, byte_count, flow_id
FROM (
SELECT
greptime_timestamp,
dst_port,
dst_addr,
byte_count,
flow_id,
ROW_NUMBER() OVER (
PARTITION BY dst_port
ORDER BY greptime_timestamp DESC
) AS rn
FROM netflow_raw
WHERE src = 'source-a'
AND greptime_timestamp >= '1970-01-01T00:00:02'::timestamp
) latest_per_port
WHERE rn = 1
ORDER BY greptime_timestamp DESC, dst_port
LIMIT 3;
-- DISTINCT ON keeps the first row per dst_port according to its own ORDER BY.
-- Wrap it to apply the final "latest groups first" ordering and LIMIT.
SELECT greptime_timestamp, dst_port, dst_addr, byte_count, flow_id
FROM (
SELECT DISTINCT ON (dst_port)
greptime_timestamp,
dst_port,
dst_addr,
byte_count,
flow_id
FROM netflow_raw
WHERE src = 'source-a'
AND greptime_timestamp >= '1970-01-01T00:00:02'::timestamp
ORDER BY dst_port, greptime_timestamp DESC
) latest_per_port
ORDER BY greptime_timestamp DESC, dst_port
LIMIT 3;
-- RANK returns all rows tied at the latest timestamp for each dst_port.
SELECT greptime_timestamp, dst_port, dst_addr, byte_count, flow_id, r
FROM (
SELECT
greptime_timestamp,
dst_port,
dst_addr,
byte_count,
flow_id,
RANK() OVER (
PARTITION BY dst_port
ORDER BY greptime_timestamp DESC
) AS r
FROM netflow_raw
WHERE src = 'source-a'
AND greptime_timestamp >= '1970-01-01T00:00:02'::timestamp
) latest_per_port
WHERE r = 1
ORDER BY greptime_timestamp DESC, dst_port
LIMIT 3;
-- Ordered aggregate variant when the caller can enumerate the columns to fetch.
SELECT
dst_port,
last_value(greptime_timestamp ORDER BY greptime_timestamp) AS latest_ts,
last_value(dst_addr ORDER BY greptime_timestamp) AS latest_dst_addr,
last_value(byte_count ORDER BY greptime_timestamp) AS latest_byte_count,
last_value(flow_id ORDER BY greptime_timestamp) AS latest_flow_id
FROM netflow_raw
WHERE src = 'source-a'
AND greptime_timestamp >= '1970-01-01T00:00:02'::timestamp
GROUP BY dst_port
ORDER BY latest_ts DESC, dst_port
LIMIT 3;
CREATE TABLE netflow_raw_ties (
greptime_timestamp TIMESTAMP TIME INDEX,
src STRING,
dst_port INT,
dst_addr STRING,
byte_count DOUBLE,
flow_id INT,
PRIMARY KEY (src, dst_port, flow_id)
);
INSERT INTO netflow_raw_ties VALUES
(1000, 'source-a', 443, 'target-443-old', 100.0, 1),
(9000, 'source-a', 443, 'target-443-tie-a', 900.0, 2),
(9000, 'source-a', 443, 'target-443-tie-b', 901.0, 3),
(8000, 'source-a', 80, 'target-80-new', 800.0, 4),
(9500, 'source-b', 443, 'target-other-src', 950.0, 5);
-- RANK with r = 1 keeps all rows tied at the latest timestamp per dst_port.
SELECT greptime_timestamp, dst_port, dst_addr, byte_count, flow_id, r
FROM (
SELECT
greptime_timestamp,
dst_port,
dst_addr,
byte_count,
flow_id,
RANK() OVER (
PARTITION BY dst_port
ORDER BY greptime_timestamp DESC
) AS r
FROM netflow_raw_ties
WHERE src = 'source-a'
AND greptime_timestamp >= '1970-01-01T00:00:02'::timestamp
) latest_per_port
WHERE r = 1
ORDER BY greptime_timestamp DESC, dst_port, flow_id;
DROP TABLE netflow_raw_ties;
DROP TABLE netflow_raw;