Merge pull request #65 from trocco-io/fix-misc

fix misc

Author: giwa

src/main/java/org/embulk/output/bigquery_java/BigqueryJavaOutputPlugin.java

@@ -153,26 +153,24 @@ public TransactionalPageOutput open(TaskSource taskSource, Schema schema, int ta
     }
 
     protected void autoCreate(PluginTask task, BigqueryClient client){
-        if (task.getAutoCreateDataset()){
-            client.createDataset(task.getDataset());
-        }else{
-            Dataset dataset = client.getDataset(task.getDataset());
-            if (dataset == null){
-                throw new BigqueryException(String.format("dataset %s is not found", task.getDataset()));
-            }
-        }
-
-        if (task.getMode().equals("replace_backup") && !task.getOldDataset().get().equals(task.getDataset())){
+        if (client.getDataset(task.getDataset()) == null){
             if (task.getAutoCreateDataset()){
                 client.createDataset(task.getDataset());
             }else{
-                Dataset dataset = client.getDataset(task.getOldDataset().get());
-                if (dataset == null){
-                    throw new BigqueryException(String.format("dataset %s is not found", task.getDataset()));
-                }
+                throw new BigqueryException(String.format("dataset %s is not found", task.getDataset()));
             }
         }
 
+        if (task.getMode().equals("replace_backup") && task.getOldDataset().isPresent() && !task.getOldDataset().get().equals(task.getDataset())){
+           if (client.getDataset(task.getOldDataset().get()) == null){
+               if (task.getAutoCreateDataset()){
+                   client.createDataset(task.getDataset());
+               }else{
+                   throw new BigqueryException(String.format("dataset %s is not found", task.getDataset()));
+               }
+           }
+        }
+
         switch (task.getMode()){
             case "delete_in_advance":
                 client.deleteTableOrPartition(task.getTable());

src/main/java/org/embulk/output/bigquery_java/BigqueryJobRunner.java

@@ -24,9 +24,16 @@ public BigqueryJobRunner(PluginTask task, Schema schema, Path path) {
     @Override
     public JobStatistics call() throws Exception {
         client = new BigqueryClient(this.task, this.schema);
+        String tableName;
+        // append_direct use table name
+        if (task.getMode().equals("append_direct")){
+            tableName = task.getTable();
+        }else{
+            tableName = task.getTempTable().get();
+        }
 
         return client.load(this.path,
-                this.task.getTempTable().get(),
+                tableName,
                 JobInfo.WriteDisposition.WRITE_APPEND);
     }
 }

src/main/java/org/embulk/output/bigquery_java/config/BigqueryTaskBuilder.java

@@ -50,12 +50,10 @@ protected static void setFileExt(PluginTask task) {
     @VisibleForTesting
     protected static void setTempTable(PluginTask task) {
         // TODO: support replace_backup
-        String[] modeForTempTable = {"replace", "append"};
+        String[] modeForTempTable = {"replace", "append", "delete_in_advance"};
         if (Arrays.asList(modeForTempTable).contains(task.getMode())) {
             String tempTable = task.getTempTable().orElse(String.format("LOAD_TEMP_%s_%s", uniqueName, task.getTable()));
             task.setTempTable(Optional.of(tempTable));
-        } else {
-            task.setTempTable(Optional.of(null));
         }
     }