Supabase Source

Connect Supabase as a source to sync tables from your Supabase project. Supaflow reads through Supabase's Data REST API using a Supabase Auth user session, so your Row Level Security policies are evaluated as they would be for an authenticated application user.

For an overview of capabilities and use cases, see the Supabase connector page. To run pipelines natively inside Snowflake, see Snowflake Native ETL.

When to Use This Connector

Use the Supabase source when you want Supaflow to read data as a Supabase Auth user and respect RLS policies. If you want database-admin style access with a PostgreSQL username and password, use the PostgreSQL source instead.

Prerequisites

Before you begin, ensure you have:

• A Supabase project
• The Project Reference for the project
• A publishable API key or legacy anon key from Supabase
• A Supabase Auth user that can sign in with email/password or magic link
• Row Level Security policies that allow that authenticated user to read the objects you want to sync
• The Supaflow metadata helper installed in your Supabase project

Supabase documents the project Data API and API keys in the Data REST API guide and API keys guide. For Auth setup, see Supabase's email/password guide and magic link guide.

Supported Objects

Supaflow discovers tables exposed through your Supabase project metadata helper and readable in the configured schemas.

Object	Sync Mode	Description
Tables	Full Refresh or Incremental	Tables in exposed schemas, subject to the authenticated user's RLS policies.

Views and foreign tables are not discovered by the Supabase source. If you need data from a view, materialize it into a table in your Supabase project.

Authentication

Supaflow authenticates with Supabase Auth. The source stores session credentials encrypted in Supaflow, refreshes the session when needed, and reads as the signed-in Supabase Auth user.

Sign-In Methods

Method	Status	Notes
Email/password	Default	Use an existing Supabase Auth user email and password.
Magic link	Supported	Sends a Supabase magic link to an existing user. The link must redirect back to Supaflow to finish setup.

Supaflow does not ask for, store, or use a Supabase service_role key. Secret API keys are rejected because they bypass Row Level Security.

Magic Link Redirect URL

If you use magic link sign-in, add the Supaflow callback URL to your Supabase Auth redirect allow list.

For the hosted Supaflow app, add:

https://app.supa-flow.io/api/supabase-auth/callback

For a local or self-hosted Supaflow app, use the equivalent callback URL for that environment.

Supabase's magic link guide explains that redirect URLs must be configured before they can be used as magic link destinations.

Metadata Helper

The Supabase source needs a Supaflow metadata helper installed in your project so Supaflow can discover schemas, tables, columns, and primary keys without using a service role key.

Run the Supaflow metadata helper SQL in your Supabase SQL Editor for each project you connect. The helper grants execute access to the authenticated role and returns metadata only; row reads still use the signed-in user's Supabase Auth session.

Supabase metadata helper

-- Supaflow metadata RPC for Supabase source connections.
--
-- Run this script in the Supabase SQL editor for each project that should be
-- readable through the Supabase Auth source connector. The function returns
-- table and column metadata to authenticated users without requiring a service
-- role key. Actual row reads still run through PostgREST with the user's
-- Supabase Auth JWT, so Row Level Security remains enforced for data access.

drop function if exists public.supaflow_supabase_metadata(text[], text[], text[], text[]);

