Merge branch '10.8.x' into master

Author: semaphore-agent-production[bot]

src/main/java/io/confluent/connect/jdbc/dialect/DatabaseDialect.java

@@ -695,4 +695,14 @@ interface ColumnConverter {
      */
     Object convert(ResultSet resultSet) throws SQLException, IOException;
   }
+
+  /**
+   * Resolve a synonym to its base table name.
+   *
+   * @param connection the database connection; may not be null
+   * @param synonymName the name of the synonym to resolve; may not be null
+   * @return the base table name if the synonym exists, null otherwise
+   * @throws SQLException if there is an error accessing the metadata
+   */
+  String resolveSynonym(Connection connection, String synonymName) throws SQLException;
 }

src/main/java/io/confluent/connect/jdbc/dialect/Db2DatabaseDialect.java

@@ -22,6 +22,8 @@
 import org.apache.kafka.connect.data.Time;
 import org.apache.kafka.connect.data.Timestamp;
 
+import java.sql.Connection;
+import java.sql.SQLException;
 import java.util.Collection;
 
 import io.confluent.connect.jdbc.dialect.DatabaseDialectProvider.SubprotocolBasedProvider;
@@ -190,4 +192,9 @@ protected String sanitizedUrl(String url) {
     return super.sanitizedUrl(url)
                 .replaceAll("(?i)([:;]password=)[^;]*", "$1****");
   }
+
+  @Override
+  public String resolveSynonym(Connection connection, String synonymName) throws SQLException {
+    throw new SQLException("DB2 does not support synonyms. Please use views instead.");
+  }
 }

src/main/java/io/confluent/connect/jdbc/dialect/DerbyDatabaseDialect.java

@@ -23,6 +23,8 @@
 import org.apache.kafka.connect.data.Timestamp;
 
 import java.util.Collection;
+import java.sql.Connection;
+import java.sql.SQLException;
 
 import io.confluent.connect.jdbc.dialect.DatabaseDialectProvider.SubprotocolBasedProvider;
 import io.confluent.connect.jdbc.sink.metadata.SinkRecordField;
@@ -176,4 +178,9 @@ protected String sanitizedUrl(String url) {
     return super.sanitizedUrl(url)
                 .replaceAll("(?i)(;password=)[^;]*", "$1****");
   }
+
+  @Override
+  public String resolveSynonym(Connection connection, String tableId) throws SQLException {
+    throw new SQLException("Derby does not support synonyms. Please use views instead.");
+  }
 }

src/main/java/io/confluent/connect/jdbc/dialect/GenericDatabaseDialect.java

@@ -2092,4 +2092,18 @@ public String identifier() {
   public String toString() {
     return name();
   }
+
+  @Override
+  public String resolveSynonym(Connection connection, String synonymName) throws SQLException {
+    DatabaseMetaData metadata = connection.getMetaData();
+    String tableName = parseTableIdentifier(synonymName).tableName();
+    ResultSet tableRs = metadata.getTables(null, null, tableName, new String[]{"SYNONYM"});
+    if (tableRs.next()) {
+      ResultSet synonymRs = metadata.getColumns(null, null, tableName, null);
+      if (synonymRs.next()) {
+        return synonymRs.getString("TABLE_NAME");
+      }
+    }
+    return null;
+  }
 }

src/main/java/io/confluent/connect/jdbc/dialect/MySqlDatabaseDialect.java

@@ -178,4 +178,9 @@ protected String sanitizedUrl(String url) {
                 .replaceAll("(?i)([(,]password=)[^,)]*", "$1****")
                 .replaceAll("(://[^:]*:)([^@]*)@", "$1****@");
   }
+
+  @Override
+  public String resolveSynonym(Connection connection, String synonymName) throws SQLException {
+    throw new SQLException("MySQL does not support synonyms. Please use views instead.");
+  }
 }

src/main/java/io/confluent/connect/jdbc/dialect/OracleDatabaseDialect.java

@@ -27,7 +27,9 @@
 import java.io.ByteArrayInputStream;
 import java.io.StringReader;
 import java.nio.ByteBuffer;
+import java.sql.Connection;
 import java.sql.PreparedStatement;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Types;
 import org.apache.kafka.common.config.AbstractConfig;
@@ -405,4 +407,20 @@ public TableId parseTableIdentifier(String fqn) {
     }
     return tableId;
   }
+
+  @Override
+  public String resolveSynonym(Connection connection, String synonymName) throws SQLException {
+    try (PreparedStatement stmt = connection.prepareStatement(
+        "SELECT TABLE_OWNER, TABLE_NAME FROM ALL_SYNONYMS WHERE OWNER = ? AND "
+        + "SYNONYM_NAME = ?")) {
+      String tableName = parseTableIdentifier(synonymName).tableName();
+      stmt.setString(1, connection.getMetaData().getUserName().toUpperCase());
+      stmt.setString(2, tableName.toUpperCase());
+      ResultSet rs = stmt.executeQuery();
+      if (rs.next()) {
+        return rs.getString("TABLE_NAME");
+      }
+    }
+    return null;
+  }
 }

src/main/java/io/confluent/connect/jdbc/dialect/PostgreSqlDatabaseDialect.java

