Send time_partitioning_expiration as an integer number of milliseconds

README.md

@@ -60,7 +60,7 @@ Because embedded gem dependency sometimes restricts ruby environment.
 | request_open_timeout_sec                      | integer       | no                                           | no           | 60                         | Bigquery API connection, and request timeout. If you send big data to Bigquery, set large value.       |
 | time_partitioning_type                        | enum          | no (either day or hour)                      | no           | nil                        | Type of bigquery time partitioning feature.                                                            |
 | time_partitioning_field                       | string        | no                                           | no           | nil                        | Field used to determine how to create a time-based partition.                                          |
-| time_partitioning_expiration                  | time          | no                                           | no           | nil                        | Expiration milliseconds for bigquery time partitioning.                                                |
+| time_partitioning_expiration                  | time          | no                                           | no           | nil                        | Expiration of each partition, as a time value (`2592000`, `720h`, `30d`). A plain number is in seconds. It is sent to BigQuery in milliseconds. |
 | clustering_fields                             | array(string) | no                                           | no           | nil                        | One or more fields on which data should be clustered. The order of the specified columns determines the sort order of the data. |
 
 #### bigquery_insert

lib/fluent/plugin/bigquery/writer.rb

@@ -313,7 +313,7 @@ def time_partitioning
           @time_partitioning = {
             type: @options[:time_partitioning_type].to_s.upcase,
             field: @options[:time_partitioning_field] ? @options[:time_partitioning_field].to_s : nil,
-            expiration_ms: @options[:time_partitioning_expiration] ? @options[:time_partitioning_expiration] * 1000 : nil,
+            expiration_ms: @options[:time_partitioning_expiration] ? (@options[:time_partitioning_expiration] * 1000).to_i : nil,
           }.reject { |_, v| v.nil? }
         else
           @time_partitioning

test/plugin/test_out_bigquery_load.rb

@@ -342,6 +342,55 @@ def test_write_with_auto_create_table
     end
   end
 
+  def test_write_with_auto_create_table_time_partitioning_expiration_in_seconds
+    driver = create_driver(<<-CONFIG)
+      table foo
+      email foo@bar.example
+      private_key_path /path/to/key
+      project yourproject_id
+      dataset yourdataset_id
+
+      <buffer>
+        @type memory
+      </buffer>
+
+      auto_create_table true
+      schema_path #{SCHEMA_PATH}
+
+      time_partitioning_type day
+      time_partitioning_expiration 2592000
+    CONFIG
+
+    job = nil
+    response_stub = stub!
+    stub_writer do |writer|
+      mock(writer).fetch_load_job(is_a(Fluent::BigQuery::Writer::JobReference)) { response_stub }
+      mock(writer).commit_load_job(is_a(String), response_stub)
+
+      mock(writer.client).get_table('yourproject_id', 'yourdataset_id', 'foo') do
+        raise Google::Apis::ClientError.new("notFound: Not found: Table yourproject_id:yourdataset_id.foo", status_code: 404)
+      end
+
+      mock(writer.client).insert_job('yourproject_id', anything, upload_source: duck_type(:write, :sync, :rewind), content_type: "application/octet-stream") do |_, configuration, _|
+        job = configuration
+        stub!.job_reference.stub! do |s|
+          s.job_id { "dummy_job_id" }
+          s.location { "us" }
+        end
+      end
+    end
+
+    driver.run do
+      driver.feed("tag", Time.now.to_i, {"a" => "b"})
+    end
+
+    time_partitioning = job[:configuration][:load][:time_partitioning]
+    assert_equal({type: "DAY", expiration_ms: 2592000000}, time_partitioning)
+    assert_kind_of(Integer, time_partitioning[:expiration_ms])
+    json = Google::Apis::BigqueryV2::Job::Representation.new(job).to_json(user_options: { skip_undefined: true })
+    assert_equal({"type" => "DAY", "expirationMs" => "2592000000"}, JSON.parse(json)["configuration"]["load"]["timePartitioning"])
+  end
+
   private
 
   def create_response_stub(response)