v0.29.0 → v0.30.4

# Using curl
curl -LsfS https://get.airbyte.com | bash -

# Or using Homebrew
brew upgrade abctl

abctl version

abctl local uninstall --persisted
rm -rf ~/.airbyte/abctl
abctl local install

airbyte-abctl-control-plane

docker exec -it airbyte-abctl-control-plane bash
# Then test connectivity:
curl https://ifconfig.io

abctl local install --chart-version=1.7.2

v0.30.4

docker exec -it airbyte-abctl-control-plane bash

root

curl <https://ifconfig.io>

curl https://ifconfig.io

airbyte-abctl-control-plane

kubectl get pods -A

ImagePullBackOff

ErrImagePull

CrashLoopBackOff

Pending

kubectl logs <pod_name> -n <namespace>

kubectl --kubeconfig ~/.airbyte/abctl/abctl.kubeconfig --namespace airbyte-abctl get pods

v0.30.4

abctl local install --low-resource-mode

kubectl get pods -A

Running

Completed

Running

ERROR: Timed out waiting for ingress

abctl

http://localhost:8000

kubectl --kubeconfig ~/.airbyte/abctl/abctl.kubeconfig -n airbyte-abctl get ingress

abctl

v0.30.4

kubectl --kubeconfig ~/.airbyte/abctl/abctl.kubeconfig -n airbyte-abctl get ingress

localhost

host.docker.internal

10.xx.xxx.xxx

80

http://ec2-xx.xx.xxx.xxx.ca-central-1.compute.amazonaws.com:8000

8000

0.0.0.0/0

--host

abctl

localhost

abctl local install --host ec2-xx.xx.xxx.xxx.ca-central-1.compute.amazonaws.com --insecure-cookies

--insecure-cookies

abctl

8000

My IP

0.0.0.0/0

--insecure-cookies

abctl local install --host ec2-xx.xx.xxx.xxx.ca-central-1.compute.amazonaws.com --insecure-cookies

abctl local install --host ec2-xx.xx.xxx.xxx.ca-central-1.compute.amazonaws.com --insecure-cookies

ERROR: Timed out waiting for ingress

abctl

abctl

abctl local uninstall
abctl local install --host ec2-xx.xx.xxx.xxx.ca-central-1.compute.amazonaws.com --insecure-cookies

http://ec2-xx.xx.xxx.xxx.ca-central-1.compute.amazonaws.com:8000

t3.xlarge

--low-resource-mode

2 x <number of concurrent connections>

t3.xlarge

rm -rf ~/.airbyte/

~/.airbyte/

~/.airbyte/abctl/abctl.kubeconfig

abctl local uninstall --persisted
rm -rf ~/.airbyte/abctl

~/.airbyte/abctl

~/.airbyte/

abctl local uninstall
abctl local install --host ec2-xx.xx.xxx.xxx.ca-central-1.compute.amazonaws.com --insecure-cookies

rm -rf ~/.airbyte/abctl

abctl

v0.30.4

--host

--insecure-cookies

8000

abctl

rm -rf ~/.airbyte/

rm -rf ~/.airbyte/

rm -rf ~/.airbyte/

rm -rf

updated_at

fetchSize

kubectl --kubeconfig ~/.airbyte/abctl/abctl.kubeconfig -n airbyte-abctl logs -f <source-pod-name>

_AIRBYTE_EMITTED_AT

_AIRBYTE_EMITTED_AT

_airbyte_extracted_at

ALTER TABLE <your_table> CLUSTER BY ();

ALTER TABLE <your_table> DROP COLUMN "_AIRBYTE_EMITTED_AT";

ALTER TABLE <your_table> ALTER SORTKEY NONE;

ALTER TABLE <your_schema>.<your_table> CLUSTER BY ();

ALTER TABLE <your_schema>.<your_table> DROP COLUMN "_AIRBYTE_EMITTED_AT";

4.0.47

4.0.47

Heartbeat not progressing

1123s → 1133s → 1143s → 1153s...

Received offset commit request on 'LSN{1C04/340000F8}'

SELECT slot_name, active, restart_lsn, confirmed_flush_lsn 
FROM pg_replication_slots;

confirmed_flush_lsn

1C04/340000F8

restart_lsn

1C04/340000C0

1C04/340000F8

confirmed_flush_lsn

3.7.2

3.7.2

