slope_corner_cases.sql

--
-- Test numeric corner cases
--

CREATE TEMPORARY TABLE slope_numeric_corners (i serial, x float8);

INSERT INTO slope_numeric_corners (x) 
SELECT x::float8 as x FROM 
        unnest(ARRAY['-inf', '-1', '0', '1', 'inf', 'nan']) x(x)
ORDER BY 1;

CREATE INDEX slope_numeric_corners_x_idx_nulllast ON slope_numeric_corners (x NULLS LAST);
CREATE INDEX slope_numeric_corners_x_idx_nullfirst ON slope_numeric_corners (x NULLS FIRST);


-- Find displaced values as the minimal set of values that need to be removed
-- from both arrays to make them equal.
-- Exponential time algorithm, acceptable for length 6 arrays.
CREATE OR REPLACE FUNCTION displaced_values(v1 int4[], v2 int4[]) RETURNS int4[]
AS $$
DECLARE
    displaced int4[];
BEGIN
    RETURN (WITH RECURSIVE
    removal(a, b, removed) AS (
        VALUES (v1, v2, '{}'::int4[])
        
        UNION ALL
        
        -- Try removing each candidate element from both arrays
        SELECT 
            array_remove(r.a, c.e),
            array_remove(r.b, c.e),
            r.removed || c.e
        FROM removal r
        CROSS JOIN LATERAL (SELECT DISTINCT unnest(r.a || r.b) AS e) AS c
        WHERE r.a <> r.b                 -- keep going until equal
          AND c.e <> ALL(r.removed)      -- don't re-remove same element
    )
    SELECT removed AS displaced
    FROM removal
    WHERE a = b
    ORDER BY array_length(removed, 1) NULLS FIRST
    LIMIT 1);
END;
$$ LANGUAGE plpgsql IMMUTABLE STRICT;

create temporary table slope_numeric_corners_results (
    seq serial,
    expr text COLLATE "C",
    sort_order text COLLATE "C",
    result text COLLATE "C",
    expected text COLLATE "C",
    nan_values text COLLATE "C",
    plan1_json json,
    plan2_json json
);

DO $$
DECLARE
    r record;
    result text;
    expected text;
    query text;
    agg_query text;
    plan_query text;
    nan_query text;
    nan_values text;
    plan1_json json;
    plan2_json json;
BEGIN

    SET enable_bitmapscan = off;
    SET enable_indexscan = off;
    SET enable_indexonlyscan = off;
    SET enable_seqscan = off;
    SET enable_sort = off;
    FOR r IN
        SELECT replace(s, 'a', a) as expr, sort_order 
        FROM 
        unnest(ARRAY['ASC', 'DESC']) WITH ORDINALITY AS so(sort_order, so_i),
        unnest(ARRAY[
        'x+a', 'x-a', 'x*a', 'x/a',
        'a+x', 'a-x', 'a*x', '-x'
        ]) WITH ORDINALITY AS s(s, s_i),
        unnest(ARRAY['''inf''::float8', '''-inf''::float8', '1::float8']) WITH ORDINALITY AS a(a, a_i)
        ORDER BY s_i, so_i
    LOOP
        query := 'SELECT *, ' || r.expr || ' as f FROM slope_numeric_corners ORDER BY f ' || r.sort_order;
        nan_query := 'SELECT array_agg(i)::text FROM slope_numeric_corners 
        WHERE ' || r.expr || ' = ''nan''::float8 AND x != ''nan''::float8';
        agg_query := 'SELECT array_agg(i)::text FROM (' || query || ') tmp';
        plan_query := 'EXPLAIN (FORMAT JSON) ' || query;
        -- slope optimization disabled
        SET enable_seqscan = on;
        SET enable_indexscan = off;
        SET enable_sort = on;
        EXECUTE agg_query into expected;
        EXECUTE nan_query into nan_values;
        EXECUTE plan_query into plan1_json;
        -- slope optimization enabled
        SET enable_seqscan = off;
        SET enable_indexscan = on;
        SET enable_sort = off;
        EXECUTE agg_query into result;
        EXECUTE plan_query into plan2_json;
        INSERT INTO slope_numeric_corners_results 
        (expr, sort_order, result, expected, nan_values, plan1_json, plan2_json) 
        VALUES (r.expr, r.sort_order, result, expected, nan_values, plan1_json, plan2_json);
    END LOOP;
END;
$$;
SELECT expr, sort_order as order, expected, result,
nan_values, displaced_values(result::int4[], expected::int4[]) as displaced
-- plan1_json->0->'Plan'->>'Node Type' as plan1,
-- plan2_json->0->'Plan'->>'Node Type' as plan2
FROM slope_numeric_corners_results
ORDER BY result != expected, seq;