Fix for DML queries

[vikasrathee-cs]

Fix for DML queries

[gemini-code-assist]

Code Review

This pull request enhances security and robustness across multiple database connectors by implementing SQL identifier quoting and escaping using SqlIdentifierUtils to prevent SQL injection. It also introduces read-only connection wrapping for data sampling operations and adds metadata-based table validation before query execution. Feedback on these changes highlights a potential connection leak in AbstractDBSpecificConnector.getConnection if wrapping the connection fails, as well as potential validation failures in validateTable for databases that do not support catalogs or have mismatched database names.

[gemini-code-assist]

If ReadOnlyConnection.wrap(connection) throws a SQLException, the connection opened by super.getConnection(connectionString, connectionArguments) is leaked because it is not closed before throwing the IllegalArgumentException. We should close the connection in the catch block to prevent resource leaks.

  @Override
  protected Connection getConnection(String connectionString, Properties connectionArguments) {
    Connection connection = super.getConnection(connectionString, connectionArguments);
    try {
      return ReadOnlyConnection.wrap(connection);
    } catch (SQLException e) {
      if (connection != null) {
        try {
          connection.close();
        } catch (SQLException closeEx) {
          e.addSuppressed(closeEx);
        }
      }
      throw new IllegalArgumentException(String.format(
        "Failed to open a read-only connection to database via connection string: %s. Error: %s.",
        connectionString, ExceptionUtils.getRootCauseMessage(e)), e);
    }
  }

[gemini-code-assist]

Passing the database parameter to connection.getMetaData().getTables(...) can cause validation to fail on databases that do not support catalogs (such as Oracle) or when the database name in the path does not match the physical database catalog name (such as in PostgreSQL/Redshift where cross-database queries are not supported and the database name is omitted from the table name).

Consider overriding validateTable in those specific connectors (e.g., OracleConnector, PostgresConnector, RedshiftConnector) to pass null for the database parameter, or handle it generically.