@@ -100,6 +100,12 @@ public PostgreSqlDatabaseDialect(AbstractConfig config) {
     super(config, new IdentifierRules(".", "\"", "\""));
   }
 
+
+  @Override
+  public String resolveSynonym(Connection connection, String synonymName) throws SQLException {
+    throw new SQLException("PostgreSQL does not support synonyms. Please use views instead.");
+  }
+
   @Override
   public Connection getConnection() throws SQLException {
     Connection result = super.getConnection();

src/main/java/io/confluent/connect/jdbc/dialect/SapHanaDatabaseDialect.java

@@ -22,6 +22,8 @@
 import org.apache.kafka.connect.data.Time;
 import org.apache.kafka.connect.data.Timestamp;
 
+import java.sql.Connection;
+import java.sql.SQLException;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -166,4 +168,11 @@ public String buildUpsertQueryStatement(
     builder.append(" WITH PRIMARY KEY");
     return builder.toString();
   }
+
+  @Override
+  public String resolveSynonym(Connection connection, String synonymName) throws SQLException {
+    throw new SQLException(
+        "Kafka JDBC Connector doesn't support Synonym Types on SAP Hana. "
+         + "Please use other table types instead.");
+  }
 }

src/main/java/io/confluent/connect/jdbc/dialect/SqlServerDatabaseDialect.java

@@ -27,13 +27,17 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.sql.Types;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
 import java.util.Calendar;
 import java.util.Collection;
 import java.util.Collections;
@@ -504,6 +508,54 @@ public void validateSpecificColumnTypes(
     verifiedSqlServerTimestamp = true;
   }
 
+  @Override
+  public List<TableId> tableIds(Connection conn) throws SQLException {
+    DatabaseMetaData metadata = conn.getMetaData();
+    String[] tableTypes = tableTypes(metadata, this.tableTypes);
+    String tableTypeDisplay = displayableTableTypes(tableTypes, ", ");
+    log.debug("Using {} dialect to get {}", this, tableTypeDisplay);
+    List<TableId> tableIds = new ArrayList<>();
+
+    // Get regular tables using metadata connection
+    try (ResultSet rs = metadata.getTables(catalogPattern(), schemaPattern(), "%", tableTypes)) {
+      while (rs.next()) {
+        String catalogName = rs.getString(1);
+        String schemaName = rs.getString(2);
+        String tableName = rs.getString(3);
+        TableId tableId = new TableId(catalogName, schemaName, tableName);
+        if (includeTable(tableId)) {
+          tableIds.add(tableId);
+        }
+      }
+    }
+
+    // Get synonyms from sys.synonyms on a separate statement
+    if (this.tableTypes.stream().anyMatch("SYNONYM"::equalsIgnoreCase)) {
+      String synonymQuery =
+          "SELECT "
+              + "DB_NAME() as catalog_name, "
+              + "SCHEMA_NAME(schema_id) as schema_name, "
+              + "name as synonym_name "
+              + "FROM sys.synonyms";
+
+      try (Statement stmt = conn.createStatement();
+          ResultSet synonymRs = stmt.executeQuery(synonymQuery)) {
+        while (synonymRs.next()) {
+          String catalogName = synonymRs.getString(1);
+          String schemaName = synonymRs.getString(2);
+          String synonymName = synonymRs.getString(3);
+
+          TableId tableId = new TableId(catalogName, schemaName, synonymName);
+          if (includeTable(tableId)) {
+            tableIds.add(tableId);
+          }
+        }
+      }
+    }
+    log.debug("Used {} dialect to find {} {}", this, tableIds.size(), tableTypeDisplay);
+    return tableIds;
+  }
+
   @Override
   protected ColumnDefinition columnDefinition(
       ResultSet resultSet,
@@ -577,4 +629,20 @@ protected String sanitizedUrl(String url) {
                 .replaceAll("(?i)(;keyStoreSecret=)[^;]*", "$1****")
                 .replaceAll("(?i)(;gsscredential=)[^;]*", "$1****");
   }
+
+  @Override
+  public String resolveSynonym(Connection connection, String synonymName) throws SQLException {
+    try (PreparedStatement stmt =
+        connection.prepareStatement(
+            "SELECT PARSENAME(s.base_object_name, 1) AS TABLE_NAME "
+                + "FROM sys.synonyms s WHERE s.name = ?")) {
+      String tableName = parseTableIdentifier(synonymName).tableName();
+      stmt.setString(1, tableName.toUpperCase());
+      ResultSet rs = stmt.executeQuery();
+      if (rs.next()) {
+        return rs.getString("TABLE_NAME");
+      }
+    }
+    return null;
+  }
 }

src/main/java/io/confluent/connect/jdbc/dialect/SqliteDatabaseDialect.java

@@ -22,6 +22,8 @@
 import org.apache.kafka.connect.data.Time;
 import org.apache.kafka.connect.data.Timestamp;
 
+import java.sql.Connection;
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -144,4 +146,9 @@ public String buildUpsertQueryStatement(
   protected String currentTimestampDatabaseQuery() {
     return "SELECT strftime('%Y-%m-%d %H:%M:%S.%f','now')";
   }
+
+  @Override
+  public String resolveSynonym(Connection connection, String synonymName) throws SQLException {
+    throw new SQLException("SQLite does not support synonyms. Please use views instead.");
+  }
 }