Merge pull request #411 from cloudsufi/bugfix/CSM-1032

PLUGIN-1636 : PostgreSQL Sink - Case Sensitivity in Schema name

Author: sgarg-CS

aurora-postgresql-plugin/src/main/java/io/cdap/plugin/auroradb/postgres/AuroraPostgresSink.java

@@ -101,6 +101,11 @@ public String getEscapedTableName() {
       return ESCAPE_CHAR + tableName + ESCAPE_CHAR;
     }
 
+    @Override
+    public String getEscapedDbSchemaName() {
+      return ESCAPE_CHAR + dbSchemaName + ESCAPE_CHAR;
+    }
+
     @Override
     public Map<String, String> getDBSpecificArguments() {
       if (connectionTimeout != null) {

cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLSink.java

@@ -183,6 +183,11 @@ public String getTransactionIsolationLevel() {
     public String getEscapedTableName() {
       return ESCAPE_CHAR + getTableName() + ESCAPE_CHAR;
     }
+    
+    @Override
+    public String getEscapedDbSchemaName() {
+      return ESCAPE_CHAR + getDBSchemaName() + ESCAPE_CHAR;
+    }
 
     @Override
     public Map<String, String> getDBSpecificArguments() {

database-commons/src/main/java/io/cdap/plugin/db/config/AbstractDBSpecificSinkConfig.java

@@ -56,11 +56,13 @@ public abstract class AbstractDBSpecificSinkConfig extends PluginConfig implemen
   @Macro
   @Nullable
   private String dbSchemaName;
+  
   @Name(OPERATION_NAME)
   @Description("Operation for the query to perform. By default, the query performs INSERT operation")
   @Macro
   @Nullable
   protected String operationName;
+  
   @Name(RELATION_TABLE_KEY)
   @Macro
   @Nullable
@@ -89,6 +91,10 @@ public String getEscapedTableName() {
     return tableName;
   }
 
+  public String getEscapedDbSchemaName() {
+    return dbSchemaName;
+  }
+
   @Override
   public boolean canConnect() {
     return !containsMacro(TABLE_NAME) && getConnection().canConnect();

database-commons/src/main/java/io/cdap/plugin/db/config/DatabaseSinkConfig.java

@@ -65,6 +65,15 @@ public interface DatabaseSinkConfig extends DatabaseConnectionConfig {
    */
   String getEscapedTableName();
 
+  /**
+   * Adds escape characters (back quotes, double quotes, etc.) to the database schema name for
+   * databases with case-sensitive identifiers.
+   *
+   * @return dBSchemaName with leading and trailing escape characters appended.
+   * Default implementation returns unchanged table name string.
+   */
+  String getEscapedDbSchemaName();
+
   /**
    * Validate the sink config
    *

database-commons/src/main/java/io/cdap/plugin/db/sink/AbstractDBSink.java

@@ -204,7 +204,7 @@ public void prepareRun(BatchSinkContext context) {
     configAccessor.getConfiguration().set(DBConfiguration.DRIVER_CLASS_PROPERTY, driverClass.getName());
     configAccessor.getConfiguration().set(DBConfiguration.URL_PROPERTY, connectionString);
     String fullyQualifiedTableName = dbSchemaName == null ? dbSinkConfig.getEscapedTableName()
-            : dbSchemaName + "." + dbSinkConfig.getEscapedTableName();
+            : dbSinkConfig.getEscapedDbSchemaName() + "." + dbSinkConfig.getEscapedTableName();
     configAccessor.getConfiguration().set(DBConfiguration.OUTPUT_TABLE_NAME_PROPERTY, fullyQualifiedTableName);
     configAccessor.getConfiguration().set(DBConfiguration.OUTPUT_FIELD_NAMES_PROPERTY, dbColumns);
     configAccessor.setOperationName(dbSinkConfig.getOperationName());
@@ -267,7 +267,7 @@ private Schema inferSchema(Class<? extends Driver> driverClass) {
     List<Schema.Field> inferredFields = new ArrayList<>();
     String dbSchemaName = dbSinkConfig.getDBSchemaName();
     String fullyQualifiedTableName = dbSchemaName == null ? dbSinkConfig.getEscapedTableName()
-            : dbSchemaName + "." + dbSinkConfig.getEscapedTableName();
+            : dbSinkConfig.getEscapedDbSchemaName() + "." + dbSinkConfig.getEscapedTableName();
     try {
       DBUtils.ensureJDBCDriverIsAvailable(driverClass, dbSinkConfig.getConnectionString(),
                                           dbSinkConfig.getJdbcPluginName());
@@ -318,7 +318,7 @@ private void setResultSetMetadata() throws Exception {
     String connectionString = dbSinkConfig.getConnectionString();
     String dbSchemaName = dbSinkConfig.getDBSchemaName();
     String fullyQualifiedTableName = dbSchemaName == null ? dbSinkConfig.getEscapedTableName()
-            : dbSchemaName + "." + dbSinkConfig.getEscapedTableName();
+            : dbSinkConfig.getEscapedDbSchemaName() + "." + dbSinkConfig.getEscapedTableName();
 
     driverCleanup = DBUtils
       .ensureJDBCDriverIsAvailable(driverClass, connectionString, dbSinkConfig.getJdbcPluginName());
@@ -381,7 +381,7 @@ private void validateSchema(FailureCollector collector, Class<? extends Driver>
                               Schema inputSchema, String dbSchemaName) {
     String connectionString = dbSinkConfig.getConnectionString();
     String fullyQualifiedTableName = dbSchemaName == null ? dbSinkConfig.getEscapedTableName()
-            : dbSchemaName + "." + dbSinkConfig.getEscapedTableName();
+            : dbSinkConfig.getEscapedDbSchemaName() + "." + dbSinkConfig.getEscapedTableName();
     try {
       DBUtils.ensureJDBCDriverIsAvailable(jdbcDriverClass, connectionString, dbSinkConfig.getJdbcPluginName());
     } catch (IllegalAccessException | InstantiationException | SQLException e) {
@@ -467,12 +467,14 @@ public abstract static class DBSinkConfig extends DBConfig implements DatabaseSi
     @Description("Name of the database schema of table.")
     @Macro
     @Nullable
-    private String dbSchemaName;
+    public String dbSchemaName;
+
     @Name(OPERATION_NAME)
     @Description("Operation for the query to perform. By default, the query performs INSERT operation")
     @Macro
     @Nullable
     protected String operationName;
+
     @Name(RELATION_TABLE_KEY)
     @Macro
     @Nullable
@@ -486,6 +488,7 @@ public String getTableName() {
     public String getDBSchemaName() {
       return dbSchemaName;
     }
+    
     @Override
     public Operation getOperationName() {
       return Strings.isNullOrEmpty(operationName) ? Operation.INSERT : Operation.valueOf(operationName.toUpperCase());
@@ -506,6 +509,17 @@ public String getEscapedTableName() {
       return tableName;
     }
 
+    /**
+     * Adds escape characters (back quotes, double quotes, etc.) to the database schema name for
+     * databases with case-sensitive identifiers.
+     *
+     * @return dbschemaName with leading and trailing escape characters appended.
+     * Default implementation returns unchanged table name string.
+     */
+    public String getEscapedDbSchemaName() {
+      return dbSchemaName;
+    }
+
     public boolean canConnect() {
       return (!containsMacro(ConnectionConfig.HOST) && !containsMacro(ConnectionConfig.PORT) &&
         !containsMacro(ConnectionConfig.DATABASE) && !containsMacro(TABLE_NAME) && !containsMacro(USER) &&

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSink.java

@@ -127,6 +127,11 @@ public String getEscapedTableName() {
       return ESCAPE_CHAR + getTableName() + ESCAPE_CHAR;
     }
 
+    @Override
+    public String getEscapedDbSchemaName() {
+      return ESCAPE_CHAR + getDBSchemaName() + ESCAPE_CHAR;
+    }
+
     @Override
     protected OracleConnectorConfig getConnection() {
       return connection;

postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresSink.java

@@ -155,6 +155,11 @@ public String getEscapedTableName() {
       return ESCAPE_CHAR + getTableName() + ESCAPE_CHAR;
     }
 
+    @Override
+    public String getEscapedDbSchemaName() {
+      return ESCAPE_CHAR + getDBSchemaName() + ESCAPE_CHAR;
+    }
+
     @Override
     public Map<String, String> getDBSpecificArguments() {
       return ImmutableMap.of(PostgresConstants.CONNECTION_TIMEOUT, String.valueOf(connectionTimeout));

saphana-plugin/src/main/java/io/cdap/plugin/saphana/SapHanaSink.java

@@ -92,5 +92,10 @@ public String getEscapedTableName() {
       return ESCAPE_CHAR + tableName + ESCAPE_CHAR;
     }
 
+    @Override
+    public String getEscapedDbSchemaName() {
+      return ESCAPE_CHAR + dbSchemaName + ESCAPE_CHAR;
+    }
+
   }
 }