import org.apache.kudu.Schema; //导入依赖的package包/类
private static Deferred<KuduTable> openOrCreateTable(final AsyncKuduClient client,
                                                     final String table,
                                                     final Schema schema,
                                                     final CreateTableOptions options) throws Exception {
  class CreateTableErrback implements Callback<Deferred<KuduTable>, Exception> {
    @Override
    public Deferred<KuduTable> call(Exception e) throws Exception {
      // TODO(danburkert): we should only do this if the error is "not found"
      LOG.debug("Creating table {}", table);
      return client.createTable(table, schema, options);
    }
    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this).add("table", table).toString();
    }
  }

  return client.openTable(table).addErrback(new CreateTableErrback());
}
 

import org.apache.kudu.Schema; //导入依赖的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.Schema; //导入依赖的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.Schema; //导入依赖的package包/类
@Override
public Table getTable(String name) {
  KuduScanSpec scanSpec = new KuduScanSpec(name);
  try {
    KuduTable table = plugin.getClient().openTable(name);
    Schema schema = table.getSchema();
    return new DrillKuduTable(schemaName, plugin, schema, scanSpec);
  } catch (Exception e) {
    logger.warn("Failure while retrieving kudu table {}", name, e);
    return null;
  }

}
 

import org.apache.kudu.Schema; //导入依赖的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.Schema; //导入依赖的package包/类
public static void createTestTable(String tableName, KuduClient client) throws Exception
{
  List<String> rangeKeys = new ArrayList<>();
  rangeKeys.add("introwkey");
  List<String> hashPartitions = new ArrayList<>();
  hashPartitions.add("stringrowkey");
  hashPartitions.add("timestamprowkey");
  CreateTableOptions thisTableOptions = new CreateTableOptions()
      .setNumReplicas(1)
      .addHashPartitions(hashPartitions,HASH_BUCKETS_SIZE_FOR_ALL_HASH_COL)
      .setRangePartitionColumns(rangeKeys);
  int stepsize = Integer.MAX_VALUE / SPLIT_COUNT_FOR_INT_ROW_KEY;
  int splitBoundary = stepsize;
  Schema schema = buildSchemaForUnitTestsTable();
  for ( int i = 0; i < SPLIT_COUNT_FOR_INT_ROW_KEY; i++) {
    PartialRow splitRowBoundary = schema.newPartialRow();
    splitRowBoundary.addInt("introwkey",splitBoundary);
    thisTableOptions = thisTableOptions.addSplitRow(splitRowBoundary);
    splitBoundary += stepsize;
  }
  try {
    client.createTable(tableName, schema,thisTableOptions);
  } catch (KuduException e) {
    LOG.error("Error while creating table for unit tests " + e.getMessage(), e);
    throw e;
  }

}
 

import org.apache.kudu.Schema; //导入依赖的package包/类
public KuduRecordConverter(Map<String, Field.Type> columnsToFieldTypes, Map<String, String> fieldsToColumns,
                           Schema schema, FieldPathConverter converter) {
  this.columnsToFieldTypes = ImmutableMap.copyOf(columnsToFieldTypes);
  this.fieldsToColumns = ImmutableMap.copyOf(fieldsToColumns);
  this.schema = schema;
  this.fieldConverter = converter;
}
 

import org.apache.kudu.Schema; //导入依赖的package包/类
@Before
public void setup() {
  // Sample table and schema
  List<ColumnSchema> columns = new ArrayList(2);
  columns.add(new ColumnSchema.ColumnSchemaBuilder("key", Type.INT32).key(true).build());
  columns.add(new ColumnSchema.ColumnSchemaBuilder("value", Type.STRING).build());
  columns.add(new ColumnSchema.ColumnSchemaBuilder("name", Type.STRING).build());
  final Schema schema = new Schema(columns);

  // Mock KuduTable class
  KuduTable table = PowerMockito.spy(PowerMockito.mock(KuduTable.class));
  PowerMockito.suppress(PowerMockito.method(AsyncKuduClient.class, "getTablesList"));
  PowerMockito.when(table.getSchema()).thenReturn(schema);

  // Mock KuduSession class
  PowerMockito.suppress(PowerMockito.method(
      AsyncKuduSession.class,
      "apply",
      Operation.class
  ));
  PowerMockito.suppress(PowerMockito.method(
      AsyncKuduSession.class,
      "flush"
  ));
  PowerMockito.suppress(PowerMockito.method(
      KuduLookupProcessor.class,
      "destroy"
  ));
}
 

