Firestore SQL connector#

The firestore connector reads and writes Table API and SQL rows in one collection of a Firestore database in Native mode, or reads every collection with one id, one document per row, through flink-connector-gcp-firestore. It maps onto the DataStream source and sink, so snapshot, paging, recovery, throttling, batching, retry, delivery, metrics and failure behavior remain the same; a lookup join reads through its own client, with the retries its section describes. A table with a PRIMARY KEY writes its rows under their key as the document id; a table without one creates a new document for every row.

CREATE TABLE orders (
  order_id STRING NOT NULL,
  customer STRING,
  total DOUBLE,
  updated_at TIMESTAMP_LTZ(6),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'firestore',
  'project' = 'my-project',
  'collection' = 'orders'
);

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

A scan reads the collection as a bounded snapshot, and a lookup join reads one document of it by id. A database in Datastore mode uses the datastore connector instead.

Use flink-sql-connector-gcp-firestore, the relocated SQL uber-jar, for SQL deployments, and place it in Flink’s lib/ before starting the cluster. It is on Maven Central and attached to the GitHub releases, in the two version lines the Quickstart index describes. The artifact bundles the connector and its runtime dependencies, for both Native mode and Datastore mode, 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-firestore instead of the SQL uber-jar.

Lineage#

The Table source and sink report their configured collection in the gcp facet of one logical SQL dataset, named by the table’s catalog identifier. A scan under scan.collection-group reports the group instead, as a firestore-collection-group resource named by the collection id. The collection is a firestore-collection resource: namespace firestore://{project}/{database}, with the default database spelled (default), and the collection path relative to the database as its name, such as orders or users/alice/audit. It names the documents directly in that collection, not those of its subcollections. Lookup joins are outside this extraction path and report no resource. See Lineage for the class loader configuration and what Flink 1.20 and 2.x each deliver.

Credentials#

service-account-key-file selects one service-account JSON key for the scan, the lookup and the sink. When it and emulator-endpoint are absent, all three use Application Default Credentials. The option stores only the path in the job graph: a scan reads the file on the JobManager, where it plans the read, and on each TaskManager that reads a split, while each lookup and sink subtask reads it on its TaskManager, so mount the same path in every container. The option is mutually exclusive with emulator-endpoint, because the emulator channel carries no credentials. See the DataStream credentials for the FIRESTORE_EMULATOR_HOST variable, which the client library reads on its own.

Schema and type mapping#

A table names one collection with collection: a collection id such as orders, or a path that alternates collection and document ids and ends with a collection id, such as users/alice/audit. A path with an even number of segments names a document and is refused. Each row is one document of that collection (or, for a collection-group scan, of any collection with its id), and each column other than the PRIMARY KEY is one top-level field, named exactly as the column is: a column named a.b is a field whose name contains a dot. A column or ROW field whose name starts and ends with __ is refused when the statement is planned, because Firestore reserves such names and would refuse every document.

The PRIMARY KEY is the document id. It must be one STRING column, declared PRIMARY KEY (...) NOT ENFORCED, and it is never stored as a field. A row whose key is empty or contains / fails the job, because the path it would compose names another document, or a document in another collection. A table without a PRIMARY KEY writes every row as a new document under an id the connector draws, 20 letters and digits, as the client library does for an added document. The write is a create, so a row never replaces an existing document: if the drawn id names one, Firestore refuses the create, the document keeps its fields, and the sink creates the row again under a new id. The client library’s retry of a create that was applied but whose answer was lost is refused the same way, and the sink cannot tell the two apart without reading the document, so such a retry stores the row twice.

Type mapping#

Every other Flink type is refused when the statement is planned, naming the field: write an INT as a BIGINT and a FLOAT as a DOUBLE. DECIMAL has no Firestore counterpart and is not mapped. A SQL NULL is written as a Firestore null value, so the field exists and holds null.

