Skip to content

Commit 73abfa3

Browse files
dwsuppleedwsupplee
committed
add create on insert logic
1 parent 1c5105c commit 73abfa3

File tree

4 files changed

+278
-7
lines changed

4 files changed

+278
-7
lines changed

‎src/BigQuery/Table.php‎

Lines changed: 96 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,11 @@
2020
use Google\Cloud\BigQuery\Connection\ConnectionInterface;
2121
use Google\Cloud\Core\ArrayTrait;
2222
use Google\Cloud\Core\ConcurrencyControlTrait;
23+
use Google\Cloud\Core\Exception\ConflictException;
2324
use Google\Cloud\Core\Exception\NotFoundException;
2425
use Google\Cloud\Core\Iterator\ItemIterator;
2526
use Google\Cloud\Core\Iterator\PageIterator;
27+
use Google\Cloud\Core\RetryDeciderTrait;
2628
use Google\Cloud\Storage\StorageObject;
2729
use Psr\Http\Message\StreamInterface;
2830

@@ -33,8 +35,12 @@
3335
*/
3436
class Table
3537
{
38+
const MAX_RETRIES = 100;
39+
const INSERT_CREATE_MAX_DELAY_MS = 60000000;
40+
3641
use ArrayTrait;
3742
use ConcurrencyControlTrait;
43+
use RetryDeciderTrait;
3844

3945
/**
4046
* @var ConnectionInterface Represents a connection to BigQuery.
@@ -81,6 +87,11 @@ public function __construct(
8187
'datasetId' => $datasetId,
8288
'projectId' => $projectId
8389
];
90+
$this->setHttpRetryCodes([]);
91+
$this->setHttpRetryMessages([
92+
'rateLimitExceeded',
93+
'backendError'
94+
]);
8495
}
8596

8697
/**
@@ -552,6 +563,19 @@ public function insertRow(array $row, array $options = [])
552563
* @param array $options [optional] {
553564
* Configuration options.
554565
*
566+
* @type bool $autoCreate Whether or not to attempt to automatically
567+
* create the table in the case it does not exist. Please note, it
568+
* will be required to provide a schema through
569+
* $tableMetadata['schema'] in the case the table does not already
570+
* exist. **Defaults to** `false`.
571+
* @type $tableMetadata Metadata to apply to table to be created. The
572+
* full set of metadata are outlined at the
573+
* [Table Resource API docs](https://cloud.google.com/bigquery/docs/reference/v2/tables#resource).
574+
* Only applies when `autoCreate` is `true`.
575+
* @type $maxRetries The maximum number of times to attempt creating the
576+
* table in the case of failure. Please note, each retry attempt
577+
* may take up to two minutes. Only applies when `autoCreate` is
578+
* `true`. **Defaults to** `100`.
555579
* @type bool $skipInvalidRows Insert all valid rows of a request, even
556580
* if invalid rows exist. The default value is `false`, which
557581
* causes the entire request to fail if any invalid rows exist.
@@ -569,11 +593,17 @@ public function insertRow(array $row, array $options = [])
569593
* for considerations when working with templates tables.
570594
* }
571595
* @return InsertResponse
572-
* @throws \InvalidArgumentException
596+
* @throws \InvalidArgumentException If a provided row does not contain a
597+
* `data` key, if a schema is not defined when `autoCreate` is
598+
* `true`, or if less than 1 row is provided.
573599
* @codingStandardsIgnoreEnd
574600
*/
575601
public function insertRows(array $rows, array $options = [])
576602
{
603+
if (count($rows) === 0) {
604+
throw new \InvalidArgumentException('Must provide at least a single row.');
605+
}
606+
577607
foreach ($rows as $row) {
578608
if (!isset($row['data'])) {
579609
throw new \InvalidArgumentException('A row must have a data key.');
@@ -593,7 +623,7 @@ public function insertRows(array $rows, array $options = [])
593623
}
594624

595625
return new InsertResponse(
596-
$this->connection->insertAllTableData($this->identity + $options),
626+
$this->handleInsert($options),
597627
$options['rows']
598628
);
599629
}
@@ -673,4 +703,68 @@ public function identity()
673703
{
674704
return $this->identity;
675705
}
706+
707+
/**
708+
* Delay execution in microseconds.
709+
*
710+
* @param int $microSeconds
711+
*/
712+
protected function usleep($microSeconds)
713+
{
714+
usleep($microSeconds);
715+
}
716+
717+
/**
718+
* Handles inserting table data and manages custom retry logic in the case
719+
* a table needs to be created.
720+
*
721+
* @param array $options Configuration options.
722+
* @return array
723+
*/
724+
private function handleInsert(array $options)
725+
{
726+
$attempt = 0;
727+
$metadata = $this->pluck('tableMetadata', $options, false) ?: [];
728+
$autoCreate = $this->pluck('autoCreate', $options, false) ?: false;
729+
$maxRetries = $this->pluck('maxRetries', $options, false) ?: self::MAX_RETRIES;
730+
731+
while (true) {
732+
try {
733+
return $this->connection->insertAllTableData(
734+
$this->identity + $options
735+
);
736+
} catch (NotFoundException $ex) {
737+
if ($autoCreate === true && $attempt <= $maxRetries) {
738+
if (!isset($metadata['schema'])) {
739+
throw new \InvalidArgumentException(
740+
'A schema is required when creating a table.'
741+
);
742+
}
743+
744+
$this->usleep(mt_rand(1, self::INSERT_CREATE_MAX_DELAY_MS));
745+
746+
try {
747+
$this->connection->insertTable($metadata + [
748+
'projectId' => $this->identity['projectId'],
749+
'datasetId' => $this->identity['datasetId'],
750+
'tableReference' => $this->identity,
751+
'retries' => 0
752+
]);
753+
} catch (ConflictException $ex) {
754+
} catch (\Exception $ex) {
755+
$retryFunction = $this->getRetryFunction();
756+
757+
if (!$retryFunction($ex)) {
758+
throw $ex;
759+
}
760+
}
761+
762+
$this->usleep(self::INSERT_CREATE_MAX_DELAY_MS);
763+
$attempt++;
764+
} else {
765+
throw $ex;
766+
}
767+
}
768+
}
769+
}
676770
}

‎tests/snippets/BigQuery/QueryJobConfigurationTest.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@
2323
use Google\Cloud\Dev\Snippet\SnippetTestCase;
2424

2525
/**
26-
* @group bigquery-2
26+
* @group bigquery
2727
*/
2828
class QueryJobConfigurationTest extends SnippetTestCase
2929
{

‎tests/system/BigQuery/LoadDataAndQueryTest.php‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -430,4 +430,33 @@ public function testInsertRowsToTable()
430430
$this->assertTrue($insertResponse->isSuccessful());
431431
$this->assertEquals(self::$expectedRows, $actualRows);
432432
}
433+
434+
public function testInsertRowsToTableWithAutoCreate()
435+
{
436+
$tName = uniqid(BigQueryTestCase::TESTING_PREFIX);
437+
$rows = [
438+
['data' => ['hello' => 'world']]
439+
];
440+
$insertResponse = self::$dataset->table($tName)
441+
->insertRows($rows, [
442+
'autoCreate' => true,
443+
'tableMetadata' => [
444+
'schema' => [
445+
'fields' => [
446+
[
447+
'name' => 'hello',
448+
'type' => 'STRING'
449+
]
450+
]
451+
]
452+
]
453+
]);
454+
$results = self::$dataset
455+
->table($tName)
456+
->rows();
457+
$actualRows = count(iterator_to_array($results));
458+
459+
$this->assertTrue($insertResponse->isSuccessful());
460+
$this->assertEquals(count($rows), $actualRows);
461+
}
433462
}

