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 neither a ROW nor a MAP, 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.
• Every MAP column is one column family too, declared without its qualifiers: each entry is a qualifier and its latest cell. The HBase connector has no such form; it is described under Map column families.
• 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.

On Flink 2.2 and 2.3 a delete carrying the key alone meets two Flink bugs, both fixed in 2.4.0. Such a delete’s non-key columns are null. Flink’s not-null enforcer checks them and fails the job when the table declares one of them NOT NULL (FLINK-40477); setting table.exec.sink.not-null-enforcer to DROP, as the error suggests, drops the delete instead, so the row is never deleted. A projection between the source and the sink also evaluates its expressions over those columns, which can fail the job when the source declares one of them NOT NULL (FLINK-40528). Declare the non-key columns nullable, in this table and in the source table, when the input’s deletes carry the key alone.

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, MULTISET, a nested ROW, RAW and TIMESTAMP WITH TIME ZONE have no encoding and are rejected — but see below for when. A MAP has no cell encoding either: a top-level MAP column is a whole column family, as Map column families describes, and is never the row key, while a MAP qualifier inside a ROW family is rejected.

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.

Map column families#

A ROW family needs its qualifiers and their types in the DDL, and Bigtable stores neither: a table’s metadata names its column families and nothing inside them. GoogleSQL for Bigtable therefore presents each family as a map from qualifier to latest value, and this connector accepts the same shape. A top-level column declared MAP<STRING, V> or MAP<BYTES, V> is the column family of that name, and one DDL may mix it with ROW families:

CREATE TABLE profile_attributes (
  rowkey STRING,
  profile ROW<name STRING>,
  attributes MAP<STRING, STRING>,
  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 profile_attributes
VALUES ('user-1', ROW('Alice'), MAP['plan', 'pro', 'region', 'eu']);

-- The GoogleSQL shape of the same families: qualifiers and values as bytes, cast in the query.
CREATE TABLE profiles_as_maps (
  rowkey STRING,
  profile MAP<BYTES, BYTES>,
  attributes MAP<BYTES, BYTES>
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'profiles'
);

SELECT rowkey, CAST(profile[CAST('name' AS BYTES)] AS STRING) AS name
FROM profiles_as_maps;

The key is a qualifier. BYTES (or VARBINARY) keeps the qualifier’s bytes as they are, which is GoogleSQL’s shape, and STRING (or VARCHAR) reads them as UTF-8 without validating them. CHAR, BINARY and every other key type are rejected, because a qualifier has no fixed length.

The value type V is any type in the type mapping, with the same byte layout and the same null convention, and every value of the family has that one type. MAP<BYTES, BYTES> reads any family. A typed value is how Flink reads a number out of a cell, since Flink has no CAST from BYTES to a numeric type: MAP<BYTES, BIGINT> reads a family of eight-byte integers, which includes the state of an int64-sum, int64-min or int64-max aggregate family, and MAP<BYTES, BYTES> reads HLL state. A cell the value type cannot decode fails the read with a message naming the cell, as it does for a ROW qualifier. A map value that is itself an ARRAY, a MAP or a ROW is rejected, and so is GoogleSQL’s with_history shape MAP<key, ARRAY<ROW<timestamp, value>>>: only the latest version of a cell is read.

A read produces one entry per qualifier that has a cell, holding that qualifier’s latest version. A family with no cell at all reads as NULL, never as an empty map, which is the rule a ROW family follows and which makes m IS NOT NULL mean that the family has a cell. That predicate is pushed to the server as a family-existence prefilter and also kept as a residual, like every cell predicate. An entry access such as m['plan'] = 'pro' is evaluated by Flink and never pushed. Row membership follows the family filter as it does for ROW families: a query that projects only a map family omits rows with no cell in it.

In the upsert, keep-latest and insert-if-absent modes, a write turns each entry into one cell; the accumulating modes are described below. A null value is written as a null ROW qualifier is, a null key fails the record, and a null map leaves the family untouched. A row that would carry more than 100,000 mutations or 200 MiB of them, the client’s limits for one row entry, fails as a serialization failure, which the sink routes like any other; under merge, keep-latest adds a column delete to each entry, which halves the number of entries a map can hold, while replace costs one family delete plus one mutation per entry in either write mode. Under checkpoint-owned delivery the staged marker takes one mutation, so the bound is 99,999, and exceeding it fails the job. The writable timestamp metadata, or the writer clock when it is absent, stamps each entry as it stamps each qualifier (see Cell timestamps). A map contains only the qualifiers it writes, so what happens to the family’s other qualifiers is a choice, made by sink.map-family.update-mode:

• merge (the default): writes the entries and leaves every other qualifier as it is. This is how a Bigtable write behaves and how a ROW family treats undeclared qualifiers. It suits a family that accumulates qualifiers across writes, and it means that no SQL write removes an entry; a delete removes the whole row. An empty map writes nothing, and a row that writes nothing is rejected like a row whose families are all null.
• replace: deletes the family, then writes the entries, in the same row mutation. Bigtable applies a row’s mutations in order and atomically, so the family reads back as the written map. An empty map clears the family. Two writes of one row in the same batch still have no defined winner.

Map families are written in every write mode except conditional, and under checkpoint-owned delivery. Keep-latest deletes each written qualifier’s older versions under merge; under replace the family delete already removes them. The option is rejected in every mode except upsert and keep-latest, and on a table that declares no map family.

The aggregate, append and increment modes turn each entry into one operation on the qualifier its key names, which is how SQL contributes to a qualifier that is data, such as a counter per day or an HLL per campaign:

CREATE TABLE page_activity (
  rowkey STRING,
  visits MAP<STRING, BIGINT>,
  visitors MAP<STRING, BIGINT>,
  bucket TIMESTAMP_LTZ(6) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'table' = 'pages',
  'sink.write-mode' = 'aggregate',
  'sink.aggregate.column-family-types' = 'visits:int64-sum,visitors:int64-hll'
);

-- The day and the campaign are data: each entry adds to the qualifier its key names.
-- A fixed bucket timestamp makes every contribution address the same cell version.
INSERT INTO page_activity
VALUES ('page-1',
        MAP['2026-09-30', CAST(1 AS BIGINT)],
        MAP['spring-sale', CAST(4211 AS BIGINT)],
        TO_TIMESTAMP_LTZ(0, 3));

• aggregate adds one AddToCell per entry. The value type must be TINYINT, SMALLINT, INT or BIGINT, and the family needs a type in sink.aggregate.column-family-types like any other family (see Aggregate contributions).
• append and increment add one read-modify-write rule per entry, after the rules of the families declared before the map. increment requires a BIGINT value and append a CHAR, VARCHAR, BINARY or VARBINARY value, as for a ROW qualifier (see Append and increment).

In all three modes a null value contributes nothing, and a null key fails the record. A row left with no operation, because its families are null and its maps are empty or hold only null values, fails the record, as a row whose cells are all null does. The entry-level limits still apply: an aggregate entry holds at most 100,000 contributions, and a read-modify-write request at most 100,000 rules. A map that takes its row past either limit fails as a serialization failure, which the sink routes like any other; the limits count every family of the row, not the map alone. Under checkpoint-owned delivery the staged marker takes one of an aggregate entry’s mutations, so the bound is 99,999, and exceeding it fails the job. sink.map-family.update-mode stays rejected in these modes, because a contribution and a rule add to a cell rather than replace it, and replace would need a family delete that a read-modify-write request cannot carry. The design record is ADR-0172.

Catalog#

A bigtable catalog resolves tables from the column families each Bigtable table already holds, so a statement reads or writes a table without a CREATE TABLE. One catalog covers one instance, which is the catalog’s one database, and the instance’s tables are the catalog’s tables. The catalog is read-only, like the BigQuery and Spanner catalogs.

CREATE CATALOG bt WITH (
  'type' = 'bigtable',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'key-type' = 'string'
);

USE CATALOG bt;

After USE CATALOG bt, orders names the orders table of my-instance, and so does bt.`my-instance`.orders. SHOW DATABASES prints the instance id, and SHOW TABLES and DESCRIBE list and describe what Bigtable holds.

What a catalog table carries#

Bigtable’s metadata names a table’s column families and each family’s value type, and nothing inside a family: neither its qualifiers nor the types of its cells. The catalog therefore resolves each family as a map column family, which reads every qualifier a row has without the schema naming them, the shape GoogleSQL for Bigtable presents. A table the catalog resolves behaves as the equivalent hand-written CREATE TABLE would, for scans, lookup joins and INSERT INTO alike.

• Row key: a _key column, GoogleSQL’s name for it, declared NOT NULL and as the PRIMARY KEY ... NOT ENFORCED, so the table serves lookup joins and keyed upserts and deletes without a declaration. Its type is BYTES, or STRING under 'key-type' = 'string', described below.

• Column families: one MAP column per family, named after it, in name order. A family name with a character outside a plain SQL identifier, such as - or ., is written in backticks: SELECT `user-events` FROM orders. The key is the qualifier, typed like the row key. The value type follows the family’s value type:

  Family value type	Column type with the default key type
  None, a raw family	MAP<BYTES, BYTES>
  int64-sum, int64-min, int64-max aggregate	MAP<BYTES, BIGINT>
  int64-hll aggregate	MAP<BYTES, BYTES>, the sketch’s bytes
  Any other	MAP<BYTES, BYTES>

  BIGINT reads an aggregate family’s state, which Bigtable stores as eight big-endian bytes, the layout the type mapping gives BIGINT. A family whose type the catalog does not recognize, such as an integer in another encoding, reads as BYTES, which returns the stored bytes and decodes nothing, so it cannot misread a cell; every catalog map value is nullable, so an empty cell reads as NULL, the null convention for a nullable value.

• Options: connector, project, instance and table, and the catalog’s service-account-key-file and emulator-endpoint when those are set. Every other option is per statement, through an OPTIONS hint: SELECT * FROM orders /*+ OPTIONS('scan.app-profile-id' = 'analytics') */.

A table with a column family named _key fails the lookup, naming the table, because the family and the row key would need the same column name; declare such a table by hand.

Contributions to an aggregate family go through an aggregate input table, as Aggregate contributions describes; the catalog table reads the state they leave.

Choosing the key type#

With the default 'key-type' = 'bytes', _key and every qualifier keep their stored bytes, which suits any key, including hashed or binary ones. Flink does not convert a string literal to BYTES implicitly, so a predicate or an entry access spells the bytes: _key = CAST('user1' AS BYTES), _key = x'7573657231' or profile[CAST('name' AS BYTES)]. The planner folds the cast to a literal, so the first form narrows the scan to the row as the second does.

'key-type' = 'string' reads _key and every qualifier as UTF-8 without validating them, which suits row keys written as human-readable strings, as Bigtable’s schema design guidance recommends. A query then compares them with string literals, and GoogleSQL for Bigtable’s own example query runs unchanged:

SELECT address['street'], address['city'] FROM myTable WHERE _key = 'user1';

The key type leaves the map values’ types as the family’s value type sets them, and CAST(address['city'] AS STRING) reads a BYTES value as text. A key or qualifier that is not valid UTF-8 keeps its bytes through a STRING column, so copying a table through the catalog with INSERT INTO ... SELECT writes the same keys back; only its printed form is unreadable. GoogleSQL’s functions, such as STARTS_WITH, are not Flink’s, and a LIKE on _key is evaluated by Flink after a full scan rather than pushed to Bigtable; a range such as _key >= 'user' AND _key < 'uses' or a scan.row-prefix hint narrows the scan instead.

What the catalog leaves out#

The catalog resolves no Change Streams table: a Bigtable table with a change stream is listed and resolved as an ordinary table, and reading its change stream takes a hand-declared table. A Change Streams table’s schema is either the fixed envelope, which an OPTIONS hint cannot give a catalog table, or the producer’s selected cell, which the catalog cannot know, so it is declared by hand. CREATE, DROP and ALTER of tables through the catalog are refused; Bigtable itself and the sink’s sink.create-disposition create tables. Table statistics are unknown to the planner.

CREATE CATALOG, USE CATALOG and SHOW DATABASES make no request and need no credentials, so a mistyped instance id shows on the first SHOW TABLES or query that names a table: the table lookup finds nothing, Flink lists the tables while validating the query, and that failure names the instance with Bigtable’s NOT_FOUND. DESCRIBE does not list, so it reports the table as not existing instead. SHOW TABLES lists the instance’s tables, which needs bigtable.tables.list, and a table lookup reads its column families, which needs bigtable.tables.get. The planner looks a table up each time a statement names it, and possibly more than once per statement.

Catalog options#

A tuning option in CREATE CATALOG is rejected as unknown: a catalog-level default would apply to statements that never mention it. key-type is the exception that belongs to the catalog, because it decides the schema, which a hint cannot change. The other keys are those of the table options they are carried as, and project and instance are checked as resource-path components. The design record is ADR-0178.

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.) A MAP family reads as Map column families describes.

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, and the value type of every map family, 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. A MAP family written with sink.map-family.update-mode = replace needs no column deletes, because the family delete ahead of its cells already removed every version. 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 of a ROW family, or of a merged MAP family, consumes two mutations, and a replaced MAP family consumes one per entry plus its family delete; 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, and a map family becomes one rule per entry, keyed by the entry’s key, with its value type checked as a cell type is. NULL families, cells and map values 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 a watermark on every table the statement scans, though not on a lookup join’s dimension table. 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.

