table.js

/*!
 * Copyright 2016 Google Inc. 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.
 */

'use strict';

var arrify = require('arrify');
var common = require('@google-cloud/common');
var commonGrpc = require('@google-cloud/common-grpc');
var concat = require('concat-stream');
var flatten = require('lodash.flatten');
var is = require('is');
var propAssign = require('prop-assign');
var pumpify = require('pumpify');
var through = require('through2');

var Family = require('./family.js');
var Filter = require('./filter.js');
var Mutation = require('./mutation.js');
var Row = require('./row.js');
const ChunkTransformer = require('./chunktransformer.js');

// See protos/google/rpc/code.proto
// (4=DEADLINE_EXCEEDED, 10=ABORTED, 14=UNAVAILABLE)
const RETRYABLE_STATUS_CODES = new Set([4, 10, 14]);

/**
 * Create a Table object to interact with a Cloud Bigtable table.
 *
 * @class
 * @param {Instance} instance Name of the table.
 * @param {string} name Name of the table.
 *
 * @example
 * const Bigtable = require('@google-cloud/bigtable');
 * const bigtable = new Bigtable();
 * const instance = bigtable.instance('my-instance');
 * const table = instance.table('prezzy');
 */
function Table(instance, name) {
  this.bigtable = instance.bigtable;
  this.instance = instance;

  var id = Table.formatName_(instance.id, name);

  this.id = id;
  this.name = id.split('/').pop();
}

/**
 * The view to be applied to the returned table's fields.
 * Defaults to schema if unspecified.
 *
 * @private
 */
Table.VIEWS = {
  unspecified: 0,
  name: 1,
  schema: 2,
  full: 4,
};

/**
 * Formats the table name to include the Bigtable cluster.
 *
 * @private
 *
 * @param {string} instanceName The formatted instance name.
 * @param {string} name The table name.
 *
 * @example
 * Table.formatName_(
 *   'projects/my-project/zones/my-zone/instances/my-instance',
 *   'my-table'
 * );
 * // 'projects/my-project/zones/my-zone/instances/my-instance/tables/my-table'
 */
Table.formatName_ = function(instanceName, name) {
  if (name.indexOf('/') > -1) {
    return name;
  }

  return instanceName + '/tables/' + name;
};

/**
 * Creates a range based off of a key prefix.
 *
 * @private
 *
 * @param {string} start The key prefix/starting bound.
 * @returns {object} range
 *
 * @example
 * Table.createPrefixRange_('start');
 * // => {
 * //   start: 'start',
 * //   end: {
 * //     value: 'staru',
 * //     inclusive: false
 * //   }
 * // }
 */
Table.createPrefixRange_ = function(start) {
  var prefix = start.replace(new RegExp('[\xff]+$'), '');
  var endKey = '';

  if (prefix) {
    var position = prefix.length - 1;
    var charCode = prefix.charCodeAt(position);
    var nextChar = String.fromCharCode(charCode + 1);

    endKey = prefix.substring(0, position) + nextChar;
  }

  return {
    start: start,
    end: {
      value: endKey,
      inclusive: !endKey,
    },
  };
};

/**
 * Create a table.
 *
 * @param {object} [options] See {@link Instance#createTable}.
 * @param {object} [options.gaxOptions]  Request configuration options, outlined
 *     here: https://googleapis.github.io/gax-nodejs/global.html#CallOptions.
 *
 * @example
 * table.create(function(err, table, apiResponse) {
 *   if (!err) {
 *     // The table was created successfully.
 *   }
 * });
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.create().then(function(data) {
 *   var table = data[0];
 *   var apiResponse = data[1];
 * });
 */
Table.prototype.create = function(options, callback) {
  if (is.fn(options)) {
    callback = options;
    options = {};
  }

  this.instance.createTable(this.name, options, callback);
};

