Spanner SQL connector#

The spanner connector reads bounded Table API and SQL scans, emits unbounded Change Streams changelogs, serves primary-key lookup joins, and writes rows through flink-connector-gcp-spanner. It maps onto the DataStream source and sink, so partitioning, snapshot, batching, retry, delivery, metrics, and failure behavior remain the same. A read-only catalog resolves tables from each database’s INFORMATION_SCHEMA, so a statement can use them without a CREATE TABLE.

CREATE TABLE orders (
  order_id BIGINT,
  customer STRING,
  total DECIMAL(38, 9),
  updated_at TIMESTAMP_LTZ(9),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'spanner',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'database' = 'orders-db',
  'table' = 'orders'
);

INSERT INTO orders SELECT order_id, customer, total, updated_at FROM staged_orders;

SELECT customer, total FROM orders;

Use flink-sql-connector-gcp-spanner, the relocated SQL uber-jar, for SQL deployments. Place flink-sql-connector-gcp-spanner-<version>.jar in Flink’s lib/ before starting the cluster, or load it for one SQL Client session:

ADD JAR '/path/to/flink-sql-connector-gcp-spanner-1.2.0.jar';

The artifact bundles the connector and its runtime dependencies while leaving Flink APIs provided by the cluster. Its bundled dependency implementations are relocated so it can coexist with other connector jars and with application dependencies. DataStream applications should depend on flink-connector-gcp-spanner instead of the SQL uber-jar.

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.

Lineage#

A bounded scan and a Table sink report their configured physical table in the gcp facet of one logical SQL dataset. The SQL catalog identifier remains Flink-owned. Index/filter pushdown and deferred scan resolution retain the same table identity without running the resolver during extraction. A Table CDC source reports its configured table and the configured Change Stream as provenance in that facet; it does not enumerate every table the stream watches. Lookup joins are outside this extraction path.

Physical tables use namespace spanner://{project}:{instance} and name {database}.{schema}.{table}, omitting an absent schema segment. The existing dialect rules determine schema/table spelling, including PostgreSQL’s public default for an unqualified native table name. A legacy native schema.table value already carries its qualification and does not receive another default schema prefix. The facet retains the original configured schema/table syntax, including quotes and case; an omitted named-schema option leaves schema absent from the identity map even when the canonical name uses a default schema. Extraction changes neither record routing nor source boundedness.

Flink 2.x retains the physical facet during planner extraction. Flink 1.20 supports direct inspection of the runtime Source/Sink metadata, without automatic FLIP-314 listener delivery. A custom listener must consume the gcp facet explicitly; an unmodified OpenLineage listener is not assumed to understand it.

Credentials#

service-account-key-file selects one service-account JSON key for the sink, bounded scan, Change Streams scan, and synchronous or asynchronous lookup paths. When it and emulator-endpoint are absent, every path uses Application Default Credentials. The option stores only the path in the job graph and reads the file in each runtime process that creates a Spanner client. Sink and lookup jobs need the path on applicable TaskManagers, while bounded and Change Streams scans need it on the JobManager and applicable TaskManagers. Mount the same configured path in all of those containers, including after a restart or restore. The option is mutually exclusive with emulator-endpoint; prefer an attached service account or Workload Identity when available. See the DataStream credential deployment contract for the exact process boundaries and failure behavior.

Schema and type mapping#

Named schemas#

Set named-schema when the destination or source table belongs to a named Spanner schema. The connector qualifies that schema with table for mutations, bounded scans, and synchronous and asynchronous lookups. When scan.index is set, the same schema also qualifies the index because Spanner requires a table and its index to share a schema. Leaving named-schema unset retains the empty GoogleSQL schema or PostgreSQL public.

