feat: add get database name

Author: hi-kbys

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<JdbcSchema> initialTargetTableSchema =
             mode.ignoreTargetTableSchema() ?
@@ -999,6 +999,9 @@ public Optional<JdbcSchema> newJdbcSchemaFromTableIfExists(JdbcOutputConnection
         final Set<String> primaryKeys = Collections.unmodifiableSet(primaryKeysBuilder);
 
         final ArrayList<JdbcColumn> 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<JdbcSchema> 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");

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;