/**
 * Create a column family.
 *
 * Optionally you can send garbage collection rules and when creating a family.
 * Garbage collection executes opportunistically in the background, so it's
 * possible for reads to return a cell even if it matches the active expression
 * for its family.
 *
 * @see [Garbage Collection Proto Docs]{@link https://github.com/googleapis/googleapis/blob/master/google/bigtable/admin/table/v1/bigtable_table_data.proto#L59}
 *
 * @throws {error} If a name is not provided.
 *
 * @param {string} name The name of column family.
 * @param {object} [options] Configuration object.
 * @param {object} [options.gaxOptions] Request configuration options, outlined
 *     here: https://googleapis.github.io/gax-nodejs/global.html#CallOptions.
 * @param {object} [options.rule] Garbage collection rule
 * @param {object} [options.rule.age] Delete cells in a column older than the
 *     given age. Values must be at least 1 millisecond.
 * @param {number} [options.rule.versions] Maximum number of versions to delete
 *     cells in a column, except for the most recent.
 * @param {boolean} [options.rule.intersect] Cells to delete should match all
 *     rules.
 * @param {boolean} [options.rule.union] Cells to delete should match any of the
 *     rules.
 * @param {function} callback The callback function.
 * @param {?error} callback.err An error returned while making this request.
 * @param {Family} callback.family The newly created Family.
 * @param {object} callback.apiResponse The full API response.
 *
 * @example
 * const Bigtable = require('@google-cloud/bigtable');
 * const bigtable = new Bigtable();
 * const instance = bigtable.instance('my-instance');
 * const table = instance.table('prezzy');
 *
 * const callback = function(err, family, apiResponse) {
 *   // `family` is a Family object
 * };
 *
 * const rule = {
 *   age: {
 *     seconds: 0,
 *     nanos: 5000
 *   },
 *   versions: 3,
 *   union: true
 * };
 *
 * table.createFamily('follows', rule, callback);
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.createFamily('follows').then(function(data) {
 *   const family = data[0];
 *   const apiResponse = data[1];
 * });
 */
Table.prototype.createFamily = function(name, options, callback) {
  var self = this;

  if (is.function(options)) {
    callback = options;
    options = {};
  }

  if (!name) {
    throw new Error('A name is required to create a family.');
  }

  var mod = {
    id: name,
    create: {},
  };

  if (options.rule) {
    mod.create.gcRule = Family.formatRule_(options.rule);
  }

  var reqOpts = {
    name: this.id,
    modifications: [mod],
  };

  this.bigtable.request(
    {
      client: 'BigtableTableAdminClient',
      method: 'modifyColumnFamilies',
      reqOpts: reqOpts,
      gaxOpts: options.gaxOptions,
    },
    function(err, resp) {
      if (err) {
        callback(err, null, resp);
        return;
      }

      var family = self.family(resp.name);
      family.metadata = resp;

      callback(null, family, resp);
    }
  );
};

/**
 * Get {@link Row} objects for the rows currently in your table as a
 * readable object stream.
 *
 * @param {object} [options] Configuration object.
 * @param {boolean} [options.decode=true] If set to `false` it will not decode
 *     Buffer values returned from Bigtable.
 * @param {string} [options.end] End value for key range.
 * @param {Filter} [options.filter] Row filters allow you to
 *     both make advanced queries and format how the data is returned.
 * @param {object} [options.gaxOptions] Request configuration options, outlined
 *     here: https://googleapis.github.io/gax-nodejs/CallSettings.html.
 * @param {string[]} [options.keys] A list of row keys.
 * @param {number} [options.limit] Maximum number of rows to be returned.
 * @param {string} [options.prefix] Prefix that the row key must match.
 * @param {object[]} [options.ranges] A list of key ranges.
 * @param {string} [options.start] Start value for key range.
 * @returns {stream}
 *
 * @example
 * const Bigtable = require('@google-cloud/bigtable');
 * const bigtable = new Bigtable();
 * const instance = bigtable.instance('my-instance');
 * const table = instance.table('prezzy');
 *
 * table.createReadStream()
 *   .on('error', console.error)
 *   .on('data', function(row) {
 *     // `row` is a Row object.
 *   })
 *   .on('end', function() {
 *     // All rows retrieved.
 *   });
 *
 * //-
 * // If you anticipate many results, you can end a stream early to prevent
 * // unnecessary processing.
 * //-
 * table.createReadStream()
 *   .on('data', function(row) {
 *     this.end();
 *   });
 *
 * //-
 * // Specify arbitrary keys for a non-contiguous set of rows.
 * // The total size of the keys must remain under 1MB, after encoding.
 * //-
 * table.createReadStream({
 *   keys: [
 *     'alincoln',
 *     'gwashington'
 *   ]
 * });
 *
 * //-
 * // Scan for row keys that contain a specific prefix.
 * //-
 * table.createReadStream({
 *   prefix: 'gwash'
 * });
 *
 * //-
 * // Specify a contiguous range of rows to read by supplying `start` and `end`
 * // keys.
 * //
 * // If the `start` key is omitted, it is interpreted as an empty string.
 * // If the `end` key is omitted, it is interpreted as infinity.
 * //-
 * table.createReadStream({
 *   start: 'alincoln',
 *   end: 'gwashington'
 * });
 *
 * //-
 * // Specify multiple ranges.
 * //-
 * table.createReadStream({
 *   ranges: [{
 *     start: 'alincoln',
 *     end: 'gwashington'
 *   }, {
 *     start: 'tjefferson',
 *     end: 'jadams'
 *   }]
 * });
 *
 * //-
 * // Apply a {@link Filter} to the contents of the specified rows.
 * //-
 * table.createReadStream({
 *   filter: [
 *     {
 *       column: 'gwashington'
 *     }, {
 *       value: 1
 *     }
 *   ]
 * });
 */
