ArrowResultSet Class Reference

#include <ArrowResultSet.h>

Public Member Functions
	ArrowResultSet (const std::shared_ptr< ResultSet > &rows, const std::vector< TargetMetaInfo > &targets_meta, const ExecutorDeviceType device_type=ExecutorDeviceType::CPU)
	ArrowResultSet (const std::shared_ptr< ResultSet > &rows, const std::vector< TargetMetaInfo > &targets_meta, const ExecutorDeviceType device_type, const size_t min_result_size_for_bulk_dictionary_fetch, const double max_dictionary_to_result_size_ratio_for_bulk_dictionary_fetch)
	ArrowResultSet (const std::shared_ptr< ResultSet > &rows, const ExecutorDeviceType device_type=ExecutorDeviceType::CPU)
std::vector< TargetValue >	getRowAt (const size_t index) const
std::vector< TargetValue >	getNextRow (const bool translate_strings, const bool decimal_to_double) const
size_t	colCount () const
SQLTypeInfo	getColType (const size_t col_idx) const
bool	definitelyHasNoRows () const
size_t	rowCount () const
size_t	entryCount () const
bool	isEmpty () const

Static Public Member Functions
static void	deallocateArrowResultBuffer (const ArrowResult &result, const ExecutorDeviceType device_type, const size_t device_id, std::shared_ptr< Data_Namespace::DataMgr > &data_mgr)

Public Attributes
	device_type

Private Member Functions
void	resultSetArrowLoopback (const ExecutorDeviceType device_type=ExecutorDeviceType::CPU)
void	resultSetArrowLoopback (const ExecutorDeviceType device_type, const size_t min_result_size_for_bulk_dictionary_fetch, const double max_dictionary_to_result_size_ratio_for_bulk_dictionary_fetch)
template<typename Type , typename ArrayType >
void	appendValue (std::vector< TargetValue > &row, const arrow::Array &column, const Type null_val, const size_t idx) const

Private Attributes
std::shared_ptr< ArrowResult >	results_
std::shared_ptr< ResultSet >	rows_
std::vector< TargetMetaInfo >	targets_meta_
std::shared_ptr < arrow::RecordBatch >	record_batch_
arrow::ipc::DictionaryMemo	dictionary_memo_
std::vector< std::shared_ptr < arrow::Array > >	columns_
size_t	crt_row_idx_
std::vector< TargetMetaInfo >	column_metainfo_

Detailed Description

Definition at line 108 of file ArrowResultSet.h.

Constructor & Destructor Documentation

ArrowResultSet::ArrowResultSet	(	const std::shared_ptr< ResultSet > &	rows,
		const std::vector< TargetMetaInfo > &	targets_meta,
		const ExecutorDeviceType	device_type = ExecutorDeviceType::CPU
	)

Definition at line 74 of file ArrowResultSet.cpp.

References column_metainfo_, columns_, field(), record_batch_, resultSetArrowLoopback(), and anonymous_namespace{ArrowResultSet.cpp}::type_from_arrow_field().

    : rows_(rows), targets_meta_(targets_meta), crt_row_idx_(0) {
  resultSetArrowLoopback(device_type);
  auto schema = record_batch_->schema();
  for (int i = 0; i < schema->num_fields(); ++i) {
    std::shared_ptr<arrow::Field> field = schema->field(i);
    SQLTypeInfo type_info = type_from_arrow_field(*schema->field(i));
    column_metainfo_.emplace_back(field->name(), type_info);
    columns_.emplace_back(record_batch_->column(i));
  }
}

Here is the call graph for this function:

ArrowResultSet::ArrowResultSet	(	const std::shared_ptr< ResultSet > &	rows,
		const std::vector< TargetMetaInfo > &	targets_meta,
		const ExecutorDeviceType	device_type,
		const size_t	min_result_size_for_bulk_dictionary_fetch,
		const double	max_dictionary_to_result_size_ratio_for_bulk_dictionary_fetch
	)

