Support field paritioning

Author: sonots

README.md

@@ -158,6 +158,8 @@ This is not transactional, i.e., if fails, the target table could have some rows
 
 ```is_skip_job_result_check``` must be false when replace mode
 
+NOTE: BigQuery does not support replacing (actually, copying into) a non-partitioned table with a paritioned table atomically. You must once delete the non-partitioned table, otherwise, you get `Incompatible table partitioning specification when copying to the column partitioned table` error.
+
 ##### replace_backup
 
 1. Load to temporary table (Create and WRITE_APPEND in parallel)

example/config_delete_in_advance_field_partitioned_table.yml

@@ -0,0 +1,33 @@
+in:
+  type: file
+  path_prefix: example/example.csv
+  parser:
+    type: csv
+    charset: UTF-8
+    newline: CRLF
+    null_string: 'NULL'
+    skip_header_lines: 1
+    comment_line_marker: '#'
+    columns:
+      - {name: date,        type: string}
+      - {name: timestamp,   type: timestamp, format: "%Y-%m-%d %H:%M:%S.%N", timezone: "+09:00"}
+      - {name: "null",      type: string}
+      - {name: long,        type: long}
+      - {name: string,      type: string}
+      - {name: double,      type: double}
+      - {name: boolean,     type: boolean}
+out:
+  type: bigquery
+  mode: delete_in_advance
+  auth_method: json_key
+  json_keyfile: example/your-project-000.json
+  dataset: your_dataset_name
+  table: your_field_partitioned_table_name
+  source_format: NEWLINE_DELIMITED_JSON
+  compression: NONE
+  auto_create_dataset: true
+  auto_create_table: true
+  schema_file: example/schema.json
+  time_partitioning:
+    type: 'DAY'
+    field: timestamp

example/config_replace_backup_field_partitioned_table.yml

@@ -0,0 +1,34 @@
+in:
+  type: file
+  path_prefix: example/example.csv
+  parser:
+    type: csv
+    charset: UTF-8
+    newline: CRLF
+    null_string: 'NULL'
+    skip_header_lines: 1
+    comment_line_marker: '#'
+    columns:
+      - {name: date,        type: string}
+      - {name: timestamp,   type: timestamp, format: "%Y-%m-%d %H:%M:%S.%N", timezone: "+09:00"}
+      - {name: "null",      type: string}
+      - {name: long,        type: long}
+      - {name: string,      type: string}
+      - {name: double,      type: double}
+      - {name: boolean,     type: boolean}
+out:
+  type: bigquery
+  mode: replace_backup
+  auth_method: json_key
+  json_keyfile: example/your-project-000.json
+  dataset: your_dataset_name
+  table: your_field_partitioned_table_name
+  table_old: your_field_partitioned_table_name_old
+  source_format: NEWLINE_DELIMITED_JSON
+  compression: NONE
+  auto_create_dataset: true
+  auto_create_table: true
+  schema_file: example/schema.json
+  time_partitioning:
+    type: 'DAY'
+    field: 'timestamp'

example/config_replace_backup_paritioned_table.yml renamed to example/config_replace_backup_partitioned_table.yml

@@ -0,0 +1,33 @@
+in:
+  type: file
+  path_prefix: example/example.csv
+  parser:
+    type: csv
+    charset: UTF-8
+    newline: CRLF
+    null_string: 'NULL'
+    skip_header_lines: 1
+    comment_line_marker: '#'
+    columns:
+      - {name: date,        type: string}
+      - {name: timestamp,   type: timestamp, format: "%Y-%m-%d %H:%M:%S.%N", timezone: "+09:00"}
+      - {name: "null",      type: string}
+      - {name: long,        type: long}
+      - {name: string,      type: string}
+      - {name: double,      type: double}
+      - {name: boolean,     type: boolean}
+out:
+  type: bigquery
+  mode: replace
+  auth_method: json_key
+  json_keyfile: example/your-project-000.json
+  dataset: your_dataset_name
+  table: your_field_partitioned_table_name
+  source_format: NEWLINE_DELIMITED_JSON
+  compression: NONE
+  auto_create_dataset: true
+  auto_create_table: true
+  schema_file: example/schema.json
+  time_partitioning:
+    type: 'DAY'
+    field: 'timestamp'

example/config_replace_field_partitioned_table.yml

@@ -306,33 +306,48 @@ def self.auto_create(task, bigquery)
 
         case task['mode']
         when 'delete_in_advance'
-          bigquery.delete_table_or_partition(task['table'])
+          bigquery.delete_partition(task['table'])
           bigquery.create_table_if_not_exists(task['table'])
-        when 'replace', 'replace_backup', 'append'
+        when 'replace'
           bigquery.create_table_if_not_exists(task['temp_table'])
-          if task['time_partitioning']
+          if Helper.has_partition_decorator?(task['table'])
             if task['auto_create_table']
               bigquery.create_table_if_not_exists(task['table'])
             else
               bigquery.get_table(task['table']) # raises NotFoundError
             end
           end
-        else # append_direct
-          if task['auto_create_table']
-            bigquery.create_table_if_not_exists(task['table'])
-          else
-            bigquery.get_table(task['table']) # raises NotFoundError
+        when 'append'
+          bigquery.create_table_if_not_exists(task['temp_table'])
+          if Helper.has_partition_decorator?(task['table'])
+            if task['auto_create_table']
+              bigquery.create_table_if_not_exists(task['table'])
+            else
+              bigquery.get_table(task['table']) # raises NotFoundError
+            end
           end