Table.prototype.createReadStream = function(options) {
  var self = this;

  options = options || {};
  let maxRetries = is.number(this.maxRetries) ? this.maxRetries : 3;

  let rowKeys;
  let ranges = options.ranges || [];
  let filter;
  let rowsLimit;
  let rowsRead = 0;
  let numRequestsMade = 0;

  if (options.start || options.end) {
    ranges.push({
      start: options.start,
      end: options.end,
    });
  }

  if (options.keys) {
    rowKeys = options.keys;
  }

  if (options.prefix) {
    ranges.push(Table.createPrefixRange_(options.prefix));
  }

  if (options.filter) {
    filter = Filter.parse(options.filter);
  }

  if (options.limit) {
    rowsLimit = options.limit;
  }

  const userStream = through.obj();
  let chunkTransformer;

  const makeNewRequest = () => {
    let lastRowKey = chunkTransformer ? chunkTransformer.lastRowKey : '';
    chunkTransformer = new ChunkTransformer({decode: options.decode});

    var reqOpts = {
      tableName: this.id,
      appProfileId: this.bigtable.appProfileId,
    };

    var retryOpts = {
      currentRetryAttempt: numRequestsMade,
    };

    if (lastRowKey) {
      const lessThan = (lhs, rhs) => {
        const lhsBytes = Mutation.convertToBytes(lhs);
        const rhsBytes = Mutation.convertToBytes(rhs);
        return lhsBytes.compare(rhsBytes) === -1;
      };
      const greaterThan = (lhs, rhs) => lessThan(rhs, lhs);
      const greaterThanOrEqualTo = (lhs, rhs) => !lessThan(rhs, lhs);

      if (ranges.length === 0) {
        ranges.push({
          start: {
            value: lastRowKey,
            inclusive: false,
          },
        });
      } else {
        // Readjust and/or remove ranges based on previous valid row reads.

        // Iterate backward since items may need to be removed.
        for (let index = ranges.length - 1; index >= 0; index--) {
          const range = ranges[index];
          const startValue = is.object(range.start)
            ? range.start.value
            : range.start;
          const endValue = is.object(range.end) ? range.end.value : range.end;
          const isWithinStart =
            !startValue || greaterThanOrEqualTo(startValue, lastRowKey);
          const isWithinEnd = !endValue || lessThan(lastRowKey, endValue);
          if (isWithinStart) {
            if (isWithinEnd) {
              // The lastRowKey is within this range, adjust the start value.
              range.start = {
                value: lastRowKey,
                inclusive: false,
              };
            } else {
              // The lastRowKey is past this range, remove this range.
              ranges.splice(index, 1);
            }
          }
        }
      }

      // Remove rowKeys already read.
      if (rowKeys) {
        rowKeys = rowKeys.filter(rowKey => greaterThan(rowKey, lastRowKey));
        if (rowKeys.length === 0) {
          rowKeys = null;
        }
      }
    }
    if (rowKeys || ranges.length) {
      reqOpts.rows = {};

      if (rowKeys) {
        reqOpts.rows.rowKeys = rowKeys.map(Mutation.convertToBytes);
      }

      if (ranges.length) {
        reqOpts.rows.rowRanges = ranges.map(function(range) {
          return Filter.createRange(range.start, range.end, 'Key');
        });
      }
    }

    if (filter) {
      reqOpts.filter = filter;
    }

    if (rowsLimit) {
      reqOpts.rowsLimit = rowsLimit - rowsRead;
    }

    const requestStream = this.bigtable.request({
      client: 'BigtableClient',
      method: 'readRows',
      reqOpts: reqOpts,
      gaxOpts: options.gaxOptions,
      retryOpts: retryOpts,
    });

    requestStream.on('request', () => numRequestsMade++);

    const rowStream = pumpify.obj([
      requestStream,
      chunkTransformer,
      through.obj(function(rowData, enc, next) {
        if (chunkTransformer._destroyed || userStream._writableState.ended) {
          return next();
        }
        numRequestsMade = 0;
        rowsRead++;
        const row = self.row(rowData.key);
        row.data = rowData.data;
        next(null, row);
      }),
    ]);

    rowStream.on('error', error => {
      rowStream.unpipe(userStream);
      if (
        numRequestsMade <= maxRetries &&
        RETRYABLE_STATUS_CODES.has(error.code)
      ) {
        makeNewRequest();
      } else {
        userStream.emit('error', error);
      }
    });
    rowStream.pipe(userStream);
  };

  makeNewRequest();
  return userStream;
};

