Bigtable SQL Connector#

The bigtable connector reads and writes a table in Cloud Bigtable through the module flink-connector-gcp-bigtable. It is a mapping onto the DataStream sink and scan source documented in Bigtable — that page carries the design, the delivery guarantees and the error handling; this one carries the DDL surface. Per-feature status is in the module README.

sink.parallelism and scan.parallelism come from Flink’s own FactoryUtil rather than from this connector. Bounded scans and sinks have no format option: a Bigtable row is a schema this DDL describes, cell by cell, and the cell encoding is the HBase ecosystem’s rather than a choice. The selected-cell Change Streams mode is the exception because one cell holds a serialized logical row and value.format decodes it.

In the ordinary family/qualifier schema, a column family is a column name, so it has to be a legal SQL identifier — a reserved word such as identity needs backticks, or a different name.

CREATE TABLE profiles (
  rowkey STRING,
  profile ROW<name STRING, email STRING>,
  usage ROW<requests BIGINT, last_seen TIMESTAMP_LTZ(3)>,
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles',
  'sink.insert-only-input-mode' = 'insert-only'
);

INSERT INTO profiles
SELECT user_id, ROW(name, email), ROW(requests, last_seen) FROM staged_profiles;

-- A bounded scan of the same table; only the families the query reads leave the server.
SELECT rowkey, profile FROM profiles;

The async SQL functions return conditional outcomes and changed cells to Flink 2.x streaming queries, with SQL-only registration and named request settings.

Getting the connector onto the classpath#

Use flink-sql-connector-gcp-bigtable, an uber-jar built for exactly this: put it in Flink’s lib/ directory, or add it with ADD JAR in the SQL client. It bundles flink-connector-gcp-bigtable together with its whole runtime tree — the Bigtable client, gRPC, protobuf, Guava, the Google auth and HTTP clients — which is 67 artifacts, not a dependency list anyone wants to assemble by hand.

The plain flink-connector-gcp-bigtable jar works too, where the deployment already resolves transitive dependencies. That is the right choice for a DataStream job built with Maven or Gradle. For SQL it usually is not.

The uber-jar does not bundle Flink format implementations. A selected-cell Change Streams job must also put the chosen format jar, such as flink-json, on the SQL client and cluster classpaths.

Relocation and shared APIs#

Bundled dependencies and internal helpers move under io.github.flink.gcp.connector.bigtable.shaded., so the versions of gRPC, protobuf and Guava this connector needs cannot collide with the ones a job, another connector, or Flink itself brings. Six third-party packages are deliberately not relocated, and none of them can collide in a way that matters: org.conscrypt, which gRPC picks up reflectively as an optional TLS provider and does without when it is unusable; and the annotation-only javax.annotation (jsr305’s classes only — javax.annotation-api, the other artifact publishing into that package, is not bundled, #352), org.jspecify, org.codehaus.mojo.animal_sniffer, android.annotation and org.checkerframework, where a duplicate class is inert because nothing ever invokes it.

The shared lineage values PhysicalResourceFacet and ResourceIdentifier also retain their original package names so one listener can consume them across SQL connector jars. See Lineage for the class loader configuration and connector adoption status.

io.grpc:grpc-netty-shaded is relocated, including the rename of its META-INF/native/ libraries that relocating an already-relocated gRPC requires. The full reasoning — why exempting it instead would trade a real collision for a hypothetical one — is on the Pub/Sub SQL page, and this jar inherits it. One consequence worth repeating: relocation rewrites netty’s system-property names along with its packages, so a -D spelled with the upstream io.grpc.netty.shaded.io.netty. prefix has no effect inside this jar.

Sharing a lib/ with the sibling GCP SQL uber-jars works, and it is measured for this jar rather than inherited: of the 560 file entries it shares with the Pub/Sub jar and the 1,034 it shares with the BigQuery jar, all but four are byte-identical in each pair, and the four are per-jar metadata Flink reads through ServiceLoader or enumeration — the manifest, the NOTICE, and two service files (measured 2026-08-10, one build of each jar; the BigQuery SQL page carries the same measurement for the pair it names). Merging jars into one fat jar is the case that does not work: without maven-shade’s ServicesResourceTransformer, one jar’s META-INF/services/org.apache.flink.table.factories.Factory entry silently shadows the other’s, as does one jar’s NOTICE. Put the jars in lib/, or add each with its own ADD JAR.

That warning has a second reader here that the sibling pages do not have: google/flink-connector-gcp publishes a Bigtable table connector that registers the same bigtable identifier (#472). With both jars in lib/, FactoryUtil discovery fails loudly, naming the ambiguous identifier — the acceptable outcome. Merged into one fat jar, the failure is silent: whichever factory registration survives owns bigtable, and the DDL then means whatever that connector says it means. The two option vocabularies overlap in only project, instance and table, so the first symptom is the other connector rejecting options it never declared.

Licensing#

META-INF/NOTICE inside the jar lists every bundled artifact grouped by licence, and META-INF/licenses/ carries the full text of each non-Apache-2.0 one — protobuf, gax and api-common, the Google auth library, the ThreeTen backport, RE2/J, animal-sniffer and the Checker Framework qualifiers.

The prose of the NOTICE is human-written, in the module’s NOTICE.template; the artifact lists are generated into it from what Maven actually resolves, so a wrong licence grouping or a stale version cannot be written at all. Each licence text has a pinned source — the artifact’s own jar where one ships a text, otherwise a curated URL matched to the bundled version — recorded with its sha256, so a text that changes upstream fails the build instead of being shipped unreviewed. just update-notice <module> regenerates both after a dependency change; just check-notice <module> verifies, offline, that what is checked in still matches the bundle and the pins.

Lineage#

Ordinary scans, Change Streams scans in both changelog modes, and every sink write mode expose the configured physical table to the supported Flink 2.x planner. Flink uses the SQL catalog identifier as the dataset name and retains namespace bigtable://{project}/{instance} with a gcp facet containing the physical project, instance and table (kind bigtable-table). The physical table name can differ from the SQL catalog name.

The upsert, keep-latest and aggregate modes report their MutateRows destination; insert-if-absent and conditional report their Conditional destination; append and increment report their ReadModifyWrite destination. Metadata inspection does not start aggregate family validation, evaluate a conditional predicate or infer whether an RPC changed a row. Projection, filtering, row ranges, app profiles and column families retain the same table identity. Change Streams names only the configured data table, and its lineage boundedness follows the configured end timestamp.

Lookup joins and the result-emitting Async SQL functions do not contribute Bigtable Source/Sink lineage through FLIP-314. Flink 1.20 provides connector metadata for direct inspection, without automatic listener extraction. See Lineage for listener configuration, physical-resource facet access and class loader requirements.

The schema#

The schema below applies to the ordinary family/qualifier modes. The write-only conditional command schema instead binds named input columns to a DDL-defined request.

The DDL model is Flink’s HBase connector’s, so a table definition moves between the two with its schema intact and a table written by either is readable by the other:

• Exactly one column is not a ROW, and that column is the row key. It may be declared anywhere among the columns. Its type decides how the key bytes are formed, so a BIGINT key is eight big-endian bytes rather than its decimal text.
• Every ROW column is one column family, and the column’s name is the family’s. Its nested fields are the qualifiers, one cell each. A nested ROW is rejected: a cell holds bytes, not a structure.
• A family name containing : is rejected. Bigtable’s family filter is a regular expression that refuses a colon even escaped, so such a family could be written but never selectively read.

A PRIMARY KEY is optional, exactly as it is in the HBase connector — a Bigtable write is keyed on the row key whether or not the DDL says so. If one is declared it must be the row-key column and nothing else.

The updating-query discussion below applies to ordinary upsert and keep-latest writes. Aggregate, insert-if-absent, conditional, append and increment modes accept only INSERT input.

Declaring it makes an updating query cheaper. A delete has to reach the sink carrying the row key, and which of two ways that is arranged depends on the primary key. With one declared, the sink tells the planner a delete may carry the upsert key alone — that key is the row key, so nothing else is needed. With none declared, the planner keys its upserts on whatever the query happens to be unique by, which need not be the row-key column at all, so the sink asks for whole rows and the planner completes each one before the delete reaches it. That completion is a ChangelogNormalize, which keeps state proportional to the keyspace. For this delete-completion decision, a query that carries no deletes needs no ChangelogNormalize either way. Flink 2.3’s separate conflict-strategy state for insert-only input is covered under Flink 2.3 may demand ON CONFLICT.

A completed delete is completed from what the job has seen. ChangelogNormalize holds the last row per key in Flink state, so a -D for a key this job never inserted has nothing to complete from and is dropped rather than applied — a row written by an earlier job, or before the state was cleared, is not deleted by it. This is the planner’s behaviour rather than the connector’s, and it already applied to every table that declares a primary key; from #470 it applies to those that do not. If deleting rows a different job wrote is the goal, emit the row key in the delete — a retract source carrying whole rows reaches the sink without a normalize.

Upgrading past #470 changes the plan of an updating query over a table with no PRIMARY KEY, because ChangelogNormalize is a new stateful operator. Flink assigns a SQL pipeline’s operator UIDs explicitly only when it came from a persisted COMPILED PLAN — table.exec.uid.generation defaults to PLAN_ONLY — and otherwise lets the lower layers generate them “taking the complete topology into account”. A savepoint taken before the upgrade therefore does not map onto the new topology, so restoring one is a plan migration rather than a version bump. A table that declares its primary key is unaffected: such a job already had the operator.

A persisted COMPILED PLAN pins the old shape, which is what it is for. The plan is the execution graph — stream-exec-changelog-normalize is one of its own node types — and its sink node stores the changelog mode it was compiled with rather than asking the connector again. A plan written before this change therefore keeps the pre-upgrade topology, and the job running it keeps the pre-upgrade behaviour, until the plan is recompiled.

Schema and option rejections happen when a statement is planned, not at CREATE TABLE. Flink does not consult a connector while registering a table, so a CREATE TABLE naming a column this connector cannot encode is accepted and the first INSERT INTO over it fails. The message arrives wrapped in Flink’s own “Unable to create a sink for writing table …” — the actionable sentence is in the cause. Per-record checks, such as an empty append operand or an input whose cells are all null, run in the sink on the TaskManager. Aggregate family existence and type checks inspect live Bigtable metadata during writer creation on the TaskManager.

Type mapping#

For ordinary cell writes and reads, a byte encoding convention has to be picked. This connector uses the HBase ecosystem’s — org.apache.hadoop.hbase.util.Bytes as Flink’s HBase connector applies it — reproduced here rather than depended on, since hbase-common drags in Hadoop. The row key takes the same encodings. Aggregate writes use integer contributions and skip nulls as described in Aggregate contributions; the cell byte encodings below apply to ordinary write modes and state reads.

ARRAY, MAP, MULTISET, a nested ROW, RAW and TIMESTAMP WITH TIME ZONE have no encoding and are rejected — but see below for when.

Precision stops at milliseconds#

A TIME or TIMESTAMP cell holds milliseconds, so a precision above 3 is rejected rather than silently truncated — the same bound the HBase connector draws. TIMESTAMP and TIMESTAMP_LTZ encode identically, to the same epoch-millisecond value.

Nulls#

A null is an empty cell for every type except a character string, where an empty cell is a legitimate value; a null there writes null-string-literal instead. Nulls are written rather than skipped: a qualifier left unwritten keeps whatever an earlier version of the row put there, which is not what “this column is null now” means. A whole column family that is null writes no cells at all, since there is no value to encode.

Two collisions the convention does not resolve, both inherited from the HBase connector. A null and a zero-length value are the same bytes in a BINARY, VARBINARY or BYTES cell — only a character string gets a marker — so a column that must tell them apart needs the distinction encoded in the value. And a character string whose value happens to equal null-string-literal reads back as a null; pick a literal the data cannot contain.

A read reverses the convention through the same option — What a read produces below.

Source#

With the default scan.mode = bounded, a SELECT is a bounded scan over the DataStream source — the same split planning, resumption and metrics that page describes — and it works in both batch and streaming jobs. The design record is ADR-0092.

Projection is pushed to the server as a family filter. The query’s retained column families become a filter the scan carries, so an unread family never leaves the server. A query that reads no family at all — SELECT rowkey, SELECT COUNT(*) — scans keys only: one cell per row, its value stripped, since Bigtable has no row without a cell. The projection is by whole columns; a retained family always arrives as its full declared ROW.

The filter is applied whether or not the query projects, naming exactly the declared families. Two consequences:

• A family the physical table has but the DDL does not declare is never read.
• A family the DDL declares but the table lacks fails the scan with the service’s NOT_FOUND (“Error while reading table ‘…’ : Requested column family not found”). The source does not pre-validate the DDL against the table — that would cost every scan a metadata read to soften an error the service already reports precisely, naming the table it read. A row-key-only query still answers, its keys-only filter naming no family.

Which rows a query sees follows from the storage model. A Bigtable row exists while it has a cell, so a query that reads families returns the rows with at least one cell in a family it reads — a row whose every read family is empty has nothing for the server-side filter to return and does not appear. SELECT * includes a row holding data in any declared family, with the empty ones NULL; a narrower projection can exclude that same row; and a query reading no family — SELECT rowkey, COUNT(*) — sees every physical row, including one whose cells all live in families the DDL never declared. Which columns a query selects therefore also decides which rows it sees. This is the wide-column model’s row existence, not an artifact of the pushdown. Flink’s HBase connector also makes row membership projection-dependent, but selects each declared qualifier; this connector filters at the family boundary, so a row holding only an undeclared qualifier in a read family appears here with that family NULL where HBase omits it. Projecting the row key alongside a family does not change membership: once a query reads a family, only rows with a cell in a read family appear. A keys-only query is the SQL shape that sees every physical row.

The latest version of each cell is read. Bigtable stores timestamped versions; the scan takes the newest per qualifier. A qualifier the declared family holds but the DDL does not name is ignored.

Filter pushdown#

The source consumes row-key predicates exactly when the SQL comparison has the same ordering as the HBase-compatible byte encoding. Exact equality, inequality and IN predicates become one or more Bigtable row ranges. AND intersects ranges, and OR unions them. The ranges configured by scan.row-prefix, scan.row-range.* and scan.row-ranges are first treated as one union, then intersected with the SQL ranges. The same final ranges and filters serve the bounded scan and a FULL-cache loader created from that filtered source plan.

Fixed-width integer and temporal decoders ignore suffix bytes by default, so equality uses a key-prefix range rather than only the canonical encoded key. Under decode.trailing-bytes = 'reject' those predicates stay with Flink instead: no range is exact for a fixed-width key there — a prefix set admits a suffix-bearing key as an = match while its complement excludes that key from a <> scan that must fail on it — so the evaluation stays on the decoded value, where the policy throws, at the cost of the pushdown. Their encodings do not preserve signed SQL order. An empty VARCHAR or VARBINARY literal remains with Flink: the SDK cannot express an empty-key range, normalising an empty bound to unbounded, which would widen the scan rather than narrow it. Fixed-width character and binary values may require SQL padding, a nonzero byte decodes as boolean true, decimal encodings carry their own scale, and floating point has signed-zero and NaN semantics. Those many-to-one or noncanonical cases remain with Flink. An expression outside the table, including a cast or computed expression around the row key, stays with Flink unless the planner has already reduced it to a supported field-literal form.

Positive predicates on a family or qualifier use a best-effort cell-existence prefilter. For example, cf1 IS NOT NULL, cf1.name IS NOT NULL, cf1.name = 'alice' and cf1.name IN ('alice', 'bob') can avoid returning rows that have no relevant cell. The source reports the same SQL expression as a residual filter, so Flink still compares the decoded value and applies its null semantics. IS NULL, NOT, an OR with an unsupported branch and other predicates that cannot yield a necessary positive existence test stay entirely with Flink.

The connector deliberately does not push raw cell-value comparisons. Bigtable compares encoded bytes rather than decoded SQL values, an empty or sentinel cell may decode as NULL, and a row may hold several timestamped versions while SQL sees only the latest. Those differences make a raw value filter unsafe as the final SQL answer. The existence prefilter is composed with projection through a Bigtable conditional row filter; Google documents conditional filters as non-atomic and warns that they can perform poorly. Treat this pushdown as an opportunity to reduce returned rows, not as a guarantee that every cell predicate makes a read faster.

What a read produces#

The cell bytes decode by the same type mapping the write side uses, and nulls reverse the write-side convention: an empty cell is NULL — except in a character-string column, where the null-string-literal is NULL and an empty cell is an empty string. A column family none of whose declared qualifiers has a cell is a NULL field, mirroring the sink, whose null family writes no cells; a family with some cells is a ROW whose absent qualifiers are null. (Flink’s HBase connector differs here: it always builds the nested row.)

Three more read-side facts worth knowing:

• A decimal wider than its column fails the read. A cell whose value, rounded to the declared scale, needs more precision than the declared DECIMAL(p, s) allows fails the scan with a message naming the cell and its row, the same way a fixed-width cell shorter than its declared layout does. Flink’s HBase connector reads such a cell as a SQL NULL instead, which silently aliases a real value onto the empty-cell null convention — or hands a NOT NULL column a null; this connector deliberately diverges. Rescaling alone is not an overflow: a cell written at another scale reads, its fractional digits rounded to the declared scale, and fails only when the rounded value no longer fits — which a rounding carry can make true of a value whose stored digits look representable.
• decode.trailing-bytes decides what a fixed-width value longer than its layout does. Under the default ignore, trailing bytes after a complete fixed-width value are discarded, exactly as HBase’s Bytes decoder discards them — which is what lets a BIGINT column read the leading component of a composite key. The cost of that compatibility is that distinct values sharing a decoded prefix read as one SQL value, so a key column whose bytes were not written under this convention is better declared BYTES or STRING. Setting the option to reject fails the read on any length but the exact layout instead, with the same message a short cell gets. BOOLEAN is the one exception under either setting: Bytes.toBoolean rejects any cell that is not exactly one byte, and so does this decoder. (A nullable column’s empty cell never reaches a decoder under either setting — it is the null convention’s, above.)
• Declare a qualifier NOT NULL only when every row carries the cell. The read path cannot manufacture a value for an absent cell, so sparse data under a NOT NULL column hands the planner a null it was told cannot exist. What an empty cell does under NOT NULL depends on the declared type: a character-string or binary column decodes it to an empty value, while every other type fails the scan, the plain decoder having no null to offer and no bytes to read. (A nullable binary column reads the same empty cell as NULL, per the convention above — the two differ only under NOT NULL.)

Bounding the scan#

scan.row-prefix, the legacy scan.row-range.* pair and scan.row-ranges bound the scan by row key, server-side, and are additive — overlapping or adjacent selections are merged, so no row is read twice. scan.row-ranges uses semicolon-separated closed-start, open-end entries such as [account-a,account-m);[account-q,). Either endpoint may be omitted, but not both. Use a backslash before \, ;, ,, [, ], ( or ) when that character belongs to a UTF-8 endpoint rather than to the range grammar. The \\ sequence is the complete input for one literal backslash; it needs no additional prefix. Malformed, empty, equal or inverted entries fail table validation with their one-based entry number. scan.row-key-encoding = UTF8, the default, preserves the original text behavior. Set it to BASE64 to express exact binary keys using canonical padded RFC 4648 standard Base64. The Base64 mode rejects the URL-safe alphabet, whitespace, missing or non-canonical padding, and malformed input. The standard alphabet does not contain ;, so the existing separator for multiple prefixes stays unambiguous: the connector splits the list before decoding each prefix. Every mode rejects a value that decodes to an empty key because the client would silently widen it, and “scan everything” is spelled by leaving the option unset. Supported SQL row-key predicates further intersect this configured union. An empty intersection returns no rows rather than widening back to the configured range or the whole table.

Sink#

Aggregate contributions#

Set sink.write-mode = aggregate to contribute integer inputs to Bigtable aggregate cells. The required sink.aggregate.column-family-types map assigns each physical family one of int64-sum, int64-min, int64-max, or int64-hll. Every qualifier must be TINYINT, SMALLINT, INT, or BIGINT; inputs widen losslessly to INT64 and all four types use AddToCell. HLL accepts an integer to count, without a client-side sketch library. Raw families, missing or extra family declarations, and unsupported input types fail when the statement is planned.

The mode accepts INSERT-only input, including repeated inputs for the same row key with or without a declared PRIMARY KEY. An updating GROUP BY is rejected because its running totals are replacement values, while each input here contributes again. The mode does not retract earlier contributions or keep per-input aggregation state. Explicit sink.insert-only-input-mode, null-string-literal, and conditional-write options are rejected. The ordinary batch writer still owns buffering, flow control, failure isolation, and checkpoint draining.

A null family or scalar contributes nothing; an input with no non-null cells fails serialization. The row-key checks and writable timestamp metadata remain in force. An absent or null timestamp uses the millisecond-aligned writer clock, read per written cell. Provide a stable bucket timestamp when multiple inputs must address the same cell version. The aggregate example demonstrates all four types and a separate read schema.

Each writer validates the destination before accepting records. With sink.create-disposition = create-if-needed, it creates the table or adds missing typed families; the existing GC-rule requirement still applies. With the default create-never, the table and every declared family must already exist with compatible types. This check requires bigtable.tables.get, in addition to the selected data-write and creation permissions. A type mismatch fails the job naming the destination, family, actual type and expected type; it never converts a family or enters the row failure handler. Existing GC rules are neither compared nor changed, and undeclared stored families are untouched.

Aggregate input DDL is sink-only: scan and lookup planning reject it. Read the same table through a separate DDL without aggregate options, using BIGINT for SUM/MIN/MAX state and BYTES for HLL sketch state. The connector does not extract cardinality from an HLL sketch; Bigtable SQL provides HLL_COUNT.EXTRACT.

Delivery defaults to at-least-once; checkpoint-owned delivery protects each staged contribution with a retained marker. Under at-least-once delivery, a stable timestamp selects a cell but does not deduplicate a SUM contribution: replaying 3, 5, 3 changes SUM from 11 to 22. Repeating those values at the same timestamp leaves MIN 3, MAX 5, and the HLL result unchanged, provided no deletion or GC intervenes. A regenerated timestamp can create another aggregate version, including for those three types. In the default eager mode, neither batching retries nor a completed checkpoint establish exactly-once aggregation.

Insert-if-absent#

Set sink.write-mode to insert-if-absent to atomically insert cells only when the stored row has no cell anywhere. A cell in an undeclared family also makes the row exist. The existing row-key and family/qualifier schema, cell codec, nullable cells and writable timestamp metadata still apply. A null family writes no cells; a row whose every family is null is rejected.

CREATE TABLE new_users (
  row_key STRING,
  profile ROW<name STRING, email STRING>
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'users',
  'sink.app-profile-id' = 'single-cluster',
  'sink.write-mode' = 'insert-if-absent'
);

INSERT INTO new_users VALUES ('u1', ROW('Alice', 'alice@example.com'));

The table and families must exist, and the application profile must use single-cluster routing with single-row transactions enabled. The mode accepts INSERT-only input and works with plain INSERT VALUES and INSERT SELECT on the supported Flink versions. It uses one CheckAndMutateRow RPC per input: a whole-row existence check, an empty true branch, and the input cells in the false branch. Concurrent requests for one absent row have no defined winner; an atomic request that finds the row absent can insert it.

sink.conditional.empty-branch-policy is ignore unless configured as fail. The latter fails the job after a successful RPC selects the empty branch because the row exists. This is at-least-once delivery: if an applied insertion loses its acknowledgement, recovery can replay it against the now-existing row and fail repeatedly under fail. A checkpoint drains requests and does not make Bigtable writes transactional with Flink state.

Tune sink.request-timeout and sink.in-flight.max-requests through BigtableRequestOptions. Idle-timeout, active-instance and per-destination-metrics options are shared with the ordinary sink. Explicit batching, in-flight entry/byte limits, automatic table creation/repair, rejection-limit and sink.insert-only-input-mode settings are rejected for this mode. Conditional-only options are rejected under upsert and keep-latest. The scan and lookup paths remain available for this ordinary schema.

ON CONFLICT controls Flink planner/job behavior; it is not the destination’s atomic existence test. The planner and ordering discussion below applies to upsert and keep-latest.

DDL-defined conditional commands#

Set sink.write-mode to conditional when each input describes a command with a fixed predicate and two ordered mutation branches. The DDL fixes the operation kinds, target cells and order; input columns supply values. Compute and convert values in ordinary SELECT expressions.

CREATE TABLE conditional_updates (
  row_key STRING,
  expected_status BYTES,
  new_status BYTES,
  activation_delta BIGINT,
  mismatch_reason BYTES
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'users',
  'sink.app-profile-id' = 'single-cluster',
  'sink.write-mode' = 'conditional',
  'sink.conditional.row-key-column' = 'row_key',
  'sink.conditional.predicate' = 'latest-cell-value-equals',
  'sink.conditional.predicate.family' = 'profile',
  'sink.conditional.predicate.qualifier' = 'status',
  'sink.conditional.predicate.value-column' = 'expected_status',
  'sink.conditional.then.0.operation' = 'set-cell',
  'sink.conditional.then.0.family' = 'profile',
  'sink.conditional.then.0.qualifier' = 'status',
  'sink.conditional.then.0.value-column' = 'new_status',
  'sink.conditional.then.1.operation' = 'add-to-cell',
  'sink.conditional.then.1.family' = 'stats',
  'sink.conditional.then.1.qualifier' = 'activated',
  'sink.conditional.then.1.timestamp-micros' = '0',
  'sink.conditional.then.1.value-column' = 'activation_delta',
  'sink.conditional.otherwise.0.operation' = 'set-cell',
  'sink.conditional.otherwise.0.family' = 'audit',
  'sink.conditional.otherwise.0.qualifier' = 'reason',
  'sink.conditional.otherwise.0.value-column' = 'mismatch_reason'
);

INSERT INTO conditional_updates VALUES (
  'u1',
  CAST('pending' AS BYTES),
  CAST('active' AS BYTES),
  1,
  CAST('status mismatch' AS BYTES)
);

The profile and audit families must exist, and stats must already be an int64-sum family for this example. The application profile must use single-cluster routing with single-row transactions enabled. A profile ID alone does not trigger an admin lookup; service rejections retain their cause and explain this prerequisite.

This is a command-input schema. sink.conditional.row-key-column names one top-level physical column with a supported cell encoding. Each value-column or timestamp-column names a top-level physical column exactly, including case; nested paths and expressions are not parsed in option strings. PRIMARY KEY and metadata declarations are rejected. Unused physical columns do not affect the request. The table is write-only: declare a separate ordinary table to scan or look up stored cells.

Predicates#

A qualifier is exactly one of qualifier (UTF-8) and qualifier-base64 (canonical padded RFC 4648 Base64). An empty qualifier is valid. A value binding is exactly one of value-column, value-utf8, value-base64 and value-int64. For predicates and SetCell, columns use their declared cell encoding, UTF-8 and Base64 literals bind bytes, and int64 literals encode eight big-endian bytes. Empty byte values are valid. Row existence rejects cell and comparison attributes; cell existence rejects comparison values. Arbitrary composable filters remain available through the DataStream conditional API.

Ordered branches#

Use expanded map keys such as sink.conditional.then.0.operation. Indexes are canonical nonnegative integers, consecutive from zero within each branch, and execute in numeric order. The standard Flink packed map spelling is also accepted, but mixing packed and expanded spellings within a branch is rejected. An omitted branch is empty; at least one branch must contain a mutation. Each branch permits at most 100,000 mutations.

AddToCell and MergeToCell value columns must be BIGINT or BYTES. BIGINT and value-int64 use the service’s typed int64 value; BYTES, value-utf8 and value-base64 use its typed bytes value. INT is not widened and bytes are not reinterpreted as integers. The service checks these inputs against the pre-existing aggregate family. For Int64 Sum MergeToCell, read the accumulator bytes from a stored aggregate cell; do not assume their encoding. The int64-sum service measurement does not establish support for every aggregate type.

Timestamps and nulls#

A timestamp binding is exactly one of timestamp-micros, a signed integral literal, and timestamp-column, a BIGINT column containing microseconds. SetCell without a binding reads a millisecond-aligned writer clock for each cell. An explicit SetCell timestamp of -1 requests server time; other negative values are rejected. AddToCell and MergeToCell require an explicit nonnegative timestamp, including zero. An aggregate timestamp never defaults to the writer clock. Explicit timestamps are passed unchanged, so a table-granularity mismatch remains a service error.

DeleteCells bounds use start-timestamp-micros or start-timestamp-column, and end-timestamp-micros or end-timestamp-column. The start is inclusive and nonnegative; the end is exclusive. An omitted bound is unbounded. Empty or reversed intervals are rejected before the RPC, including an explicit end of zero in the nonnegative timestamp domain.

NULL in any referenced column is an error before submission, including values or timestamps in the branch the service will not select. Both branches form the request and must be valid. NULL never deletes a cell, omits a mutation or selects a timestamp default. The nullable family/cell behavior of insert-if-absent is separate.

Outcomes and delivery#

Plain INSERT VALUES and INSERT SELECT are supported; updating and retracting input is rejected. The sink submits requests asynchronously, drains accepted requests at checkpoints, discards the Boolean result and records predicate match, predicate miss and selected-empty-branch counters. A false result succeeds and can execute a nonempty otherwise branch. sink.conditional.empty-branch-policy=ignore accepts an empty selected branch; fail fails the job after counting the completed RPC and its outcome. It tests whether the selected list is empty, not whether the mutations changed stored bytes, and remains valid but inactive with two nonempty branches.

The runtime uses one SDK deadline and no automatic retry. Delivery is at-least-once: recovery can replay an applied request whose acknowledgement was lost and select a different branch. With empty-branch-policy=fail, a successful insertion can therefore fail repeatedly on replay. Checkpoints drain accepted requests; they do not provide exactly-once effects or transactions across rows. Separate RPCs for the same row have no ordering guarantee. Flink ON CONFLICT does not express the Bigtable predicate.

Request timeout, in-flight request count, idle timeout, instance cap, per-destination metrics, credentials, endpoint, application profile and sink parallelism use the existing conditional runtime settings. Explicit batching, entry/byte flow control, table creation/repair, insert-only compatibility, legacy timestamp truncation, null-cell encoding, decode, aggregate-family creation and scan/lookup options are rejected for this command schema. Staged exactly-once delivery is unavailable for this mode. Command-template options are rejected in every other write mode.

Keep-latest#

Set sink.write-mode to keep-latest to replace all versions of each cell the input writes. For each target qualifier, the sink appends an unbounded column delete immediately followed by its replacement SetCell in one RowMutationEntry. Multiple qualifiers remain one atomic row operation, following Google’s delete-then-write recommendation.

A scalar null still writes the existing empty-byte or null-string-literal encoding and replaces that cell’s history. A whole null family writes and deletes nothing; stored families and qualifiers absent from the DDL are untouched. A partial SQL INSERT that materializes a null scalar therefore replaces that cell, while an omitted family is preserved. A row with every family null remains invalid. INSERT and UPDATE_AFTER replace the targeted cells; DELETE still removes the entire stored row.

GC settings are independent. sink.table-create.gc-rule.max-versions = 1 configures eventual retention when a family is created; garbage collection is asynchronous and can leave multiple versions visible after a write. Keep-latest instead leaves one logical replacement per targeted cell when its entry completes, before another write or applicable GC changes it. It does not promise immediate physical storage reclamation. The following tables share the same GC policy but use different write operations:

CREATE TABLE retained_profiles (
  row_key STRING,
  profile ROW<name STRING, email STRING>,
  PRIMARY KEY (row_key) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'retained-profiles',
  'sink.write-mode' = 'upsert',
  'sink.insert-only-input-mode' = 'insert-only',
  'sink.create-disposition' = 'create-if-needed',
  'sink.table-create.gc-rule.max-versions' = '1'
);

CREATE TABLE latest_profiles (
  row_key STRING,
  profile ROW<name STRING, email STRING>,
  PRIMARY KEY (row_key) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'latest-profiles',
  'sink.write-mode' = 'keep-latest',
  'sink.insert-only-input-mode' = 'insert-only',
  'sink.create-disposition' = 'create-if-needed',
  'sink.table-create.gc-rule.max-versions' = '1'
);

INSERT INTO retained_profiles VALUES ('u1', ROW('Alice', 'alice@example.com'));
INSERT INTO latest_profiles VALUES ('u1', ROW('Alice', 'alice@example.com'));

Both use the ordinary batch writer and its batching, flow-control and table-creation settings. Table creation still requires a GC rule; selecting keep-latest does not invent one. The sink.insert-only-input-mode option retains its planner meaning; the example selects the compatibility mode so plain inserts work across the supported Flink versions.

The cell timestamp rules are unchanged: absent or null metadata selects the per-cell writer clock, and explicit metadata retains its value and optional truncation. Reapplying a record replaces the cells again, so a new writer timestamp does not accumulate another version. It can still change the stored timestamp, overwrite a newer event, and produce repeated Change Streams mutations. Keep-latest provides neither compare-and-set nor latest-event-time-wins, and it does not order separate entries for one row or change at-least-once delivery. Each written cell consumes two mutations; the SDK’s mutation limits still apply, while batching and in-flight entry limits continue to count row entries. Age-based GC can remove a replacement carrying an old explicit timestamp. ADR-0153 records the decisions.

Append and increment#

sink.write-mode = append appends each nonnull cell to its latest stored value. Use CHAR, VARCHAR, BINARY or VARBINARY cells; the existing codec supplies UTF-8 strings and unchanged binary bytes.

CREATE TABLE account_notes (
    rowkey STRING,
    activity ROW<notes STRING, binary_log BYTES>
) WITH (
    'connector' = 'bigtable',
    'project' = 'my-project',
    'instance' = 'my-instance',
    'table' = 'accounts',
    'sink.app-profile-id' = 'single-cluster',
    'sink.write-mode' = 'append'
);
INSERT INTO account_notes
VALUES ('account-1', ROW(' payment received;', X'0102'));

sink.write-mode = increment adds each nonnull BIGINT cell to its latest stored signed integer. Negative values decrement, and zero remains an operation. For workloads that can use aggregate cells, prefer AddToCell; this mode addresses raw families.

CREATE TABLE account_counters (
    rowkey STRING,
    counters ROW<balance BIGINT, adjustments BIGINT>
) WITH (
    'connector' = 'bigtable',
    'project' = 'my-project',
    'instance' = 'my-instance',
    'table' = 'accounts',
    'sink.app-profile-id' = 'single-cluster',
    'sink.write-mode' = 'increment'
);
INSERT INTO account_counters
VALUES ('account-1', ROW(CAST(-5 AS BIGINT), CAST(1 AS BIGINT)));

One DDL selects one operation and fixes its project, instance, table and sink application profile. Mixed or incompatible cell types fail during planning. Each family and qualifier becomes a rule in declaration order. NULL families and cells omit rules; null-string-literal does not change this behavior. A row with no remaining rule, a null or empty row key, or an empty append value fails. The cell type describes the input operand rather than a maximum length or range for accumulated state. Bigtable performs increment arithmetic, including overflow, without a connector-side read or overflow policy. An existing increment target must contain the service’s eight-byte big-endian signed representation.

These modes accept INSERT-only input, including repeated same-key inputs with or without a primary key. They wait for one ReadModifyWriteRow response per input and discard the returned cells. To consume those values, use the DataStream async helper or the async SQL functions in Flink 2.x streaming mode.

The table and families must exist, and the sink profile must use single-cluster routing with single-row transactions enabled. The request timeout, in-flight request limit and client-lifecycle options use BigtableRequestOptions. Explicit batch, entry/byte flow-control, creation/repair, rejection-limit, insert-only compatibility and conditional empty-branch options are rejected. Writable timestamp metadata and sink.cell-timestamp.truncate-to-millis are also rejected because read-modify-write uses service timestamps. The existing scan and lookup paths remain available.

These are non-idempotent operations with at-least-once delivery. The connector does not retry an ambiguous RPC; it fails the job and explains that recovery can apply an already committed append or increment again. A checkpoint drains requests without committing a transaction between Bigtable and Flink state.

Flink 2.3 may demand ON CONFLICT#

Flink 2.3 (FLIP-558) refuses to plan into an upsert sink table declaring a PRIMARY KEY when the query’s upsert key differs from that key — or when it cannot infer one at all, as for INSERT INTO .. VALUES — unless the statement carries one of the ON CONFLICT DO DEDUPLICATE / DO ERROR / DO NOTHING clauses Flink 2.3 introduces. The refusal is the planner’s, not this connector’s: rows with different upsert keys mapping to one primary key reach the sink in no defined order, and Flink 2.3 asks the query to say what that should mean instead of materializing silently. An updating query whose upsert key is the primary key plans with no clause, exactly as on 2.2.

The default sink.insert-only-input-mode = upsert applies that rule to an insert-only input too. It exposes all three conflict strategies and reflects the physical write, which overwrites a row with the same key. On an insert-only input, DO DEDUPLICATE adds no materializer and leaves those writes on the ordinary overwrite path. DO NOTHING keeps the first row the job observes per key and DO ERROR fails when the job observes a conflict; both require watermarks. Neither inspects Bigtable, so DO NOTHING is not an atomic insert-if-absent — a fresh job, expired state or cleared state can overwrite an existing row.

For a plain insert that must use the same DDL on Flink 1.20, 2.2 and 2.3, set sink.insert-only-input-mode = insert-only. This table-local compatibility mode restores the append answer introduced by #488, so the statement plans without the 2.3-only clause. In exchange, Flink rejects an ON CONFLICT clause on that insert-only statement because the sink has advertised only INSERT changes. Updating inputs remain upsert and are unaffected by the option.

table.exec.sink.require-on-conflict = false is the supported planner-wide alternative. It lets 2.3 plan a statement without the clause while leaving this sink in upsert mode; measured on 2.3.0, the plain insert plan then carries upsertMaterialize=[true], while the connector’s insert-only mode carries no materializer. Flink 1.20 and 2.2 do not define the setting and ignore it. Prefer the connector option when the compatibility decision belongs to one Bigtable table and the extra keyed state is not wanted; use the Flink setting when relaxing the check for every sink planned by the session is intentional.

Two rows for one key in one batch have no defined winner#

With at-least-once upsert, keep-latest and aggregate, the writer hands entries to the client’s bulk-mutation batcher, and Bigtable’s own contract for MutateRows is that its entries “may be applied in arbitrary order (even between entries for the same row)”. So when a job produces two changelog rows for the same key close enough together to share a request, which one lands last is not defined — and if they share a millisecond they also share a cell timestamp, so ordinary upserts collapse to one version rather than two. Keep-latest removes prior versions regardless of timestamp, but does not define the winner.

Separate requests are not the fix, and setting sink.batching.element-count-threshold to 1 to force one entry per request is the shape that looks like it. The batcher sends each request without waiting for the previous one’s response, so requests a single job has in flight are concurrent rather than ordered — measured, forcing one entry per request made a delete stop taking effect. What has a defined order is separate writes. A request whose response has been awaited before the next is issued is ordered, which is what successive jobs give you.

A job that needs last-write-wins per key cannot obtain it from this sink’s submission order. It can put the version in the row key, or separate dependent mutations into writes whose completion is awaited before the next begins. Upstream aggregation helps only when it emits at most one mutation per key for the lifetime of the write; windowed aggregation can still leave requests concurrent. Google’s own guidance lists “multiple mutations to the same row” under when not to use batch writes.

The sink does not serialize same-key entries. A real-service campaign submitted 86,196 same-row pairs in mirrored arms, across request sizes from 2 through 19,998 entries, and observed no submission-order reversals on 2026-08-11. That is evidence about the tested service behaviour, not a contract: the documented arbitrary-order allowance still applies, and a future service or client change may exercise it. Enforcing one in-flight entry per key would add key-indexed pending state that grows with active keys and head-of-line blocking to protect a behaviour the campaign did not observe; ADR-0093 records why #471 therefore keeps the existing bulk path and this explicit caveat.

Cell timestamps#

The ordinary cell sink exposes writable metadata named timestamp with type TIMESTAMP_LTZ(6). The conditional command schema uses its per-mutation timestamp bindings and rejects metadata declarations. One value is applied to every cell written by that row; a delete ignores it because deleteRow has no cell timestamp. In modes that accept this metadata, a value before the Unix epoch is rejected as a serialization failure. A negative SQL timestamp cannot request server time; use NULL for the writer clock.

CREATE TABLE profiles_with_event_time (
  rowkey STRING,
  profile ROW<name STRING, email STRING>,
  cell_timestamp TIMESTAMP_LTZ(6) METADATA FROM 'timestamp',
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles',
  'sink.insert-only-input-mode' = 'insert-only'
);

INSERT INTO profiles_with_event_time
SELECT user_id, ROW(name, email), event_time FROM staged_profiles;

Flink casts the metadata column to the advertised TIMESTAMP_LTZ(6) type before the sink runtime receives it. A declaration below precision 6, such as TIMESTAMP_LTZ(3), is widened without adding fractional digits. A declaration above precision 6, such as TIMESTAMP_LTZ(9), is truncated to microseconds by that cast, independently of sink.cell-timestamp.truncate-to-millis.

An absent metadata column or a NULL value takes the writer clock: the sink stamps the cell with the TaskManager’s current millisecond when the mutation is built. The value is always millisecond-aligned, so this path cannot produce the granularity rejection below, and sub-millisecond precision is not kept — use the metadata column when you need it. The sink stamps it rather than letting the client library do so, which is what it did until ADR-0149. In upsert mode the client reuses the same RowMutationEntry when it retries an RPC, so that retry rewrites the same cell version. The insert-if-absent runtime makes one RPC attempt without retry. A Flink recovery serializes the record again, however, and therefore takes a new writer-clock value. Use a stable event timestamp from the record when the same version must be addressed across job replay.

Bigtable stores timestamps as epoch microseconds but accepts values only at millisecond granularity. By default, sink.cell-timestamp.truncate-to-millis is false: the connector sends the explicit microsecond value unchanged and lets Bigtable reject a value whose last three digits are nonzero. Set the option to true to opt into dropping those three digits before the mutation is sent. This is truncation, not rounding; every cell written by the row receives the same truncated value.

Three record-level rejections#

Each of these fails the record through the sink’s failure handler rather than skipping it:

• An UPDATE_BEFORE row. The declared changelog mode means the planner never sends one, so its arrival is a bug — and treating it as a delete, which is what a two-branch converter does, would erase the row the following UPDATE_AFTER is about to rewrite.
• A row key that is null or encodes to zero bytes. Bigtable has no such row; Flink’s HBase connector drops the record instead, which leaves an incomplete table under a green job.
• A row whose every column family is null, which would produce a mutation with no cell in it. The service refuses that with an INVALID_ARGUMENT naming neither the row nor the reason, so the connector refuses it where both can be said. A partial column list — INSERT INTO t (rowkey) VALUES (...) — is the ordinary way to reach it.

Retries, error classification and the failure handler are the DataStream sink’s and are described on its page. A SQL table has no failure-policy option, so its sink always fails the job on a routed failure.

Lookup joins#

The table supports processing-time temporal joins by equality on its row-key column. The lookup key is interpreted after projection, so the row key may appear anywhere in the DDL and the query may reorder the lookup table’s output. Composite keys, nested family fields and predicates that do not include row-key equality are rejected when the join is planned.

SELECT e.event_id, p.profile.name
FROM events AS e
LEFT JOIN profiles FOR SYSTEM_TIME AS OF e.proc_time AS p
  ON e.user_id = p.rowkey;

By default each input row performs a synchronous Bigtable point read. Set lookup.async = true for asynchronous point reads. A missing Bigtable row produces no lookup result, so a left join keeps the input row with null lookup columns and an inner join drops it. Both forms apply the same family projection filter as a scan and use scan.app-profile-id; a Data Boost application profile cannot serve the point-read modes. Flink does not currently pass an additional right-side temporal-join predicate through SupportsFilterPushDown. It keeps that expression in the lookup operator, where NONE, PARTIAL and FULL modes evaluate the same residual predicate. Configured scan.row-prefix, scan.row-range.* and scan.row-ranges bounds still apply to point reads and FULL-cache contents.

lookup.cache = PARTIAL uses Flink’s standard on-demand lookup cache around either the synchronous or asynchronous function. Configure at least one of lookup.partial-cache.max-rows, lookup.partial-cache.expire-after-access or lookup.partial-cache.expire-after-write with it. lookup.cache = FULL loads the projected table through a bounded scan into each lookup task; choose PERIODIC with lookup.full-cache.periodic-reload.interval, or TIMED with lookup.full-cache.timed-reload.iso-time. FULL is synchronous by definition, so combining it with lookup.async = true is rejected. Its scan may use a Data Boost profile. Scan prefix and range bounds apply consistently to point reads and FULL-cache contents.

Point reads retry only DEADLINE_EXCEEDED, UNAVAILABLE and ABORTED, which are the statuses the Bigtable client itself retries a point read on, so lookup.max-retries buys more attempts rather than a different retry policy. It counts retries after the first attempt and defaults to 3. Other failures surface immediately, RESOURCE_EXHAUSTED among them: Bigtable raises it for an exhausted admin API quota, node limit or node storage limit, and another point read clears none of those. The connector adds no lookup-specific metrics; Flink owns cache metrics and the Bigtable client owns RPC metrics.

Change Streams#

Set scan.mode = change-stream to read the table’s mutation log through the DataStream Change Streams source. The default remains bounded, so existing DDL keeps its current row-scan behavior.

A Change Streams table is source-only: an INSERT INTO naming one is rejected when the statement is planned, rather than writing to the table as an ordinary sink and discarding the scan.change-stream.* options. Its physical envelope has exactly the row_key and entries columns; this example also selects all optional metadata as virtual columns:

CREATE TABLE profile_mutations (
  row_key BYTES,
  entries ARRAY<ROW<
    entry_index INT,
    kind STRING,
    family STRING,
    qualifier ROW<value_type STRING, bytes_value BYTES, long_value BIGINT>,
    `timestamp` ROW<value_type STRING, bytes_value BYTES, long_value BIGINT>,
    `value` ROW<value_type STRING, bytes_value BYTES, long_value BIGINT>,
    delete_range ROW<
      start_bound STRING,
      start_micros BIGINT,
      end_bound STRING,
      end_micros BIGINT
    >
  >>,
  mutation_type STRING NOT NULL
    METADATA FROM 'mutation-type' VIRTUAL,
  source_cluster_id STRING
    METADATA FROM 'source-cluster-id' VIRTUAL,
  commit_timestamp TIMESTAMP_LTZ(9) NOT NULL
    METADATA FROM 'commit-timestamp' VIRTUAL,
  tie_breaker INT NOT NULL
    METADATA FROM 'tie-breaker' VIRTUAL,
  estimated_low_watermark TIMESTAMP_LTZ(9) NOT NULL
    METADATA FROM 'estimated-low-watermark' VIRTUAL
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles',
  'scan.mode' = 'change-stream',
  'scan.change-stream.changelog-mode' = 'envelope',
  'scan.app-profile-id' = 'single-cluster-profile'
);

The source emits one INSERT row per Bigtable mutation. entries retains the service order, and entry_index is its zero-based position in that list. The source rejects a primary key because two mutations for the same row_key are distinct log records rather than updates to one Flink row.

The metadata columns expose scalar fields attached to the mutation:

The declared column names are local to the DDL; METADATA FROM selects the stable connector key. Flink permits an explicitly castable declared type and applies the cast after the source, while the connector always emits the type in the table above. Marking the columns VIRTUAL keeps them out of the physical row, which is what the envelope schema check reads.

A generic value sets value_type to RAW_VALUE, RAW_TIMESTAMP, or INT64. RAW_VALUE populates bytes_value; the other two populate long_value, with raw timestamps in microseconds. Delete bounds use OPEN, CLOSED, or UNBOUNDED, and an unbounded endpoint has a null micros field. Fields that do not apply to an entry kind are null. If a later client library introduces an entry or value subtype the SDK converter does not know, the job fails with that subtype’s class name before emitting an incomplete mutation.

UNNEST expands the ordered entry array for relational processing:

SELECT
  row_key,
  mutation_type,
  commit_timestamp,
  entry_index,
  kind,
  family,
  qualifier,
  entry_timestamp,
  entry_value,
  delete_range
FROM profile_mutations
CROSS JOIN UNNEST(entries) AS entry_table(
  entry_index,
  kind,
  family,
  qualifier,
  entry_timestamp,
  entry_value,
  delete_range
);

SQL result rows have no implicit arrival order. entry_index carries each entry’s original zero-based service position through the expansion, so a downstream keyed computation can reconstruct the order without relying on UNNEST output order.

The envelope is a mutation record, not a reconstructed Bigtable row. Bigtable does not supply before or complete after images, so a cell or family deletion remains an inserted envelope row rather than a Flink DELETE or UPDATE.

Selected-cell upserts#

Set scan.change-stream.changelog-mode = selected-cell only when one Bigtable cell contains the complete serialized non-key part of a logical row. The mutation row key supplies exactly one declared physical primary key, which may appear anywhere in the DDL, and value.format decodes every other physical column. At least one non-key column is required. The primary key decodes under the same fixed-width rule as every other read, and this mode is where its default bites hardest: a key longer than a fixed-width primary key’s layout silently decodes as its prefix, and the DELETE or UPDATE_AFTER it keys then lands on the wrong Flink row. Set decode.trailing-bytes = 'reject' here unless the table’s keys are known to be exact, so such a mutation fails the read instead — see what a read produces.

CREATE TABLE current_profiles (
  name STRING,
  profile_id STRING NOT NULL,
  score INT,
  source_cluster_id STRING
    METADATA FROM 'source-cluster-id' VIRTUAL,
  commit_timestamp TIMESTAMP_LTZ(9) NOT NULL
    METADATA FROM 'commit-timestamp' VIRTUAL,
  PRIMARY KEY (profile_id) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles',
  'scan.mode' = 'change-stream',
  'scan.change-stream.changelog-mode' = 'selected-cell',
  'scan.app-profile-id' = 'single-cluster-profile',
  'scan.change-stream.selected-cell.family' = 'state',
  -- Base64 for the qualifier "current"; an empty qualifier is ''.
  'scan.change-stream.selected-cell.qualifier-base64' = 'Y3VycmVudA==',
  'scan.change-stream.selected-cell.source-cluster-id' = 'cluster-a',
  'value.format' = 'json'
);

The source recognizes only this atomic producer protocol:

• An upsert is one full delete of the selected column across all timestamps, or one delete of the selected family, followed by exactly one selected SetCell in the same mutation.
• A delete is that full selected-column or selected-family delete without a later selected SetCell and produces a key-only DELETE.
• Entries for other cells and families produce no row.

The upsert is emitted as UPDATE_AFTER, not an invented INSERT. A downstream keyed table can materialize the first update it observes as the current value. The source performs no point lookup, stores no previous row image, and does not scan a snapshot before its configured Change Streams start position.

The configured source cluster must match every mutation that affects the selected cell. This excludes cross-cluster conflict ordering from the changelog contract. The job fails on standalone, repeated, or out-of-order selected SetCell entries; partial timestamp deletes; garbage-collection or aggregate mutations affecting the selected cell; or a mutation from another cluster. The chosen format must be insert-only and emit exactly one non-null row for every upsert. Format metadata is not exposed because a key-only delete has no payload to decode.

For a full selected-column delete, the source also recognizes the SDK’s CLOSED(0) / OPEN(0) range: the Bigtable RPC timestamp range uses zero for the default inclusive lower bound and for an omitted, infinite upper bound. Positive lower bounds and finite upper bounds still fail the selected-cell protocol. Envelope mode preserves the SDK’s bound types and numeric values.

The keep-latest producer example stores a complete JSON object in one cell and reads it with the selected-cell JSON format. It distinguishes nullable JSON fields from a null cell payload and uses sequential completed writes to demonstrate replacement and reapplication.

These are producer requirements, not inferences the connector can make from arbitrary Bigtable traffic. A writer that does not atomically replace the complete selected value with the sequence above must use the lossless envelope mode instead.

The application profile is required and must route to one cluster, as on the DataStream source. The emulator is rejected because it does not implement Change Streams. Row ranges, the HBase-compatible cell codec, lookup options, projection pushdown, and filter pushdown belong to the bounded source and are not Change Streams settings. The factory rejects those options in Change Streams mode and rejects Change Streams options in bounded mode. A sink is held to the same rule: a table written to may carry the scan and lookup options it also reads with, but a Change Streams option on it is rejected rather than ignored.

An absent scan.startup.mode retains the DataStream builder’s latest-position default. Choose earliest or timestamp; the timestamp mode also requires scan.startup.timestamp-millis. Restored checkpoint state wins over that fresh-start setting. If a restored position has expired, the job fails unless scan.resume-fallback.mode explicitly opts into earliest, latest, or a timestamp with scan.resume-fallback.timestamp-millis. This restore contract is ADR-0094.

Set scan.bounded.timestamp-millis to make the source bounded; without it the source is continuous. scan.max-concurrent-streams-per-subtask bounds open partition reads in each source subtask and keeps the DataStream builder’s default of two when absent. Source parallelism multiplied by that value is configured job capacity, not a Bigtable quota. No option sets the service heartbeat interval, and Spanner’s scan.change-stream.heartbeat-interval has no counterpart here: Bigtable’s five seconds are fixed, because the same value paces how the reader rotates queued partitions in. ADR-0103 records that decision.

Continuation tokens and partition ranges remain internal checkpoint protocol state rather than queryable metadata. The estimated-low-watermark value belongs to the partition that produced the mutation; it is not a safe stream-wide event-time frontier, and the connector does not provide native SOURCE_WATERMARK() support. Bigtable permits a future record below a previously observed estimate, so including queued and enumerator-held partitions would still not satisfy Flink’s non-early contract.

Do not declare SOURCE_WATERMARK() for this connector. Flink can retain that expression in a separate watermark operator when the source does not support pushdown, but the Bigtable source does not provide the source-generated watermark it requests.

A DDL can instead declare an ordinary, application-owned watermark expression over commit-time metadata:

commit_timestamp TIMESTAMP_LTZ(3) NOT NULL
  METADATA FROM 'commit-timestamp' VIRTUAL,
WATERMARK FOR commit_timestamp AS commit_timestamp - INTERVAL '5' MINUTE

The five-minute delay is an example policy, not a recommended or service-backed bound. Bigtable publishes no finite maximum lateness, and source concurrency can leave partitions queued or unassigned, so the job must choose how it handles records behind that watermark. See ADR-0109 for the decision and the different Spanner heartbeat contract.

Options#

Scan, connection and write options map onto builder setters of the DataStream API, which stays their source of truth. Lookup options are table-layer or Flink-owned instead. An option left out of the DDL leaves the corresponding setter or lookup setting untouched; the full list of defaults is in the configuration reference.

scan.max-rows-per-fetch and scan.max-bytes-per-fetch bound each bounded scan fetch before it hands decoded input to Flink’s element queue. The first limit reached ends the fetch; one row larger than the byte target is handed over alone. Top-level projection and pushed cell filters can reduce the cells returned before the byte estimate is measured; pushed row-key predicates instead reduce which rows enter the fetch. The bounds are per source subtask, so increasing scan.parallelism can increase the number of batches queued at once. They do not change point lookups or FULL-cache reloads, whose loader and memory ownership are Flink’s lookup-cache path rather than the bounded ScanTableSource reader. They are rejected when scan.mode = change-stream.

scan.mode, null-string-literal, decode.trailing-bytes, scan.row-key-encoding, lookup.async, sink.cell-timestamp.truncate-to-millis, sink.insert-only-input-mode, sink.write-mode and sink.aggregate.column-family-types belong to the table layer because they configure its codec, runtime shape or planner contract rather than a DataStream builder.

Destination#

Scan#

Lookup#

Sink options#

Table creation#

The families come from the DDL, not from a key. A ROW<...> column already says a family exists, so naming the same families again in the WITH clause would only create a way for the two lists to disagree.

The garbage-collection rule does not. A GcRule is a tree of unions and intersections to any depth, and a flat WITH namespace cannot carry one; the two keys above are the contraction, and they apply the same rule to every family. A family needing anything else is created out of band, which is what create-never is for.

At least one of the two keys is required under create-if-needed, which the DataStream API does not require. A family created with no rule keeps every version of every cell forever, and this ordinary upsert path is at-least-once: a replay using a new timestamp can add another version. The requirement also applies to keep-latest, keeping retention independent of the selected write operation and covering cells that later writes omit.

Setting either key without sink.create-disposition = create-if-needed is rejected rather than ignored. A table that declares no column family is rejected outright, whatever the disposition: a mutation with no cell in it is not a write.

Checkpoint-owned delivery#

Set sink.delivery-guarantee to exactly-once with upsert, keep-latest or aggregate to select the experimental staged runtime. Its production-service recovery acceptance was recorded on 2026-09-14 under #1319. A row becomes readable one checkpoint interval later at best, plus the commit drain, which is why the Stage 2 gate was declined on 2026-09-21 under #1327 and no supported workload is claimed. At the default sink.in-flight.max-requests of 100, with 1 KiB rows on distinct keys and a one-second checkpoint interval, the measured visibility p95 on a four-processor task manager host was 3.5 to 6.6 seconds with one subtask and 7.2 to 7.6 seconds with four, against 23 to 61 milliseconds for eager writes (#1464); the DataStream page explains why each staged write costs more than an eager one and what is known about the drain. The DataStream staged contract also governs SQL recovery, visibility and marker retention.

SET 'execution.runtime-mode' = 'streaming';
SET 'execution.checkpointing.interval' = '10 s';
SET 'execution.checkpointing.mode' = 'EXACTLY_ONCE';
SET 'execution.checkpointing.checkpoints-after-tasks-finish' = 'true';

CREATE TABLE staged_orders (
  rowkey STRING,
  cf ROW<status STRING>,
  PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'orders',
  'sink.delivery-guarantee' = 'exactly-once',
  'sink.app-profile-id' = 'transactional',
  'sink.staged.marker-family' = 'flink_commit',
  'sink.insert-only-input-mode' = 'insert-only'
);

INSERT INTO staged_orders VALUES ('order#1', ROW('created'));

Staged delivery does not require sink.insert-only-input-mode; the example sets it only so the same plain insert plans on Flink 1.20, 2.2 and 2.3, as Flink 2.3 may demand ON CONFLICT explains.

Provision cf and the reserved raw flink_commit family before running this example, with no GC rule on flink_commit and transactional single-cluster routing on transactional. Keep the marker family outside the DDL’s data families. CREATE_NEVER is required; table creation and GC-rule options do not configure this mode. Batch thresholds, batch in-flight limits and auto-creation recovery knobs are rejected; use sink.request-timeout and sink.in-flight.max-requests instead. General conditional, insert-if-absent, append, increment and async operations cannot select staged delivery.

For staged upsert and keep-latest, a DELETE removes all cells in the DDL’s declared data families. The marker family and undeclared families survive, so this is not physical row removal. The default at-least-once DELETE still removes the whole row. Aggregate mode still accepts only INSERT contributions and validates the declared aggregate family types before writing. All declared data families must exist; non-aggregate modes allow raw or typed non-aggregate families and reject aggregate families before staging.

Delivery guarantees#

See Write and key-collision semantics for the cross-connector distinction between an insert-only changelog and destination-side insert-if-absent behavior.

The sink defaults to at-least-once and advertises upsert by default, including when the requested input contains inserts alone. A Bigtable write is an upsert on the row key by construction — setCell overwrites — and there is no retract path to offer instead. On Flink 2.x the upsert mode says a delete may carry the upsert key alone only when the DDL declares the primary key, which is what makes that key the row key; The schema above has the consequence for a job. Flink 1.20 has no such distinction and always completes the row first.

An append-only row design must generate a unique row key for each logical event, because reusing a row key updates that row. An application that keeps cell history within one row must use ordinary upsert and choose distinct qualifiers or stable event timestamps deliberately. With ordinary upsert, a stable timestamp makes a replay target the same cell version, while omitting the timestamp uses the writer’s wall clock and can create a new version after Flink recovery. With keep-latest, replay replaces all versions of the targeted cells again; the timestamp and winning value can still change.

With default at-least-once delivery, a -D deletes the whole row, not the declared qualifiers one by one. The row key is the primary key, so “this key is gone” is what a delete means here; removing only the declared cells would leave a row behind made of whatever else was in it.

Design decisions#

The DDL model is the HBase connector’s, and that is the whole point. Upstream google/flink-connector-gcp models a family with a value.format, which cannot give a single qualifier its own type and ties a family to a format; it was weighed and declined on #34. apache/flink-connector-hbase has no Flink 2.x release, so the population this model serves has nowhere else to go.

The encoding is normative. It exists to be byte-compatible with the HBase ecosystem, so it is pinned to exact byte arrays by a golden-vector test rather than round-tripped through this connector’s own code, which would pass while the interop was broken.

The writable metadata surface contains only the cell timestamp. It applies one value to every cell of the row and deliberately does not timestamp deletes. The opt-in truncation option exists because the SQL type can carry microseconds while Bigtable accepts only millisecond-aligned values; preserving the user’s explicit value and letting the service validate it remains the default.

Per-record table routing has no SQL surface. A DDL names one table. Writing to several is a STATEMENT SET of INSERTs, one per table, which is what SQL already offers.

Projection pushdown is family pruning, served by one filter (ADR-0092). Bigtable’s read API takes a filter per scan, so a projection is a filter to build rather than an index list to apply client-side; the edge the ADR pins is the projection retaining no family, which must become a keys-only chain and not an empty filter. Qualifier-level pruning and a latest-version filter are compatible follow-ups the ADR names.

Testing#

The emulator suite drives CREATE TABLE, INSERT INTO and SELECT through the production factory, with the emulator endpoint interpolated into the DDL rather than injected through a test-only factory, and seeds or reads rows with its own client. The gated real-GCP suite covers the production-endpoint path that authenticates with application-default credentials; sink.app-profile-id and scan.app-profile-id, which the emulator ignores entirely; split planning, which needs a pre-split real table because the emulator models no tablets; and the family filter’s server-side NOT_FOUND for a declared family the table lacks, which the emulator answers with an empty result instead. The explicit-key path cannot accompany the emulator and does not need a service RPC to prove credential injection: unit and runtime-boundary tests parse a key file and inspect every affected client settings family.

The gated suite also checks immediate keep-latest replacement and reapplication with writer-clock and explicit SQL timestamps, plus server timestamps supplied by a DataStream serializer. Those tests read all stored versions without a latest-version filter or GC rule, and check that SQL writes preserve omitted families and undeclared qualifiers.

Change Streams metadata and UNNEST(entries) are covered without a service: converter tests use the connector-owned mutation model directly, planner tests select and cast metadata, and the existing flink-sql-connector-gcp-bigtable uber-jar plans an envelope DDL through its discovered bigtable factory. Selected-cell tests drive the strict mutation classifier and row assembly directly, while a MiniCluster job executes the canonical upsert, unrelated-mutation, and key-only-delete paths. The keep-latest interoperability unit test connects the factory-built sink serializer to the factory-built JSON decoder, including null fields, reapplication, selected-column deletion and invalid payloads. The emulator implements no Change Streams RPC, so the gated production-service suite runs the envelope DDL through the discovered bigtable factory on the existing ephemeral instance and single-cluster application profile. It starts from an explicit timestamp, finishes at a finite end timestamp, and verifies a binary row key, binary qualifier and value, ordered user writes and deletes, and all five readable metadata fields through SQL. A selected-cell service case writes complete JSON values with the SQL keep-latest sink and reads the resulting updates, repeated replacement and full selected-column deletion through the SQL source, collected with an explicit upsert changelog mode. Ordinary SQL result collection can normalize those records and suppress identical replacements. The deletion uses the data client; it is not a SQL DELETE FROM statement. Garbage-collection timing and retention expiry remain deterministic model and protocol tests rather than service-timed assertions.