Fixes #19944: updates operations to point to new tables (#20945)

* Update OpenMetadataOperations to point to new tables

* Create test file for testing openmetadata operations

* Fix Issues

* Fix Issues in inti system Repository

---------

Co-authored-by: Mohit Yadav <105265192+mohityadav766@users.noreply.github.com>
Co-authored-by: Sriharsha Chintalapani <harshach@users.noreply.github.com>
Co-authored-by: mohitdeuex <mohit.y@deuexsolutions.com>

Author: 3 people

openmetadata-service/src/main/java/org/openmetadata/service/OpenMetadataApplication.java

@@ -105,6 +105,7 @@
 import org.openmetadata.service.jdbi3.EntityRelationshipRepository;
 import org.openmetadata.service.jdbi3.EntityRepository;
 import org.openmetadata.service.jdbi3.MigrationDAO;
+import org.openmetadata.service.jdbi3.SystemRepository;
 import org.openmetadata.service.jdbi3.locator.ConnectionAwareAnnotationSqlLocator;
 import org.openmetadata.service.jdbi3.locator.ConnectionType;
 import org.openmetadata.service.jobs.EnumCleanupHandler;
@@ -236,6 +237,9 @@ public void run(OpenMetadataApplicationConfig catalogConfig, Environment environ
     // Initialize the IndexMapping class
     IndexMappingLoader.init(catalogConfig.getElasticSearchConfiguration());
 
+    // Initialize the MigrationValidationClient, used in the Settings Repository
+    MigrationValidationClient.initialize(jdbi.onDemand(MigrationDAO.class), catalogConfig);
+
     // init for dataSourceFactory
     DatasourceConfig.initialize(catalogConfig.getDataSourceFactory().getDriverClass());
 
@@ -245,15 +249,14 @@ public void run(OpenMetadataApplicationConfig catalogConfig, Environment environ
     Entity.setCollectionDAO(getDao(jdbi));
     Entity.setEntityRelationshipRepository(
         new EntityRelationshipRepository(Entity.getCollectionDAO()));
+    Entity.setSystemRepository(new SystemRepository());
     Entity.setJobDAO(jdbi.onDemand(JobDAO.class));
     Entity.setJdbi(jdbi);
 
     // Initialize bulk operation executor for bounded concurrent processing
     BulkExecutor.initialize(catalogConfig.getBulkOperationConfiguration());
 
     initializeSearchRepository(catalogConfig);
-    // Initialize the MigrationValidationClient, used in the Settings Repository
-    MigrationValidationClient.initialize(jdbi.onDemand(MigrationDAO.class), catalogConfig);
     // as first step register all the repositories
     Entity.initializeRepositories(catalogConfig, jdbi);
     auditLogRepository = new AuditLogRepository(Entity.getCollectionDAO());

openmetadata-service/src/main/java/org/openmetadata/service/util/OpenMetadataOperations.java

@@ -107,6 +107,7 @@
 import org.openmetadata.service.jdbi3.TypeRepository;
 import org.openmetadata.service.jdbi3.UserRepository;
 import org.openmetadata.service.jdbi3.locator.ConnectionType;
+import org.openmetadata.service.migration.MigrationValidationClient;
 import org.openmetadata.service.migration.api.MigrationWorkflow;
 import org.openmetadata.service.resources.CollectionRegistry;
 import org.openmetadata.service.resources.apps.AppMapper;
@@ -174,6 +175,136 @@ public Integer call() {
     return 0;
   }
 
+  @Command(
+      name = "info",
+      description =
+          "Shows the list of migrations applied and the pending migration "
+              + "waiting to be applied on the target database")
+  public Integer info() {
+    try {
+      parseConfig();
+
+      // Then get the native migration info from SERVER_CHANGE_LOG and SERVER_MIGRATION_SQL_LOGS
+      LOG.info("Native System Data Migrations:");
+      MigrationDAO migrationDAO = jdbi.onDemand(MigrationDAO.class);
+      List<MigrationDAO.ServerChangeLog> serverChangeLogs =
+          migrationDAO.listMetricsFromDBMigrations();
+
+      // Create a formatted display for native migrations
+      Set<String> columns = new LinkedHashSet<>(Set.of("version", "installedOn", "status"));
+      List<List<String>> rows = new ArrayList<>();
+
+      for (MigrationDAO.ServerChangeLog serverChangeLog : serverChangeLogs) {
+        List<String> row = new ArrayList<>();
+        row.add(serverChangeLog.getVersion());
+        row.add(serverChangeLog.getInstalledOn());
+
+        if (serverChangeLog.getMetrics() != null) {
+          JsonObject metricsJson =
+              new Gson().fromJson(serverChangeLog.getMetrics(), JsonObject.class);
+          for (Map.Entry<String, JsonElement> entry : metricsJson.entrySet()) {
+            if (!columns.contains(entry.getKey())) {
+              columns.add(entry.getKey());
+            }
+            row.add(entry.getValue().toString());
+          }
+        }
+        rows.add(row);
+      }
+
+      printToAsciiTable(columns.stream().toList(), rows, "No Native Migrations Found");
+
+      return 0;
+    } catch (Exception e) {
+      LOG.error("Failed due to ", e);
+      return 1;
+    }
+  }
+
+  @Command(
+      name = "validate",
+      description =
+          "Checks if the all the migrations haven been applied " + "on the target database.")
+  public Integer validate() {
+    try {
+      parseConfig();
+      // Validate native migrations
+      ConnectionType connType = ConnectionType.from(config.getDataSourceFactory().getDriverClass());
+      DatasourceConfig.initialize(connType.label);
+      MigrationWorkflow workflow =
+          new MigrationWorkflow(
+              jdbi,
+              config.getMigrationConfiguration().getNativePath(),
+              connType,
+              config.getMigrationConfiguration().getExtensionPath(),
+              config.getMigrationConfiguration().getFlywayPath(),
+              config,
+              false);
+      workflow.loadMigrations();
+      workflow.validateMigrationsForServer();
+      return 0;
+    } catch (Exception e) {
+      LOG.error("Database migration validation failed due to ", e);
+      return 1;
+    }
+  }
+
+  @Command(
+      name = "repair",
+      description =
+          "Repairs the SERVER_MIGRATION_SQL_LOGS and SERVER_CHANGE_LOG tables which are used to track "
+              + "all the migrations on the target database This involves removing entries for the failed migrations and update"
+              + "the checksum of migrations already applied on the target database")
+  public Integer repair() {
+    try {
+      parseConfig();
+      // Get the migration workflow to repair native migrations
+      ConnectionType connType = ConnectionType.from(config.getDataSourceFactory().getDriverClass());
+      DatasourceConfig.initialize(connType.label);
+
+      // Handle repair of SERVER_MIGRATION_SQL_LOGS and SERVER_CHANGE_LOG tables
+      try {
+        List<String> failedVersions =
+            jdbi.withHandle(
+                handle ->
+                    handle
+                        .createQuery(
+                            "SELECT version FROM SERVER_CHANGE_LOG WHERE status = 'FAILED'")
+                        .mapTo(String.class)
+                        .list());
+
+        if (!failedVersions.isEmpty()) {
+          LOG.info("Found {} failed migrations in SERVER_CHANGE_LOG", failedVersions.size());
+
+          // Remove failed migrations from SERVER_CHANGE_LOG
+          jdbi.useHandle(
+              handle ->
+                  handle
+                      .createUpdate("DELETE FROM SERVER_CHANGE_LOG WHERE status = 'FAILED'")
+                      .execute());
+
+          // Clean up related entries in SERVER_MIGRATION_SQL_LOGS
+          for (String version : failedVersions) {
+            jdbi.useHandle(
+                handle ->
+                    handle
+                        .createUpdate(
+                            "DELETE FROM SERVER_MIGRATION_SQL_LOGS WHERE version = :version")
+                        .bind("version", version)
+                        .execute());
+          }
+        }
+      } catch (Exception e) {
+        LOG.error("Error repairing SERVER_CHANGE_LOG and SERVER_MIGRATION_SQL_LOGS tables", e);
+        throw e;
+      }
+      return 0;
+    } catch (Exception e) {
+      LOG.error("Repair of migration tables failed due to ", e);
+      return 1;
+    }
+  }
+
   @Command(
       name = "setOpenMetadataUrl",
       description = "Set or update the OpenMetadata URL in the system repository")
@@ -537,6 +668,40 @@ private void installApplication(String appName, AppRepository appRepository) thr
   public Integer checkConnection() {
     try {
       parseConfig();
+      // Check native tables
+      try {
+        jdbi.withHandle(
+            handle -> {
+              try {
+                handle
+                    .createQuery("SELECT COUNT(*) FROM SERVER_CHANGE_LOG")
+                    .mapTo(Integer.class)
+                    .findOne();
+                return true;
+              } catch (Exception e) {
+                LOG.warn("Could not access SERVER_CHANGE_LOG table: {}", e.getMessage());
+                return false;
+              }
+            });
+
+        // querying SERVER_MIGRATION_SQL_LOGS table
+        jdbi.withHandle(
+            handle -> {
+              try {
+                handle
+                    .createQuery("SELECT COUNT(*) FROM SERVER_MIGRATION_SQL_LOGS")
+                    .mapTo(Integer.class)
+                    .findOne();
+                return true;
+              } catch (Exception e) {
+                LOG.warn("Could not access SERVER_MIGRATION_SQL_LOGS table: {}", e.getMessage());
+                return false;
+              }
+            });
+
+      } catch (Exception e) {
+        LOG.warn("Error checking migration tables: {}", e.getMessage());
+      }
       jdbi.open().getConnection();
       return 0;
     } catch (Exception e) {
@@ -2056,6 +2221,15 @@ public void parseConfig() throws Exception {
 
     jdbi = JdbiUtils.createAndSetupJDBI(dataSourceFactory);
 
+    // Initialize the MigrationValidationClient, used in the Settings Repository
+    MigrationValidationClient.initialize(jdbi.onDemand(MigrationDAO.class), config);
+    // Init repos
+    collectionDAO = jdbi.onDemand(CollectionDAO.class);
+    Entity.setJdbi(jdbi);
+    Entity.setCollectionDAO(collectionDAO);
+    Entity.setEntityRelationshipRepository(new EntityRelationshipRepository(collectionDAO));
+    Entity.setSystemRepository(new SystemRepository());
+
     searchRepository =
         SearchRepositoryFactory.createSearchRepository(
             config.getElasticSearchConfiguration(), config.getDataSourceFactory().getMaxSize());
@@ -2065,12 +2239,7 @@ public void parseConfig() throws Exception {
         SecretsManagerFactory.createSecretsManager(
             config.getSecretsManagerConfiguration(), config.getClusterName());
 
-    collectionDAO = jdbi.onDemand(CollectionDAO.class);
     Entity.setSearchRepository(searchRepository);
-    Entity.setJdbi(jdbi);
-    Entity.setCollectionDAO(collectionDAO);
-    Entity.setEntityRelationshipRepository(new EntityRelationshipRepository(collectionDAO));
-    Entity.setSystemRepository(new SystemRepository());
     Entity.initializeRepositories(config, jdbi);
     ConnectionType connType = ConnectionType.from(config.getDataSourceFactory().getDriverClass());
     DatasourceConfig.initialize(connType.label);