1import * as bigquery from "@distilled.cloud/gcp/bigquery_v2";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { tagRecord } from "../../Tags.ts";
11import { GcpEnvironment } from "../Environment.ts";
12import {
13 createInternalLabels,
14 diffLabels,
15 hasAlchemyLabels,
16 stripInternalLabels,
17 toLabels,
18} from "../Labels.ts";
19import type { Providers } from "../Providers.ts";
20
21const MAX_NAME_LENGTH = 1024;
22const DEFAULT_TYPE = "TABLE";
23
24const TYPE_ALIASES: Record<string, string> = {
25 INT64: "INTEGER",
26 FLOAT64: "FLOAT",
27 BOOL: "BOOLEAN",
28 STRUCT: "RECORD",
29};
30
31export type TableField = {
32 /** Field name. Letters, numbers, underscores; max 300 characters. */
33 name: string;
34 /**
35 * Field type (`STRING`, `INTEGER`/`INT64`, `FLOAT`/`FLOAT64`,
36 * `BOOLEAN`/`BOOL`, `TIMESTAMP`, `DATE`, `RECORD`/`STRUCT`, …).
37 */
38 type: string;
39 /**
40 * Field mode (`NULLABLE`, `REQUIRED`, `REPEATED`).
41 * @default "NULLABLE"
42 */
43 mode?: string;
44 /** Field description. Max 1,024 characters. */
45 description?: string;
46 /** Nested fields when `type` is `RECORD` / `STRUCT`. */
47 fields?: TableField[];
48 /** Max UTF-8 length (`STRING`) or byte length (`BYTES`). */
49 maxLength?: string;
50 /** Precision for `NUMERIC` / `BIGNUMERIC`. */
51 precision?: string;
52 /** Scale for `NUMERIC` / `BIGNUMERIC`. */
53 scale?: string;
54 /** SQL expression for the column default. */
55 defaultValueExpression?: string;
56 /** Collation for `STRING` fields (`und:ci` or `""`). */
57 collation?: string;
58};
59
60export type TableTimePartitioning = {
61 /**
62 * Partition granularity (`DAY`, `HOUR`, `MONTH`, `YEAR`). Immutable —
63 * changing it replaces the table.
64 */
65 type: string;
66 /**
67 * Partitioning column (`TIMESTAMP` or `DATE`). Omit to use
68 * `_PARTITIONTIME`. Immutable — changing it replaces the table.
69 */
70 field?: string;
71 /** Partition expiration in milliseconds. */
72 expirationMs?: string;
73};
74
75export type TableClustering = {
76 /** Top-level clustering columns, most selective first. */
77 fields?: string[];
78};
79
80export type TableProps = {
81 /**
82 * Dataset id (the `{dataset}` segment of
83 * `projects/{project}/datasets/{dataset}`) or a full dataset resource
84 * name. Immutable — changing it replaces the table.
85 */
86 datasetId: string;
87 /**
88 * Table id (the `{table}` segment of
89 * `projects/{project}/datasets/{dataset}/tables/{table}`). If omitted,
90 * a unique name is generated from the stack, stage, and logical id.
91 * Must contain only letters, numbers, and underscores; max 1,024
92 * characters. Immutable — changing it replaces the table.
93 */
94 tableId?: string;
95 /**
96 * User labels. Alchemy ownership labels are merged in automatically.
97 */
98 labels?: Record<string, string>;
99 /**
100 * Table schema. Adding columns updates in place; removing or changing
101 * the type of an existing column is rejected by BigQuery.
102 */
103 schema?: TableField[];
104 /** User-friendly description of this table. */
105 description?: string;
106 /** Descriptive display name. */
107 friendlyName?: string;
108 /**
109 * Expiration time as milliseconds since the Unix epoch (string). Omit
110 * for no expiration.
111 */
112 expirationTime?: string;
113 /**
114 * Time-based partitioning. `type` and `field` are immutable.
115 */
116 timePartitioning?: TableTimePartitioning;
117 /**
118 * Clustering columns. Usually combined with time partitioning.
119 */
120 clustering?: TableClustering;
121 /**
122 * Require a partition filter in queries that can be used for partition
123 * elimination.
124 * @default false
125 */
126 requirePartitionFilter?: boolean;
127 /**
128 * Cloud KMS key used to encrypt table data, as
129 * `projects/{project}/locations/{location}/keyRings/{keyRing}/cryptoKeys/{cryptoKey}`.
130 * Immutable — changing it replaces the table.
131 */
132 kmsKeyName?: string;
133};
134
135export type Table = Resource<
136 "GCP.BigQuery.Table",
137 TableProps,
138 {
139 /**
140 * Resource path `projects/{project}/datasets/{dataset}/tables/{table}`.
141 */
142 name: string;
143 /** Opaque id `project:dataset.table`. */
144 id: string;
145 /** Table id (last path segment). */
146 tableId: string;
147 /** Dataset id. */
148 datasetId: string;
149 /** Project id. */
150 project: string;
151 /** Table type (`TABLE`, `VIEW`, `EXTERNAL`, …). */
152 type: string;
153 /** User labels (Alchemy ownership labels stripped). */
154 labels: Record<string, string>;
155 /** Schema fields, if set. */
156 schema: TableField[] | undefined;
157 /** Description, if set. */
158 description: string | undefined;
159 /** Friendly name, if set. */
160 friendlyName: string | undefined;
161 /** Expiration time in milliseconds since epoch, if set. */
162 expirationTime: string | undefined;
163 /** Dataset location (`US`, `us-central1`, …). */
164 location: string | undefined;
165 /** Time partitioning, if configured. */
166 timePartitioning: TableTimePartitioning | undefined;
167 /** Clustering, if configured. */
168 clustering: TableClustering | undefined;
169 /** Whether queries must specify a partition filter. */
170 requirePartitionFilter: boolean;
171 /** CMEK key, if set. */
172 kmsKeyName: string | undefined;
173 /** Creation time in milliseconds since epoch. */
174 creationTime: string | undefined;
175 /** Row count, excluding the streaming buffer. */
176 numRows: string | undefined;
177 /** Self-link URL. */
178 selfLink: string | undefined;
179 },
180 never,
181 Providers
182>;
183
184/**
185 * A Google BigQuery table.
186 *
187 * Native tables only — views, materialized views, snapshots, and external
188 * tables are not managed by this resource. `datasetId`, `tableId`, time
189 * partitioning `type`/`field`, and `kmsKeyName` are immutable (changing
190 * them replaces the table). Schema, labels, description, clustering, and
191 * partition expiration update in place.
192 *
193 * ### Creating a Table
194 * **Example:** Generated name in an existing dataset
195 * ```typescript
196 * const events = yield* GCP.BigQuery.Table("Events", {
197 * datasetId: "analytics",
198 * schema: [
199 * { name: "id", type: "STRING" },
200 * { name: "created_at", type: "TIMESTAMP" },
201 * ],
202 * });
203 * ```
204 *
205 * **Example:** Explicit id, labels, and partitioning
206 * ```typescript
207 * const events = yield* GCP.BigQuery.Table("Events", {
208 * datasetId: "analytics",
209 * tableId: "order_events",
210 * labels: { env: "prod" },
211 * schema: [
212 * { name: "id", type: "STRING" },
213 * { name: "created_at", type: "TIMESTAMP" },
214 * ],
215 * timePartitioning: { type: "DAY", field: "created_at" },
216 * clustering: { fields: ["id"] },
217 * });
218 * ```
219 *
220 * ### Inserting Rows
221 * **Example:** Stream rows with InsertAll
222 * ```typescript
223 * const insertAll = yield* GCP.BigQuery.InsertAll(events);
224 * yield* insertAll({
225 * body: { rows: [{ json: { id: "1", created_at: "2024-01-01T00:00:00Z" } }] },
226 * });
227 * ```
228 *
229 * @resource
230 * @category BigQuery
231 */
232export const Table = Resource<Table>("GCP.BigQuery.Table");
233
234export class TableNotResolved extends Data.TaggedError(
235 "GCP.BigQuery.TableNotResolved",
236)<{
237 name: string;
238}> {}
239
240const lastSegment = (value: string) => {
241 const trimmed = value.replace(/\/+$/, "");
242 const parts = trimmed.split("/");
243 return parts[parts.length - 1] || trimmed;
244};
245
246const datasetIdOf = (datasetId: string) =>
247 datasetId.includes("/") ? lastSegment(datasetId) : datasetId;
248
249const resourceName = (project: string, datasetId: string, tableId: string) =>
250 `projects/${project}/datasets/${datasetId}/tables/${tableId}`;
251
252const userLabels = (
253 labels: Record<string, string | undefined> | null | undefined,
254): Record<string, string> => stripInternalLabels(tagRecord(labels));
255
256const toId = (id: string, tableId: string | undefined, existing?: string) =>
257 Effect.gen(function* () {
258 if (tableId !== undefined) return tableId;
259 if (existing !== undefined) return existing;
260 const generated = yield* createPhysicalName({
261 id,
262 maxLength: MAX_NAME_LENGTH,
263 lowercase: true,
264 delimiter: "_",
265 });
266 return generated.replaceAll("-", "_");
267 });
268
269const toFieldSchema = (fields: TableField[]): bigquery.TableFieldSchema[] =>
270 fields.map((field) => ({
271 name: field.name,
272 type: field.type,
273 mode: field.mode,
274 description: field.description,
275 fields: field.fields ? toFieldSchema(field.fields) : undefined,
276 maxLength: field.maxLength,
277 precision: field.precision,
278 scale: field.scale,
279 defaultValueExpression: field.defaultValueExpression,
280 collation: field.collation,
281 }));
282
283const schemaOf = (
284 schema: bigquery.TableSchema | undefined,
285): TableField[] | undefined => {
286 const fields = schema?.fields;
287 if (fields === undefined || fields.length === 0) return undefined;
288 return fields.flatMap((field) =>
289 field.name && field.type
290 ? [
291 {
292 name: field.name,
293 type: field.type,
294 mode: field.mode,
295 description: field.description,
296 fields: schemaOf({ fields: field.fields }),
297 maxLength: field.maxLength,
298 precision: field.precision,
299 scale: field.scale,
300 defaultValueExpression: field.defaultValueExpression,
301 collation: field.collation,
302 },
303 ]
304 : [],
305 );
306};
307
308const normalizeType = (type: string) => {
309 const upper = type.toUpperCase();
310 return TYPE_ALIASES[upper] ?? upper;
311};
312
313const canonField = (field: TableField): unknown => ({
314 name: field.name,
315 type: normalizeType(field.type),
316 mode: (field.mode ?? "NULLABLE").toUpperCase(),
317 description: field.description ?? "",
318 maxLength: field.maxLength ?? "",
319 precision: field.precision ?? "",
320 scale: field.scale ?? "",
321 defaultValueExpression: field.defaultValueExpression ?? "",
322 collation: field.collation ?? "",
323 fields: (field.fields ?? []).map(canonField),
324});
325
326const jsonEqual = (left: unknown, right: unknown) =>
327 JSON.stringify(left ?? null) === JSON.stringify(right ?? null);
328
329const schemaChanged = (
330 desired: TableField[] | undefined,
331 observed: bigquery.TableSchema | undefined,
332) =>
333 desired !== undefined &&
334 !jsonEqual(
335 desired.map(canonField),
336 (schemaOf(observed) ?? []).map(canonField),
337 );
338
339const timePartitioningOf = (
340 value: bigquery.TimePartitioning | undefined,
341): TableTimePartitioning | undefined => {
342 if (value?.type === undefined) return undefined;
343 return {
344 type: value.type,
345 field: value.field,
346 expirationMs: value.expirationMs,
347 };
348};
349
350const clusteringOf = (
351 value: bigquery.Clustering | undefined,
352): TableClustering | undefined => {
353 const fields = (value?.fields ?? []).filter(
354 (field): field is string => field !== undefined && field.length > 0,
355 );
356 return fields.length > 0 ? { fields } : undefined;
357};
358
359const sameOptionalString = (
360 left: string | undefined,
361 right: string | undefined,
362) => (left ?? "") === (right ?? "");
363
364const toAttrs = (
365 table: bigquery.Table | bigquery.TableListTablesItem,
366 project: string,
367) => {
368 const full = table as bigquery.Table;
369 const ref = table.tableReference;
370 const tableId = ref?.tableId ?? "";
371 const datasetId = ref?.datasetId ?? "";
372 const projectId = ref?.projectId ?? project;
373 return {
374 name: resourceName(projectId, datasetId, tableId),
375 id: full.id ?? `${projectId}:${datasetId}.${tableId}`,
376 tableId,
377 datasetId,
378 project: projectId,
379 type: table.type ?? DEFAULT_TYPE,
380 labels: userLabels(table.labels),
381 schema: schemaOf(full.schema),
382 description: full.description,
383 friendlyName: table.friendlyName,
384 expirationTime: table.expirationTime,
385 location: full.location,
386 timePartitioning: timePartitioningOf(table.timePartitioning),
387 clustering: clusteringOf(table.clustering),
388 requirePartitionFilter: table.requirePartitionFilter === true,
389 kmsKeyName: full.encryptionConfiguration?.kmsKeyName,
390 creationTime: table.creationTime,
391 numRows: full.numRows,
392 selfLink: full.selfLink,
393 };
394};
395
396const getByRef = (projectId: string, datasetId: string, tableId: string) =>
397 bigquery
398 .getTables({
399 projectId,
400 datasetId,
401 tableId,
402 view: "FULL",
403 })
404 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
405
406const toTableBody = (
407 projectId: string,
408 datasetId: string,
409 tableId: string,
410 news: TableProps,
411 labels: Record<string, string>,
412): bigquery.Table => {
413 const body: bigquery.Table = {
414 tableReference: { projectId, datasetId, tableId },
415 labels,
416 };
417 if (news.schema !== undefined) {
418 body.schema = { fields: toFieldSchema(news.schema) };
419 }
420 if (news.description !== undefined) {
421 body.description = news.description;
422 }
423 if (news.friendlyName !== undefined) {
424 body.friendlyName = news.friendlyName;
425 }
426 if (news.expirationTime !== undefined) {
427 body.expirationTime = news.expirationTime;
428 }
429 if (news.timePartitioning !== undefined) {
430 body.timePartitioning = news.timePartitioning;
431 }
432 if (news.clustering !== undefined) {
433 body.clustering = news.clustering;
434 }
435 if (news.requirePartitionFilter !== undefined) {
436 body.requirePartitionFilter = news.requirePartitionFilter;
437 }
438 if (news.kmsKeyName !== undefined) {
439 body.encryptionConfiguration = { kmsKeyName: news.kmsKeyName };
440 }
441 return body;
442};
443
444export const TableProvider = () =>
445 Provider.succeed(Table, {
446 stables: [
447 "name",
448 "id",
449 "tableId",
450 "datasetId",
451 "project",
452 "location",
453 "creationTime",
454 "selfLink",
455 ],
456
457 diff: Effect.fn(function* ({ news, olds, output }) {
458 if (!isResolved(news)) return undefined;
459 const previousTableId = olds?.tableId ?? output?.tableId;
460 const nextTableId = news.tableId ?? previousTableId;
461 const tableIdChanged =
462 news.tableId !== undefined &&
463 previousTableId !== undefined &&
464 news.tableId !== previousTableId;
465
466 const previousDataset = olds?.datasetId ?? output?.datasetId;
467 const nextDataset = datasetIdOf(news.datasetId);
468 const datasetChanged =
469 previousDataset !== undefined &&
470 datasetIdOf(previousDataset) !== nextDataset;
471
472 const previousPartition =
473 olds?.timePartitioning ?? output?.timePartitioning;
474 const nextPartition = news.timePartitioning;
475 const partitionChanged =
476 nextPartition !== undefined &&
477 ((previousPartition?.type ?? "").toUpperCase() !==
478 nextPartition.type.toUpperCase() ||
479 (previousPartition?.field ?? "") !== (nextPartition.field ?? ""));
480
481 const previousKms = olds?.kmsKeyName ?? output?.kmsKeyName ?? "";
482 const nextKms = news.kmsKeyName ?? previousKms;
483 const kmsChanged = previousKms !== nextKms;
484
485 if (
486 !tableIdChanged &&
487 !datasetChanged &&
488 !partitionChanged &&
489 !kmsChanged
490 ) {
491 return undefined;
492 }
493 return {
494 action: "replace" as const,
495 deleteFirst:
496 !tableIdChanged &&
497 !datasetChanged &&
498 nextTableId !== undefined &&
499 previousTableId !== undefined &&
500 nextTableId === previousTableId,
501 };
502 }),
503
504 read: Effect.fn(function* ({ id, olds, output }) {
505 const env = yield* GcpEnvironment.current;
506 const tableId = yield* toId(id, olds?.tableId, output?.tableId);
507 const datasetId = datasetIdOf(olds?.datasetId ?? output?.datasetId ?? "");
508 if (datasetId.length === 0) return undefined;
509 const existing = yield* getByRef(env.project, datasetId, tableId);
510 if (existing === undefined) return undefined;
511 const attrs = toAttrs(existing, env.project);
512 return (yield* hasAlchemyLabels(id, tagRecord(existing.labels)))
513 ? attrs
514 : Unowned(attrs);
515 }),
516
517 list: () =>
518 Effect.gen(function* () {
519 const env = yield* GcpEnvironment.current;
520 const datasets = yield* bigquery.listDatasets
521 .pages({
522 projectId: env.project,
523 maxResults: 1000,
524 all: true,
525 })
526 .pipe(
527 Stream.flatMap((page) => Stream.fromIterable(page.datasets ?? [])),
528 Stream.runCollect,
529 Effect.map((chunk) => Array.from(chunk)),
530 Effect.catchTag("NotFound", () =>
531 Effect.succeed([] as bigquery.DatasetListDatasetsItem[]),
532 ),
533 );
534 const pages = yield* Effect.forEach(
535 datasets,
536 (dataset) => {
537 const datasetId = dataset.datasetReference?.datasetId;
538 if (datasetId === undefined) {
539 return Effect.succeed([] as ReturnType<typeof toAttrs>[]);
540 }
541 return bigquery.listTables
542 .pages({
543 projectId: env.project,
544 datasetId,
545 maxResults: 1000,
546 })
547 .pipe(
548 Stream.flatMap((page) =>
549 Stream.fromIterable(page.tables ?? []),
550 ),
551 Stream.filter((table) =>
552 Object.keys(table.labels ?? {}).some((key) =>
553 key.startsWith("alchemy-"),
554 ),
555 ),
556 Stream.map((table) => toAttrs(table, env.project)),
557 Stream.runCollect,
558 Effect.map((chunk) => Array.from(chunk)),
559 Effect.catchTag("NotFound", () =>
560 Effect.succeed([] as ReturnType<typeof toAttrs>[]),
561 ),
562 );
563 },
564 { concurrency: 4 },
565 );
566 return pages.flat();
567 }),
568
569 reconcile: Effect.fn(function* ({ id, news, output }) {
570 const env = yield* GcpEnvironment.current;
571 const tableId = yield* toId(id, news.tableId, output?.tableId);
572 const datasetId = datasetIdOf(news.datasetId);
573 const name = resourceName(env.project, datasetId, tableId);
574 const desiredLabels = {
575 ...toLabels(news.labels),
576 ...(yield* createInternalLabels(id)),
577 };
578
579 let current = yield* getByRef(env.project, datasetId, tableId);
580
581 if (current === undefined) {
582 const created = yield* bigquery
583 .insertTables({
584 projectId: env.project,
585 datasetId,
586 body: toTableBody(
587 env.project,
588 datasetId,
589 tableId,
590 news,
591 desiredLabels,
592 ),
593 })
594 .pipe(
595 Effect.catchTag("Conflict", () =>
596 getByRef(env.project, datasetId, tableId),
597 ),
598 );
599 current = created ?? undefined;
600 }
601
602 if (current === undefined) {
603 return yield* new TableNotResolved({ name });
604 }
605
606 const observedLabels = tagRecord(current.labels);
607 const { upsert, removed } = diffLabels(observedLabels, desiredLabels);
608 if (removed.length > 0) {
609 // PATCH merges labels; removing one needs a JSON null the typed body
610 // cannot carry, so replace the labels with PUT (get-modify-put).
611 current = yield* bigquery.updateTables({
612 projectId: env.project,
613 datasetId,
614 tableId,
615 body: { ...current, labels: desiredLabels },
616 });
617 }
618 const labelsChanged = removed.length === 0 && upsert.length > 0;
619 const descriptionChanged =
620 news.description !== undefined &&
621 !sameOptionalString(news.description, current.description);
622 const friendlyNameChanged =
623 news.friendlyName !== undefined &&
624 !sameOptionalString(news.friendlyName, current.friendlyName);
625 const expirationChanged =
626 news.expirationTime !== undefined &&
627 !sameOptionalString(news.expirationTime, current.expirationTime);
628 const fieldsChanged = schemaChanged(news.schema, current.schema);
629 const partitionExpirationChanged =
630 news.timePartitioning !== undefined &&
631 !sameOptionalString(
632 news.timePartitioning.expirationMs,
633 current.timePartitioning?.expirationMs,
634 );
635 const clusteringChanged =
636 news.clustering !== undefined &&
637 !jsonEqual(
638 news.clustering.fields ?? [],
639 current.clustering?.fields ?? [],
640 );
641 const requireFilterChanged =
642 news.requirePartitionFilter !== undefined &&
643 news.requirePartitionFilter !==
644 (current.requirePartitionFilter === true);
645
646 if (
647 labelsChanged ||
648 descriptionChanged ||
649 friendlyNameChanged ||
650 expirationChanged ||
651 fieldsChanged ||
652 partitionExpirationChanged ||
653 clusteringChanged ||
654 requireFilterChanged
655 ) {
656 const body: bigquery.Table = {};
657 if (labelsChanged) body.labels = desiredLabels;
658 if (descriptionChanged) body.description = news.description;
659 if (friendlyNameChanged) body.friendlyName = news.friendlyName;
660 if (expirationChanged) body.expirationTime = news.expirationTime;
661 if (fieldsChanged && news.schema !== undefined) {
662 body.schema = { fields: toFieldSchema(news.schema) };
663 }
664 if (partitionExpirationChanged && news.timePartitioning !== undefined) {
665 body.timePartitioning = news.timePartitioning;
666 }
667 if (clusteringChanged && news.clustering !== undefined) {
668 body.clustering = news.clustering;
669 }
670 if (requireFilterChanged) {
671 body.requirePartitionFilter = news.requirePartitionFilter;
672 }
673 current = yield* bigquery.patchTables({
674 projectId: env.project,
675 datasetId,
676 tableId,
677 body,
678 });
679 }
680
681 return toAttrs(current, env.project);
682 }),
683
684 delete: Effect.fn(function* ({ output }) {
685 yield* bigquery
686 .deleteTables({
687 projectId: output.project,
688 datasetId: output.datasetId,
689 tableId: output.tableId,
690 })
691 .pipe(Effect.catchTag("NotFound", () => Effect.void));
692 }),
693 });