fix: change range-partitioning fields to be int

Author: kitagry

README.md

@@ -113,9 +113,9 @@ Following options are same as [bq command-line tools](https://cloud.google.com/b
 |  range_partitioning               | hash     | optional  | nil     | See [Range Partitioning](#range-partitioning) |
 |  range_partitioning.field         | string   | required  | nil     | `INT64` column used for partitioning |
 |  range-partitioning.range         | hash     | required  | nil     | Defines the ranges for range paritioning |
-|  range-partitioning.range.start   | string   | required  | nil     | The start of range partitioning, inclusive. This field is an INT64 value represented as a string. |
-|  range-partitioning.range.end     | string   | required  | nil     | The end of range partitioning, exclusive. This field is an INT64 value represented as a string. |
-|  range-partitioning.range.interval| string   | required  | nil     | The width of each interval. This field is an INT64 value represented as a string. |
+|  range-partitioning.range.start   | int      | required  | nil     | The start of range partitioning, inclusive. |
+|  range-partitioning.range.end     | int      | required  | nil     | The end of range partitioning, exclusive. |
+|  range-partitioning.range.interval| int      | required  | nil     | The width of each interval. |
 |  clustering                       | hash     | optional  | nil     | Currently, clustering is supported for partitioned tables, so must be used with `time_partitioning` option. See [clustered tables](https://cloud.google.com/bigquery/docs/clustered-tables) |
 |  clustering.fields                | array    | required  | nil     | One or more fields on which data should be clustered. The order of the specified columns determines the sort order of the data. |
 |  schema_update_options            | array    | optional  | nil     | (Experimental) List of `ALLOW_FIELD_ADDITION` or `ALLOW_FIELD_RELAXATION` or both. See [jobs#configuration.load.schemaUpdateOptions](https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.load.schemaUpdateOptions). NOTE for the current status: `schema_update_options` does not work for `copy` job, that is, is not effective for most of modes such as `append`, `replace` and `replace_backup`. `delete_in_advance` deletes origin table so does not need to update schema. Only `append_direct` can utilize schema update. |
@@ -467,9 +467,9 @@ out:
   range_partitioning:
     field: customer_id
     range:
-      start: '1'
-      end: '99999'
-      range: '1'
+      start: 1
+      end: 99999
+      range: 1
 ```
 
 ## Development

example/config_replace_field_range_partitioned_table.yml

@@ -31,6 +31,6 @@ out:
   range_partitioning:
     field: 'long'
     range:
-      start: '90'
-      end: '100'
-      interval: '1'
+      start: 90
+      end: 100
+      interval: 1

lib/embulk/output/bigquery.rb

@@ -248,12 +248,24 @@ def self.configure(config, schema, task_count)
           unless task['range_partitioning']['range']['start']
             raise ConfigError.new "`range_partitioning` must have `range.start` key"
           end
+          unless task['range_partitioning']['range']['start'].is_a?(Integer)
+            raise ConfigError.new "`range_partitioning.range.start` must be an integer"
+          end
           unless task['range_partitioning']['range']['end']
             raise ConfigError.new "`range_partitioning` must have `range.end` key"
           end
+          unless task['range_partitioning']['range']['end'].is_a?(Integer)
+            raise ConfigError.new "`range_partitioning.range.end` must be an integer"
+          end
+          if task['range_partitioning']['range']['start'] >= task['range_partitioning']['range']['end']
+            raise ConfigError.new "`range_partitioning.range.start` must be less than `range_partitioning.range.end`"
+          end
           unless task['range_partitioning']['range']['interval']
             raise ConfigError.new "`range_partitioning` must have `range.interval` key"
           end
+          unless task['range_partitioning']['range']['interval'].is_a?(Integer)
+            raise ConfigError.new "`range_partitioning.range.interval` must be an integer"
+          end
         end
 
         if task['clustering']

lib/embulk/output/bigquery/bigquery_client.rb

@@ -440,9 +440,9 @@ def create_table_if_not_exists(table, dataset: nil, options: nil)
               body[:range_partitioning] = {
                 field: options['range_partitioning']['field'],
                 range: {
-                  start: options['range_partitioning']['range']['start'],
-                  end: options['range_partitioning']['range']['end'],
-                  interval: options['range_partitioning']['range']['interval'],
+                  start: options['range_partitioning']['range']['start'].to_s,
+                  end: options['range_partitioning']['range']['end'].to_s,
+                  interval: options['range_partitioning']['range']['interval'].to_s,
                 },
               }
             end

test/test_configure.rb

@@ -271,11 +271,11 @@ def test_time_partitioning
       end
 
       def test_range_partitioning
-        config = least_config.merge('range_partitioning' => {'field' => 'foo', 'range' => { 'start' => '1', 'end' => '2', 'interval' => '1' }})
+        config = least_config.merge('range_partitioning' => {'field' => 'foo', 'range' => { 'start' => 1, 'end' => 2, 'interval' => 1 }})
         assert_nothing_raised { Bigquery.configure(config, schema, processor_count) }
 
         # field is required
-        config = least_config.merge('range_partitioning' => {'range' => { 'start' => '1', 'end' => '2', 'interval' => '1' }})
+        config = least_config.merge('range_partitioning' => {'range' => { 'start' => 1, 'end' => 2, 'interval' => 1 }})
         assert_raise { Bigquery.configure(config, schema, processor_count) }
 
 
@@ -284,15 +284,19 @@ def test_range_partitioning
         assert_raise { Bigquery.configure(config, schema, processor_count) }
 
         # range.start is required
-        config = least_config.merge('range_partitioning' => {'field' => 'foo', 'range' => { 'end' => '2', 'interval' => '1' }})
+        config = least_config.merge('range_partitioning' => {'field' => 'foo', 'range' => { 'end' => 2, 'interval' => 1 }})
         assert_raise { Bigquery.configure(config, schema, processor_count) }
 
         # range.end is required
-        config = least_config.merge('range_partitioning' => {'field' => 'foo', 'range' => { 'start' => '1', 'interval' => '1' }})
+        config = least_config.merge('range_partitioning' => {'field' => 'foo', 'range' => { 'start' => 1, 'interval' => 1 }})
+        assert_raise { Bigquery.configure(config, schema, processor_count) }
+
+        # range.start should be less than range.end
+        config = least_config.merge('range_partitioning' => {'field' => 'foo', 'range' => { 'start' => 2, 'end' => 1, 'interval' => 1 }})
         assert_raise { Bigquery.configure(config, schema, processor_count) }
 
         # range.interval is required
-        config = least_config.merge('range_partitioning' => {'field' => 'foo', 'range' => { 'start' => '1', 'end' => '2' }})
+        config = least_config.merge('range_partitioning' => {'field' => 'foo', 'range' => { 'start' => 1, 'end' => 2 }})
         assert_raise { Bigquery.configure(config, schema, processor_count) }
       end