Merge pull request #7 from giwa/update-description

Author: giwa

embulk-output-bigquery.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |spec|
   spec.name          = "embulk-output-bigquery"
-  spec.version       = "0.6.5"
+  spec.version       = "0.6.6"
   spec.authors       = ["Satoshi Akama", "Naotoshi Seo"]
   spec.summary       = "Google BigQuery output plugin for Embulk"
   spec.description   = "Embulk plugin that insert records to Google BigQuery."

lib/embulk/output/bigquery.rb

@@ -398,6 +398,12 @@ def self.transaction(config, schema, task_count, &control)
               end
             end
           end
+
+          if task['mode'] == 'append' || task['mode'] == 'append_direct'
+            # update only column.description based on column_options
+            bigquery.patch_table
+          end
+
         ensure
           begin
             if task['temp_table'] # append or replace or replace_backup

lib/embulk/output/bigquery/bigquery_client.rb

@@ -511,6 +511,29 @@ def get_table_or_partition(table, dataset: nil)
             raise Error, "failed to get table #{@project}:#{dataset}.#{table}, response:#{response}"
           end
         end
+
+        # update only column.description
+        def patch_table
+          table = get_table(@task['table'])
+
+          def patch_description(fields, column_options)
+            fields.map do |field|
+              column_option = column_options.select{|col_opt| col_opt['name'] == field.name}.first
+              if column_option
+                field.update!(description: column_option['description']) if column_option['description']
+                if field.fields && column_option['fields']
+                  nested_fields = patch_description(field.fields, column_option['fields'])
+                  field.update!(fields: nested_fields)
+                end
+              end
+              field
+            end
+          end
+
+          fields = patch_description(table.schema.fields, @task['column_options'])
+          table.schema.update!(fields: fields)
+          client.patch_table(@project, @dataset, @task['table'], table)
+        end
       end
     end
   end

test/test_transaction.rb

@@ -45,6 +45,7 @@ def test_append_direc_without_auto_create
           any_instance_of(BigqueryClient) do |obj|
             mock(obj).get_dataset(config['dataset'])
             mock(obj).get_table(config['table'])
+            mock(obj).patch_table
           end
           Bigquery.transaction(config, schema, processor_count, &control)
         end
@@ -55,6 +56,7 @@ def test_append_direct_with_auto_create
           any_instance_of(BigqueryClient) do |obj|
             mock(obj).create_dataset(config['dataset'])
             mock(obj).create_table_if_not_exists(config['table'])
+            mock(obj).patch_table
           end
           Bigquery.transaction(config, schema, processor_count, &control)
         end
@@ -64,6 +66,7 @@ def test_append_direct_with_partition_without_auto_create
           any_instance_of(BigqueryClient) do |obj|
             mock(obj).get_dataset(config['dataset'])
             mock(obj).get_table(config['table'])
+            mock(obj).patch_table
           end
           Bigquery.transaction(config, schema, processor_count, &control)
         end
@@ -74,6 +77,7 @@ def test_append_direct_with_partition_with_auto_create
           any_instance_of(BigqueryClient) do |obj|
             mock(obj).create_dataset(config['dataset'])
             mock(obj).create_table_if_not_exists(config['table'])
+            mock(obj).patch_table
           end
           Bigquery.transaction(config, schema, processor_count, &control)
         end
@@ -200,6 +204,7 @@ def test_append
             mock(obj).create_table_if_not_exists(config['table'])
             mock(obj).copy(config['temp_table'], config['table'], write_disposition: 'WRITE_APPEND')
             mock(obj).delete_table(config['temp_table'])
+            mock(obj).patch_table
           end
           Bigquery.transaction(config, schema, processor_count, &control)
         end
@@ -213,6 +218,7 @@ def test_append_with_partitioning
             mock(obj).create_table_if_not_exists(config['table'])
             mock(obj).copy(config['temp_table'], config['table'], write_disposition: 'WRITE_APPEND')
             mock(obj).delete_table(config['temp_table'])
+            mock(obj).patch_table
           end
           Bigquery.transaction(config, schema, processor_count, &control)
         end