import org.apache.kudu.client.RowResult; //导入依赖的package包/类
@Override
public Iterable<Row> getExistingForFilters(Iterable<Row> filters) throws Exception {
  List<Row> existingForFilters = Lists.newArrayList();

  if (!filters.iterator().hasNext()) {
    return existingForFilters;
  }

  KuduTable table = connectToTable();
  KuduScanner scanner = scannerForFilters(filters, table);

  long startTime = System.nanoTime();
  while (scanner.hasMoreRows()) {
    for (RowResult rowResult : scanner.nextRows()) {
      Row existing = resultAsRow(rowResult, table);

      existingForFilters.add(existing);
    }
  }
  long endTime = System.nanoTime();
  if (hasAccumulators()) {
    accumulators.getDoubleAccumulators().get(ACCUMULATOR_SECONDS_SCANNING).add((endTime - startTime) / 1000.0 / 1000.0 / 1000.0);
  }

  return existingForFilters;
}
 

import org.apache.kudu.client.RowResult; //导入依赖的package包/类
public long countNumRowsInTable() throws Exception
{
  List<String> allProjectedCols = new ArrayList<>(
      unitTestStepwiseScanInputOperator.getKuduColNameToSchemaMapping().keySet());
  KuduScanner scanner = kuduClient.newScannerBuilder(kuduTable)
      .setProjectedColumnNames(allProjectedCols)
      .build();
  long counter = 0;
  while (scanner.hasMoreRows()) {
    RowResultIterator rowResultItr = scanner.nextRows();
    while (rowResultItr.hasNext()) {
      RowResult thisRow = rowResultItr.next();
      counter += 1;
    }
  }
  return counter;
}
 

import org.apache.kudu.client.RowResult; //导入依赖的package包/类
protected void lookUpAndPopulateRecord(UnitTestTablePojo keyInfo) throws Exception
{
  KuduScanner scanner = kuduClient.newScannerBuilder(kuduTable)
      .addPredicate(KuduPredicate.newComparisonPredicate(columnDefs.get("introwkey"),
      KuduPredicate.ComparisonOp.EQUAL,keyInfo.getIntrowkey()))
      .addPredicate(KuduPredicate.newComparisonPredicate(columnDefs.get("stringrowkey"),
      KuduPredicate.ComparisonOp.EQUAL,keyInfo.getStringrowkey()))
      .addPredicate(KuduPredicate.newComparisonPredicate(columnDefs.get("timestamprowkey"),
      KuduPredicate.ComparisonOp.EQUAL,keyInfo.getTimestamprowkey()))
      .build();
  RowResultIterator rowResultItr = scanner.nextRows();
  while (rowResultItr.hasNext()) {
    RowResult thisRow = rowResultItr.next();
    keyInfo.setFloatdata(thisRow.getFloat("floatdata"));
    keyInfo.setBooldata(thisRow.getBoolean("booldata"));
    keyInfo.setBinarydata(thisRow.getBinary("binarydata"));
    keyInfo.setLongdata(thisRow.getLong("longdata"));
    keyInfo.setTimestampdata(thisRow.getLong("timestampdata"));
    keyInfo.setStringdata("stringdata");
    break;
  }
}
 

import org.apache.kudu.client.RowResult; //导入依赖的package包/类
private Row resultAsRow(RowResult result, KuduTable table) throws KuduException {
  List<Object> values = Lists.newArrayList();

  for (ColumnSchema columnSchema : table.getSchema().getColumns()) {
    String columnName = columnSchema.getName();

    if (result.isNull(columnName)) {
      values.add(null);
      continue;
    }

    switch (columnSchema.getType()) {
      case DOUBLE:
        values.add(result.getDouble(columnName));
        break;
      case FLOAT:
        values.add(result.getFloat(columnName));
        break;
      case INT32:
        values.add(result.getInt(columnName));
        break;
      case INT64:
        values.add(result.getLong(columnName));
        break;
      case STRING:
        values.add(result.getString(columnName));
        break;
      case BOOL:
        values.add(result.getBoolean(columnName));
        break;
      default:
        throw new RuntimeException("Unsupported Kudu column type: " + columnSchema.getType());
    }
  }

  Row row = new RowWithSchema(getTableSchema(table), values.toArray());

  return row;
}
 

