Iceberg Target

​

Compatible versions

​

Key Features

• Any catalog: one connector type covers REST catalogs, AWS Glue, Amazon S3 Tables, a SQL catalog (hosted by your Popsink deployment, or your own PostgreSQL) and Hive Metastore.
• Vended or static storage credentials: on a REST catalog the connector can rely entirely on per-table credentials vended by the catalog. Otherwise it uses its own S3, ADLS or GCS credentials, or the pod’s cloud identity.
• Typed tables: columns are typed from the topic’s Avro schema. Numbers, decimals, dates and timestamps keep their type.
• Full change history: every insert, update and delete is appended as a row, with its operation and source timestamp.
• Managed current-state view on Snowflake: Popsink creates and maintains a <TABLE>_CURRENT view (one row per key, deletes removed) on Snowflake Horizon, or on any catalog Snowflake reads through a catalog-linked database.
• No duplicates on restart: the connector records the Kafka offsets it has committed inside the Iceberg snapshots, so a pod restart does not append the same batch twice.
• Built-in table maintenance: a daily job compacts small files, expires old snapshots, removes orphan files and can purge deleted keys.
• Backfill: MySQL and IBM i sources can load a table’s existing rows into an Iceberg subscription.

​

How It Works

1. The connector reads the subscription’s topic and turns each change event into a row: the mapped source columns, plus __op (the operation) and __ts_ms (when the change happened at the source).
2. Rows are buffered per table and committed as an Iceberg append. Each table is committed when it reaches 10,000 rows or after 60 seconds, whichever comes first. Kafka offsets are committed only once the Iceberg commit has landed, so nothing is lost if the pod stops.
3. The table therefore grows with every change: an UPDATE adds a row, and so does a DELETE. The connector never rewrites or deletes rows in the table itself.
4. The current state of the source table is a view that keeps the latest row of each primary key and hides deleted keys. See Current state: the merge view.
5. A scheduled maintenance job keeps the table healthy. See Table maintenance.

​

Prerequisites

• An Iceberg catalog the connector can reach, and a namespace in it. The connector can create the namespace if you allow it.
• A storage location for the data files (S3 bucket, ADLS Gen2 container or GCS bucket), unless your REST catalog vends storage credentials.
• Credentials for the catalog, and, when the catalog does not vend them, for the storage. The required permissions are listed per catalog below.
• A primary key on each subscription whose current state you want to read: flag the key columns in the subscription’s column mapping, or declare them on the datamodel. Without a key the table is still written, but there is nothing to collapse the history on.

​

Catalog Configuration

ALTER USER IF EXISTS <user> ADD PAT popsink_iceberg
  ROLE_RESTRICTION = '<ROLE>'
  DAYS_TO_EXPIRY = 90
  MINS_TO_BYPASS_NETWORK_POLICY_REQUIREMENT = 60;

ALTER DATABASE <db> SET EXTERNAL_VOLUME = <volume>;
ALTER DATABASE <db> SET CATALOG = 'SNOWFLAKE';
GRANT USAGE ON EXTERNAL VOLUME <volume> TO ROLE <ROLE>;

{
  "Effect": "Allow",
  "Action": ["glue:GetDatabase", "glue:CreateDatabase", "glue:GetTable", "glue:GetTables",
             "glue:CreateTable", "glue:UpdateTable",
             "s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:ListBucket"],
  "Resource": ["arn:aws:glue:<region>:<account>:catalog",
               "arn:aws:glue:<region>:<account>:database/<namespace>",
               "arn:aws:glue:<region>:<account>:table/<namespace>/*",
               "arn:aws:s3:::<bucket>", "arn:aws:s3:::<bucket>/*"]
}

python3 -c "import hmac,hashlib;print(hmac.new(b'<db password>', b'popsink-iceberg-catalog:<db user>', hashlib.sha256).hexdigest())"

CREATE ROLE "<db user>_iceberg" LOGIN PASSWORD '<derived password>';
GRANT CONNECT ON DATABASE "<db>" TO "<db user>_iceberg";
CREATE SCHEMA IF NOT EXISTS iceberg_catalog;
GRANT USAGE, CREATE ON SCHEMA iceberg_catalog TO "<db user>_iceberg";
ALTER ROLE "<db user>_iceberg" IN DATABASE "<db>" SET search_path = iceberg_catalog;