Definition at line 88 of file ArrowResultSet.cpp.

References column_metainfo_, columns_, field(), record_batch_, resultSetArrowLoopback(), and anonymous_namespace{ArrowResultSet.cpp}::type_from_arrow_field().

    : rows_(rows), targets_meta_(targets_meta), crt_row_idx_(0) {
  resultSetArrowLoopback(device_type,
                         min_result_size_for_bulk_dictionary_fetch,
                         max_dictionary_to_result_size_ratio_for_bulk_dictionary_fetch);
  auto schema = record_batch_->schema();
  for (int i = 0; i < schema->num_fields(); ++i) {
    std::shared_ptr<arrow::Field> field = schema->field(i);
    SQLTypeInfo type_info = type_from_arrow_field(*schema->field(i));
    column_metainfo_.emplace_back(field->name(), type_info);
    columns_.emplace_back(record_batch_->column(i));
  }
}

Here is the call graph for this function:

ArrowResultSet::ArrowResultSet ( const std::shared_ptr< ResultSet > &  rows, const ExecutorDeviceType  device_type = ExecutorDeviceType::CPU  )

inline

Definition at line 121 of file ArrowResultSet.h.

      : ArrowResultSet(rows, {}, device_type) {}

Member Function Documentation

template<typename Type , typename ArrayType >

void ArrowResultSet::appendValue ( std::vector< TargetValue > &  row, const arrow::Array &  column, const Type  null_val, const size_t  idx  ) const

private

Definition at line 108 of file ArrowResultSet.cpp.

                                                         {
  const auto& col = static_cast<const ArrayType&>(column);
  row.emplace_back(col.IsNull(idx) ? null_val : static_cast<Type>(col.Value(idx)));
}

size_t ArrowResultSet::colCount

(

)

const

Definition at line 245 of file ArrowResultSet.cpp.

References column_metainfo_.

                                      {
  return column_metainfo_.size();
}

void ArrowResultSet::deallocateArrowResultBuffer ( const ArrowResult &  result, const ExecutorDeviceType  device_type, const size_t  device_id, std::shared_ptr< Data_Namespace::DataMgr > &  data_mgr  )

static

Definition at line 1212 of file ArrowResultSetConverter.cpp.

References CHECK_EQ, CPU, ArrowResult::df_handle, ArrowResult::df_size, ArrowResult::sm_handle, ArrowResult::sm_size, and to_string().

Referenced by DBHandler::deallocate_df().

                                                    {
#ifndef _MSC_VER
  // CPU buffers skip the sm handle, serializing the entire RecordBatch to df.
  // Remove shared memory on sysmem
  if (!result.sm_handle.empty()) {
    CHECK_EQ(sizeof(key_t), result.sm_handle.size());
    const key_t& schema_key = *(key_t*)(&result.sm_handle[0]);
    auto shm_id = shmget(schema_key, result.sm_size, 0666);
    if (shm_id < 0) {
      throw std::runtime_error(
          "failed to get an valid shm ID w/ given shm key of the schema");
    }
    if (-1 == shmctl(shm_id, IPC_RMID, 0)) {
      throw std::runtime_error("failed to deallocate Arrow schema on errorno(" +
                               std::to_string(errno) + ")");
    }
  }

  if (device_type == ExecutorDeviceType::CPU) {
    CHECK_EQ(sizeof(key_t), result.df_handle.size());
    const key_t& df_key = *(key_t*)(&result.df_handle[0]);
    auto shm_id = shmget(df_key, result.df_size, 0666);
    if (shm_id < 0) {
      throw std::runtime_error(
          "failed to get an valid shm ID w/ given shm key of the data");
    }
    if (-1 == shmctl(shm_id, IPC_RMID, 0)) {
      throw std::runtime_error("failed to deallocate Arrow data frame");
    }
  }
  // CUDA buffers become owned by the caller, and will automatically be freed
  // TODO: What if the client never takes ownership of the result? we may want to
  // establish a check to see if the GPU buffer still exists, and then free it.
#endif
}