import org.apache.kudu.client.RowResult; //导入依赖的package包/类
/**
 * Retrieves the tagset IDs of all tagsets which contain the specified tag.
 * The tagset IDs are returned in sorted order.
 *
 * @param key the tag key
 * @param value the tag value
 * @return the sorted tagset IDs
 */
public Deferred<IntVec> getTagsetIDsForTag(final String key, final String value) {
  AsyncKuduScanner.AsyncKuduScannerBuilder scan = client.newScannerBuilder(table);
  scan.addPredicate(KuduPredicate.newComparisonPredicate(Tables.TAGS_KEY_COLUMN,
                                                         ComparisonOp.EQUAL, key));
  scan.addPredicate(KuduPredicate.newComparisonPredicate(Tables.TAGS_VALUE_COLUMN,
                                                         ComparisonOp.EQUAL, value));
  scan.setProjectedColumnIndexes(TAGSET_ID_PROJECTION);
  final AsyncKuduScanner scanner = scan.build();

  class GetTagCB implements Callback<Deferred<IntVec>, RowResultIterator> {
    private final IntVec tagsetIDs = IntVec.create();
    @Override
    public Deferred<IntVec> call(RowResultIterator results) {
      for (RowResult result : results) {
        tagsetIDs.push(result.getInt(0));
      }
      if (scanner.hasMoreRows()) {
        return scanner.nextRows().addCallbackDeferring(this);
      }
      // The Kudu java client doesn't yet allow us to specify a sorted
      // (fault-tolerant) scan, so have to sort manually.
      tagsetIDs.sort();
      return Deferred.fromResult(tagsetIDs);
    }
    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this).add("key", key).add("value", value).toString();
    }
  }

  return scanner.nextRows().addCallbackDeferring(new GetTagCB());
}
 

import org.apache.kudu.client.RowResult; //导入依赖的package包/类
@Override
public Deferred<TagsetLookupResult> call(RowResultIterator rows) {
  for (RowResult row : rows) {
    int id = row.getInt(Tables.TAGSETS_ID_INDEX);
    Preconditions.checkState(id >= probe);
    if (id != probe) {
      // We found a hole in the table where we expected the tagset.
      return Deferred.fromResult(new TagsetLookupResult(false, probe));
    }

    if (tagset.equals(row.getBinary(Tables.TAGSETS_TAGSET_INDEX))) {
      return Deferred.fromResult(new TagsetLookupResult(true, id));
    }

    probe++;
  }

  // We probed through the entire RowResult and didn't find the tagset.
  if (!scanner.hasMoreRows()) {
    if (probe <= Ints.saturatedCast((long) id + TAGSETS_PER_SCAN)) {
      // We found a hole at the end of the scan.
      return Deferred.fromResult(new TagsetLookupResult(false, probe));
    }
    // The current scanner has been exhausted; create a new scanner from the
    // latest probe point.
    scanner = tagsetScanner(probe);
    id = probe;
  }
  return scanner.nextRows().addCallbackDeferring(this);
}
 

