Allow MIN()/MAX() aggregates on arrays and composite types

[m3hm3t]

postgres/postgres@a0f1fce

postgres/postgres@2d24fd9

[m3hm3t]

[local] citus@citus:9700-96113=# CREATE SCHEMA pg18_minmax;
SET search_path TO pg18_minmax;
CREATE SCHEMA
Time: 14.576 ms
SET
Time: 0.248 ms

[local] citus@citus:9700-96113=# 

[local] citus@citus:9700-96113=# CREATE TABLE sales_data (
    product_id int,
    product text,
    monthly_sales int[]
);
CREATE TABLE
Time: 1.950 ms

[local] citus@citus:9700-96113=# SELECT create_distributed_table('sales_data', 'product_id');
┌──────────────────────────┐
│ create_distributed_table │
├──────────────────────────┤
│                          │
└──────────────────────────┘
(1 row)

Time: 17.485 ms

[local] citus@citus:9700-96113=# INSERT INTO sales_data VALUES
    (1, 'Laptop',   ARRAY[45, 52, 38]),
    (2, 'Mouse',    ARRAY[67, 71, 58]),
    (3, 'Keyboard', ARRAY[23, 28, 15]);
INSERT 0 3
Time: 4.965 ms

[local] citus@citus:9700-96113=# SELECT
    MIN(monthly_sales) AS min_sales_pattern,
    MAX(monthly_sales) AS max_sales_pattern
FROM sales_data;
ERROR:  no matching oid for function: min
Time: 0.633 ms

citus/src/backend/distributed/planner/multi_logical_optimizer.c

Lines 3604 to 3605 in cec2ab8

	ereport(ERROR, (errmsg("no matching oid for function: %s", functionName)));
	}

[m3hm3t]

citus:

CREATE TYPE product_rating AS (
    average_score DECIMAL(3,2),
    review_count INTEGER
);
CREATE TABLE product_ratings (
    id int,
    rating product_rating
);
SELECT create_distributed_table('product_ratings', 'id');
 create_distributed_table
---------------------------------------------------------------------

(1 row)

INSERT INTO product_ratings VALUES
    (1, ROW(4.5, 120)::product_rating),
    (2, ROW(4.2,  89)::product_rating),
    (3, ROW(4.8, 156)::product_rating);
SELECT
    MIN(rating) AS lowest_rating,
    MAX(rating) AS highest_rating
FROM product_ratings;
ERROR:  input of anonymous composite types is not implemented
SET citus.explain_all_tasks TO on;
EXPLAIN (COSTS FALSE, VERBOSE)
SELECT
    MIN(rating) AS lowest_rating,
    MAX(rating) AS highest_rating
FROM product_ratings;
                                                                         QUERY PLAN
---------------------------------------------------------------------
 Aggregate
   Output: min(remote_scan.lowest_rating), max(remote_scan.highest_rating)
   ->  Custom Scan (Citus Adaptive)
         Output: remote_scan.lowest_rating, remote_scan.highest_rating
         Task Count: 4
         Tasks Shown: All
         ->  Task
               Query: SELECT min(rating) AS lowest_rating, max(rating) AS highest_rating FROM pg18_minmax.product_ratings_102050 product_ratings WHERE true
               Node: host=localhost port=xxxxx dbname=regression
               ->  Aggregate
                     Output: min(rating), max(rating)
                     ->  Seq Scan on pg18_minmax.product_ratings_102050 product_ratings
                           Output: id, rating
         ->  Task
               Query: SELECT min(rating) AS lowest_rating, max(rating) AS highest_rating FROM pg18_minmax.product_ratings_102051 product_ratings WHERE true
               Node: host=localhost port=xxxxx dbname=regression
               ->  Aggregate
                     Output: min(rating), max(rating)
                     ->  Seq Scan on pg18_minmax.product_ratings_102051 product_ratings
                           Output: id, rating
         ->  Task
               Query: SELECT min(rating) AS lowest_rating, max(rating) AS highest_rating FROM pg18_minmax.product_ratings_102052 product_ratings WHERE true
               Node: host=localhost port=xxxxx dbname=regression
               ->  Aggregate
                     Output: min(rating), max(rating)
                     ->  Seq Scan on pg18_minmax.product_ratings_102052 product_ratings
                           Output: id, rating
         ->  Task
               Query: SELECT min(rating) AS lowest_rating, max(rating) AS highest_rating FROM pg18_minmax.product_ratings_102053 product_ratings WHERE true
               Node: host=localhost port=xxxxx dbname=regression
               ->  Aggregate
                     Output: min(rating), max(rating)
                     ->  Seq Scan on pg18_minmax.product_ratings_102053 product_ratings
                           Output: id, rating
(34 rows)

WITH
min_row AS (SELECT rating FROM product_ratings ORDER BY rating LIMIT 1),
max_row AS (SELECT rating FROM product_ratings ORDER BY rating DESC LIMIT 1)
SELECT (min_row.rating), (max_row.rating)
FROM min_row, max_row;
  rating   |   rating
---------------------------------------------------------------------
 (4.20,89) | (4.80,156)
(1 row)

That coordinator agg stage only works if the coordinator receives the shard values with enough type descriptor metadata to parse and compare them. In the current multi-shard path, the values arrive as anonymous record/composite (typmod/tupledesc isn’t preserved), so the coordinator can’t interpret them and we hit:

ERROR: input of anonymous composite types is not implemented.

[m3hm3t]

vanilla pg18

➜  citusdata docker exec -it pg18 psql -U postgres -d app
psql (18.1 (Debian 18.1-1.pgdg13+2))
Type "help" for help.

app=#
app=#
app=# -- Create a composite type
CREATE TYPE product_rating AS (
    average_score DECIMAL(3,2),
    review_count INTEGER
);
-- Use MIN/MAX with the composite type
SELECT
    MIN(rating) AS lowest_rating,
    MAX(rating) AS highest_rating
FROM (VALUES
    (ROW(4.5, 120)::product_rating),
    (ROW(4.2, 89)::product_rating),
    (ROW(4.8, 156)::product_rating)
) AS ratings(rating);
 lowest_rating | highest_rating
---------------+----------------
 (4.20,89)     | (4.80,156)
(1 row)