1import * as s3tables from "@distilled.cloud/aws/s3tables";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import { isResolved } from "../../Diff.ts";
5import { createPhysicalName } from "../../PhysicalName.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import type { Providers } from "../Providers.ts";
9import type { TableBucketArn } from "./TableBucket.ts";
10
11/**
12 * A field in an Apache Iceberg table schema.
13 */
14export interface IcebergSchemaField {
15 /**
16 * Name of the field.
17 */
18 name: string;
19 /**
20 * Iceberg field type, e.g. `int`, `long`, `string`, `boolean`,
21 * `timestamp`, `date`, `decimal(10,2)`.
22 */
23 type: string;
24 /**
25 * Whether the field is required (non-nullable).
26 * @default false
27 */
28 required?: boolean;
29}
30
31export interface TableProps {
32 /**
33 * ARN of the table bucket that owns the table. Changing it replaces the
34 * table.
35 */
36 tableBucket: TableBucketArn | string;
37 /**
38 * Name of the namespace that owns the table. Changing it replaces the
39 * table.
40 */
41 namespace: string;
42 /**
43 * Name of the table. Must be 1-255 characters of lowercase letters,
44 * numbers, and underscores, beginning with a letter or number. Changing
45 * the name replaces the table.
46 * @default a deterministic name derived from the app, stage, and logical ID
47 */
48 name?: string;
49 /**
50 * Open table format. Only `ICEBERG` is supported.
51 * @default "ICEBERG"
52 */
53 format?: "ICEBERG";
54 /**
55 * Iceberg schema for the table, applied at create time. Schema evolution
56 * after creation is a data-plane concern; changing this replaces the
57 * table.
58 */
59 schema?: {
60 /**
61 * Ordered list of schema fields.
62 */
63 fields: IcebergSchemaField[];
64 };
65}
66
67export interface Table extends Resource<
68 "AWS.S3Tables.Table",
69 TableProps,
70 {
71 tableArn: string;
72 name: string;
73 namespace: string;
74 tableBucketArn: string;
75 versionToken: string;
76 metadataLocation: string | undefined;
77 warehouseLocation: string;
78 format: string;
79 type: string;
80 createdAt: Date;
81 },
82 never,
83 Providers
84> {}
85
86/**
87 * A fully-managed Apache Iceberg table within an S3 Tables {@link Namespace}.
88 *
89 * S3 Tables manages the table's storage, metadata, and maintenance
90 * (compaction, snapshot expiration). Query it through engines like Amazon
91 * Athena, Amazon EMR, or Apache Spark via the S3 Tables Iceberg catalog.
92 * ### Creating Tables
93 * **Example:** Table with an Iceberg schema
94 * ```typescript
95 * import * as S3Tables from "alchemy/AWS/S3Tables";
96 *
97 * const bucket = yield* S3Tables.TableBucket("Analytics");
98 * const ns = yield* S3Tables.Namespace("Events", {
99 * tableBucket: bucket.tableBucketArn,
100 * });
101 * const table = yield* S3Tables.Table("PageViews", {
102 * tableBucket: bucket.tableBucketArn,
103 * namespace: ns.namespace,
104 * schema: {
105 * fields: [
106 * { name: "id", type: "long", required: true },
107 * { name: "url", type: "string" },
108 * { name: "ts", type: "timestamp" },
109 * ],
110 * },
111 * });
112 * ```
113 *
114 * @resource
115 */
116export const Table = Resource<Table>("AWS.S3Tables.Table");
117
118const createTableName = (id: string, props: { name?: string | undefined }) =>
119 Effect.gen(function* () {
120 if (props.name) {
121 return props.name;
122 }
123 // Table names allow lowercase letters, numbers, and underscores only.
124 const base = yield* createPhysicalName({
125 id,
126 maxLength: 60,
127 lowercase: true,
128 });
129 return base.replaceAll("-", "_");
130 });
131
132const buildMetadata = (
133 props: TableProps,
134): s3tables.TableMetadata | undefined =>
135 props.schema
136 ? {
137 iceberg: {
138 schema: {
139 fields: props.schema.fields.map((f) => ({
140 name: f.name,
141 type: f.type,
142 required: f.required,
143 })),
144 },
145 },
146 }
147 : undefined;
148
149export const TableProvider = () =>
150 Provider.succeed(Table, {
151 stables: ["tableArn", "name", "namespace", "tableBucketArn"],
152 // Tables are scoped to a parent namespace; the engine drives lifecycle
153 // from state rather than ambient enumeration.
154 list: () => Effect.succeed([]),
155 read: Effect.fn(function* ({ id, olds, output }) {
156 const tableBucketArn = output?.tableBucketArn ?? olds?.tableBucket;
157 const namespace = output?.namespace ?? olds?.namespace;
158 if (typeof tableBucketArn !== "string" || typeof namespace !== "string") {
159 return undefined;
160 }
161 const name = output?.name ?? (yield* createTableName(id, olds ?? {}));
162 return yield* s3tables
163 .getTable({ tableBucketARN: tableBucketArn, namespace, name })
164 .pipe(
165 Effect.map((t): Table["Attributes"] => ({
166 tableArn: t.tableARN,
167 name: t.name,
168 namespace: t.namespace[0] ?? namespace,
169 tableBucketArn,
170 versionToken: t.versionToken,
171 metadataLocation: t.metadataLocation,
172 warehouseLocation: t.warehouseLocation,
173 format: t.format,
174 type: t.type,
175 createdAt: t.createdAt,
176 })),
177 Effect.catchTag("NotFoundException", () => Effect.succeed(undefined)),
178 );
179 }),
180 diff: Effect.fn(function* ({ id, news, olds }) {
181 if (!isResolved(news)) return;
182 if (
183 news.tableBucket !== olds?.tableBucket ||
184 news.namespace !== olds?.namespace
185 ) {
186 return { action: "replace" } as const;
187 }
188 const oldName = yield* createTableName(id, olds ?? {});
189 const newName = yield* createTableName(id, news);
190 if (oldName !== newName) {
191 return { action: "replace" } as const;
192 }
193 if ((news.format ?? "ICEBERG") !== (olds?.format ?? "ICEBERG")) {
194 return { action: "replace" } as const;
195 }
196 // Schema is fixed at create time; schema evolution is out of band.
197 const oldSchema = JSON.stringify(olds?.schema ?? null);
198 const newSchema = JSON.stringify(news.schema ?? null);
199 if (oldSchema !== newSchema) {
200 return { action: "replace" } as const;
201 }
202 }),
203 reconcile: Effect.fn(function* ({ id, news, output, session }) {
204 const tableBucketArn = news.tableBucket as string;
205 const namespace = news.namespace as string;
206 const name = output?.name ?? (yield* createTableName(id, news));
207
208 // Observe — read live state; the table may have been deleted
209 // out-of-band even if `output` cached its ARN.
210 let table = yield* s3tables
211 .getTable({ tableBucketARN: tableBucketArn, namespace, name })
212 .pipe(
213 Effect.map((t) => t),
214 Effect.catchTag("NotFoundException", () => Effect.succeed(undefined)),
215 );
216
217 // Ensure — create if missing, tolerating a concurrent create.
218 if (table === undefined) {
219 yield* s3tables
220 .createTable({
221 tableBucketARN: tableBucketArn,
222 namespace,
223 name,
224 format: news.format ?? "ICEBERG",
225 metadata: buildMetadata(news),
226 })
227 .pipe(
228 Effect.asVoid,
229 Effect.catchTag("ConflictException", () => Effect.void),
230 );
231 // Eventual consistency: getTable can briefly 404 a table that
232 // createTable just returned.
233 table = yield* s3tables
234 .getTable({ tableBucketARN: tableBucketArn, namespace, name })
235 .pipe(
236 Effect.retry({
237 while: (e) => e._tag === "NotFoundException",
238 schedule: Schedule.max([
239 Schedule.exponential(500),
240 Schedule.recurs(8),
241 ]),
242 }),
243 );
244 }
245
246 yield* session.note(table.tableARN);
247 return {
248 tableArn: table.tableARN,
249 name: table.name,
250 namespace: table.namespace[0] ?? namespace,
251 tableBucketArn,
252 versionToken: table.versionToken,
253 metadataLocation: table.metadataLocation,
254 warehouseLocation: table.warehouseLocation,
255 format: table.format,
256 type: table.type,
257 createdAt: table.createdAt,
258 };
259 }),
260 delete: Effect.fn(function* ({ output }) {
261 // Deliberately NO versionToken: destroy is unconditional. Runtime
262 // commits (UpdateTableMetadataLocation) rotate the token, so the
263 // persisted one may be stale and would fail with ConflictException.
264 yield* s3tables
265 .deleteTable({
266 tableBucketARN: output.tableBucketArn,
267 namespace: output.namespace,
268 name: output.name,
269 })
270 .pipe(Effect.catchTag("NotFoundException", () => Effect.void));
271 }),
272 });