{"record":{"id":"6ec4a90e1bf4e263","repo":"apache/seatunnel","slug":"streaming-mode-requires-a-primary-key-in-the-schem","errorCode":null,"errorMessage":"Streaming mode requires a Primary Key in the schema of table: ","messagePattern":"Streaming mode requires a Primary Key in the schema of table: ","errorType":"exception","errorClass":"CatalogException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-bigquery/src/main/java/org/apache/seatunnel/connectors/bigquery/catalog/BigQueryCatalog.java","lineNumber":259,"sourceCode":"            if (ignoreIfExists) {\n                return;\n            }\n            throw new TableAlreadyExistException(catalogName, tablePath);\n        }\n\n        List<Field> fields = new ArrayList<>();\n        for (Column column : table.getTableSchema().getColumns()) {\n            fields.add(convertColumn(column));\n        }\n\n        // Add stream change capture tracking fields if running in streaming mode\n        boolean isBatch =\n                BigQuerySinkBatchWriter.BATCH.equals(config.get(BigQuerySinkOptions.WRITE_MODE));\n        if (!isBatch) {\n            org.apache.seatunnel.api.table.catalog.PrimaryKey seaTunnelPrimaryKey =\n                    table.getTableSchema().getPrimaryKey();\n            if (seaTunnelPrimaryKey == null || seaTunnelPrimaryKey.getColumnNames().isEmpty()) {\n                throw new CatalogException(\n                        \"Streaming mode requires a Primary Key in the schema of table: \"\n                                + tablePath.getFullName());\n            }\n        }\n\n        Schema bqSchema = Schema.of(fields);\n        TableId tableId = TableId.of(getDatasetName(tablePath), tablePath.getTableName());\n\n        TableDefinition tableDefinition;\n        org.apache.seatunnel.api.table.catalog.PrimaryKey seaTunnelPrimaryKey =\n                table.getTableSchema().getPrimaryKey();\n\n        if (seaTunnelPrimaryKey != null && !seaTunnelPrimaryKey.getColumnNames().isEmpty()) {\n            com.google.cloud.bigquery.PrimaryKey bqPrimaryKey =\n                    com.google.cloud.bigquery.PrimaryKey.newBuilder()\n                            .setColumns(seaTunnelPrimaryKey.getColumnNames())\n                            .build();\n","sourceCodeStart":241,"sourceCodeEnd":277,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-bigquery/src/main/java/org/apache/seatunnel/connectors/bigquery/catalog/BigQueryCatalog.java#L241-L277","documentation":"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).","triggerScenarios":"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\").","commonSituations":"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.","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)."],"exampleFix":"// before\nBigQuery {\n  write_mode = \"streaming\"   # table schema has no primary key -> error\n}\n\n// after: either switch to batch\nBigQuery {\n  write_mode = \"batch\"\n}\n// or ensure the source schema defines a primary key\n// source: table with primary_keys = [\"id\"]","handlingStrategy":"validation","validationCode":"// Java, before invoking createTable in streaming mode\nboolean isBatch = \"batch\".equalsIgnoreCase(config.get(\"write_mode\"));\nif (!isBatch) {\n    PrimaryKey pk = table.getTableSchema().getPrimaryKey();\n    if (pk == null || pk.getColumnNames() == null || pk.getColumnNames().isEmpty()) {\n        throw new IllegalArgumentException(\n            \"write_mode=streaming requires a primary key; table \"\n            + tablePath.getFullName() + \" has none\");\n    }\n}","typeGuard":"static boolean hasPrimaryKey(CatalogTable table) {\n    PrimaryKey pk = table == null ? null : table.getTableSchema().getPrimaryKey();\n    return pk != null && pk.getColumnNames() != null && !pk.getColumnNames().isEmpty();\n}","tryCatchPattern":"try {\n    catalog.createTable(tablePath, table, ignoreIfExists);\n} catch (CatalogException e) {\n    if (e.getMessage().startsWith(\"Streaming mode requires a Primary Key\")) {\n        // fallback: switch sink to batch write mode or add a key upstream\n        log.warn(\"Keyless schema rejected for streaming write; use batch or define a PK\");\n        return;\n    }\n    throw e;\n}","preventionTips":["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"],"tags":["bigquery","schema","primary-key","validation","streaming"],"backgroundTag":"schema-validation-failed","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}