Fix FlyWay Migration Issue (#24985)

* Fix FlyWay Migration Issue

* Separate drop flyway to it's own method

* Fix critical issue in calling getSqlQuery; Add recovery options for --force option

* fix hash <> version in getSqlQuery

* Add recover option for openmetadata-ops.sh

* Add recover option for openmetadata-ops.sh

* naming

* Add tests recover option

* handle all bots

---------

Co-authored-by: Sriharsha Chintalapani <[email protected]>
Co-authored-by: Pere Miquel Brull <[email protected]>
(cherry picked from commit f0ecc2f)

Author: 3 people

openmetadata-service/src/main/java/org/openmetadata/service/migration/api/MigrationProcessImpl.java

@@ -120,7 +120,7 @@ public static Map<String, QueryStatus> performSqlExecutionAndUpdate(
         try {
           String previouslyRanSql = null;
           try {
-            previouslyRanSql = migrationDAO.getSqlQuery(hash(sql), version);
+            previouslyRanSql = migrationDAO.getSqlQuery(version, hash(sql));
           } catch (Exception dbException) {
             // If SERVER_MIGRATION_SQL_LOGS table doesn't exist yet, assume query hasn't run
             previouslyRanSql = null;

openmetadata-service/src/main/java/org/openmetadata/service/migration/api/MigrationWorkflow.java

@@ -1,19 +1,31 @@
 package org.openmetadata.service.migration.api;
 
 import static org.openmetadata.common.utils.CommonUtil.nullOrEmpty;
+import static org.openmetadata.service.util.EntityUtil.hash;
 import static org.openmetadata.service.util.OpenMetadataOperations.printToAsciiTable;
 
 import java.io.File;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import java.util.stream.Stream;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.flywaydb.core.api.configuration.ClassicConfiguration;
+import org.flywaydb.core.api.configuration.Configuration;
+import org.flywaydb.core.internal.database.postgresql.PostgreSQLParser;
+import org.flywaydb.core.internal.parser.Parser;
+import org.flywaydb.core.internal.parser.ParsingContext;
+import org.flywaydb.core.internal.resource.filesystem.FileSystemResource;
+import org.flywaydb.core.internal.sqlscript.SqlStatementIterator;
+import org.flywaydb.database.mysql.MySQLParser;
 import org.jdbi.v3.core.Handle;
 import org.jdbi.v3.core.Jdbi;
 import org.json.JSONObject;
@@ -63,6 +75,17 @@ public MigrationWorkflow(
   }
 
   public void loadMigrations() {
+    // Migrate Flyway history if this is a force migration on an existing database that was
+    // previously managed by Flyway. This must happen BEFORE parsing SQL files.
+    // 1. Migrate DATABASE_CHANGE_LOG entries to SERVER_CHANGE_LOG
+    // 2. Pre-populate SERVER_MIGRATION_SQL_LOGS so flyway queries don't re-execute
+    // NOTE: DO NOT REMOVE
+    if (hasExistingFlywayHistory()) {
+      migrateFlywayToServerChangeLogs();
+      prePopulateFlywayMigrationSQLLogs();
+      dropFlywayTable();
+    }
+
     // Sort Migration on the basis of version
     List<MigrationFile> availableMigrations =
         getMigrationFiles(
@@ -427,4 +450,209 @@ public void updateMigrationStepInDB(
         UUID.randomUUID().toString(),
         metrics.toString());
   }
+
+  private boolean hasExistingFlywayHistory() {
+    try (Handle handle = jdbi.open()) {
+      String checkTableQuery =
+          connectionType == ConnectionType.MYSQL
+              ? "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = 'DATABASE_CHANGE_LOG'"
+              : "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = current_schema() AND table_name = 'DATABASE_CHANGE_LOG'";
+
+      Integer tableExists = handle.createQuery(checkTableQuery).mapTo(Integer.class).one();
+      if (tableExists == null || tableExists == 0) {
+        return false;
+      }
+
+      String countQuery =
+          connectionType == ConnectionType.MYSQL
+              ? "SELECT COUNT(*) FROM DATABASE_CHANGE_LOG WHERE success = true"
+              : "SELECT COUNT(*) FROM \"DATABASE_CHANGE_LOG\" WHERE success = true";
+
+      Integer count = handle.createQuery(countQuery).mapTo(Integer.class).one();
+      return count != null && count > 0;
+    } catch (Exception e) {
+      LOG.debug("Error checking for existing Flyway history: {}", e.getMessage());
+      return false;
+    }
+  }
+
+  private void dropFlywayTable() {
+    // Drop the old DATABASE_CHANGE_LOG table after successful migration
+    try (Handle handle = jdbi.open()) {
+      String dropTableQuery =
+          connectionType == ConnectionType.MYSQL
+              ? "DROP TABLE IF EXISTS DATABASE_CHANGE_LOG"
+              : "DROP TABLE IF EXISTS \"DATABASE_CHANGE_LOG\"";
+
+      try {
+        handle.createUpdate(dropTableQuery).execute();
+        LOG.info("Dropped legacy DATABASE_CHANGE_LOG table");
+      } catch (Exception e) {
+        LOG.warn("Could not drop DATABASE_CHANGE_LOG table: {}", e.getMessage());
+      }
+    }
+  }
+
+  private void migrateFlywayToServerChangeLogs() {
+    LOG.info("Migrating Flyway history from DATABASE_CHANGE_LOG to SERVER_CHANGE_LOG");
+
+    try (Handle handle = jdbi.open()) {
+      // Check if Flyway records have already been migrated
+      String checkMigratedQuery =
+          connectionType == ConnectionType.MYSQL
+              ? """
+              SELECT COUNT(*) FROM SERVER_CHANGE_LOG scl
+              INNER JOIN DATABASE_CHANGE_LOG dcl ON CONCAT('0.0.', CAST(dcl.version AS UNSIGNED)) = scl.version
+              WHERE scl.migrationfilename LIKE '%flyway%'
+              """
+              : """
+              SELECT COUNT(*) FROM SERVER_CHANGE_LOG scl
+              INNER JOIN "DATABASE_CHANGE_LOG" dcl ON '0.0.' || CAST(dcl.version AS INTEGER) = scl.version
+              WHERE scl.migrationfilename LIKE '%flyway%'
+              """;
+
+      try {
+        Integer alreadyMigrated = handle.createQuery(checkMigratedQuery).mapTo(Integer.class).one();
+        if (alreadyMigrated != null && alreadyMigrated > 0) {
+          LOG.info("Flyway records already migrated to SERVER_CHANGE_LOG, skipping");
+          return;
+        }
+      } catch (Exception e) {
+        // SERVER_CHANGE_LOG might not exist yet, continue with migration
+        LOG.debug("Could not check if already migrated: {}", e.getMessage());
+      }
+
+      // Insert v0.0.0 baseline record if not present
+      String insertBaselineQuery =
+          connectionType == ConnectionType.MYSQL
+              ? """
+              INSERT IGNORE INTO SERVER_CHANGE_LOG (version, migrationfilename, checksum, installed_on, metrics)
+              VALUES ('0.0.0', 'bootstrap/sql/migrations/flyway/com.mysql.cj.jdbc.Driver/v000__create_server_change_log.sql', '0', NOW(), NULL)
+              """
+              : """
+              INSERT INTO SERVER_CHANGE_LOG (version, migrationfilename, checksum, installed_on, metrics)
+              VALUES ('0.0.0', 'bootstrap/sql/migrations/flyway/org.postgresql.Driver/v000__create_server_change_log.sql', '0', current_timestamp, NULL)
+              ON CONFLICT (version) DO NOTHING
+              """;
+
+      try {
+        handle.createUpdate(insertBaselineQuery).execute();
+      } catch (Exception e) {
+        LOG.debug("Could not insert baseline record: {}", e.getMessage());
+      }
+
+      // Migrate Flyway migration records to SERVER_CHANGE_LOG
+      String dbDir = connectionType == ConnectionType.MYSQL ? "mysql" : "postgres";
+      String insertQuery =
+          connectionType == ConnectionType.MYSQL
+              ? String.format(
+                  """
+                  INSERT INTO SERVER_CHANGE_LOG (version, migrationfilename, checksum, installed_on, metrics)
+                  SELECT CONCAT('0.0.', CAST(version AS UNSIGNED)) as version,
+                         CASE
+                           WHEN script LIKE 'v%%__.sql' THEN CONCAT('bootstrap/sql/migrations/flyway/%s/', script)
+                           ELSE CONCAT('bootstrap/sql/migrations/flyway/%s/v', version, '__', REPLACE(LOWER(description), ' ', '_'), '.sql')
+                         END as migrationfilename,
+                         '0' as checksum,
+                         installed_on,
+                         NULL as metrics
+                  FROM DATABASE_CHANGE_LOG
+                  WHERE CONCAT('0.0.', CAST(version AS UNSIGNED)) NOT IN (SELECT version FROM SERVER_CHANGE_LOG)
+                  AND success = true
+                  """,
+                  "com.mysql.cj.jdbc.Driver", "com.mysql.cj.jdbc.Driver")
+              : String.format(
+                  """
+                  INSERT INTO SERVER_CHANGE_LOG (version, migrationfilename, checksum, installed_on, metrics)
+                  SELECT '0.0.' || CAST(version AS INTEGER) as version,
+                         CASE
+                           WHEN script LIKE 'v%%__.sql' THEN 'bootstrap/sql/migrations/flyway/%s/' || script
+                           ELSE 'bootstrap/sql/migrations/flyway/%s/v' || version || '__' || REPLACE(LOWER(description), ' ', '_') || '.sql'
+                         END as migrationfilename,
+                         '0' as checksum,
+                         installed_on,
+                         NULL as metrics
+                  FROM "DATABASE_CHANGE_LOG"
+                  WHERE '0.0.' || CAST(version AS INTEGER) NOT IN (SELECT version FROM SERVER_CHANGE_LOG)
+                  AND success = true
+                  """,
+                  "org.postgresql.Driver", "org.postgresql.Driver");
+
+      int migratedCount = handle.createUpdate(insertQuery).execute();
+      LOG.info("Migrated {} Flyway records to SERVER_CHANGE_LOG", migratedCount);
+    } catch (Exception e) {
+      LOG.error("Error during Flyway history migration to SERVER_CHANGE_LOG", e);
+    }
+  }
+
+  private void prePopulateFlywayMigrationSQLLogs() {
+    LOG.info("Pre-populating SERVER_MIGRATION_SQL_LOGS with existing Flyway SQL statements");
+
+    if (flywayPath == null || flywayPath.isEmpty()) {
+      return;
+    }
+
+    String dbSubDir =
+        connectionType == ConnectionType.MYSQL
+            ? "com.mysql.cj.jdbc.Driver"
+            : "org.postgresql.Driver";
+    File flywayDir = new File(flywayPath, dbSubDir);
+
+    if (!flywayDir.exists() || !flywayDir.isDirectory()) {
+      LOG.info("Flyway migration directory does not exist: {}", flywayDir.getPath());
+      return;
+    }
+
+    File[] sqlFiles = flywayDir.listFiles((dir, name) -> name.endsWith(".sql"));
+    if (sqlFiles == null || sqlFiles.length == 0) {
+      return;
+    }
+
+    Pattern versionPattern = Pattern.compile("v(\\d+)__.*\\.sql");
+    ParsingContext parsingContext = new ParsingContext();
+    Configuration configuration = new ClassicConfiguration();
+    Parser parser =
+        connectionType == ConnectionType.MYSQL
+            ? new MySQLParser(configuration, parsingContext)
+            : new PostgreSQLParser(configuration, parsingContext);
+
+    int totalStatements = 0;
+    for (File sqlFile : sqlFiles) {
+      Matcher matcher = versionPattern.matcher(sqlFile.getName());
+      if (!matcher.matches()) {
+        continue;
+      }
+
+      String flywayVersion = matcher.group(1);
+      // Parse as integer to remove leading zeros (e.g., "001" -> 1)
+      String omVersion = "0.0." + Integer.parseInt(flywayVersion);
+
+      try (SqlStatementIterator iterator =
+          parser.parse(
+              new FileSystemResource(
+                  null, sqlFile.getAbsolutePath(), StandardCharsets.UTF_8, true))) {
+        while (iterator.hasNext()) {
+          String sql = iterator.next().getSql();
+          if (sql != null && !sql.isBlank()) {
+            String checksum = hash(sql);
+            try {
+              String existingQuery = migrationDAO.checkIfQueryPreviouslyRan(checksum);
+              if (existingQuery == null) {
+                migrationDAO.upsertServerMigrationSQL(omVersion, sql, checksum);
+                totalStatements++;
+              }
+            } catch (Exception e) {
+              LOG.debug(
+                  "Error inserting SQL statement from {}: {}", sqlFile.getName(), e.getMessage());
+            }
+          }
+        }
+      } catch (Exception e) {
+        LOG.warn("Failed to parse SQL file {}: {}", sqlFile.getName(), e.getMessage());
+      }
+    }
+
+    LOG.info(
+        "Pre-populated {} Flyway SQL statements into SERVER_MIGRATION_SQL_LOGS", totalStatements);
+  }
 }

openmetadata-service/src/main/java/org/openmetadata/service/migration/mysql/v1110/Migration.java

@@ -34,6 +34,7 @@ public Map<String, QueryStatus> runPostDDLScripts(boolean isForceMigration) {
   @Override
   @SneakyThrows
   public void runDataMigration() {
-    this.migrationUtil.migrateFlywayHistory(handle);
+    // Flyway history migration is now handled in MigrationWorkflow.loadMigrations()
+    // before parsing SQL files, to ensure it runs before flyway migrations in force mode
   }
 }

openmetadata-service/src/main/java/org/openmetadata/service/migration/postgres/v1110/Migration.java

@@ -34,6 +34,7 @@ public Map<String, QueryStatus> runPostDDLScripts(boolean isForceMigration) {
   @Override
   @SneakyThrows
   public void runDataMigration() {
-    this.migrationUtil.migrateFlywayHistory(handle);
+    // Flyway history migration is now handled in MigrationWorkflow.loadMigrations()
+    // before parsing SQL files, to ensure it runs before flyway migrations in force mode
   }
 }