Finish scaffolding. Get fully compiling again and original parquet-arrow test suite passing

Change-Id: Icb260f6ffc4f41ee7519653bf8d3f48c2da30091

cpp/src/arrow/compute/compute-test.cc

@@ -82,6 +82,10 @@ void CheckImplicitConstructor(enum Datum::type expected_kind) {
 
 TEST(TestDatum, ImplicitConstructors) {
   CheckImplicitConstructor<Array>(Datum::ARRAY);
+
+  // Instantiate from array subclass
+  CheckImplicitConstructor<BinaryArray>(Datum::ARRAY);
+
   CheckImplicitConstructor<ChunkedArray>(Datum::CHUNKED_ARRAY);
   CheckImplicitConstructor<RecordBatch>(Datum::RECORD_BATCH);
   CheckImplicitConstructor<Table>(Datum::TABLE);

cpp/src/arrow/compute/kernel.h

@@ -65,8 +65,10 @@ struct ARROW_EXPORT Datum {
       : value(value) {}
   Datum(const std::shared_ptr<ArrayData>& value)  // NOLINT implicit conversion
       : value(value) {}
+
   Datum(const std::shared_ptr<Array>& value)  // NOLINT implicit conversion
       : Datum(value ? value->data() : NULLPTR) {}
+
   Datum(const std::shared_ptr<ChunkedArray>& value)  // NOLINT implicit conversion
       : value(value) {}
   Datum(const std::shared_ptr<RecordBatch>& value)  // NOLINT implicit conversion
@@ -76,6 +78,13 @@ struct ARROW_EXPORT Datum {
   Datum(const std::vector<Datum>& value)  // NOLINT implicit conversion
       : value(value) {}
 
+  // Cast from subtypes of Array to Datum
+  template <typename T,
+            typename = typename std::enable_if<std::is_base_of<Array, T>::value,
+                                               void>::type>
+  Datum(const std::shared_ptr<T>& value)  // NOLINT implicit conversion
+      : Datum(std::shared_ptr<Array>(value)) {}
+
   ~Datum() {}
 
   Datum(const Datum& other) noexcept { this->value = other.value; }

cpp/src/arrow/table.h

@@ -46,7 +46,8 @@ class ARROW_EXPORT ChunkedArray {
   explicit ChunkedArray(const ArrayVector& chunks);
 
   /// \brief Construct a chunked array from a single Array
-  explicit ChunkedArray(const std::shared_ptr<Array>& chunk) : ChunkedArray({chunk}) {}
+  explicit ChunkedArray(const std::shared_ptr<Array>& chunk)
+      : ChunkedArray(ArrayVector({chunk})) {}
 
   /// \brief Construct a chunked array from a vector of arrays and a data type
   ///

cpp/src/parquet/arrow/reader.cc

@@ -92,7 +92,7 @@ static inline int64_t impala_timestamp_to_nanoseconds(const Int96& impala_timest
 template <typename ArrowType>
 using ArrayType = typename ::arrow::TypeTraits<ArrowType>::ArrayType;
 
-namespace internal {
+namespace {
 
 Status GetSingleChunk(const ChunkedArray& chunked, std::shared_ptr<Array>* out) {
   DCHECK_GT(chunked.num_chunks(), 0);
@@ -103,7 +103,7 @@ Status GetSingleChunk(const ChunkedArray& chunked, std::shared_ptr<Array>* out)
   return Status::OK();
 }
 
-}  // namespace internal
+}  // namespace
 
 // ----------------------------------------------------------------------
 // Iteration utilities
@@ -814,7 +814,7 @@ Status PrimitiveImpl::WrapIntoListArray(Datum* inout_array) {
     flat_array = inout_array->chunked_array()->chunk(0);
   } else {
     DCHECK_EQ(Datum::ARRAY, inout_array->kind());
-    flat_array = inout_array->array();
+    flat_array = inout_array->make_array();
   }
 
   const int16_t* def_levels = record_reader_->def_levels();
@@ -919,7 +919,7 @@ Status PrimitiveImpl::WrapIntoListArray(Datum* inout_array) {
     valid_bits.emplace_back(std::static_pointer_cast<BooleanArray>(array)->values());
   }
 
-  std::shared_ptr<Array> output(*flat_array);
+  std::shared_ptr<Array> output = flat_array;
   for (int64_t j = list_depth - 1; j >= 0; j--) {
     auto list_type =
         ::arrow::list(::arrow::field("item", output->type(), nullable[j + 1]));
@@ -1089,8 +1089,11 @@ struct TransferFunctor<::arrow::Date64Type, Int32Type> {
 template <typename ArrowType, typename ParquetType>
 struct TransferFunctor<
     ArrowType, ParquetType,
-    typename std::enable_if<std::is_same<ParquetType, ByteArrayType>::value ||
-                            std::is_same<ParquetType, FLBAType>::value>::type> {
+    typename std::enable_if<
+        (std::is_base_of<::arrow::BinaryType, ArrowType>::value ||
+         std::is_same<::arrow::FixedSizeBinaryType, ArrowType>::value) &&
+        (std::is_same<ParquetType, ByteArrayType>::value ||
+         std::is_same<ParquetType, FLBAType>::value)>::type> {
   Status operator()(RecordReader* reader, MemoryPool* pool,
                     const std::shared_ptr<::arrow::DataType>& type, Datum* out) {
     std::vector<std::shared_ptr<Array>> chunks = reader->GetBuilderChunks();
@@ -1214,6 +1217,12 @@ static inline void RawBytesToDecimalBytes(const uint8_t* value, int32_t byte_wid
 // ----------------------------------------------------------------------
 // BYTE_ARRAY / FIXED_LEN_BYTE_ARRAY -> Decimal128
 
+template <typename T>
+Status ConvertToDecimal128(const Array& array, const std::shared_ptr<::arrow::DataType>&,
+                           MemoryPool* pool, std::shared_ptr<Array>*) {
+  return Status::NotImplemented("not implemented");
+}
+
 template <>
 Status ConvertToDecimal128<FLBAType>(const Array& array,
                                      const std::shared_ptr<::arrow::DataType>& type,
@@ -1263,10 +1272,10 @@ Status ConvertToDecimal128<FLBAType>(const Array& array,
 }
 
 template <>
-    static Status ConvertToDecimal128 <
-    ByteArrayType(const Array& array, const std::shared_ptr<::arrow::DataType>& type,
-                  std::shared_ptr<Array>* out) {
-  const auto& binary_array = static_cast<const ::arrow::BinaryArray&>(*array);
+Status ConvertToDecimal128<ByteArrayType>(const Array& array,
+                                          const std::shared_ptr<::arrow::DataType>& type,
+                                          MemoryPool* pool, std::shared_ptr<Array>* out) {
+  const auto& binary_array = static_cast<const ::arrow::BinaryArray&>(array);
   const int64_t length = binary_array.length();
 
   const auto& decimal_type = static_cast<const ::arrow::Decimal128Type&>(*type);
@@ -1311,9 +1320,10 @@ template <>
 /// 2. Allocating a buffer for the arrow::Decimal128Array
 /// 3. Converting the big-endian bytes in each BinaryArray entry to two integers
 ///    representing the high and low bits of each decimal value.
+template <typename ArrowType, typename ParquetType>
 struct TransferFunctor<
     ArrowType, ParquetType,
-    typename std::enable_if<std::is_same<ArrowType, ::arrow::Decimal128Type> &&
+    typename std::enable_if<std::is_same<ArrowType, ::arrow::Decimal128Type>::value &&
                             (std::is_same<ParquetType, ByteArrayType>::value ||
                              std::is_same<ParquetType, FLBAType>::value)>::type> {
   Status operator()(RecordReader* reader, MemoryPool* pool,
@@ -1408,7 +1418,7 @@ struct TransferFunctor<::arrow::Decimal128Type, Int64Type> {
 #define TRANSFER_DATA(ArrowType, ParquetType)                                \
   TransferFunctor<ArrowType, ParquetType> func;                              \
   RETURN_NOT_OK(func(record_reader_.get(), pool_, field_->type(), &result)); \
-  RETURN_NOT_OK(WrapIntoListArray<ParquetType>(out))
+  RETURN_NOT_OK(WrapIntoListArray<ParquetType>(&result))
 
 #define TRANSFER_CASE(ENUM, ArrowType, ParquetType) \
   case ::arrow::Type::ENUM: {                       \
@@ -1505,7 +1515,7 @@ Status PrimitiveImpl::NextBatch(int64_t records_to_read,
   DCHECK_NE(result.kind(), Datum::NONE);
 
   if (result.kind() == Datum::ARRAY) {
-    *out = std::make_shared<ChunkedArray>(result.array());
+    *out = std::make_shared<ChunkedArray>(result.make_array());
   } else if (result.kind() == Datum::CHUNKED_ARRAY) {
     *out = result.chunked_array();
   } else {
@@ -1542,7 +1552,7 @@ Status ColumnReader::NextBatch(int64_t records_to_read,
 }
 
 Status ColumnReader::NextBatch(int64_t records_to_read, std::shared_ptr<Array>* out) {
-  std::make_shared<ChunkedArray> chunked_out;
+  std::shared_ptr<ChunkedArray> chunked_out;
   RETURN_NOT_OK(impl_->NextBatch(records_to_read, &chunked_out));
   return GetSingleChunk(*chunked_out, out);
 }
@@ -1689,7 +1699,7 @@ Status ColumnChunkReader::Read(std::shared_ptr<::arrow::ChunkedArray>* out) {
 }
 
 Status ColumnChunkReader::Read(std::shared_ptr<::arrow::Array>* out) {
-  std::make_shared<ChunkedArray> chunked_out;
+  std::shared_ptr<ChunkedArray> chunked_out;
   RETURN_NOT_OK(impl_->ReadColumnChunk(column_index_, row_group_index_, &chunked_out));
   return GetSingleChunk(*chunked_out, out);
 }

cpp/src/parquet/arrow/reader.h

@@ -128,7 +128,8 @@ class PARQUET_EXPORT FileReader {
   // Read column as a whole into an Array.
   ::arrow::Status ReadColumn(int i, std::shared_ptr<::arrow::ChunkedArray>* out);
 
-  PARQUET_DEPRECATED("Use version with ChunkedArray output")
+  /// \note Deprecated since 0.12
+  ARROW_DEPRECATED("Use version with ChunkedArray output")
   ::arrow::Status ReadColumn(int i, std::shared_ptr<::arrow::Array>* out);
 
   // NOTE: Experimental API
@@ -145,7 +146,8 @@ class PARQUET_EXPORT FileReader {
   // i=0 will read the entire foo struct, i=1 the foo2 primitive column etc
   ::arrow::Status ReadSchemaField(int i, std::shared_ptr<::arrow::ChunkedArray>* out);
 
-  PARQUET_DEPRECATED("Use version with ChunkedArray output")
+  /// \note Deprecated since 0.12
+  ARROW_DEPRECATED("Use version with ChunkedArray output")
   ::arrow::Status ReadSchemaField(int i, std::shared_ptr<::arrow::Array>* out);
 
   /// \brief Return a RecordBatchReader of row groups selected from row_group_indices, the
@@ -237,7 +239,8 @@ class PARQUET_EXPORT ColumnChunkReader {
  public:
   ::arrow::Status Read(std::shared_ptr<::arrow::ChunkedArray>* out);
 
-  PARQUET_DEPRECATED("Use version with ChunkedArray output")
+  /// \note Deprecated since 0.12
+  ARROW_DEPRECATED("Use version with ChunkedArray output")
   ::arrow::Status Read(std::shared_ptr<::arrow::Array>* out);
 
   virtual ~ColumnChunkReader();
@@ -274,7 +277,8 @@ class PARQUET_EXPORT ColumnReader {
   ::arrow::Status NextBatch(int64_t batch_size,
                             std::shared_ptr<::arrow::ChunkedArray>* out);
 
-  PARQUET_DEPRECATED("Use version with ChunkedArray output")
+  /// \note Deprecated since 0.12
+  ARROW_DEPRECATED("Use version with ChunkedArray output")
   ::arrow::Status NextBatch(int64_t batch_size, std::shared_ptr<::arrow::Array>* out);
 
  private:

cpp/src/parquet/arrow/record_reader.cc

@@ -381,7 +381,7 @@ class RecordReader::RecordReaderImpl {
 
   virtual void DebugPrintState() = 0;
 
-  virtual std::vector<std::shared_ptr<Array>> GetBuilderChunks() = 0;
+  virtual std::vector<std::shared_ptr<::arrow::Array>> GetBuilderChunks() = 0;
 
  protected:
   virtual bool ReadNewPage() = 0;
@@ -440,12 +440,17 @@ class RecordReader::RecordReaderImpl {
 
 template <typename DType>
 struct RecordReaderTraits {
-  using BuilderType = void;
+  using BuilderType = ::arrow::ArrayBuilder;
 };
 
 template <>
 struct RecordReaderTraits<ByteArrayType> {
-  using BuilderType = ::arrow::ChunkedBinaryBuilder;
+  using BuilderType = ::arrow::internal::ChunkedBinaryBuilder;
+};
+
+template <>
+struct RecordReaderTraits<FLBAType> {
+  using BuilderType = ::arrow::FixedSizeBinaryBuilder;
 };
 
 template <typename DType>
@@ -455,18 +460,9 @@ class TypedRecordReader : public RecordReader::RecordReaderImpl {
 
   using BuilderType = typename RecordReaderTraits<DType>::BuilderType;
 
-  // Maximum of 16MB chunks
-  static constexpr int32_t kBinaryChunksize = 1 << 24;
-
   TypedRecordReader(const ColumnDescriptor* descr, ::arrow::MemoryPool* pool)
       : RecordReader::RecordReaderImpl(descr, pool), current_decoder_(nullptr) {
-    if (descr->physical_type() == Type::BYTE_ARRAY) {
-      builder_.reset(new ::arrow::internal::ChunkedBinaryBuilder(kBinaryChunksize, pool));
-    } else if (descr->physical_type() == Type::FIXED_LEN_BYTE_ARRAY) {
-      int byte_width = descr->type_length();
-      std::shared_ptr<::arrow::DataType> type = ::arrow::fixed_size_binary(byte_width);
-      builder_.reset(new ::arrow::FixedSizeBinaryBuilder(type, pool));
-    }
+    InitializeBuilder();
   }
 
   void ResetDecoders() override { decoders_.clear(); }
@@ -558,7 +554,7 @@ class TypedRecordReader : public RecordReader::RecordReaderImpl {
     std::cout << std::endl;
   }
 
-  std::vector<std::shared_ptr<Array>> GetBuilderChunks() override {
+  std::vector<std::shared_ptr<::arrow::Array>> GetBuilderChunks() override {
     throw ParquetException("GetChunks only implemented for binary types");
   }
 
@@ -577,6 +573,8 @@ class TypedRecordReader : public RecordReader::RecordReaderImpl {
   // Advance to the next data page
   bool ReadNewPage() override;
 
+  void InitializeBuilder() {}
+
   void ConfigureDictionary(const DictionaryPage* page);
 };
 
@@ -590,6 +588,22 @@ void TypedRecordReader<ByteArrayType>::DebugPrintState() {}
 template <>
 void TypedRecordReader<FLBAType>::DebugPrintState() {}
 
+template <>
+void TypedRecordReader<ByteArrayType>::InitializeBuilder() {
+  // Maximum of 16MB chunks
+  constexpr int32_t kBinaryChunksize = 1 << 24;
+  DCHECK_EQ(descr_->physical_type(), Type::BYTE_ARRAY);
+  builder_.reset(new ::arrow::internal::ChunkedBinaryBuilder(kBinaryChunksize, pool_));
+}
+
+template <>
+void TypedRecordReader<FLBAType>::InitializeBuilder() {
+  DCHECK_EQ(descr_->physical_type(), Type::FIXED_LEN_BYTE_ARRAY);
+  int byte_width = descr_->type_length();
+  std::shared_ptr<::arrow::DataType> type = ::arrow::fixed_size_binary(byte_width);
+  builder_.reset(new ::arrow::FixedSizeBinaryBuilder(type, pool_));
+}
+
 template <>
 ::arrow::ArrayVector TypedRecordReader<ByteArrayType>::GetBuilderChunks() {
   ::arrow::ArrayVector chunks;
@@ -599,7 +613,7 @@ template <>
 
 template <>
 ::arrow::ArrayVector TypedRecordReader<FLBAType>::GetBuilderChunks() {
-  std::shared_ptr<Array> chunk;
+  std::shared_ptr<::arrow::Array> chunk;
   PARQUET_THROW_NOT_OK(builder_->Finish(&chunk));
   return ::arrow::ArrayVector({chunk});
 }
@@ -888,6 +902,10 @@ void RecordReader::SetPageReader(std::unique_ptr<PageReader> reader) {
   impl_->SetPageReader(std::move(reader));
 }
 
+::arrow::ArrayVector RecordReader::GetBuilderChunks() {
+  return impl_->GetBuilderChunks();
+}
+
 void RecordReader::DebugPrintState() { impl_->DebugPrintState(); }
 
 }  // namespace internal