1import * as glue from "@distilled.cloud/aws/glue";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
10import { AWSEnvironment } from "../Environment.ts";
11import type { Providers } from "../Providers.ts";
12import {
13 cleanMap,
14 retryWhileConcurrentModification,
15 tableArn,
16} from "./internal.ts";
17
18export interface GlueColumn {
19 /** Column name. */
20 name: string;
21 /** Column data type (Hive/Glue type string, e.g. `string`, `bigint`). */
22 type?: string;
23 /** Free-text comment. */
24 comment?: string;
25 /** Free-form key/value properties for the column. */
26 parameters?: Record<string, string>;
27}
28
29export interface GlueSerDeInfo {
30 /** Name of the SerDe. */
31 name?: string;
32 /**
33 * The serialization library class, e.g.
34 * `org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe`.
35 */
36 serializationLibrary?: string;
37 /** SerDe configuration properties. */
38 parameters?: Record<string, string>;
39}
40
41export interface GlueStorageDescriptor {
42 /** Columns of the table (schema). */
43 columns?: GlueColumn[];
44 /** Physical location of the data (usually an S3 path). */
45 location?: string;
46 /**
47 * The input format class, e.g.
48 * `org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat`.
49 */
50 inputFormat?: string;
51 /**
52 * The output format class, e.g.
53 * `org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat`.
54 */
55 outputFormat?: string;
56 /** Whether the data is compressed. */
57 compressed?: boolean;
58 /** Number of buckets used to bucket the table (if `bucketColumns` is set). */
59 numberOfBuckets?: number;
60 /** Serialization/deserialization metadata. */
61 serdeInfo?: GlueSerDeInfo;
62 /** Columns the table is bucketed by. */
63 bucketColumns?: string[];
64 /** Free-form storage descriptor properties. */
65 parameters?: Record<string, string>;
66 /** Whether the table stores data in subdirectories. */
67 storedAsSubDirectories?: boolean;
68}
69
70export interface TableProps {
71 /**
72 * Name of the database the table belongs to. Changing it replaces the
73 * table.
74 */
75 databaseName: string;
76 /**
77 * Name of the table. Glue lowercases table names. If omitted, a unique
78 * lowercase name is generated. Changing the name replaces the table.
79 * @default a generated lowercase physical name
80 */
81 tableName?: string;
82 /**
83 * A description of the table.
84 */
85 description?: string;
86 /**
87 * The table owner.
88 */
89 owner?: string;
90 /**
91 * The type of the table. Common values are `EXTERNAL_TABLE`,
92 * `GOVERNED`, and `VIRTUAL_VIEW`.
93 * @default "EXTERNAL_TABLE"
94 */
95 tableType?: string;
96 /**
97 * Retention time (in days) for the table.
98 */
99 retention?: number;
100 /**
101 * The physical storage descriptor: columns (schema), data location, input/
102 * output formats, and SerDe. This is what Athena reads to query the data.
103 */
104 storageDescriptor?: GlueStorageDescriptor;
105 /**
106 * Partition keys. Athena prunes partitions on these columns. Note: changing
107 * partition keys after creation requires `Force`; Alchemy sends the desired
108 * set on every update.
109 */
110 partitionKeys?: GlueColumn[];
111 /**
112 * Free-form key/value properties stored on the table. Alchemy adds its own
113 * `alchemy::*` ownership markers here (Glue tables are not ARN-taggable) —
114 * user keys are preserved.
115 */
116 parameters?: Record<string, string>;
117 /**
118 * The AWS account ID of the Data Catalog. Changing it replaces the table.
119 * @default the caller's account (the default Data Catalog)
120 */
121 catalogId?: string;
122}
123
124export interface Table extends Resource<
125 "AWS.Glue.Table",
126 TableProps,
127 {
128 /** The (lowercase) name of the table. */
129 tableName: string;
130 /** The name of the database the table belongs to. */
131 databaseName: string;
132 /** The ARN of the table. */
133 tableArn: string;
134 /** The AWS account ID of the Data Catalog the table lives in. */
135 catalogId: string;
136 },
137 {},
138 Providers
139> {}
140
141/**
142 * An AWS Glue Data Catalog table — a schema (columns), storage location, and
143 * SerDe over data in S3 (or another store). This is the unit Athena, Redshift
144 * Spectrum, and EMR query; it is the analytics foundation of a Glue database.
145 * ### Creating Tables
146 * **Example:** Parquet Table over S3
147 * ```typescript
148 * import * as AWS from "alchemy/AWS";
149 *
150 * const database = yield* AWS.Glue.Database("Analytics", {
151 * databaseName: "analytics",
152 * });
153 *
154 * const events = yield* AWS.Glue.Table("Events", {
155 * databaseName: database.databaseName,
156 * tableName: "events",
157 * tableType: "EXTERNAL_TABLE",
158 * storageDescriptor: {
159 * location: "s3://my-data-lake/events/",
160 * inputFormat:
161 * "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat",
162 * outputFormat:
163 * "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat",
164 * serdeInfo: {
165 * serializationLibrary:
166 * "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe",
167 * },
168 * columns: [
169 * { name: "id", type: "string" },
170 * { name: "amount", type: "double" },
171 * ],
172 * },
173 * partitionKeys: [{ name: "dt", type: "string" }],
174 * parameters: { classification: "parquet" },
175 * });
176 * ```
177 *
178 * @resource
179 */
180export const Table = Resource<Table>("AWS.Glue.Table");
181
182const toColumn = (column: GlueColumn) => ({
183 Name: column.name,
184 Type: column.type,
185 Comment: column.comment,
186 Parameters: column.parameters,
187});
188
189export const TableProvider = () =>
190 Provider.effect(
191 Table,
192 Effect.gen(function* () {
193 const createName = Effect.fn(function* (
194 id: string,
195 props: { tableName?: string | undefined },
196 ) {
197 return (
198 props.tableName ??
199 (yield* createPhysicalName({ id, maxLength: 255, lowercase: true }))
200 );
201 });
202
203 const observe = Effect.fn(function* (
204 databaseName: string,
205 name: string,
206 catalogId: string | undefined,
207 ) {
208 return yield* glue
209 .getTable({
210 DatabaseName: databaseName,
211 Name: name,
212 CatalogId: catalogId,
213 })
214 .pipe(
215 Effect.map((r) => r.Table),
216 Effect.catchTag("EntityNotFoundException", () =>
217 Effect.succeed(undefined),
218 ),
219 );
220 });
221
222 const buildTableInput = (name: string, props: TableProps) => {
223 const sd = props.storageDescriptor;
224 return {
225 Name: name,
226 Description: props.description,
227 Owner: props.owner,
228 Retention: props.retention,
229 TableType: props.tableType ?? "EXTERNAL_TABLE",
230 StorageDescriptor: sd
231 ? {
232 Columns: sd.columns?.map(toColumn),
233 Location: sd.location,
234 InputFormat: sd.inputFormat,
235 OutputFormat: sd.outputFormat,
236 Compressed: sd.compressed,
237 NumberOfBuckets: sd.numberOfBuckets,
238 SerdeInfo: sd.serdeInfo
239 ? {
240 Name: sd.serdeInfo.name,
241 SerializationLibrary: sd.serdeInfo.serializationLibrary,
242 Parameters: sd.serdeInfo.parameters,
243 }
244 : undefined,
245 BucketColumns: sd.bucketColumns,
246 Parameters: sd.parameters,
247 StoredAsSubDirectories: sd.storedAsSubDirectories,
248 }
249 : undefined,
250 PartitionKeys: props.partitionKeys?.map(toColumn),
251 };
252 };
253
254 return Table.Provider.of({
255 stables: ["tableName", "databaseName", "tableArn", "catalogId"],
256
257 list: () =>
258 Effect.gen(function* () {
259 const { accountId, region } = yield* AWSEnvironment.current;
260 const dbPages = yield* glue.getDatabases
261 .pages({})
262 .pipe(Stream.runCollect);
263 const databases = Array.from(dbPages).flatMap(
264 (page) => page.DatabaseList ?? [],
265 );
266 const nested = yield* Effect.forEach(
267 databases,
268 (db) =>
269 Effect.gen(function* () {
270 const tablePages = yield* glue.getTables
271 .pages({ DatabaseName: db.Name, CatalogId: db.CatalogId })
272 .pipe(
273 Stream.runCollect,
274 Effect.map((chunk) => Array.from(chunk)),
275 Effect.catchTag("EntityNotFoundException", () =>
276 Effect.succeed([] as glue.GetTablesResponse[]),
277 ),
278 );
279 return tablePages
280 .flatMap((page) => page.TableList ?? [])
281 .map((table) => ({
282 tableName: table.Name,
283 databaseName: table.DatabaseName ?? db.Name,
284 tableArn: tableArn(
285 region,
286 table.CatalogId ?? db.CatalogId ?? accountId,
287 table.DatabaseName ?? db.Name,
288 table.Name,
289 ),
290 catalogId: table.CatalogId ?? db.CatalogId ?? accountId,
291 }));
292 }),
293 { concurrency: 5 },
294 );
295 return nested.flat();
296 }),
297
298 read: Effect.fn(function* ({ id, olds, output }) {
299 const { accountId, region } = yield* AWSEnvironment.current;
300 const databaseName = output?.databaseName ?? olds?.databaseName;
301 if (databaseName === undefined) return undefined;
302 const catalogId = output?.catalogId ?? olds?.catalogId ?? accountId;
303 const name = output?.tableName ?? (yield* createName(id, olds ?? {}));
304 const table = yield* observe(databaseName, name, catalogId);
305 if (table === undefined) return undefined;
306 const attrs = {
307 tableName: table.Name,
308 databaseName: table.DatabaseName ?? databaseName,
309 tableArn: tableArn(
310 region,
311 table.CatalogId ?? catalogId,
312 table.DatabaseName ?? databaseName,
313 table.Name,
314 ),
315 catalogId: table.CatalogId ?? catalogId,
316 };
317 return (yield* hasAlchemyTags(id, cleanMap(table.Parameters)))
318 ? attrs
319 : Unowned(attrs);
320 }),
321
322 diff: Effect.fn(function* ({ id, news, olds }) {
323 if (!isResolved(news)) return undefined;
324 const oldName = yield* createName(id, olds);
325 const newName = yield* createName(id, news);
326 if (oldName !== newName) return { action: "replace" } as const;
327 if (olds.databaseName !== news.databaseName) {
328 return { action: "replace" } as const;
329 }
330 if ((olds.catalogId ?? undefined) !== (news.catalogId ?? undefined)) {
331 return { action: "replace" } as const;
332 }
333 // schema / storage / partition keys / parameters → update
334 }),
335
336 reconcile: Effect.fn(function* ({ id, news, output, session }) {
337 const { accountId, region } = yield* AWSEnvironment.current;
338 const catalogId = news.catalogId ?? output?.catalogId ?? accountId;
339 const name = output?.tableName ?? (yield* createName(id, news));
340 const internalTags = yield* createInternalTags(id);
341 const tableInput = {
342 ...buildTableInput(name, news),
343 Parameters: { ...news.parameters, ...internalTags },
344 };
345
346 // 1. OBSERVE
347 let table = yield* observe(news.databaseName, name, catalogId);
348
349 // 2. ENSURE / 3. SYNC
350 if (table === undefined) {
351 yield* glue
352 .createTable({
353 CatalogId: catalogId,
354 DatabaseName: news.databaseName,
355 TableInput: tableInput,
356 })
357 .pipe(
358 Effect.catchTag("AlreadyExistsException", () => Effect.void),
359 retryWhileConcurrentModification,
360 );
361 } else {
362 // UpdateTable replaces the full TableInput. Force lets partition
363 // keys change without a version conflict.
364 yield* glue
365 .updateTable({
366 CatalogId: catalogId,
367 DatabaseName: news.databaseName,
368 TableInput: tableInput,
369 Force: true,
370 })
371 .pipe(retryWhileConcurrentModification);
372 }
373 table = yield* observe(news.databaseName, name, catalogId);
374
375 yield* session.note(name);
376 return {
377 tableName: name,
378 databaseName: news.databaseName,
379 tableArn: tableArn(
380 region,
381 table?.CatalogId ?? catalogId,
382 news.databaseName,
383 name,
384 ),
385 catalogId: table?.CatalogId ?? catalogId,
386 };
387 }),
388
389 delete: Effect.fn(function* ({ output }) {
390 yield* glue
391 .deleteTable({
392 CatalogId: output.catalogId,
393 DatabaseName: output.databaseName,
394 Name: output.tableName,
395 })
396 .pipe(
397 retryWhileConcurrentModification,
398 Effect.catchTag("EntityNotFoundException", () => Effect.void),
399 );
400 }),
401 });
402 }),
403 );