create or replace function public.supaflow_supabase_metadata(
    metadata_level text default 'FULL'
)
returns table (
    table_schema text,
    table_name text,
    table_type text,
    table_description text,
    column_name text,
    ordinal_position integer,
    data_type text,
    udt_name text,
    is_nullable boolean,
    column_default text,
    character_maximum_length integer,
    numeric_precision integer,
    numeric_scale integer,
    datetime_precision integer,
    is_primary_key boolean,
    primary_key_ordinal integer,
    column_description text
)
language sql
stable
security definer
set search_path = pg_catalog, public
set statement_timeout = '60s'
as $$
with params as (
    select
        case
            when upper(coalesce(nullif(btrim(metadata_level), ''), 'FULL'))
                in ('SCHEMAS_ONLY', 'TABLES_ONLY', 'FULL')
            then upper(coalesce(nullif(btrim(metadata_level), ''), 'FULL'))
            else 'FULL'
        end as metadata_level,

        -- Configure Supaflow's allowed metadata scope here.
        -- Leave included_schemas empty to allow public plus PostgREST-exposed schemas.
        -- Use table names as either table_name or schema_name.table_name.
        array[]::text[] as included_schemas,
        array[]::text[] as excluded_schemas,
        array[]::text[] as included_tables,
        array[]::text[] as excluded_tables
),
config as (
    select
        p.metadata_level,
        array(
            select btrim(value)
            from unnest(p.included_schemas) as value
            where btrim(value) <> ''
        ) as included_schemas,
        array(
            select btrim(value)
            from unnest(p.excluded_schemas) as value
            where btrim(value) <> ''
        ) as excluded_schemas,
        array(
            select btrim(value)
            from unnest(p.included_tables) as value
            where btrim(value) <> ''
        ) as included_tables,
        array(
            select btrim(value)
            from unnest(p.excluded_tables) as value
            where btrim(value) <> ''
        ) as excluded_tables
    from params p
),
postgrest_schemas as (
    select btrim(value) as schema_name
    from regexp_split_to_table(
        coalesce(nullif(current_setting('pgrst.db_schemas', true), ''), 'public'),
        ','
    ) as value
    where btrim(value) <> ''
),
requested_schemas as (
    select unnest(c.included_schemas) as schema_name
    from config c
),
candidate_schemas as (
    select 'public'::text as schema_name
    union
    select schema_name from postgrest_schemas
    union
    select schema_name from requested_schemas
),
filtered_schemas as (
    select distinct n.nspname as schema_name
    from pg_namespace n
    join candidate_schemas cs on cs.schema_name = n.nspname
    cross join config c
    where (cardinality(c.included_schemas) = 0 or n.nspname = any(c.included_schemas))
      and (cardinality(c.excluded_schemas) = 0 or n.nspname <> all(c.excluded_schemas))
      and has_schema_privilege('authenticated', n.oid, 'USAGE')
),
tables_in_scope as (
    select
        ns.nspname::text as table_schema,
        cls.relname::text as table_name,
        'BASE TABLE'::text as table_type,
        cls.oid as table_oid,
        obj_description(cls.oid, 'pg_class')::text as table_description
    from pg_class cls
    join pg_namespace ns
      on ns.oid = cls.relnamespace
    join filtered_schemas fs
      on fs.schema_name = ns.nspname
    cross join config c
    -- Supaflow replication supports table relations only. Views and foreign
    -- tables are intentionally excluded from Supabase source discovery.
    where cls.relkind in ('r', 'p')
      and has_table_privilege('authenticated', cls.oid, 'SELECT')
      and (
          cardinality(c.included_tables) = 0
          or cls.relname = any(c.included_tables)
          or (ns.nspname || '.' || cls.relname) = any(c.included_tables)
      )
      and (
          cardinality(c.excluded_tables) = 0
          or (
              cls.relname <> all(c.excluded_tables)
              and (ns.nspname || '.' || cls.relname) <> all(c.excluded_tables)
          )
      )
),
primary_keys as (
    select
        t.table_oid,
        attr.attname::text as column_name,
        key_column.ordinality::integer as primary_key_ordinal
    from tables_in_scope t
    join pg_index idx
      on idx.indrelid = t.table_oid
     and idx.indisprimary
    join unnest(idx.indkey) with ordinality as key_column(attnum, ordinality)
      on true
    join pg_attribute attr
      on attr.attrelid = idx.indrelid
     and attr.attnum = key_column.attnum
),
schema_rows as (
    select
        fs.schema_name::text as table_schema,
        null::text as table_name,
        null::text as table_type,
        null::text as table_description,
        null::text as column_name,
        null::integer as ordinal_position,
        null::text as data_type,
        null::text as udt_name,
        null::boolean as is_nullable,
        null::text as column_default,
        null::integer as character_maximum_length,
        null::integer as numeric_precision,
        null::integer as numeric_scale,
        null::integer as datetime_precision,
        null::boolean as is_primary_key,
        null::integer as primary_key_ordinal,
        null::text as column_description
    from filtered_schemas fs
    cross join config c
    where c.metadata_level = 'SCHEMAS_ONLY'
),
table_rows as (
    select
        t.table_schema::text,
        t.table_name::text,
        t.table_type::text,
        t.table_description::text,
        null::text as column_name,
        null::integer as ordinal_position,
        null::text as data_type,
        null::text as udt_name,
        null::boolean as is_nullable,
        null::text as column_default,
        null::integer as character_maximum_length,
        null::integer as numeric_precision,
        null::integer as numeric_scale,
        null::integer as datetime_precision,
        null::boolean as is_primary_key,
        null::integer as primary_key_ordinal,
        null::text as column_description
    from tables_in_scope t
    cross join config c
    where c.metadata_level = 'TABLES_ONLY'
),
column_rows as (
    select
        t.table_schema::text,
        t.table_name::text,
        t.table_type::text,
        t.table_description::text,
        attr.attname::text as column_name,
        attr.attnum::integer as ordinal_position,
        format_type(attr.atttypid, attr.atttypmod)::text as data_type,
        typ.typname::text as udt_name,
        (not attr.attnotnull) as is_nullable,
        pg_get_expr(def.adbin, def.adrelid)::text as column_default,
        case
            when typ.typname in ('varchar', 'bpchar') and attr.atttypmod > 4
            then attr.atttypmod - 4
            else null
        end::integer as character_maximum_length,
        case
            when typ.typname = 'numeric' and attr.atttypmod >= 0
            then ((attr.atttypmod - 4) >> 16) & 65535
            else null
        end::integer as numeric_precision,
        case
            when typ.typname = 'numeric' and attr.atttypmod >= 0
            then (attr.atttypmod - 4) & 65535
            else null
        end::integer as numeric_scale,
        case
            when typ.typname in ('time', 'timetz', 'timestamp', 'timestamptz', 'interval')
                 and attr.atttypmod >= 0
            then attr.atttypmod
            else null
        end::integer as datetime_precision,
        (pk.column_name is not null) as is_primary_key,
        pk.primary_key_ordinal,
        col_description(t.table_oid, attr.attnum)::text as column_description
    from tables_in_scope t
    join pg_attribute attr
      on attr.attrelid = t.table_oid
     and attr.attnum > 0
     and not attr.attisdropped
    join pg_type typ
      on typ.oid = attr.atttypid
    left join pg_attrdef def
      on def.adrelid = attr.attrelid
     and def.adnum = attr.attnum
    cross join config cfg
    left join primary_keys pk
      on pk.table_oid = t.table_oid
     and pk.column_name = attr.attname
    where cfg.metadata_level = 'FULL'
)
select * from schema_rows
union all
select * from table_rows
union all
select * from column_rows
order by table_schema, table_name nulls first, ordinal_position nulls first;
$$;