Here is the call graph for this function:

Here is the caller graph for this function:

bool ArrowResultSet::definitelyHasNoRows

(

)

const

Definition at line 254 of file ArrowResultSet.cpp.

References rowCount().

                                               {
  return !rowCount();
}

Here is the call graph for this function:

size_t ArrowResultSet::entryCount

(

)

const

Definition at line 264 of file ArrowResultSet.cpp.

References rowCount().

                                        {
  return rowCount();
}

Here is the call graph for this function:

SQLTypeInfo ArrowResultSet::getColType

(

const size_t

col_idx

)

const

Definition at line 249 of file ArrowResultSet.cpp.

References CHECK_LT, and column_metainfo_.

Referenced by getRowAt().

                                                                 {
  CHECK_LT(col_idx, column_metainfo_.size());
  return column_metainfo_[col_idx].get_type_info();
}

Here is the caller graph for this function:

std::vector< TargetValue > ArrowResultSet::getNextRow	(	const bool	translate_strings,
		const bool	decimal_to_double
	)		const

Definition at line 234 of file ArrowResultSet.cpp.

References CHECK_LT, crt_row_idx_, getRowAt(), and rowCount().

                                                                                        {
  if (crt_row_idx_ == rowCount()) {
    return {};
  }
  CHECK_LT(crt_row_idx_, rowCount());
  auto row = getRowAt(crt_row_idx_);
  ++crt_row_idx_;
  return row;
}

Here is the call graph for this function:

std::vector< TargetValue > ArrowResultSet::getRowAt

(

const size_t

index

)

const

Definition at line 142 of file ArrowResultSet.cpp.

References CHECK, CHECK_EQ, CHECK_LT, columns_, getColType(), inline_fp_null_value< double >(), inline_fp_null_value< float >(), inline_int_null_val(), kBIGINT, kDATE, kDOUBLE, kENCODING_DICT, kFLOAT, kINT, kSMALLINT, kTEXT, kTIME, kTIMESTAMP, kTINYINT, record_batch_, and rowCount().

