Ctrlk
WarpStream.comSlackDiscordContact UsCreate Account
For the complete documentation index, see llms.txt. This page is also available as Markdown.

Tableflow Setup

This page describes how to setup WarpStream Tableflow.

Introduction

Tableflow automates the tedious process of transforming a topic in an Apache Kafka-compatible data streaming system into an Apache Iceberg table. Instead of writing custom code and manually configuring a data pipeline for each table you want to build, Tableflow allows you to declaratively specify which topics to build tables from and what schema and data format to expect. When schemas inevitably need to change, you can update the schema in Tableflow's editor and WarpStream will handle the schema migration automatically.

Compaction and table maintenance is included out-of-the-box with no tuning required. Tableflow continuously compacts the table in the background with intelligent heuristics to ensure readers get the best performance.

Tableflow is available as Bring-Your-Own-Cloud (BYOC) where the compute and storage live inside your cloud account inside your VPC. The raw data for your table is only ever stored inside your object storage bucket and never leaves your VPC during the table ingestion and maintenance process. Tableflow maintains a metadata store inside WarpStream Cloud as the Iceberg metadata layer that is periodically synced into your object storage bucket.

Getting Started

To get started with Tableflow, you first need to create a Tableflow cluster from the WarpStream Console, or using one of infrastructure-as-code deployment options. The WarpStream Agents that join this cluster will only perform Tableflow operations and do not expose the Apache Kafka protocol. Please refer to our other documentation for how to install and configure the WarpStream Agents in your environment as the process does not differ for Tableflow.

As part of deploying the Agents, you'll also need to setup and configure an object storage bucket and/or provide the Agents with access to one of your existing buckets. See our object storage configuration documentation for more details on that.

Once the Agents are running, you can open the Configuration table and start defining your source clusters, topics, tables, and schemas.

Managed Tables

Tableflow tables are fully managed by WarpStream, or what we call "managed tables". You cannot use another system for performing writes, compactions, or other table maintenance operations. This is in contrast to a connector-based approach where you would be forced to combine multiple distinct systems or operations together to implement all of these functions.

Configuration

Tableflow is configured with and is fully controllable from a single YAML file which can be edited through the WarpStream console or the Pipelines API.

Overview

Currently, there are two methods for defining schemas:

  • an inline mode which doesn't require using an external Schema Registry. This supports all schema types (JSON, Avro, Protobuf). The full schema must be fully contained in the YAML config: you cannot import external schemas with this mode.

  • a schema_registry mode which requires using an external Schema Registry. Protobuf and JSON schemas are supported from agent version v813+, Avro schemas are supported from agent version v820+. When using the Schema Registry, a schema may reference and import other schemas (as long as they are also registered in the Schema Registry), and Tableflow will resolve the final schema.

An example YAML for the inline mode is the following:

An example YAML for the schema_registry mode is the following:

The YAML specifies

  • The source clusters Tableflow should connect to.

  • The schema registries Tableflow should connect to (for the schema_registry mode).

  • For each cluster, the topic that Tableflow should create Iceberg tables from.

  • For each topic, the schema to deserialize the Kafka records with, either as an inline schema, or as referenced by a subject of a schema registry.

  • The destination bucket to store the table.

Configure Source Clusters

Source clusters are the Apache Kafka-compatible systems like WarpStream that store the topics you'd like to convert to tables. You define clusters by giving them a name, a list of brokers, and credentials if they are needed. You define source clusters at the root of the configuration YAML.

You can define multiple source clusters so a single Tableflow cluster can centralize data from multiple clusters into one unified place.

Configure Schema Registries

Schema registries are the Confluent-compatible schema registries that store the schemas you'd like to use to deserialize the data from your topics. In schema_mode: schema_registry, Protobuf and JSON tables require agent v813+ and Avro tables require agent v820+. You define schema registries by giving them a name, a URL and credentials if they are needed. You define schema registries at the root of the configuration YAML.

All credential fields reference environment variable names. The fields you may define for the credentials are the following:

Field
Description

username_env

Env var holding the basic-auth username. Must be set together with password_env.

password_env

Env var holding the basic-auth password. Must be set together with username_env.

use_tls

Enables TLS / mTLS for the registry connection.

mtls_client_cert_env

Env var holding the path to the client certificate (PEM). Must be set together with mtls_client_private_key_env.

mtls_client_private_key_env

Env var holding the path to the client private key (PEM). Must be set together with mtls_client_cert_env.

mtls_client_private_key_password_env

Env var holding the private key password. Only allowed if mtls_client_private_key_env is also set.

server_ca_cert_env

Env var holding the path to the server CA certificate (PEM).

Please note that for all those credentials, contrary to the Configure Connection and Credentials section below, the agents do not automatically add a TABLEFLOW_ prefix to the values of the fields before using the environment variables. So the environment variables in the Agent should be exactly the same as those stated in the configuration.

You can define multiple schema registries.

Configure Connection and Credentials

If credentials are needed to connect to the Kafka cluster, the connection information can be provided under the credentials block for each cluster.

Note that if the source Kafka cluster is a WarpStream cluster, credentials still need to be provided if authentication is required. This is different from the Managed Data Pipelines setup where credentials are injected automatically.

