Configure Tableflow in Confluent Cloud

Tableflow in Confluent Cloud provides table-level settings you can configure, including snapshot retention for time-travel queries, data time-to-live (TTL) for automatically expiring old rows, and an error-handling mode for managing per-record materialization failures. You can also set a metadata column naming scheme for compatibility with query engines that reject $ characters in column names.

Snapshot retention

Snapshot retention involves managing metadata that enables you to run time-travel queries, which query a previous state of your table. Tableflow creates a snapshot every time it commits a change to your table. This includes any time Tableflow adds data to or updates data in your table, and when it performs maintenance tasks, like compaction.

Tableflow always maintains a minimum number of snapshots, but to control how long Tableflow retains additional snapshots before expiring them, set the retention_ms configuration. When a snapshot is past its expiration time or more than 100 snapshots exist, Tableflow asynchronously removes the snapshot from the table and any data files that no remaining snapshot needs.

Data TTL

With the data time-to-live (TTL) setting, you can configure Tableflow to automatically delete old data in your tables. Tableflow deletes table rows when they become older than the retention period that you specify. By setting the TTL, you can reduce storage costs for situations where you only need data for a limited time. For example, data TTL is useful if you analyze only recent data or if you run periodic aggregations, such as hourly or daily summaries.

Note

Data TTL in Tableflow is available to some Confluent customers as a Limited Availability feature. For those with access, Confluent fully supports data TTL, which is recommended for production use. To request access, contact Confluent Support.

Tableflow measures the age of a row based on the timestamp of the corresponding Apache Kafka® record.

Data TTL differs from snapshot retention, which manages metadata for time-travel queries but does not delete rows from your table.

How data TTL works

When you configure data TTL on a table, Tableflow asynchronously deletes each row that exceeds your retention period.

First, Tableflow logically deletes expired rows in your table after a waiting period. Tableflow waits up to 24 hours after at least one row in a Parquet file expires. Then, Tableflow logically deletes the expired rows, which makes them invisible to queries. If all rows in the file expire within that time, Tableflow deletes the file. If only some rows expire after 24 hours, Tableflow removes those rows. This action requires compaction, which means Tableflow creates a new file without the expired rows. The waiting period minimizes two impacts on query performance: the logical deletions and the compaction.

Subsequently, Tableflow physically deletes the Parquet file. After every snapshot that references a Parquet file expires, Tableflow physically deletes the file from storage. By waiting for snapshots to expire, Tableflow ensures that you can do time-travel queries for as long as your snapshot retention settings permit.

Configure data TTL

To configure the data TTL setting, use the Confluent Cloud API. Set the data_retention_ms parameter when you create or update a table. Specify the retention period in milliseconds. Tableflow rounds up this period to the nearest day.

The following example configures data TTL to retain data for 90 days:

{
  "spec": {
    "display_name": "high-volume-events",
    "config": {
      "data_retention_ms": "7776000000"
    },
    "table_formats": ["ICEBERG"],
    "environment": {"id": "env-xxxxx"},
    "kafka_cluster": {"id": "lkc-xxxxx"}
  }
}

Error-handling mode

Tableflow offers three modes for handling per-record materialization failures: suspend, skip, and log. The default mode is suspend. Assign the value to the error-handling.mode table property, for example:

ALTER TABLE my_table SET (
 'error-handling.mode' = 'log'
);

You can configure error-handling mode by using Tableflow configurations or the Apache Flink® error-handling.mode table property.

Suspend mode

The default mode, suspend, causes Tableflow to enter the Pause state whenever a record can’t be materialized and added to the table. This means that in situations where your topic ingests a corrupted record, Tableflow pauses processing on that record.

Failures that occur for reasons that are not record-specific always cause Tableflow to enter the Pause state, regardless of the configured error-handling.mode. This includes, but is not limited to, errors related to catalog or storage access and invalid schema changes.

Skip mode

When you set the error-handling mode to skip, Tableflow skips over records that fail to materialize. Tableflow reports the number of skipped records in the rows_skipped metric.

Log mode

When you set the error-handling mode to log, Tableflow sends records that fail to materialize to a dead letter queue (DLQ) topic. The DLQ is a Kafka topic in the same cluster as your Tableflow table. Tableflow also counts records that it sends to the DLQ in the rows_skipped metric.

Important

Failure handling with log mode is available only for topics that are configured with Avro or Protobuf schemas.

Error-handling log target

When the Tableflow error-handling mode is set to log, you can optionally set a specific topic by creating a new topic or using an existing topic for your DLQ. The DLQ topic must follow the DLQ topic schema.

If you don’t specify a DLQ topic, Tableflow uses a topic named error_log.

The account enabling or updating Tableflow to use the log error-handling mode must have write permissions on the DLQ topic. Otherwise, Tableflow encounters an error and enters the Pause state.

Assign the value to the error-handling.log.target table property, for example:

ALTER TABLE my_table SET (
 'error-handling.log.target' = 'my_error_topic'
);

