From 9aa96acb3a53014f52ae1e72bf3203cf9e5301dd Mon Sep 17 00:00:00 2001
From: hi-kbys <hiroyuki.kobayashi@estie.co.jp>
Date: Tue, 4 Jun 2024 14:04:47 +0900
Subject: [PATCH] feat: add get database name

---
 .../org/embulk/output/jdbc/AbstractJdbcOutputPlugin.java    | 6 +++++-
 .../java/org/embulk/output/jdbc/JdbcOutputConnection.java   | 5 +++++
 2 files changed, 10 insertions(+), 1 deletion(-)

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<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");
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;