TLS

  • use_tls specifies whether the Agents should use TLS when connecting to your source clusters.

  • tls_insecure_skip_verify specifies whether a client verifies the server's certificate chain and host name.

SASL

  • Both the sasl_username_env and the sasl_password_env fields refer to environment variable names. The Agents will append a TABLEFLOW_ prefix to the values of the fields before using the environment variables, so the environment variables in the Agent should be configured as TABLEFLOW_SASL_USERNAME_ENV_VAR and TABLEFLOW_SASL_PASSWORD_ENV_VAR respectively.

  • The default value of sasl_mechanism is plain. Supported mechanisms include: plain, scram-256, and scram-512.

mTLS PEM encoded certs

  • The mtls_client_cert_env, mtls_client_key_env, and mtls_server_ca_cert_env fields refer to environment variable names. The Agents will append a TABLEFLOW_ prefix to the values of the fields before using the environment variables, so the environment variables in the Agent should be configured as MTLS_CERT_PATH_ENV_VAR, MTLS_KEY_PATH_ENV_VAR, and MTLS_SERVER_CA_CERT_PATH_ENV_VAR respectively.

  • mtls_client_cert_env specifies the environment variable that contains the path to the X.509 certificate file in PEM format.

  • mtls_client_key_env specifies the environment variable that contains the path to the X.509 private key file in PEM format.

  • mtls_server_ca_cert_env is optional and specifies the environment variable that contains the path to the X.509 certificate file in PEM format for the client certificate authority. "If not specified, the host's root certificate pool will be used for client certificate verification.

Configure the Destination Bucket URL

To specify where your table data should be stored, use the destination_bucket_url field at the root of the configuration YAML. This configures the default destination bucket URL for all tables.

Alternatively, you can specify per-table bucket URL overrides within each tables configuration:

The destination bucket URL for a table can only be changed if a table is completely empty. This means that once your table has started ingesting data successfully, the destination bucket cannot be changed.

Check our object storage configuration documentation for more details on how to configure this URL for various different cloud providers, as well as for a complete list of permissions that the Agents will require.

Note that Tables will be created under the <bucket-name>/warpstream/_tableflow path. Optionally, a prefix can be specified in the bucket URL, which will result in Tables being created under the <bucket-name>/prefix/warpstream/_tableflow path.

Configure Topics and Tables

The next step is defining your tables and topics. Each table you define has exactly one source topic, and the table will be named the same as <cluster-name>+<topic+name>.

Tableflow currently supports append-only tables. If you ingest data from a compacted topic in the source cluster, rows will not be deduplicated and any tombstones may not comply with the schema. Support for compacted topics is coming soon.

In both inline and schema_registry modes, the following must be provided:

  • the source_cluster_name to which TableFlow must connect. It must match the name of one source clusters declared (see Source Clusters section)

  • the source_topic from which TableFlow must ingest records.

  • the source_format of the records (which can be avro, json or protobuf).

  • the schema_mode: whether inline or schema_registry.

The schema_registry mode also requires a schema_registry section containing:

  • the name of the source schema registry. It must match the name of one source schema registries declared (see Source Schema Registry section)

  • the subject of that schema registry in which all versions of your schema will be registered.

An example declaration for the schema_registry mode is the following:

As for the inline mode, the schema must be provided (see the Schema Definitions section).

An example declaration for the inline mode is the following:

Note that the Kafka Topic is read with an IsolationLevel of read_committed. It's currently not possible to change it. Please ask warpstream support if you are interested in creating Tableflow Tables from read_uncommitted data.

Schema Definitions

Tableflow uses schemas for two related but different purposes:

  • input_schema tells Tableflow how to decode records from the source Kafka topic.

  • schema defines the final Iceberg table schema that will be written and queried (the output schema).

schema_mode controls where schema definitions come from:

  • inline: schema is declared directly in the Tableflow config.

  • schema_registry: schema is declared in an external Schema Registry.

The guidance in this section is about how to declare schemas when using schema_mode: inline.

For input_schema, provide a raw schema string in the same format as your source records:

  • If your records are JSON, provide a JSON Schema.

  • If your records are Avro, provide an Avro schema.

  • If your records are Protobuf, provide a .proto schema.

Schema Definition for the schema_registry mode

For the schema_registry mode, there is no need to declare a schema inline as it will be fetched from the external Schema Registry. However, each table requires a schema_registry section containing the name of the Schema Registry to connect to, and the subject that registers every schema version to fetch.

Recommended Default: Define Only input_schema

In most cases, defining only input_schema is enough. Tableflow uses the input schema to decode source records and infer the table schema automatically.

Use this pattern when:

  • The table should have the same logical shape as the source records.

  • You are not using transforms that rename, flatten, remove, or add fields.

  • You are happy with Tableflow choosing the corresponding table types for your input types.

Here, input_schema is a raw JSON Schema string because source_format is json. Tableflow infers the Iceberg table schema from the input schema, so there is no need to declare a separate output schema.

Example: Json Input Schema

Supported types are:

boolean, int, long, float, double, decimal, date, time, timestamp, timestamptz, string, uuid, fixed, and binary.

