import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
public ExternalSortBatch(ExternalSort popConfig, FragmentContext context, RecordBatch incoming) throws OutOfMemoryException {
  super(popConfig, context, true);
  this.incoming = incoming;
  DrillConfig config = context.getConfig();
  Configuration conf = new Configuration();
  conf.set("fs.default.name", config.getString(ExecConstants.EXTERNAL_SORT_SPILL_FILESYSTEM));
  try {
    this.fs = FileSystem.get(conf);
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
  SPILL_BATCH_GROUP_SIZE = config.getInt(ExecConstants.EXTERNAL_SORT_SPILL_GROUP_SIZE);
  SPILL_THRESHOLD = config.getInt(ExecConstants.EXTERNAL_SORT_SPILL_THRESHOLD);
  dirs = Iterators.cycle(config.getStringList(ExecConstants.EXTERNAL_SORT_SPILL_DIRS));
  copierAllocator = oContext.getAllocator().getChildAllocator(
      context, PriorityQueueCopier.INITIAL_ALLOCATION, PriorityQueueCopier.MAX_ALLOCATION, true);
  FragmentHandle handle = context.getHandle();
  fileName = String.format("%s/major_fragment_%s/minor_fragment_%s/operator_%s", QueryIdHelper.getQueryId(handle.getQueryId()),
      handle.getMajorFragmentId(), handle.getMinorFragmentId(), popConfig.getOperatorId());
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
public InternalBatch(RecordBatch incoming, VectorWrapper[] ignoreWrappers){
  switch(incoming.getSchema().getSelectionVectorMode()){
  case FOUR_BYTE:
    this.sv4 = incoming.getSelectionVector4().createNewWrapperCurrent();
    this.sv2 = null;
    break;
  case TWO_BYTE:
    this.sv4 = null;
    this.sv2 = incoming.getSelectionVector2().clone();
    break;
  default:
    this.sv4 = null;
    this.sv2 = null;
  }
  this.schema = incoming.getSchema();
  this.container = VectorContainer.getTransferClone(incoming, ignoreWrappers);
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
public static Copier getGenerated4Copier(RecordBatch batch, FragmentContext context, BufferAllocator allocator, VectorContainer container, RecordBatch outgoing, SchemaChangeCallBack callBack) throws SchemaChangeException{

    for(VectorWrapper<?> vv : batch){
      ValueVector v = vv.getValueVectors()[0];
      v.makeTransferPair(container.addOrGet(v.getField(), callBack));
    }

    try {
      final CodeGenerator<Copier> cg = CodeGenerator.get(Copier.TEMPLATE_DEFINITION4, context.getFunctionRegistry());
      CopyUtil.generateCopies(cg.getRoot(), batch, true);
      Copier copier = context.getImplementationClass(cg);
      copier.setupRemover(context, batch, outgoing);

      return copier;
    } catch (ClassTransformationException | IOException e) {
      throw new SchemaChangeException("Failure while attempting to load generated class", e);
    }
  }
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
/** Create RootExec and its children (RecordBatches) for given FragmentRoot */
private RootExec getRootExec(final FragmentRoot root, final FragmentContext context) throws ExecutionSetupException {
  final List<RecordBatch> childRecordBatches = getChildren(root, context);

  if (context.isImpersonationEnabled()) {
    final UserGroupInformation proxyUgi = ImpersonationUtil.createProxyUgi(root.getUserName(), context.getQueryUserName());
    try {
      return proxyUgi.doAs(new PrivilegedExceptionAction<RootExec>() {
        @Override
        public RootExec run() throws Exception {
          return ((RootCreator<PhysicalOperator>) getOpCreator(root, context)).getRoot(context, root, childRecordBatches);
        }
      });
    } catch (InterruptedException | IOException e) {
      final String errMsg = String.format("Failed to create RootExec for operator with id '%d'", root.getOperatorId());
      logger.error(errMsg, e);
      throw new ExecutionSetupException(errMsg, e);
    }
  } else {
    return ((RootCreator<PhysicalOperator>) getOpCreator(root, context)).getRoot(context, root, childRecordBatches);
  }
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
@Override
public ScanBatch getBatch(FragmentContext context, HBaseSubScan subScan, List<RecordBatch> children)
    throws ExecutionSetupException {
  Preconditions.checkArgument(children.isEmpty());
  List<RecordReader> readers = Lists.newArrayList();
  List<SchemaPath> columns = null;
  for(HBaseSubScan.HBaseSubScanSpec scanSpec : subScan.getRegionScanSpecList()){
    try {
      if ((columns = subScan.getColumns())==null) {
        columns = GroupScan.ALL_COLUMNS;
      }
      readers.add(new HBaseRecordReader(subScan.getStorageConfig().getHBaseConf(), scanSpec, columns, context));
    } catch (Exception e1) {
      throw new ExecutionSetupException(e1);
    }
  }
  return new ScanBatch(subScan, context, readers.iterator());
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
public RecordBatchData(VectorAccessible batch) {
  List<ValueVector> vectors = Lists.newArrayList();
  recordCount = batch.getRecordCount();

  if (batch instanceof RecordBatch && batch.getSchema().getSelectionVectorMode() == SelectionVectorMode.TWO_BYTE) {
    this.sv2 = ((RecordBatch)batch).getSelectionVector2().clone();
  } else {
    this.sv2 = null;
  }

  for (VectorWrapper<?> v : batch) {
    if (v.isHyper()) {
      throw new UnsupportedOperationException("Record batch data can't be created based on a hyper batch.");
    }
    TransferPair tp = v.getValueVector().getTransferPair();
    tp.transfer();
    vectors.add(tp.getTo());
  }

  container.addCollection(vectors);
  container.setRecordCount(recordCount);
  container.buildSchema(batch.getSchema().getSelectionVectorMode());
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
@Override
public final void setup(FragmentContext context, VectorAccessible incoming, RecordBatch outgoing, List<TransferPair> transfers,
                        VectorContainer partitionVectors, int partitions, SchemaPath outputField)  throws SchemaChangeException{

  this.svMode = incoming.getSchema().getSelectionVectorMode();
  this.outBatch = outgoing;
  this.outputField = outputField;
  partitionValues = (IntVector) outBatch.getValueAccessorById(IntVector.class, outBatch.getValueVectorId(outputField).getFieldIds()).getValueVector();
  switch(svMode){
  case FOUR_BYTE:
  case TWO_BYTE:
    throw new UnsupportedOperationException("Selection vector not supported");
  }
  this.transfers = ImmutableList.copyOf(transfers);
  this.partitions = partitions;
  doSetup(context, incoming, outgoing, partitionVectors);
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
public TraceRecordBatch(Trace pop, RecordBatch incoming, FragmentContext context) throws ExecutionSetupException {
  super(pop, context, incoming);
  this.traceTag = pop.traceTag;
  logLocation = context.getConfig().getString(ExecConstants.TRACE_DUMP_DIRECTORY);

  String fileName = getFileName();

  /* Create the log file we will dump to and initialize the file descriptors */
  try {
    Configuration conf = new Configuration();
    conf.set(FileSystem.FS_DEFAULT_NAME_KEY, context.getConfig().getString(ExecConstants.TRACE_DUMP_FILESYSTEM));
    FileSystem fs = FileSystem.get(conf);

    /* create the file */
    fos = fs.create(new Path(fileName));
  } catch (IOException e) {
      throw new ExecutionSetupException("Unable to create file: " + fileName + " check permissions or if directory exists", e);
  }
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
protected MergeJoinBatch(MergeJoinPOP popConfig, FragmentContext context, RecordBatch left, RecordBatch right) throws OutOfMemoryException {
  super(popConfig, context, true);

  if (popConfig.getConditions().size() == 0) {
    throw new UnsupportedOperationException("Merge Join currently does not support cartesian join.  This join operator was configured with 0 conditions");
  }
  this.left = left;
  this.right = right;
  this.joinType = popConfig.getJoinType();
  this.status = new JoinStatus(left, right, this);
  this.batchBuilder = new MergeJoinBatchBuilder(oContext.getAllocator(), status);
  this.conditions = popConfig.getConditions();

  JoinComparator comparator = JoinComparator.NONE;
  for (JoinCondition condition : conditions) {
    comparator = JoinUtils.checkAndSetComparison(condition, comparator);
  }
  assert comparator != JoinComparator.NONE;
  areNullsEqual = (comparator == JoinComparator.IS_NOT_DISTINCT_FROM);
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
@Override
public RootExec getRoot(FragmentContext context, Screen config, List<RecordBatch> children)
    throws ExecutionSetupException {
  Preconditions.checkNotNull(children);
  Preconditions.checkArgument(children.size() == 1);
  return new ScreenRoot(context, children.iterator().next(), config);
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
@Override
public ScanBatch getBatch(FragmentContext context, HiveSubScan config, List<RecordBatch> children)
    throws ExecutionSetupException {
  List<RecordReader> readers = Lists.newArrayList();
  Table table = config.getTable();
  List<InputSplit> splits = config.getInputSplits();
  List<Partition> partitions = config.getPartitions();
  boolean hasPartitions = (partitions != null && partitions.size() > 0);
  int i = 0;

  // Native hive text record reader doesn't handle all types currently. For now use HiveRecordReader which uses
  // Hive InputFormat and SerDe classes to read the data.
  //if (table.getSd().getInputFormat().equals(TextInputFormat.class.getCanonicalName()) &&
  //        table.getSd().getSerdeInfo().getSerializationLib().equals(LazySimpleSerDe.class.getCanonicalName()) &&
  //        config.getColumns() != null) {
  //  for (InputSplit split : splits) {
  //    readers.add(new HiveTextRecordReader(table,
  //        (hasPartitions ? partitions.get(i++) : null),
  //        split, config.getColumns(), context));
  //  }
  //} else {
    for (InputSplit split : splits) {
      readers.add(new HiveRecordReader(table,
          (hasPartitions ? partitions.get(i++) : null),
          split, config.getColumns(), context, config.getHiveReadEntry().hiveConfigOverride));
    }
  //}

  // If there are no readers created (which is possible when the table is empty), create an empty RecordReader to
  // output the schema
  if (readers.size() == 0) {
    readers.add(new HiveRecordReader(table, null, null, config.getColumns(), context,
        config.getHiveReadEntry().hiveConfigOverride));
  }

  return new ScanBatch(config, context, readers.iterator());
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
public void add(RecordBatch batch, boolean newSchema) throws SchemaChangeException {
  if (!schemas.isEmpty() && newSchema) {
    throw new SchemaChangeException("Currently, the sv4 builder doesn't support embedded types");
  }
  if (newSchema) {
    schemas.add(batch.getSchema());
  }
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
@Override
public final void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, List<TransferPair> transfers)  throws SchemaChangeException{

  this.svMode = incoming.getSchema().getSelectionVectorMode();
  switch (svMode) {
  case FOUR_BYTE:
    this.vector4 = incoming.getSelectionVector4();
    break;
  case TWO_BYTE:
    this.vector2 = incoming.getSelectionVector2();
    break;
  }
  this.transfers = ImmutableList.copyOf(transfers);
  doSetup(context, incoming, outgoing);
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes", "unchecked" })
@Override
public ScanBatch getBatch(final FragmentContext context, final SystemTableScan scan,
                            final List<RecordBatch> children)
  throws ExecutionSetupException {
  final SystemTable table = scan.getTable();
  final Iterator<Object> iterator = table.getIterator(context);
  final RecordReader reader = new PojoRecordReader(table.getPojoClass(), iterator);

  return new ScanBatch(scan, context, Collections.singleton(reader).iterator());
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
@Override
public void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, TransferPair[] transfers)
    throws SchemaChangeException {
  this.transfers = transfers;
  this.outgoingSelectionVector = outgoing.getSelectionVector4();
  this.incomingSelectionVector = incoming.getSelectionVector4();
  doSetup(context, incoming, outgoing);
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
public VectorRecordMaterializer(FragmentContext context, RecordBatch batch) {
    this.queryId = context.getHandle().getQueryId();
    this.batch = batch;
    BatchSchema schema = batch.getSchema();
    assert schema != null : "Schema must be defined.";

//    for (MaterializedField f : batch.getSchema()) {
//      logger.debug("New Field: {}", f);
//    }
  }
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
@Override
public UnorderedReceiverBatch getBatch(FragmentContext context, UnorderedReceiver receiver, List<RecordBatch> children)
    throws ExecutionSetupException {
  assert children == null || children.isEmpty();
  IncomingBuffers bufHolder = context.getBuffers();
  assert bufHolder != null : "IncomingBuffers must be defined for any place a receiver is declared.";

  RawBatchBuffer[] buffers = bufHolder.getBuffers(receiver.getOppositeMajorFragmentId());
  assert buffers.length == 1;
  RawBatchBuffer buffer = buffers[0];
  return new UnorderedReceiverBatch(context, buffer, receiver);
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
public SingleSenderRootExec(FragmentContext context, RecordBatch batch, SingleSender config) throws OutOfMemoryException {
  super(context, context.newOperatorContext(config, null, false), config);
  this.incoming = batch;
  assert incoming != null;
  handle = context.getHandle();
  recMajor = config.getOppositeMajorFragmentId();
  tunnel = context.getDataTunnel(config.getDestination());
  oppositeHandle = handle.toBuilder()
      .setMajorFragmentId(config.getOppositeMajorFragmentId())
      .setMinorFragmentId(config.getOppositeMinorFragmentId())
      .build();
  tunnel = context.getDataTunnel(config.getDestination());
  tunnel.setTestInjectionControls(injector, context.getExecutionControls(), logger);
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
public LimitRecordBatch(Limit popConfig, FragmentContext context, RecordBatch incoming)
    throws OutOfMemoryException {
  super(popConfig, context, incoming);
  outgoingSv = new SelectionVector2(oContext.getAllocator());
  recordsToSkip = popConfig.getFirst();
  noEndLimit = popConfig.getLast() == null;
  if(!noEndLimit) {
    recordsLeft = popConfig.getLast() - recordsToSkip;
  }
  skipBatch = false;
}
 

import org.apache.drill.exec.record.RecordBatch; //导入依赖的package包/类
public CloseableRecordBatch getWriterBatch(FragmentContext context, RecordBatch incoming, EasyWriter writer)
    throws ExecutionSetupException {
  try {
    return new WriterRecordBatch(writer, incoming, context, getRecordWriter(context, writer));
  } catch(IOException e) {
    throw new ExecutionSetupException(String.format("Failed to create the WriterRecordBatch. %s", e.getMessage()), e);
  }
}