ArrowImporter.h

Go to the documentation of this file.

/*
 * Copyright 2019 OmniSci, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#ifndef ARROW_IMPORTER_H
#define ARROW_IMPORTER_H

#include <cstdlib>
#include <ctime>
#include <map>
#include <mutex>

#include <arrow/api.h>
#include <arrow/io/api.h>
#include <boost/algorithm/string.hpp>
#include <boost/variant.hpp>

#include "Shared/SqlTypesLayout.h"
#include "Shared/ThreadController.h"
#include "Shared/sqltypes.h"

using namespace arrow;

struct ArrowImporterException : std::runtime_error {
  using std::runtime_error::runtime_error;
};

template <typename T = ArrowImporterException>
inline void arrow_throw_if(const bool cond, const std::string& message) {
  if (cond) {
    // work around race from goooogle log
    static std::mutex mtx;
    std::unique_lock<std::mutex> lock(mtx);
    LOG(ERROR) << message;
    throw T(message);
  }
}

#ifdef ENABLE_IMPORT_PARQUET
#include <parquet/api/reader.h>
#include <parquet/api/writer.h>
#include <parquet/arrow/reader.h>
#include <parquet/exception.h>
#endif  // ENABLE_IMPORT_PARQUET

#include "arrow/util/decimal.h"

namespace {

using VarValue =
    boost::variant<bool, float, double, int64_t, std::string, void*, Decimal128>;

template <typename T>
using enable_if_integral = typename std::enable_if_t<std::is_integral<T>::value, T>;
template <typename T>
using enable_if_integral_not_bool =
    typename std::enable_if_t<std::is_integral<T>::value && !std::is_same<T, bool>::value,
                              T>;
template <typename T>
using enable_if_floating = typename std::enable_if_t<std::is_floating_point<T>::value, T>;

#define exprtype(expr) std::decay_t<decltype(expr)>

inline std::string error_context(const ColumnDescriptor* cd,
                                 Importer_NS::BadRowsTracker* const bad_rows_tracker) {
  return bad_rows_tracker ? "File " + bad_rows_tracker->file_name + ", row-group " +
                                std::to_string(bad_rows_tracker->row_group) +
                                (cd ? ", column " + cd->columnName + ": " : "")
                          : std::string();
}

template <typename SrcType, typename DstType>
inline VarValue get_numeric_value(const Array& array, const int64_t idx) {
  using ArrayType = typename TypeTraits<SrcType>::ArrayType;
  return (DstType) static_cast<const ArrayType&>(array).Value(idx);
}

template <typename SrcType>
inline VarValue get_string_value(const Array& array, const int64_t idx) {
  using ArrayType = typename TypeTraits<SrcType>::ArrayType;
  return static_cast<const ArrayType&>(array).GetString(idx);
}

#define NUMERIC_CASE(tid, src_type, var_type) \
  case Type::tid:                             \
    return get_numeric_value<src_type, var_type>;
#define STRING_CASE(tid, src_type) \
  case Type::tid:                  \
    return get_string_value<src_type>;

inline auto value_getter(const Array& array,
                         const ColumnDescriptor* cd,
                         Importer_NS::BadRowsTracker* const bad_rows_tracker) {
  switch (array.type_id()) {
    NUMERIC_CASE(BOOL, BooleanType, bool)
    NUMERIC_CASE(UINT8, UInt8Type, int64_t)
    NUMERIC_CASE(UINT16, UInt16Type, int64_t)
    NUMERIC_CASE(UINT32, UInt32Type, int64_t)
    NUMERIC_CASE(UINT64, Int64Type, int64_t)
    NUMERIC_CASE(INT8, Int8Type, int64_t)
    NUMERIC_CASE(INT16, Int16Type, int64_t)
    NUMERIC_CASE(INT32, Int32Type, int64_t)
    NUMERIC_CASE(INT64, Int64Type, int64_t)
    NUMERIC_CASE(FLOAT, FloatType, float)
    NUMERIC_CASE(DOUBLE, DoubleType, double)
    NUMERIC_CASE(DATE32, Date32Type, int64_t)
    NUMERIC_CASE(DATE64, Date64Type, int64_t)
    NUMERIC_CASE(TIME64, Time64Type, int64_t)
    NUMERIC_CASE(TIME32, Time32Type, int64_t)
    NUMERIC_CASE(TIMESTAMP, TimestampType, int64_t)
    NUMERIC_CASE(DECIMAL, Decimal128Type, Decimal128)
    STRING_CASE(STRING, StringType)
    STRING_CASE(BINARY, BinaryType)
    default:
      arrow_throw_if(true,
                     error_context(cd, bad_rows_tracker) + "Parquet type " +
                         array.type()->name() + " is not supported");
      throw;
  }
}

inline void type_conversion_error(const std::string pt,
                                  const ColumnDescriptor* cd,
                                  Importer_NS::BadRowsTracker* const bad_rows_tracker) {
  arrow_throw_if(true,
                 error_context(cd, bad_rows_tracker) +
                     "Invalid type conversion from parquet " + pt + " type to " +
                     cd->columnType.get_type_name());
}

template <typename DATA_TYPE, typename VALUE_TYPE>
inline void data_conversion_error(const VALUE_TYPE v,
                                  const ColumnDescriptor* cd,
                                  Importer_NS::BadRowsTracker* const bad_rows_tracker) {
  arrow_throw_if(true,
                 error_context(cd, bad_rows_tracker) +
                     "Invalid data conversion from parquet value " + std::to_string(v) +
                     " to " + std::to_string(DATA_TYPE(v)));
}

inline void data_conversion_error(const std::string& v,
                                  const ColumnDescriptor* cd,
                                  Importer_NS::BadRowsTracker* const bad_rows_tracker) {
  arrow_throw_if(true,
                 error_context(cd, bad_rows_tracker) +
                     "Invalid data conversion from parquet string '" + v + "' to " +
                     cd->columnType.get_type_name() + " column type");
}

// models the variant data buffers of TypedImportBuffer (LHS)
struct DataBufferBase {
  const ColumnDescriptor* cd;
  const Array& array;
  Importer_NS::BadRowsTracker* const bad_rows_tracker;
  // in case of arrow-decimal to omni-decimal conversion
  // dont get/set these info on every row of arrow array
  const DataType& arrow_type;
  const int arrow_decimal_scale;
  const SQLTypeInfo old_type;
  const SQLTypeInfo new_type;
  DataBufferBase(const ColumnDescriptor* cd,
                 const Array& array,
                 Importer_NS::BadRowsTracker* const bad_rows_tracker)
      : cd(cd)
      , array(array)
      , bad_rows_tracker(bad_rows_tracker)
      , arrow_type(*array.type())
      , arrow_decimal_scale(arrow_type.id() == Type::DECIMAL
                                ? static_cast<const Decimal128Type&>(arrow_type).scale()
                                : 0)
      , old_type(cd->columnType.get_type(),
                 cd->columnType.get_dimension(),
                 arrow_decimal_scale,
                 true)
      , new_type(cd->columnType.get_type(),
                 cd->columnType.get_dimension(),
                 cd->columnType.get_scale(),
                 true) {}
};

template <typename DATA_TYPE>
struct DataBuffer : DataBufferBase {
  std::vector<DATA_TYPE>& buffer;
  DataBuffer(const ColumnDescriptor* cd,
             const Array& array,
             std::vector<DATA_TYPE>& buffer,
             Importer_NS::BadRowsTracker* const bad_rows_tracker)
      : DataBufferBase(cd, array, bad_rows_tracker), buffer(buffer) {}
};

constexpr int64_t kMillisecondsInSecond = 1000L;
constexpr int64_t kMicrosecondsInSecond = 1000L * 1000L;
constexpr int64_t kNanosecondsinSecond = 1000L * 1000L * 1000L;
constexpr int32_t kSecondsInDay = 86400;

static const std::map<std::pair<int32_t, arrow::TimeUnit::type>,
                      std::pair<SQLOps, int64_t>>
    _precision_scale_lookup{{{0, TimeUnit::MILLI}, {kDIVIDE, kMillisecondsInSecond}},
                            {{0, TimeUnit::MICRO}, {kDIVIDE, kMicrosecondsInSecond}},
                            {{0, TimeUnit::NANO}, {kDIVIDE, kNanosecondsinSecond}},
                            {{3, TimeUnit::SECOND}, {kMULTIPLY, kMicrosecondsInSecond}},
                            {{3, TimeUnit::MICRO}, {kDIVIDE, kMillisecondsInSecond}},
                            {{3, TimeUnit::NANO}, {kDIVIDE, kMicrosecondsInSecond}},
                            {{6, TimeUnit::SECOND}, {kMULTIPLY, kMicrosecondsInSecond}},
                            {{6, TimeUnit::MILLI}, {kMULTIPLY, kMillisecondsInSecond}},
                            {{6, TimeUnit::NANO}, {kDIVIDE, kMillisecondsInSecond}},
                            {{9, TimeUnit::SECOND}, {kMULTIPLY, kNanosecondsinSecond}},
                            {{9, TimeUnit::MILLI}, {kMULTIPLY, kMicrosecondsInSecond}},
                            {{9, TimeUnit::MICRO}, {kMULTIPLY, kMillisecondsInSecond}}};

// models the variant values of Arrow Array (RHS)
template <typename VALUE_TYPE>
struct ArrowValueBase {
  const DataBufferBase& data;
  const VALUE_TYPE v;
  const int32_t dimension;
  ArrowValueBase(const DataBufferBase& data, const VALUE_TYPE& v)
      : data(data)
      , v(v)
      , dimension(data.cd->columnType.is_high_precision_timestamp()
                      ? data.cd->columnType.get_dimension()
                      : 0) {}
  template <bool enabled = std::is_integral<VALUE_TYPE>::value>
  int64_t resolve_time(const VALUE_TYPE& v, std::enable_if_t<enabled>* = 0) const {
    const auto& type_id = data.arrow_type.id();
    if (type_id == Type::DATE32 || type_id == Type::DATE64) {
      auto& date_type = static_cast<const DateType&>(data.arrow_type);
      switch (date_type.unit()) {
        case DateUnit::DAY:
          return v * kSecondsInDay;
        case DateUnit::MILLI:
          return v / kMillisecondsInSecond;
      }
    } else if (type_id == Type::TIME32 || type_id == Type::TIME64 ||
               type_id == Type::TIMESTAMP) {
      auto& time_type = static_cast<const TimeType&>(data.arrow_type);
      const auto result =
          _precision_scale_lookup.find(std::make_pair(dimension, time_type.unit()));
      if (result != _precision_scale_lookup.end()) {
        const auto scale = result->second;
        return scale.first == kMULTIPLY ? v * scale.second : v / scale.second;
      } else {
        return v;
      }
    }
    UNREACHABLE() << data.arrow_type << " is not a valid Arrow time or date type";
    return 0;
  }
  template <bool enabled = std::is_integral<VALUE_TYPE>::value>
  int64_t resolve_time(const VALUE_TYPE& v, std::enable_if_t<!enabled>* = 0) const {
    static_assert(enabled, "unreachable");
    return 0;
  }
};

template <typename VALUE_TYPE>
struct ArrowValue : ArrowValueBase<VALUE_TYPE> {};

template <>
struct ArrowValue<void*> : ArrowValueBase<void*> {
  using VALUE_TYPE = void*;
  ArrowValue(const DataBufferBase& data, const VALUE_TYPE& v)
      : ArrowValueBase<VALUE_TYPE>(data, v) {}
  template <typename DATA_TYPE, typename = enable_if_integral<DATA_TYPE>>
  explicit operator const DATA_TYPE() const {
    return inline_fixed_encoding_null_val(data.cd->columnType);
  }
  template <typename DATA_TYPE, typename = enable_if_floating<DATA_TYPE>>
  explicit operator DATA_TYPE() const {
    return inline_fp_null_val(data.cd->columnType);
  }
  explicit operator const std::string() const { return std::string(); }
};

template <>
struct ArrowValue<bool> : ArrowValueBase<bool> {
  using VALUE_TYPE = bool;
  ArrowValue(const DataBufferBase& data, const VALUE_TYPE& v)
      : ArrowValueBase<VALUE_TYPE>(data, v) {}
  template <typename DATA_TYPE, typename = enable_if_integral<DATA_TYPE>>
  explicit operator const DATA_TYPE() const {
    if (!(data.cd->columnType.is_number() || data.cd->columnType.is_boolean())) {
      type_conversion_error("bool", data.cd, data.bad_rows_tracker);
    }
    return v;
  }
  template <typename DATA_TYPE, typename = enable_if_floating<DATA_TYPE>>
  explicit operator DATA_TYPE() const {
    return v ? 1 : 0;
  }
  explicit operator const std::string() const { return v ? "T" : "F"; }
};

template <>
struct ArrowValue<float> : ArrowValueBase<float> {
  using VALUE_TYPE = float;
  ArrowValue(const DataBufferBase& data, const VALUE_TYPE& v)
      : ArrowValueBase<VALUE_TYPE>(data, v) {}
  template <typename DATA_TYPE, typename = enable_if_integral<DATA_TYPE>>
  explicit operator const DATA_TYPE() const {
    const auto ti = data.cd->columnType;
    DATA_TYPE v = ti.is_decimal() ? this->v * pow(10, ti.get_scale()) : this->v;
    if (!(std::numeric_limits<DATA_TYPE>::lowest() < v &&
          v <= std::numeric_limits<DATA_TYPE>::max())) {
      data_conversion_error<DATA_TYPE>(v, data.cd, data.bad_rows_tracker);
    }
    return v;
  }
  template <typename DATA_TYPE, typename = enable_if_floating<DATA_TYPE>>
  explicit operator DATA_TYPE() const {
    return v;
  }
  explicit operator const std::string() const { return std::to_string(v); }
};

template <>
struct ArrowValue<double> : ArrowValueBase<double> {
  using VALUE_TYPE = double;
  ArrowValue(const DataBufferBase& data, const VALUE_TYPE& v)
      : ArrowValueBase<VALUE_TYPE>(data, v) {}
  template <typename DATA_TYPE, typename = enable_if_integral<DATA_TYPE>>
  explicit operator const DATA_TYPE() const {
    const auto ti = data.cd->columnType;
    DATA_TYPE v = ti.is_decimal() ? this->v * pow(10, ti.get_scale()) : this->v;
    if (!(std::numeric_limits<DATA_TYPE>::lowest() < v &&
          v <= std::numeric_limits<DATA_TYPE>::max())) {
      data_conversion_error<DATA_TYPE>(v, data.cd, data.bad_rows_tracker);
    }
    return v;
  }
  template <typename DATA_TYPE, typename = enable_if_floating<DATA_TYPE>>
  explicit operator DATA_TYPE() const {
    if (std::is_same<DATA_TYPE, float>::value) {
      if (!(std::numeric_limits<float>::lowest() < v &&
            v <= std::numeric_limits<float>::max())) {
        data_conversion_error<float>(v, data.cd, data.bad_rows_tracker);
      }
    }
    return v;
  }
  explicit operator const std::string() const { return std::to_string(v); }
};

template <>
struct ArrowValue<int64_t> : ArrowValueBase<int64_t> {
  using VALUE_TYPE = int64_t;
  ArrowValue(const DataBufferBase& data, const VALUE_TYPE& v)
      : ArrowValueBase<VALUE_TYPE>(data, v) {}
  template <typename DATA_TYPE, typename = enable_if_integral<DATA_TYPE>>
  explicit operator const DATA_TYPE() const {
    int64_t v = this->v;
    if (std::is_same<int64_t, DATA_TYPE>::value) {
    } else if (std::numeric_limits<DATA_TYPE>::lowest() < v &&
               v <= std::numeric_limits<DATA_TYPE>::max()) {
    } else {
      data_conversion_error<DATA_TYPE>(v, data.cd, data.bad_rows_tracker);
    }
    if (data.cd->columnType.is_time()) {
      v = this->resolve_time(v);
    }
    return v;
  }
  template <typename DATA_TYPE, typename = enable_if_floating<DATA_TYPE>>
  explicit operator DATA_TYPE() const {
    return v;
  }
  explicit operator const std::string() const {
    const auto& type_id = data.arrow_type.id();
    if (type_id == Type::DATE32 || type_id == Type::DATE64) {
      auto& date_type = static_cast<const DateType&>(data.arrow_type);
      SQLTypeInfo ti(kDATE);
      Datum datum{.bigintval = date_type.unit() == DateUnit::MILLI
                                   ? v / kMicrosecondsInSecond
                                   : v};
      return DatumToString(datum, ti);
    } else if (type_id == Type::TIME32 || type_id == Type::TIME64 ||
               type_id == Type::TIMESTAMP) {
      auto& time_type = static_cast<const TimeType&>(data.arrow_type);
      const auto result =
          _precision_scale_lookup.find(std::make_pair(0, time_type.unit()));
      int64_t divisor{1};
      if (result != _precision_scale_lookup.end()) {
        divisor = result->second.second;
      }
      SQLTypeInfo ti(kTIMESTAMP);
      Datum datum{.bigintval = v / divisor};
      auto time_str = DatumToString(datum, ti);
      if (divisor != 1 && v % divisor) {
        time_str += "." + std::to_string(v % divisor);
      }
      return time_str;
    }
    return std::to_string(v);
  }
};

template <>
struct ArrowValue<std::string> : ArrowValueBase<std::string> {
  using VALUE_TYPE = std::string;
  ArrowValue(const DataBufferBase& data, const VALUE_TYPE& v)
      : ArrowValueBase<VALUE_TYPE>(data, v) {}
  explicit operator const bool() const {
    if (v.size() == 0) {
      return inline_int_null_value<int8_t>();
    }
    try {
      SQLTypeInfo ti(kBOOLEAN);
      auto datum = StringToDatum(v, ti);
      return datum.boolval;
    } catch (...) {
      data_conversion_error(v, data.cd, data.bad_rows_tracker);
      return false;
    }
  }
  template <typename DATA_TYPE, typename = enable_if_integral_not_bool<DATA_TYPE>>
  explicit operator const DATA_TYPE() const {
    if (v.size() == 0) {
      return inline_fixed_encoding_null_val(data.cd->columnType);
    }
    try {
      auto ti = data.cd->columnType;
      auto datum = StringToDatum(v, ti);
      return datum.bigintval;
    } catch (...) {
      data_conversion_error(v, data.cd, data.bad_rows_tracker);
      return 0;
    }
  }
  template <typename DATA_TYPE, typename = enable_if_floating<DATA_TYPE>>
  explicit operator DATA_TYPE() const {
    return atof(v.data());
  }
  explicit operator const std::string() const { return v; }
};

template <>
struct ArrowValue<Decimal128> : ArrowValueBase<Decimal128> {
  using VALUE_TYPE = Decimal128;
  ArrowValue(const DataBufferBase& data, const VALUE_TYPE& v)
      : ArrowValueBase<VALUE_TYPE>(data, v) {
    // omni decimal has only 64 bits
    arrow_throw_if(!(v.high_bits() == 0 || v.high_bits() == -1),
                   error_context(data.cd, data.bad_rows_tracker) +
                       "Truncation error on Arrow Decimal128 value");
  }
  template <typename DATA_TYPE, typename = enable_if_integral<DATA_TYPE>>
  explicit operator const DATA_TYPE() const {
    int64_t v = static_cast<int64_t>(this->v);
    if (data.cd->columnType.is_decimal()) {
      return convert_decimal_value_to_scale(v, data.old_type, data.new_type);
    }
    if (data.arrow_decimal_scale) {
      v = std::llround(v / pow(10, data.arrow_decimal_scale));
    }
    if (std::is_same<int64_t, DATA_TYPE>::value) {
    } else if (std::numeric_limits<DATA_TYPE>::lowest() < v &&
               v <= std::numeric_limits<DATA_TYPE>::max()) {
    } else {
      data_conversion_error<DATA_TYPE>(v, data.cd, data.bad_rows_tracker);
    }
    return v;
  }
  template <typename DATA_TYPE, typename = enable_if_floating<DATA_TYPE>>
  explicit operator DATA_TYPE() const {
    int64_t v = static_cast<int64_t>(this->v);
    return data.arrow_decimal_scale ? v / pow(10, data.arrow_decimal_scale) : v;
  }
  explicit operator const std::string() const {
    return v.ToString(data.arrow_decimal_scale);
  }
};

// appends a converted RHS value to LHS data block
template <typename DATA_TYPE>
inline auto& operator<<(DataBuffer<DATA_TYPE>& data, const VarValue& var) {
  boost::apply_visitor(
      [&data](const auto& v) {
        data.buffer.push_back(DATA_TYPE(ArrowValue<exprtype(v)>(data, v)));
      },
      var);
  return data;
}

}  // namespace
#endif  // ARROW_IMPORTER_H