revoke all on function public.supaflow_supabase_metadata(text) from public, anon, authenticated;
grant execute on function public.supaflow_supabase_metadata(text) to authenticated;

-- Make the new/updated RPC visible to PostgREST without waiting for the next
-- automatic schema-cache refresh.
notify pgrst, 'reload schema';

If Test & Save reports that the metadata helper is missing or not visible, run the helper SQL again and retry after Supabase refreshes the API schema cache.

Sync Modes

Full refresh: Supaflow reads all selected rows from the source object on each run.

Incremental sync: Supaflow can use cursor fields such as updated_at, modified_at, last_modified, or last_modified_at when they are present and reliable. Incremental runs fetch rows changed since the last cursor position.

For large full-refresh reads, keep a primary key on source tables. Tables without a primary key cannot be paged safely beyond one REST page because unordered offset pagination can duplicate or skip rows.

Configuration

In Supaflow, create a new Supabase source with these settings:

Project Reference*

The Supabase project reference from your project URL.
Example: abcdefghijklmnopqrst from https://abcdefghijklmnopqrst.supabase.co

Public API Key*

Your Supabase publishable key (sb_publishable_...) or legacy anon key.
Stored encrypted

Authentication Type*

Supabase Auth session. This is selected automatically for the Supabase source.

Sign in Method*

