import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
@Override
public void updateSchema(VectorAccessible batch) throws IOException {
  BatchSchema schema = batch.getSchema();
  int i = 0;

  try {
    if (!checkForTable(name)) {
      List<ColumnSchema> columns = new ArrayList<>();
      for (MaterializedField f : schema) {
        columns.add(new ColumnSchema.ColumnSchemaBuilder(f.getName(), getType(f.getType()))
            .nullable(f.getType().getMode() == DataMode.OPTIONAL)
            .key(i == 0).build());
        i++;
      }
      Schema kuduSchema = new Schema(columns);
      table = client.createTable(name, kuduSchema, new CreateTableOptions());
    }
  } catch (Exception e) {
    throw new IOException(e);
  }
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
/**
 * Builds a set of scan tokens. The list of scan tokens are generated as if the entire table is being scanned
 * i.e. a SELECT * FROM TABLE equivalent expression. This list is used to assign the partition pie assignments
 * for all of the planned partition of operators. Each operator gets a part of the PIE as if all columns were
 * selected. Subsequently when a query is to be processed, the query is used to generate the scan tokens applicable
 * for that query. Given that partition pie represents the entire data set, the scan assignments for the current
 * query will be a subset.
 * @return The list of scan tokens as if the entire table is getting scanned.
 * @throws Exception in cases when the connection to kudu cluster cannot be closed.
 */
public List<KuduScanToken> getKuduScanTokensForSelectAllColumns() throws Exception
{
  // We are not using the current query for deciding the partition strategy but a SELECT * as
  // we do not want to want to optimize on just the current query. This prevents rapid throttling of operator
  // instances when the scan patterns are erratic. On the other hand, this might result on under utilized
  // operator resources in the DAG but will be consistent at a minimum.
  ApexKuduConnection apexKuduConnection = prototypeKuduInputOperator.getApexKuduConnectionInfo().build();
  KuduClient clientHandle = apexKuduConnection.getKuduClient();
  KuduTable table = apexKuduConnection.getKuduTable();
  KuduScanToken.KuduScanTokenBuilder builder = clientHandle.newScanTokenBuilder(table);
  List<String> allColumns = new ArrayList<>();
  List<ColumnSchema> columnList = apexKuduConnection.getKuduTable().getSchema().getColumns();
  for ( ColumnSchema column : columnList) {
    allColumns.add(column.getName());
  }
  builder.setProjectedColumnNames(allColumns);
  LOG.debug("Building the partition pie assignments for the input operator");
  List<KuduScanToken> allPossibleTokens = builder.build();
  apexKuduConnection.close();
  return allPossibleTokens;
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
/**
 * Scans the metadata for the kudu table that this operator is scanning for and
 * returns back the mapping for the kudu column name to the ColumnSchema metadata definition.
 * Note that the Kudu columns names are case sensitive.
 * @return A Map with Kudu column names as keys and value as the Column Definition.
 * @throws Exception
 */
private Map<String,ColumnSchema> buildColumnSchemaForTable() throws Exception
{
  if (kuduColNameToSchemaMapping == null) {
    ApexKuduConnection connectionForMetaDataScan = apexKuduConnectionInfo.build();
    KuduTable table = connectionForMetaDataScan.getKuduTable();
    List<ColumnSchema> tableColumns =  table.getSchema().getColumns();
    connectionForMetaDataScan.close();
    Map<String,ColumnSchema> columnSchemaMap = new HashMap<>();
    for (ColumnSchema aColumn: tableColumns) {
      columnSchemaMap.put(aColumn.getName(),aColumn);
    }
    kuduColNameToSchemaMapping = columnSchemaMap;
  }
  return kuduColNameToSchemaMapping;
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
@KuduClusterTestContext(kuduClusterBasedTest = true)
@Test
public void testSettersForPojo() throws Exception
{
  initOperatorState();
  AbstractKuduPartitionScanner<UnitTestTablePojo,InputOperatorControlTuple> currentScanner =
      unitTestStepwiseScanInputOperator.getScanner();
  SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator(
      "select introwkey as intColumn from unittests",
      new ArrayList<ColumnSchema>(columnDefs.values()));
  List<KuduPartitionScanAssignmentMeta> scansForThisQuery = currentScanner.preparePlanForScanners(translator);
  KuduPartitionScannerCallable<UnitTestTablePojo,InputOperatorControlTuple> threadToScan = new
      KuduPartitionScannerCallable<>(unitTestStepwiseScanInputOperator,scansForThisQuery.get(0),
      currentScanner.getConnectionPoolForThreads().get(0),
      unitTestStepwiseScanInputOperator.extractSettersForResultObject(translator),translator);
  long countOfScan = threadToScan.call();
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
public void convert(Record record, PartialRow row, int operation) throws OnRecordErrorException {
  for (Map.Entry<String, String> entry : fieldsToColumns.entrySet()) {
    String fieldName =  entry.getKey();
    if (fieldConverter != null) {
      fieldName = fieldConverter.getFieldPath(fieldName, operation);
    }
    String column = entry.getValue();  // column name in Kudu table
    // For delete, we only need to fill primary key column name & value in PartialRow
    if (operation == KuduOperationType.DELETE.code) {
      for(ColumnSchema col : schema.getPrimaryKeyColumns()) {
        if (col.getName().equals(column))
          recordToRow(record, row, fieldName, column, operation);
      }
    } else {
      // For other operations, we need to know the operation
      // to correctly fill the record.
      recordToRow(record, row, fieldName, column, operation);
    }
  }
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
public void init() throws KuduException {
    client = new KuduClient.KuduClientBuilder(KUDU_MASTER).build();
    List<ColumnSchema> columns = new ArrayList(2);
    columns.add(new ColumnSchema.ColumnSchemaBuilder("id", Type.INT32).key(true).build());
    columns.add(new ColumnSchema.ColumnSchemaBuilder("value", Type.STRING).build());
    columns.add(new ColumnSchema.ColumnSchemaBuilder("hash", Type.INT32).build());
    Schema schema = new Schema(columns);
    List<String> rangeColumns = new ArrayList<String>(1);
    rangeColumns.add("id");
    client.createTable(KUDU_TABLE_NAME, schema, new CreateTableOptions().setRangePartitionColumns(rangeColumns));
    table = client.openTable(KUDU_TABLE_NAME);
    session = client.newSession();
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的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.ColumnSchema; //导入依赖的package包/类
private StructType schemaFor(KuduTable table) {
  List<String> fieldNames = Lists.newArrayList();
  List<String> fieldTypes = Lists.newArrayList();

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

    switch (columnSchema.getType()) {
      case DOUBLE:
        fieldType = "double";
        break;
      case FLOAT:
        fieldType = "float";
        break;
      case INT32:
        fieldType = "int";
        break;
      case INT64:
        fieldType = "long";
        break;
      case STRING:
        fieldType = "string";
        break;
      case BOOL:
        fieldType = "boolean";
        break;
      default:
        throw new RuntimeException("Unsupported Kudu column type: " + columnSchema.getType());
    }

    fieldNames.add(fieldName);
    fieldTypes.add(fieldType);
  }

  StructType tableSchema = RowUtils.structTypeFor(fieldNames, fieldTypes);

  return tableSchema;
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
private KuduScanner scannerForFilters(Iterable<Row> filters, KuduTable table) {
  List<Row> filtersList = Lists.newArrayList(filters);

  if (filtersList.size() == 0) {
    throw new RuntimeException("Kudu existing filter was not provided.");
  }
  
  if (filtersList.get(0).schema() == null) {
    throw new RuntimeException("Kudu existing filter did not contain a schema.");
  }
  
  if (hasAccumulators()) {
    accumulators.getLongAccumulators().get(ACCUMULATOR_NUMBER_OF_SCANNERS).add(1);
    accumulators.getLongAccumulators().get(ACCUMULATOR_NUMBER_OF_FILTERS_SCANNED).add(filtersList.size());
  }
  
  KuduScannerBuilder builder = client.newScannerBuilder(table);

  for (String fieldName : filtersList.get(0).schema().fieldNames()) {
    ColumnSchema columnSchema = table.getSchema().getColumn(fieldName);

    List<Object> columnValues = Lists.newArrayList();
    for (Row filter : filtersList) {
      Object columnValue = RowUtils.get(filter, fieldName);
      columnValues.add(columnValue);
    }

    KuduPredicate predicate = KuduPredicate.newInListPredicate(columnSchema, columnValues);

    builder = builder.addPredicate(predicate);
  }

  KuduScanner scanner = builder.build();

  return scanner;
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
@Override
public RelDataType getRowType(RelDataTypeFactory typeFactory) {

  List<String> names = Lists.newArrayList();
  List<RelDataType> types = Lists.newArrayList();
  for (ColumnSchema column : schema.getColumns()) {
    names.add(column.getName());
    RelDataType type = getSqlTypeFromKuduType(typeFactory, column.getType());
    type = typeFactory.createTypeWithNullability(type, column.isNullable());
    types.add(type);
  }

  return typeFactory.createStructType(types, names);
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
private void buildGettersForPojoPayload()
{
  Class payloadClass = getTuplePayloadClass();
  checkNotNull(payloadClass,"Payload class cannot be null");
  Field[] classFields = payloadClass.getDeclaredFields();
  Schema schemaInfo = kuduTable.getSchema();
  List<ColumnSchema> allColumns = schemaInfo.getColumns();
  Set<String> allKuduTableColumnNames = new HashSet<>();
  Map<String,ColumnSchema> normalizedColumns = new HashMap();
  for ( ColumnSchema aColumnDef : allColumns) {
    allColumnDefs.put(aColumnDef.getName(), aColumnDef);
    normalizedColumns.put(aColumnDef.getName().toLowerCase(), aColumnDef);
    allKuduTableColumnNames.add(aColumnDef.getName().toLowerCase());
  }
  List<ColumnSchema> primaryKeyColumns = schemaInfo.getPrimaryKeyColumns();
  for (ColumnSchema primaryKeyInfo : primaryKeyColumns) {
    primaryKeyColumnNames.add(primaryKeyInfo.getName());
  }
  Map<String,String> columnNameOverrides = getOverridingColumnNameMap();
  if (columnNameOverrides == null) {
    columnNameOverrides = new HashMap(); // to avoid null checks further down the line
  }
  for ( Field aFieldDef : classFields) {
    String currentFieldName = aFieldDef.getName().toLowerCase();
    if (allKuduTableColumnNames.contains(currentFieldName)) {
      extractGetterForColumn(normalizedColumns.get(currentFieldName),aFieldDef);
    } else {
      if (columnNameOverrides.containsKey(aFieldDef.getName())) {
        extractGetterForColumn(normalizedColumns.get(columnNameOverrides.get(aFieldDef.getName()).toLowerCase()),
            aFieldDef);
      } else if (columnNameOverrides.containsKey(aFieldDef.getName().toLowerCase())) {
        extractGetterForColumn(normalizedColumns.get(columnNameOverrides.get(aFieldDef.getName().toLowerCase())
            .toLowerCase()),aFieldDef);
      }
    }
  }
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
public void setColumnSchemaList(List<ColumnSchema> listOfColumnsForCurrentTable)
{
  Preconditions.checkNotNull(listOfColumnsForCurrentTable,"Column schemas " +
      "cannot be null for kudu table");
  for (ColumnSchema aColumnDef : listOfColumnsForCurrentTable) {
    columnSchemaLookups.put(aColumnDef.getName(),aColumnDef);
    aliases.put(aColumnDef.getName(),aColumnDef.getName()); // By default each column is its own alias in POJO.
  }
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
public SQLToKuduPredicatesTranslator(String sqlExpresssionForParsing, List<ColumnSchema> tableColumns)
  throws Exception
{
  Preconditions.checkNotNull(tableColumns,"Kudu table cannot have null or empty columns");
  Preconditions.checkNotNull(sqlExpresssionForParsing,"Kudu SQL expression cannot be null");
  sqlExpresssion = sqlExpresssionForParsing;
  allColumnsForThisTable = tableColumns;
  parseKuduExpression();
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
@KuduClusterTestContext(kuduClusterBasedTest = false)
@Test
public void testForCompletenessOfSQLExpression() throws Exception
{
  SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator(
      "select from unittests",new ArrayList<ColumnSchema>(columnDefs.values()));
  translator.parseKuduExpression();
  KuduSQLExpressionErrorListener errorListener = translator.getErrorListener();
  assertEquals(true,errorListener.isSyntaxError());
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
@KuduClusterTestContext(kuduClusterBasedTest = false)
@Test
public void testForErrorsInColumnAliasesInSQLExpression() throws Exception
{
  SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator(
      "select intkey as intColumn from unittests",
      new ArrayList<ColumnSchema>(columnDefs.values()));
  KuduSQLExpressionErrorListener errorListener = translator.getErrorListener();
  assertEquals(false,errorListener.isSyntaxError());

  translator = new SQLToKuduPredicatesTranslator(
    "select intkey as intColumn, 'from' as fgh from unittests",
    new ArrayList<ColumnSchema>(columnDefs.values()));
  errorListener = translator.getErrorListener();
  assertEquals(false,errorListener.isSyntaxError());

  translator = new SQLToKuduPredicatesTranslator(
    "select intkey, 'from' from unittests",
    new ArrayList<ColumnSchema>(columnDefs.values()));
  errorListener = translator.getErrorListener();
  assertEquals(false,errorListener.isSyntaxError());

  translator = new SQLToKuduPredicatesTranslator(
    "select intkey, 'from' as fgh from unittests",
    new ArrayList<ColumnSchema>(columnDefs.values()));
  errorListener = translator.getErrorListener();
  assertEquals(false,errorListener.isSyntaxError());
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
@KuduClusterTestContext(kuduClusterBasedTest = false)
@Test
public void testForErrorsInOptionsInSQLExpression() throws Exception
{
  SQLToKuduPredicatesTranslator translator = null;
  KuduSQLExpressionErrorListener errorListener = null;

  translator = new SQLToKuduPredicatesTranslator(
    "select intkey as intColumn from unittests using Options",
    new ArrayList<ColumnSchema>(columnDefs.values()));
  errorListener = translator.getErrorListener();
  assertEquals(true,errorListener.isSyntaxError());

  translator = new SQLToKuduPredicatesTranslator(
    "select intkey as intColumn, 'from' as fgh from " +
      " unittests using options READ_SNAPSHOT_TIME = aASDAD",
    new ArrayList<ColumnSchema>(columnDefs.values()));
  errorListener = translator.getErrorListener();
  assertEquals(true,errorListener.isSyntaxError());

  translator = new SQLToKuduPredicatesTranslator(
    "select intkey, 'from' from unittests using options READ_SNAPSHOT_TIME = 2342345",
    new ArrayList<ColumnSchema>(columnDefs.values()));
  errorListener = translator.getErrorListener();
  assertEquals(false,errorListener.isSyntaxError());

  translator = new SQLToKuduPredicatesTranslator(
    "select intkey, xcv as fgh from unittests using options CONTROLTUPLE_MESSAGE = \"done\"",
    new ArrayList<ColumnSchema>(columnDefs.values()));
  errorListener = translator.getErrorListener();
  assertEquals(false,errorListener.isSyntaxError());
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
@KuduClusterTestContext(kuduClusterBasedTest = false)
@Test
public void testForSelectStarInSQLExpression() throws Exception
{
  SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator(
      " select * from unittests",
      new ArrayList<ColumnSchema>(columnDefs.values()));
  assertEquals(true,translator.getKuduSQLParseTreeListener().isSelectStarExpressionEnabled());

  translator = new SQLToKuduPredicatesTranslator(
    "select intkey as intColumn from unittests ",
    new ArrayList<ColumnSchema>(columnDefs.values()));
  assertEquals(false,translator.getKuduSQLParseTreeListener().isSelectStarExpressionEnabled());

}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
@KuduClusterTestContext(kuduClusterBasedTest = false)
@Test
public void testForColumnNameExtractionInSQLExpression() throws Exception
{
  SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator(
      " select introwkey as intColumn, '      from' as flColumn, stringCol from unittests",
      new ArrayList<ColumnSchema>(columnDefs.values()));
  assertEquals(1, translator.getKuduSQLParseTreeListener().getListOfColumnsUsed().size());
  assertEquals(9, translator.getKuduSQLParseTreeListener().getAliases().size());
  assertEquals("intColumn", translator.getKuduSQLParseTreeListener().getAliases().get("introwkey"));
}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
@Test
@KuduClusterTestContext(kuduClusterBasedTest = false)
public void testForReadSnapshotTimeExpression() throws Exception
{
  SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator(
      " select introwkey as intColumn using options read_snapshot_time = 12345",
      new ArrayList<ColumnSchema>(columnDefs.values()));
  assertEquals(12345L, translator.getKuduSQLParseTreeListener().getReadSnapshotTime().longValue());
  SQLToKuduPredicatesTranslator translatorForNoReadSnapshotTime = new SQLToKuduPredicatesTranslator(
      " select introwkey as intColumn",
      new ArrayList<ColumnSchema>(columnDefs.values()));
  assertEquals(null, translatorForNoReadSnapshotTime.getKuduSQLParseTreeListener().getReadSnapshotTime());

}
 

import org.apache.kudu.ColumnSchema; //导入依赖的package包/类
@KuduClusterTestContext(kuduClusterBasedTest = true)
@Test
public void testRowScansForAllDataAcrossAllPartitions() throws Exception
{
  partitonScanStrategy = KuduPartitionScanStrategy.MANY_TABLETS_PER_OPERATOR;
  numberOfKuduInputOperatorPartitions = 1;
  initOperatorState();
  AbstractKuduPartitionScanner<UnitTestTablePojo,InputOperatorControlTuple> currentScanner =
      unitTestStepwiseScanInputOperator.getScanner();
  // truncate and add some data to the unit test table
  truncateTable();
  addTestDataRows(10); // This is per partition and there are 12 partitions
  assertEquals((KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE * 10 ),countNumRowsInTable());
  SQLToKuduPredicatesTranslator translator = new SQLToKuduPredicatesTranslator(
      "select * from unittests",
      new ArrayList<ColumnSchema>(columnDefs.values()));
  List<KuduPartitionScanAssignmentMeta> scansForThisQuery = currentScanner.preparePlanForScanners(translator);
  // Now scan for exact match of counts
  long totalRowsRead = 0;
  unitTestStepwiseScanInputOperator.getBuffer().clear();
  for (KuduPartitionScanAssignmentMeta aSegmentToScan :  scansForThisQuery) {
    KuduPartitionScannerCallable<UnitTestTablePojo,InputOperatorControlTuple> threadToScan = new
        KuduPartitionScannerCallable<>(unitTestStepwiseScanInputOperator,aSegmentToScan,
        currentScanner.getConnectionPoolForThreads().get(0),
        unitTestStepwiseScanInputOperator.extractSettersForResultObject(translator),translator);
    totalRowsRead += threadToScan.call();
  }
  // 144 = 120 records + 12 * 2 markers
  int expectedCount = ( 10 * KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE) +
      ( 2 * KuduClientTestCommons.TOTAL_KUDU_TABLETS_FOR_UNITTEST_TABLE);
  assertEquals(expectedCount,unitTestStepwiseScanInputOperator.getBuffer().size());
  // revert all configs to default
  partitonScanStrategy = KuduPartitionScanStrategy.ONE_TABLET_PER_OPERATOR;
  numberOfKuduInputOperatorPartitions = 5;
}