Skip to content

Commit be3bc61

Browse files
sonotssonots
committed
Breaking changes:
* Change default value of `auto_create_table` to `true` from `false`. * Drop supporting `auto_create_table: false` for mode: `append`, `replace`, `replace_backup`, `delete_in_advance` which could be used for specifying a partitioned table with a partition decorator on behalf of simplicity.
1 parent 90a83fc commit be3bc61

File tree

3 files changed

+29
-94
lines changed

3 files changed

+29
-94
lines changed

‎README.md

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ v0.3.x has incompatibility changes with v0.2.x. Please see [CHANGELOG.md](CHANGE
4747
| location | string | optional | nil | geographic location of dataset. See [Location](#location) |
4848
| table | string | required | | table name, or table name with a partition decorator such as `table_name$20160929`|
4949
| auto_create_dataset | boolean | optional | false | automatically create dataset |
50-
| auto_create_table | boolean | optional | false | See [Dynamic Table Creating](#dynamic-table-creating) and [Time Partitioning](#time-partitioning) |
50+
| auto_create_table | boolean | optional | true | `false` is available only for `append_direct` mode. Other modes requires `true`. See [Dynamic Table Creating](#dynamic-table-creating) and [Time Partitioning](#time-partitioning) |
5151
| schema_file | string | optional | | /path/to/schema.json |
5252
| template_table | string | optional | | template table name. See [Dynamic Table Creating](#dynamic-table-creating) |
5353
| job_status_max_polling_time | int | optional | 3600 sec | Max job status polling time |
@@ -250,11 +250,6 @@ out:
250250

251251
### Dynamic table creating
252252

253-
This plugin tries to create a table using BigQuery API when
254-
255-
* mode is either of `delete_in_advance`, `replace`, `replace_backup`, `append`.
256-
* mode is `append_direct` and `auto_create_table` is true.
257-
258253
There are 3 ways to set schema.
259254

260255
#### Set schema.json
@@ -383,32 +378,31 @@ To load into a partition, specify `table` parameter with a partition decorator a
383378
out:
384379
type: bigquery
385380
table: table_name$20160929
386-
auto_create_table: true
387381
```
388382

389-
You may configure `time_partitioning` parameter together to create table via `auto_create_table: true` option as:
383+
You may configure `time_partitioning` parameter together as:
390384

391385
```yaml
392386
out:
393387
type: bigquery
394388
table: table_name$20160929
395-
auto_create_table: true
396389
time_partitioning:
397390
type: DAY
398391
expiration_ms: 259200000
399392
```
400393

401394
You can also create column-based partitioning table as:
395+
402396
```yaml
403397
out:
404398
type: bigquery
405399
mode: replace
406-
auto_create_table: true
407400
table: table_name
408401
time_partitioning:
409402
type: DAY
410403
field: timestamp
411404
```
405+
412406
Note the `time_partitioning.field` should be top-level `DATE` or `TIMESTAMP`.
413407

414408
Use [Tables: patch](https://cloud.google.com/bigquery/docs/reference/v2/tables/patch) API to update the schema of the partitioned table, embulk-output-bigquery itself does not support it, though.

‎lib/embulk/output/bigquery.rb

Lines changed: 10 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ def self.configure(config, schema, task_count)
4545
'table_old' => config.param('table_old', :string, :default => nil),
4646
'table_name_old' => config.param('table_name_old', :string, :default => nil), # lower version compatibility
4747
'auto_create_dataset' => config.param('auto_create_dataset', :bool, :default => false),
48-
'auto_create_table' => config.param('auto_create_table', :bool, :default => false),
48+
'auto_create_table' => config.param('auto_create_table', :bool, :default => true),
4949
'schema_file' => config.param('schema_file', :string, :default => nil),
5050
'template_table' => config.param('template_table', :string, :default => nil),
5151

@@ -104,10 +104,14 @@ def self.configure(config, schema, task_count)
104104
raise ConfigError.new "`mode` must be one of append, append_direct, replace, delete_in_advance, replace_backup"
105105
end
106106

107+
if %w[append replace delete_in_advance replace_backup].include?(task['mode']) and !task['auto_create_table']
108+
raise ConfigError.new "`mode: #{task['mode']}` requires `auto_create_table: true`"
109+
end
110+
107111
if task['mode'] == 'replace_backup'
108112
task['table_old'] ||= task['table_name_old'] # for lower version compatibility
109113
if task['dataset_old'].nil? and task['table_old'].nil?
110-
raise ConfigError.new "`mode replace_backup` requires either of `dataset_old` or `table_old`"
114+
raise ConfigError.new "`mode: replace_backup` requires either of `dataset_old` or `table_old`"
111115
end
112116
task['dataset_old'] ||= task['dataset']
113117
task['table_old'] ||= task['table']
@@ -309,38 +313,14 @@ def self.auto_create(task, bigquery)
309313
bigquery.create_table_if_not_exists(task['table'])
310314
when 'replace'
311315
bigquery.create_table_if_not_exists(task['temp_table'])
312-
if Helper.has_partition_decorator?(task['table'])
313-
if task['auto_create_table']
314-
bigquery.create_table_if_not_exists(task['table'])
315-
else
316-
bigquery.get_table(task['table']) # raises NotFoundError
317-
end
318-
end
316+
bigquery.create_table_if_not_exists(task['table'])
319317
when 'append'
320318
bigquery.create_table_if_not_exists(task['temp_table'])
321-
if Helper.has_partition_decorator?(task['table'])
322-
if task['auto_create_table']
323-
bigquery.create_table_if_not_exists(task['table'])
324-
else
325-
bigquery.get_table(task['table']) # raises NotFoundError
326-
end
327-
end
319+
bigquery.create_table_if_not_exists(task['table'])
328320
when 'replace_backup'
329321
bigquery.create_table_if_not_exists(task['temp_table'])
330-
if Helper.has_partition_decorator?(task['table'])
331-
if task['auto_create_table']
332-
bigquery.create_table_if_not_exists(task['table'])
333-
else
334-
bigquery.get_table(task['table']) # raises NotFoundError
335-
end
336-
end
337-
if Helper.has_partition_decorator?(task['table_old'])
338-
if task['auto_create_table']
339-
bigquery.create_table_if_not_exists(task['table_old'], dataset: task['dataset_old'])
340-
else
341-
bigquery.get_table(task['table_old'], dataset: task['dataset_old']) # raises NotFoundError
342-
end
343-
end
322+
bigquery.create_table_if_not_exists(task['table'])
323+
bigquery.create_table_if_not_exists(task['table_old'], dataset: task['dataset_old'])
344324
else # append_direct
345325
if task['auto_create_table']
346326
bigquery.create_table_if_not_exists(task['table'])

‎test/test_transaction.rb

Lines changed: 15 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -41,8 +41,8 @@ def setup
4141
end
4242

4343
sub_test_case "append_direct" do
44-
def test_append_direct
45-
config = least_config.merge('mode' => 'append_direct')
44+
def test_append_direc_without_auto_create
45+
config = least_config.merge('mode' => 'append_direct', 'auto_create_dataset' => false, 'auto_create_table' => false)
4646
any_instance_of(BigqueryClient) do |obj|
4747
mock(obj).get_dataset(config['dataset'])
4848
mock(obj).get_table(config['table'])
@@ -60,8 +60,8 @@ def test_append_direct_with_auto_create
6060
Bigquery.transaction(config, schema, processor_count, &control)
6161
end
6262

63-
def test_append_direct_with_partition
64-
config = least_config.merge('mode' => 'append_direct', 'table' => 'table$20160929')
63+
def test_append_direct_with_partition_without_auto_create
64+
config = least_config.merge('mode' => 'append_direct', 'table' => 'table$20160929', 'auto_create_dataset' => false, 'auto_create_table' => false)
6565
any_instance_of(BigqueryClient) do |obj|
6666
mock(obj).get_dataset(config['dataset'])
6767
mock(obj).get_table(config['table'])
@@ -111,6 +111,7 @@ def test_replace
111111
any_instance_of(BigqueryClient) do |obj|
112112
mock(obj).get_dataset(config['dataset'])
113113
mock(obj).create_table_if_not_exists(config['temp_table'])
114+
mock(obj).create_table_if_not_exists(config['table'])
114115
mock(obj).copy(config['temp_table'], config['table'], write_disposition: 'WRITE_TRUNCATE')
115116
mock(obj).delete_table(config['temp_table'])
116117
end
@@ -120,19 +121,6 @@ def test_replace
120121
def test_replace_with_partitioning
121122
config = least_config.merge('mode' => 'replace', 'table' => 'table$20160929')
122123
task = Bigquery.configure(config, schema, processor_count)
123-
any_instance_of(BigqueryClient) do |obj|
124-
mock(obj).get_dataset(config['dataset'])
125-
mock(obj).create_table_if_not_exists(config['temp_table'])
126-
mock(obj).get_table(config['table'])
127-
mock(obj).copy(config['temp_table'], config['table'], write_disposition: 'WRITE_TRUNCATE')
128-
mock(obj).delete_table(config['temp_table'])
129-
end
130-
Bigquery.transaction(config, schema, processor_count, &control)
131-
end
132-
133-
def test_replace_with_partitioning_with_auto_create_table
134-
config = least_config.merge('mode' => 'replace', 'table' => 'table$20160929', 'auto_create_table' => true)
135-
task = Bigquery.configure(config, schema, processor_count)
136124
any_instance_of(BigqueryClient) do |obj|
137125
mock(obj).get_dataset(config['dataset'])
138126
mock(obj).create_table_if_not_exists(config['temp_table'])
@@ -152,8 +140,10 @@ def test_replace_backup
152140
mock(obj).get_dataset(config['dataset'])
153141
mock(obj).get_dataset(config['dataset_old'])
154142
mock(obj).create_table_if_not_exists(config['temp_table'])
143+
mock(obj).create_table_if_not_exists(config['table'])
144+
mock(obj).create_table_if_not_exists(config['table_old'], dataset: config['dataset_old'])
155145

156-
mock(obj).get_table_or_partition(task['table'])
146+
mock(obj).get_table_or_partition(config['table'])
157147
mock(obj).copy(config['table'], config['table_old'], config['dataset_old'])
158148

159149
mock(obj).copy(config['temp_table'], config['table'], write_disposition: 'WRITE_TRUNCATE')
@@ -168,9 +158,11 @@ def test_replace_backup_auto_create_dataset
168158
any_instance_of(BigqueryClient) do |obj|
169159
mock(obj).create_dataset(config['dataset'])
170160
mock(obj).create_dataset(config['dataset_old'], reference: config['dataset'])
161+
mock(obj).create_table_if_not_exists(config['table'])
171162
mock(obj).create_table_if_not_exists(config['temp_table'])
163+
mock(obj).create_table_if_not_exists(config['table_old'], dataset: config['dataset_old'])
172164

173-
mock(obj).get_table_or_partition(task['table'])
165+
mock(obj).get_table_or_partition(config['table'])
174166
mock(obj).copy(config['table'], config['table_old'], config['dataset_old'])
175167

176168
mock(obj).copy(config['temp_table'], config['table'], write_disposition: 'WRITE_TRUNCATE')
@@ -180,35 +172,16 @@ def test_replace_backup_auto_create_dataset
180172
end
181173

182174
def test_replace_backup_with_partitioning
183-
config = least_config.merge('mode' => 'replace_backup', 'table' => 'table$20160929', 'dataset_old' => 'dataset_old', 'table_old' => 'table_old$20190929', 'temp_table' => 'temp_table')
184-
task = Bigquery.configure(config, schema, processor_count)
185-
any_instance_of(BigqueryClient) do |obj|
186-
mock(obj).get_dataset(config['dataset'])
187-
mock(obj).get_dataset(config['dataset_old'])
188-
mock(obj).create_table_if_not_exists(config['temp_table'])
189-
mock(obj).get_table(task['table'])
190-
mock(obj).get_table(task['table_old'], dataset: config['dataset_old'])
191-
192-
mock(obj).get_table_or_partition(task['table'])
193-
mock(obj).copy(config['table'], config['table_old'], config['dataset_old'])
194-
195-
mock(obj).copy(config['temp_table'], config['table'], write_disposition: 'WRITE_TRUNCATE')
196-
mock(obj).delete_table(config['temp_table'])
197-
end
198-
Bigquery.transaction(config, schema, processor_count, &control)
199-
end
200-
201-
def test_replace_backup_with_partitioning_auto_create_table
202175
config = least_config.merge('mode' => 'replace_backup', 'table' => 'table$20160929', 'dataset_old' => 'dataset_old', 'table_old' => 'table_old$20160929', 'temp_table' => 'temp_table', 'auto_create_table' => true)
203176
task = Bigquery.configure(config, schema, processor_count)
204177
any_instance_of(BigqueryClient) do |obj|
205178
mock(obj).get_dataset(config['dataset'])
206179
mock(obj).get_dataset(config['dataset_old'])
207180
mock(obj).create_table_if_not_exists(config['temp_table'])
208-
mock(obj).create_table_if_not_exists(task['table'])
209-
mock(obj).create_table_if_not_exists(task['table_old'], dataset: config['dataset_old'])
181+
mock(obj).create_table_if_not_exists(config['table'])
182+
mock(obj).create_table_if_not_exists(config['table_old'], dataset: config['dataset_old'])
210183

211-
mock(obj).get_table_or_partition(task['table'])
184+
mock(obj).get_table_or_partition(config['table'])
212185
mock(obj).copy(config['table'], config['table_old'], config['dataset_old'])
213186

214187
mock(obj).copy(config['temp_table'], config['table'], write_disposition: 'WRITE_TRUNCATE')
@@ -225,26 +198,14 @@ def test_append
225198
any_instance_of(BigqueryClient) do |obj|
226199
mock(obj).get_dataset(config['dataset'])
227200
mock(obj).create_table_if_not_exists(config['temp_table'])
201+
mock(obj).create_table_if_not_exists(config['table'])
228202
mock(obj).copy(config['temp_table'], config['table'], write_disposition: 'WRITE_APPEND')
229203
mock(obj).delete_table(config['temp_table'])
230204
end
231205
Bigquery.transaction(config, schema, processor_count, &control)
232206
end
233207

234208
def test_append_with_partitioning
235-
config = least_config.merge('mode' => 'append', 'table' => 'table$20160929')
236-
task = Bigquery.configure(config, schema, processor_count)
237-
any_instance_of(BigqueryClient) do |obj|
238-
mock(obj).get_dataset(config['dataset'])
239-
mock(obj).create_table_if_not_exists(config['temp_table'])
240-
mock(obj).get_table(config['table'])
241-
mock(obj).copy(config['temp_table'], config['table'], write_disposition: 'WRITE_APPEND')
242-
mock(obj).delete_table(config['temp_table'])
243-
end
244-
Bigquery.transaction(config, schema, processor_count, &control)
245-
end
246-
247-
def test_append_with_partitioning_with_auto_create_table
248209
config = least_config.merge('mode' => 'append', 'table' => 'table$20160929', 'auto_create_table' => true)
249210
task = Bigquery.configure(config, schema, processor_count)
250211
any_instance_of(BigqueryClient) do |obj|

0 commit comments

Comments
 (0)