-- Dynamic Filter Test: Hash Join
-- Tests dynamic filters generated by hash join
-- Test 1.1: Basic Hash Join
CREATE TABLE orders ("id" INT, customer_id INT, amount DOUBLE, ts TIMESTAMP TIME INDEX);

Affected Rows: 0

CREATE TABLE customers (customer_id INT, "name" STRING, tier STRING, ts TIMESTAMP TIME INDEX);

Affected Rows: 0

-- Insert sufficient 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, 1, 150.0, '2024-01-03 00:00:00');

Affected Rows: 3

-- Insert more rows to trigger dynamic filter
INSERT INTO customers VALUES
  (1, 'Alice', 'gold', '2024-01-01 00:00:00'),
  (2, 'Bob', 'silver', '2024-01-02 00:00:00');

Affected Rows: 2

-- Test: Basic hash join should produce dynamic filter
-- The hash join will build hash table from customers (small table)
-- and probe with orders (larger table), potentially using dynamic filter
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE (partitioning.*) REDACTED
EXPLAIN SELECT o."id", o.amount, c."name", c.tier
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
WHERE c.tier = 'gold';

+---------------+-----------------------------------------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                                                              |
+---------------+-----------------------------------------------------------------------------------------------------------------------------------+
| logical_plan  | Projection: o.id, o.amount, c.name, c.tier                                                                                        |
|               |   Inner Join: o.customer_id = c.customer_id                                                                                       |
|               |     Projection: o.id, o.customer_id, o.amount                                                                                     |
|               |       MergeScan [is_placeholder=false, remote_input=[                                                                             |
|               | SubqueryAlias: o                                                                                                                  |
|               |   TableScan: orders                                                                                                               |
|               | ]]                                                                                                                                |
|               |     Filter: c.tier = Utf8("gold")                                                                                                 |
|               |       Projection: c.customer_id, c.name, c.tier                                                                                   |
|               |         MergeScan [is_placeholder=false, remote_input=[                                                                           |
|               | SubqueryAlias: c                                                                                                                  |
|               |   TableScan: customers                                                                                                            |
|               | ]]                                                                                                                                |
| physical_plan | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] |
|               |   RepartitionExec: REDACTED
|               |     ProjectionExec: expr=[id@0 as id, customer_id@1 as customer_id, amount@2 as amount]                                           |
|               |       MergeScanExec: REDACTED
|               |   RepartitionExec: REDACTED
|               |     FilterExec: tier@2 = gold                                                                                                     |
|               |       ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier]                                         |
|               |         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
EXPLAIN ANALYZE VERBOSE SELECT o."id", o.amount, c."name", c.tier
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
WHERE c.tier = 'gold';

+-+-+-+
| stage | node | plan_|
+-+-+-+
| 0_| 0_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(customer_id@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] metrics=REDACTED_|
|_|_|_RepartitionExec: partitioning=REDACTED
|_|_|_ProjectionExec: expr=[id@0 as id, customer_id@1 as customer_id, amount@2 as amount] metrics=REDACTED_|
|_|_|_MergeScanExec: REDACTED
|_|_|_RepartitionExec: partitioning=REDACTED
|_|_|_FilterExec: tier@2 = gold metrics=REDACTED_|
|_|_|_ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] metrics=REDACTED_|
|_|_|_MergeScanExec: REDACTED
|_|_|_|
| 1_| 0_|_CooperativeExec metrics=REDACTED_|
|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "amount", "ts"], "flat_format":REDACTED, "metrics_per_partition": REDACTED metrics=REDACTED |
|_|_|_|
| 1_| 0_|_CooperativeExec metrics=REDACTED_|
|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier", "ts"], "flat_format":REDACTED, "metrics_per_partition": REDACTED metrics=REDACTED_|
|_|_|_|
|_|_| Total rows: REDACTED_|
+-+-+-+

-- Verify correctness
-- SQLNESS SORT_RESULT
SELECT o."id", o.amount, c."name", c.tier
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
WHERE c.tier = 'gold';

+----+--------+-------+------+
+----+--------+-------+------+
+----+--------+-------+------+
| 1  | 100.0  | Alice | gold |
| 3  | 150.0  | Alice | gold |
| id | amount | name  | tier |

DROP TABLE orders;

Affected Rows: 0

DROP TABLE customers;

Affected Rows: 0

-- Test 1.2: Hash Join with Projection Alias
CREATE TABLE orders ("id" INT, customer_id INT, amount DOUBLE, ts TIMESTAMP TIME INDEX);

Affected Rows: 0

CREATE TABLE customers (customer_id INT, "name" STRING, tier STRING, 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, 1, 150.0, '2024-01-03 00:00:00'),
  (4, 3, 300.0, '2024-01-04 00:00:00');

Affected Rows: 4