All of this is streaming behaviour. In batch execution Flink 2.3 plans a plain insert, VALUES included, with no clause, and accepts an ON CONFLICT clause without watermarks but drops it: the plan carries no materializer, so DO NOTHING keeps no first row and DO ERROR reports no conflict (measured on 2.3.0).

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. A MAP family written with replace is the same: each entry replaces the family atomically, so the family reads back as one of the two maps, never a mixture, but not necessarily the later one.

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, or an empty map a merge writes nothing for, 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.

A MAP family’s null key fails the record too, since a cell has no null qualifier.

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. A join without row-key equality is rejected during planning.

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. An additional equality key may name a whole physical family ROW whose children are scalar. Only the row key addresses the point read; the connector compares every field of the converted family row before returning it, independently of Flink’s internal row representation. Binary children compare by content, and two NULL children compare equal; a NULL whole-family key or result joins no row. MAP families, nested lookup key paths, and metadata keys are rejected during planning. Whole-family equality keys support NONE and PARTIAL, in sync and async mode; FULL rejects them because its standard row-key index does not normalize nested row representations. In this example, events.expected_profile is ROW<name STRING, email STRING>, matching the profiles.profile family.

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 AND e.expected_profile = p.profile;

Scalar qualifier conditions such as p.profile.name = 'Alice' remain Flink residual predicates. Flink does not pass them through SupportsFilterPushDown for a temporal join, and NONE, PARTIAL and row-key-only FULL evaluate the same residual. 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. The cache stores the filtered result under the complete tuple of row-key and family equality values; a failed equality follows the configured missing-key cache policy. 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.

A key-only DELETE carries NULL in every non-key physical column, including one the DDL declares NOT NULL. On Flink 2.x the planner hands such a delete on as it is to a sink that accepts key-only deletes, unless the query needs a ChangelogNormalize: reading a metadata column, filtering on a non-key column and aggregating each do. That operator, which Flink 1.20 always keeps, completes each delete from the last row it holds for the key and drops the delete of a key the job never saw. On Flink 2.2 and 2.3, two Flink bugs fixed in 2.4.0 affect a delete handed on as it is: a projection evaluates its expressions over the null columns, which can fail the job when this table declares one of them NOT NULL (FLINK-40528), and Flink’s not-null enforcer fails the job when the sink table declares one of them NOT NULL (FLINK-40477). Declare the non-key columns nullable, in this table and in the sink table.

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<...> or MAP<...> 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.

Map column families are this connector’s addition to that model (ADR-0172). They follow GoogleSQL for Bigtable, whose map-per-family shape Bigtable’s metadata alone can describe, and they are what the catalog derives. A DDL using one does not move to the HBase connector.

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. It reads the state an aggregate write leaves through a MAP family DDL, a ROW family DDL and GoogleSQL for Bigtable and requires all three to agree, which the emulator cannot show because it has no aggregate families. 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.