You can configure the error-handling log target by using Tableflow configurations or the Flink error-handling.log.target table property.

DLQ topic schema

Tableflow DLQ topics must use the following Avro schema to operate correctly. If you create the topic as part of DLQ enablement, it’s automatically registered with this schema. For the full schema definitions in Avro, JSON Schema, and Protobuf (including both key and value schemas), see Schema definitions for manual registration.

DLQ topic schema

{
  "fields": [
    {
      "name": "error_timestamp",
      "type": {
        "logicalType": "timestamp-millis",
        "type": "long"
      }
    },
    {
      "name": "error_code",
      "type": "int"
    },
    {
      "name": "error_reason",
      "type": "string"
    },
    {
      "name": "error_message",
      "type": "string"
    },
    {
      "name": "error_details",
      "type": {
        "type": "map",
        "values": [
          "null",
          "string"
        ]
      }
    },
    {
      "name": "processor",
      "type": "string"
    },
    {
      "default": null,
      "name": "statement_name",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "affected_type",
      "type": "string"
    },
    {
      "default": null,
      "name": "affected_catalog",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "default": null,
      "name": "affected_database",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "default": null,
      "name": "affected_name",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "default": null,
      "name": "source_record",
      "type": [
        "null",
        {
          "fields": [
            {
              "default": null,
              "doc": "The topic of the Kafka source record.",
              "name": "topic",
              "type": [
                "null",
                "string"
              ]
            },
            {
              "default": null,
              "doc": "The partition of the Kafka source record.",
              "name": "partition",
              "type": [
                "null",
                "int"
              ]
            },
            {
              "default": null,
              "doc": "The offset of the Kafka source record.",
              "name": "offset",
              "type": [
                "null",
                "long"
              ]
            },
            {
              "default": null,
              "doc": "The timestamp of the Kafka source record. The value is in milliseconds since epoch. The specific meaning of the timestamp depends on timestamp_type.",
              "name": "timestamp",
              "type": [
                "null",
                {
                  "logicalType": "timestamp-millis",
                  "type": "long"
                }
              ]
            },
            {
              "default": null,
              "doc": "The type of the timestamp in the Kafka source record. Possible values are 'CREATE_TIME' and 'LOG_APPEND_TIME'.",
              "name": "timestamp_type",
              "type": [
                "null",
                "string"
              ]
            },
            {
              "default": null,
              "doc": "The headers of the Kafka source record. The keys are strings and values are byte arrays. Note that for multiple headers with the same key, only the first one is kept in the map.",
              "name": "headers",
              "type": [
                "null",
                {
                  "type": "map",
                  "values": [
                    "null",
                    "bytes"
                  ]
                }
              ]
            },
            {
              "default": null,
              "doc": "The key of the Kafka source record. May be null if the record has no key.",
              "name": "key",
              "type": [
                "null",
                "bytes"
              ]
            },
            {
              "default": null,
              "doc": "The value of the Kafka source record. May be null if the record has no value (tombstone record).",
              "name": "value",
              "type": [
                "null",
                "bytes"
              ]
            }
          ],
          "name": "stocks_realtime_feed_dlq_value_source_record",
          "type": "record"
        }
      ]
    }
  ],
  "name": "stocks_realtime_feed_dlq_value",
  "namespace": "org.apache.flink.avro.generated.record",
  "type": "record"
}

Metadata column naming

Tableflow adds metadata columns to every materialized table, such as $$topic, $$partition, and $$offset. Some query engines, including BigQuery, reject column names that contain a $ character, which can prevent you from querying these tables.

To support these engines, Tableflow can rename its metadata columns to a normalized cflt_metadata_ prefix instead of $$. The metadata_column_naming_scheme property controls this per Tableflow-enabled topic:

PORTABLE: Renames the columns to the normalized cflt_metadata_ prefix. For example, $$topic becomes cflt_metadata_topic, $$partition becomes cflt_metadata_partition, and $$offset becomes cflt_metadata_offset.
DEFAULT: Keeps the original $$ prefix.

The property values are case-sensitive. Use uppercase.

When you enable Tableflow on a topic, the Confluent Cloud Console selects PORTABLE by default. You can set or change this property from the Cloud Console or the Confluent Cloud API, either when you enable Tableflow or later on a table that’s already materialized.

If you change the scheme on a table that’s already materialized, the table then exposes only the columns for the selected scheme. For Iceberg tables, Tableflow applies this change as a metadata-only change, without rewriting any existing data.

The following example sets the metadata column naming scheme to PORTABLE:

{
  "spec": {
    "display_name": "high-volume-events",
    "config": {
      "metadata_column_naming_scheme": "PORTABLE"
    },
    "table_formats": ["ICEBERG"],
    "environment": {"id": "env-xxxxx"},
    "kafka_cluster": {"id": "lkc-xxxxx"}
  }
}

If you switch schemes, update any existing queries that reference the old metadata column names.