/**
 * Delete the table.
 *
 * @param {object} [gaxOptions] Request configuration options, outlined
 *     here: https://googleapis.github.io/gax-nodejs/CallSettings.html.
 * @param {function} [callback] The callback function.
 * @param {?error} callback.err An error returned while making this
 *     request.
 * @param {object} callback.apiResponse The full API response.
 *
 * @example
 * table.delete(function(err, apiResponse) {});
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.delete().then(function(data) {
 *   var apiResponse = data[0];
 * });
 */
Table.prototype.delete = function(gaxOptions, callback) {
  if (is.fn(gaxOptions)) {
    callback = gaxOptions;
    gaxOptions = {};
  }

  this.bigtable.request(
    {
      client: 'BigtableTableAdminClient',
      method: 'deleteTable',
      reqOpts: {
        name: this.id,
      },
      gaxOpts: gaxOptions,
    },
    callback
  );
};

/**
 * Delete all rows in the table, optionally corresponding to a particular
 * prefix.
 *
 * @param {object} [options] Configuration object.
 * @param {string} [options.prefix] Row key prefix, when omitted all rows
 *     will be deleted.
 * @param {function} callback The callback function.
 * @param {?error} callback.err An error returned while making this request.
 * @param {object} callback.apiResponse The full API response.
 *
 * @example
 * const Bigtable = require('@google-cloud/bigtable');
 * const bigtable = new Bigtable();
 * const instance = bigtable.instance('my-instance');
 * const table = instance.table('prezzy');
 *
 * //-
 * // You can supply a prefix to delete all corresponding rows.
 * //-
 * const callback = function(err, apiResponse) {
 *   if (!err) {
 *     // Rows successfully deleted.
 *   }
 * };
 *
 * table.deleteRows({
 *   prefix: 'alincoln'
 * }, callback);
 *
 * //-
 * // If you choose to omit the prefix, all rows in the table will be deleted.
 * //-
 * table.deleteRows(callback);
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.deleteRows().then(function(data) {
 *   const apiResponse = data[0];
 * });
 */
Table.prototype.deleteRows = function(options, callback) {
  if (is.function(options)) {
    callback = options;
    options = {};
  }

  var reqOpts = {
    name: this.id,
  };

  if (options.prefix) {
    reqOpts.rowKeyPrefix = Mutation.convertToBytes(options.prefix);
  } else {
    reqOpts.deleteAllDataFromTable = true;
  }

  this.bigtable.request(
    {
      client: 'BigtableTableAdminClient',
      method: 'dropRowRange',
      reqOpts: reqOpts,
      gaxOpts: options.gaxOptions,
    },
    callback
  );
};

/**
 * Check if a table exists.
 *
 * @param {object} [gaxOptions] Request configuration options, outlined
 *     here: https://googleapis.github.io/gax-nodejs/CallSettings.html.
 * @param {function} callback The callback function.
 * @param {?error} callback.err An error returned while making this
 *     request.
 * @param {boolean} callback.exists Whether the table exists or not.
 *
 * @example
 * table.exists(function(err, exists) {});
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.exists().then(function(data) {
 *   var exists = data[0];
 * });
 */