The named-schema, table, and scan.index values each name one identifier component when named-schema is set. Do not put a dot-qualified name in any of these options. GoogleSQL accepts an unquoted identifier or a backtick-quoted identifier, and PostgreSQL accepts an unquoted identifier or a double-quoted identifier. PostgreSQL folds unquoted values to lower case and preserves quoted values, including their case. GoogleSQL compares schema-object names case-insensitively. Quoted values use one canonical spelling: PostgreSQL doubles an embedded double quote, while GoogleSQL escapes an embedded backtick or backslash with a backslash. Numeric, octal, and Unicode escape spellings are not accepted in these options. The connector rejects blank values, multipart values, mismatched quotes, and non-canonical quote escapes before opening a client. It does not duplicate Spanner’s identifier character, length, or keyword rules. The connector removes SQL quoting after decoding each component because Spanner’s native data APIs accept catalog names, not SQL identifier tokens. The bounded scan resolves those decoded names through INFORMATION_SCHEMA, while sink mutations and lookup reads leave final name validation to the Spanner API. Quote a reserved word when it is used as the schema, the first component of the qualified name.

CREATE TABLE sales_orders (
  order_id BIGINT,
  total DECIMAL(38, 9),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'spanner',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'database' = 'orders-db',
  'named-schema' = 'sales',
  'table' = 'orders',
  'scan.index' = 'orders_by_total'
);

Type mapping#

Spanner’s GoogleSQL data types and PostgreSQL data types define the native column and key constraints that this mapping enforces. A catalog table is derived from the live schema instead of declared; the last column gives the Flink type each Spanner column resolves to. JSON (jsonb), PROTO and ENUM columns resolve as their marked STRING, BYTES and BIGINT carriers, as the type markers describe.

The mapping applies to one-dimensional array elements, and nullable Flink values become null Spanner values. Nested arrays are rejected because neither Spanner dialect permits an array whose element is another array. ROW is rejected because Spanner STRUCT cannot be stored in a table column. Plain TIMESTAMP is rejected because it has no time-zone semantics; use TIMESTAMP_LTZ for an instant. GoogleSQL accepts only DECIMAL(38, 9), which exactly matches its fixed NUMERIC precision and scale. PostgreSQL accepts every Flink-supported DECIMAL(p, s), but the physical numeric column is wider than Flink’s maximum precision of 38 and has no DDL-level precision or scale modifier. On reads, a PostgreSQL numeric value must fit the declared Flink precision without reducing non-zero fractional digits; trailing fractional zeros may be removed without changing the value. Precision overflow, scale loss, and PostgreSQL numeric NaN fail conversion instead of rounding or turning a non-null value into null. The error names the physical column and declared Flink shape without including the stored value. FLOAT cannot be a primary-key column in either dialect, and no PostgreSQL decimal can be a primary-key column because PostgreSQL numeric is not a key type.

Catalog#

A spanner catalog resolves tables from the schema each Spanner database already holds, so a statement reads or writes a table without a CREATE TABLE. One catalog covers one instance: its databases are the catalog’s databases, in either dialect, and their base tables are the catalog’s tables. The catalog is read-only, like the BigQuery catalog and the JDBC connector’s.

CREATE CATALOG sp WITH (
  'type' = 'spanner',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'default-database' = 'orders-db'
);

USE CATALOG sp;

After USE CATALOG sp, orders names the orders table in the default schema of orders-db, and `sales.orders` names the orders table in its sales schema. SHOW DATABASES, SHOW TABLES and DESCRIBE list and describe what Spanner holds.

Table names#

A Spanner table has three levels of address below its instance, a database, a schema and the table, while a Flink catalog has two, so a named schema travels inside the table name. A table in the default schema, the empty GoogleSQL schema or PostgreSQL public, is listed by its name alone, and a table in a named schema as schema.table. Each part uses the canonical quoting the named-schema and table options accept. A GoogleSQL part outside the plain identifier grammar is backtick-quoted. A PostgreSQL part is double-quoted when it has an upper-case letter or a character outside the plain grammar, since an unquoted PostgreSQL part folds to lower case. In Flink SQL the whole name is one identifier, so it is written in Flink backticks: `sales.orders`, or `"Sales"."Orders"` for a PostgreSQL table whose schema and table were created with quoted mixed-case names.

Every name SHOW TABLES prints resolves to the table it came from. A GoogleSQL name is compared case-insensitively, as Spanner compares it, and a PostgreSQL name exactly after its quoting is decoded.

What a catalog table carries#

A table the catalog resolves behaves as the equivalent hand-written CREATE TABLE would, for scans, lookup joins and INSERT INTO alike.

