1import * as dataplex from "@distilled.cloud/gcp/dataplex_v1";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import { Unowned } from "../../AdoptPolicy.ts";
6import { isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import type { Providers } from "../Providers.ts";
11import {
12 fingerprint,
13 lastSegment,
14 listChildResources,
15 listEntities,
16 listLakes,
17 listZones,
18 parseResourceName,
19 replaceIfChanged,
20 toPhysicalSnake,
21 hasAlchemyLabelMap,
22} from "./shared.ts";
23
24export type EntitySchemaField = {
25 /** Field name (letters, numbers, underscores). */
26 name: string;
27 /** Field description. */
28 description?: string;
29 /** Field type (`STRING`, `INT64`, …). */
30 type: dataplex.GoogleCloudDataplexV1SchemaSchemaFieldTypeEnum | (string & {});
31 /** Field mode (`REQUIRED`, `NULLABLE`, `REPEATED`). */
32 mode: dataplex.GoogleCloudDataplexV1SchemaSchemaFieldModeEnum | (string & {});
33 /** Nested fields for `RECORD` types. */
34 fields?: EntitySchemaField[];
35};
36
37export type EntityPartitionField = {
38 /** Partition field name. Immutable. */
39 name: string;
40 /** Partition field type. Immutable. */
41 type:
42 | dataplex.GoogleCloudDataplexV1SchemaPartitionFieldTypeEnum
43 | (string & {});
44};
45
46export type EntitySchema = {
47 /**
48 * When true, Dataplex will not evolve the schema.
49 * @default true
50 */
51 userManaged?: boolean;
52 /** Sequence of data fields. */
53 fields?: EntitySchemaField[];
54 /** Partition keys. Immutable once set. */
55 partitionFields?: EntityPartitionField[];
56 /** Partition path style (`HIVE_COMPATIBLE`). */
57 partitionStyle?:
58 | dataplex.GoogleCloudDataplexV1SchemaPartitionStyleEnum
59 | (string & {});
60};
61
62export type EntityStorageFormat = {
63 /** File format (`PARQUET`, `CSV`, `JSON`, …). */
64 format?:
65 | dataplex.GoogleCloudDataplexV1StorageFormatFormatEnum
66 | (string & {});
67 /** Compression (`GZIP`, `BZIP2`). */
68 compressionFormat?:
69 | dataplex.GoogleCloudDataplexV1StorageFormatCompressionFormatEnum
70 | (string & {});
71 /**
72 * MIME type. Derived from `format` for PARQUET, AVRO, ORC, CSV, and JSON
73 * when omitted.
74 */
75 mimeType?: string;
76};
77
78// Dataplex requires `format.mimeType`; derive it from the file format when
79// the caller only names the format.
80const MIME_TYPES: Record<string, string> = {
81 PARQUET: "application/x-parquet",
82 AVRO: "application/x-avro",
83 ORC: "application/x-orc",
84 CSV: "text/csv",
85 JSON: "application/json",
86};
87
88const withMimeType = (
89 format: EntityStorageFormat | undefined,
90): EntityStorageFormat | undefined => {
91 if (format === undefined || format.mimeType !== undefined) return format;
92 const mimeType = MIME_TYPES[(format.format ?? "").toUpperCase()];
93 return mimeType === undefined ? format : { ...format, mimeType };
94};
95
96export type LakesEntityProps = {
97 /**
98 * Parent zone. Full name
99 * `projects/{project}/locations/{location}/lakes/{lake}/zones/{zone}`.
100 * Immutable — changing it replaces the entity.
101 */
102 zone: string;
103 /**
104 * Entity id used as the published table name. Letters, numbers, and
105 * underscores; at most 256 characters. Immutable — changing it
106 * replaces the entity.
107 */
108 entityId?: string;
109 /**
110 * User-friendly display name (at most 256 characters).
111 */
112 displayName?: string;
113 /**
114 * Description (at most 1024 characters).
115 */
116 description?: string;
117 /**
118 * Entity type. Immutable — changing it replaces the entity.
119 * @default "TABLE"
120 */
121 type?: dataplex.GoogleCloudDataplexV1EntityTypeEnum | (string & {});
122 /**
123 * Asset id (not a full resource name) that stores the entity data.
124 * Immutable — changing it replaces the entity.
125 */
126 asset: string;
127 /**
128 * Storage path (`gs://bucket/path` or a BigQuery table name). Immutable
129 * — changing it replaces the entity.
130 */
131 dataPath: string;
132 /**
133 * Glob of objects that constitute the entity data.
134 */
135 dataPathPattern?: string;
136 /**
137 * Storage system. Immutable — changing it replaces the entity.
138 * @default "CLOUD_STORAGE"
139 */
140 system?: dataplex.GoogleCloudDataplexV1EntitySystemEnum | (string & {});
141 /**
142 * Storage format. Required for Cloud Storage entities.
143 */
144 format?: EntityStorageFormat;
145 /**
146 * Schema. Defaults to a user-managed empty schema when omitted.
147 */
148 schema?: EntitySchema;
149};
150
151export type LakesEntity = Resource<
152 "GCP.Dataplex.LakesEntity",
153 LakesEntityProps,
154 {
155 /** Full resource name `.../zones/{zone}/entities/{entity}`. */
156 name: string;
157 /** Entity id. */
158 entityId: string;
159 /** Parent zone resource name. */
160 zone: string;
161 /** Project id. */
162 project: string;
163 /** Location id. */
164 location: string;
165 /** User-friendly display name. */
166 displayName: string | undefined;
167 /** Entity description. */
168 description: string | undefined;
169 /** Entity type (`TABLE` or `FILESET`). */
170 type: string | undefined;
171 /** Asset id. */
172 asset: string | undefined;
173 /** Storage path. */
174 dataPath: string | undefined;
175 /** Storage system. */
176 system: string | undefined;
177 /** Storage format. */
178 format: string | undefined;
179 /** Data Catalog entry name. */
180 catalogEntry: string | undefined;
181 /** Server etag. */
182 etag: string | undefined;
183 /** Server-assigned uid. */
184 uid: string | undefined;
185 /** RFC3339 creation timestamp. */
186 createTime: string | undefined;
187 /** RFC3339 last-update timestamp. */
188 updateTime: string | undefined;
189 },
190 never,
191 Providers
192>;
193
194/**
195 * A Dataplex metadata entity — a table or fileset registered in a zone.
196 *
197 * Entities have no labels field: `read` reports an entity it finds
198 * without prior state as unowned (adopt it with `--adopt`), and `list`
199 * returns the entities of Alchemy-labeled lakes. Changing `zone`, `entityId`, `type`,
200 * `asset`, `dataPath`, or `system` replaces the entity. Display name,
201 * description, format, and schema update in place.
202 *
203 * ### Creating an Entity
204 * **Example:** User-managed Parquet table
205 * ```typescript
206 * const entity = yield* GCP.Dataplex.LakesEntity("Events", {
207 * zone: zone.name,
208 * asset: asset.assetId,
209 * type: "TABLE",
210 * system: "CLOUD_STORAGE",
211 * dataPath: `gs://${bucket.bucketName}/events`,
212 * format: { format: "PARQUET" },
213 * schema: {
214 * userManaged: true,
215 * fields: [{ name: "id", type: "STRING", mode: "REQUIRED" }],
216 * },
217 * });
218 * ```
219 *
220 * @resource
221 * @category Dataplex
222 */
223export const LakesEntity = Resource<LakesEntity>("GCP.Dataplex.LakesEntity");
224
225export class LakesEntityNotResolved extends Data.TaggedError(
226 "GCP.Dataplex.LakesEntityNotResolved",
227)<{
228 name: string;
229}> {}
230
231const DEFAULT_TYPE = "TABLE";
232const DEFAULT_SYSTEM = "CLOUD_STORAGE";
233
234const resourceNameOf = (zone: string, entityId: string) =>
235 `${zone}/entities/${entityId}`;
236
237const assetIdOf = (asset: string) => lastSegment(asset);
238
239const schemaBody = (
240 schema: EntitySchema | undefined,
241): dataplex.GoogleCloudDataplexV1Schema => ({
242 userManaged: schema?.userManaged !== false,
243 fields: schema?.fields,
244 partitionFields: schema?.partitionFields,
245 partitionStyle: schema?.partitionStyle,
246});
247
248const toAttrs = (
249 entity: dataplex.GoogleCloudDataplexV1Entity,
250 project: string,
251) => {
252 const name = entity.name ?? "";
253 const parsed = parseResourceName(name, "entities");
254 return {
255 name,
256 entityId: entity.id ?? parsed.id,
257 zone: parsed.parent,
258 project: parsed.project || project,
259 location: parsed.location,
260 displayName: entity.displayName,
261 description: entity.description,
262 type: entity.type,
263 asset: entity.asset,
264 dataPath: entity.dataPath,
265 system: entity.system,
266 format: entity.format?.format,
267 catalogEntry: entity.catalogEntry,
268 etag: entity.etag,
269 uid: entity.uid,
270 createTime: entity.createTime,
271 updateTime: entity.updateTime,
272 };
273};
274
275const getByName = (name: string) =>
276 dataplex
277 .getProjectsLocationsLakesZonesEntities({ name, view: "FULL" })
278 .pipe(Effect.catchTag("NotFound", () => Effect.succeed(undefined)));
279
280const listOwnedEntities = (project: string, region: string) =>
281 Effect.gen(function* () {
282 const lakes = (yield* listLakes(project, region)).filter((lake) =>
283 hasAlchemyLabelMap(lake.labels),
284 );
285 const zones = yield* listChildResources(lakes, listZones);
286 const named = zones.filter((zone) => (zone.name ?? "").length > 0);
287 const tables = yield* Effect.forEach(
288 named,
289 (zone) => listEntities(zone.name!, "TABLES"),
290 { concurrency: 4 },
291 );
292 const filesets = yield* Effect.forEach(
293 named,
294 (zone) => listEntities(zone.name!, "FILESETS"),
295 { concurrency: 4 },
296 );
297 return [...tables.flat(), ...filesets.flat()];
298 });
299
300export const LakesEntityProvider = () =>
301 Provider.succeed(LakesEntity, {
302 stables: [
303 "name",
304 "entityId",
305 "zone",
306 "project",
307 "location",
308 "type",
309 "asset",
310 "system",
311 "uid",
312 "createTime",
313 ],
314
315 diff: Effect.fn(function* ({ news, olds, output }) {
316 if (!isResolved(news)) return undefined;
317 const previousId = olds?.entityId ?? output?.entityId;
318 const nextId = news.entityId ?? previousId;
319 const previousZone = olds?.zone ?? output?.zone;
320 const nextZone = news.zone ?? previousZone;
321 const previousType = (
322 olds?.type ??
323 output?.type ??
324 DEFAULT_TYPE
325 ).toUpperCase();
326 const nextType = (news.type ?? previousType).toUpperCase();
327 const previousAsset = assetIdOf(olds?.asset ?? output?.asset ?? "");
328 const nextAsset = assetIdOf(news.asset);
329 const previousPath = olds?.dataPath ?? output?.dataPath ?? "";
330 const nextPath = news.dataPath;
331 const previousSystem = (
332 olds?.system ??
333 output?.system ??
334 DEFAULT_SYSTEM
335 ).toUpperCase();
336 const nextSystem = (news.system ?? previousSystem).toUpperCase();
337 if (
338 replaceIfChanged(previousId, nextId) ||
339 replaceIfChanged(previousZone, nextZone) ||
340 previousType !== nextType ||
341 previousAsset !== nextAsset ||
342 previousPath !== nextPath ||
343 previousSystem !== nextSystem
344 ) {
345 return {
346 action: "replace" as const,
347 deleteFirst:
348 previousZone === nextZone &&
349 previousId !== undefined &&
350 nextId === previousId,
351 };
352 }
353 return undefined;
354 }),
355
356 read: Effect.fn(function* ({ id, olds, output }) {
357 const env = yield* GcpEnvironment.current;
358 const zone = olds?.zone ?? output?.zone ?? "";
359 const entityId = yield* toPhysicalSnake(
360 id,
361 olds?.entityId,
362 output?.entityId,
363 );
364 const name = output?.name ?? resourceNameOf(zone, entityId);
365 const existing = yield* getByName(name);
366 if (existing === undefined) return undefined;
367 const attrs = toAttrs(existing, env.project);
368 // No labels field: without prior state the entity may not be ours.
369 return output !== undefined ? attrs : Unowned(attrs);
370 }),
371
372 list: () =>
373 Effect.gen(function* () {
374 const env = yield* GcpEnvironment.current;
375 const entities = yield* listOwnedEntities(env.project, env.region);
376 const resolved = yield* Effect.forEach(
377 entities,
378 (entity) =>
379 entity.name
380 ? getByName(entity.name).pipe(
381 Effect.map((full) => full ?? entity),
382 )
383 : Effect.succeed(entity),
384 { concurrency: 4 },
385 );
386 return resolved.map((entity) => toAttrs(entity, env.project));
387 }),
388
389 reconcile: Effect.fn(function* ({ id, news, output }) {
390 const env = yield* GcpEnvironment.current;
391 const zone = news.zone;
392 const entityId = yield* toPhysicalSnake(
393 id,
394 news.entityId,
395 output?.entityId,
396 );
397 const name = output?.name ?? resourceNameOf(zone, entityId);
398 const description = news.description;
399 const type = (news.type ?? DEFAULT_TYPE).toUpperCase();
400 const system = (news.system ?? DEFAULT_SYSTEM).toUpperCase();
401 const asset = assetIdOf(news.asset);
402 const schema = schemaBody(news.schema);
403 const format = withMimeType(news.format);
404
405 let current = yield* getByName(name);
406
407 if (current === undefined) {
408 const created = yield* dataplex
409 .createProjectsLocationsLakesZonesEntities({
410 parent: zone,
411 body: {
412 id: entityId,
413 displayName: news.displayName,
414 description,
415 type,
416 asset,
417 dataPath: news.dataPath,
418 dataPathPattern: news.dataPathPattern,
419 system,
420 format,
421 schema,
422 },
423 })
424 .pipe(Effect.catchTag("Conflict", () => getByName(name)));
425 current = created ?? undefined;
426 }
427
428 if (current === undefined) {
429 return yield* new LakesEntityNotResolved({ name });
430 }
431
432 const descriptionChanged =
433 (current.description ?? "") !== (description ?? "");
434 const displayNameChanged =
435 (current.displayName ?? "") !== (news.displayName ?? "");
436 const patternChanged =
437 (current.dataPathPattern ?? "") !== (news.dataPathPattern ?? "");
438 const formatChanged = fingerprint(current.format) !== fingerprint(format);
439 const schemaChanged = fingerprint(current.schema) !== fingerprint(schema);
440
441 if (
442 descriptionChanged ||
443 displayNameChanged ||
444 patternChanged ||
445 formatChanged ||
446 schemaChanged
447 ) {
448 current = yield* dataplex.updateProjectsLocationsLakesZonesEntities({
449 name: current.name ?? name,
450 body: {
451 id: entityId,
452 etag: current.etag,
453 displayName: news.displayName,
454 description,
455 type,
456 asset,
457 dataPath: news.dataPath,
458 dataPathPattern: news.dataPathPattern,
459 system,
460 format,
461 schema,
462 },
463 });
464 }
465
466 return toAttrs(current, env.project);
467 }),
468
469 delete: Effect.fn(function* ({ output }) {
470 // Delete requires the current etag, which moves while partitions are
471 // removed and discovery runs, so re-read it on every attempt.
472 yield* getByName(output.name).pipe(
473 Effect.flatMap((existing) =>
474 existing === undefined
475 ? Effect.void
476 : dataplex.deleteProjectsLocationsLakesZonesEntities({
477 name: output.name,
478 etag: existing.etag,
479 }),
480 ),
481 Effect.retry({
482 while: (error) => error._tag === "Conflict",
483 times: 10,
484 schedule: Schedule.spaced("3 seconds"),
485 }),
486 Effect.catchTag("NotFound", () => Effect.void),
487 );
488 }),
489 });