Required and Nullable Fields (Agent v801+)

Fields listed in the required array are validated at decode time: if a required field is missing from the JSON record, the record is rejected. By default, all fields are optional.

To declare a field as nullable, use a type array with "null":

In this example, event_id is required and non-nullable — the record is rejected if it is missing or null. nickname is required and nullable — {"nickname": null} is accepted but {} (missing) is rejected. user_id is optional — it can be missing, null, or present.

The combination of required and nullable controls validation as follows:

Configuration

Value present

null

Missing

Optional (default)

accepted

accepted

accepted

Required ("required" array)

accepted

rejected

rejected

Required + Nullable ("required" + ["type", "null"])

accepted

accepted

rejected

In Iceberg, there is no distinction between nullable and optional — a column is either required or optional. Tableflow maps this as: a field is optional in Iceberg if it is not in the required array or if it is nullable. Requiredness and nullability are only distinguished at JSON decode time.

Map keys in JSON

Note that map keys can only be string and are declared using additionalProperties in json inline schemas:

The additionalProperties must contain a valid type and cannot be mixed with explicit fields. The following is invalid:

Example: Avro Input Schema

Supported types are:

boolean, int, long, float, double, decimal, date, time, timestamp, timestamptz, string, uuid, fixed, and binary.

Note that map keys can only be string.

Example: Protobuf Input Schema (Agent v796+)

When source_format is protobuf, set wire_format based on the payload encoding:

  • raw: protobuf binary payload with no prefix

  • confluent: Confluent wire format (magic byte + schema ID prefix)

The list of supported types are:

boolean, int32, sint32, uint32, fixed32, sfixed32, int64, sint64, uint64, fixed64, sfixed64, float, double, string, and bytes .

Enum values are stored by name in Iceberg but identified by number on the wire. This has implications for schema evolution:

  • adding new enum values is safe

  • renaming enum values is forbidden in WarpStream's TableFlow because renaming would cause inconsistent data (old records would have old names, new records would have new names)

  • removing old enum values is safe. But note that if we decode a record whose number is not in the current schema, it will be stored in Iceberg as the number in string form (e.g. "99")

Note: WarpStream's TableFlow also validates that sibling enum fields have identical sets if they share any value name. This prevents accidental inconsistencies between different fields using the same enum type.

The Protobuf types are converted as follows to the Iceberg types:

Protobuf Type
Iceberg Type
Notes

boolean

boolean

int32

integer

sint32

integer

sfixed32

integer

uint32

decimal(10,0)

Stored as decimal(10,0) to prevent overflow (max uint32 > max int32)

fixed32

decimal(10,0)

Stored as decimal(10,0) to prevent overflow (max uint32 > max int32)

int64

long

sint64

long

sfixed64

long

uint64

decimal(20,0)

Stored as decimal(20,0) to prevent overflow (max uint64 > max int64)

fixed64

decimal(20,0)

Stored as decimal(20,0) to prevent overflow (max uint64 > max int64)

float

float

double

double

enum

string

Stored as the enum value name

string

string

bytes

binary

message

struct

map

map

repeated

list

oneof

struct

Converted to a struct where each option is an optional field

google.protobuf.Timestamp

timestamptz

Nanosecond precision is truncated to microseconds (Iceberg limitation)

Note: Iceberg does not have unsigned integer types. To prevent overflow when storing large unsigned values:

  • uint32 and fixed32 are stored as decimal(10,0) (the max uint32 number 4,294,967,295 has 10 digits)

  • uint64 and fixed64 are stored as decimal(20,0) (the max uint64 number 18,446,744,073,709,551,615 has 20 digits)

JSON Schema Features Currently Not Supported

When using JSON input_schema, Tableflow currently rejects JSON Schema documents that use the following keywords:

  • prefixItems

  • contains

  • patternProperties

  • dependentRequired

  • dependentSchemas

  • if

  • then

  • else

  • $defs

  • $ref

  • allOf

  • oneOf

  • anyOf

If any of these are present, schema conversion fails with an unsupported-feature validation error.

When to Declare schema (Output Schema)

Define schema only when you need control over the final table shape. Common reasons include:

  • You use transforms that change the record shape (rename, flatten, remove, or add fields).

  • You want full control over the table type chosen for a field (for example, storing a JSON number as a specific decimal precision and scale).

  • You want to control nullability explicitly.

The input schema describes records before transforms. The output schema describes records after transforms. Conceptually:

  1. Source Kafka record

  2. → decoded using input_schema

  3. → transformed using transforms, if configured

  4. → written using schema, if configured

  5. → otherwise written using the schema inferred from input_schema

When you define schema, use a YAML object with a fields list. The types here are expressed using [Iceberg types](nested structure explicitly) so do not define schema as a raw JSON Schema, Avro schema, or .proto string. Raw schema strings should only be used for input_schema.

In this example:

  • input_schema matches the nested source record and is a raw Avro schema string.

  • The transform flattens the record.

  • schema describes the flattened table as a YAML object.

Output Schema Format

When you define schema, use YAML:

Supported Field Types

Primitive Types

Each field has a name, a type, and an optional optional flag (defaults to false):