import org.apache.kudu.Schema; //导入依赖的package包/类
private void initCols(Schema schema) throws SchemaChangeException {
  ImmutableList.Builder<ProjectedColumnInfo> pciBuilder = ImmutableList.builder();

  for (int i = 0; i < schema.getColumnCount(); i++) {
    ColumnSchema col = schema.getColumnByIndex(i);

    final String name = col.getName();
    final Type kuduType = col.getType();
    MinorType minorType = TYPES.get(kuduType);
    if (minorType == null) {
      logger.warn("Ignoring column that is unsupported.", UserException
          .unsupportedError()
          .message(
              "A column you queried has a data type that is not currently supported by the Kudu storage plugin. "
                  + "The column's name was %s and its Kudu data type was %s. ",
              name, kuduType.toString())
          .addContext("column Name", name)
          .addContext("plugin", "kudu")
          .build(logger));

      continue;
    }
    MajorType majorType;
    if (col.isNullable()) {
      majorType = Types.optional(minorType);
    } else {
      majorType = Types.required(minorType);
    }
    MaterializedField field = MaterializedField.create(name, majorType);
    final Class<? extends ValueVector> clazz = TypeHelper.getValueVectorClass(
        minorType, majorType.getMode());
    ValueVector vector = output.addField(field, clazz);
    vector.allocateNew();

    ProjectedColumnInfo pci = new ProjectedColumnInfo();
    pci.vv = vector;
    pci.kuduColumn = col;
    pci.index = i;
    pciBuilder.add(pci);
  }

  projectedCols = pciBuilder.build();
}
 

import org.apache.kudu.Schema; //导入依赖的package包/类
public DrillKuduTable(String storageEngineName, KuduStoragePlugin plugin, Schema schema, KuduScanSpec scanSpec) {
  super(plugin, storageEngineName, scanSpec);
  this.schema = schema;
}
 

import org.apache.kudu.Schema; //导入依赖的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.Schema; //导入依赖的package包/类
@Before
public void setup() throws Exception {
  context = ContextInfoCreator.createSourceContext("i", false, OnRecordError.TO_ERROR, Collections.EMPTY_LIST);
  record = context.createRecord("123");
  record.set("/", Field.create(new LinkedHashMap<String, Field>()));
  record.set("/byte", Field.create((byte)1));
  record.set("/short", Field.create((short)123));
  record.set("/int", Field.create(123));
  record.set("/long", Field.create(123L));
  record.set("/float", Field.create(123.0f));
  record.set("/double", Field.create(123.0d));
  record.set("/bytes", Field.create("ABC".getBytes(StandardCharsets.UTF_8)));
  record.set("/str", Field.create("ABC"));
  record.set("/bool", Field.create(true));
  DateTime dt = new DateTime(2017, 8, 24, 9, 15, 30, DateTimeZone.UTC); // 2017/8/24 9:15:30
  record.set("/unixtime", Field.create(dt.getMillis() * 1000L));
  Map<String, Field.Type> columnsToFieldTypes = ImmutableMap.<String, Field.Type>builder()
    .put("byte1", Field.Type.BYTE)
    .put("short1", Field.Type.SHORT)
    .put("int1", Field.Type.INTEGER)
    .put("long1", Field.Type.LONG)
    .put("float1", Field.Type.FLOAT)
    .put("double1", Field.Type.DOUBLE)
    .put("bytes", Field.Type.BYTE_ARRAY)
    .put("str", Field.Type.STRING)
    .put("bool1", Field.Type.BOOLEAN)
    .put("unixtime_micro", Field.Type.LONG)
    .build();
  Map<String, String> fieldsToColumns = ImmutableMap.<String, String>builder()
    .put("/byte", "byte1")
    .put("/short", "short1")
    .put("/int", "int1")
    .put("/long", "long1")
    .put("/float", "float1")
    .put("/double", "double1")
    .put("/bytes", "bytes")
    .put("/str", "str")
    .put("/bool", "bool1")
    .put("/unixtime", "unixtime_micro")
    .build();
  Schema schema = new Schema(Arrays.asList(
    new ColumnSchema.ColumnSchemaBuilder("str", Type.STRING).key(true).build(),
    new ColumnSchema.ColumnSchemaBuilder("byte1", Type.INT8).build(),
    new ColumnSchema.ColumnSchemaBuilder("short1", Type.INT16).nullable(true).build(),
    new ColumnSchema.ColumnSchemaBuilder("int1", Type.INT32).build(),
    new ColumnSchema.ColumnSchemaBuilder("long1", Type.INT64).build(),
    new ColumnSchema.ColumnSchemaBuilder("float1", Type.FLOAT).build(),
    new ColumnSchema.ColumnSchemaBuilder("double1", Type.DOUBLE).build(),
    new ColumnSchema.ColumnSchemaBuilder("bytes", Type.BINARY).build(),
    new ColumnSchema.ColumnSchemaBuilder("bool1", Type.BOOL).build(),
    new ColumnSchema.ColumnSchemaBuilder("unixtime_micro", Type.UNIXTIME_MICROS).build()
    ));
  partialRow = new PartialRow(schema);
  kuduRecordConverter = new KuduRecordConverter(columnsToFieldTypes, fieldsToColumns, schema, null);
}