Choose email_password or magic_link. Email/password is the default.

Read Page Size

Rows fetched per Supabase read request.
Default: 1000
Range: 1 to 1000
Lower this value for very wide tables. For statement timeouts caused by Row Level Security scans, see Statement timeout.

Test & Save

After entering the project details and signing in with Supabase Auth, click Test & Save to verify your connection and save the source.

Rate Limiting and Performance

Supabase applies API limits to project API requests. Supaflow retries transient rate limiting with backoff, but large syncs can still take longer if the source project is rate limited.

Row Level Security policies are evaluated during reads. Complex policies or wide tables can make source reads slower. If a table hits the source project's Postgres statement_timeout, increase the timeout for the authenticated role or simplify the RLS policy for the syncing user; lowering Read Page Size may not avoid the timeout because Postgres can need to evaluate many rows before returning visible rows.

Schema Evolution

Supaflow refreshes schema metadata from the Supabase metadata helper.

• New tables and columns appear after metadata refresh
• Removed source columns stop appearing in refreshed source metadata
• Type changes are reflected after metadata refresh and may require pipeline review before the next sync

Troubleshooting

Metadata helper is missing or permission denied

Problem:

• Test & Save reports that the metadata helper is missing
• Test & Save reports permission denied for metadata discovery

Solutions:

1. Run the Supaflow metadata helper SQL in the Supabase SQL Editor for the project
2. Confirm the helper grants execute access to the authenticated role
3. Retry Test & Save after Supabase refreshes the API schema cache

Authentication failed

Problem:

• The source cannot sign in or refresh the Supabase Auth session

Solutions:

1. Confirm the user exists in Supabase Auth and can sign in to your application
2. For email/password, verify the password and any email-confirmation requirement in Supabase Auth
3. For magic link, open the latest link in the same browser session where you started setup
4. Reconnect the source if the refresh token has been revoked or expired

Secret key rejected

Problem:

• Supaflow rejects a key beginning with sb_secret_

Solutions:

1. Use a publishable key (sb_publishable_...) or legacy anon key
2. Do not use a Supabase secret key or legacy service_role key for this source
3. Confirm your RLS policies allow the authenticated user to read the intended rows

Tables or rows are missing

Problem:

• A table does not appear in schema discovery
• A view or foreign table does not appear in schema discovery
• A table appears, but fewer rows sync than expected

Solutions:

1. Confirm the schema is exposed through Supabase's Data API
2. Confirm the metadata helper includes the schema and table
3. If the missing object is a view or foreign table, copy or materialize the data into a table; the Supabase source discovers tables only
4. Confirm the authenticated user's RLS policies allow SELECT access to the expected rows
5. Test the same user session in your application to verify row visibility

Large table fails without a primary key

Problem:

• A large full-refresh table fails with a pagination error

Solutions:

1. Add a primary key to the source table
2. Use a reliable incremental cursor field where possible
3. If the object is small, keep the selected read within one page

Statement timeout

Problem:

• A read fails because the Supabase project's Postgres statement_timeout was exceeded

Solutions:

1. Increase the source project's timeout for the authenticated role. Expensive RLS policies can require Postgres to scan many rows before returning even one visible row, so lowering Read Page Size may not avoid the timeout.
2. Run these commands in the Supabase SQL Editor:

   alter role authenticated set statement_timeout = '60s';
   notify pgrst, 'reload config';

3. Review expensive Row Level Security policies for the authenticated role and simplify them where possible.
4. Reduce selected columns or split large objects into smaller pipelines if the read still returns very wide rows after the RLS check completes.

Support

Need help? Contact us at support@supa-flow.io