-- Test ORDER BY collation safety for pushdown
-- Ensures that ParadeDB correctly refuses to push down ORDER BY when a non-C collation
-- is in use, since Tantivy sorts by raw byte order (equivalent to C/POSIX/C.UTF-8 only).
--
-- ASSUMPTION: This test file assumes the database's default collation (datcollate) is
-- byte-ordered (one of: C, POSIX, C.UTF-8, C.utf8, POSIX.UTF-8, POSIX.utf8). CI runs
-- against C.UTF-8 so these tests pass there. On a local dev box whose `initdb` locale
-- defaults to a non byte-ordered collation (e.g. en_US.UTF-8), Sections 7 and 8 will
-- fail because the "default collation -> pushdown safe" expectations no longer hold.
--
-- Tests cover:
-- 1. TopK (base scan ORDER BY ... LIMIT) with C vs non-C collation
-- 2. Aggregate scan grouping and ordering with C vs non-C collation
-- 3. Sorted index scan (sort_by) with C vs non-C collation
-- 4. Non-text columns (integers) are unaffected by collation checks
-- 5. Result correctness
-- 6. JoinScan ORDER BY collation check
-- 7. Aggregate-on-Join TopK collation check
-- 8. Default database collation (no explicit COLLATE) -> pushdown safe when default is C-like
-- 9. Mixed ORDER BY with safe and unsafe columns

\i common/common_setup.sql

-- =============================================================================
-- SETUP
-- =============================================================================

-- Create an ICU collation for testing non-C ordering (always available in PG15+)
CREATE COLLATION IF NOT EXISTS test_icu (
    provider = icu,
    locale = 'en-US'
);

CREATE COLLATION IF NOT EXISTS test_icu_case_insensitive (
    provider = icu,
    locale = 'und-u-ks-level2',
    deterministic = false
);

DROP TABLE IF EXISTS collation_test CASCADE;

CREATE TABLE collation_test (
    id SERIAL PRIMARY KEY,
    name_c TEXT COLLATE "C",
    name_icu TEXT COLLATE "test_icu",
    name_case_insensitive TEXT COLLATE "test_icu_case_insensitive",
    name_default TEXT,
    priority INTEGER
);

INSERT INTO
    collation_test (
        name_c,
        name_icu,
        name_case_insensitive,
        name_default,
        priority
    )
VALUES ('apple', 'apple', 'Electronics', 'apple', 10),
    (
        'Banana',
        'Banana',
        'electronics',
        'Banana',
        20
    ),
    (
        'cherry',
        'cherry',
        'Clothing',
        'cherry',
        30
    ),
    ('Date', 'Date', 'Home', 'Date', 40),
    (
        'elderberry',
        'elderberry',
        'home',
        'elderberry',
        50
    );

CREATE INDEX collation_test_idx ON collation_test USING bm25 (
    id,
    name_c,
    name_icu,
    name_case_insensitive,
    name_default,
    priority
)
WITH (
        key_field = 'id',
        text_fields = '{"name_c": {"indexed": true, "fast": true}, "name_icu": {"indexed": true, "fast": true}, "name_case_insensitive": {"indexed": true, "fast": true}, "name_default": {"indexed": true, "fast": true}}',
        numeric_fields = '{"priority": {"indexed": true, "fast": true}}'
    );

ANALYZE collation_test;

-- =============================================================================
-- SECTION 1: TopK (Base Scan ORDER BY ... LIMIT)
-- =============================================================================


\echo '=== SECTION 1: TopK Base Scan ORDER BY ... LIMIT ==='

\echo 'Test 1.1: ORDER BY C-collation text column -> TopK pushdown (no Sort node)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, name_c FROM collation_test
WHERE id @@@ paradedb.all()
ORDER BY name_c
LIMIT 5;

\echo 'Test 1.2: ORDER BY ICU-collation text column -> Sort node expected'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, name_icu FROM collation_test
WHERE id @@@ paradedb.all()
ORDER BY name_icu
LIMIT 5;

\echo 'Test 1.3: ORDER BY integer column -> TopK pushdown (collation is InvalidOid for non-text)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, priority FROM collation_test
WHERE id @@@ paradedb.all()
ORDER BY priority
LIMIT 5;

\echo 'Test 1.4: ORDER BY with explicit COLLATE "C" on ICU column -> TopK pushdown'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, name_icu FROM collation_test
WHERE id @@@ paradedb.all()
ORDER BY name_icu COLLATE "C"
LIMIT 5;

