<?php

/**

* Copyright 2018, Google LLC All rights reserved.

*

* Licensed under the Apache License, Version 2.0 (the "License");

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/

namespace Google\Cloud\Bigtable;

use Google\ApiCore\ApiException;

use Google\ApiCore\Serializer;

use Google\Cloud\Bigtable\Exception\BigtableDataOperationException;

use Google\Cloud\Bigtable\Filter\FilterInterface;

use Google\Cloud\Bigtable\Mutations;

use Google\Cloud\Bigtable\ReadModifyWriteRowRules;

use Google\Cloud\Bigtable\V2\BigtableClient as GapicClient;

use Google\Cloud\Bigtable\V2\MutateRowsRequest\Entry;

use Google\Cloud\Bigtable\V2\Row;

use Google\Cloud\Bigtable\V2\RowRange;

use Google\Cloud\Bigtable\V2\RowSet;

use Google\Cloud\Core\ArrayTrait;

use Google\Rpc\Code;

/**

* A table instance can be used to read rows and to perform insert, update, and

* delete operations.

*

* Example:

* ```

* use Google\Cloud\Bigtable\BigtableClient;

*

* $bigtable = new BigtableClient();

* $table = $bigtable->table('my-instance', 'my-table');

* ```

*/

class Table

{

use ArrayTrait;

/**

* @var GapicClient

*/

private $gapicClient;

/**

* @var array

*/

private $options;

/**

* @var string

*/

private $tableName;

/**

* @var Serializer

*/

private $serializer;

/**

* Create a table instance.

*

* @param GapicClient $gapicClient The GAPIC client used to make requests.

* @param string $tableName The full table name. Must match the following

* pattern: projects/{project}/instances/{instance}/tables/{table}.

* @param array $options [optional] {

* Configuration options.

*

* @type string $appProfileId This value specifies routing for

* replication. **Defaults to** the "default" application profile.

* @type array $headers Headers to be passed with each request.

* @type int $retries Number of times to retry. **Defaults to** `3`.

* This settings only applies to {@see Google\Cloud\Bigtable\Table::mutateRows()},

* {@see Google\Cloud\Bigtable\Table::upsert()} and

* {@see Google\Cloud\Bigtable\Table::readRows()}.

* }

*/

public function __construct(

GapicClient $gapicClient,

$tableName,

array $options = []

) {

$this->gapicClient = $gapicClient;

$this->tableName = $tableName;

$this->options = $options;

$this->serializer = new Serializer();

}

/**

* Mutates rows in a table.

*

* Example:

* ```

* use Google\Cloud\Bigtable\Mutations;

*

* $mutations = (new Mutations)

* ->upsert('cf1', 'cq1', 'value1', 1534183334215000);

*

* $table->mutateRows(['r1' => $mutations]);

* ```

*

* @param array $rowMutations An associative array with the key being the

* row key and the value being the

* {@see Google\Cloud\Bigtable\Mutations} to perform.

* @param array $options [optional] Configuration options.

* @return void

* @throws ApiException|BigtableDataOperationException If the remote call fails or operation fails.

* @throws InvalidArgumentException If rowMutations is a list instead of associative array indexed by row key.

*/

public function mutateRows(array $rowMutations, array $options = [])

{

if (!$this->isAssoc($rowMutations)) {

throw new \InvalidArgumentException(

'Expected rowMutations to be of type associative array, instead got list.'

);

}

$entries = [];

foreach ($rowMutations as $rowKey => $mutations) {

$entries[] = $this->toEntry($rowKey, $mutations);

}

$this->mutateRowsWithEntries($entries, $options);

}

/**

* Mutates a row atomically. Cells already present in the row are left

* unchanged unless explicitly changed by `mutations`.

*

* Example:

* ```

* use Google\Cloud\Bigtable\Mutations;

*

* $mutations = (new Mutations)

* ->upsert('cf1', 'cq1', 'value1', 1534183334215000);

*

* $table->mutateRow('r1', $mutations);

* ```

*

* @param string $rowKey The row key of the row to mutate.

* @param Mutations $mutations Mutations to apply on row.

* @param array $options [optional] {

* Configuration options.

*

* @type int $retries Number of times to retry. **Defaults to** `3`.

* }

* @return void

* @throws ApiException If the remote call fails.

*/

public function mutateRow($rowKey, Mutations $mutations, array $options = [])

{

$this->gapicClient->mutateRow(

$this->tableName,

$rowKey,

$mutations->toProto(),

$options + $this->options

);

}

/**

* Insert/update rows in a table.

*

* Example:

* ```

* $table->upsert([

* 'r1' => [

* 'cf1' => [

* 'cq1' => [

* 'value'=>'value1',

* 'timeStamp' => 1534183334215000

* ]

* ]

* ]

*]);

* ```

*

* @param array[] $rows An array of rows.

* @param array $options [optional] {

* Configuration options.

*

* @type int $retries Number of times to retry. **Defaults to** `3`.

* }

* @return void

* @throws ApiException|BigtableDataOperationException If the remote call fails or operation fails

*/

public function upsert(array $rows, array $options = [])

{

$entries = [];

foreach ($rows as $rowKey => $families) {

$mutations = new Mutations;

foreach ($families as $family => $qualifiers) {

foreach ($qualifiers as $qualifier => $value) {

if (isset($value['timeStamp'])) {

$mutations->upsert(

$family,

$qualifier,

$value['value'],

$value['timeStamp']

);

} else {

$mutations->upsert($family, $qualifier, $value['value']);

}

}

}

$entries[] = $this->toEntry($rowKey, $mutations);

}

$this->mutateRowsWithEntries($entries, $options);

}

/**

* Read rows from the table.

*

* Example:

* ```

* $rows = $table->readRows();

*

* foreach ($rows as $key => $row) {

* echo $key . ': ' . print_r($row, true) . PHP_EOL;

* }

* ```

*

* ```

* // Specify a set of row ranges.

* $rows = $table->readRows([

* 'rowRanges' => [

* [

* 'startKeyOpen' => 'jefferson',

* 'endKeyOpen' => 'lincoln'

* ]

* ]

* ]);

*

* foreach ($rows as $key => $row) {

* echo $key . ': ' . print_r($row, true) . PHP_EOL;

* }

* ```

*

* @param array $options [optional] {

* Configuration options.

*

* @type string[] $rowKeys A set of row keys to read.

* @type array $rowRanges A set of row ranges. Each row range is an

* associative array which may contain a start key

* (`startKeyClosed` or `startKeyOpen`) and/or an end key

* (`endKeyOpen` or `endKeyClosed`).

* @type FilterInterface $filter A filter used to take an input row and

* produce an alternate view of the row based on the specified rules.

* To learn more please see {@see Google\Cloud\Bigtable\Filter} which

* provides static factory methods for the various filter types.

* @type int $rowsLimit The number of rows to scan.

* @type int $retries Number of times to retry. **Defaults to** `3`.

* }

* @return ChunkFormatter

*/

public function readRows(array $options = [])

{

$rowKeys = $this->pluck('rowKeys', $options, false) ?: [];

$ranges = $this->pluck('rowRanges', $options, false) ?: [];

$filter = $this->pluck('filter', $options, false) ?: null;

array_walk($ranges, function (&$range) {

$range = $this->serializer->decodeMessage(

new RowRange(),

$range

);

});

if (!is_array($rowKeys)) {

throw new \InvalidArgumentException(

sprintf(

'Expected rowKeys to be of type array, instead got \'%s\'.',

gettype($rowKeys)

)

);

}

if ($ranges || $rowKeys) {

$options['rows'] = $this->serializer->decodeMessage(

new RowSet,

[

'rowKeys' => $rowKeys,

'rowRanges' => $ranges

]

);

}

if ($filter !== null) {

if (!$filter instanceof FilterInterface) {

throw new \InvalidArgumentException(

sprintf(

'Expected filter to be of type \'%s\', instead got \'%s\'.',

FilterInterface::class,

gettype($filter)

)

);

}

$options['filter'] = $filter->toProto();

}

return new ChunkFormatter(

[$this->gapicClient, 'readRows'],

$this->tableName,

$options + $this->options

);

}

/**

* Read a single row from the table.

*

* Example:

* ```

* $row = $table->readRow('jefferson');

*

* print_r($row);

* ```

*

* @param string $rowKey The row key to read.

* @param array $options [optional] Configuration options.

* @return array|null

*/

public function readRow($rowKey, array $options = [])

{

return $this->readRows(

['rowKeys' => [$rowKey]] + $options + $this->options

)

->readAll()

->current();

}

/**

* Modifies a row atomically on the server. The method reads the latest

* existing timestamp and value from the specified columns and writes a new

* entry based on pre-defined read/modify/write rules. The new value for the

* timestamp is the greater of the existing timestamp or the current server

* time. The method returns the new contents of all modified cells.

*

* Example:

* ```

* use Google\Cloud\Bigtable\ReadModifyWriteRowRules;

*

* $rules = (new ReadModifyWriteRowRules)

* ->append('cf1', 'cq1', 'value12');

* $row = $table->readModifyWriteRow('rk1', $rules);

*

* print_r($row);

* ```

*

* ```

* // Increments value

* use Google\Cloud\Bigtable\DataUtil;

* use Google\Cloud\Bigtable\ReadModifyWriteRowRules;

* use Google\Cloud\Bigtable\Mutations;

*

* $mutations = (new Mutations)

* ->upsert('cf1', 'cq1', DataUtil::intToByteString(2));

*

* $table->mutateRows(['rk1' => $mutations]);

*

* $rules = (new ReadModifyWriteRowRules)

* ->increment('cf1', 'cq1', 3);

* $row = $table->readModifyWriteRow('rk1', $rules);

*

* print_r($row);

* ```

*

* @param string $rowKey The row key to read.

* @param ReadModifyWriteRowRules $rules Rules to apply on row.

* @param array $options [optional] Configuration options.

* @return array Returns array containing all column family keyed by family name.

* @throws ApiException If the remote call fails or operation fails

*/

public function readModifyWriteRow($rowKey, ReadModifyWriteRowRules $rules, array $options = [])

{

$readModifyWriteRowResponse = $this->gapicClient->readModifyWriteRow(

$this->tableName,

$rowKey,

$rules->toProto(),

$options + $this->options

);

return $this->convertToArray($readModifyWriteRowResponse->getRow());

}

/**

* Returns a sample of row keys in the table. The returned row keys will

* delimit contiguous sections of the table of approximately equal size,

* which can be used to break up the data for distributed tasks like

* mapreduces.

*

* Example:

* ```

* $rowKeyStream = $table->sampleRowKeys();

* foreach ($rowKeyStream as $rowKey) {

* print_r($rowKey) . PHP_EOL;

* }

* ```

*

* @param array $options [optional] Configuration options.

* @return \Generator<array> A list of associative arrays, each with the keys `rowKey` and `offset`.

* @throws ApiException If the remote call fails or operation fails

*/

public function sampleRowKeys(array $options = [])

{

$stream = $this->gapicClient->sampleRowKeys(

$this->tableName,

$options + $this->options

);

foreach ($stream->readAll() as $response) {

yield [

'rowKey' => $response->getRowKey(),

'offset' => $response->getOffsetBytes()

];

}

}

/**

* Mutates the specified row atomically based on output of the filter.

*

* Example:

* ```

* use Google\Cloud\Bigtable\Mutations;

*

* $mutations = (new Mutations)->upsert('family', 'qualifier', 'value');

* $result = $table->checkAndMutateRow('rk1', ['trueMutations' => $mutations]);

* ```

*

* ```

* // With predicate filter

* use Google\Cloud\Bigtable\Filter;

* use Google\Cloud\Bigtable\Mutations;

*

* $mutations = (new Mutations)->upsert('family', 'qualifier', 'value');

* $predicateFilter = Filter::qualifier()->exactMatch('cq');

* $options = ['predicateFilter' => $predicateFilter, 'trueMutations' => $mutations];

* $result = $table->checkAndMutateRow('rk1', $options);

* ```

*

* @param string $rowKey The row key to mutate row conditionally.

* @param array $options [optional] {

* Configuration options.

*

* @type FilterInterface $predicateFilter The filter to be applied to the specified

* row. Depending on whether or not any results are yielded, either the

* trueMutations or falseMutations will be executed. If unset, checks that the

* row contains any values at all. Only a single condition can be set, however

* that filter can be {@see Google\Cloud\Bigtable\Filter::chain()} or

* {@see Google\Cloud\Bigtable\Filter::interleave()} which can wrap multiple other

* filters.

* WARNING: {@see Google\Cloud\Bigtable\Filter::condition()} is not supported.

* @type Mutations $trueMutations Mutations to be atomically applied when the predicate

* filter's condition yields at least one cell when applied to the row. Please note

* either `trueMutations` or `falseMutations` must be provided.

* @type Mutations $falseMutations Mutations to be atomically applied when the predicate

* filter's condition does not yield any cells when applied to the row. Please note

* either `trueMutations` or `falseMutations` must be provided.

* }

*

* @return bool Returns true if predicate filter yielded any output, false otherwise.

* @throws ApiException If the remote call fails or operation fails

* @throws \InvalidArgumentException If neither of $trueMutations or $falseMutations is set.

* If $predicateFilter is not instance of {@see Google\Cloud\Bigtable\Filter\FilterInterface}.

* If $trueMutations is set and is not instance of {@see Google\Cloud\Bigtable\Mutations}.

* If $falseMutations is set and is not instance of {@see Google\Cloud\Bigtable\Mutations}.

*/

public function checkAndMutateRow($rowKey, array $options = [])

{

$hasSetMutations = false;

if (isset($options['predicateFilter'])) {

$this->convertToProto($options, 'predicateFilter', FilterInterface::class);

}

if (isset($options['trueMutations'])) {

$this->convertToProto($options, 'trueMutations', Mutations::class);

$hasSetMutations = true;

}

if (isset($options['falseMutations'])) {

$this->convertToProto($options, 'falseMutations', Mutations::class);

$hasSetMutations = true;

}

if (!$hasSetMutations) {

throw new \InvalidArgumentException('checkAndMutateRow must have either trueMutations or falseMutations.');

}

return $this->gapicClient

->checkAndMutateRow(

$this->tableName,

$rowKey,

$options + $this->options

)

->getPredicateMatched();

}

private function mutateRowsWithEntries(array $entries, array $options = [])

{

$rowMutationsFailedResponse = [];

$options = $options + $this->options;

$argumentFunction = function () use (&$entries, &$rowMutationsFailedResponse, $options) {

if (count($rowMutationsFailedResponse) > 0) {

$entries = array_values($entries);

$rowMutationsFailedResponse = [];

}

return [$this->tableName, $entries, $options];

};

$statusCode = Code::OK;

$lastProcessedIndex = -1;

$retryFunction = function ($ex) use (&$statusCode, &$lastProcessedIndex) {

if (($ex && ResumableStream::isRetryable($ex->getCode())) || (ResumableStream::isRetryable($statusCode))) {

$statusCode = Code::OK;

$lastProcessedIndex = -1;

return true;

}

return false;

};

$retryingStream = new ResumableStream(

[$this->gapicClient, 'mutateRows'],

$argumentFunction,

$retryFunction,

$this->pluck('retries', $options, false)

);

$message = 'partial failure';

try {

foreach ($retryingStream as $mutateRowsResponse) {

foreach ($mutateRowsResponse->getEntries() as $mutateRowsResponseEntry) {

if ($mutateRowsResponseEntry->getStatus()->getCode() !== Code::OK) {

if ($statusCode === Code::OK

|| !ResumableStream::isRetryable($mutateRowsResponseEntry->getStatus()->getCode())) {

$statusCode = $mutateRowsResponseEntry->getStatus()->getCode();

}

$rowMutationsFailedResponse[] = [

'rowKey' => $entries[$mutateRowsResponseEntry->getIndex()]->getRowKey(),

'statusCode' => $mutateRowsResponseEntry->getStatus()->getCode(),

'message' => $mutateRowsResponseEntry->getStatus()->getMessage()

];

} else {

unset($entries[$mutateRowsResponseEntry->getIndex()]);

}

$lastProcessedIndex = $mutateRowsResponseEntry->getIndex();

}

}

} catch (ApiException $ex) {

$this->appendPendingEntryToFailedMutations(

$lastProcessedIndex,

$entries,

$rowMutationsFailedResponse,

$ex->getCode(),

$ex->getMessage()

);

if ($ex->getErrorInfoMetadata()) {

$rowMutationsFailedResponse = array_merge(

$rowMutationsFailedResponse,

$ex->getErrorInfoMetadata()

);

}

throw new BigtableDataOperationException(

$ex->getMessage(),

$ex->getCode(),

$rowMutationsFailedResponse

);

}

if (!empty($rowMutationsFailedResponse)) {

$this->appendPendingEntryToFailedMutations(

$lastProcessedIndex,

$entries,

$rowMutationsFailedResponse,

$statusCode,

$message

);

throw new BigtableDataOperationException($message, $statusCode, $rowMutationsFailedResponse);

}

}

private function appendPendingEntryToFailedMutations(

$lastProcessedIndex,

&$entries,

&$rowMutationsFailedResponse,

$statusCode,

$message

) {

if (count($rowMutationsFailedResponse) !== count($entries)) {

end($entries);

foreach (range($lastProcessedIndex + 1, key($entries)) as $index) {

$rowMutationsFailedResponse[] = [

'rowKey' => $entries[$index]->getRowKey(),

'statusCode' => $statusCode,

'message' => $message

];

}

}

}

private function toEntry($rowKey, Mutations $mutations)

{

return (new Entry)

->setRowKey($rowKey)

->setMutations($mutations->toProto());

}

private function convertToArray(Row $row)

{

if ($row === null) {

return [];

}

$families = [];

foreach ($row->getFamilies() as $family) {

$qualifiers = [];

foreach ($family->getColumns() as $column) {

$values = [];

foreach ($column->getCells() as $cell) {

$values[] = [

'value' => $cell->getValue(),

'timeStamp' => $cell->getTimestampMicros(),

'labels' => ($cell->getLabels()->getIterator()->valid())

? implode(iterator_to_array($cell->getLabels()->getIterator()))

: ''

];

}

$qualifiers[$column->getQualifier()] = $values;

}

$families[$family->getName()] = $qualifiers;

}

return $families;

}

private function convertToProto(array &$options, $key, $expectedType)

{

if (!$options[$key] instanceof $expectedType) {

throw new \InvalidArgumentException(

sprintf(

'Expected %s to be of type \'%s\', instead got \'%s\'.',

$key,

$expectedType,

gettype($options[$key])

)

);

}

$options[$key] = $options[$key]->toProto();

}

}