Add a new column in attempts table (#17806)

* migration attempts

* migrator script

* version fix

airbyte-bootloader/src/test/java/io/airbyte/bootloader/BootloaderAppTest.java

@@ -134,7 +134,7 @@ void testBootloaderAppBlankDb() throws Exception {
       bootloader.load();
 
       val jobsMigrator = new JobsDatabaseMigrator(jobDatabase, jobsFlyway);
-      assertEquals("0.40.4.002", jobsMigrator.getLatestMigration().getVersion().getVersion());
+      assertEquals("0.40.14.001", jobsMigrator.getLatestMigration().getVersion().getVersion());
 
       val configsMigrator = new ConfigsDatabaseMigrator(configDatabase, configsFlyway);
       // this line should change with every new migration

airbyte-db/db-lib/src/main/java/io/airbyte/db/instance/jobs/migrations/V0_40_14_001__AddProcessingTaskQueueInAttempts.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.db.instance.jobs.migrations;
+
+import org.flywaydb.core.api.migration.BaseJavaMigration;
+import org.flywaydb.core.api.migration.Context;
+import org.jooq.DSLContext;
+import org.jooq.impl.DSL;
+import org.jooq.impl.SQLDataType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+// TODO: update migration description in the class name
+public class V0_40_14_001__AddProcessingTaskQueueInAttempts extends BaseJavaMigration {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(
+      V0_40_14_001__AddProcessingTaskQueueInAttempts.class);
+
+  @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());
+    addProtocolVersionColumn(ctx);
+  }
+
+  private static void addProtocolVersionColumn(final DSLContext ctx) {
+    ctx.alterTable("attempts")
+        .addColumnIfNotExists(DSL.field(
+            "processing_task_queue",
+            SQLDataType.VARCHAR(255).nullable(true)))
+        .execute();
+  }
+
+}

airbyte-db/db-lib/src/main/resources/jobs_database/schema_dump.txt

@@ -34,6 +34,7 @@ create table "public"."attempts"(
   "ended_at" timestamptz(35) null,
   "temporal_workflow_id" varchar(256) null,
   "failure_summary" jsonb null,
+  "processing_task_queue" varchar(255) null,
   constraint "attempts_pkey"
     primary key ("id")
 );