\echo 'Test 1.5: ORDER BY with explicit COLLATE "test_icu" on C column -> Sort node expected'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, name_c FROM collation_test
WHERE id @@@ paradedb.all()
ORDER BY name_c COLLATE "test_icu"
LIMIT 5;

-- =============================================================================
-- SECTION 2: Aggregate Scan ORDER BY
-- =============================================================================

SET paradedb.enable_aggregate_custom_scan TO on;


\echo '=== SECTION 2: Aggregate Scan ORDER BY ==='

\echo 'Test 2.1: GROUP BY + ORDER BY C-collation text -> aggregate pushdown (no Sort node)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT name_c, COUNT(*) FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_c
ORDER BY name_c;

\echo 'Test 2.2: deterministic ICU GROUP BY stays pushed down; ORDER BY stays above'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT name_icu, COUNT(*) FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_icu
ORDER BY name_icu;

-- Deterministic ICU equality is compatible with byte-based grouping. Keep
-- pdb.agg() on AggregateScan rather than falling back to PostgreSQL, which
-- cannot execute its placeholder.
SELECT name_icu, pdb.agg('{"value_count": {"field": "id"}}'::jsonb)
FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_icu
ORDER BY name_icu;

\echo 'Test 2.3: GROUP BY + ORDER BY integer -> aggregate pushdown (collation is InvalidOid)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT priority, COUNT(*) FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY priority
ORDER BY priority;

\echo 'Test 2.4: safe GROUP BY + unsafe ORDER BY -> Sort above AggregateScan'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT name_c, COUNT(*) FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_c
ORDER BY name_c COLLATE "test_icu";

\echo 'Test 2.5: nondeterministic ICU GROUP BY -> PostgreSQL combines equivalent values'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT lower(name_case_insensitive) AS name, COUNT(*)
FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_case_insensitive
ORDER BY name;

SELECT lower(name_case_insensitive) AS name, COUNT(*)
FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_case_insensitive
ORDER BY name;

\echo 'Test 2.6: GROUPING SETS with a deterministic key -> AggregateScan declined'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT name_c, COUNT(*)
FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY GROUPING SETS ((name_c), ());

SELECT COUNT(*) AS group_count
FROM (
    SELECT name_c
    FROM collation_test
    WHERE id @@@ paradedb.all()
    GROUP BY GROUPING SETS ((name_c), ())
) AS grouped;

SELECT name_c, pdb.agg('{"value_count": {"field": "id"}}'::jsonb)
FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY GROUPING SETS ((name_c), ());

\echo 'Test 2.7: constant-equality GROUP BY key with no pathkeys -> AggregateScan declined'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT name_c, COUNT(*)
FROM collation_test
WHERE name_c = 'apple' AND id @@@ paradedb.all()
GROUP BY name_c;

SELECT name_c, pdb.agg('{"value_count": {"field": "id"}}'::jsonb)
FROM collation_test
WHERE name_c = 'apple' AND id @@@ paradedb.all()
GROUP BY name_c;

\echo 'Test 2.8: hash aggregation with a nondeterministic collation -> AggregateScan declined'
SET enable_sort TO off;

EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT name_case_insensitive, COUNT(*)
FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_case_insensitive;

SELECT COUNT(*) AS group_count
FROM (
    SELECT name_case_insensitive
    FROM collation_test
    WHERE id @@@ paradedb.all()
    GROUP BY name_case_insensitive
) AS grouped;

RESET enable_sort;

\echo 'Test 2.9: pdb.agg() with a nondeterministic grouping collation warns before failing'
SELECT lower(name_case_insensitive) AS name,
       pdb.agg('{"value_count": {"field": "id"}}'::jsonb)
FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_case_insensitive;

\echo 'Test 2.10: mixed safe and unsafe GROUP BY keys -> AggregateScan declined'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT name_c, name_case_insensitive, COUNT(*) FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_c, name_case_insensitive
ORDER BY name_c, name_case_insensitive;

RESET paradedb.enable_aggregate_custom_scan;

-- =============================================================================
-- SECTION 3: Sorted Index Scan (sort_by pathkey matching)
-- =============================================================================

\echo '=== SECTION 3: Sorted Index Scan (sort_by) ==='


DROP TABLE IF EXISTS collation_sortby_test CASCADE;

CREATE TABLE collation_sortby_test (
    id SERIAL PRIMARY KEY,
    city TEXT COLLATE "C",
    population INTEGER
);

INSERT INTO
    collation_sortby_test (city, population)
VALUES ('berlin', 3600000),
    ('Amsterdam', 900000),
    ('chicago', 2700000),
    ('Delhi', 32000000),
    ('edmonton', 1000000);

CREATE INDEX collation_sortby_test_idx ON collation_sortby_test USING bm25 (id, city, population)
WITH (
        key_field = 'id',
        text_fields = '{"city": {"indexed": true, "fast": true}}',
        numeric_fields = '{"population": {"indexed": true, "fast": true}}',
        sort_by = 'city ASC NULLS FIRST'
    );

ANALYZE collation_sortby_test;

\echo 'Test 3.1: ORDER BY city (C collation, matches sort_by) -> no Sort node'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, city FROM collation_sortby_test
WHERE id @@@ paradedb.all()
ORDER BY city ASC NULLS FIRST;

\echo 'Test 3.2: ORDER BY city with explicit COLLATE "test_icu" -> Sort node expected'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, city FROM collation_sortby_test
WHERE id @@@ paradedb.all()
ORDER BY city COLLATE "test_icu" ASC NULLS FIRST;

-- =============================================================================
-- SECTION 4: Result Correctness
-- =============================================================================


\echo '=== SECTION 4: Result Correctness ==='

\echo 'Test 4.1: C collation ORDER BY -> byte order (uppercase before lowercase in ASCII)'
SELECT id, city FROM collation_sortby_test
WHERE id @@@ paradedb.all()
ORDER BY city ASC NULLS FIRST;

\echo 'Test 4.2: ICU collation ORDER BY -> linguistic order (case-insensitive from Postgres Sort)'
SELECT id, city FROM collation_sortby_test
WHERE id @@@ paradedb.all()
ORDER BY city COLLATE "test_icu" ASC NULLS FIRST;

-- =============================================================================
-- SECTION 5: JoinScan ORDER BY collation check
-- =============================================================================

\echo '=== SECTION 5: JoinScan ORDER BY ==='

SET paradedb.enable_join_custom_scan = on;

DROP TABLE IF EXISTS collation_join_products CASCADE;

DROP TABLE IF EXISTS collation_join_suppliers CASCADE;

CREATE TABLE collation_join_products (
    id INTEGER PRIMARY KEY,
    name_c TEXT COLLATE "C" NOT NULL,
    name_icu TEXT COLLATE "test_icu" NOT NULL,
    description TEXT COLLATE "C"
);

CREATE TABLE collation_join_suppliers (
    id INTEGER PRIMARY KEY,
    product_id INTEGER NOT NULL,
    supplier_name TEXT COLLATE "C" NOT NULL
);

INSERT INTO
    collation_join_products (
        id,
        name_c,
        name_icu,
        description
    )
VALUES (
        1,
        'Wireless Mouse',
        'Wireless Mouse',
        'ergonomic wireless mouse'
    ),
    (
        2,
        'USB Cable',
        'USB Cable',
        'high-speed cable'
    ),
    (
        3,
        'Keyboard',
        'Keyboard',
        'mechanical keyboard wireless'
    ),
    (
        4,
        'Monitor Stand',
        'Monitor Stand',
        'adjustable monitor stand'
    ),
    (
        5,
        'Webcam',
        'Webcam',
        'HD webcam wireless'
    );

INSERT INTO
    collation_join_suppliers (id, product_id, supplier_name)
VALUES (1, 1, 'TechCorp'),
    (2, 2, 'CableCo'),
    (3, 3, 'TechCorp'),
    (4, 4, 'FurniPro'),
    (5, 5, 'TechCorp');

CREATE INDEX collation_join_products_idx ON collation_join_products USING bm25 (
    id,
    name_c,
    name_icu,
    description
)
WITH (
        key_field = 'id',
        text_fields = '{"name_c": {"fast": true}, "name_icu": {"fast": true}, "description": {}}'
    );

CREATE INDEX collation_join_suppliers_idx ON collation_join_suppliers USING bm25 (id, product_id, supplier_name)
WITH (
        key_field = 'id',
        text_fields = '{"supplier_name": {"fast": true}}',
        numeric_fields = '{"product_id": {"fast": true}}'
    );

ANALYZE collation_join_products;

ANALYZE collation_join_suppliers;

\echo 'Test 5.1: JoinScan ORDER BY C-collation column -> JoinScan used'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT p.name_c, s.supplier_name
FROM collation_join_products p
JOIN collation_join_suppliers s ON p.id = s.product_id
WHERE p.description @@@ 'wireless'
ORDER BY p.name_c
LIMIT 5;

\echo 'Test 5.2: JoinScan ORDER BY ICU-collation column -> JoinScan declined (Sort node expected)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT p.name_icu, s.supplier_name
FROM collation_join_products p
JOIN collation_join_suppliers s ON p.id = s.product_id
WHERE p.description @@@ 'wireless'
ORDER BY p.name_icu
LIMIT 5;