Referenced by getNextRow(), and ArrowResultSetRowIterator::operator*().

                                                                        {
  if (index >= rowCount()) {
    return {};
  }

  CHECK_LT(index, rowCount());
  std::vector<TargetValue> row;
  for (int i = 0; i < record_batch_->num_columns(); ++i) {
    const auto& column = *columns_[i];
    const auto& column_typeinfo = getColType(i);
    switch (column_typeinfo.get_type()) {
      case kTINYINT: {
        CHECK_EQ(arrow::Type::INT8, column.type_id());
        appendValue<int64_t, arrow::Int8Array>(
            row, column, inline_int_null_val(column_typeinfo), index);
        break;
      }
      case kSMALLINT: {
        CHECK_EQ(arrow::Type::INT16, column.type_id());
        appendValue<int64_t, arrow::Int16Array>(
            row, column, inline_int_null_val(column_typeinfo), index);
        break;
      }
      case kINT: {
        CHECK_EQ(arrow::Type::INT32, column.type_id());
        appendValue<int64_t, arrow::Int32Array>(
            row, column, inline_int_null_val(column_typeinfo), index);
        break;
      }
      case kBIGINT: {
        CHECK_EQ(arrow::Type::INT64, column.type_id());
        appendValue<int64_t, arrow::Int64Array>(
            row, column, inline_int_null_val(column_typeinfo), index);
        break;
      }
      case kFLOAT: {
        CHECK_EQ(arrow::Type::FLOAT, column.type_id());
        appendValue<float, arrow::FloatArray>(
            row, column, inline_fp_null_value<float>(), index);
        break;
      }
      case kDOUBLE: {
        CHECK_EQ(arrow::Type::DOUBLE, column.type_id());
        appendValue<double, arrow::DoubleArray>(
            row, column, inline_fp_null_value<double>(), index);
        break;
      }
      case kTEXT: {
        CHECK_EQ(kENCODING_DICT, column_typeinfo.get_compression());
        CHECK_EQ(arrow::Type::DICTIONARY, column.type_id());
        const auto& dict_column = static_cast<const arrow::DictionaryArray&>(column);
        if (dict_column.IsNull(index)) {
          row.emplace_back(NullableString(nullptr));
        } else {
          const auto& indices =
              static_cast<const arrow::Int32Array&>(*dict_column.indices());
          const auto& dictionary =
              static_cast<const arrow::StringArray&>(*dict_column.dictionary());
          row.emplace_back(dictionary.GetString(indices.Value(index)));
        }
        break;
      }
      case kTIMESTAMP: {
        CHECK_EQ(arrow::Type::TIMESTAMP, column.type_id());
        appendValue<int64_t, arrow::TimestampArray>(
            row, column, inline_int_null_val(column_typeinfo), index);
        break;
      }
      case kDATE: {
        // TODO(wamsi): constexpr?
        CHECK(arrow::Type::DATE32 == column.type_id() ||
              arrow::Type::DATE64 == column.type_id());
        column_typeinfo.is_date_in_days()
            ? appendValue<int64_t, arrow::Date32Array>(
                  row, column, inline_int_null_val(column_typeinfo), index)
            : appendValue<int64_t, arrow::Date64Array>(
                  row, column, inline_int_null_val(column_typeinfo), index);
        break;
      }
      case kTIME: {
        CHECK_EQ(arrow::Type::TIME32, column.type_id());
        appendValue<int64_t, arrow::Time32Array>(
            row, column, inline_int_null_val(column_typeinfo), index);
        break;
      }
      default:
        CHECK(false);
    }
  }
  return row;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bool ArrowResultSet::isEmpty

(

)

const

Definition at line 270 of file ArrowResultSet.cpp.

References rowCount().

                                   {
  return rowCount() == static_cast<size_t>(0);
}

Here is the call graph for this function:

void ArrowResultSet::resultSetArrowLoopback ( const ExecutorDeviceType  device_type = ExecutorDeviceType::CPU )

private

Definition at line 274 of file ArrowResultSet.cpp.

References ArrowResultSetConverter::default_min_result_size_for_bulk_dictionary_fetch.

Referenced by ArrowResultSet().

                                                                                {
  resultSetArrowLoopback(
      device_type,
      ArrowResultSetConverter::default_min_result_size_for_bulk_dictionary_fetch,
      ArrowResultSetConverter::
          default_max_dictionary_to_result_size_ratio_for_bulk_dictionary_fetch);
}

Here is the caller graph for this function:

void ArrowResultSet::resultSetArrowLoopback ( const ExecutorDeviceType  device_type, const size_t  min_result_size_for_bulk_dictionary_fetch, const double  max_dictionary_to_result_size_ratio_for_bulk_dictionary_fetch  )

private

Definition at line 282 of file ArrowResultSet.cpp.

References ARROW_ASSIGN_OR_THROW, ARROW_THROW_NOT_OK, CHECK_EQ, device_type, dictionary_memo_, record_batch_, results_, rows_, targets_meta_, to_string(), and WIRE.

                                                                                {
  std::vector<std::string> col_names;

  if (!targets_meta_.empty()) {
    for (auto& meta : targets_meta_) {
      col_names.push_back(meta.get_resname());
    }
  } else {
    for (unsigned int i = 0; i < rows_->colCount(); i++) {
      col_names.push_back("col_" + std::to_string(i));
    }
  }

  // We convert the given rows to arrow, which gets serialized
  // into a buffer by Arrow Wire.
  auto converter = ArrowResultSetConverter(
      rows_,
      col_names,
      -1,
      min_result_size_for_bulk_dictionary_fetch,
      max_dictionary_to_result_size_ratio_for_bulk_dictionary_fetch);
  converter.transport_method_ = ArrowTransport::WIRE;
  converter.device_type_ = device_type;

  // Lifetime of the result buffer is that of ArrowResultSet
  results_ = std::make_shared<ArrowResult>(converter.getArrowResult());

  // Create a reader for reading back serialized
  arrow::io::BufferReader reader(
      reinterpret_cast<const uint8_t*>(results_->df_buffer.data()), results_->df_size);

  ARROW_ASSIGN_OR_THROW(auto batch_reader,
                        arrow::ipc::RecordBatchStreamReader::Open(&reader));

  ARROW_THROW_NOT_OK(batch_reader->ReadNext(&record_batch_));

  // Collect dictionaries from the record batch into the dictionary memo.
  ARROW_THROW_NOT_OK(
      arrow::ipc::internal::CollectDictionaries(*record_batch_, &dictionary_memo_));

  CHECK_EQ(record_batch_->schema()->num_fields(), record_batch_->num_columns());
}

Here is the call graph for this function:

size_t ArrowResultSet::rowCount

(

)

const

Definition at line 258 of file ArrowResultSet.cpp.

References record_batch_.

Referenced by definitelyHasNoRows(), entryCount(), getNextRow(), getRowAt(), and isEmpty().

                                      {
  return record_batch_->num_rows();
}

Here is the caller graph for this function:

Member Data Documentation

std::vector<TargetMetaInfo> ArrowResultSet::column_metainfo_

private

Definition at line 190 of file ArrowResultSet.h.

Referenced by ArrowResultSet(), colCount(), and getColType().

std::vector<std::shared_ptr<arrow::Array> > ArrowResultSet::columns_

private

Definition at line 188 of file ArrowResultSet.h.

Referenced by ArrowResultSet(), and getRowAt().

size_t ArrowResultSet::crt_row_idx_

mutableprivate

Definition at line 189 of file ArrowResultSet.h.

Referenced by getNextRow().

ArrowResultSet::device_type

Initial value:

{}

  ArrowResultSetRowIterator rowIterator(size_t from_index,
                                        bool translate_strings,
                                        bool decimal_to_double) const {
    ArrowResultSetRowIterator iter(this);
    for (size_t i = 0; i < from_index; i++) {
      ++iter;
    }

    return iter;
  }

  ArrowResultSetRowIterator rowIterator(bool translate_strings,
                                        bool decimal_to_double) const {
    return rowIterator(0, translate_strings, decimal_to_double);
  }

  std::vector<std::string> getDictionaryStrings(const size_t col_idx) const

Definition at line 123 of file ArrowResultSet.h.

Referenced by resultSetArrowLoopback().

arrow::ipc::DictionaryMemo ArrowResultSet::dictionary_memo_

private

Definition at line 184 of file ArrowResultSet.h.

Referenced by resultSetArrowLoopback().

std::shared_ptr<arrow::RecordBatch> ArrowResultSet::record_batch_

private

Definition at line 183 of file ArrowResultSet.h.

Referenced by ArrowResultSet(), getRowAt(), resultSetArrowLoopback(), and rowCount().

std::shared_ptr<ArrowResult> ArrowResultSet::results_

private

Definition at line 180 of file ArrowResultSet.h.

Referenced by resultSetArrowLoopback().

std::shared_ptr<ResultSet> ArrowResultSet::rows_

private

Definition at line 181 of file ArrowResultSet.h.

Referenced by resultSetArrowLoopback().

std::vector<TargetMetaInfo> ArrowResultSet::targets_meta_

private

Definition at line 182 of file ArrowResultSet.h.

Referenced by resultSetArrowLoopback().

---

The documentation for this class was generated from the following files:

• /home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/ArrowResultSet.h
• /home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/ArrowResultSet.cpp
• /home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/ArrowResultSetConverter.cpp