• Options: connector, project, instance, database and dialect, the table’s name, and the catalog’s service-account-key-file and emulator-endpoint when those are set. A default-schema table carries table alone, holding the native name; a named-schema table carries named-schema and table in canonical quoting. The markers its columns need are set from the column metadata: the type markers json-field-paths, uuid-field-paths, proto-type-names and enum-type-names, and generated-columns, which is a marker but not a type marker. Every other option is per statement, through an OPTIONS hint, which resolves an index in the table’s own schema: SELECT * FROM `sales.orders` /*+ OPTIONS('scan.index' = 'orders_by_total') */.
• Schema: each column resolves as the last column of the type mapping gives, and a column Spanner declares NOT NULL resolves as NOT NULL. Spanner names a PROTO or ENUM column’s type as PROTO<pkg.Message> or ENUM<pkg.Enum>; the emulator names only the fully qualified type, so there the catalog reads the database’s proto bundle to tell a message from an enum. A PostgreSQL numeric resolves as DECIMAL(38, 9), so a stored value with more fractional or integer digits than that fails the read, naming the column, rather than rounding.
• Primary key: the Spanner primary key becomes a PRIMARY KEY ... NOT ENFORCED in key order, over NOT NULL columns, so the table serves lookup joins and keyed upserts and deletes without a declaration. A GoogleSQL key column that Spanner allows to hold NULL is declared NOT NULL all the same, because the planner requires it of a key; a row whose key is NULL breaks that declaration. A key whose type the connector cannot key on, such as FLOAT32, PostgreSQL numeric or an ENUM, resolves, and a statement over the table is refused when it is planned, as for a hand-written table with that key.
• Stored generated columns resolve as ordinary columns listed in generated-columns: reads return them, and the sink leaves them out of every mutation, since Spanner refuses a value for a generated column. A generated column that is not stored is left out, because Spanner’s read API, which scans and lookups go through, refuses to return one; the DataStream source reading through a SQL query can select it. A generated column outside the key is nullable in the Flink schema, so an INSERT INTO that names the other columns need not supply it. A generated key column is NOT NULL like any key column, so an INSERT INTO must supply it, and the value supplied must be the one Spanner computes, as the sink explains.
• Hidden columns, such as the TOKENLIST columns a search index reads, are left out, as Spanner’s own SELECT * leaves them out; a hidden key column would be kept, since the key names it.

A column whose Spanner type the mapping has no row for, such as INTERVAL, fails the lookup, naming the table, the column and its Spanner type; the table never resolves with the column missing.

Change streams#

Change streams are not listed. An OPTIONS hint turns a catalog table into its change-stream source instead: SELECT * FROM `sales.orders` /*+ OPTIONS('scan.mode' = 'change-stream', 'scan.change-stream.name' = 'order_changes', 'scan.change-stream.changelog-mode' = 'upsert') */. Spanner change streams do not watch a generated column outside the primary key, so a table with one refuses the hint when the statement is planned. Such a table’s change stream is read through a hand-written table that leaves those columns out, as Change Streams scan behavior describes. On Flink 2.2 and 2.3 the hint’s key-only deletes can meet two Flink bugs (FLINK-40477, FLINK-40528) when a non-key column is NOT NULL, which a catalog table keeps from Spanner; Change Streams scan behavior describes them and the hand-written table that avoids them.

What the catalog leaves out#

Views are not listed: the table source reads through Spanner’s read API, which takes a table. CREATE, DROP and ALTER of tables and databases through the catalog are refused; Spanner’s own DDL creates them. Table statistics are unknown to the planner.

CREATE CATALOG and USE CATALOG make no request and need no credentials. SHOW DATABASES and a database’s dialect come from the database admin API, which needs the spanner.databases.list and spanner.databases.get permissions respectively. Where a column’s type names a proto type without saying whether it is a message or an enum, as on the emulator, the catalog also reads the database’s proto bundle, which needs spanner.databases.getDdl. Tables and columns come from INFORMATION_SCHEMA, read with the catalog’s credentials. The catalog asks a database’s dialect once, and asks again only after finding the database dropped; the planner looks a table’s columns 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. The keys are those of the table options they are carried as. project, instance and default-database are checked as resource-path components, and default-database also against Spanner’s database-id grammar, since a name outside it names no database.