Table.prototype.exists = function(gaxOptions, callback) {
  if (is.fn(gaxOptions)) {
    callback = gaxOptions;
    gaxOptions = {};
  }

  this.getMetadata(gaxOptions, function(err) {
    if (err) {
      if (err.code === 5) {
        callback(null, false);
        return;
      }

      callback(err);
      return;
    }

    callback(null, true);
  });
};

/**
 * Get a reference to a Table Family.
 *
 * @throws {error} If a name is not provided.
 *
 * @param {string} name The family name.
 * @returns {Family}
 *
 * @example
 * const family = table.family('my-family');
 */
Table.prototype.family = function(name) {
  if (!name) {
    throw new Error('A family name must be provided.');
  }

  return new Family(this, name);
};

/**
 * Get a table if it exists.
 *
 * You may optionally use this to "get or create" an object by providing an
 * object with `autoCreate` set to `true`. Any extra configuration that is
 * normally required for the `create` method must be contained within this
 * object as well.
 *
 * @param {object} [options] Configuration object.
 * @param {boolean} [options.autoCreate=false] Automatically create the
 *     instance if it does not already exist.
 * @param {object} [options.gaxOptions] Request configuration options, outlined
 *     here: https://googleapis.github.io/gax-nodejs/CallSettings.html.
 * @param {?error} callback.error An error returned while making this request.
 * @param {Table} callback.table The Table object.
 * @param {object} callback.apiResponse The resource as it exists in the API.
 *
 * @example
 * table.get(function(err, table, apiResponse) {
 *   // The `table` data has been populated.
 * });
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.get().then(function(data) {
 *   var table = data[0];
 *   var apiResponse = data[0];
 * });
 */
Table.prototype.get = function(options, callback) {
  var self = this;

  if (is.fn(options)) {
    callback = options;
    options = {};
  }

  var autoCreate = !!options.autoCreate;
  var gaxOptions = options.gaxOptions;

  this.getMetadata({gaxOptions}, function(err, metadata) {
    if (err) {
      if (err.code === 5 && autoCreate) {
        self.create({gaxOptions}, callback);
        return;
      }

      callback(err);
      return;
    }

    callback(null, self, metadata);
  });
};

/**
 * Get Family objects for all the column familes in your table.
 *
 * @param {object} [gaxOptions] Request configuration options, outlined here:
 *     https://googleapis.github.io/gax-nodejs/CallSettings.html.
 * @param {function} callback The callback function.
 * @param {?error} callback.err An error returned while making this request.
 * @param {Family[]} callback.families The list of families.
 * @param {object} callback.apiResponse The full API response.
 *
 * @example
 * const Bigtable = require('@google-cloud/bigtable');
 * const bigtable = new Bigtable();
 * const instance = bigtable.instance('my-instance');
 * const table = instance.table('prezzy');
 *
 * table.getFamilies(function(err, families, apiResponse) {
 *   // `families` is an array of Family objects.
 * });
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.getFamilies().then(function(data) {
 *   var families = data[0];
 *   var apiResponse = data[1];
 * });
 */
Table.prototype.getFamilies = function(gaxOptions, callback) {
  var self = this;

  if (is.fn(gaxOptions)) {
    callback = gaxOptions;
    gaxOptions = {};
  }

  this.getMetadata({gaxOptions}, function(err, metadata) {
    if (err) {
      callback(err);
      return;
    }

    var families = Object.keys(metadata.columnFamilies).map(function(familyId) {
      var family = self.family(familyId);
      family.metadata = metadata.columnFamilies[familyId];
      return family;
    });

    callback(null, families, metadata.columnFamilies);
  });
};

/**
 * Get the table's metadata.
 *
 * @param {object} [options] Table request options.
 * @param {object} [options.gaxOptions] Request configuration options, outlined
 *     here: https://googleapis.github.io/gax-nodejs/CallSettings.html.
 * @param {string} [options.view] The view to be applied to the table fields.
 * @param {function} [callback] The callback function.
 * @param {?error} callback.err An error returned while making this
 *     request.
 * @param {object} callback.metadata The table's metadata.
 *
 * @example
 * const Bigtable = require('@google-cloud/bigtable');
 * const bigtable = new Bigtable();
 * const instance = bigtable.instance('my-instance');
 * const table = instance.table('prezzy');
 *
 * table.getMetadata(function(err, metadata) {});
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.getMetadata().then(function(data) {
 *   var metadata = data[0];
 *   var apiResponse = data[1];
 * });
 */
