BQ Sink Connector Partition Decorator Support for Write API

Failure Scenarios

Use of partition decorator syntax fails (on the connector side) in the following scenarios:

• Table is not partitioned by DAY: The use of decorator syntax requires the target table to be partitioned by DAY. If the table is partitioned by another type (e.g., MONTH, HOUR, etc.), the connector fails on first write with the following: “Cannot use decorator syntax to write to table as it is partitioned by MONTH and not by day”
• Record is missing timestamp: When bigQueryMessageTimePartitioning=true, the connector expects each record to have a timestamp set. If a record’s timestamp is NO_TIMESTAMP_TYPE, the connector will not be able to determine the partition and fails with the following: “Message has no timestamp type, cannot use message timestamp to partition.”

To enable partition decorator syntax with Storage Write API, we need to ensure we are maintaining the same failure scenarios.

Re-introducing Partition Decorator Syntax

As of version 2.7.0, the connector disabled support for partition decorator syntax altogether. This was based on the assumption that partition decorator syntax is not supported by Storage Write API, combined with the intent to eventually deprecate InsertAll API support (see the following). To restore support for partition decorators, the deprecated warning will need to be removed.

Enabling Partition Decorator Syntax in Storage Write Api mode

Given that with the enablement of partition decorator for Storage Write Api most of the table resolution logic will be shared with the insertAll Api, we can consolidate the two existing methods (getStorageWriteApiRecordTable and getRecordTable) into a unified getRecordTable method. This will avoid unnecessary code duplication and ensure consistent handling of partition decorators across both write paths.

The connector already enforces a constraint during startup that disallows enabling both useStorageWriteApi and upsertDelete simultaneously. As seen in StorageWriteApiValidator:

if (config.getBoolean(UPSERT_ENABLED_CONFIG)) {
      return Optional.of(upsertNotSupportedError);
} else if (config.getBoolean(DELETE_ENABLED_CONFIG)) {
        return Optional.of(deleteNotSupportedError);
}

This makes it safe for getRecordTable to contain logic specific to resolving intermediate tables for upsert/delete mode even when the connector is running with Storage Write Api enabled. However to safeguard it further, we can also enable the upsertDelete flag in BigQuerySinkTask only if the Storage Write API is disabled. Eg:

BigQuerySinkTask.java

upsertDelete = !useStorageWriteApi && config.getBoolean(BigQuerySinkConfig.UPSERT_ENABLED_CONFIG) || config.getBoolean(BigQuerySinkConfig.DELETE_ENABLED_CONFIG) 

Optimization of Table Resolution

The current implementation of getRecordTable performs unnecessary work, leading to inefficiency, especially for high-throughput topics. These issues will become more critical as getRecordTable will now be used by both StorageWriteApi path and insertAll, so will need to be addressed.

1. Redundant Base Table ID Computation: The current implementation recomputes the base table ID for every record. The base table ID is then used to derive the final table name (if upsert/delete or partition decorator is enabled) or is returned directly for standard flow. This re-computation, however, is redundant, since all records from a specific topic will always resolve to the same base table ID.
2. Unnecessary Partition Validation: When the usePartitionDecorator setting is enabled, the getRecordTable further validates that the table is partitioned by DAY (repeatedly for every record). This re-validation is also redundant because table partitioning in BigQuery is immutable once the table is created. It is, therefore, enough to verify table partitioning once per topic instead of for every individual record.

To address this redundant work, we can introduce a caching mechanism. The proposed approach is to:

• On the first record for a topic, retrieve the StandardTableDefinition from BigQuery, validate that the table is partitioned by day, and then cache the base table ID for that topic.
• For all subsequent records from the same topic, the function will retrieve the table ID directly from the cache. This bypasses the need to retrieve and revalidate the table definition, allowing the function to immediately apply the partition decorator syntax.

Handling Decorated vs Undecorated Table Names in the Storage Write API Flow

Currently, in the Storage Write API flow, the connector uses the undecorated (base) table name when constructing the StorageWriteApiWriter:

BigQuerySinkTask.java

if (useStorageApi) {
    tableWriterBuilder = new StorageWriteApiWriter.Builder(
    storageApiWriter,
    TableNameUtils.tableName(table.getBaseTableId()),
    recordConverter,
    batchHandler
 );
}

This behaviour must be revised to support partition decorator syntax. However, we cannot simply replace the table.getBaseTableId() with table.getFullTableId() (the partition decorated name) as to maintain the existing behaviour, the stream writer also needs to be able to perform table operations (auto-creating or updating tables) via the StorageWriteApiRetryHandler if the connector is configured to do so:

StorageWriteApiRetryHandler.java

public void attemptTableOperation(BiConsumer<TableId, List<SinkRecord>> tableOperation) {
    try {
      tableOperation.accept(tableId(), records);
      // Table takes time to be available for after creation
      setAdditionalRetriesAndWait();
    }

These table operations must use the base table name to avoid BigQuery API rejecting the requests with the following errors: “Invalid table ID “table$20250710”. Table Ids must be alphanumeric (plus underscores) and must be at most 1024 characters long. Also, table decorators cannot be used”.

To ensure correctness and avoid breaking table manipulation functionality, we need to ensure we are using correct table name representations within the Storage Write Api flow:

• The decorated table name must be used when creating and caching JsonStreamWriters in map (decoratedTableName to JsonStreamWriter)
• The base table name must be used for table-related operations (creations and updates).

Accounting for Storage Write Api Partition Decorator Limitations

Currently, BigQuery Storage Write API only supports partition decorator syntax with DEFAULT stream. It does not support decorators when using Application Streams, e.g, PENDING which is used for batch loading (when enableBatchMode=true).

To make it explicit that use of partition decorator is not available when using batch mode, we will need to add to the validation logic in the StorageWriteApiValidator. On connector startup, we will return an error if the user explicitly requested partition decorator syntax and batch mode, or silently disable partition decorator otherwise.