Method MaterializeNextRow

be/src/exec/data-source-scan-node.cc:342–452

}

Status DataSourceScanNode::MaterializeNextRow(const Timezone* local_tz,
    MemPool* tuple_pool, Tuple* tuple) {
  const vector<TColumnData>& cols = input_batch_->rows.cols;
  tuple->Init(tuple_desc_->byte_size());

  for (int i = 0; i < tuple_desc_->slots().size(); ++i) {
    const SlotDescriptor* slot_desc = tuple_desc_->slots()[i];
    void* slot = tuple->GetSlot(slot_desc->tuple_offset());
    const TColumnData& col = cols[i];

    if (col.is_null[next_row_idx_]) {
      tuple->SetNull(slot_desc->null_indicator_offset());
      continue;
    }

    // Get and increment the index into the values array (e.g. int_vals) for this col.
    int val_idx = cols_next_val_idx_[i]++;
    switch (slot_desc->type().type) {
      case TYPE_STRING: {
          if (val_idx >= col.string_vals.size()) {
            return Status(Substitute(ERROR_INVALID_COL_DATA, "STRING"));
          }
          const string& val = col.string_vals[val_idx];
          size_t val_size = val.size();
          char* buffer = reinterpret_cast<char*>(
              tuple_pool->TryAllocateUnaligned(val_size));
          if (UNLIKELY(buffer == NULL)) {
            string details = Substitute(ERROR_MEM_LIMIT_EXCEEDED, "MaterializeNextRow",
                val_size, "string slot");
            return tuple_pool->mem_tracker()->MemLimitExceeded(NULL, details, val_size);
          }
          memcpy(buffer, val.data(), val_size);
          reinterpret_cast<StringValue*>(slot)->Assign(buffer, val_size);
          break;
        }
      case TYPE_TINYINT:
        if (val_idx >= col.byte_vals.size()) {
          return Status(Substitute(ERROR_INVALID_COL_DATA, "TINYINT"));
        }
        *reinterpret_cast<int8_t*>(slot) = col.byte_vals[val_idx];
        break;
      case TYPE_SMALLINT:
        if (val_idx >= col.short_vals.size()) {
          return Status(Substitute(ERROR_INVALID_COL_DATA, "SMALLINT"));
        }
        *reinterpret_cast<int16_t*>(slot) = col.short_vals[val_idx];
        break;
      case TYPE_INT:
        if (val_idx >= col.int_vals.size()) {
          return Status(Substitute(ERROR_INVALID_COL_DATA, "INT"));
        }
        *reinterpret_cast<int32_t*>(slot) = col.int_vals[val_idx];
        break;
      case TYPE_BIGINT:
        if (val_idx >= col.long_vals.size()) {
          return Status(Substitute(ERROR_INVALID_COL_DATA, "BIGINT"));
        }
        *reinterpret_cast<int64_t*>(slot) = col.long_vals[val_idx];