Table.prototype.getMetadata = function(options, callback) {
  var self = this;

  if (is.function(options)) {
    callback = options;
    options = {};
  }

  var reqOpts = {
    name: this.id,
    view: Table.VIEWS[options.view || 'unspecified'],
  };

  this.bigtable.request(
    {
      client: 'BigtableTableAdminClient',
      method: 'getTable',
      reqOpts: reqOpts,
      gaxOpts: options.gaxOptions,
    },
    function() {
      if (arguments[1]) {
        self.metadata = arguments[1];
      }

      callback.apply(null, arguments);
    }
  );
};

/**
 * Get {@link Row} objects for the rows currently in your table.
 *
 * This method is not recommended for large datasets as it will buffer all rows
 * before returning the results. Instead we recommend using the streaming API
 * via {@link Table#createReadStream}.
 *
 * @param {object} [options] Configuration object. See
 *     {@link Table#createReadStream} for a complete list of options.
 * @param {object} [options.gaxOptions] Request configuration options, outlined
 *     here: https://googleapis.github.io/gax-nodejs/CallSettings.html.
 * @param {function} callback The callback function.
 * @param {?error} callback.err An error returned while making this request.
 * @param {Row[]} callback.rows List of Row objects.
 *
 * @example
 * const Bigtable = require('@google-cloud/bigtable');
 * const bigtable = new Bigtable();
 * const instance = bigtable.instance('my-instance');
 * const table = instance.table('prezzy');
 *
 * table.getRows(function(err, rows) {
 *   if (!err) {
 *     // `rows` is an array of Row objects.
 *   }
 * });
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.getRows().then(function(data) {
 *   var rows = data[0];
 * });
 */
Table.prototype.getRows = function(options, callback) {
  if (is.function(options)) {
    callback = options;
    options = {};
  }

  this.createReadStream(options)
    .on('error', callback)
    .pipe(
      concat(function(rows) {
        callback(null, rows);
      })
    );
};

/**
 * Insert or update rows in your table. It should be noted that gRPC only allows
 * you to send payloads that are less than or equal to 4MB. If you're inserting
 * more than that you may need to send smaller individual requests.
 *
 * @param {object|object[]} entries List of entries to be inserted.
 *     See {@link Table#mutate}.
 * @param {object} [gaxOptions] Request configuration options, outlined here:
 *     https://googleapis.github.io/gax-nodejs/CallSettings.html.
 * @param {function} callback The callback function.
 * @param {?error} callback.err An error returned while making this request.
 * @param {object[]} callback.err.errors If present, these represent partial
 *     failures. It's possible for part of your request to be completed
 *     successfully, while the other part was not.
 *
 * @example
 * const Bigtable = require('@google-cloud/bigtable');
 * const bigtable = new Bigtable();
 * const instance = bigtable.instance('my-instance');
 * const table = instance.table('prezzy');
 *
 * const callback = function(err) {
 *   if (err) {
 *     // An API error or partial failure occurred.
 *
 *     if (err.name === 'PartialFailureError') {
 *       // err.errors[].code = 'Response code'
 *       // err.errors[].message = 'Error message'
 *       // err.errors[].entry = The original entry
 *     }
 *   }
 * };
 *
 * const entries = [
 *  {
 *     key: 'alincoln',
 *     data: {
 *       follows: {
 *         gwashington: 1
 *       }
 *     }
 *   }
 * ];
 *
 * table.insert(entries, callback);
 *
 * //-
 * // By default whenever you insert new data, the server will capture a
 * // timestamp of when your data was inserted. It's possible to provide a
 * // date object to be used instead.
 * //-
 * const entries = [
 *   {
 *     key: 'gwashington',
 *     data: {
 *       follows: {
 *         jadams: {
 *           value: 1,
 *           timestamp: new Date('March 22, 2016')
 *         }
 *       }
 *     }
 *   }
 * ];
 *
 * table.insert(entries, callback);
 *
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.insert(entries).then(function() {
 *   // All requested inserts have been processed.
 * });
 * //-
 */
Table.prototype.insert = function(entries, gaxOptions, callback) {
  if (is.fn(gaxOptions)) {
    callback = gaxOptions;
    gaxOptions = {};
  }

  entries = arrify(entries).map(propAssign('method', Mutation.methods.INSERT));

  return this.mutate(entries, gaxOptions, callback);
};

