Add an expiration option of temporary table to clean up

Author: takegue

README.md

@@ -79,6 +79,13 @@ Options for intermediate local files
 |  delete_from_local_when_job_end      | boolean     | optional   | true                     | If set to true, delete generate local files when job is end |
 |  compression                         | string      | optional   | "NONE"                   | Compression of local files (`GZIP` or `NONE`) |
 
+
+Options for intermediate tables on BigQuery
+
+| name                                 | type        | required?  | default                  | description            |
+|:-------------------------------------|:------------|:-----------|:-------------------------|:-----------------------|
+|  temporary_table_expiration          | integer     | optional   |                          | Temporary table's expiration time in seconds |
+
 `source_format` is also used to determine formatter (csv or jsonl).
 
 #### Same options of bq command-line tools or BigQuery job's property

lib/embulk/output/bigquery.rb

@@ -90,6 +90,8 @@ def self.configure(config, schema, task_count)
           '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),
 
+          'temporary_table_expiration'     => config.param('temporary_table_expiration',     :integer, :default => nil),
+
           # for debug
           'skip_load'                      => config.param('skip_load',                      :bool,    :default => false),
           'temp_table'                     => config.param('temp_table',                     :string,  :default => nil),
@@ -300,19 +302,23 @@ def self.auto_create(task, bigquery)
           end
         end
 
+        temp_table_expiration = task['temporary_table_expiration']
+        temp_options = {'expiration_time' => temp_table_expiration}
+
         case task['mode']
         when 'delete_in_advance'
           bigquery.delete_table_or_partition(task['table'])
           bigquery.create_table_if_not_exists(task['table'])
         when 'replace'
-          bigquery.create_table_if_not_exists(task['temp_table'])
+          bigquery.create_table_if_not_exists(task['temp_table'], options: temp_options)
           bigquery.create_table_if_not_exists(task['table']) # needs for when task['table'] is a partition
         when 'append'
-          bigquery.create_table_if_not_exists(task['temp_table'])
+          bigquery.create_table_if_not_exists(task['temp_table'], options: temp_options)
           bigquery.create_table_if_not_exists(task['table']) # needs for when task['table'] is a partition
         when 'replace_backup'
-          bigquery.create_table_if_not_exists(task['temp_table'])
+          bigquery.create_table_if_not_exists(task['temp_table'], options: temp_options)
           bigquery.create_table_if_not_exists(task['table'])
+
           bigquery.create_table_if_not_exists(task['table_old'], dataset: task['dataset_old']) # needs for when a partition
         else # append_direct
           if task['auto_create_table']

lib/embulk/output/bigquery/bigquery_client.rb

@@ -440,6 +440,12 @@ def create_table_if_not_exists(table, dataset: nil, options: nil)
               }
             end
 
+            options['expiration_time'] ||= nil
+            if options['expiration_time']
+              # expiration_time is expressed in milliseconds
+              body[:expiration_time] = (Time.now.to_i + options['expiration_time']) * 1000
+            end
+
             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) }