Merge pull request #11 from yas-okadatech/27783-catalog

fix: listing table and columns were executed with null category because it specify database name instead of category name.

src/main/java/org/embulk/output/DatabricksOutputPlugin.java

@@ -180,11 +180,12 @@ public Optional<JdbcSchema> newJdbcSchemaFromTableIfExists(
       return Optional.empty();
     }
 
+    DatabricksOutputConnection conn = (DatabricksOutputConnection) connection;
     DatabaseMetaData dbm = connection.getMetaData();
     String escape = dbm.getSearchStringEscape();
 
     ResultSet rs =
-        dbm.getPrimaryKeys(table.getDatabase(), table.getSchemaName(), table.getTableName());
+        dbm.getPrimaryKeys(conn.getCatalogName(), table.getSchemaName(), table.getTableName());
     final HashSet<String> primaryKeysBuilder = new HashSet<>();
     try {
       while (rs.next()) {
@@ -207,7 +208,7 @@ public Optional<JdbcSchema> newJdbcSchemaFromTableIfExists(
     // https://docs.databricks.com/en/sql/language-manual/data-types/timestamp-ntz-type.html#notes
     rs =
         dbm.getColumns(
-            JdbcUtils.escapeSearchString(table.getDatabase(), escape),
+            JdbcUtils.escapeSearchString(conn.getCatalogName(), escape),
             JdbcUtils.escapeSearchString(table.getSchemaName(), escape),
             JdbcUtils.escapeSearchString(table.getTableName(), escape),
             null);

src/main/java/org/embulk/output/databricks/DatabricksOutputConnection.java

@@ -45,7 +45,7 @@ public boolean tableExists(TableIdentifier table) throws SQLException {
     try (ResultSet rs =
         connection
             .getMetaData()
-            .getTables(table.getDatabase(), table.getSchemaName(), table.getTableName(), null)) {
+            .getTables(catalogName, table.getSchemaName(), table.getTableName(), null)) {
       while (rs.next()) {
         if (isAvailableTableMetadataInConnection(rs, table)) {
           return true;
@@ -62,6 +62,10 @@ public boolean isAvailableTableMetadataInConnection(ResultSet rs, TableIdentifie
     // null
     // and one Databricks connection has multiple available catalogs　(databases).
 
+    // NOTE: maybe this logic is not necessary anymore after this PR:
+    //    https://github.com/trocco-io/embulk-output-databricks/pull/11
+    // But I'm not sure, so I'll keep it for now.
+
     if (tableIdentifier.getDatabase() == null) {
       logger.trace("tableIdentifier.getDatabase() == null, check by instance variable");
       if (!rs.getString("TABLE_CAT").equalsIgnoreCase(catalogName)) {
@@ -295,4 +299,8 @@ private String buildColumns(JdbcSchema schema, String prefix) {
     }
     return sb.toString();
   }
+
+  public String getCatalogName() {
+    return catalogName;
+  }
 }