3.7.2

CREATE TABLE airbyte_heartbeat (
    id SERIAL PRIMARY KEY,
    timestamp TIMESTAMP NOT NULL DEFAULT current_timestamp,
    text TEXT
);

ALTER PUBLICATION <publicationName> ADD TABLE airbyte_heartbeat;

INSERT INTO airbyte_heartbeat (text) VALUES ('heartbeat')

3.7.2

confirmed_flush_lsn

confirmed_flush_lsn

lsn

CREATE TABLE airbyte_heartbeat (
    id SERIAL PRIMARY KEY,
    timestamp TIMESTAMP NOT NULL DEFAULT current_timestamp,
    text TEXT
);

ALTER PUBLICATION <publicationName> ADD TABLE airbyte_heartbeat;

INSERT INTO airbyte_heartbeat (text) VALUES ('heartbeat')

Re-sync data

Fail sync

airbyte_heartbeat

airbyte_heartbeat

public

public

CREATE TABLE public.airbyte_heartbeat (
    id SERIAL PRIMARY KEY,
    timestamp TIMESTAMP NOT NULL DEFAULT current_timestamp,
    text TEXT
);

ALTER PUBLICATION <publicationName> ADD TABLE public.airbyte_heartbeat;

INSERT INTO public.airbyte_heartbeat (text) VALUES ('heartbeat')

CREATE TABLE <your_schema>.airbyte_heartbeat (
    id SERIAL PRIMARY KEY,
    timestamp TIMESTAMP NOT NULL DEFAULT current_timestamp,
    text TEXT
);

ALTER PUBLICATION <publicationName> ADD TABLE <your_schema>.airbyte_heartbeat;

INSERT INTO <your_schema>.airbyte_heartbeat (text) VALUES ('heartbeat')

184.62 MB | 522,451 records loaded | 26m 57s

Default User cancelled a sync

airbyte_extracted_at

airbyte_cdc_updated_at

_airbyte_extracted_at

_ab_cdc_updated_at

SELECT 
    _airbyte_extracted_at,
    _ab_cdc_updated_at,
    _airbyte_extracted_at = _ab_cdc_updated_at AS are_equal
FROM <your_schema>.<your_table>
LIMIT 100;

airbyte_heartbeat

Cannot drop column '_AIRBYTE_EMITTED_AT' which belongs to a clustering key

SELECT 
    table_schema,
    table_name,
    column_name
FROM information_schema.columns
WHERE column_name = '_AIRBYTE_EMITTED_AT'
AND table_schema = '<your_schema>'
ORDER BY table_schema, table_name;

SELECT 
    table_schema,
    table_name,
    column_name
FROM information_schema.columns
WHERE column_name = '_AIRBYTE_EMITTED_AT'
ORDER BY table_schema, table_name;

-- Step 1: Remove the clustering key
ALTER TABLE <your_schema>.<your_table> CLUSTER BY ();

-- Step 2: Drop the old column
ALTER TABLE <your_schema>.<your_table> DROP COLUMN "_AIRBYTE_EMITTED_AT";

airbyte_emitted_at column

_AIRBYTE_EMITTED_AT

_airbyte_extracted_at

-- Step 1: Generate ALTER statements for all affected tables
SELECT 
    'ALTER TABLE ' || table_schema || '.' || table_name || ' CLUSTER BY ();' AS remove_cluster,
    'ALTER TABLE ' || table_schema || '.' || table_name || ' DROP COLUMN "_AIRBYTE_EMITTED_AT";' AS drop_column
FROM information_schema.columns
WHERE column_name = '_AIRBYTE_EMITTED_AT'
AND table_schema = '<your_schema>'
ORDER BY table_schema, table_name;

_AIRBYTE_EMITTED_AT

CLUSTER BY ()

()

-- This removes the clustering key entirely
ALTER TABLE <your_schema>.<your_table> CLUSTER BY ();

-- Then drop the column
ALTER TABLE <your_schema>.<your_table> DROP COLUMN "_AIRBYTE_EMITTED_AT";

-- Setting a new clustering key
ALTER TABLE <your_schema>.<your_table> CLUSTER BY (some_other_column);

_AIRBYTE_EMITTED_AT

_SCD

_SCD

_SCD

_SCD

_SCD

_SCD

_SCD

_SCD

_SCD

_AIRBYTE_EMITTED_AT