/**
 * Apply a set of changes to be atomically applied to the specified row(s).
 * Mutations are applied in order, meaning that earlier mutations can be masked
 * by later ones.
 *
 * @param {object|object[]} entries List of entities to be inserted or
 *     deleted.
 * @param {object} [gaxOptions] Request configuration options, outlined here:
 *     https://googleapis.github.io/gax-nodejs/CallSettings.html.
 * @param {function} callback The callback function.
 * @param {?error} callback.err An error returned while making this request.
 * @param {object[]} callback.err.errors If present, these represent partial
 *     failures. It's possible for part of your request to be completed
 *     successfully, while the other part was not.
 *
 * @example
 * const Bigtable = require('@google-cloud/bigtable');
 * const bigtable = new Bigtable();
 * const instance = bigtable.instance('my-instance');
 * const table = instance.table('prezzy');
 *
 * //-
 * // Insert entities. See {@link Table#insert}.
 * //-
 * const callback = function(err) {
 *   if (err) {
 *     // An API error or partial failure occurred.
 *
 *     if (err.name === 'PartialFailureError') {
 *       // err.errors[].code = 'Response code'
 *       // err.errors[].message = 'Error message'
 *       // err.errors[].entry = The original entry
 *     }
 *   }
 * };
 *
 * const entries = [
 *   {
 *     method: 'insert',
 *     key: 'gwashington',
 *     data: {
 *       follows: {
 *         jadams: 1
 *       }
 *     }
 *   }
 * ];
 *
 * table.mutate(entries, callback);
 *
 * //-
 * // Delete entities. See {@link Row#deleteCells}.
 * //-
 * const entries = [
 *   {
 *     method: 'delete',
 *     key: 'gwashington'
 *   }
 * ];
 *
 * table.mutate(entries, callback);
 *
 * //-
 * // Delete specific columns within a row.
 * //-
 * const entries = [
 *   {
 *     method: 'delete',
 *     key: 'gwashington',
 *     data: [
 *       'follows:jadams'
 *     ]
 *   }
 * ];
 *
 * table.mutate(entries, callback);
 *
 * //-
 * // Mix and match mutations. This must contain at least one entry and at
 * // most 100,000.
 * //-
 * const entries = [
 *   {
 *     method: 'insert',
 *     key: 'alincoln',
 *     data: {
 *       follows: {
 *         gwashington: 1
 *       }
 *     }
 *   }, {
 *     method: 'delete',
 *     key: 'jadams',
 *     data: [
 *       'follows:gwashington'
 *     ]
 *   }
 * ];
 *
 * table.mutate(entries, callback);
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.mutate(entries).then(function() {
 *   // All requested mutations have been processed.
 * });
 */
Table.prototype.mutate = function(entries, gaxOptions, callback) {
  var self = this;

  if (is.fn(gaxOptions)) {
    callback = gaxOptions;
    gaxOptions = {};
  }

  entries = flatten(arrify(entries));

  var numRequestsMade = 0;

  var maxRetries = is.number(this.maxRetries) ? this.maxRetries : 3;
  var pendingEntryIndices = new Set(entries.map((entry, index) => index));
  var entryToIndex = new Map(entries.map((entry, index) => [entry, index]));
  var mutationErrorsByEntryIndex = new Map();

  function onBatchResponse(err) {
    if (err) {
      // The error happened before a request was even made, don't retry.
      callback(err);
      return;
    }
    if (pendingEntryIndices.size !== 0 && numRequestsMade <= maxRetries) {
      makeNextBatchRequest();
      return;
    }

    if (mutationErrorsByEntryIndex.size !== 0) {
      var mutationErrors = Array.from(mutationErrorsByEntryIndex.values());
      err = new common.util.PartialFailureError({
        errors: mutationErrors,
      });
    }

    callback(err);
  }

  function makeNextBatchRequest() {
    var entryBatch = entries.filter((entry, index) => {
      return pendingEntryIndices.has(index);
    });

    var reqOpts = {
      tableName: self.id,
      appProfileId: self.bigtable.appProfileId,
      entries: entryBatch.map(Mutation.parse),
    };

    var retryOpts = {
      currentRetryAttempt: numRequestsMade,
    };

    self.bigtable
      .request({
        client: 'BigtableClient',
        method: 'mutateRows',
        reqOpts: reqOpts,
        gaxOpts: gaxOptions,
        retryOpts: retryOpts,
      })
      .on('request', () => numRequestsMade++)
      .on('error', err => {
        if (numRequestsMade === 0) {
          callback(err); // Likely a "projectId not detected" error.
          return;
        }

        onBatchResponse(err);
      })
      .on('data', function(obj) {
        obj.entries.forEach(function(entry) {
          var originalEntry = entryBatch[entry.index];
          var originalEntriesIndex = entryToIndex.get(originalEntry);

          // Mutation was successful.
          if (entry.status.code === 0) {
            pendingEntryIndices.delete(originalEntriesIndex);
            mutationErrorsByEntryIndex.delete(originalEntriesIndex);
            return;
          }

          if (!RETRYABLE_STATUS_CODES.has(entry.status.code)) {
            pendingEntryIndices.delete(originalEntriesIndex);
          }

          var status = commonGrpc.Service.decorateStatus_(entry.status);
          status.entry = originalEntry;

          mutationErrorsByEntryIndex.set(originalEntriesIndex, status);
        });
      })
      .on('end', onBatchResponse);
  }

  makeNextBatchRequest();
};