‎tests/unit/BigQuery/TableTest.php‎

Lines changed: 152 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
use Google\Cloud\BigQuery\LoadJobConfiguration;
2727
use Google\Cloud\BigQuery\Table;
2828
use Google\Cloud\BigQuery\ValueMapper;
29+
use Google\Cloud\Core\Exception\ConflictException;
2930
use Google\Cloud\Core\Exception\NotFoundException;
3031
use Google\Cloud\Core\Upload\AbstractUploader;
3132
use Google\Cloud\Storage\Connection\ConnectionInterface as StorageConnectionInterface;
@@ -479,22 +480,169 @@ public function testInsertsRows()
479480
->willReturn([])
480481
->shouldBeCalledTimes(1);
481482
$table = $this->getTable($this->connection);
482-
483483
$insertResponse = $table->insertRows($rowData);
484484

485485
$this->assertInstanceOf(InsertResponse::class, $insertResponse);
486486
$this->assertTrue($insertResponse->isSuccessful());
487487
}
488488

489+
public function testInsertsRowsWithAutoCreate()
490+
{
491+
$insertId = '1';
492+
$data = ['key' => 'value'];
493+
$rowData = [
494+
[
495+
'insertId' => $insertId,
496+
'data' => $data
497+
]
498+
];
499+
$schema = [
500+
'fields' => [
501+
[
502+
'name' => 'key',
503+
'type' => 'STRING'
504+
]
505+
]
506+
];
507+
$expectedInsertTableDataArguments = [
508+
'tableId' => self::TABLE_ID,
509+
'projectId' => self::PROJECT_ID,
510+
'datasetId' => self::DATASET_ID,
511+
'rows' => [
512+
[
513+
'json' => $data,
514+
'insertId' => $insertId
515+
]
516+
]
517+
];
518+
$expectedInsertTableArguments = [
519+
'schema' => $schema,
520+
'retries' => 0,
521+
'projectId' => self::PROJECT_ID,
522+
'datasetId' => self::DATASET_ID,
523+
'tableReference' => [
524+
'projectId' => self::PROJECT_ID,
525+
'datasetId' => self::DATASET_ID,
526+
'tableId' => self::TABLE_ID
527+
]
528+
];
529+
$callCount = 0;
530+
$this->connection->insertAllTableData($expectedInsertTableDataArguments)
531+
->will(function () use (&$callCount) {
532+
if ($callCount === 0) {
533+
$callCount++;
534+
throw new NotFoundException(null);
535+
}
536+
537+
return [];
538+
})
539+
->shouldBeCalledTimes(2);
540+
$this->connection->insertTable($expectedInsertTableArguments)
541+
->willReturn([]);
542+
$table = $this->getTable($this->connection);
543+
$insertResponse = $table->insertRows($rowData, [
544+
'autoCreate' => true,
545+
'tableMetadata' => [
546+
'schema' => $schema
547+
]
548+
]);
549+
550+
$this->assertInstanceOf(InsertResponse::class, $insertResponse);
551+
$this->assertTrue($insertResponse->isSuccessful());
552+
}
553+
489554
/**
490555
* @expectedException \InvalidArgumentException
556+
* @expectedMessage A schema is required when creating a table.
491557
*/
492-
public function testInsertRowsThrowsException()
558+
public function testInsertRowsThrowsExceptionWithoutSchema()
559+
{
560+
$options = [
561+
'autoCreate' => true
562+
];
563+
$this->connection->insertAllTableData(Argument::any())
564+
->willThrow(new NotFoundException(null));
565+
$table = $this->getTable($this->connection);
566+
$table->insertRows([
567+
[
568+
'data' => [
569+
'city' => 'state'
570+
]
571+
]
572+
], $options);
573+
}
574+
575+
/**
576+
* @expectedException \Exception
577+
*/
578+
public function testInsertRowsThrowsExceptionWithUnretryableTableFailure()
579+
{
580+
$options = [
581+
'autoCreate' => true,
582+
'tableMetadata' => [
583+
'schema' => []
584+
]
585+
];
586+
$this->connection->insertAllTableData(Argument::any())
587+
->willThrow(new NotFoundException(null));
588+
$this->connection->insertTable(Argument::any())
589+
->willThrow(new \Exception());
590+
$table = $this->getTable($this->connection);
591+
$table->insertRows([
592+
[
593+
'data' => [
594+
'city' => 'state'
595+
]
596+
]
597+
], $options);
598+
}
599+
600+
/**
601+
* @expectedException Google\Cloud\Core\Exception\NotFoundException
602+
*/
603+
public function testInsertRowsThrowsExceptionWhenMaxRetryLimitHit()
604+
{
605+
$options = [
606+
'autoCreate' => true,
607+
'maxRetries' => 0,
608+
'tableMetadata' => [
609+
'schema' => []
610+
]
611+
];
612+
$this->connection->insertAllTableData(Argument::any())
613+
->willThrow(new NotFoundException(null));
614+
$this->connection->insertTable(Argument::any())
615+
->willThrow(new ConflictException(null));
616+
$table = $this->getTable($this->connection);
617+
$table->insertRows([
618+
[
619+
'data' => [
620+
'city' => 'state'
621+
]
622+
]
623+
], $options);
624+
}
625+
626+
/**
627+
* @expectedException \InvalidArgumentException
628+
* @expectedMessage A row must have a data key.
629+
*/
630+
public function testInsertRowsThrowsExceptionWithoutDataKey()
493631
{
494632
$table = $this->getTable($this->connection);
495633
$table->insertRows([[], []]);
496634
}
497635

636+
/**
637+
* @expectedException \InvalidArgumentException
638+
* @expectedMessage Must provide at least a single row.
639+
*/
640+
public function testInsertRowsThrowsExceptionWithZeroRows()
641+
{
642+
$table = $this->getTable($this->connection);
643+
$table->insertRows([]);
644+
}
645+
498646
public function testGetsInfo()
499647
{
500648
$tableInfo = ['tableReference' => ['tableId' => self::TABLE_ID]];
@@ -533,8 +681,8 @@ public function testGetsIdentity()
533681

534682
class TableStub extends Table
535683
{
536-
protected function generateJobId($jobIdPrefix = null)
684+
protected function usleep($ms)
537685
{
538-
return $jobIdPrefix ? $jobIdPrefix . '-' . BigQueryClientTest::JOBID : BigQueryClientTest::JOBID;
686+
return;
539687
}
540688
}

0 commit comments

Comments
 (0)