Supported primitive types are Iceberg primitive types: int, long, float, double, string, boolean, date, timestamp, timestamptz, uuid, and binary.

Parameterized types are also supported:

  • decimal — requires decimal_precision (1–38) and decimal_scale (0 to precision):

  • fixed — requires fixed_length:

Struct

A struct is a tuple of named, typed fields. Fields inside a struct follow the same rules as top-level fields and can be of any type, including other structs:

Map

A map is a key/value pair. The key field and value field are defined as child fields. Map keys are required; map values can be optional. Keys must be of type string:

List

A list contains a single child field named element. Elements can be of any type:

Lists of structs are also supported:

Field IDs Are Deprecated

Do not define field IDs in new Tableflow schemas. Tableflow assigns and manages field IDs internally. User-defined field IDs are deprecated and will be ignored. Existing legacy examples that include id should not be copied into new configurations.

This applies to table schema field IDs only. Protobuf field numbers inside a .proto input_schema are still part of the Protobuf schema and should remain there.

Deprecated schema definitions

This section contains documentation for the old way of declaring schemas definitions in tableflow. This was cumbersome because it required you to manually specify field-ids and used a schema definition that didn't necessarily map 1 to 1 with the type used to store your data. We still support it for backward compatibility but you shouldn't use it.

As shown in the above example, schemas specified with the inline mode contain a list of fields. Each field is named and has a unique integer id that will be mapped to the field ID for your Iceberg table as well as a type that will be used as the Iceberg date type for the corresponding column.

Primitive Types

The syntax for defining a primitive field looks like the following:

where field-type is one of the supported fields for your input record type (refer to the Protobuf / Avro / Json sections above).

Nested Types

For Avro and JSON only

A struct is specified as a tuple of typed values. Each field in the tuple is named and has an integer id that is unique in the table schema. Fields can be of any type.

For Avro, JSON and Protobuf

A map is specified as a key/value pair. Both the key field and value field have an integer id that is unique in the table schema. While map values can be either optional or required, map keys are required. For both Avro and JSON schemas, map keys can only be of type string, but values can be of any type. For Protobuf, map keys can be of any map key type allowed by the Protobuf specs, namely: string, int32, int64, uint32, uint64, sint32, sint64, sfixed32, sfixed64, fixed32, fixed64, boolean.

Note that for Protobuf, the proto_field_number must be added to the map field but should not be set for the key nor value. So a full example for a Protobuf map is:

A list (in JSON and Avro) or repeated (in Protobuf) is specified with a single element field. The element field is named element and has an integer id that is unique in the table schema. Elements can be either optional or required and can be of any type.

Note that for Protobuf, the proto_field_number must be added to the repeated field but should not be set for the element. So a full example for a Protobuf repeated is:

For Avro only

AVRO's binary encoding does not include field names nor type information and instead values are concatenated strictly based on the schema. Consequently, the decoder interprets the stream of bytes strictly according to the sequence defined in the schema.

You must list fields in your table schema in the same order as in the producer’s AVRO schema. Reordering fields (e.g. by id or by logical group) can cause decode failures such as avro: ReadBool: invalid bool or unexpected EOF. If you see these errors, compare your schema field order to the producer’s (e.g. the .avsc file or the schema in your schema registry) and align the order.

For Protobuf only

A message is defined using type: message. A proto_field_number must be provided for every nested field.

An enum is stored as a string in the Iceberg table using the enum value name. You must define all enum values with their corresponding numbers.

Enum values are stored by name in Iceberg but identified by number on the wire. This has implications for schema evolution:

  • adding new enum values is safe

  • renaming enum values is forbidden in WarpStream's TableFlow because renaming would cause inconsistent data (old records would have old names, new records would have new names)

  • removing old enum values is safe. But note that if we decode a record whose number is not in the current schema, it will be stored in Iceberg as the number in string form (e.g. "99")

Note: WarpStream's TableFlow also validates that sibling enum fields have identical sets if they share any value name. This prevents accidental inconsistencies between different fields using the same enum type.

A oneof field is defined with type: oneof. Each option must be explicitly set to optional and with a proto_field_number, but the oneof field itself must be defined as required and without any proto_field_number.

Field Type Remapping (Agent v796+)

Requires Agent v796 or higher.

Field type remapping lets you override specific Iceberg column types in the inferred table schema without declaring a full output schema. Use it when the input schema is correct for decoding, but the default Iceberg type is not the type you want to expose in the table. This is also useful when Tableflow infers a wider type than you need (for example, long when int is sufficient) and you want tighter typing for storage efficiency or downstream compatibility.

For example, if your JSON producer serializes numeric IDs as strings, you can keep the input schema as string (so decoding works) but store the column as long in Iceberg:

Here, user_id is decoded as a JSON string but the Iceberg table column is created as long. The input schema is not modified.

Requirements

  • schema_mode must be inline.

  • No explicit output schema can be defined. Remapping only applies when Tableflow infers the table schema from input_schema.

Syntax

Each entry specifies a path (dot-separated field path) and a type (Iceberg primitive type):

Nested fields use dot notation:

For lists and maps, use the internal child names from the schema conversion. For arrays use element and for maps use key and value:

Only scalar leaf fields can be remapped. Struct, list, map, and other container types cannot be remapped.

Supported Target Types

The target type must be a valid Iceberg primitive type. Simple types include: int, long, float, double, string, boolean, date, timestamp, timestamptz, uuid, and binary.

Parameterized types are also supported:

Decimal precision must be between 1 and 38, and scale must be non-negative and no greater than precision.

Common Use Cases

String to date:

Valid input: { "event_date": "2026-05-18" }. Date strings must be in YYYY-MM-DD format.

String to decimal:

Valid input: { "price": "123.45" }.

Important Considerations

  • Field type remapping does not change how source records are decoded. Decoding always uses input_schema.

  • Paths must match resolved schema field names exactly. Missing or duplicate paths are rejected.

  • Changing an existing table's columns type in a non-widening manner is an incompatible Iceberg schema change. Use recreation_key to trigger a table rebuild when intentionally changing column types.

  • Decimal scale handling truncates extra fractional digits rather than rounding. If exact decimal precision matters, prefer using a string field type.

  • Config validation confirms the target is a valid table schema type, but runtime success depends on whether actual values can be converted. For example, JSON "123" can become int, but "not-a-number" will fail.

Partitioning

Tableflow supports unpartitioned, timestamp partitioned tables using the record timestamp, and custom partitioning. Tables are unpartitioned by default and migrating between partitioning schemes is currently not supported.

The partitioning scheme is specified on a per-table basis. For convenience the partitioning_scheme option can be used to define unpartitioned tables and timestamp partitioned tables using the record timestamp. Supported values for partitioning_scheme include unpartitioned, hour, day, month, and year.

If a custom partitioning scheme is needed, then the custom_partitioning option can be used as follows:

Field
Description

source_field_path

The Input Column. The input field to partition on. Use the field path from the source schema, for example user_id, created_at, or customer.address.country. For WarpStream metadata fields, use paths like warpstream.timestamp or warpstream.partition

name

The Partition Alias. This is the name given to the partition itself. It does not have to match the source field name. It must start with a letter and contain only alphanumerics or underscores.

transform

An object defining the transform to be applied to the source column to produce a partition value.

The transform object requires a name and, depending on the type, additional parameters:

  • Time-based: year, month, day, hour.

  • bucket: Requires an n field (e.g., { name: "bucket", n: 16 }) to specify the number of buckets.

  • truncate: Requires a w field (e.g., { name: "truncate", w: 10 }) to specify the width of the truncation.

  • identity: Uses the source value as-is.

Note that to use custom partitioning the Agent version needs to be at least v748.

For example, to create hourly partitions on the Kafka timestamp and bucket the Kafka partitions into fours bins the configuration would look like this:

Sorting

Requires Agent v810 or higher. Every Agent in the cluster must be on a supported version before you configure sorting. An Agent that is too old cannot create a sorted table, so enabling sorting while any of the cluster's Agents are still on an older version will break ingestion for that table until all of them are upgraded.

Tableflow can physically sort the records within each table's data files by one or more fields. Sorting clusters similar values from a sort field together within a file, which improves compression and allows query engines to prune data files and Parquet row groups. This can significantly speed up queries that filter or range-scan on the sort fields.

Sorting is opt-in: a table is sorted only when you set the sorting: option and is left unsorted otherwise.

sorting: is an ordered list of fields. The order is significant: records are sorted by the first field, ties are broken by the second field, and so on.

Field
Description

field_path

(Required) The dot-separated path of the field to sort by, taken from the table schema (for example user_id or customer.address.country). WarpStream metadata fields such as warpstream.timestamp are also valid.

direction

The sort direction. One of asc (default) or desc.

null_order

Where null values are placed. One of nulls_last (default) or nulls_first.

transform

An optional transform applied to the field before sorting. Supports the same transforms as custom partitioning: identity (default), year, month, day, hour, bucket (requires n), and truncate (requires w).

Only primitive fields can be used as sort fields. Supported types are boolean, int, long, float, double, decimal, date, time, timestamp, timestamptz, string, binary, uuid, and fixed. struct, list, and map container fields cannot be sorted directly; reference a primitive leaf field with dot notation instead.

Decimal sort fields require Agent v822 or higher.

A table can have at most 5 sort fields. Every additional field also adds ordering work during ingestion and compaction, so sort only by the fields you actually query on. Contact support if you need a higher limit.

Sorting only affects newly written data; existing data is not rewritten to match a new sort order. This applies both when you first enable sorting and when you later change a table's sort fields.

Enabling sorting sends the minimum and maximum value of each sort field to the WarpStream control plane, to be used for Iceberg metadata generation. See Security and Privacy Considerations for the full list of metadata WarpStream stores.

Column Statistics

Requires Agent v810 or higher. As with sorting, every Agent in the cluster must be on a supported version before you configure statistics: on additional columns, or ingestion for that table will break until all of them are upgraded.

Tableflow can track per-file minimum and maximum value bounds for specific columns. These bounds let query engines prune data files that cannot match a query's filter, which accelerates selective queries on the tracked columns. Use the statistics: option to list the columns to track, by dot-separated field path:

Tableflow always tracks statistics for the built-in warpstream.timestamp and warpstream.partition fields, as well as for every field used in the table's sorting: configuration. Use statistics: to track additional columns.

The supported column types are the same as for sorting: primitive fields only (boolean, int, long, float, double, decimal, date, time, timestamp, timestamptz, string, binary, uuid, and fixed). Container fields (struct, list, and map) cannot be tracked directly.

Decimal column statistics require Agent v822 or higher.

A table can track statistics for at most 10 columns. Each tracked column also adds to the per-file metadata Tableflow maintains, so track only the columns you filter on. Contact support if you need a higher limit. Bounds for string and binary columns are stored as short truncated prefixes rather than full values, so pruning on those columns is conservative: a query may scan some files that do not actually match, but it never skips a file that could.

Enabling column statistics sends the minimum and maximum value of each tracked field to the WarpStream control plane, to be used for Iceberg metadata generation. See Security and Privacy Considerations for the full list of metadata WarpStream stores.

The per-file minimum and maximum bounds are written into the table's Iceberg metadata, so external query engines (not just WarpStream) can use them to skip files that cannot match a query's filter.

Transforms (Agent v730+)

Tableflow supports applying stateless transformations to ingested records. This can be helpful to massage the data into the desired shape before inserting it into the table without having to reprocess the data into a completely different topic first.

For example, imagine the topic contains a CDC change stream from debezium that looks like the following:

Without a transform, the table schema would have to be defined as follows:

This is unfortunate because users querying the data would always have to write their queries in the form: SELECT payload.after.field1 instead of simply SELECT field1.

Transforms solve this problem by rewriting the structure of the record before applying the table schema. Define input_schema to match the source records, apply transforms to reshape the data, and define schema as a YAML object to describe the final table shape.

For a JSON workload:

For an Avro workload, input_schema is especially important because Avro records cannot be deserialized without a schema. The same pattern applies: input_schema is a raw Avro schema matching the source records, and schema is a YAML object describing the post-transform table:

In summary, records before transforms must match input_schema and records after transforms must match schema. If schema is not defined, Tableflow infers the table schema from input_schema. If transforms only normalize values without changing field names, nesting, or types, you may still be able to omit schema and rely on inference.

Separately, keep in mind that transforms can be chained:

Tableflow transformations are executed by running arbitrary Bento Bloblang programs, but are limited to "pure" Bloblang functions that have no external side-effects.

Bloblang is a rich turing-complete programming language with many features, functions, methods, conditionals, and even error-handling. You can read more about Bloblang and its capabilities in the Bento Bloblang documentation, but but the basics are quite straightforward and can be grasped with a few examples.

The key thing to understand about Bloblang transformations is that they're mapping functions that mutate the input record into the desired shape. Within the context of a Tableflow Bloblang mapping, the this keyword refers to the input record and the root keyword refers to the output record. See the examples below to learn how to perform the most common transformations.

The Bento website has a powerful and interactive Bloblang playground that can be used to experiment with Bloblang mapping programs.

In schema_mode: schema_registry, bento transforms are supported. To branch on writer schema metadata during transform execution, use warpstream.sr.* fields (requires agent v817+; see Schema Registry mode).

Rename a field

Delete a field

Add a field

Drop / filter out an entire record

Type Conversions

Data Retention and TTL

By default, data is retained in the table indefinitely. Optionally, a retention period can specified using the retention_ttl field. Retention must be expressed in units of hour (h).

Starting ingestion at the latest offsets

By default, data is ingested from the start of the topic. You can optionally override this setting to instead start at the latest offsets (skipping any data previously stored in your input topic) via the start_ingestion_at field:

The supported values are:

  • latest

  • earliest

Dead Letter Queue (DLQ) Mode

Requires Agent v737 or higher.

By default, Tableflow stops ingestion when it sees records that are incompatible with the provided schema to avoid head of line blocking. This behavior can be overridden using the dlq_mode field. Supported values include:

  • stop, which blocks ingestion upon encountering an invalid record (this is the default)

  • skip, which skips invalid records during ingestion

  • keep, (Requires Agent v792 or higher) which parks invalid records to an internal topic that can then be re-ingested. You can fix those records using transforms if they were not complying with the table schema for instance.

DLQ Mode Keep settings

Requires Agent v792 or higher.

Setting the DLQ mode to keep ensures continuous ingestion from your main topic even when some records cannot be processed. Those records will be sent to an internal topic so you can then manually trigger ingestion for from the UI.

You can control this behavior via the dlq_keep_settings settings in your table configuration (required when setting the mode to keep).

dlq_keep_settings supports:

  • retention: how long records are stored in the DLQ topic (for instance 10h or 5d)

  • dlq_replay_mode: the mode used when replaying traffic from the DLQ (must be skip or stop)

    • skip: ignores records that fail ingestion again during the replay

    • stop: halts the entire replay process if any DLQ record fails to ingest

  • circuit_breaker: controls what happens when too many records are being DLQ'ed

    • on_open: defines the fallback behavior if the circuit breaker trips. Supported values are skip or stop. When triggered, this value overrides the keep behavior with this new setting.

  • last_n_records: the only circuit breaker policy allowed for now

    • count: the number of consecutive invalid records the system must encounter before tripping the circuit breaker and triggering the on_open logic.