Source#

Scan behavior#

The source is bounded and reads the table through Spanner partitionRead at one shared snapshot. Top-level projection is pushed into the requested Spanner column list, so unused columns do not cross the network. Nested projection is not advertised. If the planner requests no physical column, the connector reads the first declared column as a carrier but emits zero-field rows in the requested shape.

The source translates =, <, <=, >, >=, and conjunctions over consecutive key columns into Spanner KeySet points and lexicographic ranges. A complete primary-key equality is an exact point read. A leading equality prefix, optionally followed by a range on the next primary-key column, is exact and is removed from Flink’s residual filter. For DOUBLE keys (FLOAT64 / PostgreSQL FLOAT8), equality and ordered comparisons accept finite literals. The literal must reach the connector as DOUBLE; a DECIMAL literal that Flink leaves unconverted remains residual. The translator also handles infinite DOUBLE literals. Both infinities are valid key values; a range includes or excludes them according to its comparison. Flink leaves CAST('Infinity' AS DOUBLE) and CAST('-Infinity' AS DOUBLE) as expressions, so comparisons with these SQL string casts remain residual. Both signed zeros are equal, including at open and closed range boundaries. Comparisons with a NaN literal remain with Flink. Spanner rejects NaN in FLOAT64 / PostgreSQL FLOAT8 key columns; the emulator accepts these keys, as recorded in the emulator deviations. The PostgreSQL type contract also states this restriction explicitly. Predicates on later key columns, OR, IN, <>, computed expressions, casts that survive planning, field-to-field comparisons, and null literals remain with Flink. A literal the connector cannot convert to its key column’s Spanner type, such as a non-integral value compared with an INT64 column or text that is not a canonical UUID, also remains with Flink instead of failing the job during planning.

Set scan.index to read a bounded scan through a named secondary index. The connector resolves the live index key order, sort direction, state, null filtering, and readable columns from the GoogleSQL or PostgreSQL INFORMATION_SCHEMA at the batch transaction’s exact snapshot. It then uses matching predicates as a best-effort index-key prefilter and leaves every such predicate with Flink to preserve SQL semantics. EXPLAIN lists parsed filter candidates under the source’s filter as well as in the residual calculation. Only candidates usable with the live index key order narrow the read. Reporting these candidates keeps differently filtered scans distinct during Flink source reuse. The selected index must be READ_WRITE, belong to the configured table and schema, cover every column the scan reads, and be safe for nullable key rows. GoogleSQL STORING and PostgreSQL INCLUDE columns are readable through the index together with its key and the base-table primary key. A null-filtered index is accepted only when the pushed filters prove every nullable index key is not null. An unusable configured index fails the job during partition planning instead of falling back to the base table. The connector resolves schema, table, and index metadata together at the scan snapshot. A missing or invisible schema is reported separately from a missing or invisible table access path.

Partition count and size are service hints, not exact split controls. The default timestamp bound is strong; set either a read timestamp or exact staleness, never both. There are deliberately no column-range partition options because Spanner chooses partition boundaries from physical storage.

Sink#

Mutation behavior#

See Write and key-collision semantics for the cross-connector distinction between a planner key and a destination write operation.

A declared PRIMARY KEY makes the sink an upsert sink. INSERT and UPDATE_AFTER rows use Spanner insertOrUpdate, and DELETE rows use a key built in the declared column order. The key may be composite, but every member must map to a Spanner key type.

On Flink 2.x a delete may carry the key alone, which is all the sink reads from it. On Flink 2.2 and 2.3, two Flink bugs fixed in 2.4.0 affect such a delete, whose non-key columns are null. Flink’s not-null enforcer checks those columns 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. A catalog table carries Spanner’s NOT NULL over to Flink, so write such an input through a hand-written table that declares those columns nullable.

A field listed in generated-columns is left out of every mutation, because Spanner refuses a value for a generated column. An upsert lets Spanner compute a generated key column from the columns written, but a delete still names the row by the value the row carries, and the planner keys its upserts by it, so that value must equal the one Spanner computes. Rows read from the same table carry it; a value a statement invents does not, and a delete keyed by it removes nothing.