-        end
-
-        if task['mode'] == 'replace_backup'
-          if task['time_partitioning'] and Helper.has_partition_decorator?(task['table_old'])
+        when 'replace_backup'
+          bigquery.create_table_if_not_exists(task['temp_table'])
+          if Helper.has_partition_decorator?(task['table'])
+            if task['auto_create_table']
+              bigquery.create_table_if_not_exists(task['table'])
+            else
+              bigquery.get_table(task['table']) # raises NotFoundError
+            end
+          end
+          if Helper.has_partition_decorator?(task['table_old'])
             if task['auto_create_table']
               bigquery.create_table_if_not_exists(task['table_old'], dataset: task['dataset_old'])
             else
               bigquery.get_table(task['table_old'], dataset: task['dataset_old']) # raises NotFoundError
             end
           end
+        else # append_direct
+          if task['auto_create_table']
+            bigquery.create_table_if_not_exists(task['table'])
+          else
+            bigquery.get_table(task['table']) # raises NotFoundError
+          end
         end
       end
 

example/config_replace_paritioned_table.yml renamed to example/config_replace_partitioned_table.yml

@@ -470,6 +470,10 @@ def delete_table(table, dataset: nil)
           delete_table_or_partition(table, dataset: dataset)
         end
 
+        def delete_partition(table, dataset: nil)
+          delete_table_or_partition(table, dataset: dataset)
+        end
+
         # if `table` with a partition decorator is given, a partition is deleted.
         def delete_table_or_partition(table, dataset: nil)
           begin

example/example.jsonl

@@ -96,20 +96,20 @@ def test_get_dataset_not_found
           end
         end
 
-        sub_test_case "create_table" do
-          def test_create_table
+        sub_test_case "create_table_if_not_exists" do
+          def test_create_table_if_not_exists
             client.delete_table('your_table_name')
-            assert_nothing_raised { client.create_table('your_table_name') }
+            assert_nothing_raised { client.create_table_if_not_exists('your_table_name') }
           end
 
-          def test_create_table_already_exists
-            assert_nothing_raised { client.create_table('your_table_name') }
+          def test_create_table_if_not_exists_already_exists
+            assert_nothing_raised { client.create_table_if_not_exists('your_table_name') }
           end
 
           def test_create_partitioned_table
             client.delete_table('your_table_name')
             assert_nothing_raised do
-              client.create_table('your_table_name$20160929', options:{
+              client.create_table_if_not_exists('your_table_name$20160929', options:{
                 'time_partitioning' => {'type'=>'DAY', 'expiration_ms'=>1000}
               })
             end
@@ -118,7 +118,7 @@ def test_create_partitioned_table
 
         sub_test_case "delete_table" do
           def test_delete_table
-            client.create_table('your_table_name')
+            client.create_table_if_not_exists('your_table_name')
             assert_nothing_raised { client.delete_table('your_table_name') }
           end
 
@@ -127,14 +127,14 @@ def test_delete_table_not_found
           end
 
           def test_delete_partitioned_table
-            client.create_table('your_table_name')
+            client.create_table_if_not_exists('your_table_name')
             assert_nothing_raised { client.delete_table('your_table_name$20160929') }
           end
         end
 
         sub_test_case "get_table" do
           def test_get_table
-            client.create_table('your_table_name')
+            client.create_table_if_not_exists('your_table_name')
             assert_nothing_raised { client.get_table('your_table_name') }
           end
 
@@ -146,23 +146,23 @@ def test_get_table_not_found
           end
 
           def test_get_partitioned_table
-            client.create_table('your_table_name')
+            client.create_table_if_not_exists('your_table_name')
             assert_nothing_raised { client.get_table('your_table_name$20160929') }
           end
         end
 
         sub_test_case "delete_partition" do
           def test_delete_partition
             client.delete_table('your_table_name')
-            client.create_table('your_table_name$20160929')
+            client.create_table_if_not_exists('your_table_name$20160929')
             assert_nothing_raised { client.delete_partition('your_table_name$20160929') }
           ensure
             client.delete_table('your_table_name')
           end
 
           def test_delete_partition_of_non_partitioned_table
             client.delete_table('your_table_name')
-            client.create_table('your_table_name')
+            client.create_table_if_not_exists('your_table_name')
             assert_raise { client.delete_partition('your_table_name$20160929') }
           ensure
             client.delete_table('your_table_name')
@@ -175,7 +175,7 @@ def test_delete_partition_table_not_found
 
         sub_test_case "fields" do
           def test_fields_from_table
-            client.create_table('your_table_name')
+            client.create_table_if_not_exists('your_table_name')
             fields = client.fields_from_table('your_table_name')
             expected = [
               {:type=>"BOOLEAN", :name=>"boolean"},
@@ -190,15 +190,15 @@ def test_fields_from_table
         end
 
         sub_test_case "copy" do
-          def test_create_table
-            client.create_table('your_table_name')
+          def test_create_table_if_not_exists
+            client.create_table_if_not_exists('your_table_name')
             assert_nothing_raised { client.copy('your_table_name', 'your_table_name_old') }
           end
         end
 
         sub_test_case "load" do
           def test_load
-            client.create_table('your_table_name')
+            client.create_table_if_not_exists('your_table_name')
             File.write("tmp/your_file_name.csv", record.to_csv)
             assert_nothing_raised { client.load("/tmp/your_file_name.csv", 'your_table_name') }
           end