Here's an example of a complete DLQ keep configuration

Re-processing DLQ records

As mentioned in the previous section, if you have some records that were written to the DLQ, you can go into the UI and manually replay them. However, if your records were put into the DLQ it often means that there was an issue ingesting them, which could be because of an invalid input or table schema.

If this happens you can write a transform and scope it to your DLQ records, for instance if your session_id field was wrong you could do:

Records coming from the DLQ are automatically injected with metadata fields that you can access within your transforms.

Field
Description

warpstream.dlq_timestamp

The exact time the record was written to the DLQ.

warpstream.dlq.failure_reason

A brief error message explaining why the record failed ingestion.

warpstream.dlq.original_offset

The offset of the record in the source topic.

warpstream.dlq.original_partition

The partition of the record in the source topic.

warpstream.dlq.source_topic

The name of the source topic.

Compression codecs

Requires Agent v748 or higher.

Tableflow supports a few compression codecs for the stored data files. The default one is snappy .

This codec can be overridden using the compression field. Supported values include:

  • snappy

  • gzip

  • lz4

  • zstd

  • brotli

  • none To disable compression

Skipping raw record values

Requires Agent v749 or higher.

If you don't need to access the raw record values (the ones coming from the kafka topics) you can set the skip_raw_record_values to true in your config. This will result in smaller data files.

Handling topic re-creation

To define the table ingestion behavior when the source topic is recreated, use the topic_recreation_policy setting. Currently, the only supported policy is recreate_table. This ensures data integrity by creating a new table (with a different identifier) whenever the system detects that the source topic has been re-created.

Pausing a Table

To temporarily stop ingestion for a specific table without removing it from your configuration, set paused to true:

To resume, set paused: false (or remove the field) and deploy the updated configuration. Ingestion will pick up from where it left off.

Schema Registry mode

Tableflow supports Protobuf and JSON schemas from an external Confluent-compatible Schema Registry from agent version v813+ and Avro schemas from agent version v820+.

In schema_mode: schema_registry, records are expected to be serialized with Confluent's wire format. You may omit wire_format (defaults to Confluent) or set wire_format: confluent explicitly.

A config using the schema_registry mode requires some specific sections:

  • a schema_registries section (see Configure Source Schema Registry).

  • inside each tables section, a schema_registry subsection that contains the name of the source schema registry and the subject in which schema versions are registered.

An example config for that mode is the following:

In schema_mode: schema_registry, bento transforms are supported. During transform execution, writer schema metadata is available under warpstream.sr (these fields are not stored in the Iceberg table):

Requires Agent v817 or higher. To use warpstream.sr.writer_schema_id and warpstream.sr.writer_schema_version in your Bento transform logic every Agent in the Tableflow cluster must be on at least v817.

Field
Description

warpstream.sr.writer_schema_id

Confluent schema ID used to produce the record

warpstream.sr.writer_schema_version

Schema Registry subject version of the writer schema

Example table config with a transform that tags records based on writer schema version:

Important requirements and recommendations for use in Schema Registry mode

There are some requirements for the schema registry integration to work properly:

In the Schema Registry mode, the ingested records are always deserialized using the schema they were serialized with and written to Parquet files using the latest schema that was synced (which is the last registered schema in the subject). This means that we need to convert the record in the writer schema version to the latest schema version. This step requires for every new version to be backwards-compatible with any existing version used by a producer. Therefore, we recommend setting your schema registry to have a compatibility level set to BACKWARD_TRANSITIVE to guarantee that a record produced with any schema version will be convertible to the latest version. That being said, you may relax this configuration to only BACKWARD compatibility level if you always produce with the two most recent schema versions for example.

If the latest schema version synced by TableFlow is not backwards-compatible with the schema used to produce a record, we may DLQ that record or decode it incorrectly.

Any schema evolution must be Iceberg backwards-compatible. In the Schema Registry mode, contrary to the inline mode, that check is made asynchronously after fetching the latest version from the schema registry. The same is true for checking that some sections of the config (like partitioning, sorting, ...) are compatible with that schema. It's preferable that you wait until your new schema has been synced (and thus that we have made sure it is valid) before you start producing records with it.

Our recommended flow when deploying a schema change is the following:

  • Register your new version in the Schema Registry

  • This new schema version will be processed by TableFlow: checking that it is Iceberg backward-compatible and that it's compatible with the rest of the config (partitioning, sorting, ...)

  • Once that version has been validated and synced, you can start upgrading your producers to use that new version.

To that end, we also highly recommend that you disable auto schema registration on your serializer (see docs).

If you do not follow our recommended flow, the following may happen:

  • If you start producing records with a schema version newer than the version that was synced, ingestion of those records will be transiently paused until we have synced a newer version.

  • If you start producing records with a schema version newer than the version that was synced AND it turns out the version cannot be synced because it is invalid (for example not backwards-compatible), ingestion of these records will be transiently paused until you have submitted a new version that is valid.

Table Schema

Tableflow creates an Iceberg table with a struct schema, containing all the fields from the configured schema as well as the following default fields:

