add retain_column_descriptions option

README.md

@@ -61,6 +61,7 @@ Under construction
 |  auto_create_gcs_bucket (x)              | boolean     | optional   | false                    | See [GCS Bucket](#gcs-bucket) |
 |  progress_log_interval  (x)            | float       | optional   | nil (Disabled)           | Progress log interval. The progress log is disabled by nil (default). NOTE: This option may be removed in a future because a filter plugin can achieve the same goal |
 |  before_load          | string       | optional   | nil            |  if set, this SQL will be executed before loading all records in append mode. In replace mode, SQL is not executed. |
+|  retain_column_descriptions          | boolean     | optional   | false                    | In case of replace mode, the column's descriptions are taken over. |
 
 
 Client or request options

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

@@ -9,13 +9,15 @@
 import com.google.cloud.bigquery.Dataset;
 import com.google.cloud.bigquery.DatasetInfo;
 import com.google.cloud.bigquery.Field;
+import com.google.cloud.bigquery.FieldList;
 import com.google.cloud.bigquery.FieldValue;
 import com.google.cloud.bigquery.FormatOptions;
 import com.google.cloud.bigquery.Job;
 import com.google.cloud.bigquery.JobId;
 import com.google.cloud.bigquery.JobInfo;
 import com.google.cloud.bigquery.JobStatistics;
 import com.google.cloud.bigquery.LegacySQLTypeName;
+import com.google.cloud.bigquery.PolicyTags;
 import com.google.cloud.bigquery.QueryJobConfiguration;
 import com.google.cloud.bigquery.StandardSQLTypeName;
 import com.google.cloud.bigquery.StandardTableDefinition;
@@ -75,15 +77,16 @@ public class BigqueryClient {
     private PluginTask task;
     private Schema schema;
     private List<BigqueryColumnOption> columnOptions;
+    private Table srcTable;
 
     public BigqueryClient(PluginTask task, Schema schema) {
         this.task = task;
         this.schema = schema;
         this.dataset = task.getDataset();
-        if (task.getLocation().isPresent()){
+        if (task.getLocation().isPresent()) {
             this.location = task.getLocation().get();
             this.locationForLog = task.getLocation().get();
-        }else{
+        } else {
             this.locationForLog = "us/eu";
         }
         this.columnOptions = this.task.getColumnOptions().orElse(Collections.emptyList());
@@ -92,6 +95,7 @@ public BigqueryClient(PluginTask task, Schema schema) {
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
+        this.srcTable = (task.getMode().equals("replace") && task.getRetainColumnDescriptions()) ? getTable(task.getTable()) : null;
     }
 
     private static BigQuery getClientWithJsonKey(String key) throws IOException {
@@ -616,6 +620,15 @@ protected com.google.cloud.bigquery.Schema buildSchema(Schema schema, List<Bigqu
             Optional<BigqueryColumnOption> columnOption = BigqueryUtil.findColumnOption(col.getName(), columnOptions);
             Field.Builder fieldBuilder = createFieldBuilder(task, col, columnOption);
 
+            if(srcTable != null && task.getRetainColumnDescriptions()){
+                FieldList srcFields = srcTable.getDefinition().getSchema().getFields();
+                if(srcFields != null) {
+                    srcFields.stream().filter(x -> x.getName().equals(col.getName()))
+                            .findFirst()
+                            .ifPresent(field -> fieldBuilder.setDescription(field.getDescription()));
+                }
+            }
+
             if (columnOption.isPresent()) {
                 BigqueryColumnOption colOpt = columnOption.get();
                 if (!colOpt.getMode().isEmpty()) {

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

@@ -162,4 +162,8 @@ public interface PluginTask
     @Config("merge_rule")
     @ConfigDefault("null")
     Optional<List<String>> getMergeRule();
+
+    @Config("retain_column_descriptions")
+    @ConfigDefault("false")
+    Boolean getRetainColumnDescriptions();
 }