diff --git a/embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/AbstractJdbcOutputPlugin.java b/embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/AbstractJdbcOutputPlugin.java index 9f2c1cb9..41cf480d 100644 --- a/embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/AbstractJdbcOutputPlugin.java +++ b/embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/AbstractJdbcOutputPlugin.java @@ -583,7 +583,7 @@ protected void doBegin(JdbcOutputConnection con, } } } - task.setActualTable(new TableIdentifier(null, con.getSchemaName(), actualTable)); + task.setActualTable(new TableIdentifier(con.getDatabaseName(), con.getSchemaName(), actualTable)); Optional initialTargetTableSchema = mode.ignoreTargetTableSchema() ? @@ -999,6 +999,9 @@ public Optional newJdbcSchemaFromTableIfExists(JdbcOutputConnection final Set primaryKeys = Collections.unmodifiableSet(primaryKeysBuilder); final ArrayList builder = new ArrayList<>(); + logger.info("table database: {}", table.getDatabase()); + logger.info("table schema: {}", table.getSchemaName()); + logger.info("table name: {}", table.getTableName()); rs = dbm.getColumns( JdbcUtils.escapeSearchString(table.getDatabase(), escape), JdbcUtils.escapeSearchString(table.getSchemaName(), escape), @@ -1007,6 +1010,7 @@ public Optional newJdbcSchemaFromTableIfExists(JdbcOutputConnection try { while (rs.next()) { String columnName = rs.getString("COLUMN_NAME"); + logger.info("column name {}", columnName); String simpleTypeName = rs.getString("TYPE_NAME").toUpperCase(Locale.ENGLISH); boolean isUniqueKey = primaryKeys.contains(columnName); int sqlType = rs.getInt("DATA_TYPE"); diff --git a/embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/JdbcOutputConnection.java b/embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/JdbcOutputConnection.java index 24287eca..e2f70cf1 100644 --- a/embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/JdbcOutputConnection.java +++ b/embulk-output-jdbc/src/main/java/org/embulk/output/jdbc/JdbcOutputConnection.java @@ -61,6 +61,11 @@ public void close() throws SQLException } } + public String getDatabaseName() throws SQLException + { + return connection.getCatalog(); + } + public String getSchemaName() { return schemaName;