Without a primary key, the sink accepts insert-only input and uses Spanner insert. This preserves duplicate-key errors instead of pretending that an unknown physical key can support upserts or deletes. As with other Flink connectors, PRIMARY KEY ... NOT ENFORCED describes the contract to the planner; the connector does not verify uniqueness.

Flink 2.3 may demand ON CONFLICT#

Flink 2.3 (FLIP-558) refuses to plan an updating statement into a table with a PRIMARY KEY unless the PRIMARY KEY contains an upsert key the planner infers for the query, or the statement carries one of the ON CONFLICT DO DEDUPLICATE, DO ERROR or DO NOTHING clauses that Flink 2.3 introduces. A query keyed by other columns, such as an aggregation written into a table keyed by its result, is refused, and so is one whose upsert key the planner cannot infer. The refusal is the planner’s, not this connector’s: rows with different upsert keys that map to one primary key reach the sink in no defined order, and Flink 2.3 asks the statement to say what that should mean instead of materializing silently, as Flink 2.2 does. An aggregation grouped by the key columns plans with no clause, and so does an insert-only statement, INSERT INTO .. VALUES included, because the sink answers an insert-only input with an insert-only changelog. ON CONFLICT DO DEDUPLICATE, or the planner-wide table.exec.sink.require-on-conflict = false, plans a refused statement with the materialization Flink 2.2 uses; DO ERROR and DO NOTHING plan it only when every table the statement scans declares a watermark, while a lookup join’s dimension table needs none (measured on 2.3.0). Flink 1.20 and 2.2 do not parse the clause, and ignore the setting. All of this is streaming behaviour: in batch execution Flink 2.3 plans such a statement with no clause, and accepts an ON CONFLICT clause without watermarks but drops it, so the rows reach the sink with no materializer and the clause has no effect (measured on 2.3.0).

Lookup behavior#

The source supports temporal lookup joins when the equality key contains every column of the declared PRIMARY KEY. Composite keys are encoded in the DDL declaration order even when the planner supplies the predicates in another order. Additional equality keys on top-level physical scalar columns are supported, including constants in ON or WHERE and comparisons with an input column. The complete primary key alone addresses the point read; the connector compares the additional keys against the converted Flink row before returning it. Binary values compare by content, and schema-marked values compare in their Flink carrier type: UUID and JSON strings, PROTO bytes, and ENUM integers. Nested key paths, metadata columns, and additional ARRAY, MAP, or ROW keys are rejected during planning. A null lookup key, an absent Spanner row, or a failed additional equality produces no joined row. Exact pushed primary-key predicates also gate synchronous and asynchronous lookups before an RPC, while Flink evaluates any residual predicate normally. scan.index does not change lookup keys or lookup access paths. The following example uses an order_events input with event_id, order_id, and processing-time proc_time columns, and the orders table declared above.

SELECT e.event_id, o.customer
FROM order_events AS e
LEFT JOIN orders FOR SYSTEM_TIME AS OF e.proc_time AS o
  ON e.order_id = o.order_id AND o.customer = 'Alice';

lookup.async chooses Spanner’s synchronous readRow or asynchronous readRowAsync API. Flink’s standard lookup.cache = NONE and PARTIAL modes are supported; FULL is rejected because it would require a scan-backed cache with different snapshot and refresh semantics. The standard partial-cache expiry, size, and missing-key options apply unchanged. The cache stores the filtered result under the complete lookup-key tuple, so different additional values for the same primary key have separate entries. An additional equality that fails follows the configured missing-key cache policy. lookup.max-retries retries only ABORTED, DEADLINE_EXCEEDED, and UNAVAILABLE point-read failures and counts retries after the initial request. Of those, only UNAVAILABLE is also retried by the Spanner client, so this option is what buys a second attempt at the other two. RESOURCE_EXHAUSTED is not retried here because the client already retries it whenever the server asks for a delay, and waits that delay; re-issuing the read at once would spend the budget against the wait rather than observe it.

UUID, JSON, protocol buffers, and enums share carrier types with ordinary columns, so the DDL marks them explicitly:

WITH (
  'uuid-field-paths' = 'id;related_ids',
  'json-field-paths' = 'metadata;payloads',
  'proto-type-names' = 'event:example.events.Event',
  'enum-type-names' = 'status:example.events.Status'
)

UUID and JSON fields use STRING, PROTO fields use BYTES, and ENUM fields use BIGINT in the Flink schema. The UUID marker maps STRING and ARRAY<STRING> carriers to native UUID columns in both dialects, including primary-key columns and composite lookup keys. UUID input must use the complete 36-character 8-4-4-4-12 hexadecimal form, with either letter case; Java’s shortened accepted forms are rejected. Reads always emit lowercase canonical UUID strings, and nullable UUID scalars, arrays, and array elements remain null. Malformed sink or lookup input fails conversion with the physical column name but without including the input value. Set dialect = 'POSTGRESQL' for PostgreSQL-dialect databases; JSON markers then produce jsonb values. PROTO and ENUM markers require GOOGLE_STANDARD_SQL and are rejected for PostgreSQL databases. A marker names one top-level physical field or an entire array field. Every marker must resolve to exactly one physical field and no field may have more than one type marker; generated-columns is not a type marker and combines with them.

The connector does not inspect or migrate the live Spanner schema. Changing an existing column from STRING to UUID therefore requires coordinating the Spanner DDL and this option before redeploying the Flink job, and every stored string must already be valid canonical UUID input before migration.

Change Streams scan behavior#

Set scan.mode = 'change-stream' to replace the bounded scan with an unbounded CDC source over one Spanner Change Stream. The default remains bounded, so existing DDL keeps its snapshot behavior. Change-stream mode maps the declared named-schema and table to the exact dialect-aware native table name and silently advances past records for other watched tables. It does not support lookup joins, projection or filter pushdown, bounded-scan fetch, partition, and timestamp options, or lookup options. A change-stream table is source-only. Its declared columns are the watched table’s own, so an INSERT INTO naming one would otherwise write into the very table being watched; it is rejected when the statement is planned instead. A table written to may still carry the bounded-scan and lookup options it is also read with, but a scan.change-stream.* option on it is rejected rather than ignored.

scan.change-stream.changelog-mode is required. full accepts only records captured with NEW_ROW_AND_OLD_VALUES and emits INSERT, adjacent UPDATE_BEFORE and UPDATE_AFTER, and full DELETE rows. For an update, the complete new row is copied and each reported old value replaces its new value to reconstruct the before row. upsert accepts NEW_ROW and NEW_ROW_AND_OLD_VALUES, requires a declared primary key, and emits INSERT, UPDATE_AFTER, and key-only DELETE rows. 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; a catalog table keeps Spanner’s NOT NULL, so read the stream through a hand-written table instead.

CREATE TABLE order_changes (
  order_id BIGINT,
  customer STRING,
  status STRING,
  commit_timestamp TIMESTAMP_LTZ(3) METADATA FROM 'commit-timestamp' VIRTUAL,
  record_sequence STRING METADATA FROM 'sequence' VIRTUAL,
  server_transaction_id STRING METADATA FROM 'server-transaction-id' VIRTUAL,
  mod_number INT METADATA FROM 'mod-number' VIRTUAL,
  WATERMARK FOR commit_timestamp AS SOURCE_WATERMARK(),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'spanner',
  'project' = 'my-project',
  'instance' = 'my-instance',
  'database' = 'orders-db',
  'table' = 'orders',
  'scan.mode' = 'change-stream',
  'scan.change-stream.name' = 'order_changes',
  'scan.change-stream.changelog-mode' = 'upsert',
  'scan.startup.mode' = 'latest'
);

The source exposes the stable scalar identity, transaction, and record fields listed in the Spanner metadata reference. The vocabulary follows Debezium’s Spanner source metadata where it represents the same Spanner field, with hyphenated keys matching this connector’s SQL conventions. The native commit-timestamp metadata type is non-null TIMESTAMP_LTZ(9), preserving Spanner’s nanosecond precision. Flink permits watermark columns only through precision 3, so the example declares that metadata column as TIMESTAMP_LTZ(3) and lets the planner apply the compatible precision cast. Omit the watermark declaration and use TIMESTAMP_LTZ(9) when downstream SQL must retain the full commit timestamp. SOURCE_WATERMARK() uses the Change Streams source’s existing commit-timestamped records and coordinator-owned unfinished-ledger heartbeat frontier; it does not introduce a second clock or a second out-of-orderness policy.

