time partition

lib/embulk/output/bigquery.rb

@@ -63,8 +63,6 @@ def self.configure(config, schema, task_count)
           'payload_column'                 => config.param('payload_column',                 :string,  :default => nil),
           'payload_column_index'           => config.param('payload_column_index',           :integer, :default => nil),
 
-          'description'                    => config.param('description',                    :string,  :default => nil),
-
           'open_timeout_sec'               => config.param('open_timeout_sec',               :integer, :default => nil),
           'timeout_sec'                    => config.param('timeout_sec',                    :integer, :default => nil), # google-api-ruby-client < v0.11.0
           'send_timeout_sec'               => config.param('send_timeout_sec',               :integer, :default => nil), # google-api-ruby-client >= v0.11.0
@@ -89,6 +87,7 @@ def self.configure(config, schema, task_count)
           'ignore_unknown_values'          => config.param('ignore_unknown_values',          :bool,    :default => false),
           'allow_quoted_newlines'          => config.param('allow_quoted_newlines',          :bool,    :default => false),
           'time_partitioning'              => config.param('time_partitioning',              :hash,    :default => nil),
+          'range_partitioning'              => config.param('range_partitioning',            :hash,    :default => nil),
           'clustering'                     => config.param('clustering',                     :hash,    :default => nil), # google-api-ruby-client >= v0.21.0
           'schema_update_options'          => config.param('schema_update_options',          :array,   :default => nil),
 

lib/embulk/output/bigquery/bigquery_client.rb

@@ -121,7 +121,7 @@ def load_from_gcs(object_uris, table)
               opts = {}
 
               Embulk.logger.debug { "embulk-output-bigquery: insert_job(#{@project}, #{body}, #{opts})" }
-              response = with_network_retry { client.insert_job(@project, body, **opts) }
+              response = with_network_retry { client.insert_job(@project, body, opts) }
               unless @task['is_skip_job_result_check']
                 response = wait_load('Load', response)
               end
@@ -222,7 +222,7 @@ def load(path, table, write_disposition: 'WRITE_APPEND')
                 # },
               }
               Embulk.logger.debug { "embulk-output-bigquery: insert_job(#{@project}, #{body}, #{opts})" }
-              response = with_network_retry { client.insert_job(@project, body, **opts) }
+              response = with_network_retry { client.insert_job(@project, body, opts) }
               if @task['is_skip_job_result_check']
                 response
               else
@@ -278,7 +278,7 @@ def copy(source_table, destination_table, destination_dataset = nil, write_dispo
 
               opts = {}
               Embulk.logger.debug { "embulk-output-bigquery: insert_job(#{@project}, #{body}, #{opts})" }
-              response = with_network_retry { client.insert_job(@project, body, **opts) }
+              response = with_network_retry { client.insert_job(@project, body, opts) }
               wait_load('Copy', response)
             rescue Google::Apis::ServerError, Google::Apis::ClientError, Google::Apis::AuthorizationError => e
               response = {status_code: e.status_code, message: e.message, error_class: e.class}
@@ -372,7 +372,7 @@ def create_dataset(dataset = nil, reference: nil)
             end
             opts = {}
             Embulk.logger.debug { "embulk-output-bigquery: insert_dataset(#{@project}, #{dataset}, #{@location_for_log}, #{body}, #{opts})" }
-            with_network_retry { client.insert_dataset(@project, body, **opts) }
+            with_network_retry { client.insert_dataset(@project, body, opts) }
           rescue Google::Apis::ServerError, Google::Apis::ClientError, Google::Apis::AuthorizationError => e
             if e.status_code == 409 && /Already Exists:/ =~ e.message
               # ignore 'Already Exists' error
@@ -410,6 +410,7 @@ def create_table_if_not_exists(table, dataset: nil, options: nil)
             dataset ||= @dataset
             options ||= {}
             options['time_partitioning'] ||= @task['time_partitioning']
+            options['range_partitioning'] ||= @task['range_partitioning']
             if Helper.has_partition_decorator?(table)
               options['time_partitioning'] ||= {'type' => 'DAY'}
               table = Helper.chomp_partition_decorator(table)
@@ -420,7 +421,6 @@ def create_table_if_not_exists(table, dataset: nil, options: nil)
               table_reference: {
                 table_id: table,
               },
-              description: @task['description'],
               schema: {
                 fields: fields,
               }
@@ -434,6 +434,17 @@ def create_table_if_not_exists(table, dataset: nil, options: nil)
               }
             end
 
+            if options['range_partitioning']
+              body[:range_partitioning] = {
+                field: options['range_partitioning']['field'],
+              }
+              body[:range_partitioning][:range] = {
+                start: options['range_partitioning']['range']['start'],
+                end: options['range_partitioning']['range']['end'],
+                interval: options['range_partitioning']['range']['interval'],
+              }
+            end
+
             options['clustering'] ||= @task['clustering']
             if options['clustering']
               body[:clustering] = {
@@ -447,8 +458,8 @@ def create_table_if_not_exists(table, dataset: nil, options: nil)
             end
 
             opts = {}
-            Embulk.logger.debug { "embulk-output-bigquery: insert_table(#{@destination_project}, #{dataset}, #{@location_for_log}, #{body}, #{opts})" }
-            with_network_retry { client.insert_table(@destination_project, dataset, body, **opts) }
+            Embulk.logger.debug { "embulk-output-bigquery: insert_table(#{@project}, #{dataset}, #{@location_for_log}, #{body}, #{opts})" }
+            with_network_retry { client.insert_table(@project, dataset, body, opts) }
           rescue Google::Apis::ServerError, Google::Apis::ClientError, Google::Apis::AuthorizationError => e
             if e.status_code == 409 && /Already Exists:/ =~ e.message
               # ignore 'Already Exists' error
@@ -457,7 +468,7 @@ def create_table_if_not_exists(table, dataset: nil, options: nil)
 
             response = {status_code: e.status_code, message: e.message, error_class: e.class}
             Embulk.logger.error {
-              "embulk-output-bigquery: insert_table(#{@destination_project}, #{dataset}, #{@location_for_log}, #{body}, #{opts}), response:#{response}"
+              "embulk-output-bigquery: insert_table(#{@project}, #{dataset}, #{@location_for_log}, #{body}, #{opts}), response:#{response}"
             }
             raise Error, "failed to create table #{@destination_project}:#{dataset}.#{table} in #{@location_for_log}, response:#{response}"
           end