import org.apache.kudu.client.RowResult; //导入依赖的package包/类
public void truncateTable() throws Exception
{
  AbstractKuduPartitionScanner<UnitTestTablePojo,InputOperatorControlTuple> scannerForDeletingRows =
      unitTestStepwiseScanInputOperator.getScanner();
  List<KuduScanToken> scansForAllTablets = unitTestStepwiseScanInputOperator
      .getPartitioner().getKuduScanTokensForSelectAllColumns();
  ApexKuduConnection aCurrentConnection = scannerForDeletingRows.getConnectionPoolForThreads().get(0);
  KuduSession aSessionForDeletes = aCurrentConnection.getKuduClient().newSession();
  KuduTable currentTable = aCurrentConnection.getKuduTable();
  for ( KuduScanToken aTabletScanToken : scansForAllTablets) {
    KuduScanner aScanner = aTabletScanToken.intoScanner(aCurrentConnection.getKuduClient());
    while ( aScanner.hasMoreRows()) {
      RowResultIterator itrForRows = aScanner.nextRows();
      while ( itrForRows.hasNext()) {
        RowResult aRow = itrForRows.next();
        int intRowKey = aRow.getInt("introwkey");
        String stringRowKey = aRow.getString("stringrowkey");
        long timestampRowKey = aRow.getLong("timestamprowkey");
        Delete aDeleteOp = currentTable.newDelete();
        aDeleteOp.getRow().addInt("introwkey",intRowKey);
        aDeleteOp.getRow().addString("stringrowkey", stringRowKey);
        aDeleteOp.getRow().addLong("timestamprowkey",timestampRowKey);
        aSessionForDeletes.apply(aDeleteOp);
      }
    }
  }
  aSessionForDeletes.close();
  Thread.sleep(2000); // Sleep to allow for scans to complete
}
 

import org.apache.kudu.client.RowResult; //导入依赖的package包/类
/**
 * Create a field and assign a value off of RowResult.
 * @param result Result obtained from scan
 * @param fieldName Field name to create
 * @param type Kudu Type for the field
 * @return Generated field
 * @throws StageException
 */
public static Field createField(RowResult result, String fieldName, Type type) throws StageException {
  switch (type) {
    case INT8:
      return Field.create(Field.Type.BYTE, result.getByte(fieldName));
    case INT16:
      return Field.create(Field.Type.SHORT, result.getShort(fieldName));
    case INT32:
      return Field.create(Field.Type.INTEGER, result.getInt(fieldName));
    case INT64:
      return Field.create(Field.Type.LONG, result.getLong(fieldName));
    case BINARY:
      try {
        return Field.create(Field.Type.BYTE_ARRAY, result.getBinary(fieldName));
      } catch (IllegalArgumentException ex) {
        throw new OnRecordErrorException(Errors.KUDU_35, fieldName);
      }
    case STRING:
      return Field.create(Field.Type.STRING, result.getString(fieldName));
    case BOOL:
      return Field.create(Field.Type.BOOLEAN, result.getBoolean(fieldName));
    case FLOAT:
      return Field.create(Field.Type.FLOAT, result.getFloat(fieldName));
    case DOUBLE:
      return Field.create(Field.Type.DOUBLE, result.getDouble(fieldName));
    case UNIXTIME_MICROS:
      //UNIXTIME_MICROS is in microsecond
      return Field.create(Field.Type.DATETIME, new Date(result.getLong(fieldName)/1000L));
    default:
      throw new StageException(Errors.KUDU_10, fieldName, type.getName());
  }
}
 

