1import * as GCP from "@/GCP";
2import * as Output from "@/Output";
3import * as Test from "@/Test/Alchemy";
4import * as dataplex from "@distilled.cloud/gcp/dataplex_v1";
5import { expect } from "alchemy-test";
6import * as Effect from "effect/Effect";
7import { MinimumLogLevel } from "effect/References";
8import * as Schedule from "effect/Schedule";
9import { withDataplexSlot } from "./quota.ts";
10
11const { test } = Test.make({ providers: GCP.providers() });
12
13const logLevel = Effect.provideService(
14 MinimumLogLevel,
15 process.env.DEBUG ? "Debug" : "Info",
16);
17
18const waitUntilGone = (name: string) =>
19 dataplex.getProjectsLocationsLakesZonesEntitiesPartitions({ name }).pipe(
20 Effect.as("found" as const),
21 Effect.catchTag("NotFound", () => Effect.succeed("gone" as const)),
22 Effect.repeat({
23 schedule: Schedule.spaced("2 seconds"),
24 until: (status) => status === "gone",
25 times: 10,
26 }),
27 );
28
29// A lake (~2.5 min) plus a bucket asset (~1.5 min), then teardown of both,
30// runs past 5 minutes.
31test.provider.skipIf(!process.env.GCP_TEST_SLOW || !!process.env.FAST)(
32 "create and delete a lake entity partition",
33 (stack) =>
34 Effect.gen(function* () {
35 yield* stack.destroy();
36
37 const created = yield* stack.deploy(
38 Effect.gen(function* () {
39 const bucket = yield* GCP.Storage.Bucket("PartitionData", {
40 location: "US-CENTRAL1",
41 forceDestroy: true,
42 });
43 const lake = yield* GCP.Dataplex.Lake("Warehouse", {
44 location: "us-central1",
45 labels: { env: "test" },
46 });
47 const zone = yield* GCP.Dataplex.LakesZone("Landing", {
48 lake: lake.name,
49 type: "RAW",
50 locationType: "SINGLE_REGION",
51 discoverySpec: { enabled: false },
52 labels: { env: "test" },
53 });
54 const asset = yield* GCP.Dataplex.LakesAsset("RawBucket", {
55 zone: zone.name,
56 resourceSpec: {
57 type: "STORAGE_BUCKET",
58 name: Output.interpolate`projects/${bucket.projectNumber}/buckets/${bucket.bucketName}`,
59 },
60 discoverySpec: { enabled: false },
61 labels: { env: "test" },
62 });
63 const entity = yield* GCP.Dataplex.LakesEntity("Events", {
64 zone: zone.name,
65 asset: asset.assetId,
66 type: "TABLE",
67 system: "CLOUD_STORAGE",
68 dataPath: Output.interpolate`gs://${bucket.bucketName}/events`,
69 format: { format: "PARQUET" },
70 schema: {
71 userManaged: true,
72 fields: [{ name: "id", type: "STRING", mode: "REQUIRED" }],
73 partitionFields: [{ name: "dt", type: "STRING" }],
74 partitionStyle: "HIVE_COMPATIBLE",
75 },
76 });
77 const partition = yield* GCP.Dataplex.LakesEntitiesPartition("Day", {
78 entity: entity.name,
79 values: ["2024-01-01"],
80 location: Output.interpolate`gs://${bucket.bucketName}/events/dt=2024-01-01`,
81 });
82 return { bucket, lake, zone, asset, entity, partition };
83 }),
84 );
85
86 expect(created.partition.name).toContain("/partitions/");
87 expect(created.partition.values).toEqual(["2024-01-01"]);
88 expect(created.partition.entity).toEqual(created.entity.name);
89 expect(created.partition.location).toContain("dt=2024-01-01");
90
91 const fetched =
92 yield* dataplex.getProjectsLocationsLakesZonesEntitiesPartitions({
93 name: created.partition.name,
94 });
95 expect(fetched.name).toEqual(created.partition.name);
96 expect(fetched.values).toEqual(["2024-01-01"]);
97
98 yield* stack.destroy();
99 const gone = yield* waitUntilGone(created.partition.name);
100 expect(gone).toEqual("gone");
101 }).pipe(logLevel, withDataplexSlot),
102 { tags: ["provider:gcp", "provider:gcp:dataplex", "live"], timeout: 900_000 },
103);