mod-number is the zero-based position of the mod in its original Spanner data-change record. The adjacent before and after rows produced for one full-mode update carry the same mod-number, so metadata never splits the logical identity of that mod.

Each data-change record is validated against the DDL’s physical names, native Spanner types, and, in upsert mode, primary-key column membership before any row from that record is emitted. Extra watched columns are ignored, while a missing declared column, a type mismatch, an incompatible value-capture mode, or an absent required row value fails deserialization. Spanner change streams do not watch a generated column outside the primary key, so a change-stream table that lists a column outside its declared PRIMARY KEY in generated-columns is refused when it is planned; one declared without the marker fails on the first record, as any missing declared column does, with a message saying the record omits a declared column. Explicit JSON null becomes SQL null, but an absent JSON member is never substituted with null when a complete row is required. A conversion failure identifies the table, commit timestamp, transaction, record sequence, and mod index without including row JSON or credential paths.

Startup, expired-restore fallback, retention fallback, heartbeat interval, RPC priority, per-subtask query concurrency, emulator endpoint, credential path, and source parallelism map to the DataStream Change Streams builder. The checkpoint, retention, delivery, and capacity contracts therefore remain those of the DataStream Change Streams source.

Options#

The sink options map directly onto the DataStream writer options; their validation limits are listed in the Spanner reference.

Former option keys#

Releases up to 1.1.0 named five of these options with a schema prefix. A catalog that stores a table as properties, such as Flink’s Hive catalog for a generic table, loses an option under such a key without an error. Flink rebuilds the table’s options from those properties and drops every key it takes for part of the serialized table schema: every key starting with schema., and from Flink 2.2 a bare schema too. The options therefore have keys without the prefix, and each former key is still accepted as a deprecated key:

A former key keeps working in a WITH clause or an OPTIONS hint, and Flink logs a warning naming the current key when it reads one. A table stored in such a catalog under a former key that Flink drops is read back without the option: without the named schema, writes and scans address the same-named table in the default schema, and without a marker, the column is written and read as its plain carrier type. Recreate such a table with the current keys. Setting one option under both its current and its former key is refused when a statement is planned, because Flink would read the current key and ignore the other. An OPTIONS hint counts, because Flink merges it into the table’s options: a hint overrides an option only under the key the table already sets, which for a spanner catalog table is the current key. generated-columns arrived after 1.1.0 and has no former key, so schema.generated-columns is an unsupported option. The spanner catalog sets the current keys, and the catalogs that keep a table’s options as written, such as Flink’s in-memory catalog, lose no key.

Delivery guarantee#

Both source modes are at least once. The bounded source can replay one partition after recovery or when a reader is interrupted mid-partition, while the Change Streams source can replay records at the inclusive checkpoint timestamp. The sink is at-least-once. Replaying one upsert or delete is idempotent, while insert-only writes can fail on a replayed primary key. The writer submits records as separate BatchWrite mutation groups, whose application order is not guaranteed, so successive updates to the same key do not provide a latest-input-value guarantee.

Design decisions and testing#

The Table connector composes the bounded source, Change Streams source, and sink implementations documented on the DataStream connector page; its Table-only lookup behavior is documented above. That page records the underlying design decisions, emulator deviations, and gated real-GCP coverage.

The catalog derives tables and never creates them. It follows the read-only shape ADR-0168 records for every connector catalog, and ADR-0176 records the Spanner mapping: the SPANNER_TYPE grammar of both dialects, the schema.table naming, and the answers for generated, hidden, PROTO and ENUM columns, views and change streams. The catalog’s unit tests drive it against an in-memory instance, including every listed name resolving back to its table in both dialects. SpannerCatalogITCase then runs it through the planner against the emulator in both dialects: listing, the type mapping, a write Spanner completes with a generated column, a lookup join and an index hint on a named schema, a change-stream hint, and PROTO and ENUM columns.