/**
 * Get a reference to a table row.
 *
 * @throws {error} If a key is not provided.
 *
 * @param {string} key The row key.
 * @returns {Row}
 *
 * @example
 * var row = table.row('lincoln');
 */
Table.prototype.row = function(key) {
  if (!key) {
    throw new Error('A row key must be provided.');
  }

  return new Row(this, key);
};

/**
 * Returns a sample of row keys in the table. The returned row keys will delimit
 * contigous sections of the table of approximately equal size, which can be
 * used to break up the data for distributed tasks like mapreduces.
 *
 * @param {object} [gaxOptions] Request configuration options, outlined here:
 *     https://googleapis.github.io/gax-nodejs/CallSettings.html.
 * @param {function} [callback] The callback function.
 * @param {?error} callback.err An error returned while making this request.
 * @param {object[]} callback.keys The list of keys.
 *
 * @example
 * table.sampleRowKeys(function(err, keys) {
 *   // keys = [
 *   //   {
 *   //     key: '',
 *   //     offset: '805306368'
 *   //   },
 *   //   ...
 *   // ]
 * });
 *
 * //-
 * // If the callback is omitted, we'll return a Promise.
 * //-
 * table.sampleRowKeys().then(function(data) {
 *   var keys = data[0];
 * });
 */
Table.prototype.sampleRowKeys = function(gaxOptions, callback) {
  if (is.fn(gaxOptions)) {
    callback = gaxOptions;
    gaxOptions = {};
  }

  this.sampleRowKeysStream(gaxOptions)
    .on('error', callback)
    .pipe(
      concat(function(keys) {
        callback(null, keys);
      })
    );
};

/**
 * Returns a sample of row keys in the table as a readable object stream.
 *
 * See {@link Table#sampleRowKeys} for more details.
 *
 * @param {object} [gaxOptions] Request configuration options, outlined here:
 *     https://googleapis.github.io/gax-nodejs/CallSettings.html.
 * @returns {stream}
 *
 * @example
 * table.sampleRowKeysStream()
 *   .on('error', console.error)
 *   .on('data', function(key) {
 *     // Do something with the `key` object.
 *   });
 *
 * //-
 * // If you anticipate many results, you can end a stream early to prevent
 * // unnecessary processing.
 * //-
 * table.sampleRowKeysStream()
 *   .on('data', function(key) {
 *     this.end();
 *   });
 */
Table.prototype.sampleRowKeysStream = function(gaxOptions) {
  var reqOpts = {
    tableName: this.id,
    appProfileId: this.bigtable.appProfileId,
  };

  return pumpify.obj([
    this.bigtable.request({
      client: 'BigtableClient',
      method: 'sampleRowKeys',
      reqOpts: reqOpts,
      gaxOpts: gaxOptions,
    }),
    through.obj(function(key, enc, next) {
      next(null, {
        key: key.rowKey,
        offset: key.offsetBytes,
      });
    }),
  ]);
};

/*! Developer Documentation
 *
 * All async methods (except for streams) will return a Promise in the event
 * that a callback is omitted.
 */
common.util.promisifyAll(Table, {
  exclude: ['family', 'row'],
});

module.exports = Table;