INSERT INTO customers VALUES
  (1, 'Alice', 'gold', '2024-01-01 00:00:00'),
  (2, 'Bob', 'silver', '2024-01-02 00:00:00'),
  (3, 'Charlie', 'bronze', '2024-01-03 00:00:00');

Affected Rows: 3

-- Test: Hash join with projection alias
-- The alias on customer_id tests that dynamic filter correctly handles column references
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE (partitioning.*) REDACTED
EXPLAIN SELECT o."id", o.amount, c."name", c.tier
FROM (SELECT "id", customer_id as cid, amount, ts FROM orders) o
JOIN customers c ON o.cid = c.customer_id
WHERE c.tier IN ('gold', 'silver');

+---------------+---------------------------------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                                                      |
+---------------+---------------------------------------------------------------------------------------------------------------------------+
| logical_plan  | Projection: o.id, o.amount, c.name, c.tier                                                                                |
|               |   Inner Join: o.cid = c.customer_id                                                                                       |
|               |     Projection: o.id, o.cid, o.amount                                                                                     |
|               |       MergeScan [is_placeholder=false, remote_input=[                                                                     |
|               | SubqueryAlias: o                                                                                                          |
|               |   Projection: orders.id, orders.customer_id AS cid, orders.amount, orders.ts                                              |
|               |     TableScan: orders                                                                                                     |
|               | ]]                                                                                                                        |
|               |     Filter: c.tier = Utf8("gold") OR c.tier = Utf8("silver")                                                              |
|               |       Projection: c.customer_id, c.name, c.tier                                                                           |
|               |         MergeScan [is_placeholder=false, remote_input=[                                                                   |
|               | SubqueryAlias: c                                                                                                          |
|               |   TableScan: customers                                                                                                    |
|               | ]]                                                                                                                        |
| physical_plan | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(cid@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] |
|               |   RepartitionExec: REDACTED
|               |     ProjectionExec: expr=[id@0 as id, cid@1 as cid, amount@2 as amount]                                                   |
|               |       MergeScanExec: REDACTED
|               |   RepartitionExec: REDACTED
|               |     FilterExec: tier@2 = gold OR tier@2 = silver                                                                          |
|               |       ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier]                                 |
|               |         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
EXPLAIN ANALYZE VERBOSE SELECT o."id", o.amount, c."name", c.tier
FROM (SELECT "id", customer_id as cid, amount, ts FROM orders) o
JOIN customers c ON o.cid = c.customer_id
WHERE c.tier IN ('gold', 'silver');

+-+-+-+
| stage | node | plan_|
+-+-+-+
| 0_| 0_|_HashJoinExec: mode=Partitioned, join_type=Inner, on=[(cid@1, customer_id@0)], projection=[id@0, amount@2, name@4, tier@5] metrics=REDACTED_|
|_|_|_RepartitionExec: partitioning=REDACTED
|_|_|_ProjectionExec: expr=[id@0 as id, cid@1 as cid, amount@2 as amount] metrics=REDACTED_|
|_|_|_MergeScanExec: REDACTED
|_|_|_RepartitionExec: partitioning=REDACTED
|_|_|_FilterExec: tier@2 = gold OR tier@2 = silver metrics=REDACTED_|
|_|_|_ProjectionExec: expr=[customer_id@0 as customer_id, name@1 as name, tier@2 as tier] metrics=REDACTED_|
|_|_|_MergeScanExec: REDACTED
|_|_|_|
| 1_| 0_|_ProjectionExec: expr=[id@0 as id, customer_id@1 as cid, amount@2 as amount, ts@3 as ts] metrics=REDACTED_|
|_|_|_CooperativeExec metrics=REDACTED_|
|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["id", "customer_id", "amount", "ts"], "flat_format":REDACTED, "metrics_per_partition": REDACTED metrics=REDACTED |
|_|_|_|
| 1_| 0_|_CooperativeExec metrics=REDACTED_|
|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["customer_id", "name", "tier", "ts"], "flat_format":REDACTED, "metrics_per_partition": REDACTED metrics=REDACTED_|
|_|_|_|
|_|_| Total rows: REDACTED_|
+-+-+-+

-- Verify correctness
-- SQLNESS SORT_RESULT
SELECT o."id", o.amount, c."name", c.tier
FROM (SELECT "id", customer_id as cid, amount, ts FROM orders) o
JOIN customers c ON o.cid = c.customer_id
WHERE c.tier IN ('gold', 'silver');

+----+--------+-------+--------+
+----+--------+-------+--------+
+----+--------+-------+--------+
| 1  | 100.0  | Alice | gold   |
| 2  | 200.0  | Bob   | silver |
| 3  | 150.0  | Alice | gold   |
| id | amount | name  | tier   |

DROP TABLE orders;

Affected Rows: 0

DROP TABLE customers;

Affected Rows: 0

