diff --git a/README.md b/README.md
index d2583d5..40f0faf 100644
--- a/README.md
+++ b/README.md
@@ -47,7 +47,8 @@ create type realtime.user_defined_filter as (
 and `realtime.equality_op`s are a subset of [postgrest ops](https://postgrest.org/en/v4.1/api.html#horizontal-filtering-rows). Specifically:
 ```sql
 create type realtime.equality_op as enum(
-    'eq', 'neq', 'lt', 'lte', 'gt', 'gte', 'in'
+    'eq', 'neq', 'lt', 'lte', 'gt', 'gte', 'in',
+    'like', 'ilike', 'is', 'match', 'imatch', 'isdistinct'
 );
 ```
 
@@ -57,6 +58,42 @@ insert into realtime.subscription(subscription_id, entity, filters, claims)
 values ('832bd278-dac7-4bef-96be-e21c8a0023c4', 'public.notes', array[('id', 'eq', '6')], '{"role", "authenticated"}');
 ```
 
+### Extended operators and negation (`filters_v2`)
+
+The operators `like`, `ilike`, `is`, `match` (`~`), `imatch` (`~*`) and `isdistinct`,
+along with negation of any operator, are stored on a separate `filters_v2`
+column whose composite type carries an extra `negate` boolean:
+```sql
+create type realtime.user_defined_filter_v2 as (
+    column_name text,
+    op realtime.equality_op,
+    value text,
+    negate boolean
+);
+```
+This is an expand/contract rollout: the legacy 3-field `filters` column is left
+untouched (so older Realtime server instances keep writing to it during a rolling
+deploy), and the new operators are only accepted on `filters_v2`. `apply_rls`
+evaluates both columns; for any given subscription one of them is always empty.
+
+When `negate` is `true` the operator's result is inverted (e.g. `like` →
+`NOT LIKE`, `in` → `NOT IN`). To subscribe to `public.notes` where `body` does
+**not** match the pattern `%draft%`:
+```sql
+insert into realtime.subscription(subscription_id, entity, filters_v2, claims)
+values (
+    '832bd278-dac7-4bef-96be-e21c8a0023c4',
+    'public.notes',
+    array[('body', 'like', '%draft%', true)]::realtime.user_defined_filter_v2[],
+    '{"role": "authenticated"}'
+);
+```
+
+Operator notes:
+- `is` requires a keyword value: `null`, `true`, `false` or `unknown`. `is true`/`false`/`unknown` are only valid on boolean columns; `is null` works on any type.
+- `like`/`ilike`/`match`/`imatch` require a text-compatible column type.
+- `match`/`imatch` patterns are validated as regular expressions when the subscription is created.
+
 To subscribe to `INSERT`s only on a table named `public.notes` where the `id` is `6` as the `authenticated` role:
 ```sql
 insert into realtime.subscription(subscription_id, entity, filters, claims, action_filter)
diff --git a/bin/installcheck b/bin/installcheck
index 7496065..4855f17 100755
--- a/bin/installcheck
+++ b/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 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 test/fixtures.sql -d contrib_regression
 
 # Run tests
 ${REGRESS} --use-existing --dbname=contrib_regression --inputdir=${TESTDIR} ${TESTS}
diff --git a/sql/walrus_migration_0015_like_ilike_is_not_ops.sql b/sql/walrus_migration_0015_like_ilike_is_not_ops.sql
new file mode 100644
index 0000000..3157c24
--- /dev/null
+++ b/sql/walrus_migration_0015_like_ilike_is_not_ops.sql
@@ -0,0 +1,726 @@
+-- New equality operators (postgrest parity): like, ilike, is, match, imatch,
+-- isdistinct, plus `negate` (postgrest `not.`).
+--
+-- ROLLOUT — EXPAND/CONTRACT. The realtime.subscription.filters column is a
+-- composite-typed array that the Realtime server writes positionally. Changing
+-- the arity of realtime.user_defined_filter in place would break every old
+-- server instance still writing 3-field literals during a rolling deploy (the
+-- ALTER lands while old code is alive -> "cannot cast type record"). So instead
+-- this migration EXPANDS:
+--   * the existing 3-field `filters` column / type / functions are left intact,
+--     so old instances keep inserting successfully;
+--   * a new `filters_v2` column (4-field type carrying `negate`) is added with a
+--     default of '{}', so it is invisible to old writers;
+--   * apply_rls evaluates BOTH columns (one is always '{}' for a given row).
+-- New server instances write filters_v2 once deployed. A later CONTRACT
+-- migration drops the legacy column/type/functions after all old instances are
+-- retired.
+--
+-- Adding enum values is additive on its own, but it also makes the new ops legal
+-- on the legacy 3-field column, where the legacy check_equality_op cannot
+-- evaluate them ("UNKNOWN OP"). The trigger below therefore rejects the new ops
+-- on the legacy `filters` column so they can only be stored on filters_v2.
+-- All DDL below is written to be idempotent so the migration can be re-run or
+-- recover from a partial application without erroring.
+alter type realtime.equality_op add value if not exists 'like';
+alter type realtime.equality_op add value if not exists 'ilike';
+alter type realtime.equality_op add value if not exists 'is';
+alter type realtime.equality_op add value if not exists 'match';
+alter type realtime.equality_op add value if not exists 'imatch';
+alter type realtime.equality_op add value if not exists 'isdistinct';
+
+do $$
+begin
+    if not exists (
+        select 1
+        from pg_type t
+        join pg_namespace n on n.oid = t.typnamespace
+        where n.nspname = 'realtime' and t.typname = 'user_defined_filter_v2'
+    ) then
+        create type realtime.user_defined_filter_v2 as (
+            column_name text,
+            op realtime.equality_op,
+            value text,
+            negate boolean
+        );
+    end if;
+end $$;
+
+
+-- v2 overload: same name, distinct 5-arg signature. It takes no default, so
+-- there is no ambiguity with the existing 4-arg check_equality_op; a 4-arg call
+-- resolves to the original function, a 5-arg call resolves to this one.
+create or replace function realtime.check_equality_op(
+    op realtime.equality_op,
+    type_ regtype,
+    val_1 text,
+    val_2 text,
+    negate boolean
+)
+    returns bool
+    stable  -- uses EXECUTE, so cannot be immutable
+    language plpgsql
+as $$
+declare
+    op_symbol text;
+    res boolean;
+begin
+    -- IS DISTINCT FROM / IS NOT DISTINCT FROM: infix, both sides typed literals
+    if op = 'isdistinct' then
+        execute format(
+            'select %L::%s %s %L::%s',
+            val_1,
+            type_::text,
+            case when negate then 'IS NOT DISTINCT FROM' else 'IS DISTINCT FROM' end,
+            val_2,
+            type_::text
+        ) into res;
+        return res;
+    end if;
+
+    -- IS requires a keyword RHS (NULL, TRUE, FALSE, UNKNOWN), not a typed literal
+    if op = 'is' then
+        if val_2 not in ('null', 'true', 'false', 'unknown') then
+            raise exception 'invalid value for is filter: must be null, true, false, or unknown';
+        end if;
+        execute format(
+            'select %L::%s %s %s',
+            val_1,
+            type_::text,
+            case when negate then 'IS NOT' else 'IS' end,
+            upper(val_2)
+        ) into res;
+        return res;
+    end if;
+
+    op_symbol = case
+        when op = 'eq'    then '='
+        when op = 'neq'   then '!='
+        when op = 'lt'    then '<'
+        when op = 'lte'   then '<='
+        when op = 'gt'    then '>'
+        when op = 'gte'   then '>='
+        when op = 'in'    then '= any'
+        when op = 'like'   then 'LIKE'
+        when op = 'ilike'  then 'ILIKE'
+        when op = 'match'  then '~'
+        when op = 'imatch' then '~*'
+        else null
+    end;
+
+    if op_symbol is null then
+        raise exception 'unsupported equality operator: %', op::text;
+    end if;
+
+    execute format(
+        'select %L::%s %s (%L::%s)',
+        val_1,
+        type_::text,
+        op_symbol,
+        val_2,
+        case when op = 'in' then type_::text || '[]' else type_::text end
+    ) into res;
+
+    return case when negate then not res else res end;
+end;
+$$;
+
+
+-- v2 overload: evaluates a record against the v2 filter type (with negate).
+-- The original is_visible_through_filters(wal_column[], user_defined_filter[])
+-- overload is left in place for the legacy filters column.
+create or replace function realtime.is_visible_through_filters(columns realtime.wal_column[], filters realtime.user_defined_filter_v2[])
+    returns bool
+    language sql
+    stable  -- calls a stable function, so cannot be immutable
+as $$
+    select
+        filters is null
+        or array_length(filters, 1) is null
+        or coalesce(
+            -- Fail closed: every filter must match a column present in the WAL
+            -- payload, otherwise an unevaluable filter would default to visible.
+            count(col.name) = count(1)
+            and sum(
+                realtime.check_equality_op(
+                    op:=f.op,
+                    type_:=coalesce(col.type_oid::regtype, col.type_name::regtype),
+                    val_1:=col.value #>> '{}',
+                    val_2:=f.value,
+                    negate:=coalesce(f.negate, false)
+                )::int
+            ) filter (where col.name is not null) = count(col.name),
+            false
+        )
+    from
+        unnest(filters) f
+        left join unnest(columns) col
+            on f.column_name = col.name;
+$$;
+
+
+-- Additive new column. Default '{}' keeps it invisible to old writers that only
+-- list the legacy `filters` column in their INSERTs.
+alter table realtime.subscription
+    add column if not exists filters_v2 realtime.user_defined_filter_v2[] not null default '{}';
+
+-- The unique constraint must cover both filter columns so new-style
+-- subscriptions (filters = '{}') are still de-duplicated by their filters_v2.
+-- Guarded so a re-run lands in the same state regardless of the starting point.
+alter table realtime.subscription
+    drop constraint if exists subscription_subscription_id_entity_filters_key;
+do $$
+begin
+    if not exists (
+        select 1
+        from pg_constraint
+        where conrelid = 'realtime.subscription'::regclass
+          and conname = 'subscription_subscription_id_entity_filters_key'
+    ) then
+        alter table realtime.subscription
+            add constraint subscription_subscription_id_entity_filters_key
+            unique (subscription_id, entity, filters, filters_v2);
+    end if;
+end $$;
+
+
+-- Trigger validates both filter columns and normalizes their order.
+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 empty selected_columns: array_agg would silently convert '{}' to NULL
+    -- during normalization, making it indistinguishable from "all columns"
+    if new.selected_columns = '{}' then
+        raise exception 'selected_columns cannot be empty. Remove the select parameter to capture all columns.';
+    end if;
+
+    -- 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
+    new.selected_columns = (
+        select array_agg(c order by c)
+        from unnest(new.selected_columns) c
+    );
+
+    return new;
+end;
+$$;
+
+
+-- apply_rls re-defined so the visibility check evaluates BOTH filter columns.
+-- For a given subscription one of the two is always '{}' (which evaluates to
+-- visible), so this reduces to "enforce whichever column the writer populated".
+create or replace function realtime.apply_rls(wal jsonb, max_record_bytes int = 1024 * 1024)
+    returns setof realtime.wal_rls
+    language plpgsql
+    volatile
+as $$
+declare
+    -- Regclass of the table e.g. public.notes
+    entity_ regclass = (quote_ident(wal ->> 'schema') || '.' || quote_ident(wal ->> 'table'))::regclass;
+
+    -- I, U, D, T: insert, update ...
+    action realtime.action = (
+        case wal ->> 'action'
+            when 'I' then 'INSERT'
+            when 'U' then 'UPDATE'
+            when 'D' then 'DELETE'
+            else 'ERROR'
+        end
+    );
+
+    -- Is row level security enabled for the table
+    is_rls_enabled bool = relrowsecurity from pg_class where oid = entity_;
+
+    subscriptions realtime.subscription[] = array_agg(subs)
+        from
+            realtime.subscription subs
+        where
+            subs.entity = entity_
+            -- Filter by action early - only get subscriptions interested in this action
+            -- action_filter column can be: '*' (all), 'INSERT', 'UPDATE', or 'DELETE'
+            and (subs.action_filter = '*' or subs.action_filter = action::text);
+
+    -- Subscription vars
+    working_role regrole;
+    working_selected_columns text[];
+    claimed_role regrole;
+    claims jsonb;
+
+    subscription_id uuid;
+    subscription_has_access bool;
+    visible_to_subscription_ids uuid[] = '{}';
+
+    -- structured info for wal's columns
+    columns realtime.wal_column[];
+    -- previous identity values for update/delete
+    old_columns realtime.wal_column[];
+
+    error_record_exceeds_max_size boolean = octet_length(wal::text) > max_record_bytes;
+
+    -- Primary jsonb output for record
+    output jsonb;
+
+    -- Loop record for iterating unique roles (outer loop)
+    role_record record;
+    -- Loop record for iterating unique selected_columns within a role (inner loop)
+    cols_record record;
+    -- Subscription ids visible at the role level (before fanning out by selected_columns)
+    visible_role_sub_ids uuid[] = '{}';
+
+begin
+    perform set_config('role', null, true);
+
+    columns =
+        array_agg(
+            (
+                x->>'name',
+                x->>'type',
+                x->>'typeoid',
+                realtime.cast(
+                    (x->'value') #>> '{}',
+                    coalesce(
+                        (x->>'typeoid')::regtype, -- null when wal2json version <= 2.4
+                        (x->>'type')::regtype
+                    )
+                ),
+                (pks ->> 'name') is not null,
+                true
+            )::realtime.wal_column
+        )
+        from
+            jsonb_array_elements(wal -> 'columns') x
+            left join jsonb_array_elements(wal -> 'pk') pks
+                on (x ->> 'name') = (pks ->> 'name');
+
+    old_columns =
+        array_agg(
+            (
+                x->>'name',
+                x->>'type',
+                x->>'typeoid',
+                realtime.cast(
+                    (x->'value') #>> '{}',
+                    coalesce(
+                        (x->>'typeoid')::regtype, -- null when wal2json version <= 2.4
+                        (x->>'type')::regtype
+                    )
+                ),
+                (pks ->> 'name') is not null,
+                true
+            )::realtime.wal_column
+        )
+        from
+            jsonb_array_elements(wal -> 'identity') x
+            left join jsonb_array_elements(wal -> 'pk') pks
+                on (x ->> 'name') = (pks ->> 'name');
+
+    for role_record in
+        select claims_role
+        from (select distinct claims_role from unnest(subscriptions)) t
+        order by claims_role::text
+    loop
+        working_role := role_record.claims_role;
+
+        -- Update `is_selectable` for columns and old_columns (once per role)
+        columns =
+            array_agg(
+                (
+                    c.name,
+                    c.type_name,
+                    c.type_oid,
+                    c.value,
+                    c.is_pkey,
+                    pg_catalog.has_column_privilege(working_role, entity_, c.name, 'SELECT')
+                )::realtime.wal_column
+            )
+            from
+                unnest(columns) c;
+
+        old_columns =
+                array_agg(
+                    (
+                        c.name,
+                        c.type_name,
+                        c.type_oid,
+                        c.value,
+                        c.is_pkey,
+                        pg_catalog.has_column_privilege(working_role, entity_, c.name, 'SELECT')
+                    )::realtime.wal_column
+                )
+                from
+                    unnest(old_columns) c;
+
+        if action <> 'DELETE' and count(1) = 0 from unnest(columns) c where c.is_pkey then
+            -- Fan out 400 error per distinct selected_columns for this role
+            for cols_record in
+                select selected_columns
+                from (select distinct selected_columns from unnest(subscriptions) s where s.claims_role = working_role) t
+                order by coalesce(array_to_string(selected_columns, ','), '')
+            loop
+                working_selected_columns := cols_record.selected_columns;
+                return next (
+                    jsonb_build_object(
+                        'schema', wal ->> 'schema',
+                        'table', wal ->> 'table',
+                        'type', action
+                    ),
+                    is_rls_enabled,
+                    (select array_agg(s.subscription_id) from unnest(subscriptions) as s where s.claims_role = working_role and (s.selected_columns is not distinct from working_selected_columns)),
+                    array['Error 400: Bad Request, no primary key']
+                )::realtime.wal_rls;
+            end loop;
+
+        -- The claims role does not have SELECT permission to the primary key of entity
+        elsif action <> 'DELETE' and sum(c.is_selectable::int) <> count(1) from unnest(columns) c where c.is_pkey then
+            -- Fan out 401 error per distinct selected_columns for this role
+            for cols_record in
+                select selected_columns
+                from (select distinct selected_columns from unnest(subscriptions) s where s.claims_role = working_role) t
+                order by coalesce(array_to_string(selected_columns, ','), '')
+            loop
+                working_selected_columns := cols_record.selected_columns;
+                return next (
+                    jsonb_build_object(
+                        'schema', wal ->> 'schema',
+                        'table', wal ->> 'table',
+                        'type', action
+                    ),
+                    is_rls_enabled,
+                    (select array_agg(s.subscription_id) from unnest(subscriptions) as s where s.claims_role = working_role and (s.selected_columns is not distinct from working_selected_columns)),
+                    array['Error 401: Unauthorized']
+                )::realtime.wal_rls;
+            end loop;
+
+        else
+            -- Create the prepared statement (once per role)
+            if is_rls_enabled and action <> 'DELETE' then
+                if (select 1 from pg_prepared_statements where name = 'walrus_rls_stmt' limit 1) > 0 then
+                    deallocate walrus_rls_stmt;
+                end if;
+                execute realtime.build_prepared_statement_sql('walrus_rls_stmt', entity_, columns);
+            end if;
+
+            -- Collect all visible subscription IDs for this role (filter check + RLS check)
+            visible_role_sub_ids = '{}';
+
+            for subscription_id, claims in (
+                    select
+                        subs.subscription_id,
+                        subs.claims
+                    from
+                        unnest(subscriptions) subs
+                    where
+                        subs.entity = entity_
+                        and subs.claims_role = working_role
+                        and (
+                            (
+                                realtime.is_visible_through_filters(columns, subs.filters)
+                                and realtime.is_visible_through_filters(columns, subs.filters_v2)
+                            )
+                            or (
+                              action = 'DELETE'
+                              and realtime.is_visible_through_filters(old_columns, subs.filters)
+                              and realtime.is_visible_through_filters(old_columns, subs.filters_v2)
+                            )
+                        )
+            ) loop
+
+                if not is_rls_enabled or action = 'DELETE' then
+                    visible_role_sub_ids = visible_role_sub_ids || subscription_id;
+                else
+                    -- Check if RLS allows the role to see the record
+                    perform
+                        -- Trim leading and trailing quotes from working_role because set_config
+                        -- doesn't recognize the role as valid if they are included
+                        set_config('role', trim(both '"' from working_role::text), true),
+                        set_config('request.jwt.claims', claims::text, true);
+
+                    execute 'execute walrus_rls_stmt' into subscription_has_access;
+
+                    if subscription_has_access then
+                        visible_role_sub_ids = visible_role_sub_ids || subscription_id;
+                    end if;
+                end if;
+            end loop;
+
+            perform set_config('role', null, true);
+
+            -- Inner loop: per distinct selected_columns for this role
+            for cols_record in
+                select selected_columns
+                from (select distinct selected_columns from unnest(subscriptions) s where s.claims_role = working_role) t
+                order by coalesce(array_to_string(selected_columns, ','), '')
+            loop
+                working_selected_columns := cols_record.selected_columns;
+
+                output = jsonb_build_object(
+                    'schema', wal ->> 'schema',
+                    'table', wal ->> 'table',
+                    'type', action,
+                    'commit_timestamp', to_char(
+                        ((wal ->> 'timestamp')::timestamptz at time zone 'utc'),
+                        'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'
+                    ),
+                    'columns', (
+                        select
+                            jsonb_agg(
+                                jsonb_build_object(
+                                    'name', pa.attname,
+                                    'type', pt.typname
+                                )
+                                order by pa.attnum asc
+                            )
+                        from
+                            pg_attribute pa
+                            join pg_type pt
+                                on pa.atttypid = pt.oid
+                            left join (
+                                select unnest(conkey) as pkey_attnum
+                                from pg_constraint
+                                where conrelid = entity_ and contype = 'p'
+                            ) pk on pk.pkey_attnum = pa.attnum
+                        where
+                            attrelid = entity_
+                            and attnum > 0
+                            and pg_catalog.has_column_privilege(working_role, entity_, pa.attname, 'SELECT')
+                            and (working_selected_columns is null or pa.attname = any(working_selected_columns) or pk.pkey_attnum is not null)
+                    )
+                )
+                -- Add "record" key for insert and update
+                || case
+                    when action in ('INSERT', 'UPDATE') then
+                        jsonb_build_object(
+                            'record',
+                            (
+                                select
+                                    jsonb_object_agg(
+                                        -- if unchanged toast, get column name and value from old record
+                                        coalesce((c).name, (oc).name),
+                                        case
+                                            when (c).name is null then (oc).value
+                                            else (c).value
+                                        end
+                                    )
+                                from
+                                    unnest(columns) c
+                                    full outer join unnest(old_columns) oc
+                                        on (c).name = (oc).name
+                                where
+                                    coalesce((c).is_selectable, (oc).is_selectable)
+                                    and (working_selected_columns is null or coalesce((c).name, (oc).name) = any(working_selected_columns) or coalesce((c).is_pkey, (oc).is_pkey))
+                                    and ( not error_record_exceeds_max_size or (octet_length((c).value::text) <= 64))
+                            )
+                        )
+                    else '{}'::jsonb
+                end
+                -- Add "old_record" key for update and delete
+                || case
+                    when action = 'UPDATE' then
+                        jsonb_build_object(
+                                'old_record',
+                                (
+                                    select jsonb_object_agg((c).name, (c).value)
+                                    from unnest(old_columns) c
+                                    where
+                                        (c).is_selectable
+                                        and (working_selected_columns is null or (c).name = any(working_selected_columns) or (c).is_pkey)
+                                        and ( not error_record_exceeds_max_size or (octet_length((c).value::text) <= 64))
+                                )
+                            )
+                    when action = 'DELETE' then
+                        jsonb_build_object(
+                            'old_record',
+                            (
+                                select jsonb_object_agg((c).name, (c).value)
+                                from unnest(old_columns) c
+                                where
+                                    (c).is_selectable
+                                    and (working_selected_columns is null or (c).name = any(working_selected_columns) or (c).is_pkey)
+                                    and ( not error_record_exceeds_max_size or (octet_length((c).value::text) <= 64))
+                                    and ( not is_rls_enabled or (c).is_pkey ) -- if RLS enabled, we can't secure deletes so filter to pkey
+                            )
+                        )
+                    else '{}'::jsonb
+                end;
+
+                -- Filter visible_role_sub_ids to those matching the current selected_columns group
+                visible_to_subscription_ids = coalesce(
+                    (
+                        select array_agg(s.subscription_id)
+                        from unnest(subscriptions) s
+                        where s.claims_role = working_role
+                          and (s.selected_columns is not distinct from working_selected_columns)
+                          and s.subscription_id = any(visible_role_sub_ids)
+                    ),
+                    '{}'::uuid[]
+                );
+
+                return next (
+                    output,
+                    is_rls_enabled,
+                    visible_to_subscription_ids,
+                    case
+                        when error_record_exceeds_max_size then array['Error 413: Payload Too Large']
+                        else '{}'
+                    end
+                )::realtime.wal_rls;
+            end loop;
+
+        end if;
+    end loop;
+
+    perform set_config('role', null, true);
+end;
+$$;
diff --git a/test/expected/test_integration_in_filter.out b/test/expected/test_integration_in_filter.out
index f2245d1..ca44f0c 100644
--- a/test/expected/test_integration_in_filter.out
+++ b/test/expected/test_integration_in_filter.out
@@ -105,7 +105,7 @@ select
     ),
     array[('body', 'in', array[1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1,1])::realtime.user_defined_filter];
 ERROR:  too many values for `in` filter. Maximum 100
-CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 41 at RAISE
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 52 at RAISE
 drop table public.notes;
 select pg_drop_replication_slot('realtime');
  pg_drop_replication_slot 
diff --git a/test/expected/test_integration_like_ilike_is_not_filters.out b/test/expected/test_integration_like_ilike_is_not_filters.out
new file mode 100644
index 0000000..fb91e4e
--- /dev/null
+++ b/test/expected/test_integration_like_ilike_is_not_filters.out
@@ -0,0 +1,191 @@
+create table public.notes(
+    id int primary key,
+    body text,
+    nullable_body text
+);
+-- ── Expand/contract: legacy `filters` and new `filters_v2` coexist ──
+-- Old writers keep inserting 3-field legacy filters successfully (filters_v2
+-- defaults to '{}').
+insert into realtime.subscription(subscription_id, entity, claims, filters)
+values (
+    '00000000-0000-0000-0000-000000000001',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000001'),
+    array[('id', 'eq', '6')]::realtime.user_defined_filter[]
+);
+-- New writers insert v2 filters (with negate) into filters_v2.
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000002',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000002'),
+    array[
+        ('body', 'ilike', '%world%', false),
+        ('body', 'like',  '%x%',     true)
+    ]::realtime.user_defined_filter_v2[]
+);
+-- filters_v2 is normalized (sorted by column_name, op, value, negate).
+select filters, filters_v2 from realtime.subscription order by subscription_id;
+    filters    |                   filters_v2                   
+---------------+------------------------------------------------
+ {"(id,eq,6)"} | {}
+ {}            | {"(body,like,%x%,t)","(body,ilike,%world%,f)"}
+(2 rows)
+
+-- The new operators are rejected on the LEGACY filters column (they can only be
+-- evaluated on filters_v2; otherwise apply_rls would hit the UNKNOWN OP path).
+insert into realtime.subscription(subscription_id, entity, claims, filters)
+values (
+    '00000000-0000-0000-0000-000000000003',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000003'),
+    array[('body', 'like', '%x%')]::realtime.user_defined_filter[]
+);
+ERROR:  operator like is only supported on the filters_v2 column
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 46 at RAISE
+-- `is` with an invalid keyword value is rejected at subscription time.
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000004',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000004'),
+    array[('body', 'is', 'maybe', false)]::realtime.user_defined_filter_v2[]
+);
+ERROR:  invalid value for is filter: must be null, true, false, or unknown
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 84 at RAISE
+-- like/ilike on a non-text column is rejected at subscription time.
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000005',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000005'),
+    array[('id', 'like', '%5%', false)]::realtime.user_defined_filter_v2[]
+);
+ERROR:  operator like requires a text-compatible column type, got integer
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 99 at RAISE
+-- An invalid regex is rejected at subscription time.
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000006',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000006'),
+    array[('body', 'match', '(unclosed', false)]::realtime.user_defined_filter_v2[]
+);
+ERROR:  invalid regular expression for match filter: invalid regular expression: parentheses () not balanced
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 119 at RAISE
+-- match/imatch on a non-text column is rejected at subscription time (the regex
+-- operator has no overload for the column type, which would abort apply_rls).
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000007',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000007'),
+    array[('id', 'match', 'foo.*', false)]::realtime.user_defined_filter_v2[]
+);
+ERROR:  operator match requires a text-compatible column type, got integer
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 112 at RAISE
+-- `is true` on a non-boolean column is rejected at subscription time (only
+-- `is null` is type-agnostic).
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000008',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000008'),
+    array[('body', 'is', 'true', false)]::realtime.user_defined_filter_v2[]
+);
+ERROR:  is true filter requires a boolean column, got text
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 90 at RAISE
+-- ── Evaluation path (realtime.is_visible_through_filters over v2 filters) ──
+-- A row equivalent to public.notes(id=1, body='hello world', nullable_body=null).
+-- like: matches                                       → visible (true)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('body', 'like', '%world%', false)]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ t
+(1 row)
+
+-- ilike: case-insensitive match                       → visible (true)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('body', 'ilike', '%WORLD%', false)]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ t
+(1 row)
+
+-- NOT LIKE: row matches pattern                        → not visible (false)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('body', 'like', '%world%', true)]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ f
+(1 row)
+
+-- NOT IN: value inside the list                        → not visible (false)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('body', 'in', '{hello world,other}', true)]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ f
+(1 row)
+
+-- is null on a null column                            → visible (true)
+select realtime.is_visible_through_filters(
+    array[('nullable_body', 'text', null, 'null'::jsonb, false, true)]::realtime.wal_column[],
+    array[('nullable_body', 'is', 'null', false)]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ t
+(1 row)
+
+-- is not null on a null column                        → not visible (false)
+select realtime.is_visible_through_filters(
+    array[('nullable_body', 'text', null, 'null'::jsonb, false, true)]::realtime.wal_column[],
+    array[('nullable_body', 'is', 'null', true)]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ f
+(1 row)
+
+-- isdistinct vs a different value                     → visible (true)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('body', 'isdistinct', 'other', false)]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ t
+(1 row)
+
+-- Fail closed: filter references a column absent from the WAL payload → not visible (false)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('missing', 'eq', 'x', false)]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ f
+(1 row)
+
+-- No filters                                           → visible (true)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    '{}'::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ t
+(1 row)
+
+truncate table realtime.subscription;
+drop table public.notes;
diff --git a/test/expected/test_integration_v2_filter_composition.out b/test/expected/test_integration_v2_filter_composition.out
new file mode 100644
index 0000000..a7c10af
--- /dev/null
+++ b/test/expected/test_integration_v2_filter_composition.out
@@ -0,0 +1,144 @@
+-- Composite use cases: multiple v2 filters combined with AND semantics, mixing
+-- every new operator and negations. realtime.is_visible_through_filters requires
+-- ALL filters in the array to hold, so these exercise complex real-world
+-- subscriptions like "title matches X AND status not in Y AND score >= Z".
+create table public.articles(
+    id int primary key,
+    title text,
+    status text,
+    score int,
+    nickname text
+);
+-- A row equivalent to:
+--   articles(id=42, title='Hello World', status='active', score=10, nickname=null)
+-- expressed as a realtime.wal_column[] payload reused across the checks below.
+-- ── Subscription path: the trigger accepts and normalizes a complex composite ──
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-0000000000aa',
+    'public.articles',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-0000000000aa'),
+    array[
+        ('title',  'ilike', '%world%',          false),  -- matches
+        ('status', 'in',    '{pending,closed}', true),    -- NOT IN  → matches (active not listed)
+        ('score',  'gte',   '5',                false),   -- 10 >= 5 → matches
+        ('nickname','is',   'null',             false),   -- null    → matches
+        ('id',     'isdistinct', '0',           false)    -- 42 distinct from 0 → matches
+    ]::realtime.user_defined_filter_v2[]
+);
+select filters_v2 from realtime.subscription where subscription_id = '00000000-0000-0000-0000-0000000000aa';
+                                                           filters_v2                                                            
+---------------------------------------------------------------------------------------------------------------------------------
+ {"(id,isdistinct,0,f)","(nickname,is,null,f)","(score,gte,5,f)","(status,in,\"{pending,closed}\",t)","(title,ilike,%world%,f)"}
+(1 row)
+
+-- ── Evaluation: all-AND composites over the example row ──
+-- 1. Every operator satisfied (ilike + NOT IN + gte + is null + isdistinct) → visible
+select realtime.is_visible_through_filters(
+    array[
+        ('id',      'int4', null, '42'::jsonb,            true,  true),
+        ('title',   'text', null, '"Hello World"'::jsonb, false, true),
+        ('status',  'text', null, '"active"'::jsonb,      false, true),
+        ('score',   'int4', null, '10'::jsonb,            false, true),
+        ('nickname','text', null, 'null'::jsonb,          false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('title',   'ilike',      '%world%',          false),
+        ('status',  'in',         '{pending,closed}', true),
+        ('score',   'gte',        '5',                false),
+        ('nickname','is',         'null',             false),
+        ('id',      'isdistinct', '0',                false)
+    ]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ t
+(1 row)
+
+-- 2. Same composite but one negation flips a clause (NOT ilike on a matching
+--    pattern) → the AND fails → not visible
+select realtime.is_visible_through_filters(
+    array[
+        ('title',   'text', null, '"Hello World"'::jsonb, false, true),
+        ('status',  'text', null, '"active"'::jsonb,      false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('title',   'ilike', '%world%', true),   -- NOT ilike, but it matches → clause false
+        ('status',  'eq',    'active',  false)
+    ]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ f
+(1 row)
+
+-- 3. Regex composite with negation: imatch matches AND NOT match a different
+--    pattern AND neq → visible
+select realtime.is_visible_through_filters(
+    array[
+        ('title',  'text', null, '"Hello World"'::jsonb, false, true),
+        ('status', 'text', null, '"active"'::jsonb,      false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('title',  'imatch', '^hello',   false),  -- case-insensitive match
+        ('title',  'match',  '^Goodbye', true),   -- NOT match → true (no match)
+        ('status', 'neq',    'archived', false)
+    ]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ t
+(1 row)
+
+-- 4. Negated isdistinct used as "IS NOT DISTINCT FROM" inside a composite:
+--    score IS NOT DISTINCT FROM 10 (true) AND id between via gt/lt → visible
+select realtime.is_visible_through_filters(
+    array[
+        ('id',    'int4', null, '42'::jsonb, true,  true),
+        ('score', 'int4', null, '10'::jsonb, false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('score', 'isdistinct', '10', true),   -- IS NOT DISTINCT FROM 10 → true
+        ('id',    'gt',         '40', false),
+        ('id',    'lt',         '100', false)
+    ]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ t
+(1 row)
+
+-- 5. `is not null` inside a composite where the column is NULL → clause false → not visible
+select realtime.is_visible_through_filters(
+    array[
+        ('title',   'text', null, '"Hello World"'::jsonb, false, true),
+        ('nickname','text', null, 'null'::jsonb,          false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('title',   'like', 'Hello%', false),
+        ('nickname','is',   'null',   true)   -- IS NOT NULL, but nickname is null → false
+    ]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ f
+(1 row)
+
+-- 6. Fail-closed within a composite: one clause references a column missing from
+--    the payload → whole composite not visible even though the others match
+select realtime.is_visible_through_filters(
+    array[
+        ('title', 'text', null, '"Hello World"'::jsonb, false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('title',  'like', 'Hello%', false),  -- matches
+        ('status', 'eq',   'active', false)   -- status absent from payload → fail closed
+    ]::realtime.user_defined_filter_v2[]
+);
+ is_visible_through_filters 
+----------------------------
+ f
+(1 row)
+
+truncate table realtime.subscription;
+drop table public.articles;
diff --git a/test/expected/test_select_columns_empty.out b/test/expected/test_select_columns_empty.out
index 20b8cf0..dfcb515 100644
--- a/test/expected/test_select_columns_empty.out
+++ b/test/expected/test_select_columns_empty.out
@@ -16,6 +16,6 @@ select
     ),
     '{}'::text[];
 ERROR:  selected_columns cannot be empty. Remove the select parameter to capture all columns.
-CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 52 at RAISE
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 130 at RAISE
 drop table public.notes;
 truncate table realtime.subscription;
diff --git a/test/expected/test_select_columns_invalid.out b/test/expected/test_select_columns_invalid.out
index 5a3901c..faf83dc 100644
--- a/test/expected/test_select_columns_invalid.out
+++ b/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 64 at RAISE
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 142 at RAISE
 drop table public.notes;
 truncate table realtime.subscription;
diff --git a/test/expected/test_select_columns_null_element.out b/test/expected/test_select_columns_null_element.out
index 42b0357..7690ccf 100644
--- a/test/expected/test_select_columns_null_element.out
+++ b/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 57 at RAISE
+CONTEXT:  PL/pgSQL function realtime.subscription_check_filters() line 135 at RAISE
 drop table public.notes;
 truncate table realtime.subscription;
diff --git a/test/expected/test_unit_like_ilike_is_not_filters.out b/test/expected/test_unit_like_ilike_is_not_filters.out
new file mode 100644
index 0000000..82acf01
--- /dev/null
+++ b/test/expected/test_unit_like_ilike_is_not_filters.out
@@ -0,0 +1,230 @@
+-- like (negate=false)
+select realtime.check_equality_op('like', 'text', 'hello world', '%world%', false);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('like', 'text', 'hello world', '%xyz%',   false);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- like negated (NOT LIKE)
+select realtime.check_equality_op('like', 'text', 'hello world', '%xyz%',   true);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('like', 'text', 'hello world', '%world%', true);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- ilike (negate=false)
+select realtime.check_equality_op('ilike', 'text', 'Hello World', '%world%', false);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('ilike', 'text', 'Hello World', '%xyz%',   false);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- ilike negated (NOT ILIKE)
+select realtime.check_equality_op('ilike', 'text', 'Hello World', '%xyz%',   true);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('ilike', 'text', 'Hello World', '%world%', true);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- in (negate=false)
+select realtime.check_equality_op('in', 'bigint', '2', '{1,2,3}', false);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('in', 'bigint', '4', '{1,2,3}', false);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- in negated (NOT IN)
+select realtime.check_equality_op('in', 'bigint', '4', '{1,2,3}', true);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('in', 'bigint', '2', '{1,2,3}', true);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- is null
+select realtime.check_equality_op('is', 'text', null,    'null', false);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('is', 'text', 'value', 'null', false);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- is not null (negate=true)
+select realtime.check_equality_op('is', 'text', 'value', 'null', true);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('is', 'text', null,    'null', true);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- is true / false on boolean
+select realtime.check_equality_op('is', 'boolean', 'true',  'true',  false);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('is', 'boolean', 'false', 'true',  false);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+select realtime.check_equality_op('is', 'boolean', 'false', 'true',  true);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+-- match (regex)
+select realtime.check_equality_op('match',  'text', 'hello world', 'hel+o',  false);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('match',  'text', 'hello world', '^world', false);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+select realtime.check_equality_op('match',  'text', 'hello world', 'hel+o',  true);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- imatch (case-insensitive regex)
+select realtime.check_equality_op('imatch', 'text', 'Hello World', 'hel+o',  false);
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('imatch', 'text', 'Hello World', '^world', false);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+select realtime.check_equality_op('imatch', 'text', 'Hello World', 'hel+o',  true);
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- isdistinct (IS DISTINCT FROM — differs from neq in NULL handling)
+select realtime.check_equality_op('isdistinct', 'text', 'aaa', 'bbb',  false);  -- true
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('isdistinct', 'text', 'aaa', 'aaa',  false);  -- false
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+select realtime.check_equality_op('isdistinct', 'text', null,  'aaa',  false);  -- true (NULL IS DISTINCT FROM 'aaa')
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('isdistinct', 'text', null,  'aaa',  true);   -- false (IS NOT DISTINCT FROM)
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+select realtime.check_equality_op('isdistinct', 'text', null,  null,   false);  -- false (NULL IS NOT DISTINCT FROM NULL)
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- negate works on all existing operators
+select realtime.check_equality_op('eq',  'text',   'aaa', 'aaa',      true);  -- NOT eq  → false
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+select realtime.check_equality_op('neq', 'text',   'aaa', 'bbb',      true);  -- NOT neq → false
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+select realtime.check_equality_op('lt',  'bigint', '1',   '2',        true);  -- NOT lt  → false
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+select realtime.check_equality_op('gte', 'bigint', '5',   '3',        true);  -- NOT gte → false
+ check_equality_op 
+-------------------
+ f
+(1 row)
+
+-- negate defaults to false (existing callers unaffected)
+select realtime.check_equality_op('eq', 'text', 'aaa', 'aaa');
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
+select realtime.check_equality_op('in', 'bigint', '2', '{1,2,3}');
+ check_equality_op 
+-------------------
+ t
+(1 row)
+
diff --git a/test/sql/test_integration_like_ilike_is_not_filters.sql b/test/sql/test_integration_like_ilike_is_not_filters.sql
new file mode 100644
index 0000000..07560d0
--- /dev/null
+++ b/test/sql/test_integration_like_ilike_is_not_filters.sql
@@ -0,0 +1,149 @@
+create table public.notes(
+    id int primary key,
+    body text,
+    nullable_body text
+);
+
+-- ── Expand/contract: legacy `filters` and new `filters_v2` coexist ──
+
+-- Old writers keep inserting 3-field legacy filters successfully (filters_v2
+-- defaults to '{}').
+insert into realtime.subscription(subscription_id, entity, claims, filters)
+values (
+    '00000000-0000-0000-0000-000000000001',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000001'),
+    array[('id', 'eq', '6')]::realtime.user_defined_filter[]
+);
+
+-- New writers insert v2 filters (with negate) into filters_v2.
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000002',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000002'),
+    array[
+        ('body', 'ilike', '%world%', false),
+        ('body', 'like',  '%x%',     true)
+    ]::realtime.user_defined_filter_v2[]
+);
+
+-- filters_v2 is normalized (sorted by column_name, op, value, negate).
+select filters, filters_v2 from realtime.subscription order by subscription_id;
+
+-- The new operators are rejected on the LEGACY filters column (they can only be
+-- evaluated on filters_v2; otherwise apply_rls would hit the UNKNOWN OP path).
+insert into realtime.subscription(subscription_id, entity, claims, filters)
+values (
+    '00000000-0000-0000-0000-000000000003',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000003'),
+    array[('body', 'like', '%x%')]::realtime.user_defined_filter[]
+);
+
+-- `is` with an invalid keyword value is rejected at subscription time.
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000004',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000004'),
+    array[('body', 'is', 'maybe', false)]::realtime.user_defined_filter_v2[]
+);
+
+-- like/ilike on a non-text column is rejected at subscription time.
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000005',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000005'),
+    array[('id', 'like', '%5%', false)]::realtime.user_defined_filter_v2[]
+);
+
+-- An invalid regex is rejected at subscription time.
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000006',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000006'),
+    array[('body', 'match', '(unclosed', false)]::realtime.user_defined_filter_v2[]
+);
+
+-- match/imatch on a non-text column is rejected at subscription time (the regex
+-- operator has no overload for the column type, which would abort apply_rls).
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000007',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000007'),
+    array[('id', 'match', 'foo.*', false)]::realtime.user_defined_filter_v2[]
+);
+
+-- `is true` on a non-boolean column is rejected at subscription time (only
+-- `is null` is type-agnostic).
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-000000000008',
+    'public.notes',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-000000000008'),
+    array[('body', 'is', 'true', false)]::realtime.user_defined_filter_v2[]
+);
+
+-- ── Evaluation path (realtime.is_visible_through_filters over v2 filters) ──
+-- A row equivalent to public.notes(id=1, body='hello world', nullable_body=null).
+
+-- like: matches                                       → visible (true)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('body', 'like', '%world%', false)]::realtime.user_defined_filter_v2[]
+);
+
+-- ilike: case-insensitive match                       → visible (true)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('body', 'ilike', '%WORLD%', false)]::realtime.user_defined_filter_v2[]
+);
+
+-- NOT LIKE: row matches pattern                        → not visible (false)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('body', 'like', '%world%', true)]::realtime.user_defined_filter_v2[]
+);
+
+-- NOT IN: value inside the list                        → not visible (false)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('body', 'in', '{hello world,other}', true)]::realtime.user_defined_filter_v2[]
+);
+
+-- is null on a null column                            → visible (true)
+select realtime.is_visible_through_filters(
+    array[('nullable_body', 'text', null, 'null'::jsonb, false, true)]::realtime.wal_column[],
+    array[('nullable_body', 'is', 'null', false)]::realtime.user_defined_filter_v2[]
+);
+
+-- is not null on a null column                        → not visible (false)
+select realtime.is_visible_through_filters(
+    array[('nullable_body', 'text', null, 'null'::jsonb, false, true)]::realtime.wal_column[],
+    array[('nullable_body', 'is', 'null', true)]::realtime.user_defined_filter_v2[]
+);
+
+-- isdistinct vs a different value                     → visible (true)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('body', 'isdistinct', 'other', false)]::realtime.user_defined_filter_v2[]
+);
+
+-- Fail closed: filter references a column absent from the WAL payload → not visible (false)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    array[('missing', 'eq', 'x', false)]::realtime.user_defined_filter_v2[]
+);
+
+-- No filters                                           → visible (true)
+select realtime.is_visible_through_filters(
+    array[('body', 'text', null, '"hello world"'::jsonb, false, true)]::realtime.wal_column[],
+    '{}'::realtime.user_defined_filter_v2[]
+);
+
+truncate table realtime.subscription;
+drop table public.notes;
diff --git a/test/sql/test_integration_v2_filter_composition.sql b/test/sql/test_integration_v2_filter_composition.sql
new file mode 100644
index 0000000..96a3176
--- /dev/null
+++ b/test/sql/test_integration_v2_filter_composition.sql
@@ -0,0 +1,120 @@
+-- Composite use cases: multiple v2 filters combined with AND semantics, mixing
+-- every new operator and negations. realtime.is_visible_through_filters requires
+-- ALL filters in the array to hold, so these exercise complex real-world
+-- subscriptions like "title matches X AND status not in Y AND score >= Z".
+
+create table public.articles(
+    id int primary key,
+    title text,
+    status text,
+    score int,
+    nickname text
+);
+
+-- A row equivalent to:
+--   articles(id=42, title='Hello World', status='active', score=10, nickname=null)
+-- expressed as a realtime.wal_column[] payload reused across the checks below.
+
+-- ── Subscription path: the trigger accepts and normalizes a complex composite ──
+insert into realtime.subscription(subscription_id, entity, claims, filters_v2)
+values (
+    '00000000-0000-0000-0000-0000000000aa',
+    'public.articles',
+    jsonb_build_object('role', 'authenticated', 'email', 'a@example.com', 'sub', '00000000-0000-0000-0000-0000000000aa'),
+    array[
+        ('title',  'ilike', '%world%',          false),  -- matches
+        ('status', 'in',    '{pending,closed}', true),    -- NOT IN  → matches (active not listed)
+        ('score',  'gte',   '5',                false),   -- 10 >= 5 → matches
+        ('nickname','is',   'null',             false),   -- null    → matches
+        ('id',     'isdistinct', '0',           false)    -- 42 distinct from 0 → matches
+    ]::realtime.user_defined_filter_v2[]
+);
+select filters_v2 from realtime.subscription where subscription_id = '00000000-0000-0000-0000-0000000000aa';
+
+-- ── Evaluation: all-AND composites over the example row ──
+
+-- 1. Every operator satisfied (ilike + NOT IN + gte + is null + isdistinct) → visible
+select realtime.is_visible_through_filters(
+    array[
+        ('id',      'int4', null, '42'::jsonb,            true,  true),
+        ('title',   'text', null, '"Hello World"'::jsonb, false, true),
+        ('status',  'text', null, '"active"'::jsonb,      false, true),
+        ('score',   'int4', null, '10'::jsonb,            false, true),
+        ('nickname','text', null, 'null'::jsonb,          false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('title',   'ilike',      '%world%',          false),
+        ('status',  'in',         '{pending,closed}', true),
+        ('score',   'gte',        '5',                false),
+        ('nickname','is',         'null',             false),
+        ('id',      'isdistinct', '0',                false)
+    ]::realtime.user_defined_filter_v2[]
+);
+
+-- 2. Same composite but one negation flips a clause (NOT ilike on a matching
+--    pattern) → the AND fails → not visible
+select realtime.is_visible_through_filters(
+    array[
+        ('title',   'text', null, '"Hello World"'::jsonb, false, true),
+        ('status',  'text', null, '"active"'::jsonb,      false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('title',   'ilike', '%world%', true),   -- NOT ilike, but it matches → clause false
+        ('status',  'eq',    'active',  false)
+    ]::realtime.user_defined_filter_v2[]
+);
+
+-- 3. Regex composite with negation: imatch matches AND NOT match a different
+--    pattern AND neq → visible
+select realtime.is_visible_through_filters(
+    array[
+        ('title',  'text', null, '"Hello World"'::jsonb, false, true),
+        ('status', 'text', null, '"active"'::jsonb,      false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('title',  'imatch', '^hello',   false),  -- case-insensitive match
+        ('title',  'match',  '^Goodbye', true),   -- NOT match → true (no match)
+        ('status', 'neq',    'archived', false)
+    ]::realtime.user_defined_filter_v2[]
+);
+
+-- 4. Negated isdistinct used as "IS NOT DISTINCT FROM" inside a composite:
+--    score IS NOT DISTINCT FROM 10 (true) AND id between via gt/lt → visible
+select realtime.is_visible_through_filters(
+    array[
+        ('id',    'int4', null, '42'::jsonb, true,  true),
+        ('score', 'int4', null, '10'::jsonb, false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('score', 'isdistinct', '10', true),   -- IS NOT DISTINCT FROM 10 → true
+        ('id',    'gt',         '40', false),
+        ('id',    'lt',         '100', false)
+    ]::realtime.user_defined_filter_v2[]
+);
+
+-- 5. `is not null` inside a composite where the column is NULL → clause false → not visible
+select realtime.is_visible_through_filters(
+    array[
+        ('title',   'text', null, '"Hello World"'::jsonb, false, true),
+        ('nickname','text', null, 'null'::jsonb,          false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('title',   'like', 'Hello%', false),
+        ('nickname','is',   'null',   true)   -- IS NOT NULL, but nickname is null → false
+    ]::realtime.user_defined_filter_v2[]
+);
+
+-- 6. Fail-closed within a composite: one clause references a column missing from
+--    the payload → whole composite not visible even though the others match
+select realtime.is_visible_through_filters(
+    array[
+        ('title', 'text', null, '"Hello World"'::jsonb, false, true)
+    ]::realtime.wal_column[],
+    array[
+        ('title',  'like', 'Hello%', false),  -- matches
+        ('status', 'eq',   'active', false)   -- status absent from payload → fail closed
+    ]::realtime.user_defined_filter_v2[]
+);
+
+truncate table realtime.subscription;
+drop table public.articles;
diff --git a/test/sql/test_unit_like_ilike_is_not_filters.sql b/test/sql/test_unit_like_ilike_is_not_filters.sql
new file mode 100644
index 0000000..56ea901
--- /dev/null
+++ b/test/sql/test_unit_like_ilike_is_not_filters.sql
@@ -0,0 +1,63 @@
+-- like (negate=false)
+select realtime.check_equality_op('like', 'text', 'hello world', '%world%', false);
+select realtime.check_equality_op('like', 'text', 'hello world', '%xyz%',   false);
+
+-- like negated (NOT LIKE)
+select realtime.check_equality_op('like', 'text', 'hello world', '%xyz%',   true);
+select realtime.check_equality_op('like', 'text', 'hello world', '%world%', true);
+
+-- ilike (negate=false)
+select realtime.check_equality_op('ilike', 'text', 'Hello World', '%world%', false);
+select realtime.check_equality_op('ilike', 'text', 'Hello World', '%xyz%',   false);
+
+-- ilike negated (NOT ILIKE)
+select realtime.check_equality_op('ilike', 'text', 'Hello World', '%xyz%',   true);
+select realtime.check_equality_op('ilike', 'text', 'Hello World', '%world%', true);
+
+-- in (negate=false)
+select realtime.check_equality_op('in', 'bigint', '2', '{1,2,3}', false);
+select realtime.check_equality_op('in', 'bigint', '4', '{1,2,3}', false);
+
+-- in negated (NOT IN)
+select realtime.check_equality_op('in', 'bigint', '4', '{1,2,3}', true);
+select realtime.check_equality_op('in', 'bigint', '2', '{1,2,3}', true);
+
+-- is null
+select realtime.check_equality_op('is', 'text', null,    'null', false);
+select realtime.check_equality_op('is', 'text', 'value', 'null', false);
+
+-- is not null (negate=true)
+select realtime.check_equality_op('is', 'text', 'value', 'null', true);
+select realtime.check_equality_op('is', 'text', null,    'null', true);
+
+-- is true / false on boolean
+select realtime.check_equality_op('is', 'boolean', 'true',  'true',  false);
+select realtime.check_equality_op('is', 'boolean', 'false', 'true',  false);
+select realtime.check_equality_op('is', 'boolean', 'false', 'true',  true);
+
+-- match (regex)
+select realtime.check_equality_op('match',  'text', 'hello world', 'hel+o',  false);
+select realtime.check_equality_op('match',  'text', 'hello world', '^world', false);
+select realtime.check_equality_op('match',  'text', 'hello world', 'hel+o',  true);
+
+-- imatch (case-insensitive regex)
+select realtime.check_equality_op('imatch', 'text', 'Hello World', 'hel+o',  false);
+select realtime.check_equality_op('imatch', 'text', 'Hello World', '^world', false);
+select realtime.check_equality_op('imatch', 'text', 'Hello World', 'hel+o',  true);
+
+-- isdistinct (IS DISTINCT FROM — differs from neq in NULL handling)
+select realtime.check_equality_op('isdistinct', 'text', 'aaa', 'bbb',  false);  -- true
+select realtime.check_equality_op('isdistinct', 'text', 'aaa', 'aaa',  false);  -- false
+select realtime.check_equality_op('isdistinct', 'text', null,  'aaa',  false);  -- true (NULL IS DISTINCT FROM 'aaa')
+select realtime.check_equality_op('isdistinct', 'text', null,  'aaa',  true);   -- false (IS NOT DISTINCT FROM)
+select realtime.check_equality_op('isdistinct', 'text', null,  null,   false);  -- false (NULL IS NOT DISTINCT FROM NULL)
+
+-- negate works on all existing operators
+select realtime.check_equality_op('eq',  'text',   'aaa', 'aaa',      true);  -- NOT eq  → false
+select realtime.check_equality_op('neq', 'text',   'aaa', 'bbb',      true);  -- NOT neq → false
+select realtime.check_equality_op('lt',  'bigint', '1',   '2',        true);  -- NOT lt  → false
+select realtime.check_equality_op('gte', 'bigint', '5',   '3',        true);  -- NOT gte → false
+
+-- negate defaults to false (existing callers unaffected)
+select realtime.check_equality_op('eq', 'text', 'aaa', 'aaa');
+select realtime.check_equality_op('in', 'bigint', '2', '{1,2,3}');