faucet-sink-spanner 1.0.1

Google Cloud Spanner sink connector for the faucet-stream ecosystem
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

# faucet-sink-spanner

Google Cloud Spanner sink connector for the [faucet-stream](https://github.com/PawanSikawat/faucet-stream) ecosystem.

Writes JSON records to a Spanner table as batched **mutations** — `Insert` for append, Spanner's native `InsertOrUpdate` for `write_mode: upsert`, and `Delete` mutations for delete mode — one atomic commit per chunk. Supports exactly-once delivery (`delivery: exactly_once`) via a `faucet_commit_token` watermark row committed in the same read-write transaction as the page, and additive schema evolution through the database-admin DDL API.

## Example

```yaml
version: 1
name: kafka-to-spanner
pipeline:
  source:
    kind: kafka
    config:
      brokers: ["localhost:9092"]
      topics: ["users"]
      group_id: faucet
      max_messages: 10000
  sink:
    kind: spanner
    config:
      project_id: my-project
      instance: my-instance
      database: my-db
      table_name: users
      write_mode: upsert
      key: [id]                  # must equal the table's PRIMARY KEY
      auth:
        type: application_default
```

## Configuration

| Field | Type | Default | Description |
|---|---|---|---|
| `project_id` | string | — | GCP project that owns the Spanner instance. |
| `instance` | string | — | Spanner instance ID. |
| `database` | string | — | Database name within the instance. |
| `table_name` | string | — | Target table. Must already exist (mutations require a schema). |
| `auth` | object | `application_default` | Credentials — see [Authentication](#authentication). |
| `max_sessions` | integer | `100` | Upper bound on pooled Spanner sessions. Channels are sized at one per 100 sessions. |
| `emulator_host` | string | — | Emulator endpoint (`host:port`). Plaintext + unauthenticated; scoped to this connector (no env-var races). `SPANNER_EMULATOR_HOST` is still honored when unset. |
| `batch_size` | integer | `1000` | Rows per commit. `0` = no row-count re-chunking (the cell budget below still applies). |
| `ddl_timeout_secs` | integer | `300` | Bound on schema-change (long-running DDL operation) waits. |
| `write_mode` | string | `append` | `append` \| `upsert` \| `delete`. |
| `key` | array | `[]` | Key columns for upsert/delete. **Must equal the table's PRIMARY KEY columns** (order-insensitive) — Spanner mutations always key on the PK. |
| `delete_marker` | object | — | Upsert only: `{ field, values }` — rows whose `field` matches become deletes; the marker field is stripped. |

### Authentication

The `auth` block uses the project-wide `{ type, config }` shape:

```yaml
auth:
  type: service_account_key_path   # or service_account_key | application_default
  config:
    path: /secrets/sa.json
```

- `application_default` (default) — `GOOGLE_APPLICATION_CREDENTIALS`, `gcloud auth application-default login`, or the GCE/GKE metadata server.
- `service_account_key_path` — `config.path` to a key file.
- `service_account_key` — `config.json` inline key (prefer `${vault:…}` / `${env:…}` interpolation).

## Value encoding

Values are encoded against the destination column types read once from `INFORMATION_SCHEMA`:

| Spanner column | Accepted JSON | Notes |
|---|---|---|
| `INT64` | integer, integer-string | String-encoded on the wire — lossless above 2^53. |
| `FLOAT64` / `FLOAT32` | number, `"NaN"`/`"Infinity"`/`"-Infinity"` | |
| `BOOL` | boolean | |
| `STRING` | any | Scalars coerce to text; containers serialize as JSON text. |
| `TIMESTAMP` / `DATE` | RFC 3339 string | |
| `BYTES` | base64 string | Same form the Spanner source decodes to. |
| `NUMERIC` | string or number | Strings pass through losslessly. |
| `JSON` | any | The value's JSON serialization. |
| `ARRAY<T>` | array | Elements encoded as `T`. |

Record fields with no matching column are dropped with a one-shot warning per field. A type mismatch fails that row with an error naming the column.

## Write modes

- **`append`** (default) — `Insert` mutations. A duplicate PRIMARY KEY fails the whole commit (chunk); route pages through a DLQ with `on_batch_error: dlq_all` if you need to absorb duplicates, or use `upsert`.
- **`upsert`** — `InsertOrUpdate` keyed on the table's PRIMARY KEY. `key` must equal the PK column set; this is validated on first write and by `faucet doctor`.
- **`delete`** / `delete_marker` — `Delete` mutations. Key values are re-ordered into PK order automatically. Note Spanner cascades deletes to `INTERLEAVE IN PARENT … ON DELETE CASCADE` child tables.

## Batching and the 80,000-cell commit limit

Spanner rejects commits above ~80,000 mutated cells (rows × columns, **plus** secondary-index amplification the client cannot see). The sink chunks each page at `batch_size` rows *and* a conservative 60,000-cell budget, whichever comes first. If the target table has many secondary indexes, lower `batch_size` further.

## Exactly-once delivery

`delivery: exactly_once` commits each page's mutations **and** an `InsertOrUpdate` on the watermark table in one read-write transaction:

```sql
CREATE TABLE IF NOT EXISTS faucet_commit_token (
  scope      STRING(MAX) NOT NULL,
  token      STRING(MAX) NOT NULL,
  updated_at TIMESTAMP OPTIONS (allow_commit_timestamp=true)
) PRIMARY KEY (scope)
```

The table is auto-created via the admin DDL API on first use. On crash either the page and its token both landed or neither did; on resume the pipeline reads `last_committed_token` and skips already-committed pages. The idempotent page is a **single commit** (no re-chunking), so it is bounded by the 80,000-cell limit — size the source's `batch_size` down for very wide tables. `ABORTED` commit contention is retried automatically by the client with its recommended backoff.

## Schema evolution (drift `on_drift: evolve`)

- **New columns** — `ALTER TABLE … ADD COLUMN IF NOT EXISTS` (`integer→INT64`, `number→FLOAT64`, `boolean→BOOL`, `string→STRING(MAX)`, `object→JSON`), always nullable.
- **Nullability relaxations** — the column is re-emitted at its current type without `NOT NULL`.
- **Base-type widenings (e.g. INT64→FLOAT64) are not supported by Spanner.** The sink returns a typed error advising `allow_type_widening: false`, which makes the drift policy classify the change `incompatible` and route it via `on_incompatible` (`fail` or `quarantine`).

DDL runs as a Spanner long-running operation, bounded by `ddl_timeout_secs`.

## Emulator

Point the sink at the [Cloud Spanner emulator](https://cloud.google.com/spanner/docs/emulator) (`gcr.io/cloud-spanner-emulator/emulator`, gRPC port 9010) with `emulator_host: localhost:9010`. Credentials are ignored against the emulator. The crate's integration tests bootstrap instance + database programmatically through the admin API.

## License

MIT OR Apache-2.0