\echo 'Test 5.3: JoinScan ORDER BY integer -> JoinScan used (collation is InvalidOid)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT p.name_c, s.supplier_name
FROM collation_join_products p
JOIN collation_join_suppliers s ON p.id = s.product_id
WHERE p.description @@@ 'wireless'
ORDER BY p.id
LIMIT 5;

RESET paradedb.enable_join_custom_scan;

-- =============================================================================
-- SECTION 6: Aggregate-on-Join TopK collation check
-- =============================================================================

\echo '=== SECTION 6: Aggregate-on-Join TopK ==='

SET paradedb.enable_aggregate_custom_scan TO on;

\echo 'Test 6.1: GROUP BY + ORDER BY C-collation text + LIMIT -> aggregate TopK pushdown'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT name_c, COUNT(*) FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_c
ORDER BY name_c
LIMIT 3;

\echo 'Test 6.2: deterministic ICU GROUP BY + LIMIT -> grouping pushed down, ordering remains above'
SET paradedb.max_term_agg_buckets = 1;
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT name_icu, COUNT(*) FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_icu
ORDER BY name_icu
LIMIT 3;
RESET paradedb.max_term_agg_buckets;

\echo 'Test 6.3: GROUP BY + ORDER BY integer + LIMIT -> aggregate TopK pushdown'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT priority, COUNT(*) FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY priority
ORDER BY priority
LIMIT 3;

RESET paradedb.enable_aggregate_custom_scan;

-- =============================================================================
-- SECTION 7: Default database collation (no explicit COLLATE)
-- The database default is C.UTF-8 (safe for byte-order pushdown)
-- =============================================================================


\echo '=== SECTION 7: Default Database Collation (C.UTF-8) ==='

\echo 'Test 7.1: ORDER BY default-collation text column -> TopK pushdown (default is C.UTF-8, safe)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, name_default FROM collation_test
WHERE id @@@ paradedb.all()
ORDER BY name_default
LIMIT 5;

SET paradedb.enable_aggregate_custom_scan TO on;

\echo 'Test 7.2: GROUP BY + ORDER BY default-collation text -> aggregate pushdown (no Sort node)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT name_default, COUNT(*) FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_default
ORDER BY name_default;

\echo 'Test 7.3: GROUP BY + ORDER BY default-collation text + LIMIT -> aggregate TopK pushdown'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT name_default, COUNT(*) FROM collation_test
WHERE id @@@ paradedb.all()
GROUP BY name_default
ORDER BY name_default
LIMIT 3;

RESET paradedb.enable_aggregate_custom_scan;

-- =============================================================================
-- SECTION 8: Mixed ORDER BY with safe and unsafe columns
-- =============================================================================


\echo '=== SECTION 8: Mixed ORDER BY (safe + unsafe) ==='

\echo 'Test 8.1: ORDER BY safe_col (C), unsafe_col (ICU) -> Sort node (mixed is unsafe)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, name_c, name_icu FROM collation_test
WHERE id @@@ paradedb.all()
ORDER BY name_c, name_icu
LIMIT 5;

\echo 'Test 8.2: ORDER BY safe_col (integer), unsafe_col (ICU) -> Sort node (mixed is unsafe)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, priority, name_icu FROM collation_test
WHERE id @@@ paradedb.all()
ORDER BY priority, name_icu
LIMIT 5;

\echo 'Test 8.3: ORDER BY safe_col (C), safe_col (default) -> TopK pushdown (both safe)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, name_c, name_default FROM collation_test
WHERE id @@@ paradedb.all()
ORDER BY name_c, name_default
LIMIT 5;

\echo 'Test 8.4: ORDER BY safe_col (integer), safe_col (default) -> TopK pushdown (both safe)'
EXPLAIN (FORMAT TEXT, COSTS OFF, TIMING OFF)
SELECT id, priority, name_default FROM collation_test
WHERE id @@@ paradedb.all()
ORDER BY priority, name_default
LIMIT 5;

-- =============================================================================
-- CLEANUP
-- =============================================================================


DROP TABLE IF EXISTS collation_test CASCADE;

DROP TABLE IF EXISTS collation_sortby_test CASCADE;

DROP TABLE IF EXISTS collation_join_products CASCADE;

DROP TABLE IF EXISTS collation_join_suppliers CASCADE;

DROP COLLATION IF EXISTS test_icu_case_insensitive;

DROP COLLATION IF EXISTS test_icu;

\i common/common_cleanup.sql