​

Storage Credentials

• Vended by the catalog, per table (REST catalogs only, default): no storage credential on the connector.
• The pod’s own identity (other catalogs, default): the cloud identity of the worker pod (IRSA / EKS Pod Identity, Workload Identity…).
• Static credentials of this connector: pick the storage type and authentication below.

​

Proxy

​

Table Defaults

​

Connection Test

​

Subscription Options

​

What Is Written

​

Column types

​

Schema evolution

• New source columns are added to the table automatically, as optional columns.
• Columns are never dropped or retyped. The type a column was committed with wins: each value is cast to it ("00123" into a long column becomes 123), and a value that cannot be cast without losing information ("1.5" into a long column) refuses the whole row. Refused rows are counted as skipped in the subscription’s metrics and logged without their value. They are not retried.
• A table that already exists keeps the types it was created with.

​

Current State: the Merge View

1. Group the rows by the subscription’s primary key.
2. Keep the latest change of each key: the highest __ts_ms (the source change time). With metadata columns on, __kafka_offset breaks ties between two changes made in the same millisecond.
3. Remove deleted keys: if the latest change of a key is a delete (__op = 'd'), the key is hidden. This filter runs after the ranking: filtering deletes first would bring a deleted row’s previous version back.

​

Managed view on Snowflake

CREATE VIEW "<db>"."<schema>"."orders_current"
COMMENT = 'popsink-current-state:v1 … maintained by Popsink'
AS
SELECT * FROM (
  SELECT * FROM "<db>"."<schema>"."orders"
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY "id"
    ORDER BY "__ts_ms" DESC NULLS LAST, "__kafka_offset" DESC NULLS LAST
  ) = 1
)
WHERE "__op" IS DISTINCT FROM 'd';

• PARTITION BY lists the subscription’s primary key columns. The __kafka_offset tie-breaker is added only when the table has metadata columns.
• A subscription gets a view only if it has a primary key. Its key columns, __op and __ts_ms must exist in the table: until they do, the view waits.
• Snowflake freezes SELECT * when a view is created, so the view is rebuilt when the table gains a column. Each build is checked against the table before it counts as built. Until then the view may lack the newest column.
• Popsink only manages a view whose comment starts with popsink-current-state:v1. A view of the same name that Popsink did not create, for example one you built yourself, is left as it is and only logged: drop it to let Popsink take over. Grants on a Popsink view are kept across rebuilds.
• If the view cannot be built (a missing grant, for instance), the table is still written: the subscription shows the view as degraded and the connector retries, after 1 hour for a missing grant.
• The view lives in a single-level namespace: a nested namespace gets no view.

GRANT CREATE VIEW ON SCHEMA <db>.<schema> TO ROLE <ROLE>;

GRANT USAGE ON DATABASE <view_db> TO ROLE <ROLE>;
GRANT USAGE, CREATE VIEW ON SCHEMA <view_db>.<view_schema> TO ROLE <ROLE>;
GRANT USAGE ON DATABASE <linked_db> TO ROLE <ROLE>;
GRANT USAGE ON SCHEMA <linked_db>."<namespace>" TO ROLE <ROLE>;
GRANT SELECT ON ALL ICEBERG TABLES IN SCHEMA <linked_db>."<namespace>" TO ROLE <ROLE>;
GRANT SELECT ON FUTURE ICEBERG TABLES IN SCHEMA <linked_db>."<namespace>" TO ROLE <ROLE>;

​

Creating the view yourself

CREATE OR REPLACE VIEW MY_DB.MY_SCHEMA.ORDERS_CURRENT AS
SELECT * EXCLUDE (_ppsk_rank)
FROM (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY "id"
      ORDER BY "__ts_ms" DESC NULLS LAST, "__kafka_offset" DESC NULLS LAST
    ) AS _ppsk_rank
  FROM MY_DB.MY_SCHEMA."orders"
)
WHERE _ppsk_rank = 1
  AND "__op" IS DISTINCT FROM 'd';