import org.apache.kudu.client.RowResult; //导入依赖的package包/类
public static void createKuduTable(String tableName, int tablets, int replicas, int rows) throws Exception {

    try (KuduClient client = new KuduClient.KuduClientBuilder(KUDU_MASTER).build()) {

      ListTablesResponse tables = client.getTablesList(tableName);
      if (!tables.getTablesList().isEmpty()) {
        client.deleteTable(tableName);
      }

      List<ColumnSchema> columns = new ArrayList<>(5);
      columns.add(new ColumnSchema.ColumnSchemaBuilder("key", Type.INT32).key(true).build());
      columns.add(new ColumnSchema.ColumnSchemaBuilder("binary", Type.BINARY).nullable(false).build());
      columns.add(new ColumnSchema.ColumnSchemaBuilder("boolean", Type.BOOL).nullable(true).build());
      columns.add(new ColumnSchema.ColumnSchemaBuilder("float", Type.FLOAT).nullable(false).build());
      columns.add(new ColumnSchema.ColumnSchemaBuilder("string", Type.STRING).nullable(true).build());

      Schema schema = new Schema(columns);

      CreateTableOptions builder = new CreateTableOptions();
      builder.setNumReplicas(replicas);
      builder.setRangePartitionColumns(Arrays.asList("key"));
      for (int i = 1; i < tablets; i++) {
        PartialRow splitRow = schema.newPartialRow();
        splitRow.addInt("key", i*1000);
        builder.addSplitRow(splitRow);
      }

      client.createTable(tableName, schema, builder);

      KuduTable table = client.openTable(tableName);

      KuduSession session = client.newSession();
      session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
      for (int i = 0; i < rows; i++) {
        Insert insert = table.newInsert();
        PartialRow row = insert.getRow();
        row.addInt(0, i);
        row.addBinary(1, ("Row " + i).getBytes());
        row.addBoolean(2, i % 2 == 0);
        row.addFloat(3, i + 0.01f);
        row.addString(4, ("Row " + i));
        session.apply(insert);
      }

      List<String> projectColumns = new ArrayList<>(1);
      projectColumns.add("float");
      KuduScanner scanner = client.newScannerBuilder(table)
          .setProjectedColumnNames(projectColumns)
          .build();
      while (scanner.hasMoreRows()) {
        RowResultIterator results = scanner.nextRows();
        while (results.hasNext()) {
          RowResult result = results.next();
          System.out.println(result.toStringLongFormat());
        }
      }
    }
  }
 

import org.apache.kudu.client.RowResult; //导入依赖的package包/类
public void setValuesInPOJO(RowResult aRow, T payload)
{
  Set<String> columnsUsed = parsedQuery.getKuduSQLParseTreeListener().getListOfColumnsUsed();
  for (String aColumnName : columnsUsed) {
    ColumnSchema schemaForThisColumn = tableSchema.get(aColumnName);
    if (aRow.isNull(aColumnName)) {
      continue;
    }
    switch ( schemaForThisColumn.getType().getDataType().getNumber()) {
      case Common.DataType.BINARY_VALUE:
        ((PojoUtils.Setter<T,ByteBuffer>)settersForThisQueryScan.get(aColumnName)).set(
            payload,aRow.getBinary(aColumnName));
        break;
      case Common.DataType.STRING_VALUE:
        ((PojoUtils.Setter<T,String>)settersForThisQueryScan.get(aColumnName)).set(
            payload,aRow.getString(aColumnName));
        break;
      case Common.DataType.BOOL_VALUE:
        ((PojoUtils.SetterBoolean<T>)settersForThisQueryScan.get(aColumnName)).set(
            payload,aRow.getBoolean(aColumnName));
        break;
      case Common.DataType.DOUBLE_VALUE:
        ((PojoUtils.SetterDouble<T>)settersForThisQueryScan.get(aColumnName)).set(
            payload,aRow.getDouble(aColumnName));
        break;
      case Common.DataType.FLOAT_VALUE:
        ((PojoUtils.SetterFloat<T>)settersForThisQueryScan.get(aColumnName)).set(
            payload,aRow.getFloat(aColumnName));
        break;
      case Common.DataType.INT8_VALUE:
        ((PojoUtils.SetterByte<T>)settersForThisQueryScan.get(aColumnName)).set(
            payload,aRow.getByte(aColumnName));
        break;
      case Common.DataType.INT16_VALUE:
        ((PojoUtils.SetterShort<T>)settersForThisQueryScan.get(aColumnName)).set(
            payload,aRow.getShort(aColumnName));
        break;
      case Common.DataType.INT32_VALUE:
        ((PojoUtils.SetterInt<T>)settersForThisQueryScan.get(aColumnName)).set(
            payload,aRow.getInt(aColumnName));
        break;
      case Common.DataType.UNIXTIME_MICROS_VALUE:
      case Common.DataType.INT64_VALUE:
        ((PojoUtils.SetterLong<T>)settersForThisQueryScan.get(aColumnName)).set(
            payload,aRow.getLong(aColumnName));
        break;
      case Common.DataType.UINT8_VALUE:
        LOG.error("Unsigned int 8 not supported yet");
        throw new RuntimeException("uint8 not supported in Kudu schema yet");
      case Common.DataType.UINT16_VALUE:
        LOG.error("Unsigned int 16 not supported yet");
        throw new RuntimeException("uint16 not supported in Kudu schema yet");
      case Common.DataType.UINT32_VALUE:
        LOG.error("Unsigned int 32 not supported yet");
        throw new RuntimeException("uint32 not supported in Kudu schema yet");
      case Common.DataType.UINT64_VALUE:
        LOG.error("Unsigned int 64 not supported yet");
        throw new RuntimeException("uint64 not supported in Kudu schema yet");
      case Common.DataType.UNKNOWN_DATA_VALUE:
        LOG.error("unknown data type ( complex types ? )  not supported yet");
        throw new RuntimeException("Unknown data type  ( complex types ? ) not supported in Kudu schema yet");
      default:
        LOG.error("unknown type/default  ( complex types ? )  not supported yet");
        throw new RuntimeException("Unknown type/default  ( complex types ? ) not supported in Kudu schema yet");
    }
  }
}
 

