-- Dynamic Filter Test: TopK (Sort + Limit)
-- Tests dynamic filters generated by TopK operator
-- Test 2.1: Basic TopK
CREATE TABLE orders ("id" INT, customer_id INT, amount DOUBLE, ts TIMESTAMP TIME INDEX);

Affected Rows: 0

-- Insert sufficient test data
INSERT INTO orders VALUES
  (1, 1, 1100.0, '2024-01-01 00:00:00'),
  (2, 2, 1000.0, '2024-01-02 00:00:00'),
  (3, 3, 900.0, '2024-01-03 00:00:00');

Affected Rows: 3

admin flush_table('orders');

+-----------------------------+
| ADMIN flush_table('orders') |
+-----------------------------+
| 0                           |
+-----------------------------+

INSERT INTO orders VALUES
  (4, 1, 325.0, '2024-01-04 00:00:00'),
  (5, 2, 300.0, '2024-01-05 00:00:00'),
  (6, 1, 250.0, '2024-01-06 00:00:00'),
  (7, 3, 225.0, '2024-01-07 00:00:00');

Affected Rows: 4

admin flush_table('orders');

+-----------------------------+
| ADMIN flush_table('orders') |
+-----------------------------+
| 0                           |
+-----------------------------+

INSERT INTO orders VALUES
  (8, 2, 200.0, '2024-01-08 00:00:00'),
  (9, 1, 175.0, '2024-01-09 00:00:00'),
  (10, 3, 150.0, '2024-01-10 00:00:00'),
  (11, 2, 100.0, '2024-01-11 00:00:00'),
  (12, 1, 50.0, '2024-01-12 00:00:00');

Affected Rows: 5

-- Test: Basic TopK query - get top 3 orders by amount
-- TopK (Sort + Limit) should generate dynamic filter on amount
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
-- SQLNESS REPLACE (partitioning.*) REDACTED
EXPLAIN SELECT "id", customer_id, amount
FROM orders
ORDER BY amount DESC
LIMIT 3;

+---------------+--------------------------------------------------------------+
| plan_type     | plan                                                         |
+---------------+--------------------------------------------------------------+
| logical_plan  | MergeScan [is_placeholder=false, remote_input=[              |
|               | Limit: skip=0, fetch=3                                       |
|               |   Sort: orders.amount DESC NULLS FIRST                       |
|               |     Projection: orders.id, orders.customer_id, orders.amount |
|               |       TableScan: orders                                      |
|               | ]]                                                           |
| physical_plan | CooperativeExec                                              |
|               |   MergeScanExec: REDACTED
|               |                                                              |
+---------------+--------------------------------------------------------------+

-- SQLNESS REPLACE ("metrics_per_partition":\s*.*metrics=) "metrics_per_partition": REDACTED metrics=
-- SQLNESS REPLACE (metrics=\{.*\}) metrics=REDACTED
-- SQLNESS REPLACE (metrics=\[[^\]]*\]) metrics=REDACTED
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (=Hash.*) =REDACTED
-- SQLNESS REPLACE (-+) -
-- SQLNESS REPLACE (\s\s+) _
-- SQLNESS REPLACE "(file_id|time_range_start|time_range_end)":"[^"]+" "$1":"REDACTED"
-- SQLNESS REPLACE ("[a-z_]+":"[0-9\.]+(ns|us|µs|ms|s)") "DURATION": REDACTED
-- SQLNESS REPLACE "(size|flat_format)":\s*(\d+|true|false) "$1":REDACTED
-- SQLNESS REPLACE ,\s*filter=.*?metrics=  metrics=
-- SQLNESS REPLACE Total\s+rows:\s+\d+ Total rows: REDACTED
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
-- SQLNESS REPLACE "partition_count":\{(.*?)\} "partition_count":REDACTED
-- SQLNESS REPLACE DynamicFilter\s\[[^\]]*\] DynamicFilter [ REDACTED ]
-- SQLNESS REPLACE "files":\s\[[^\]]*\] "files": REDACTED
EXPLAIN ANALYZE VERBOSE SELECT "id", customer_id, amount
FROM orders
ORDER BY amount DESC
LIMIT 3;