CREATE OR REPLACE VIEW my_catalog.my_schema.orders_current AS
SELECT * EXCEPT (_ppsk_rank)
FROM (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY `id`
      ORDER BY `__ts_ms` DESC NULLS LAST, `__kafka_offset` DESC NULLS LAST
    ) AS _ppsk_rank
  FROM my_catalog.my_schema.orders
)
WHERE _ppsk_rank = 1
  AND `__op` IS DISTINCT FROM 'd';

CREATE OR REPLACE VIEW iceberg.bronze.orders_current AS
SELECT id, status, amount, __op, __ts_ms   -- list your columns: Trino has no SELECT * EXCEPT
FROM (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY id
      ORDER BY __ts_ms DESC NULLS LAST, __kafka_offset DESC NULLS LAST
    ) AS _ppsk_rank
  FROM iceberg.bronze.orders
)
WHERE _ppsk_rank = 1
  AND __op IS DISTINCT FROM 'd';

​

Table Maintenance

1. Purges deleted keys (opt-in): removes every row of a key whose latest change is a delete, so its data leaves storage once older snapshots expire. A key re-inserted after its delete keeps its newer rows. Skipped on tables without a primary key.
2. Compacts small data files into files of about 128 MB, without changing a row.
3. Expires snapshots older than 7 days, always keeping the last 20.
4. Deletes orphan files, files no snapshot references, once they are older than 72 hours. Skipped on Amazon S3 Tables, which does this itself.

​

Backfill

• The subscription needs a primary key: without one there is no merge view to collapse the loaded rows with the streamed ones.
• Truncate history is not supported: the connector only appends, so a reload adds rows and the merge view keeps the newest row per key.
• Sources reached through an SSH tunnel, and IBM i multi-member files, are not supported yet.
• An IBM i table is first read into the backfill’s own scratch space (up to 55 GiB compressed), then loaded.
• A backfill does not resume: relaunching it appends the rows again, and the merge view collapses them.

​

Restarts and Replays

• Restarts do not duplicate data. Each append records the last Kafka offset it holds in the Iceberg snapshot. On restart, the connector skips rows the table already holds, so a pod restart, a rebalance or an append that landed but answered an error is not written twice.
• A replay appends the history again. Replaying a pipeline or a subscription, or triggering a backfill from the beginning of the topic, starts a new delivery: every event Kafka still retains is appended a second time. The merge view still returns the right current state, since duplicates share a key, __ts_ms and __op, but the history table holds both copies.
• Rows refused by typing are skipped, not retried (see Schema evolution).
• A table that hits an error no retry can fix (an invalid table definition, a missing namespace) is quarantined: it stops buffering, its rows are dropped and the connector reports the error. Other tables keep flowing.

​

Upgrading from the Previous Iceberg Target

• Polaris connectors become Iceberg REST connectors with the same catalog, tables and Kafka offsets. Nothing is replayed.
• Former Azure-only connectors keep the same storage account, container, service principal or connection string, and proxy. Their catalog becomes a SQL catalog hosted by the data plane:
  • New tables, at a new path. The old target kept its catalog in a file inside its pod, so there is nothing to carry over. Tables are created again under <container>/popsink/popsink/<table>, in namespace popsink. The old files stay where they were, untouched: point readers at the new path. Connectors that wrote to the same container share one catalog (popsink-<account>-<container>).
  • Only what Kafka still holds is reloaded. Every subscription replays its topic from the beginning. Rows older than the topic’s retention exist only in the old tables.
  • Load columns change. Migrated subscriptions get __kafka_partition, __kafka_offset, __kafka_timestamp and __ingested_at in place of the previous custom columns, which are not reproduced.

​

Limitations

• Iceberg format v1 is not supported.
• Format version, partitioning, table properties and metadata columns apply only when a table is created.
• Columns are never dropped or retyped. Nested records and arrays are written as JSON text. Avro timestamps lose their sub-second part.
• The managed merge view is available on Snowflake only, for single-level namespaces.
• The SQL catalog supports PostgreSQL only. A Hadoop / filesystem catalog is not supported.
• The proxy option covers the Entra ID token call only. There is no proxy for catalog, AWS or GCS traffic.
• The maintenance job cannot use an S3 role that requires an external ID.