Name
Field ID
Type

warpstream

10000000

struct

warpstream.partition

10000001

int

warpstream.offset

10000002

long

warpstream.key

10000003

binary

warpstream.value

10000004

binary

warpstream.timestamp

10000005

timestamp

The warpstream.value field can be ommited with the skip_raw_record_values option.

Schema Migrations

Tableflow follows the Apache Iceberg schema evolution rules. Schema migrations are supported for adding columns, changing fields from required to optional, and widening integer and floating point types. To perform any of these operations, update the schema in the Configuration editor and deploy it. In the next few syncs of the table metadata into your bucket, the schema change will be reflected.

Ensure that you execute a schema change before attempting to send data with the new schema. If you fail to do this, you will potentially lose the data written with the newer schema.

Breaking Changes

Any schema change that is not one of the compatible operations listed above is considered a breaking change and requires Table Recreation. Specifically, the following changes are breaking:

  • Make an optional field required — Existing data may contain nulls for that field, so the constraint cannot be applied retroactively.

  • Change a type in a non-widening way — For example, long → int, double → float, or string → int. This includes any data type change that is not a supported numeric widening.

  • Drop a column — Removing a column from the schema is not supported as an in-place migration.

  • Reorder columns — Changing the order of columns is not supported as an in-place migration.

  • Rename a column — Renaming a column is not currently supported as an in-place migration.

To apply breaking changes, use the recreation_key mechanism described in Table Recreation below.

Table Recreation

Certain operations require an existing table to be completely deleted and recreated. While this is most commonly necessary to apply breaking schema changes (such as converting an optional field to required, performing non-widening type conversions, or dropping columns), a full rebuild may also be required for other operational reasons. The recreation_key parameter is designed to automate this workflow.

To utilize this feature, assign an initial string value to the recreation_key in your table configuration. Any change to this value acts as a direct trigger for table recreation. When you need to rebuild a table, simply update the recreation_key to a new string and deploy the new configuration. Tableflow will detect the change, automatically drop the existing table, and provision a new one with the updated config.

Note that adding a recreation key to a table that does not already specify this option will trigger a recreation.

The following example demonstrates how to use the recreation_key to apply backward-incompatible schema changes:

To make the service field required and add a required severity field, bump the value (e.g. "v1" to "v2") alongside your schema change:

After recreation, Tableflow will re-ingest data from the earliest available offset in the source topic (subject to the topic's retention policy) into the new table.

Changing recreation_key will hard-delete the existing table's metadata. Data already written to the object store is not removed. The new table will only contain data that is still available in the source topic based on its retention settings.

Table Deletion

Tableflow does not delete any tables from the object storage bucket when they are removed from the Configuration in order to prevent accidental data deletion. To delete a table, first delete it from the Configuration and then use your cloud provider's UI or CLI to delete the directory containing your table from within the warpstream/_tableflow directory. To programmatically delete and recreate a table (e.g. for incompatible schema changes), see Table Recreation.

Object Storage Path Layout

Tableflow writes all table data and metadata into your object storage bucket under a predictable directory structure. Understanding this layout is useful for debugging, removing an entire table directory, or integrating with query engines that need direct file paths.

Given a bucket URL of s3://my-bucket (or s3://my-bucket?prefix=my-prefix), the structure is:

Path
Contents

warpstream/_tableflow/

Root directory for all Tableflow tables.

<table_name>-<table_uuid>/

One directory per table.

data/

Parquet data files.

metadata/

Iceberg metadata: JSON metadata files (v1.metadata.json, v2.metadata.json, ...), manifest files (mani-*.avro), manifest lists (snap-*.avro), and a version-hint.text that points to the latest metadata version.

When a table is deleted from the configuration or via the API, Tableflow only removes the control plane metadata. The data and metadata files in object storage are not deleted. You must clean them up manually using your cloud provider's UI or CLI.

Terraform / Infrastructure as Code / APIs

Tableflow Agents are deployed using the standard WarpStream Agent chart, and there is full support for Tableflow clusters in the WarpStream Terraform provider.

Click here for a complete Terraform example of creating a Tableflow cluster and configuring it to ingest a single topic into an Iceberg table.

Tableflow configuration can also be modified with the pipelines API.

Observability

Tableflow clusters emit metrics for ingestion lag, offset lag, and end-to-end query lag, as well as diagnostics and events for tracking pipeline health. Ingestion lag is also available visually in the WarpStream Console.

For a complete guide on which metrics to monitor, how to interpret diagnostics, how to use events for troubleshooting, and recommended alerts, see Monitoring Tableflow.

Agent Version Requirements

Tableflow requires agents running on version v797 or higher.

Feature
Minimum agent version

schema_mode: schema_registry with source_format: protobuf or json

v813

schema_mode: schema_registry with source_format: avro

v820

warpstream.sr.* writer schema metadata in transforms

v817

Always check the Change Log for the latest feature additions and bug fixes. When in doubt, use the latest stable Agent version.

Tableflow UI

The Tableflow UI available in the WarpStream console allows editing the Configuration.

PreviousAuto MigrationNextMonitoring Tableflow

Last updated

Was this helpful?