fix: selecting empty columns returns primary keys

README.md

@@ -109,7 +109,7 @@ values ('832bd278-dac7-4bef-96be-e21c8a0023c4', 'public.notes', '{}', '{"role":
 `selected_columns` behaviour:
 - `NULL` (default) — all columns are returned
 - `array['col1', 'col2']` — only the listed columns are returned (plus primary keys)
-- `'{}'` (empty array) — raises an error; use `NULL` to capture all columns
+- `'{}'` (empty array) — only the primary key columns are returned; use `NULL` to capture all columns
 
 
 ### Reading WAL

bin/installcheck

@@ -41,7 +41,7 @@ REGRESS="${PGXS}/../test/regress/pg_regress"
 TESTS=$(ls ${TESTDIR}/sql | sed -e 's/\..*$//' | sort )
 
 # Execute the test fixtures
-psql -v ON_ERROR_STOP=1 -f sql/setup.sql -f sql/walrus--0.1.sql -f sql/walrus_migration_0001*.sql -f sql/walrus_migration_0002*.sql -f sql/walrus_migration_0003*.sql -f sql/walrus_migration_0004*.sql -f sql/walrus_migration_0005*.sql -f sql/walrus_migration_0006*.sql -f sql/walrus_migration_0007*.sql -f sql/walrus_migration_0008*.sql -f sql/walrus_migration_0009*.sql -f sql/walrus_migration_0010*.sql -f sql/walrus_migration_0011*.sql -f sql/walrus_migration_0012*.sql -f sql/walrus_migration_0013*.sql -f sql/walrus_migration_0014*.sql -f sql/walrus_migration_0015*.sql -f sql/walrus_migration_0016*.sql -f test/fixtures.sql -d contrib_regression
+psql -v ON_ERROR_STOP=1 -f sql/setup.sql -f sql/walrus--0.1.sql -f sql/walrus_migration_0001*.sql -f sql/walrus_migration_0002*.sql -f sql/walrus_migration_0003*.sql -f sql/walrus_migration_0004*.sql -f sql/walrus_migration_0005*.sql -f sql/walrus_migration_0006*.sql -f sql/walrus_migration_0007*.sql -f sql/walrus_migration_0008*.sql -f sql/walrus_migration_0009*.sql -f sql/walrus_migration_0010*.sql -f sql/walrus_migration_0011*.sql -f sql/walrus_migration_0012*.sql -f sql/walrus_migration_0013*.sql -f sql/walrus_migration_0014*.sql -f sql/walrus_migration_0015*.sql -f sql/walrus_migration_0016*.sql -f sql/walrus_migration_0017*.sql -f test/fixtures.sql -d contrib_regression
 
 # Run tests
 ${REGRESS} --use-existing --dbname=contrib_regression --inputdir=${TESTDIR} ${TESTS}

sql/walrus_migration_0017_empty_select_primary_keys.sql

@@ -0,0 +1,177 @@
+-- Allow an empty selected_columns array to mean "primary keys only" instead of
+-- raising an error. apply_rls already emits just the primary key columns for an
+-- empty array (a column matches when it is in selected_columns OR it is a pkey),
+-- so the only work here is to stop rejecting '{}' and to stop the normalization
+-- step from collapsing '{}' back to NULL (which would mean "all columns").
+create or replace function realtime.subscription_check_filters()
+    returns trigger
+    language plpgsql
+as $$
+declare
+    col_names text[] = coalesce(
+            array_agg(c.column_name order by c.ordinal_position),
+            '{}'::text[]
+        )
+        from
+            information_schema.columns c
+        where
+            format('%I.%I', c.table_schema, c.table_name)::regclass = new.entity
+            and pg_catalog.has_column_privilege(
+                (new.claims ->> 'role'),
+                format('%I.%I', c.table_schema, c.table_name)::regclass,
+                c.column_name,
+                'SELECT'
+            );
+    filter realtime.user_defined_filter;
+    filter_v2 realtime.user_defined_filter_v2;
+    col_type regtype;
+    in_val jsonb;
+    selected_col text;
+begin
+    -- Legacy 3-field filters: only the original operators are evaluable by the
+    -- legacy check_equality_op path. Reject the new operators here so they
+    -- cannot be stored on this column and later crash apply_rls with UNKNOWN OP.
+    for filter in select * from unnest(new.filters) loop
+        if not filter.column_name = any(col_names) then
+            raise exception 'invalid column for filter %', filter.column_name;
+        end if;
+
+        col_type = (
+            select atttypid::regtype
+            from pg_catalog.pg_attribute
+            where attrelid = new.entity
+                  and attname = filter.column_name
+        );
+        if col_type is null then
+            raise exception 'failed to lookup type for column %', filter.column_name;
+        end if;
+
+        if filter.op in (
+            'like'::realtime.equality_op, 'ilike'::realtime.equality_op,
+            'is'::realtime.equality_op, 'match'::realtime.equality_op,
+            'imatch'::realtime.equality_op, 'isdistinct'::realtime.equality_op
+        ) then
+            raise exception 'operator % is only supported on the filters_v2 column', filter.op::text;
+        end if;
+
+        if filter.op = 'in'::realtime.equality_op then
+            in_val = realtime.cast(filter.value, (col_type::text || '[]')::regtype);
+            if coalesce(jsonb_array_length(in_val), 0) > 100 then
+                raise exception 'too many values for `in` filter. Maximum 100';
+            end if;
+        else
+            -- raises an exception if value is not coercable to type
+            perform realtime.cast(filter.value, col_type);
+        end if;
+    end loop;
+
+    -- v2 filters: full validation including the new operators.
+    for filter_v2 in select * from unnest(new.filters_v2) loop
+        if not filter_v2.column_name = any(col_names) then
+            raise exception 'invalid column for filter %', filter_v2.column_name;
+        end if;
+
+        col_type = (
+            select atttypid::regtype
+            from pg_catalog.pg_attribute
+            where attrelid = new.entity
+                  and attname = filter_v2.column_name
+        );
+        if col_type is null then
+            raise exception 'failed to lookup type for column %', filter_v2.column_name;
+        end if;
+
+        if filter_v2.op = 'in'::realtime.equality_op then
+            in_val = realtime.cast(filter_v2.value, (col_type::text || '[]')::regtype);
+            if coalesce(jsonb_array_length(in_val), 0) > 100 then
+                raise exception 'too many values for `in` filter. Maximum 100';
+            end if;
+        elsif filter_v2.op = 'is'::realtime.equality_op then
+            -- `is` requires a keyword RHS rather than a typed literal
+            if filter_v2.value not in ('null', 'true', 'false', 'unknown') then
+                raise exception 'invalid value for is filter: must be null, true, false, or unknown';
+            end if;
+            -- IS NULL works for any type, but IS TRUE/FALSE/UNKNOWN require a
+            -- boolean operand. Reject the non-null keywords on non-boolean
+            -- columns here so they don't abort apply_rls at WAL time.
+            if filter_v2.value <> 'null' and col_type <> 'boolean'::regtype then
+                raise exception 'is % filter requires a boolean column, got %', filter_v2.value, col_type::text;
+            end if;
+        elsif filter_v2.op in ('like'::realtime.equality_op, 'ilike'::realtime.equality_op) then
+            -- like/ilike apply the text pattern operator (~~); reject column
+            -- types that have no such operator instead of failing at WAL time
+            if not exists (
+                select 1 from pg_catalog.pg_operator
+                where oprname = '~~' and oprleft = col_type
+            ) then
+                raise exception 'operator % requires a text-compatible column type, got %', filter_v2.op::text, col_type::text;
+            end if;
+        elsif filter_v2.op in ('match'::realtime.equality_op, 'imatch'::realtime.equality_op) then
+            -- match/imatch apply the regex operators ~ / ~*; reject column types
+            -- that have no such operator (e.g. integer) instead of failing at WAL
+            -- time, mirroring the like/ilike guard above.
+            if not exists (
+                select 1 from pg_catalog.pg_operator
+                where oprname = case when filter_v2.op = 'imatch'::realtime.equality_op then '~*' else '~' end
+                  and oprleft = col_type
+                  and oprright = col_type
+                  and oprresult = 'boolean'::regtype
+            ) then
+                raise exception 'operator % requires a text-compatible column type, got %', filter_v2.op::text, col_type::text;
+            end if;
+            -- validate the regex eagerly so a bad pattern is rejected here, not
+            -- inside apply_rls where it would abort the WAL stream for the entity
+            begin
+                perform '' ~ filter_v2.value;
+            exception when others then
+                raise exception 'invalid regular expression for % filter: %', filter_v2.op::text, sqlerrm;
+            end;
+        else
+            -- eq/neq/lt/lte/gt/gte/isdistinct: value must be coercable to the type
+            perform realtime.cast(filter_v2.value, col_type);
+        end if;
+    end loop;
+
+    -- Reject arrays with NULL elements which bypass column validation
+    if new.selected_columns is not null and array_position(new.selected_columns, null::text) is not null then
+        raise exception 'selected_columns cannot contain null values.';
+    end if;
+
+    -- Validate that selected_columns reference columns the role can SELECT
+    if new.selected_columns is not null then
+        for selected_col in select * from unnest(new.selected_columns) loop
+            if not selected_col = any(col_names) then
+                raise exception 'invalid column for select %', selected_col;
+            end if;
+        end loop;
+    end if;
+
+    -- Apply consistent order to filters so the unique constraint can't be
+    -- tricked by a different filter order. negate is part of the v2 sort key.
+    new.filters = coalesce(
+        array_agg(f order by f.column_name, f.op, f.value),
+        '{}'
+    ) from unnest(new.filters) f;
+
+    new.filters_v2 = coalesce(
+        array_agg(f order by f.column_name, f.op, f.value, f.negate),
+        '{}'
+    ) from unnest(new.filters_v2) f;
+
+    -- Normalize selected_columns order so ARRAY['a','b'] and ARRAY['b','a'] are
+    -- treated as the same subscription group in apply_rls. Preserve an empty
+    -- array as '{}' ("primary keys only") so it stays distinct from NULL ("all
+    -- columns"); array_agg over an empty set would otherwise collapse it to NULL.
+    if new.selected_columns is not null then
+        new.selected_columns = coalesce(
+            (
+                select array_agg(c order by c)
+                from unnest(new.selected_columns) c
+            ),
+            '{}'::text[]
+        );
+    end if;
+
+    return new;
+end;
+$$;

test/expected/test_select_columns_empty.out

@@ -1,9 +1,17 @@
 /*
-Tests that subscribing with an empty selected_columns raises an exception
+Tests that subscribing with an empty selected_columns defaults to returning
+only the primary key columns (rather than raising an exception)
 */
+select 1 from pg_create_logical_replication_slot('realtime', 'wal2json', false);
+ ?column? 
+----------
+        1
+(1 row)
+
 create table public.notes(
     id int primary key,
-    body text
+    body text,
+    extra text
 );
 insert into realtime.subscription(subscription_id, entity, claims, selected_columns)
 select
@@ -15,7 +23,53 @@ select
         'sub', seed_uuid(1)::text
     ),
     '{}'::text[];
-ERROR:  selected_columns cannot be empty. Remove the select parameter to capture all columns.
-CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 130 at RAISE
+-- The empty array is preserved (not collapsed to NULL) so it stays distinct
+-- from "all columns"
+select selected_columns from realtime.subscription;
+ selected_columns 
+------------------
+ {}
+(1 row)
+
+select clear_wal();
+ clear_wal 
+-----------
+
+(1 row)
+
+insert into public.notes(id, body, extra) values (1, 'take out trash', 'extra data');
+-- Only the primary key (id) is present in record and columns
+select
+    rec,
+    is_rls_enabled,
+    subscription_ids,
+    errors
+from
+   walrus;
+                        rec                         | is_rls_enabled |            subscription_ids            | errors 
+----------------------------------------------------+----------------+----------------------------------------+--------
+ {                                                 +| f              | {f4539ebe-c779-5788-bbc1-2421ffaa8954} | {}
+     "type": "INSERT",                             +|                |                                        | 
+     "table": "notes",                             +|                |                                        | 
+     "record": {                                   +|                |                                        | 
+         "id": 1                                   +|                |                                        | 
+     },                                            +|                |                                        | 
+     "schema": "public",                           +|                |                                        | 
+     "columns": [                                  +|                |                                        | 
+         {                                         +|                |                                        | 
+             "name": "id",                         +|                |                                        | 
+             "type": "int4"                        +|                |                                        | 
+         }                                         +|                |                                        | 
+     ],                                            +|                |                                        | 
+     "commit_timestamp": "2000-01-01T08:01:01.000Z"+|                |                                        | 
+ }                                                  |                |                                        | 
+(1 row)
+
 drop table public.notes;
+select pg_drop_replication_slot('realtime');
+ pg_drop_replication_slot 
+--------------------------
+
+(1 row)
+
 truncate table realtime.subscription;

test/expected/test_select_columns_invalid.out

@@ -16,6 +16,6 @@ select
     ),
     array['nonexistent_column'];
 ERROR:  invalid column for select nonexistent_column
-CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 142 at RAISE
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 136 at RAISE
 drop table public.notes;
 truncate table realtime.subscription;

test/expected/test_select_columns_null_element.out

@@ -16,6 +16,6 @@ select
     ),
     array[null]::text[];
 ERROR:  selected_columns cannot contain null values.
-CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 135 at RAISE
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 129 at RAISE
 drop table public.notes;
 truncate table realtime.subscription;

test/sql/test_select_columns_empty.sql

@@ -1,10 +1,14 @@
 /*
-Tests that subscribing with an empty selected_columns raises an exception
+Tests that subscribing with an empty selected_columns defaults to returning
+only the primary key columns (rather than raising an exception)
 */
 
+select 1 from pg_create_logical_replication_slot('realtime', 'wal2json', false);
+
 create table public.notes(
     id int primary key,
-    body text
+    body text,
+    extra text
 );
 
 insert into realtime.subscription(subscription_id, entity, claims, selected_columns)
@@ -18,6 +22,23 @@ select
     ),
     '{}'::text[];
 
+-- The empty array is preserved (not collapsed to NULL) so it stays distinct
+-- from "all columns"
+select selected_columns from realtime.subscription;
+
+select clear_wal();
+insert into public.notes(id, body, extra) values (1, 'take out trash', 'extra data');
+
+-- Only the primary key (id) is present in record and columns
+select
+    rec,
+    is_rls_enabled,
+    subscription_ids,
+    errors
+from
+   walrus;
+
 
 drop table public.notes;
+select pg_drop_replication_slot('realtime');
 truncate table realtime.subscription;