+-+-+-+
| stage | node | plan_|
+-+-+-+
| 0_| 0_|_CooperativeExec metrics=REDACTED_|
|_|_|_MergeScanExec: REDACTED
|_|_|_|
| 1_| 0_|_SortPreservingMergeExec: [amount@2 DESC], fetch=3 metrics=REDACTED_|
|_|_|_SortExec: TopK(fetch=3), expr=[amount@2 DESC], preserve_partitioning=[true] metrics=REDACTED_|
|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "amount"], "dyn_filters": ["DynamicFilter [ REDACTED ]"], "files": REDACTED, "flat_format":REDACTED, "metrics_per_partition": REDACTED metrics=REDACTED |
|_|_|_|
|_|_| Total rows: REDACTED_|
+-+-+-+

-- Verify correctness
SELECT "id", customer_id, amount
FROM orders
ORDER BY amount DESC
LIMIT 3;

+----+-------------+--------+
| id | customer_id | amount |
+----+-------------+--------+
| 1  | 1           | 1100.0 |
| 2  | 2           | 1000.0 |
| 3  | 3           | 900.0  |
+----+-------------+--------+

DROP TABLE orders;

Affected Rows: 0

-- Test 2.2: TopK with Projection Alias
-- USER NOTE: must make alias for topk's sort column to be able to test dynamic filter pushdown
CREATE TABLE orders ("id" INT, customer_id INT, amount DOUBLE, ts TIMESTAMP TIME INDEX);

Affected Rows: 0

-- Insert test data
INSERT INTO orders VALUES
  (1, 1, 100.0, '2024-01-01 00:00:00'),
  (2, 2, 200.0, '2024-01-02 00:00:00'),
  (3, 3, 300.0, '2024-01-03 00:00:00'),
  (4, 1, 150.0, '2024-01-04 00:00:00'),
  (5, 2, 250.0, '2024-01-05 00:00:00'),
  (6, 1, 175.0, '2024-01-06 00:00:00'),
  (7, 3, 325.0, '2024-01-07 00:00:00'),
  (8, 2, 225.0, '2024-01-08 00:00:00');

Affected Rows: 8

-- Test: TopK with projection alias
-- The alias on amount tests that dynamic filter correctly handles column references after projection
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
-- SQLNESS REPLACE (partitioning.*) REDACTED
EXPLAIN SELECT "id", customer_id, total_amount
FROM (SELECT "id", customer_id, amount as total_amount, ts FROM orders)
ORDER BY total_amount DESC
LIMIT 3;

+---------------+-------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                      |
+---------------+-------------------------------------------------------------------------------------------+
| logical_plan  | MergeScan [is_placeholder=false, remote_input=[                                           |
|               | Limit: skip=0, fetch=3                                                                    |
|               |   Sort: total_amount DESC NULLS FIRST                                                     |
|               |     Projection: orders.id, orders.customer_id, total_amount                               |
|               |       Projection: orders.id, orders.customer_id, orders.amount AS total_amount, orders.ts |
|               |         TableScan: orders                                                                 |
|               | ]]                                                                                        |
| physical_plan | CooperativeExec                                                                           |
|               |   MergeScanExec: REDACTED
|               |                                                                                           |
+---------------+-------------------------------------------------------------------------------------------+

-- SQLNESS REPLACE (-+) -
-- SQLNESS REPLACE (\s\s+) _
-- SQLNESS REPLACE (metrics.*) REDACTED
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (=Hash.*) =REDACTED
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
-- SQLNESS REPLACE "partition_count":\{(.*?)\} "partition_count":REDACTED
-- some order problem between standalone&distributed, comment out for now
-- EXPLAIN ANALYZE SELECT "id", customer_id, total_amount
-- FROM (SELECT "id", customer_id, amount as total_amount, ts FROM orders)
-- ORDER BY total_amount DESC
-- LIMIT 3;
-- Verify correctness
SELECT "id", customer_id, total_amount
FROM (SELECT "id", customer_id, amount as total_amount, ts FROM orders)
ORDER BY total_amount DESC
LIMIT 3;

+-+-+-+
| id | customer_id | total_amount |
+-+-+-+
| 7_| 3_| 325.0_|
| 3_| 3_| 300.0_|
| 5_| 2_| 250.0_|
+-+-+-+

DROP TABLE orders;

Affected Rows: 0

