fix(workflows): make Flowable schema upgrades idempotent to survive partial migrations (#27234)

* fix(workflows): make Flowable schema upgrades idempotent to survive partial migrations

Fixes #26048.

When the server crashed mid-startup during a Flowable schema upgrade, the DB
was left in a partially-migrated state. On restart, Flowable re-ran the same
DDL and failed on already-existing objects (indexes, tables, columns), permanently
wedging both the server and migrate --force.

Changes:
1. WorkflowHandler: webserver now uses DB_SCHEMA_UPDATE_FALSE — it validates the
   schema but never runs DDL. Only migrate CLI uses DB_SCHEMA_UPDATE_TRUE.
2. OpenMetadataOperations: explicit WorkflowHandler.initialize(config, true) inside
   the migrate command so Flowable DDL always runs during migration.
3. WorkflowHandler: catches FlowableWrongDbException on webserver startup and
   rethrows with an actionable message directing the operator to run migrate.
4. IdempotentDdlDataSource + IdempotentDdlStatement: JDBC DataSource wrapper used
   exclusively in migration context. Intercepts execute(sql) for CREATE INDEX,
   CREATE TABLE, and ALTER TABLE ADD COLUMN and pre-checks existence via standard
   DatabaseMetaData (getIndexInfo, getTables, getColumns) before executing. If the
   object already exists it logs a skip and returns — no SQL state codes, no string
   matching, works on MySQL and PostgreSQL.

Unit tests cover schema-update mode selection in both contexts.

* fix(workflows): address review comments on idempotent DDL wrapper

- Extract shouldSkip() helper; apply idempotency checks to all execute()
  and executeUpdate() overloads, not just execute(String)
- Tighten ALTER TABLE regex with negative lookahead to exclude SQL keywords
  (CONSTRAINT, PRIMARY, UNIQUE, FOREIGN, CHECK, INDEX, KEY) from being
  matched as column names
- IdempotentDdlDataSource now wraps a DataSource delegate instead of calling
  DriverManager directly; uses migrationDataSource() helper in WorkflowHandler
  to resolve from existing DataSource or JDBC params
- Fix InvocationTargetException wrapping in Connection proxy — unwrap cause
  so callers receive the original SQLException
- Wrap all createStatement() variants in the proxy, not just the no-arg form
- Contextual error message in WorkflowHandler — distinguish between server
  startup and migration context
- Add IdempotentDdlStatementTest: 11 tests covering skip/execute for
  CREATE INDEX, CREATE UNIQUE INDEX, CREATE TABLE, ALTER TABLE ADD COLUMN,
  keyword-guarded ALTER TABLE, executeUpdate overload, and pass-through

* fix(workflows): include DB/library versions in FlowableWrongDbException message

* test(workflows): add IdempotentDdlDataSourceTest for proxy wrapping and exception surfacing

* test(workflows): assert exception identity in proxy exception-surfacing tests

* fix(workflows): catalog-aware identifier normalization in IdempotentDdlStatement

On MySQL with lower_case_table_names=0 (default on Linux), table names are
stored as-is and catalog=null metadata lookups can miss existing objects.

- Use connection.getCatalog() for all getIndexInfo/getTables/getColumns calls
- Normalize identifiers via DatabaseMetaData.storesLowerCaseIdentifiers() /
  storesUpperCaseIdentifiers() instead of unconditional toLowerCase()
- stripIdentifierQuotes() handles backtick, double-quote and bracket quoting
- extractObjectName() handles schema-qualified names (schema.table)
- columnExists now iterates and normalizes COLUMN_NAME from ResultSet
- Test: added MySQL uppercase storage case to IdempotentDdlStatementTest

* fix(workflows): null guard in shouldSkip, drop-create Flowable init, robust test indexing

- shouldSkip() returns false immediately for null SQL, preserving JDBC contract
  (delegate handles null and throws the driver's own error)
- drop-create command now calls WorkflowHandler.initialize(config, true) after
  native migrations so it produces a fully startable DB including Flowable tables
- WorkflowHandlerSchemaUpdateTest: replace brittle get(1) with getLast() so the
  test is not sensitive to how many StandaloneProcessEngineConfiguration instances
  are constructed before initializeNewProcessEngine runs

Author: yan-3005

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/IdempotentDdlDataSource.java

@@ -0,0 +1,112 @@
+/*
+ *  Copyright 2021 Collate
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.openmetadata.service.governance.workflows;
+
+import java.io.PrintWriter;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.sql.Statement;
+import java.util.logging.Logger;
+import javax.sql.DataSource;
+
+/**
+ * A {@link DataSource} wrapper that intercepts every {@link Connection} it vends and ensures
+ * that all {@link Statement} variants created from that connection go through
+ * {@link IdempotentDdlStatement}. Only used in migration context so Flowable upgrade scripts
+ * skip already-existing DDL objects instead of failing.
+ */
+final class IdempotentDdlDataSource implements DataSource {
+
+  private final DataSource delegate;
+
+  IdempotentDdlDataSource(DataSource delegate) {
+    this.delegate = delegate;
+  }
+
+  @Override
+  public Connection getConnection() throws SQLException {
+    return wrapConnection(delegate.getConnection());
+  }
+
+  @Override
+  public Connection getConnection(String username, String password) throws SQLException {
+    return wrapConnection(delegate.getConnection(username, password));
+  }
+
+  private Connection wrapConnection(Connection real) {
+    return (Connection)
+        Proxy.newProxyInstance(
+            real.getClass().getClassLoader(),
+            new Class<?>[] {Connection.class},
+            (proxy, method, args) -> {
+              if ("createStatement".equals(method.getName())) {
+                Statement stmt = (Statement) invokeDelegate(method, real, args);
+                return new IdempotentDdlStatement(stmt, real);
+              }
+              return invokeDelegate(method, real, args);
+            });
+  }
+
+  /**
+   * Invokes a method on the real connection, unwrapping any {@link InvocationTargetException}
+   * so callers receive the original exception type (e.g. {@link SQLException}).
+   */
+  private static Object invokeDelegate(Method method, Object target, Object[] args)
+      throws Throwable {
+    try {
+      return method.invoke(target, args);
+    } catch (InvocationTargetException e) {
+      throw e.getCause();
+    }
+  }
+
+  @Override
+  public PrintWriter getLogWriter() throws SQLException {
+    return delegate.getLogWriter();
+  }
+
+  @Override
+  public void setLogWriter(PrintWriter out) throws SQLException {
+    delegate.setLogWriter(out);
+  }
+
+  @Override
+  public void setLoginTimeout(int seconds) throws SQLException {
+    delegate.setLoginTimeout(seconds);
+  }
+
+  @Override
+  public int getLoginTimeout() throws SQLException {
+    return delegate.getLoginTimeout();
+  }
+
+  @Override
+  public Logger getParentLogger() throws SQLFeatureNotSupportedException {
+    return delegate.getParentLogger();
+  }
+
+  @Override
+  public <T> T unwrap(Class<T> iface) throws SQLException {
+    return delegate.unwrap(iface);
+  }
+
+  @Override
+  public boolean isWrapperFor(Class<?> iface) throws SQLException {
+    return delegate.isWrapperFor(iface);
+  }
+}