_AIRBYTE_EMITTED_AT

_AIRBYTE_EMITTED_AT

_AIRBYTE_EMITTED_AT

_airbyte_extracted_at

_airbyte_emitted_at

_airbyte_extracted_at

Cannot drop column '_AIRBYTE_EMITTED_AT' which belongs to a clustering key

4.0.47+

-- Step 1: Remove clustering key FIRST
ALTER TABLE <your_schema>.<your_table> CLUSTER BY ();

-- Step 2: Then drop the column
ALTER TABLE <your_schema>.<your_table> DROP COLUMN "_AIRBYTE_EMITTED_AT";

_AIRBYTE_EMITTED_AT

_airbyte_extracted_at

_airbyte_emitted_at

_airbyte_extracted_at

_airbyte_raw_*

_airbyte_raw_*

_AIRBYTE_EMITTED_AT

SELECT 
    table_schema,
    table_name,
    column_name
FROM information_schema.columns
WHERE column_name = '_AIRBYTE_EMITTED_AT'
AND table_name LIKE '%airbyte_raw%'
AND table_schema = '<your_schema>'
ORDER BY table_schema, table_name;

Cannot drop column '_AIRBYTE_EMITTED_AT'

SELECT *
FROM information_schema.columns
WHERE column_name = '_AIRBYTE_EMITTED_AT'
AND table_schema = '<your_schema>';

Heartbeat not progressing

_AIRBYTE_EMITTED_AT

_AIRBYTE_EMITTED_AT

Cannot drop column '_AIRBYTE_EMITTED_AT'

flush.lsn.source = false

flush.lsn.source = false

Heartbeat not progressing

flush.lsn.source = false

DELETE /jobs/{jobId}

DELETE /jobs/{jobId}

_AIRBYTE_EMITTED_AT

Cannot drop column '_AIRBYTE_EMITTED_AT'

Heartbeat not progressing

confirmed_flush_lsn

Heartbeat not progressing

Heartbeat not progressing

_AIRBYTE_EMITTED_AT

Cannot drop column '_AIRBYTE_EMITTED_AT'

Heartbeat not progressing

CREATE TABLE public.airbyte_heartbeat (
    id SERIAL PRIMARY KEY,
    timestamp TIMESTAMP NOT NULL DEFAULT current_timestamp,
    text TEXT
);

ALTER PUBLICATION <publicationName> ADD TABLE public.airbyte_heartbeat;

INSERT INTO public.airbyte_heartbeat (text) VALUES ('heartbeat')

airbyte_heartbeat

_AIRBYTE_EMITTED_AT

Heartbeat not progressing

Cannot drop column '_AIRBYTE_EMITTED_AT'

Previous cancelled sync state
        ↓
Airbyte reads cursor value from last saved state
        ↓
Only fetches records newer than last cursor value
        ↓
Appends new records to Snowflake
        ↓
Deduplicates final table by primary key
        ↓
Saves new cursor state

airbyte_heartbeat

-- Grant usage on the schema
GRANT USAGE ON SCHEMA public TO <airbyte_user>;

-- Grant permissions on the heartbeat table
GRANT INSERT, SELECT, UPDATE, DELETE ON TABLE public.airbyte_heartbeat TO <airbyte_user>;

-- Grant permissions on the sequence (this is what's failing)
GRANT USAGE, SELECT ON SEQUENCE public.airbyte_heartbeat_id_seq TO <airbyte_user>;

<airbyte_user>

Could not execute heartbeat action

INSERT

UPDATE

permission denied for sequence airbyte_heartbeat_id_seq

INSERT

id

SERIAL

-- This is the missing permission that caused the error
GRANT USAGE, SELECT ON SEQUENCE public.airbyte_heartbeat_id_seq TO <airbyte_user>;

-- Table permissions
GRANT INSERT, UPDATE, SELECT ON TABLE public.airbyte_heartbeat TO <airbyte_user>;

-- Sequence permission (critical - this was the failing part)
GRANT USAGE, SELECT ON SEQUENCE public.airbyte_heartbeat_id_seq TO <airbyte_user>;

Could not execute heartbeat action

airbyte_heartbeat_id_seq

airbyte_heartbeat_id_seq

CREATE TABLE public.airbyte_heartbeat (
    id SERIAL PRIMARY KEY,
    timestamp TIMESTAMP NOT NULL DEFAULT current_timestamp,
    text TEXT
);

SERIAL

airbyte_heartbeat_id_seq

GRANT USAGE, SELECT ON SEQUENCE public.airbyte_heartbeat_id_seq TO <airbyte_user>;

SELECT sequence_name 
FROM information_schema.sequences 
WHERE sequence_name = 'airbyte_heartbeat_id_seq';

SELECT 
    table_schema,
    table_name,
    column_name
FROM information_schema.columns
WHERE column_name = '_AIRBYTE_EMITTED_AT'
AND table_schema = '<your_schema>'
ORDER BY table_schema, table_name;

SELECT * FROM pg_publication_tables 
WHERE tablename = 'airbyte_heartbeat';

GRANT USAGE, SELECT ON SEQUENCE public.airbyte_heartbeat_id_seq TO <airbyte_user>;

_AIRBYTE_EMITTED_AT

_airbyte_extracted_at

invalid identifier '_AIRBYTE_RAW_ID'

_AIRBYTE_RAW_ID

SnowflakeAbMetaAndGenIdMigration

_AIRBYTE_META

_AIRBYTE_GENERATION_ID

COPY INTO

SELECT table_name
FROM information_schema.tables t
WHERE table_schema = '<your_schema>'
AND NOT EXISTS (
    SELECT 1 
    FROM information_schema.columns c
    WHERE c.table_name = t.table_name
    AND c.table_schema = t.table_schema
    AND c.column_name = '_AIRBYTE_RAW_ID'
);

-- Add all missing meta columns
ALTER TABLE <your_schema>.<your_table> 
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_RAW_ID" VARCHAR,
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_EXTRACTED_AT" TIMESTAMP_TZ,
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_META" VARIANT,
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_GENERATION_ID" NUMBER;

ERROR Error closing resource io.airbyte.container.orchestrator.worker.io.LocalContainerAirbyteDestination@3054cdd3; recording failure but continuing.

invalid identifier '_AIRBYTE_RAW_ID'

_AIRBYTE_RAW_ID

ALTER TABLE <your_schema>.<your_table> 
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_RAW_ID" VARCHAR,
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_EXTRACTED_AT" TIMESTAMP_TZ,
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_META" VARIANT,
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_GENERATION_ID" NUMBER;

_AIRBYTE_RAW_ID

Error closing resource

_AIRBYTE_META

_AIRBYTE_GENERATION_ID

COPY INTO

_AIRBYTE_RAW_ID

_AIRBYTE_RAW_ID

_AIRBYTE_META

_AIRBYTE_GENERATION_ID

COPY INTO

_AIRBYTE_META

_AIRBYTE_GENERATION_ID

COPY INTO

SQL compilation error: invalid identifier '_AIRBYTE_META'

SELECT 
    table_name,
    MAX(CASE WHEN column_name = '_AIRBYTE_RAW_ID' THEN 'EXISTS' ELSE 'MISSING' END) AS raw_id,
    MAX(CASE WHEN column_name = '_AIRBYTE_EXTRACTED_AT' THEN 'EXISTS' ELSE 'MISSING' END) AS extracted_at,
    MAX(CASE WHEN column_name = '_AIRBYTE_META' THEN 'EXISTS' ELSE 'MISSING' END) AS meta,
    MAX(CASE WHEN column_name = '_AIRBYTE_GENERATION_ID' THEN 'EXISTS' ELSE 'MISSING' END) AS generation_id
FROM information_schema.columns
WHERE table_schema = '<your_schema>'
AND column_name IN ('_AIRBYTE_RAW_ID', '_AIRBYTE_EXTRACTED_AT', '_AIRBYTE_META', '_AIRBYTE_GENERATION_ID')
GROUP BY table_name
ORDER BY table_name;

ALTER TABLE <your_schema>.<your_table>
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_RAW_ID" VARCHAR,
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_EXTRACTED_AT" TIMESTAMP_TZ,
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_META" VARIANT,
    ADD COLUMN IF NOT EXISTS "_AIRBYTE_GENERATION_ID" NUMBER;

_AIRBYTE_RAW_ID

invalid identifier

_AIRBYTE_META

_AIRBYTE_GENERATION_ID

ALTER TABLE

SELECT 
    'ALTER TABLE ' || '"' || table_schema || '"."' || table_name || '"' ||
    ' ADD COLUMN IF NOT EXISTS "_AIRBYTE_RAW_ID" VARCHAR' ||
    ', ADD COLUMN IF NOT EXISTS "_AIRBYTE_EXTRACTED_AT" TIMESTAMP_TZ' ||
    ', ADD COLUMN IF NOT EXISTS "_AIRBYTE_META" VARIANT' ||
    ', ADD COLUMN IF NOT EXISTS "_AIRBYTE_GENERATION_ID" NUMBER;' AS alter_statement
FROM information_schema.tables
WHERE table_schema = '<your_schema>'
AND table_type = 'BASE TABLE'
AND NOT EXISTS (
    SELECT 1 
    FROM information_schema.columns c
    WHERE c.table_name = tables.table_name
    AND c.table_schema = tables.table_schema
    AND c.column_name = '_AIRBYTE_RAW_ID'
)
ORDER BY table_name;

ALTER TABLE

alter table

ADD COLUMN IF NOT EXISTS

ALTER TABLE

ALTER TABLE

ALTER TABLE "<your_schema>"."<your_table>" ADD COLUMN IF NOT EXISTS "_AIRBYTE_RAW_ID" VARCHAR;
ALTER TABLE "<your_schema>"."<your_table>" ADD COLUMN IF NOT EXISTS "_AIRBYTE_EXTRACTED_AT" TIMESTAMP_TZ;
ALTER TABLE "<your_schema>"."<your_table>" ADD COLUMN IF NOT EXISTS "_AIRBYTE_META" VARIANT;
ALTER TABLE "<your_schema>"."<your_table>" ADD COLUMN IF NOT EXISTS "_AIRBYTE_GENERATION_ID" NUMBER;

SELECT 
    'ALTER TABLE "' || table_schema || '"."' || table_name || '" ADD COLUMN IF NOT EXISTS "_AIRBYTE_RAW_ID" VARCHAR;' || chr(10) ||
    'ALTER TABLE "' || table_schema || '"."' || table_name || '" ADD COLUMN IF NOT EXISTS "_AIRBYTE_EXTRACTED_AT" TIMESTAMP_TZ;' || chr(10) ||
    'ALTER TABLE "' || table_schema || '"."' || table_name || '" ADD COLUMN IF NOT EXISTS "_AIRBYTE_META" VARIANT;' || chr(10) ||
    'ALTER TABLE "' || table_schema || '"."' || table_name || '" ADD COLUMN IF NOT EXISTS "_AIRBYTE_GENERATION_ID" NUMBER;' AS alter_statements
FROM information_schema.tables
WHERE table_schema = '<your_schema>'
AND table_type = 'BASE TABLE'
AND NOT EXISTS (
    SELECT 1 
    FROM information_schema.columns c
    WHERE c.table_name = tables.table_name
    AND c.table_schema = tables.table_schema
    AND c.column_name = '_AIRBYTE_RAW_ID'
)
ORDER BY table_name;

_SCD

_SCD

_SCD

_SCD

SELECT 
    'ALTER TABLE "' || table_schema || '"."' || table_name || '" ADD COLUMN IF NOT EXISTS "_AIRBYTE_RAW_ID" VARCHAR;' || chr(10) ||
    'ALTER TABLE "' || table_schema || '"."' || table_name || '" ADD COLUMN IF NOT EXISTS "_AIRBYTE_EXTRACTED_AT" TIMESTAMP_TZ;' || chr(10) ||
    'ALTER TABLE "' || table_schema || '"."' || table_name || '" ADD COLUMN IF NOT EXISTS "_AIRBYTE_META" VARIANT;' || chr(10) ||
    'ALTER TABLE "' || table_schema || '"."' || table_name || '" ADD COLUMN IF NOT EXISTS "_AIRBYTE_GENERATION_ID" NUMBER;' AS alter_statements
FROM information_schema.tables
WHERE table_schema = '<your_schema>'
AND table_type = 'BASE TABLE'
AND table_name NOT LIKE '%_SCD'
AND NOT EXISTS (
    SELECT 1 
    FROM information_schema.columns c
    WHERE c.table_name = tables.table_name
    AND c.table_schema = tables.table_schema
    AND c.column_name = '_AIRBYTE_RAW_ID'
)
ORDER BY table_name;

_SCD

_airbyte_extracted_at

_airbyte_emitted_at