A geographical point and a reference share their Flink types with ordinary values, so the table names them with two markers, each a semicolon-separated list of field paths. A path names a column, then a ROW field by .name; an ARRAY is passed through, so a path through an array names its elements, and a MAP’s values are .value. A marker that names a path the table does not declare, a field of the wrong type, the PRIMARY KEY, or a path two fields share (a column named a.b beside a ROW a with a field b), is refused when the statement is planned.

CREATE TABLE stores (
  store_id STRING NOT NULL,
  name STRING,
  location ROW<latitude DOUBLE, longitude DOUBLE>,
  manager STRING,
  departments ARRAY<ROW<name STRING, head STRING>>,
  PRIMARY KEY (store_id) NOT ENFORCED
) WITH (
  'connector' = 'firestore',
  'project' = 'my-project',
  'database' = 'retail',
  'collection' = 'stores',
  'geo-point-field-paths' = 'location',
  'reference-field-paths' = 'manager;departments.head'
);

A reference’s value is a document path relative to the same database, such as staff/alice; the connector stores it as a reference to that document in the database the table writes to, and a value that is not a document path fails the job. A point’s latitude and longitude must both be set and within range, or the row fails the job.

Reading values back#

A scan reads each column from the top-level field of its name, and the PRIMARY KEY column from the document id. A field the document does not have reads as NULL, as does a field that holds null. An integer reads into a DOUBLE column only when the DOUBLE represents it exactly, up to 2^53 in magnitude; a 32-bit BSON integer reads only into a BIGINT, and a table sink writes it back as a 64-bit integer. A timestamp reads at the column’s precision, truncated. A string or bytes read whole, whatever length a CHAR, VARCHAR, BINARY or VARBINARY column declares: the scan neither checks the length nor pads a fixed-length value, so declare STRING and BYTES for values of any length. A reference reads as its document path into a column reference-field-paths names, and a geographical point into a ROW geo-point-field-paths names.

A collection has no schema, so a document can hold, under a column’s name, a value of another type: a string where the column is BIGINT, a reference where the column is an unmarked STRING, a reference into another database, or a value outside the mapping such as bytes of a BSON subtype, a vector or one of the BSON value types an Enterprise-edition database stores. type-mismatch-policy decides what such a value does. Under fail the read fails, naming the document and the field’s path in the schema, with a map’s values as .value. Under null the innermost nullable field around the value reads as NULL instead; a NOT NULL field cannot, so the nearest nullable field around it does, which also covers a NOT NULL nested field the document lacks. A NOT NULL column has no nullable field around it, so a mismatched or missing value there fails the read under either policy.

Source#

A scan reads the documents directly in the table’s collection, as one query on one split, at one read time. With scan.collection-group it reads every collection whose id is the last segment of collection, at any depth of the database, as a collection-group scan that the service partitions; scan.partition.max-partitions caps the partitions it asks for. Both read a consistent snapshot at scan.read-time (truncated to the microsecond), or at the service’s time when the read is planned.

CREATE TABLE order_snapshot (
  order_id STRING NOT NULL,
  customer STRING,
  total DOUBLE,
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'firestore',
  'project' = 'my-project',
  'collection' = 'orders',
  'scan.read-time' = '2026-10-04T00:00:00Z'
);

SELECT customer, SUM(total) FROM order_snapshot GROUP BY customer;

A scan reads only the fields of the columns it produces: the declared columns, or the projected ones when the planner pushes a projection down, each named as one literal field, so a document’s undeclared top-level fields never travel; a ROW or MAP column’s field is read whole, with any nested keys the ROW does not declare. Filters are not pushed down.

A collection-group scan reads documents of many collections, whose ids repeat across them, so its table cannot declare a PRIMARY KEY: the planner trusts a declared key to be unique and removes a GROUP BY or a DISTINCT over it, so repeated ids would come back unaggregated. The document-path metadata column, which is unique, identifies a document instead.

A scan or a lookup can read four metadata columns:

CREATE TABLE all_audit_events (
  action STRING,
  path STRING METADATA FROM 'document-path' VIRTUAL,
  updated TIMESTAMP_LTZ(6) METADATA FROM 'update-time' VIRTUAL
) WITH (
  'connector' = 'firestore',
  'project' = 'my-project',
  'collection' = 'audit',
  'scan.collection-group' = 'true',
  'type-mismatch-policy' = 'null'
);

SELECT path, action, updated FROM all_audit_events;

The DataStream source describes the snapshot’s time window, paging and recovery.

Lookup#

A lookup join reads one document of the table’s collection by the PRIMARY KEY column, its document id. It also accepts equality keys on other top-level physical scalar columns, whether a constant such as c.tier = 'gold' appears in ON or WHERE, or an input column supplies the comparison value. Only the document id addresses the read; the additional keys compare against the converted Flink row before it is returned, with binary values compared by content. Nested key paths, metadata columns, and additional ARRAY, MAP, or ROW keys are rejected during planning. A NULL lookup key or a failed additional equality joins no row. A condition that is not an equality, such as c.tier <> 'gold', stays a filter applied after the lookup. A table without a PRIMARY KEY, which a collection-group table always is, cannot be looked up. A lookup that reaches the service reads the document as it is at that moment, and under a PARTIAL cache a hit returns the row as it was cached until it expires; scan.read-time and the other scan.* options do not apply to it.

CREATE TABLE customers (
  customer_id STRING NOT NULL,
  name STRING,
  tier STRING,
  PRIMARY KEY (customer_id) NOT ENFORCED
) WITH (
  'connector' = 'firestore',
  'project' = 'my-project',
  'collection' = 'customers',
  'lookup.async' = 'true',
  'lookup.cache' = 'PARTIAL',
  'lookup.partial-cache.max-rows' = '10000',
  'lookup.partial-cache.expire-after-write' = '10 min'
);

SELECT e.order_id, c.name, c.tier
FROM order_events AS e
LEFT JOIN customers FOR SYSTEM_TIME AS OF e.proc_time AS c
  ON e.customer_id = c.customer_id AND c.tier = 'gold';

A document that does not exist joins no row, so a LEFT JOIN fills its columns with NULL. A key that cannot be a document id in the collection joins no row without a read: NULL, empty, ., .., a key matching __.*__ (two underscores, anything, two underscores), one longer than 1,500 bytes, and any key holding a /, which would otherwise name another document (a/ names a, and a/sub/b a document of a subcollection). Every other key is sent to the service as it is.

A lookup reads only the fields of the columns the join uses, like a scan, and fills the metadata columns from the document it read; read-time is the time of that read. type-mismatch-policy applies to a looked-up value as it does to a scanned one.

lookup.async runs the join as Flink’s asynchronous lookup, with several reads in flight per subtask, instead of waiting for each read; both modes read through the same BatchGetDocuments call. Flink’s lookup.cache modes NONE and PARTIAL are supported, and Flink owns the partial cache; FULL is refused, because a full cache is loaded by a scan with a snapshot and a reload contract of its own. The cache stores the filtered result under the complete lookup-key tuple, so different additional values for the same document id have separate entries. An additional equality that fails follows the configured missing-key cache policy; expiry and size options keep their standard Flink meaning. lookup.max-retries reads again after UNAVAILABLE, INTERNAL or DEADLINE_EXCEEDED, the statuses the client library already retries this call on, so it adds attempts on top of the library’s rather than retrying anything the library does not; it counts retries after the first read, and every other failure fails the join at once.

Sink#

Write modes#

With a PRIMARY KEY, sink.write-mode says what an insert or an update does to the document:

• merge, the default, merges the row into the document, creating it if it is missing; fields the table does not declare keep their values, and a ROW or MAP column is merged key by key rather than replaced, as Firestore’s own merge option does. A key removed from a MAP therefore stays stored, and an empty MAP leaves the stored map as it is; a NULL column, or an ARRAY, replaces the stored field.
• replace replaces the document with the row, creating it if it is missing; fields the table does not declare are removed.
• update replaces the table’s fields of an existing document; a missing document fails the job. An update table takes inserts and updates but no deletes, and an input that carries deletes is refused when the statement is planned: an update reordered after a delete of its document, or replayed after a restart, would meet a missing document and fail the job again on every restart. The planner can still send a delete when the query’s key differs from the table’s PRIMARY KEY, as in an aggregation written into a table keyed by its result, because it then materializes the upsert and turns a retraction into a delete; such a delete fails the job without deleting anything, and the message says to key the query by the PRIMARY KEY or to use merge or replace. Flink 2.3 refuses such a statement while planning it, so nothing reaches the sink, as Flink 2.3 may demand ON CONFLICT explains; with ON CONFLICT DO DEDUPLICATE or table.exec.sink.require-on-conflict = false, the delete reaches the sink and fails the job as above without deleting anything (measured on the emulator).

Under merge and replace, a delete deletes the document; deleting a missing document succeeds. The default keeps what other writers store in the collection: a job that maintains some fields of documents another application also writes leaves that application’s fields in place, and so does a table that declares fewer fields than the documents hold. Choose replace when the row is the whole document, so that a field the row no longer carries, or a key it removed from a MAP, does not stay stored. sink.write-mode is refused on a table without a PRIMARY KEY, whose rows are always created as new documents.

CREATE TABLE order_snapshots (
  order_id STRING NOT NULL,
  status STRING,
  total DOUBLE,
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'firestore',
  'project' = 'my-project',
  'collection' = 'order_snapshots',
  'sink.write-mode' = 'replace'
);

A table without a PRIMARY KEY accepts inserts only, and creates one document per row:

CREATE TABLE audit_events (
  actor STRING,
  action STRING,
  occurred_at TIMESTAMP_LTZ(3)
) WITH (
  'connector' = 'firestore',
  'project' = 'my-project',
  'collection' = 'users/alice/audit'
);

INSERT INTO audit_events SELECT actor, action, occurred_at FROM staged_events;

Changelog and ordering#

A table with a PRIMARY KEY consumes an upsert changelog: inserts, updates after and, except under update, deletes, keyed by the document id. 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 document 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.

The sink does not keep the order of two writes to one document. It sends writes through the client library’s BulkWriter, which sends several requests at once and applies the writes of one request in no guaranteed order, so a later change of a key can be applied before an earlier one. That is harmless when every key changes at most once in a job, as when the input is a table’s rows written once, but an input that updates a key repeatedly, such as an aggregation, can leave an older value in the document. The Spanner and Bigtable Table sinks share the limitation, and an opt-in mode that keeps each document’s order is #1556.

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, under any sink.write-mode, 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).

What a WITH clause cannot set#

A refused write fails the job: a WITH clause cannot name a serializable failure handler, so the DataStream sink’s routing of refused writes to one is not available, and neither is sink.max-consecutive-rejections, which only bounds a handler that drops them. No table write carries a precondition.

Options#

The configuration reference explains each writer option, and a refusal of a scan.*, lookup.* or sink.* value names the option key the WITH clause spells.

Delivery guarantee#

A scan restored from a checkpoint resumes each split just after the last document it passed, at the same read time, without reading a document twice or skipping one; a job that restarts before its first checkpoint plans again, at a new read time unless scan.read-time is set.

The sink is at-least-once: it waits for every write it has sent at each checkpoint, and a restart replays the records after the last completed one. With a PRIMARY KEY, merge, replace and a delete are idempotent, so a replayed row writes the same document again; an update replayed after something outside the job deleted the document fails the job, and fails again after each restart until the document exists. Without a PRIMARY KEY, a replayed row creates another document under a new id, so a restart can leave duplicates, and so can a retried write whose answer was lost; the idsRedrawn metric counts the second kind, together with drawn ids that named an existing document. See the DataStream delivery guarantee.

Design decisions and testing#

The table mapping, the markers and the declined alternatives are recorded in ADR-0179. The emulator integration tests write every type through SQL and read the documents back with the client library, read every type back through a SQL scan, and join against documents through blocking and asynchronous lookups; the uber-jar’s tests write through its relocated classes.