apache/seatunnel · error · CatalogException

Streaming mode requires a Primary Key in the schema of table

Error message

Streaming mode requires a Primary Key in the schema of table:

What it means

BigQueryCatalog.createTable throws this when the sink write mode is not 'batch' (i.e. streaming) but the SeaTunnel TableSchema has no PrimaryKey defined. Streaming writes to BigQuery in this connector upsert on a primary key, so a keyless schema is rejected before any API call is made. This is a local validation error — the message ends with the table path's full name (project.dataset.table).

Source

Thrown at seatunnel-connectors-v2/connector-bigquery/src/main/java/org/apache/seatunnel/connectors/bigquery/catalog/BigQueryCatalog.java:259

            if (ignoreIfExists) {
                return;
            }
            throw new TableAlreadyExistException(catalogName, tablePath);
        }

        List<Field> fields = new ArrayList<>();
        for (Column column : table.getTableSchema().getColumns()) {
            fields.add(convertColumn(column));
        }

        // Add stream change capture tracking fields if running in streaming mode
        boolean isBatch =
                BigQuerySinkBatchWriter.BATCH.equals(config.get(BigQuerySinkOptions.WRITE_MODE));
        if (!isBatch) {
            org.apache.seatunnel.api.table.catalog.PrimaryKey seaTunnelPrimaryKey =
                    table.getTableSchema().getPrimaryKey();
            if (seaTunnelPrimaryKey == null || seaTunnelPrimaryKey.getColumnNames().isEmpty()) {
                throw new CatalogException(
                        "Streaming mode requires a Primary Key in the schema of table: "
                                + tablePath.getFullName());
            }
        }

        Schema bqSchema = Schema.of(fields);
        TableId tableId = TableId.of(getDatasetName(tablePath), tablePath.getTableName());

        TableDefinition tableDefinition;
        org.apache.seatunnel.api.table.catalog.PrimaryKey seaTunnelPrimaryKey =
                table.getTableSchema().getPrimaryKey();

        if (seaTunnelPrimaryKey != null && !seaTunnelPrimaryKey.getColumnNames().isEmpty()) {
            com.google.cloud.bigquery.PrimaryKey bqPrimaryKey =
                    com.google.cloud.bigquery.PrimaryKey.newBuilder()
                            .setColumns(seaTunnelPrimaryKey.getColumnNames())
                            .build();

View on GitHub (pinned to cf67b549a7)

Solutions

Declare a primary key in the schema: if the connector supports defining one in the sink/table options, set it (e.g. a `primary_keys` option or via the source's key definition).
Switch to batch mode: set `WRITE_MODE = "batch"` in the BigQuery sink options if you don't need upsert semantics.
If the upstream table truly has a key, verify the source connector is propagating it (e.g. JDBC catalog metadata with PKs present) rather than a transform stripping it.
If no natural key exists, add a synthetic unique key column upstream (e.g. row id/uuid) before the sink.
Pick a different sink supporting keyless streaming (e.g. append-only BigQuery via batch loads at intervals).

Example fix

// before
BigQuery {
  write_mode = "streaming"   # table schema has no primary key -> error
}

// after: either switch to batch
BigQuery {
  write_mode = "batch"
}
// or ensure the source schema defines a primary key
// source: table with primary_keys = ["id"]

Defensive patterns

Strategy: validation

Validate before calling

// Java, before invoking createTable in streaming mode
boolean isBatch = "batch".equalsIgnoreCase(config.get("write_mode"));
if (!isBatch) {
    PrimaryKey pk = table.getTableSchema().getPrimaryKey();
    if (pk == null || pk.getColumnNames() == null || pk.getColumnNames().isEmpty()) {
        throw new IllegalArgumentException(
            "write_mode=streaming requires a primary key; table "
            + tablePath.getFullName() + " has none");
    }
}

Type guard

static boolean hasPrimaryKey(CatalogTable table) {
    PrimaryKey pk = table == null ? null : table.getTableSchema().getPrimaryKey();
    return pk != null && pk.getColumnNames() != null && !pk.getColumnNames().isEmpty();
}

Try / catch

try {
    catalog.createTable(tablePath, table, ignoreIfExists);
} catch (CatalogException e) {
    if (e.getMessage().startsWith("Streaming mode requires a Primary Key")) {
        // fallback: switch sink to batch write mode or add a key upstream
        log.warn("Keyless schema rejected for streaming write; use batch or define a PK");
        return;
    }
    throw e;
}

Prevention

Choose write_mode=batch when the source has no primary key (Kafka, files, keyless JDBC)
Ensure source connectors propagate PK metadata and no transform strips it
Add a synthetic unique id column upstream if no natural key exists
Validate the CatalogTable schema before schema_save_mode=CREATE pipelines run
Document the PK requirement for streaming jobs in your team's job templates

When it happens

Trigger: Calling createTable with a CatalogTable whose `tableSchema.getPrimaryKey()` returns null or has an empty getColumnNames(), while `BigQuerySinkOptions.WRITE_MODE` is set to something other than "batch" (e.g. "streaming").

Common situations: User configures write_mode = "streaming" but the source data has no primary key declared (e.g. reading from Kafka, files, or a JDBC table without PK); schema transformed by an earlier transform dropped the key; someone copied a batch config template and only changed the mode; using auto-create table (schema_save_mode=CREATE) with keyless sources.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

AI-assisted analysis of apache/seatunnel@cf67b549a7 (2026-09-10). Data as JSON: /api/errors/6ec4a90e1bf4e263. Report an issue: GitHub.