import org.apache.kudu.client.RowResult; //导入依赖的package包/类
@Override
public Long call() throws Exception
{
  long numRowsScanned = 0;
  KuduScanner aPartitionSpecificScanner = KuduScanToken.deserializeIntoScanner(
      kuduPartitionScanAssignmentMeta.getSerializedKuduScanToken(), kuduClientHandle);
  LOG.info("Scanning the following tablet " + KuduScanToken.stringifySerializedToken(kuduPartitionScanAssignmentMeta
      .getSerializedKuduScanToken(), kuduClientHandle));
  KuduRecordWithMeta<T> beginScanRecord = new KuduRecordWithMeta<>();
  beginScanRecord.setBeginScanMarker(true);
  beginScanRecord.setTabletMetadata(kuduPartitionScanAssignmentMeta);
  bufferForTransmittingRecords.add(beginScanRecord); // Add a record entry that denotes the end of this scan.
  while ( aPartitionSpecificScanner.hasMoreRows()) {
    LOG.debug("Number of columns being returned for this read " +
        aPartitionSpecificScanner.getProjectionSchema().getColumnCount());
    RowResultIterator resultIterator = aPartitionSpecificScanner.nextRows();
    if (resultIterator == null) {
      break;
    } else {
      while (resultIterator.hasNext()) {
        KuduRecordWithMeta<T> recordWithMeta = new KuduRecordWithMeta<>();
        RowResult aRow = resultIterator.next();
        recordWithMeta.setPositionInScan(numRowsScanned);
        T payload = clazzForResultObject.newInstance();
        recordWithMeta.setThePayload(payload);
        recordWithMeta.setEndOfScanMarker(false);
        recordWithMeta.setTabletMetadata(kuduPartitionScanAssignmentMeta);
        setValuesInPOJO(aRow,payload);
        bufferForTransmittingRecords.add(recordWithMeta);
        numRowsScanned += 1;
      }
    }
  }
  aPartitionSpecificScanner.close();
  KuduRecordWithMeta<T> endScanRecord = new KuduRecordWithMeta<>();
  endScanRecord.setEndOfScanMarker(true);
  endScanRecord.setTabletMetadata(kuduPartitionScanAssignmentMeta);
  bufferForTransmittingRecords.add(endScanRecord); // Add a record entry that denotes the end of this scan.
  LOG.info(" Scanned a total of " + numRowsScanned + " for this scanner thread @tablet " +
      KuduScanToken.stringifySerializedToken(kuduPartitionScanAssignmentMeta.getSerializedKuduScanToken(),
      kuduClientHandle));
  return numRowsScanned;
}