Skip to content

Send time_partitioning_expiration as an integer number of milliseconds - #231

Open
lbarrosby wants to merge 2 commits into
fluent-plugins-nursery:masterfrom
lbarrosby:fix-partition-expiration-integer-ms
Open

lbarrosby wants to merge 2 commits into
fluent-plugins-nursery:masterfrom
lbarrosby:fix-partition-expiration-integer-ms

Conversation

@lbarrosby

Copy link
Copy Markdown

Fixes #220

time_partitioning_expiration is declared as a Fluentd :time parameter, and the writer sends value * 1000 as expiration_ms. A unit value (30d, 720h) is parsed as an Integer, but a plain number (2592000) is parsed as a Float, so expiration_ms became 2592000000.0. The client serializes that int64 field as "2592000000.0" and BigQuery refuses the load job (or insert_table for bigquery_insert with auto_create_table):

Invalid value at 'job.configuration.load.time_partitioning.expiration_ms.value' (TYPE_INT64), "2592000000.0"

This converts the milliseconds with to_i. The README described the parameter as "Expiration milliseconds", which is what led to the report in #220 (users entered milliseconds); it now says it takes a time value and is sent in milliseconds.

Reproduction

<match **>
  @type bigquery_load
  ...
  auto_create_table true
  time_partitioning_type day
  time_partitioning_expiration 2592000
</match>

Request body built by the client for the load job (Job::Representation), master:

{"configuration":{"load":{"timePartitioning":{"expirationMs":"2592000000.0","type":"DAY"}, ...}}}

this branch:

{"configuration":{"load":{"timePartitioning":{"expirationMs":"2592000000","type":"DAY"}, ...}}}

How the parameter is parsed (Fluentd 1.20.1):

2592000  -> 2592000.0 -> expiration_ms 2592000000.0
2592000s -> 2592000   -> expiration_ms 2592000000
30d      -> 2592000   -> expiration_ms 2592000000

Tests

  • New test_write_with_auto_create_table_time_partitioning_expiration_in_seconds in test/plugin/test_out_bigquery_load.rb: captures the job passed to insert_job and serializes it with Google::Apis::BigqueryV2::Job::Representation, asserting "expirationMs" => "2592000000". Fails on master with "2592000000.0".
  • bundle exec rake test: 35 tests, 0 failures (Ruby 4.0.7).

time_partitioning_expiration is a Fluentd time parameter, so a plain
number such as 2592000 is parsed as a Float. Multiplied by 1000 it
became 2592000000.0, which the client serializes as "2592000000.0",
and BigQuery rejects the load job or the table creation with
"Invalid value at 'job.configuration.load.time_partitioning.expiration_ms.value' (TYPE_INT64)".
Values with a unit such as 30d were Integers and worked.

Convert the milliseconds to an Integer, and say in the README that the
parameter takes a time value and is sent in milliseconds.

Signed-off-by: Lucas Barros <lucasbarros.kam@gmail.com>

@Watson1978 Watson1978 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the fix and the clear write-up.

I checked that a plain number such as 2592000 now gives "expirationMs":"2592000000" for both the load job and insert_table with auto_create_table. Values with a unit are unchanged. The new test fails on master as described.

Two requests, with details inline. The new test reaches the real BigQuery API, so it fails without network access. The README should also say that a plain number is in seconds. The NIT on the assertion is optional.

Comment on lines +364 to +366
job = nil
stub_writer do |writer|
mock(writer.client).get_table('yourproject_id', 'yourdataset_id', 'foo') do

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SHOULD: This test sends a real request to BigQuery. fetch_load_job and commit_load_job are not mocked, so the polling timer calls client.get_job against bigquery.googleapis.com. When I ran it with network access, the API returned 401 and the test passed. Without network access, the chunk is rolled back and written again, get_table is called twice, and the test fails with RR::Errors::TimesCalledError before any assertion runs.

Could you mock them in this test the same way test_write does? With this change the test passes offline, and it still fails on master with "2592000000.0".

Suggested change
job = nil
stub_writer do |writer|
mock(writer.client).get_table('yourproject_id', 'yourdataset_id', 'foo') do
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

end

time_partitioning = job[:configuration][:load][:time_partitioning]
assert_equal({type: "DAY", expiration_ms: 2592000000}, time_partitioning)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

NIT: This assertion also passes on master. Hash#== compares the values with ==, and 2592000000 == 2592000000.0 is true. Only the JSON assertion below catches the bug. Checking the class here would make the intent clear.

Suggested change
assert_equal({type: "DAY", expiration_ms: 2592000000}, time_partitioning)
assert_equal({type: "DAY", expiration_ms: 2592000000}, time_partitioning)
assert_kind_of(Integer, time_partitioning[:expiration_ms])

Comment thread README.md Outdated
| 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`). It is sent to BigQuery in milliseconds. |

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SHOULD: #220 started from users entering milliseconds. The new text says the value is sent in milliseconds, but not which unit a plain number like 2592000 uses. Please state that it is seconds, so the same mistake does not happen again.

Suggested change
| time_partitioning_expiration | time | no | no | nil | Expiration of each partition, as a time value (`2592000`, `720h`, `30d`). It is sent to BigQuery in milliseconds. |
| 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. |

A plain time_partitioning_expiration is in seconds. The test now mocks
fetch_load_job and commit_load_job so it runs without network access,
and checks that expiration_ms is an Integer.

Signed-off-by: Lucas Barros <lucasbarros.kam@gmail.com>
@lbarrosby

Copy link
Copy Markdown
Author

Thanks for checking. Pushed f02fa82:

The new test now mocks fetch_load_job and commit_load_job like test_write, so it no longer reaches BigQuery. I ran it with the network blocked through a dead proxy and it passes. I also added the assert_kind_of(Integer, ...) check, and the README now says a plain number is in seconds.

bundle exec rake test: 35 tests, 0 failures.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

time_partitioning_expiration should be provided in days

2 participants