forked from airbytehq/airbyte-platform
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
migration: backfill actor.default_version_id and mark column as non-n…
…ull (#8502)
- Loading branch information
1 parent
4a71129
commit 57a6d4a
Showing
8 changed files
with
276 additions
and
19 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
72 changes: 72 additions & 0 deletions
72
...b/instance/configs/migrations/V0_50_21_001__BackfillActorDefaultVersionAndSetNonNull.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* | ||
| * Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.db.instance.configs.migrations; | ||
|
|
||
| import com.google.common.annotations.VisibleForTesting; | ||
| import java.util.UUID; | ||
| import org.flywaydb.core.api.migration.BaseJavaMigration; | ||
| import org.flywaydb.core.api.migration.Context; | ||
| import org.jooq.DSLContext; | ||
| import org.jooq.Field; | ||
| import org.jooq.Record; | ||
| import org.jooq.Table; | ||
| import org.jooq.impl.DSL; | ||
| import org.jooq.impl.SQLDataType; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| /** | ||
| * Sets all actor's default_version_id to its actor_definition's default_version_id, and sets the | ||
| * column to be non-null. | ||
| */ | ||
| public class V0_50_21_001__BackfillActorDefaultVersionAndSetNonNull extends BaseJavaMigration { | ||
|
|
||
| private static final Logger LOGGER = LoggerFactory.getLogger(V0_50_21_001__BackfillActorDefaultVersionAndSetNonNull.class); | ||
|
|
||
| private static final Table<Record> ACTOR_DEFINITION = DSL.table("actor_definition"); | ||
| private static final Table<Record> ACTOR = DSL.table("actor"); | ||
|
|
||
| private static final Field<UUID> ID = DSL.field("id", SQLDataType.UUID); | ||
| private static final Field<UUID> DEFAULT_VERSION_ID = DSL.field("default_version_id", SQLDataType.UUID); | ||
| private static final Field<UUID> ACTOR_DEFINITION_ID = DSL.field("actor_definition_id", SQLDataType.UUID); | ||
|
|
||
| @Override | ||
| public void migrate(final Context context) throws Exception { | ||
| LOGGER.info("Running migration: {}", this.getClass().getSimpleName()); | ||
|
|
||
| // Warning: please do not use any jOOQ generated code to write a migration. | ||
| // As database schema changes, the generated jOOQ code can be deprecated. So | ||
| // old migration may not compile if there is any generated code. | ||
| final DSLContext ctx = DSL.using(context.getConnection()); | ||
| backfillActorDefaultVersionId(ctx); | ||
| setNonNull(ctx); | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| static void backfillActorDefaultVersionId(final DSLContext ctx) { | ||
| final var actorDefinitions = ctx.select(ID, DEFAULT_VERSION_ID) | ||
| .from(ACTOR_DEFINITION) | ||
| .fetch(); | ||
|
|
||
| for (final var actorDefinition : actorDefinitions) { | ||
| final UUID actorDefinitionId = actorDefinition.get(ID); | ||
| final UUID defaultVersionId = actorDefinition.get(DEFAULT_VERSION_ID); | ||
|
|
||
| ctx.update(ACTOR) | ||
| .set(DEFAULT_VERSION_ID, defaultVersionId) | ||
| .where(ACTOR_DEFINITION_ID.eq(actorDefinitionId)) | ||
| .execute(); | ||
| } | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| static void setNonNull(final DSLContext ctx) { | ||
| ctx.alterTable(ACTOR) | ||
| .alterColumn(DEFAULT_VERSION_ID) | ||
| .setNotNull() | ||
| .execute(); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
164 changes: 164 additions & 0 deletions
164
...stance/configs/migrations/V0_50_21_001__BackfillActorDefaultVersionAndSetNonNullTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,164 @@ | ||
| /* | ||
| * Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.db.instance.configs.migrations; | ||
|
|
||
| import static org.junit.jupiter.api.Assertions.assertEquals; | ||
| import static org.junit.jupiter.api.Assertions.assertNull; | ||
|
|
||
| import io.airbyte.db.factory.FlywayFactory; | ||
| import io.airbyte.db.instance.configs.AbstractConfigsDatabaseTest; | ||
| import io.airbyte.db.instance.configs.ConfigsDatabaseMigrator; | ||
| import io.airbyte.db.instance.configs.migrations.V0_32_8_001__AirbyteConfigDatabaseDenormalization.ActorType; | ||
| import io.airbyte.db.instance.development.DevDatabaseMigrator; | ||
| import java.util.Objects; | ||
| import java.util.UUID; | ||
| import org.flywaydb.core.Flyway; | ||
| import org.flywaydb.core.api.migration.BaseJavaMigration; | ||
| import org.jooq.DSLContext; | ||
| import org.jooq.Field; | ||
| import org.jooq.JSONB; | ||
| import org.jooq.Record; | ||
| import org.jooq.Table; | ||
| import org.jooq.exception.DataAccessException; | ||
| import org.jooq.impl.DSL; | ||
| import org.jooq.impl.SQLDataType; | ||
| import org.junit.jupiter.api.Assertions; | ||
| import org.junit.jupiter.api.BeforeEach; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| class V0_50_21_001__BackfillActorDefaultVersionAndSetNonNullTest extends AbstractConfigsDatabaseTest { | ||
|
|
||
| private static final Table<Record> ACTOR = DSL.table("actor"); | ||
| private static final Table<Record> ACTOR_DEFINITION = DSL.table("actor_definition"); | ||
| private static final Table<Record> ACTOR_DEFINITION_VERSION = DSL.table("actor_definition_version"); | ||
| private static final Table<Record> WORKSPACE = DSL.table("workspace"); | ||
|
|
||
| private static final Field<UUID> ID_COL = DSL.field("id", SQLDataType.UUID); | ||
| private static final Field<UUID> DEFAULT_VERSION_ID_COL = DSL.field("default_version_id", SQLDataType.UUID); | ||
| private static final Field<UUID> ACTOR_DEFINITION_ID_COL = DSL.field("actor_definition_id", SQLDataType.UUID); | ||
|
|
||
| private static final UUID WORKSPACE_ID = UUID.randomUUID(); | ||
| private static final UUID ACTOR_DEFINITION_ID = UUID.randomUUID(); | ||
| private static final UUID ACTOR_ID = UUID.randomUUID(); | ||
| private static final UUID VERSION_ID = UUID.randomUUID(); | ||
|
|
||
| @BeforeEach | ||
| void beforeEach() { | ||
| final Flyway flyway = | ||
| FlywayFactory.create(dataSource, "V0_50_21_001__BackfillActorDefaultVersionAndSetNonNullTest.java", ConfigsDatabaseMigrator.DB_IDENTIFIER, | ||
| ConfigsDatabaseMigrator.MIGRATION_FILE_LOCATION); | ||
| final ConfigsDatabaseMigrator configsDbMigrator = new ConfigsDatabaseMigrator(database, flyway); | ||
|
|
||
| final BaseJavaMigration previousMigration = new V0_50_20_001__MakeManualNullableForRemoval(); | ||
| final DevDatabaseMigrator devConfigsDbMigrator = new DevDatabaseMigrator(configsDbMigrator, previousMigration.getVersion()); | ||
| devConfigsDbMigrator.createBaseline(); | ||
| } | ||
|
|
||
| private UUID getDefaultVersionIdForActorId(final DSLContext ctx, final UUID actorId) { | ||
| final var actor = ctx.select(DEFAULT_VERSION_ID_COL) | ||
| .from(ACTOR) | ||
| .where(ID_COL.eq(actorId)) | ||
| .fetchOne(); | ||
|
|
||
| if (Objects.isNull(actor)) { | ||
| return null; | ||
| } | ||
|
|
||
| return actor.get(DEFAULT_VERSION_ID_COL); | ||
| } | ||
|
|
||
| static void insertDependencies(final DSLContext ctx) { | ||
| ctx.insertInto(WORKSPACE) | ||
| .columns( | ||
| ID_COL, | ||
| DSL.field("name"), | ||
| DSL.field("slug"), | ||
| DSL.field("initial_setup_complete")) | ||
| .values( | ||
| WORKSPACE_ID, | ||
| "name1", | ||
| "default", | ||
| true) | ||
| .execute(); | ||
|
|
||
| ctx.insertInto(ACTOR_DEFINITION) | ||
| .columns( | ||
| ID_COL, | ||
| DSL.field("name"), | ||
| DSL.field("actor_type")) | ||
| .values( | ||
| ACTOR_DEFINITION_ID, | ||
| "source def name", | ||
| ActorType.source) | ||
| .execute(); | ||
|
|
||
| ctx.insertInto(ACTOR_DEFINITION_VERSION) | ||
| .columns(ID_COL, ACTOR_DEFINITION_ID_COL, DSL.field("docker_repository"), DSL.field("docker_image_tag"), DSL.field("spec")) | ||
| .values(VERSION_ID, ACTOR_DEFINITION_ID, "airbyte/some-source", "1.0.0", JSONB.valueOf("{}")) | ||
| .execute(); | ||
|
|
||
| ctx.update(ACTOR_DEFINITION) | ||
| .set(DEFAULT_VERSION_ID_COL, VERSION_ID) | ||
| .where(ID_COL.eq(ACTOR_DEFINITION_ID)) | ||
| .execute(); | ||
| } | ||
|
|
||
| @Test | ||
| void testBackFillActorDefaultVersionId() { | ||
| final DSLContext ctx = getDslContext(); | ||
| insertDependencies(ctx); | ||
|
|
||
| ctx.insertInto(ACTOR) | ||
| .columns( | ||
| ID_COL, | ||
| ACTOR_DEFINITION_ID_COL, | ||
| DSL.field("workspace_id"), | ||
| DSL.field("name"), | ||
| DSL.field("configuration"), | ||
| DSL.field("actor_type")) | ||
| .values( | ||
| ACTOR_ID, | ||
| ACTOR_DEFINITION_ID, | ||
| WORKSPACE_ID, | ||
| "My Source", | ||
| JSONB.valueOf("{}"), | ||
| ActorType.source) | ||
| .execute(); | ||
|
|
||
| assertNull(getDefaultVersionIdForActorId(ctx, ACTOR_ID)); | ||
|
|
||
| V0_50_21_001__BackfillActorDefaultVersionAndSetNonNull.backfillActorDefaultVersionId(ctx); | ||
|
|
||
| assertEquals(VERSION_ID, getDefaultVersionIdForActorId(ctx, ACTOR_ID)); | ||
| } | ||
|
|
||
| @Test | ||
| void testActorDefaultVersionIdIsNotNull() { | ||
| final DSLContext context = getDslContext(); | ||
|
|
||
| V0_50_21_001__BackfillActorDefaultVersionAndSetNonNull.setNonNull(context); | ||
|
|
||
| final Exception e = Assertions.assertThrows(DataAccessException.class, () -> { | ||
| context.insertInto(ACTOR) | ||
| .columns( | ||
| ID_COL, | ||
| ACTOR_DEFINITION_ID_COL, | ||
| DSL.field("workspace_id"), | ||
| DSL.field("name"), | ||
| DSL.field("configuration"), | ||
| DSL.field("actor_type")) | ||
| .values( | ||
| UUID.randomUUID(), | ||
| UUID.randomUUID(), | ||
| UUID.randomUUID(), | ||
| "My Source", | ||
| JSONB.valueOf("{}"), | ||
| ActorType.source) | ||
| .execute(); | ||
| }); | ||
| Assertions.assertTrue(e.getMessage().contains("null value in column \"default_version_id\" of relation \"actor\" violates not-null constraint")); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters