Split up builder.h, builder.cc into smaller headers, compilation units. add

failing test case for ARROW-3762. Add ChunkedBinaryBuilder, make BinaryBuilder
Append methods inline

cpp/src/arrow/CMakeLists.txt

@@ -55,9 +55,6 @@ set(ARROW_SRCS
   array.cc
   buffer.cc
   builder.cc
-  builder-adaptive.cc
-  builder-binary.cc
-  builder-dict.cc
   compare.cc
   memory_pool.cc
   pretty_print.cc
@@ -69,6 +66,14 @@ set(ARROW_SRCS
   type.cc
   visitor.cc
 
+  columnar/builder_adaptive.cc
+  columnar/builder_base.cc
+  columnar/builder_binary.cc
+  columnar/builder_decimal.cc
+  columnar/builder_dict.cc
+  columnar/builder_nested.cc
+  columnar/builder_primitive.cc
+
   csv/converter.cc
   csv/chunker.cc
   csv/column-builder.cc
@@ -297,6 +302,7 @@ ADD_ARROW_TEST(tensor-test)
 ADD_ARROW_BENCHMARK(builder-benchmark)
 ADD_ARROW_BENCHMARK(column-benchmark)
 
+add_subdirectory(columnar)
 add_subdirectory(csv)
 add_subdirectory(io)
 add_subdirectory(util)

cpp/src/arrow/allocator-test.cc

@@ -17,6 +17,7 @@
 
 #include <cstdint>
 #include <limits>
+#include <memory>
 #include <new>
 
 #include <gtest/gtest.h>

cpp/src/arrow/array-binary-test.cc

@@ -15,10 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include <algorithm>
 #include <cstdint>
 #include <cstring>
-#include <limits>
 #include <memory>
 #include <string>
 #include <vector>
@@ -28,10 +26,14 @@
 #include "arrow/array.h"
 #include "arrow/buffer.h"
 #include "arrow/builder.h"
+#include "arrow/memory_pool.h"
 #include "arrow/status.h"
 #include "arrow/test-common.h"
 #include "arrow/test-util.h"
 #include "arrow/type.h"
+#include "arrow/type_traits.h"
+#include "arrow/util/bit-util.h"
+#include "arrow/util/checked_cast.h"
 
 namespace arrow {
 
@@ -676,4 +678,104 @@ TEST_F(TestStringArray, TestSliceEquality) { CheckSliceEquality<BinaryType>(); }
 
 TEST_F(TestBinaryArray, LengthZeroCtor) { BinaryArray array(0, nullptr, nullptr); }
 
+// ----------------------------------------------------------------------
+// ChunkedBinaryBuilder tests
+
+class TestChunkedBinaryBuilder : public ::testing::Test {
+ public:
+  void SetUp() {}
+
+  void Init(int32_t chunksize) {
+    builder_.reset(new internal::ChunkedBinaryBuilder(chunksize));
+  }
+
+ protected:
+  std::unique_ptr<internal::ChunkedBinaryBuilder> builder_;
+};
+
+TEST_F(TestChunkedBinaryBuilder, BasicOperation) {
+  const int32_t chunksize = 1000;
+  Init(chunksize);
+
+  const int elem_size = 10;
+  uint8_t buf[elem_size];
+
+  BinaryBuilder unchunked_builder;
+
+  const int iterations = 1000;
+  for (int i = 0; i < iterations; ++i) {
+    random_bytes(elem_size, i, buf);
+
+    ASSERT_OK(unchunked_builder.Append(buf, elem_size));
+    ASSERT_OK(builder_->Append(buf, elem_size));
+  }
+
+  std::shared_ptr<Array> unchunked;
+  ASSERT_OK(unchunked_builder.Finish(&unchunked));
+
+  ArrayVector chunks;
+  ASSERT_OK(builder_->Finish(&chunks));
+
+  // This assumes that everything is evenly divisible
+  ArrayVector expected_chunks;
+  const int elems_per_chunk = chunksize / elem_size;
+  for (int i = 0; i < iterations / elems_per_chunk; ++i) {
+    expected_chunks.emplace_back(unchunked->Slice(i * elems_per_chunk, elems_per_chunk));
+  }
+
+  ASSERT_EQ(expected_chunks.size(), chunks.size());
+  for (size_t i = 0; i < chunks.size(); ++i) {
+    AssertArraysEqual(*expected_chunks[i], *chunks[i]);
+  }
+}
+
+TEST_F(TestChunkedBinaryBuilder, NoData) {
+  Init(1000);
+
+  ArrayVector chunks;
+  ASSERT_OK(builder_->Finish(&chunks));
+
+  ASSERT_EQ(1, chunks.size());
+  ASSERT_EQ(0, chunks[0]->length());
+}
+
+TEST_F(TestChunkedBinaryBuilder, LargeElements) {
+  Init(100);
+
+  const int bufsize = 101;
+  uint8_t buf[bufsize];
+
+  const int iterations = 100;
+  for (int i = 0; i < iterations; ++i) {
+    random_bytes(bufsize, i, buf);
+    ASSERT_OK(builder_->Append(buf, bufsize));
+  }
+
+  ArrayVector chunks;
+  ASSERT_OK(builder_->Finish(&chunks));
+  ASSERT_EQ(iterations, static_cast<int>(chunks.size()));
+}
+
+TEST(TestChunkedStringBuilder, BasicOperation) {
+  const int chunksize = 100;
+  internal::ChunkedStringBuilder builder(chunksize);
+
+  std::string value = "0123456789";
+
+  const int iterations = 100;
+  for (int i = 0; i < iterations; ++i) {
+    ASSERT_OK(builder.Append(value));
+  }
+
+  ArrayVector chunks;
+  ASSERT_OK(builder.Finish(&chunks));
+
+  ASSERT_EQ(10, chunks.size());
+
+  // Type is correct
+  for (auto chunk : chunks) {
+    ASSERT_TRUE(chunk->type()->Equals(*::arrow::utf8()));
+  }
+}
+
 }  // namespace arrow

cpp/src/arrow/array-dict-test.cc

@@ -15,23 +15,23 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include <algorithm>
+#include <array>
 #include <cstdint>
-#include <cstring>
-#include <limits>
 #include <memory>
+#include <ostream>
 #include <string>
 #include <vector>
 
 #include <gtest/gtest.h>
 
 #include "arrow/array.h"
-#include "arrow/buffer.h"
 #include "arrow/builder.h"
+#include "arrow/memory_pool.h"
 #include "arrow/status.h"
 #include "arrow/test-common.h"
 #include "arrow/test-util.h"
 #include "arrow/type.h"
+#include "arrow/util/decimal.h"
 
 namespace arrow {
 

cpp/src/arrow/array-list-test.cc

@@ -15,10 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include <algorithm>
 #include <cstdint>
 #include <cstring>
-#include <limits>
 #include <memory>
 #include <string>
 #include <vector>
@@ -32,6 +30,8 @@
 #include "arrow/test-common.h"
 #include "arrow/test-util.h"
 #include "arrow/type.h"
+#include "arrow/util/bit-util.h"
+#include "arrow/util/checked_cast.h"
 
 namespace arrow {
 

cpp/src/arrow/array-struct-test.cc

@@ -15,10 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include <algorithm>
 #include <cstdint>
 #include <cstring>
-#include <limits>
 #include <memory>
 #include <string>
 #include <vector>
@@ -26,12 +24,12 @@
 #include <gtest/gtest.h>
 
 #include "arrow/array.h"
-#include "arrow/buffer.h"
 #include "arrow/builder.h"
 #include "arrow/status.h"
 #include "arrow/test-common.h"
 #include "arrow/test-util.h"
 #include "arrow/type.h"
+#include "arrow/util/checked_cast.h"
 
 namespace arrow {
 

cpp/src/arrow/array-test.cc

@@ -23,7 +23,6 @@
 #include <limits>
 #include <memory>
 #include <numeric>
-#include <ostream>
 #include <string>
 #include <type_traits>
 #include <vector>
@@ -40,7 +39,6 @@
 #include "arrow/test-common.h"
 #include "arrow/test-util.h"
 #include "arrow/type.h"
-#include "arrow/type_traits.h"
 #include "arrow/util/bit-util.h"
 #include "arrow/util/checked_cast.h"
 #include "arrow/util/decimal.h"

cpp/src/arrow/array.cc

@@ -18,6 +18,7 @@
 #include "arrow/array.h"
 
 #include <algorithm>
+#include <cstddef>
 #include <cstdint>
 #include <limits>
 #include <sstream>

cpp/src/arrow/array.h

@@ -18,7 +18,6 @@
 #ifndef ARROW_ARRAY_H
 #define ARROW_ARRAY_H
 
-#include <cstddef>
 #include <cstdint>
 #include <iosfwd>
 #include <memory>

cpp/src/arrow/builder-benchmark.cc

@@ -177,6 +177,25 @@ static void BM_BuildBinaryArray(benchmark::State& state) {  // NOLINT non-const
   state.SetBytesProcessed(state.iterations() * iterations * value.size());
 }
 
+static void BM_BuildChunkedBinaryArray(
+    benchmark::State& state) {  // NOLINT non-const reference
+  const int64_t iterations = 1 << 20;
+
+  std::string value = "1234567890";
+  while (state.KeepRunning()) {
+    // 1MB chunks
+    const int32_t chunksize = 1 << 20;
+    internal::ChunkedBinaryBuilder builder(chunksize);
+    for (int64_t i = 0; i < iterations; i++) {
+      ABORT_NOT_OK(builder.Append(reinterpret_cast<const uint8_t*>(value.data()),
+                                  static_cast<int32_t>(value.size())));
+    }
+    ArrayVector out;
+    ABORT_NOT_OK(builder.Finish(&out));
+  }
+  state.SetBytesProcessed(state.iterations() * iterations * value.size());
+}
+
 static void BM_BuildFixedSizeBinaryArray(
     benchmark::State& state) {  // NOLINT non-const reference
   const int64_t iterations = 1 << 20;
@@ -372,6 +391,9 @@ BENCHMARK(BM_BuildAdaptiveUIntNoNullsScalarAppend)
     ->Unit(benchmark::kMicrosecond);
 
 BENCHMARK(BM_BuildBinaryArray)->Repetitions(kRepetitions)->Unit(benchmark::kMicrosecond);
+BENCHMARK(BM_BuildChunkedBinaryArray)
+    ->Repetitions(kRepetitions)
+    ->Unit(benchmark::kMicrosecond);
 BENCHMARK(BM_BuildFixedSizeBinaryArray)
     ->Repetitions(kRepetitions)
     ->Unit(benchmark::kMicrosecond);

cpp/src/arrow/builder.cc

@@ -15,513 +15,20 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include <algorithm>
-#include <cstddef>
-#include <cstdint>
-#include <cstring>
+#include "arrow/builder.h"
+
 #include <sstream>
+#include <string>
 #include <utility>
 #include <vector>
 
-#include "arrow/array.h"
-#include "arrow/buffer.h"
-#include "arrow/builder.h"
 #include "arrow/status.h"
 #include "arrow/type.h"
-#include "arrow/type_traits.h"
-#include "arrow/util/bit-util.h"
 #include "arrow/util/checked_cast.h"
-#include "arrow/util/int-util.h"
-#include "arrow/util/logging.h"
 
 namespace arrow {
 
-using internal::checked_cast;
-
-Status ArrayBuilder::TrimBuffer(const int64_t bytes_filled, ResizableBuffer* buffer) {
-  if (buffer) {
-    if (bytes_filled < buffer->size()) {
-      // Trim buffer
-      RETURN_NOT_OK(buffer->Resize(bytes_filled));
-    }
-    // zero the padding
-    buffer->ZeroPadding();
-  } else {
-    // Null buffers are allowed in place of 0-byte buffers
-    DCHECK_EQ(bytes_filled, 0);
-  }
-  return Status::OK();
-}
-
-Status ArrayBuilder::AppendToBitmap(bool is_valid) {
-  if (length_ == capacity_) {
-    // If the capacity was not already a multiple of 2, do so here
-    // TODO(emkornfield) doubling isn't great default allocation practice
-    // see https://github.com/facebook/folly/blob/master/folly/docs/FBVector.md
-    // fo discussion
-    RETURN_NOT_OK(Resize(BitUtil::NextPower2(capacity_ + 1)));
-  }
-  UnsafeAppendToBitmap(is_valid);
-  return Status::OK();
-}
-
-Status ArrayBuilder::AppendToBitmap(const uint8_t* valid_bytes, int64_t length) {
-  RETURN_NOT_OK(Reserve(length));
-
-  UnsafeAppendToBitmap(valid_bytes, length);
-  return Status::OK();
-}
-
-Status ArrayBuilder::Resize(int64_t capacity) {
-  // Target size of validity (null) bitmap data
-  const int64_t new_bitmap_size = BitUtil::BytesForBits(capacity);
-  RETURN_NOT_OK(CheckCapacity(capacity, capacity_));
-
-  if (capacity_ == 0) {
-    RETURN_NOT_OK(AllocateResizableBuffer(pool_, new_bitmap_size, &null_bitmap_));
-    null_bitmap_data_ = null_bitmap_->mutable_data();
-
-    // Padding is zeroed by AllocateResizableBuffer
-    memset(null_bitmap_data_, 0, static_cast<size_t>(new_bitmap_size));
-  } else {
-    const int64_t old_bitmap_capacity = null_bitmap_->capacity();
-    RETURN_NOT_OK(null_bitmap_->Resize(new_bitmap_size));
-
-    const int64_t new_bitmap_capacity = null_bitmap_->capacity();
-    null_bitmap_data_ = null_bitmap_->mutable_data();
-
-    // Zero the region between the original capacity and the new capacity,
-    // including padding, which has not been zeroed, unlike
-    // AllocateResizableBuffer
-    if (old_bitmap_capacity < new_bitmap_capacity) {
-      memset(null_bitmap_data_ + old_bitmap_capacity, 0,
-             static_cast<size_t>(new_bitmap_capacity - old_bitmap_capacity));
-    }
-  }
-  capacity_ = capacity;
-  return Status::OK();
-}
-
-Status ArrayBuilder::Advance(int64_t elements) {
-  if (length_ + elements > capacity_) {
-    return Status::Invalid("Builder must be expanded");
-  }
-  length_ += elements;
-  return Status::OK();
-}
-
-Status ArrayBuilder::Finish(std::shared_ptr<Array>* out) {
-  std::shared_ptr<ArrayData> internal_data;
-  RETURN_NOT_OK(FinishInternal(&internal_data));
-  *out = MakeArray(internal_data);
-  return Status::OK();
-}
-
-Status ArrayBuilder::Reserve(int64_t additional_elements) {
-  if (length_ + additional_elements > capacity_) {
-    // TODO(emkornfield) power of 2 growth is potentially suboptimal
-    int64_t new_size = BitUtil::NextPower2(length_ + additional_elements);
-    return Resize(new_size);
-  }
-  return Status::OK();
-}
-
-void ArrayBuilder::Reset() {
-  capacity_ = length_ = null_count_ = 0;
-  null_bitmap_ = nullptr;
-}
-
-Status ArrayBuilder::SetNotNull(int64_t length) {
-  RETURN_NOT_OK(Reserve(length));
-  UnsafeSetNotNull(length);
-  return Status::OK();
-}
-
-void ArrayBuilder::UnsafeAppendToBitmap(const uint8_t* valid_bytes, int64_t length) {
-  if (valid_bytes == nullptr) {
-    UnsafeSetNotNull(length);
-    return;
-  }
-  UnsafeAppendToBitmap(valid_bytes, valid_bytes + length);
-}
-
-void ArrayBuilder::UnsafeAppendToBitmap(const std::vector<bool>& is_valid) {
-  UnsafeAppendToBitmap(is_valid.begin(), is_valid.end());
-}
-
-void ArrayBuilder::UnsafeSetNotNull(int64_t length) {
-  const int64_t new_length = length + length_;
-
-  // Fill up the bytes until we have a byte alignment
-  int64_t pad_to_byte = std::min<int64_t>(8 - (length_ % 8), length);
-
-  if (pad_to_byte == 8) {
-    pad_to_byte = 0;
-  }
-  for (int64_t i = length_; i < length_ + pad_to_byte; ++i) {
-    BitUtil::SetBit(null_bitmap_data_, i);
-  }
-
-  // Fast bitsetting
-  int64_t fast_length = (length - pad_to_byte) / 8;
-  memset(null_bitmap_data_ + ((length_ + pad_to_byte) / 8), 0xFF,
-         static_cast<size_t>(fast_length));
-
-  // Trailing bits
-  for (int64_t i = length_ + pad_to_byte + (fast_length * 8); i < new_length; ++i) {
-    BitUtil::SetBit(null_bitmap_data_, i);
-  }
-
-  length_ = new_length;
-}
-
-// ----------------------------------------------------------------------
-// Null builder
-
-Status NullBuilder::FinishInternal(std::shared_ptr<ArrayData>* out) {
-  *out = ArrayData::Make(null(), length_, {nullptr}, length_);
-  length_ = null_count_ = 0;
-  return Status::OK();
-}
-
-// ----------------------------------------------------------------------
-
-template <typename T>
-void PrimitiveBuilder<T>::Reset() {
-  data_.reset();
-  raw_data_ = nullptr;
-}
-
-template <typename T>
-Status PrimitiveBuilder<T>::Resize(int64_t capacity) {
-  RETURN_NOT_OK(CheckCapacity(capacity, capacity_));
-  capacity = std::max(capacity, kMinBuilderCapacity);
-
-  int64_t nbytes = TypeTraits<T>::bytes_required(capacity);
-  if (capacity_ == 0) {
-    RETURN_NOT_OK(AllocateResizableBuffer(pool_, nbytes, &data_));
-  } else {
-    RETURN_NOT_OK(data_->Resize(nbytes));
-  }
-
-  raw_data_ = reinterpret_cast<value_type*>(data_->mutable_data());
-  return ArrayBuilder::Resize(capacity);
-}
-
-template <typename T>
-Status PrimitiveBuilder<T>::AppendValues(const value_type* values, int64_t length,
-                                         const uint8_t* valid_bytes) {
-  RETURN_NOT_OK(Reserve(length));
-
-  if (length > 0) {
-    std::memcpy(raw_data_ + length_, values,
-                static_cast<std::size_t>(TypeTraits<T>::bytes_required(length)));
-  }
-
-  // length_ is update by these
-  ArrayBuilder::UnsafeAppendToBitmap(valid_bytes, length);
-  return Status::OK();
-}
-
-template <typename T>
-Status PrimitiveBuilder<T>::AppendValues(const value_type* values, int64_t length,
-                                         const std::vector<bool>& is_valid) {
-  RETURN_NOT_OK(Reserve(length));
-  DCHECK_EQ(length, static_cast<int64_t>(is_valid.size()));
-
-  if (length > 0) {
-    std::memcpy(raw_data_ + length_, values,
-                static_cast<std::size_t>(TypeTraits<T>::bytes_required(length)));
-  }
-
-  // length_ is update by these
-  ArrayBuilder::UnsafeAppendToBitmap(is_valid);
-  return Status::OK();
-}
-
-template <typename T>
-Status PrimitiveBuilder<T>::AppendValues(const std::vector<value_type>& values,
-                                         const std::vector<bool>& is_valid) {
-  return AppendValues(values.data(), static_cast<int64_t>(values.size()), is_valid);
-}
-
-template <typename T>
-Status PrimitiveBuilder<T>::AppendValues(const std::vector<value_type>& values) {
-  return AppendValues(values.data(), static_cast<int64_t>(values.size()));
-}
-
-template <typename T>
-Status PrimitiveBuilder<T>::FinishInternal(std::shared_ptr<ArrayData>* out) {
-  RETURN_NOT_OK(TrimBuffer(BitUtil::BytesForBits(length_), null_bitmap_.get()));
-  RETURN_NOT_OK(TrimBuffer(TypeTraits<T>::bytes_required(length_), data_.get()));
-
-  *out = ArrayData::Make(type_, length_, {null_bitmap_, data_}, null_count_);
-
-  data_ = null_bitmap_ = nullptr;
-  capacity_ = length_ = null_count_ = 0;
-
-  return Status::OK();
-}
-
-template class PrimitiveBuilder<UInt8Type>;
-template class PrimitiveBuilder<UInt16Type>;
-template class PrimitiveBuilder<UInt32Type>;
-template class PrimitiveBuilder<UInt64Type>;
-template class PrimitiveBuilder<Int8Type>;
-template class PrimitiveBuilder<Int16Type>;
-template class PrimitiveBuilder<Int32Type>;
-template class PrimitiveBuilder<Int64Type>;
-template class PrimitiveBuilder<Date32Type>;
-template class PrimitiveBuilder<Date64Type>;
-template class PrimitiveBuilder<Time32Type>;
-template class PrimitiveBuilder<Time64Type>;
-template class PrimitiveBuilder<TimestampType>;
-template class PrimitiveBuilder<HalfFloatType>;
-template class PrimitiveBuilder<FloatType>;
-template class PrimitiveBuilder<DoubleType>;
-
-BooleanBuilder::BooleanBuilder(MemoryPool* pool)
-    : ArrayBuilder(boolean(), pool), data_(nullptr), raw_data_(nullptr) {}
-
-BooleanBuilder::BooleanBuilder(const std::shared_ptr<DataType>& type, MemoryPool* pool)
-    : BooleanBuilder(pool) {
-  DCHECK_EQ(Type::BOOL, type->id());
-}
-
-void BooleanBuilder::Reset() {
-  ArrayBuilder::Reset();
-  data_.reset();
-  raw_data_ = nullptr;
-}
-
-Status BooleanBuilder::Resize(int64_t capacity) {
-  RETURN_NOT_OK(CheckCapacity(capacity, capacity_));
-  capacity = std::max(capacity, kMinBuilderCapacity);
-
-  const int64_t new_bitmap_size = BitUtil::BytesForBits(capacity);
-  if (capacity_ == 0) {
-    RETURN_NOT_OK(AllocateResizableBuffer(pool_, new_bitmap_size, &data_));
-    raw_data_ = reinterpret_cast<uint8_t*>(data_->mutable_data());
-
-    // We zero the memory for booleans to keep things simple; for some reason if
-    // we do not, even though we may write every bit (through in-place | or &),
-    // valgrind will still show a warning. If we do not zero the bytes here, we
-    // will have to be careful to zero them in AppendNull and AppendNulls. Also,
-    // zeroing the bits results in deterministic bits when each byte may have a
-    // mix of nulls and not nulls.
-    //
-    // We only zero up to new_bitmap_size because the padding was zeroed by
-    // AllocateResizableBuffer
-    memset(raw_data_, 0, static_cast<size_t>(new_bitmap_size));
-  } else {
-    const int64_t old_bitmap_capacity = data_->capacity();
-    RETURN_NOT_OK(data_->Resize(new_bitmap_size));
-    const int64_t new_bitmap_capacity = data_->capacity();
-    raw_data_ = reinterpret_cast<uint8_t*>(data_->mutable_data());
-
-    // See comment above about why we zero memory for booleans
-    memset(raw_data_ + old_bitmap_capacity, 0,
-           static_cast<size_t>(new_bitmap_capacity - old_bitmap_capacity));
-  }
-
-  return ArrayBuilder::Resize(capacity);
-}
-
-Status BooleanBuilder::FinishInternal(std::shared_ptr<ArrayData>* out) {
-  int64_t bit_offset = length_ % 8;
-  if (bit_offset > 0) {
-    // Adjust last byte
-    data_->mutable_data()[length_ / 8] &= BitUtil::kPrecedingBitmask[bit_offset];
-  }
-
-  RETURN_NOT_OK(TrimBuffer(BitUtil::BytesForBits(length_), null_bitmap_.get()));
-  RETURN_NOT_OK(TrimBuffer(BitUtil::BytesForBits(length_), data_.get()));
-
-  *out = ArrayData::Make(boolean(), length_, {null_bitmap_, data_}, null_count_);
-
-  data_ = null_bitmap_ = nullptr;
-  capacity_ = length_ = null_count_ = 0;
-  return Status::OK();
-}
-
-Status BooleanBuilder::AppendValues(const uint8_t* values, int64_t length,
-                                    const uint8_t* valid_bytes) {
-  RETURN_NOT_OK(Reserve(length));
-
-  int64_t i = 0;
-  internal::GenerateBitsUnrolled(raw_data_, length_, length,
-                                 [values, &i]() -> bool { return values[i++] != 0; });
-
-  // this updates length_
-  ArrayBuilder::UnsafeAppendToBitmap(valid_bytes, length);
-  return Status::OK();
-}
-
-Status BooleanBuilder::AppendValues(const uint8_t* values, int64_t length,
-                                    const std::vector<bool>& is_valid) {
-  RETURN_NOT_OK(Reserve(length));
-  DCHECK_EQ(length, static_cast<int64_t>(is_valid.size()));
-
-  int64_t i = 0;
-  internal::GenerateBitsUnrolled(raw_data_, length_, length,
-                                 [values, &i]() -> bool { return values[i++]; });
-
-  // this updates length_
-  ArrayBuilder::UnsafeAppendToBitmap(is_valid);
-  return Status::OK();
-}
-
-Status BooleanBuilder::AppendValues(const std::vector<uint8_t>& values,
-                                    const std::vector<bool>& is_valid) {
-  return AppendValues(values.data(), static_cast<int64_t>(values.size()), is_valid);
-}
-
-Status BooleanBuilder::AppendValues(const std::vector<uint8_t>& values) {
-  return AppendValues(values.data(), static_cast<int64_t>(values.size()));
-}
-
-Status BooleanBuilder::AppendValues(const std::vector<bool>& values,
-                                    const std::vector<bool>& is_valid) {
-  const int64_t length = static_cast<int64_t>(values.size());
-  RETURN_NOT_OK(Reserve(length));
-  DCHECK_EQ(length, static_cast<int64_t>(is_valid.size()));
-
-  int64_t i = 0;
-  internal::GenerateBitsUnrolled(raw_data_, length_, length,
-                                 [&values, &i]() -> bool { return values[i++]; });
-
-  // this updates length_
-  ArrayBuilder::UnsafeAppendToBitmap(is_valid);
-  return Status::OK();
-}
-
-Status BooleanBuilder::AppendValues(const std::vector<bool>& values) {
-  const int64_t length = static_cast<int64_t>(values.size());
-  RETURN_NOT_OK(Reserve(length));
-
-  int64_t i = 0;
-  internal::GenerateBitsUnrolled(raw_data_, length_, length,
-                                 [&values, &i]() -> bool { return values[i++]; });
-
-  // this updates length_
-  ArrayBuilder::UnsafeSetNotNull(length);
-  return Status::OK();
-}
-
-// ----------------------------------------------------------------------
-// ListBuilder
-
-ListBuilder::ListBuilder(MemoryPool* pool,
-                         std::shared_ptr<ArrayBuilder> const& value_builder,
-                         const std::shared_ptr<DataType>& type)
-    : ArrayBuilder(type ? type
-                        : std::static_pointer_cast<DataType>(
-                              std::make_shared<ListType>(value_builder->type())),
-                   pool),
-      offsets_builder_(pool),
-      value_builder_(value_builder) {}
-
-Status ListBuilder::AppendValues(const int32_t* offsets, int64_t length,
-                                 const uint8_t* valid_bytes) {
-  RETURN_NOT_OK(Reserve(length));
-  UnsafeAppendToBitmap(valid_bytes, length);
-  offsets_builder_.UnsafeAppend(offsets, length);
-  return Status::OK();
-}
-
-Status ListBuilder::AppendNextOffset() {
-  int64_t num_values = value_builder_->length();
-  if (ARROW_PREDICT_FALSE(num_values > kListMaximumElements)) {
-    std::stringstream ss;
-    ss << "ListArray cannot contain more then INT32_MAX - 1 child elements,"
-       << " have " << num_values;
-    return Status::CapacityError(ss.str());
-  }
-  return offsets_builder_.Append(static_cast<int32_t>(num_values));
-}
-
-Status ListBuilder::Append(bool is_valid) {
-  RETURN_NOT_OK(Reserve(1));
-  UnsafeAppendToBitmap(is_valid);
-  return AppendNextOffset();
-}
-
-Status ListBuilder::Resize(int64_t capacity) {
-  DCHECK_LE(capacity, kListMaximumElements);
-  RETURN_NOT_OK(CheckCapacity(capacity, capacity_));
-
-  // one more then requested for offsets
-  RETURN_NOT_OK(offsets_builder_.Resize((capacity + 1) * sizeof(int32_t)));
-  return ArrayBuilder::Resize(capacity);
-}
-
-Status ListBuilder::FinishInternal(std::shared_ptr<ArrayData>* out) {
-  RETURN_NOT_OK(AppendNextOffset());
-
-  // Offset padding zeroed by BufferBuilder
-  std::shared_ptr<Buffer> offsets;
-  RETURN_NOT_OK(offsets_builder_.Finish(&offsets));
-
-  std::shared_ptr<ArrayData> items;
-  if (values_) {
-    items = values_->data();
-  } else {
-    if (value_builder_->length() == 0) {
-      // Try to make sure we get a non-null values buffer (ARROW-2744)
-      RETURN_NOT_OK(value_builder_->Resize(0));
-    }
-    RETURN_NOT_OK(value_builder_->FinishInternal(&items));
-  }
-
-  *out = ArrayData::Make(type_, length_, {null_bitmap_, offsets}, null_count_);
-  (*out)->child_data.emplace_back(std::move(items));
-  Reset();
-  return Status::OK();
-}
-
-void ListBuilder::Reset() {
-  ArrayBuilder::Reset();
-  values_.reset();
-  offsets_builder_.Reset();
-  value_builder_->Reset();
-}
-
-ArrayBuilder* ListBuilder::value_builder() const {
-  DCHECK(!values_) << "Using value builder is pointless when values_ is set";
-  return value_builder_.get();
-}
-
-// ----------------------------------------------------------------------
-// Struct
-
-StructBuilder::StructBuilder(const std::shared_ptr<DataType>& type, MemoryPool* pool,
-                             std::vector<std::shared_ptr<ArrayBuilder>>&& field_builders)
-    : ArrayBuilder(type, pool), field_builders_(std::move(field_builders)) {}
-
-void StructBuilder::Reset() {
-  ArrayBuilder::Reset();
-  for (const auto& field_builder : field_builders_) {
-    field_builder->Reset();
-  }
-}
-
-Status StructBuilder::FinishInternal(std::shared_ptr<ArrayData>* out) {
-  RETURN_NOT_OK(TrimBuffer(BitUtil::BytesForBits(length_), null_bitmap_.get()));
-  *out = ArrayData::Make(type_, length_, {null_bitmap_}, null_count_);
-
-  (*out)->child_data.resize(field_builders_.size());
-  for (size_t i = 0; i < field_builders_.size(); ++i) {
-    if (length_ == 0) {
-      // Try to make sure the child buffers are initialized
-      RETURN_NOT_OK(field_builders_[i]->Resize(0));
-    }
-    RETURN_NOT_OK(field_builders_[i]->FinishInternal(&(*out)->child_data[i]));
-  }
-
-  null_bitmap_ = nullptr;
-  capacity_ = length_ = null_count_ = 0;
-  return Status::OK();
-}
+class MemoryPool;
 
 // ----------------------------------------------------------------------
 // Helper functions
@@ -566,7 +73,7 @@ Status MakeBuilder(MemoryPool* pool, const std::shared_ptr<DataType>& type,
     case Type::LIST: {
       std::unique_ptr<ArrayBuilder> value_builder;
       std::shared_ptr<DataType> value_type =
-          checked_cast<const ListType&>(*type).value_type();
+          internal::checked_cast<const ListType&>(*type).value_type();
       RETURN_NOT_OK(MakeBuilder(pool, value_type, &value_builder));
       out->reset(new ListBuilder(pool, std::move(value_builder)));
       return Status::OK();

cpp/src/arrow/columnar/builder_adaptive.cc

@@ -15,13 +15,15 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include "arrow/columnar/builder_adaptive.h"
+
+#include <algorithm>
 #include <cstddef>
 #include <cstdint>
 #include <utility>
 
 #include "arrow/array.h"
 #include "arrow/buffer.h"
-#include "arrow/builder.h"
 #include "arrow/status.h"
 #include "arrow/type.h"
 #include "arrow/type_traits.h"

cpp/src/arrow/columnar/builder_adaptive.h

@@ -0,0 +1,174 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#pragma once
+
+#include <memory>
+
+#include "arrow/columnar/builder_base.h"
+
+namespace arrow {
+
+namespace internal {
+
+class ARROW_EXPORT AdaptiveIntBuilderBase : public ArrayBuilder {
+ public:
+  explicit AdaptiveIntBuilderBase(MemoryPool* pool);
+
+  /// Write nulls as uint8_t* (0 value indicates null) into pre-allocated memory
+  Status AppendNulls(const uint8_t* valid_bytes, int64_t length) {
+    ARROW_RETURN_NOT_OK(CommitPendingData());
+    ARROW_RETURN_NOT_OK(Reserve(length));
+    memset(data_->mutable_data() + length_ * int_size_, 0, int_size_ * length);
+    UnsafeAppendToBitmap(valid_bytes, length);
+    return Status::OK();
+  }
+
+  Status AppendNull() {
+    pending_data_[pending_pos_] = 0;
+    pending_valid_[pending_pos_] = 0;
+    pending_has_nulls_ = true;
+    ++pending_pos_;
+
+    if (ARROW_PREDICT_FALSE(pending_pos_ >= pending_size_)) {
+      return CommitPendingData();
+    }
+    return Status::OK();
+  }
+
+  void Reset() override;
+  Status Resize(int64_t capacity) override;
+
+ protected:
+  virtual Status CommitPendingData() = 0;
+
+  std::shared_ptr<ResizableBuffer> data_;
+  uint8_t* raw_data_;
+  uint8_t int_size_;
+
+  static constexpr int32_t pending_size_ = 1024;
+  uint8_t pending_valid_[pending_size_];
+  uint64_t pending_data_[pending_size_];
+  int32_t pending_pos_;
+  bool pending_has_nulls_;
+};
+
+}  // namespace internal
+
+class ARROW_EXPORT AdaptiveUIntBuilder : public internal::AdaptiveIntBuilderBase {
+ public:
+  explicit AdaptiveUIntBuilder(MemoryPool* pool ARROW_MEMORY_POOL_DEFAULT);
+
+  using ArrayBuilder::Advance;
+  using internal::AdaptiveIntBuilderBase::Reset;
+
+  /// Scalar append
+  Status Append(const uint64_t val) {
+    pending_data_[pending_pos_] = val;
+    pending_valid_[pending_pos_] = 1;
+    ++pending_pos_;
+
+    if (ARROW_PREDICT_FALSE(pending_pos_ >= pending_size_)) {
+      return CommitPendingData();
+    }
+    return Status::OK();
+  }
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values a contiguous C array of values
+  /// \param[in] length the number of values to append
+  /// \param[in] valid_bytes an optional sequence of bytes where non-zero
+  /// indicates a valid (non-null) value
+  /// \return Status
+  Status AppendValues(const uint64_t* values, int64_t length,
+                      const uint8_t* valid_bytes = NULLPTR);
+
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+
+ protected:
+  Status CommitPendingData() override;
+  Status ExpandIntSize(uint8_t new_int_size);
+
+  Status AppendValuesInternal(const uint64_t* values, int64_t length,
+                              const uint8_t* valid_bytes);
+
+  template <typename new_type, typename old_type>
+  typename std::enable_if<sizeof(old_type) >= sizeof(new_type), Status>::type
+  ExpandIntSizeInternal();
+#define __LESS(a, b) (a) < (b)
+  template <typename new_type, typename old_type>
+  typename std::enable_if<__LESS(sizeof(old_type), sizeof(new_type)), Status>::type
+  ExpandIntSizeInternal();
+#undef __LESS
+
+  template <typename new_type>
+  Status ExpandIntSizeN();
+};
+
+class ARROW_EXPORT AdaptiveIntBuilder : public internal::AdaptiveIntBuilderBase {
+ public:
+  explicit AdaptiveIntBuilder(MemoryPool* pool ARROW_MEMORY_POOL_DEFAULT);
+
+  using ArrayBuilder::Advance;
+  using internal::AdaptiveIntBuilderBase::Reset;
+
+  /// Scalar append
+  Status Append(const int64_t val) {
+    auto v = static_cast<uint64_t>(val);
+
+    pending_data_[pending_pos_] = v;
+    pending_valid_[pending_pos_] = 1;
+    ++pending_pos_;
+
+    if (ARROW_PREDICT_FALSE(pending_pos_ >= pending_size_)) {
+      return CommitPendingData();
+    }
+    return Status::OK();
+  }
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values a contiguous C array of values
+  /// \param[in] length the number of values to append
+  /// \param[in] valid_bytes an optional sequence of bytes where non-zero
+  /// indicates a valid (non-null) value
+  /// \return Status
+  Status AppendValues(const int64_t* values, int64_t length,
+                      const uint8_t* valid_bytes = NULLPTR);
+
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+
+ protected:
+  Status CommitPendingData() override;
+  Status ExpandIntSize(uint8_t new_int_size);
+
+  Status AppendValuesInternal(const int64_t* values, int64_t length,
+                              const uint8_t* valid_bytes);
+
+  template <typename new_type, typename old_type>
+  typename std::enable_if<sizeof(old_type) >= sizeof(new_type), Status>::type
+  ExpandIntSizeInternal();
+#define __LESS(a, b) (a) < (b)
+  template <typename new_type, typename old_type>
+  typename std::enable_if<__LESS(sizeof(old_type), sizeof(new_type)), Status>::type
+  ExpandIntSizeInternal();
+#undef __LESS
+
+  template <typename new_type>
+  Status ExpandIntSizeN();
+};
+
+}  // namespace arrow

cpp/src/arrow/columnar/builder_base.cc

@@ -0,0 +1,176 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#include "arrow/columnar/builder_base.h"
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <sstream>
+#include <utility>
+#include <vector>
+
+#include "arrow/array.h"
+#include "arrow/buffer.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/type_traits.h"
+#include "arrow/util/bit-util.h"
+#include "arrow/util/int-util.h"
+#include "arrow/util/logging.h"
+
+namespace arrow {
+
+Status ArrayBuilder::TrimBuffer(const int64_t bytes_filled, ResizableBuffer* buffer) {
+  if (buffer) {
+    if (bytes_filled < buffer->size()) {
+      // Trim buffer
+      RETURN_NOT_OK(buffer->Resize(bytes_filled));
+    }
+    // zero the padding
+    buffer->ZeroPadding();
+  } else {
+    // Null buffers are allowed in place of 0-byte buffers
+    DCHECK_EQ(bytes_filled, 0);
+  }
+  return Status::OK();
+}
+
+Status ArrayBuilder::AppendToBitmap(bool is_valid) {
+  if (length_ == capacity_) {
+    // If the capacity was not already a multiple of 2, do so here
+    // TODO(emkornfield) doubling isn't great default allocation practice
+    // see https://github.com/facebook/folly/blob/master/folly/docs/FBVector.md
+    // fo discussion
+    RETURN_NOT_OK(Resize(BitUtil::NextPower2(capacity_ + 1)));
+  }
+  UnsafeAppendToBitmap(is_valid);
+  return Status::OK();
+}
+
+Status ArrayBuilder::AppendToBitmap(const uint8_t* valid_bytes, int64_t length) {
+  RETURN_NOT_OK(Reserve(length));
+
+  UnsafeAppendToBitmap(valid_bytes, length);
+  return Status::OK();
+}
+
+Status ArrayBuilder::Resize(int64_t capacity) {
+  // Target size of validity (null) bitmap data
+  const int64_t new_bitmap_size = BitUtil::BytesForBits(capacity);
+  RETURN_NOT_OK(CheckCapacity(capacity, capacity_));
+
+  if (capacity_ == 0) {
+    RETURN_NOT_OK(AllocateResizableBuffer(pool_, new_bitmap_size, &null_bitmap_));
+    null_bitmap_data_ = null_bitmap_->mutable_data();
+
+    // Padding is zeroed by AllocateResizableBuffer
+    memset(null_bitmap_data_, 0, static_cast<size_t>(new_bitmap_size));
+  } else {
+    const int64_t old_bitmap_capacity = null_bitmap_->capacity();
+    RETURN_NOT_OK(null_bitmap_->Resize(new_bitmap_size));
+
+    const int64_t new_bitmap_capacity = null_bitmap_->capacity();
+    null_bitmap_data_ = null_bitmap_->mutable_data();
+
+    // Zero the region between the original capacity and the new capacity,
+    // including padding, which has not been zeroed, unlike
+    // AllocateResizableBuffer
+    if (old_bitmap_capacity < new_bitmap_capacity) {
+      memset(null_bitmap_data_ + old_bitmap_capacity, 0,
+             static_cast<size_t>(new_bitmap_capacity - old_bitmap_capacity));
+    }
+  }
+  capacity_ = capacity;
+  return Status::OK();
+}
+
+Status ArrayBuilder::Advance(int64_t elements) {
+  if (length_ + elements > capacity_) {
+    return Status::Invalid("Builder must be expanded");
+  }
+  length_ += elements;
+  return Status::OK();
+}
+
+Status ArrayBuilder::Finish(std::shared_ptr<Array>* out) {
+  std::shared_ptr<ArrayData> internal_data;
+  RETURN_NOT_OK(FinishInternal(&internal_data));
+  *out = MakeArray(internal_data);
+  return Status::OK();
+}
+
+Status ArrayBuilder::Reserve(int64_t additional_elements) {
+  if (length_ + additional_elements > capacity_) {
+    // TODO(emkornfield) power of 2 growth is potentially suboptimal
+    int64_t new_size = BitUtil::NextPower2(length_ + additional_elements);
+    return Resize(new_size);
+  }
+  return Status::OK();
+}
+
+void ArrayBuilder::Reset() {
+  capacity_ = length_ = null_count_ = 0;
+  null_bitmap_ = nullptr;
+}
+
+Status ArrayBuilder::SetNotNull(int64_t length) {
+  RETURN_NOT_OK(Reserve(length));
+  UnsafeSetNotNull(length);
+  return Status::OK();
+}
+
+void ArrayBuilder::UnsafeAppendToBitmap(const uint8_t* valid_bytes, int64_t length) {
+  if (valid_bytes == nullptr) {
+    UnsafeSetNotNull(length);
+    return;
+  }
+  UnsafeAppendToBitmap(valid_bytes, valid_bytes + length);
+}
+
+void ArrayBuilder::UnsafeAppendToBitmap(const std::vector<bool>& is_valid) {
+  UnsafeAppendToBitmap(is_valid.begin(), is_valid.end());
+}
+
+void ArrayBuilder::UnsafeSetNotNull(int64_t length) {
+  const int64_t new_length = length + length_;
+
+  // Fill up the bytes until we have a byte alignment
+  int64_t pad_to_byte = std::min<int64_t>(8 - (length_ % 8), length);
+
+  if (pad_to_byte == 8) {
+    pad_to_byte = 0;
+  }
+  for (int64_t i = length_; i < length_ + pad_to_byte; ++i) {
+    BitUtil::SetBit(null_bitmap_data_, i);
+  }
+
+  // Fast bitsetting
+  int64_t fast_length = (length - pad_to_byte) / 8;
+  memset(null_bitmap_data_ + ((length_ + pad_to_byte) / 8), 0xFF,
+         static_cast<size_t>(fast_length));
+
+  // Trailing bits
+  for (int64_t i = length_ + pad_to_byte + (fast_length * 8); i < new_length; ++i) {
+    BitUtil::SetBit(null_bitmap_data_, i);
+  }
+
+  length_ = new_length;
+}
+
+}  // namespace arrow

cpp/src/arrow/columnar/builder_base.h

@@ -0,0 +1,228 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#pragma once
+
+#include "arrow/columnar/builder_base.h"
+
+#include <algorithm>  // IWYU pragma: keep
+#include <array>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <iterator>
+#include <limits>
+#include <memory>
+#include <string>
+#include <type_traits>
+#include <vector>
+
+#include "arrow/buffer.h"
+#include "arrow/memory_pool.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/type_traits.h"
+#include "arrow/util/bit-util.h"
+#include "arrow/util/macros.h"
+#include "arrow/util/string_view.h"
+#include "arrow/util/type_traits.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+class Array;
+struct ArrayData;
+
+constexpr int64_t kMinBuilderCapacity = 1 << 5;
+constexpr int64_t kListMaximumElements = std::numeric_limits<int32_t>::max() - 1;
+
+
+/// Base class for all data array builders.
+///
+/// This class provides a facilities for incrementally building the null bitmap
+/// (see Append methods) and as a side effect the current number of slots and
+/// the null count.
+///
+/// \note Users are expected to use builders as one of the concrete types below.
+/// For example, ArrayBuilder* pointing to BinaryBuilder should be downcast before use.
+class ARROW_EXPORT ArrayBuilder {
+ public:
+  explicit ArrayBuilder(const std::shared_ptr<DataType>& type, MemoryPool* pool)
+      : type_(type),
+        pool_(pool),
+        null_bitmap_(NULLPTR),
+        null_count_(0),
+        null_bitmap_data_(NULLPTR),
+        length_(0),
+        capacity_(0) {}
+
+  virtual ~ArrayBuilder() = default;
+
+  /// For nested types. Since the objects are owned by this class instance, we
+  /// skip shared pointers and just return a raw pointer
+  ArrayBuilder* child(int i) { return children_[i].get(); }
+
+  int num_children() const { return static_cast<int>(children_.size()); }
+
+  int64_t length() const { return length_; }
+  int64_t null_count() const { return null_count_; }
+  int64_t capacity() const { return capacity_; }
+
+  /// \brief Ensure that enough memory has been allocated to fit the indicated
+  /// number of total elements in the builder, including any that have already
+  /// been appended. Does not account for reallocations that may be due to
+  /// variable size data, like binary values. To make space for incremental
+  /// appends, use Reserve instead.
+  ///
+  /// \param[in] capacity the minimum number of total array values to
+  ///            accommodate. Must be greater than the current capacity.
+  /// \return Status
+  virtual Status Resize(int64_t capacity);
+
+  /// \brief Ensure that there is enough space allocated to add the indicated
+  /// number of elements without any further calls to Resize. The memory
+  /// allocated is rounded up to the next highest power of 2 similar to memory
+  /// allocations in STL containers like std::vector
+  /// \param[in] additional_capacity the number of additional array values
+  /// \return Status
+  Status Reserve(int64_t additional_capacity);
+
+  /// Reset the builder.
+  virtual void Reset();
+
+  /// For cases where raw data was memcpy'd into the internal buffers, allows us
+  /// to advance the length of the builder. It is your responsibility to use
+  /// this function responsibly.
+  Status Advance(int64_t elements);
+
+  /// \brief Return result of builder as an internal generic ArrayData
+  /// object. Resets builder except for dictionary builder
+  ///
+  /// \param[out] out the finalized ArrayData object
+  /// \return Status
+  virtual Status FinishInternal(std::shared_ptr<ArrayData>* out) = 0;
+
+  /// \brief Return result of builder as an Array object.
+  ///
+  /// The builder is reset except for DictionaryBuilder.
+  ///
+  /// \param[out] out the finalized Array object
+  /// \return Status
+  Status Finish(std::shared_ptr<Array>* out);
+
+  std::shared_ptr<DataType> type() const { return type_; }
+
+ protected:
+  ArrayBuilder() {}
+
+  /// Append to null bitmap
+  Status AppendToBitmap(bool is_valid);
+
+  /// Vector append. Treat each zero byte as a null.   If valid_bytes is null
+  /// assume all of length bits are valid.
+  Status AppendToBitmap(const uint8_t* valid_bytes, int64_t length);
+
+  /// Set the next length bits to not null (i.e. valid).
+  Status SetNotNull(int64_t length);
+
+  // Unsafe operations (don't check capacity/don't resize)
+
+  void UnsafeAppendNull() { UnsafeAppendToBitmap(false); }
+
+  // Append to null bitmap, update the length
+  void UnsafeAppendToBitmap(bool is_valid) {
+    if (is_valid) {
+      BitUtil::SetBit(null_bitmap_data_, length_);
+    } else {
+      ++null_count_;
+    }
+    ++length_;
+  }
+
+  template <typename IterType>
+  void UnsafeAppendToBitmap(const IterType& begin, const IterType& end) {
+    int64_t byte_offset = length_ / 8;
+    int64_t bit_offset = length_ % 8;
+    uint8_t bitset = null_bitmap_data_[byte_offset];
+
+    for (auto iter = begin; iter != end; ++iter) {
+      if (bit_offset == 8) {
+        bit_offset = 0;
+        null_bitmap_data_[byte_offset] = bitset;
+        byte_offset++;
+        // TODO: Except for the last byte, this shouldn't be needed
+        bitset = null_bitmap_data_[byte_offset];
+      }
+
+      if (*iter) {
+        bitset |= BitUtil::kBitmask[bit_offset];
+      } else {
+        bitset &= BitUtil::kFlippedBitmask[bit_offset];
+        ++null_count_;
+      }
+
+      bit_offset++;
+    }
+
+    if (bit_offset != 0) {
+      null_bitmap_data_[byte_offset] = bitset;
+    }
+
+    length_ += std::distance(begin, end);
+  }
+
+  // Vector append. Treat each zero byte as a nullzero. If valid_bytes is null
+  // assume all of length bits are valid.
+  void UnsafeAppendToBitmap(const uint8_t* valid_bytes, int64_t length);
+
+  void UnsafeAppendToBitmap(const std::vector<bool>& is_valid);
+
+  // Set the next length bits to not null (i.e. valid).
+  void UnsafeSetNotNull(int64_t length);
+
+  static Status TrimBuffer(const int64_t bytes_filled, ResizableBuffer* buffer);
+
+  static Status CheckCapacity(int64_t new_capacity, int64_t old_capacity) {
+    if (new_capacity < 0) {
+      return Status::Invalid("Resize capacity must be positive");
+    }
+    if (new_capacity < old_capacity) {
+      return Status::Invalid("Resize cannot downsize");
+    }
+    return Status::OK();
+  }
+
+  std::shared_ptr<DataType> type_;
+  MemoryPool* pool_;
+
+  // When null_bitmap are first appended to the builder, the null bitmap is allocated
+  std::shared_ptr<ResizableBuffer> null_bitmap_;
+  int64_t null_count_;
+  uint8_t* null_bitmap_data_;
+
+  // Array length, so far. Also, the index of the next element to be added
+  int64_t length_;
+  int64_t capacity_;
+
+  // Child value array builders. These are owned by this class
+  std::vector<std::unique_ptr<ArrayBuilder>> children_;
+
+ private:
+  ARROW_DISALLOW_COPY_AND_ASSIGN(ArrayBuilder);
+};
+
+}  // namespace arrow

cpp/src/arrow/columnar/builder_binary.cc

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include "arrow/columnar/builder_binary.h"
+
 #include <algorithm>
 #include <cstddef>
 #include <cstdint>
@@ -27,7 +29,6 @@
 
 #include "arrow/array.h"
 #include "arrow/buffer.h"
-#include "arrow/builder.h"
 #include "arrow/status.h"
 #include "arrow/type.h"
 #include "arrow/type_traits.h"
@@ -79,23 +80,6 @@ Status BinaryBuilder::AppendNextOffset() {
   return offsets_builder_.Append(static_cast<int32_t>(num_bytes));
 }
 
-Status BinaryBuilder::Append(const uint8_t* value, int32_t length) {
-  RETURN_NOT_OK(Reserve(1));
-  RETURN_NOT_OK(AppendNextOffset());
-  RETURN_NOT_OK(value_data_builder_.Append(value, length));
-
-  UnsafeAppendToBitmap(true);
-  return Status::OK();
-}
-
-Status BinaryBuilder::AppendNull() {
-  RETURN_NOT_OK(AppendNextOffset());
-  RETURN_NOT_OK(Reserve(1));
-
-  UnsafeAppendToBitmap(false);
-  return Status::OK();
-}
-
 Status BinaryBuilder::FinishInternal(std::shared_ptr<ArrayData>* out) {
   // Write final offset (values length)
   RETURN_NOT_OK(AppendNextOffset());
@@ -292,24 +276,50 @@ util::string_view FixedSizeBinaryBuilder::GetView(int64_t i) const {
 }
 
 // ----------------------------------------------------------------------
-// Decimal128Builder
+// ChunkedArray builders
 
-Decimal128Builder::Decimal128Builder(const std::shared_ptr<DataType>& type,
-                                     MemoryPool* pool)
-    : FixedSizeBinaryBuilder(type, pool) {}
+namespace internal {
 
-Status Decimal128Builder::Append(const Decimal128& value) {
-  RETURN_NOT_OK(FixedSizeBinaryBuilder::Reserve(1));
-  return FixedSizeBinaryBuilder::Append(value.ToBytes());
-}
-
-Status Decimal128Builder::FinishInternal(std::shared_ptr<ArrayData>* out) {
-  std::shared_ptr<Buffer> data;
-  RETURN_NOT_OK(byte_builder_.Finish(&data));
-
-  *out = ArrayData::Make(type_, length_, {null_bitmap_, data}, null_count_);
+ChunkedBinaryBuilder::ChunkedBinaryBuilder(int32_t max_chunk_size, MemoryPool* pool)
+    : max_chunk_size_(max_chunk_size),
+      chunk_data_size_(0),
+      builder_(new BinaryBuilder(pool)) {}
 
+Status ChunkedBinaryBuilder::Finish(ArrayVector* out) {
+  if (builder_->length() > 0 || chunks_.size() == 0) {
+    std::shared_ptr<Array> chunk;
+    RETURN_NOT_OK(builder_->Finish(&chunk));
+    chunks_.emplace_back(std::move(chunk));
+  }
+  *out = std::move(chunks_);
   return Status::OK();
 }
 
+Status ChunkedBinaryBuilder::NextChunk() {
+  std::shared_ptr<Array> chunk;
+  RETURN_NOT_OK(builder_->Finish(&chunk));
+  chunks_.emplace_back(std::move(chunk));
+
+  chunk_data_size_ = 0;
+  return Status::OK();
+}
+
+Status ChunkedStringBuilder::Finish(ArrayVector* out) {
+  ArrayVector temp;
+  RETURN_NOT_OK(ChunkedBinaryBuilder::Finish(&temp));
+
+  out->clear();
+  out->reserve(temp.size());
+
+  // XXX: Change data type to string/utf8
+  for (const auto& chunk : temp) {
+    std::shared_ptr<ArrayData> data = chunk->data();
+    data->type = ::arrow::utf8();
+    out->emplace_back(std::make_shared<StringArray>(data));
+  }
+  return Status::OK();
+}
+
+}  // namespace internal
+
 }  // namespace arrow

cpp/src/arrow/columnar/builder_binary.h

@@ -0,0 +1,298 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#pragma once
+
+#include <limits>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "arrow/array.h"
+#include "arrow/columnar/builder_base.h"
+#include "arrow/status.h"
+#include "arrow/type_traits.h"
+#include "arrow/util/macros.h"
+#include "arrow/util/string_view.h"
+
+namespace arrow {
+
+constexpr int64_t kBinaryMemoryLimit = std::numeric_limits<int32_t>::max() - 1;
+
+// ----------------------------------------------------------------------
+// Binary and String
+
+/// \class BinaryBuilder
+/// \brief Builder class for variable-length binary data
+class ARROW_EXPORT BinaryBuilder : public ArrayBuilder {
+ public:
+  explicit BinaryBuilder(MemoryPool* pool ARROW_MEMORY_POOL_DEFAULT);
+
+  BinaryBuilder(const std::shared_ptr<DataType>& type, MemoryPool* pool);
+
+  Status Append(const uint8_t* value, int32_t length) {
+    ARROW_RETURN_NOT_OK(Reserve(1));
+    ARROW_RETURN_NOT_OK(AppendNextOffset());
+    ARROW_RETURN_NOT_OK(value_data_builder_.Append(value, length));
+
+    UnsafeAppendToBitmap(true);
+    return Status::OK();
+  }
+
+  Status AppendNull() {
+    ARROW_RETURN_NOT_OK(AppendNextOffset());
+    ARROW_RETURN_NOT_OK(Reserve(1));
+    UnsafeAppendToBitmap(false);
+    return Status::OK();
+  }
+
+  Status Append(const char* value, int32_t length) {
+    return Append(reinterpret_cast<const uint8_t*>(value), length);
+  }
+
+  Status Append(util::string_view value) {
+    return Append(value.data(), static_cast<int32_t>(value.size()));
+  }
+
+  /// \brief Append without checking capacity
+  ///
+  /// Offsets and data should have been presized using Reserve() and
+  /// ReserveData(), respectively.
+  void UnsafeAppend(const uint8_t* value, int32_t length) {
+    UnsafeAppendNextOffset();
+    value_data_builder_.UnsafeAppend(value, length);
+    UnsafeAppendToBitmap(true);
+  }
+
+  void UnsafeAppend(const char* value, int32_t length) {
+    UnsafeAppend(reinterpret_cast<const uint8_t*>(value), length);
+  }
+
+  void UnsafeAppend(const std::string& value) {
+    UnsafeAppend(value.c_str(), static_cast<int32_t>(value.size()));
+  }
+
+  void UnsafeAppendNull() {
+    const int64_t num_bytes = value_data_builder_.length();
+    offsets_builder_.UnsafeAppend(static_cast<int32_t>(num_bytes));
+    UnsafeAppendToBitmap(false);
+  }
+
+  void Reset() override;
+  Status Resize(int64_t capacity) override;
+
+  /// \brief Ensures there is enough allocated capacity to append the indicated
+  /// number of bytes to the value data buffer without additional allocations
+  Status ReserveData(int64_t elements);
+
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+
+  /// \return size of values buffer so far
+  int64_t value_data_length() const { return value_data_builder_.length(); }
+  /// \return capacity of values buffer
+  int64_t value_data_capacity() const { return value_data_builder_.capacity(); }
+
+  /// Temporary access to a value.
+  ///
+  /// This pointer becomes invalid on the next modifying operation.
+  const uint8_t* GetValue(int64_t i, int32_t* out_length) const;
+
+  /// Temporary access to a value.
+  ///
+  /// This view becomes invalid on the next modifying operation.
+  util::string_view GetView(int64_t i) const;
+
+ protected:
+  TypedBufferBuilder<int32_t> offsets_builder_;
+  TypedBufferBuilder<uint8_t> value_data_builder_;
+
+  Status AppendNextOffset();
+
+  void UnsafeAppendNextOffset() {
+    const int64_t num_bytes = value_data_builder_.length();
+    offsets_builder_.UnsafeAppend(static_cast<int32_t>(num_bytes));
+  }
+};
+
+/// \class StringBuilder
+/// \brief Builder class for UTF8 strings
+class ARROW_EXPORT StringBuilder : public BinaryBuilder {
+ public:
+  using BinaryBuilder::BinaryBuilder;
+  explicit StringBuilder(MemoryPool* pool ARROW_MEMORY_POOL_DEFAULT);
+
+  using BinaryBuilder::Append;
+  using BinaryBuilder::Reset;
+  using BinaryBuilder::UnsafeAppend;
+
+  /// \brief Append a sequence of strings in one shot.
+  ///
+  /// \param[in] values a vector of strings
+  /// \param[in] valid_bytes an optional sequence of bytes where non-zero
+  /// indicates a valid (non-null) value
+  /// \return Status
+  Status AppendValues(const std::vector<std::string>& values,
+                      const uint8_t* valid_bytes = NULLPTR);
+
+  /// \brief Append a sequence of nul-terminated strings in one shot.
+  ///        If one of the values is NULL, it is processed as a null
+  ///        value even if the corresponding valid_bytes entry is 1.
+  ///
+  /// \param[in] values a contiguous C array of nul-terminated char *
+  /// \param[in] length the number of values to append
+  /// \param[in] valid_bytes an optional sequence of bytes where non-zero
+  /// indicates a valid (non-null) value
+  /// \return Status
+  Status AppendValues(const char** values, int64_t length,
+                      const uint8_t* valid_bytes = NULLPTR);
+};
+
+// ----------------------------------------------------------------------
+// FixedSizeBinaryBuilder
+
+class ARROW_EXPORT FixedSizeBinaryBuilder : public ArrayBuilder {
+ public:
+  FixedSizeBinaryBuilder(const std::shared_ptr<DataType>& type,
+                         MemoryPool* pool ARROW_MEMORY_POOL_DEFAULT);
+
+  Status Append(const uint8_t* value) {
+    ARROW_RETURN_NOT_OK(Reserve(1));
+    UnsafeAppendToBitmap(true);
+    return byte_builder_.Append(value, byte_width_);
+  }
+
+  Status Append(const char* value) {
+    return Append(reinterpret_cast<const uint8_t*>(value));
+  }
+
+  Status Append(const util::string_view& view) {
+#ifndef NDEBUG
+    CheckValueSize(static_cast<int64_t>(view.size()));
+#endif
+    return Append(reinterpret_cast<const uint8_t*>(view.data()));
+  }
+
+  Status Append(const std::string& s) {
+#ifndef NDEBUG
+    CheckValueSize(static_cast<int64_t>(s.size()));
+#endif
+    return Append(reinterpret_cast<const uint8_t*>(s.data()));
+  }
+
+  template <size_t NBYTES>
+  Status Append(const std::array<uint8_t, NBYTES>& value) {
+    ARROW_RETURN_NOT_OK(Reserve(1));
+    UnsafeAppendToBitmap(true);
+    return byte_builder_.Append(value);
+  }
+
+  Status AppendValues(const uint8_t* data, int64_t length,
+                      const uint8_t* valid_bytes = NULLPTR);
+  Status AppendNull();
+
+  void Reset() override;
+  Status Resize(int64_t capacity) override;
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+
+  /// \return size of values buffer so far
+  int64_t value_data_length() const { return byte_builder_.length(); }
+
+  int32_t byte_width() const { return byte_width_; }
+
+  /// Temporary access to a value.
+  ///
+  /// This pointer becomes invalid on the next modifying operation.
+  const uint8_t* GetValue(int64_t i) const;
+
+  /// Temporary access to a value.
+  ///
+  /// This view becomes invalid on the next modifying operation.
+  util::string_view GetView(int64_t i) const;
+
+ protected:
+  int32_t byte_width_;
+  BufferBuilder byte_builder_;
+
+#ifndef NDEBUG
+  void CheckValueSize(int64_t size);
+#endif
+};
+
+// ----------------------------------------------------------------------
+// Chunked builders: build a sequence of BinaryArray or StringArray that are
+// limited to a particular size (to the upper limit of 2GB)
+
+namespace internal {
+
+class ARROW_EXPORT ChunkedBinaryBuilder {
+ public:
+  ChunkedBinaryBuilder(int32_t max_chunk_size,
+                       MemoryPool* pool ARROW_MEMORY_POOL_DEFAULT);
+
+  virtual ~ChunkedBinaryBuilder() = default;
+
+  Status Append(const uint8_t* value, int32_t length) {
+    if (ARROW_PREDICT_FALSE(length + chunk_data_size_ > max_chunk_size_)) {
+      // Move onto next chunk, unless the builder length is currently 0, which
+      // means that max_chunk_size_ is less than the item length
+      if (builder_->length() > 0) {
+        ARROW_RETURN_NOT_OK(NextChunk());
+      }
+      // else fall through
+    }
+
+    chunk_data_size_ += length;
+    return builder_->Append(value, length);
+  }
+
+  Status Append(const util::string_view& value) {
+    return Append(reinterpret_cast<const uint8_t*>(value.data()),
+                  static_cast<int32_t>(value.size()));
+  }
+
+  Status AppendNull() {
+    if (ARROW_PREDICT_FALSE(builder_->length() == std::numeric_limits<int32_t>::max())) {
+      ARROW_RETURN_NOT_OK(NextChunk());
+    }
+    return builder_->AppendNull();
+  }
+
+  virtual Status Finish(ArrayVector* out);
+
+ protected:
+  ChunkedBinaryBuilder(const std::shared_ptr<DataType>& type, int32_t max_chunk_size,
+                       MemoryPool* pool);
+
+  Status NextChunk();
+
+  int32_t max_chunk_size_;
+  int32_t chunk_data_size_;
+
+  std::unique_ptr<BinaryBuilder> builder_;
+  std::vector<std::shared_ptr<Array>> chunks_;
+};
+
+class ARROW_EXPORT ChunkedStringBuilder : public ChunkedBinaryBuilder {
+ public:
+  using ChunkedBinaryBuilder::ChunkedBinaryBuilder;
+
+  Status Finish(ArrayVector* out) override;
+};
+
+}  // namespace internal
+
+}  // namespace arrow

cpp/src/arrow/columnar/builder_decimal.cc

@@ -0,0 +1,64 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#include "arrow/columnar/builder_decimal.h"
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <memory>
+#include <numeric>
+#include <sstream>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/array.h"
+#include "arrow/buffer.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/type_traits.h"
+#include "arrow/util/bit-util.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/decimal.h"
+#include "arrow/util/logging.h"
+
+namespace arrow {
+
+// ----------------------------------------------------------------------
+// Decimal128Builder
+
+Decimal128Builder::Decimal128Builder(const std::shared_ptr<DataType>& type,
+                                     MemoryPool* pool)
+    : FixedSizeBinaryBuilder(type, pool) {}
+
+Status Decimal128Builder::Append(const Decimal128& value) {
+  RETURN_NOT_OK(FixedSizeBinaryBuilder::Reserve(1));
+  return FixedSizeBinaryBuilder::Append(value.ToBytes());
+}
+
+Status Decimal128Builder::FinishInternal(std::shared_ptr<ArrayData>* out) {
+  std::shared_ptr<Buffer> data;
+  RETURN_NOT_OK(byte_builder_.Finish(&data));
+
+  *out = ArrayData::Make(type_, length_, {null_bitmap_, data}, null_count_);
+
+  return Status::OK();
+}
+
+}  // namespace arrow

cpp/src/arrow/columnar/builder_decimal.h

@@ -0,0 +1,45 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#pragma once
+
+#include <memory>
+
+#include "arrow/columnar/builder_base.h"
+#include "arrow/columnar/builder_binary.h"
+
+namespace arrow {
+
+class Decimal128;
+
+class ARROW_EXPORT Decimal128Builder : public FixedSizeBinaryBuilder {
+ public:
+  explicit Decimal128Builder(const std::shared_ptr<DataType>& type,
+                             MemoryPool* pool ARROW_MEMORY_POOL_DEFAULT);
+
+  using FixedSizeBinaryBuilder::Append;
+  using FixedSizeBinaryBuilder::AppendValues;
+  using FixedSizeBinaryBuilder::Reset;
+
+  Status Append(const Decimal128& val);
+
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+};
+
+using DecimalBuilder = Decimal128Builder;
+
+}  // namespace arrow

cpp/src/arrow/columnar/builder_dict.cc

@@ -15,13 +15,15 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include "arrow/columnar/builder_dict.h"
+
+#include <algorithm>
 #include <cstdint>
 #include <utility>
 #include <vector>
 
 #include "arrow/array.h"
 #include "arrow/buffer.h"
-#include "arrow/builder.h"
 #include "arrow/status.h"
 #include "arrow/type.h"
 #include "arrow/type_traits.h"

cpp/src/arrow/columnar/builder_dict.h

@@ -0,0 +1,168 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#pragma once
+
+#include <memory>
+
+#include "arrow/columnar/builder_adaptive.h"  // IWYU pragma: export
+#include "arrow/columnar/builder_base.h"      // IWYU pragma: export
+
+namespace arrow {
+
+
+// ----------------------------------------------------------------------
+// Dictionary builder
+
+namespace internal {
+
+template <typename T>
+struct DictionaryScalar {
+  using type = typename T::c_type;
+};
+
+template <>
+struct DictionaryScalar<BinaryType> {
+  using type = util::string_view;
+};
+
+template <>
+struct DictionaryScalar<StringType> {
+  using type = util::string_view;
+};
+
+template <>
+struct DictionaryScalar<FixedSizeBinaryType> {
+  using type = util::string_view;
+};
+
+}  // namespace internal
+
+/// \brief Array builder for created encoded DictionaryArray from dense array
+///
+/// Unlike other builders, dictionary builder does not completely reset the state
+/// on Finish calls. The arrays built after the initial Finish call will reuse
+/// the previously created encoding and build a delta dictionary when new terms
+/// occur.
+///
+/// data
+template <typename T>
+class ARROW_EXPORT DictionaryBuilder : public ArrayBuilder {
+ public:
+  using Scalar = typename internal::DictionaryScalar<T>::type;
+
+  // WARNING: the type given below is the value type, not the DictionaryType.
+  // The DictionaryType is instantiated on the Finish() call.
+  DictionaryBuilder(const std::shared_ptr<DataType>& type, MemoryPool* pool);
+
+  template <typename T1 = T>
+  explicit DictionaryBuilder(
+      typename std::enable_if<TypeTraits<T1>::is_parameter_free, MemoryPool*>::type pool)
+      : DictionaryBuilder<T1>(TypeTraits<T1>::type_singleton(), pool) {}
+
+  ~DictionaryBuilder() override;
+
+  /// \brief Append a scalar value
+  Status Append(const Scalar& value);
+
+  /// \brief Append a fixed-width string (only for FixedSizeBinaryType)
+  template <typename T1 = T>
+  Status Append(typename std::enable_if<std::is_base_of<FixedSizeBinaryType, T1>::value,
+                                        const uint8_t*>::type value) {
+    return Append(util::string_view(reinterpret_cast<const char*>(value), byte_width_));
+  }
+
+  /// \brief Append a fixed-width string (only for FixedSizeBinaryType)
+  template <typename T1 = T>
+  Status Append(typename std::enable_if<std::is_base_of<FixedSizeBinaryType, T1>::value,
+                                        const char*>::type value) {
+    return Append(util::string_view(value, byte_width_));
+  }
+
+  /// \brief Append a scalar null value
+  Status AppendNull();
+
+  /// \brief Append a whole dense array to the builder
+  Status AppendArray(const Array& array);
+
+  void Reset() override;
+  Status Resize(int64_t capacity) override;
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+
+  /// is the dictionary builder in the delta building mode
+  bool is_building_delta() { return delta_offset_ > 0; }
+
+ protected:
+  class MemoTableImpl;
+  std::unique_ptr<MemoTableImpl> memo_table_;
+
+  int32_t delta_offset_;
+  // Only used for FixedSizeBinaryType
+  int32_t byte_width_;
+
+  AdaptiveIntBuilder values_builder_;
+};
+
+template <>
+class ARROW_EXPORT DictionaryBuilder<NullType> : public ArrayBuilder {
+ public:
+  DictionaryBuilder(const std::shared_ptr<DataType>& type, MemoryPool* pool);
+  explicit DictionaryBuilder(MemoryPool* pool);
+
+  /// \brief Append a scalar null value
+  Status AppendNull();
+
+  /// \brief Append a whole dense array to the builder
+  Status AppendArray(const Array& array);
+
+  Status Resize(int64_t capacity) override;
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+
+ protected:
+  AdaptiveIntBuilder values_builder_;
+};
+
+class ARROW_EXPORT BinaryDictionaryBuilder : public DictionaryBuilder<BinaryType> {
+ public:
+  using DictionaryBuilder::Append;
+  using DictionaryBuilder::DictionaryBuilder;
+
+  Status Append(const uint8_t* value, int32_t length) {
+    return Append(reinterpret_cast<const char*>(value), length);
+  }
+
+  Status Append(const char* value, int32_t length) {
+    return Append(util::string_view(value, length));
+  }
+};
+
+/// \brief Dictionary array builder with convenience methods for strings
+class ARROW_EXPORT StringDictionaryBuilder : public DictionaryBuilder<StringType> {
+ public:
+  using DictionaryBuilder::Append;
+  using DictionaryBuilder::DictionaryBuilder;
+
+  Status Append(const uint8_t* value, int32_t length) {
+    return Append(reinterpret_cast<const char*>(value), length);
+  }
+
+  Status Append(const char* value, int32_t length) {
+    return Append(util::string_view(value, length));
+  }
+};
+
+}  // namespace arrow

cpp/src/arrow/columnar/builder_nested.cc

@@ -0,0 +1,156 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#include "arrow/columnar/builder_nested.h"
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <sstream>
+#include <utility>
+#include <vector>
+
+#include "arrow/array.h"
+#include "arrow/buffer.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/type_traits.h"
+#include "arrow/util/bit-util.h"
+#include "arrow/util/int-util.h"
+#include "arrow/util/logging.h"
+
+namespace arrow {
+
+// ----------------------------------------------------------------------
+// ListBuilder
+
+ListBuilder::ListBuilder(MemoryPool* pool,
+                         std::shared_ptr<ArrayBuilder> const& value_builder,
+                         const std::shared_ptr<DataType>& type)
+    : ArrayBuilder(type ? type
+                        : std::static_pointer_cast<DataType>(
+                              std::make_shared<ListType>(value_builder->type())),
+                   pool),
+      offsets_builder_(pool),
+      value_builder_(value_builder) {}
+
+Status ListBuilder::AppendValues(const int32_t* offsets, int64_t length,
+                                 const uint8_t* valid_bytes) {
+  RETURN_NOT_OK(Reserve(length));
+  UnsafeAppendToBitmap(valid_bytes, length);
+  offsets_builder_.UnsafeAppend(offsets, length);
+  return Status::OK();
+}
+
+Status ListBuilder::AppendNextOffset() {
+  int64_t num_values = value_builder_->length();
+  if (ARROW_PREDICT_FALSE(num_values > kListMaximumElements)) {
+    std::stringstream ss;
+    ss << "ListArray cannot contain more then INT32_MAX - 1 child elements,"
+       << " have " << num_values;
+    return Status::CapacityError(ss.str());
+  }
+  return offsets_builder_.Append(static_cast<int32_t>(num_values));
+}
+
+Status ListBuilder::Append(bool is_valid) {
+  RETURN_NOT_OK(Reserve(1));
+  UnsafeAppendToBitmap(is_valid);
+  return AppendNextOffset();
+}
+
+Status ListBuilder::Resize(int64_t capacity) {
+  DCHECK_LE(capacity, kListMaximumElements);
+  RETURN_NOT_OK(CheckCapacity(capacity, capacity_));
+
+  // one more then requested for offsets
+  RETURN_NOT_OK(offsets_builder_.Resize((capacity + 1) * sizeof(int32_t)));
+  return ArrayBuilder::Resize(capacity);
+}
+
+Status ListBuilder::FinishInternal(std::shared_ptr<ArrayData>* out) {
+  RETURN_NOT_OK(AppendNextOffset());
+
+  // Offset padding zeroed by BufferBuilder
+  std::shared_ptr<Buffer> offsets;
+  RETURN_NOT_OK(offsets_builder_.Finish(&offsets));
+
+  std::shared_ptr<ArrayData> items;
+  if (values_) {
+    items = values_->data();
+  } else {
+    if (value_builder_->length() == 0) {
+      // Try to make sure we get a non-null values buffer (ARROW-2744)
+      RETURN_NOT_OK(value_builder_->Resize(0));
+    }
+    RETURN_NOT_OK(value_builder_->FinishInternal(&items));
+  }
+
+  *out = ArrayData::Make(type_, length_, {null_bitmap_, offsets}, null_count_);
+  (*out)->child_data.emplace_back(std::move(items));
+  Reset();
+  return Status::OK();
+}
+
+void ListBuilder::Reset() {
+  ArrayBuilder::Reset();
+  values_.reset();
+  offsets_builder_.Reset();
+  value_builder_->Reset();
+}
+
+ArrayBuilder* ListBuilder::value_builder() const {
+  DCHECK(!values_) << "Using value builder is pointless when values_ is set";
+  return value_builder_.get();
+}
+
+// ----------------------------------------------------------------------
+// Struct
+
+StructBuilder::StructBuilder(const std::shared_ptr<DataType>& type, MemoryPool* pool,
+                             std::vector<std::shared_ptr<ArrayBuilder>>&& field_builders)
+    : ArrayBuilder(type, pool) {
+  children_ = std::move(field_builders);
+}
+
+void StructBuilder::Reset() {
+  ArrayBuilder::Reset();
+  for (const auto& field_builder : children_) {
+    field_builder->Reset();
+  }
+}
+
+Status StructBuilder::FinishInternal(std::shared_ptr<ArrayData>* out) {
+  RETURN_NOT_OK(TrimBuffer(BitUtil::BytesForBits(length_), null_bitmap_.get()));
+  *out = ArrayData::Make(type_, length_, {null_bitmap_}, null_count_);
+
+  (*out)->child_data.resize(children_.size());
+  for (size_t i = 0; i < children_.size(); ++i) {
+    if (length_ == 0) {
+      // Try to make sure the child buffers are initialized
+      RETURN_NOT_OK(children_[i]->Resize(0));
+    }
+    RETURN_NOT_OK(children_[i]->FinishInternal(&(*out)->child_data[i]));
+  }
+
+  null_bitmap_ = nullptr;
+  capacity_ = length_ = null_count_ = 0;
+  return Status::OK();
+}
+
+}  // namespace arrow

cpp/src/arrow/columnar/builder_nested.h

@@ -0,0 +1,125 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#pragma once
+
+#include <memory>
+#include <vector>
+
+#include "arrow/columnar/builder_base.h"
+
+namespace arrow {
+
+
+// ----------------------------------------------------------------------
+// List builder
+
+/// \class ListBuilder
+/// \brief Builder class for variable-length list array value types
+///
+/// To use this class, you must append values to the child array builder and use
+/// the Append function to delimit each distinct list value (once the values
+/// have been appended to the child array) or use the bulk API to append
+/// a sequence of offests and null values.
+///
+/// A note on types.  Per arrow/type.h all types in the c++ implementation are
+/// logical so even though this class always builds list array, this can
+/// represent multiple different logical types.  If no logical type is provided
+/// at construction time, the class defaults to List<T> where t is taken from the
+/// value_builder/values that the object is constructed with.
+class ARROW_EXPORT ListBuilder : public ArrayBuilder {
+ public:
+  /// Use this constructor to incrementally build the value array along with offsets and
+  /// null bitmap.
+  ListBuilder(MemoryPool* pool, std::shared_ptr<ArrayBuilder> const& value_builder,
+              const std::shared_ptr<DataType>& type = NULLPTR);
+
+  Status Resize(int64_t capacity) override;
+  void Reset() override;
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+
+  /// \brief Vector append
+  ///
+  /// If passed, valid_bytes is of equal length to values, and any zero byte
+  /// will be considered as a null for that slot
+  Status AppendValues(const int32_t* offsets, int64_t length,
+                      const uint8_t* valid_bytes = NULLPTR);
+
+  /// \brief Start a new variable-length list slot
+  ///
+  /// This function should be called before beginning to append elements to the
+  /// value builder
+  Status Append(bool is_valid = true);
+
+  Status AppendNull() { return Append(false); }
+
+  ArrayBuilder* value_builder() const;
+
+ protected:
+  TypedBufferBuilder<int32_t> offsets_builder_;
+  std::shared_ptr<ArrayBuilder> value_builder_;
+  std::shared_ptr<Array> values_;
+
+  Status AppendNextOffset();
+};
+
+// ----------------------------------------------------------------------
+// Struct
+
+// ---------------------------------------------------------------------------------
+// StructArray builder
+/// Append, Resize and Reserve methods are acting on StructBuilder.
+/// Please make sure all these methods of all child-builders' are consistently
+/// called to maintain data-structure consistency.
+class ARROW_EXPORT StructBuilder : public ArrayBuilder {
+ public:
+  StructBuilder(const std::shared_ptr<DataType>& type, MemoryPool* pool,
+                std::vector<std::shared_ptr<ArrayBuilder>>&& field_builders);
+
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+
+  /// Null bitmap is of equal length to every child field, and any zero byte
+  /// will be considered as a null for that field, but users must using app-
+  /// end methods or advance methods of the child builders' independently to
+  /// insert data.
+  Status AppendValues(int64_t length, const uint8_t* valid_bytes) {
+    ARROW_RETURN_NOT_OK(Reserve(length));
+    UnsafeAppendToBitmap(valid_bytes, length);
+    return Status::OK();
+  }
+
+  /// Append an element to the Struct. All child-builders' Append method must
+  /// be called independently to maintain data-structure consistency.
+  Status Append(bool is_valid = true) {
+    ARROW_RETURN_NOT_OK(Reserve(1));
+    UnsafeAppendToBitmap(is_valid);
+    return Status::OK();
+  }
+
+  Status AppendNull() { return Append(false); }
+
+  void Reset() override;
+
+  ArrayBuilder* field_builder(int i) const { return field_builders_[i].get(); }
+
+  int num_fields() const { return static_cast<int>(field_builders_.size()); }
+
+ protected:
+  std::vector<std::shared_ptr<ArrayBuilder>> field_builders_;
+};
+
+}  // namespace arrow

cpp/src/arrow/columnar/builder_primitive.cc

@@ -0,0 +1,272 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#include "arrow/columnar/builder_primitive.h"
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <sstream>
+#include <utility>
+#include <vector>
+
+#include "arrow/array.h"
+#include "arrow/buffer.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/type_traits.h"
+#include "arrow/util/bit-util.h"
+#include "arrow/util/int-util.h"
+#include "arrow/util/logging.h"
+
+namespace arrow {
+
+// ----------------------------------------------------------------------
+// Null builder
+
+Status NullBuilder::FinishInternal(std::shared_ptr<ArrayData>* out) {
+  *out = ArrayData::Make(null(), length_, {nullptr}, length_);
+  length_ = null_count_ = 0;
+  return Status::OK();
+}
+
+// ----------------------------------------------------------------------
+
+template <typename T>
+void PrimitiveBuilder<T>::Reset() {
+  data_.reset();
+  raw_data_ = nullptr;
+}
+
+template <typename T>
+Status PrimitiveBuilder<T>::Resize(int64_t capacity) {
+  RETURN_NOT_OK(CheckCapacity(capacity, capacity_));
+  capacity = std::max(capacity, kMinBuilderCapacity);
+
+  int64_t nbytes = TypeTraits<T>::bytes_required(capacity);
+  if (capacity_ == 0) {
+    RETURN_NOT_OK(AllocateResizableBuffer(pool_, nbytes, &data_));
+  } else {
+    RETURN_NOT_OK(data_->Resize(nbytes));
+  }
+
+  raw_data_ = reinterpret_cast<value_type*>(data_->mutable_data());
+  return ArrayBuilder::Resize(capacity);
+}
+
+template <typename T>
+Status PrimitiveBuilder<T>::AppendValues(const value_type* values, int64_t length,
+                                         const uint8_t* valid_bytes) {
+  RETURN_NOT_OK(Reserve(length));
+
+  if (length > 0) {
+    std::memcpy(raw_data_ + length_, values,
+                static_cast<std::size_t>(TypeTraits<T>::bytes_required(length)));
+  }
+
+  // length_ is update by these
+  ArrayBuilder::UnsafeAppendToBitmap(valid_bytes, length);
+  return Status::OK();
+}
+
+template <typename T>
+Status PrimitiveBuilder<T>::AppendValues(const value_type* values, int64_t length,
+                                         const std::vector<bool>& is_valid) {
+  RETURN_NOT_OK(Reserve(length));
+  DCHECK_EQ(length, static_cast<int64_t>(is_valid.size()));
+
+  if (length > 0) {
+    std::memcpy(raw_data_ + length_, values,
+                static_cast<std::size_t>(TypeTraits<T>::bytes_required(length)));
+  }
+
+  // length_ is update by these
+  ArrayBuilder::UnsafeAppendToBitmap(is_valid);
+  return Status::OK();
+}
+
+template <typename T>
+Status PrimitiveBuilder<T>::AppendValues(const std::vector<value_type>& values,
+                                         const std::vector<bool>& is_valid) {
+  return AppendValues(values.data(), static_cast<int64_t>(values.size()), is_valid);
+}
+
+template <typename T>
+Status PrimitiveBuilder<T>::AppendValues(const std::vector<value_type>& values) {
+  return AppendValues(values.data(), static_cast<int64_t>(values.size()));
+}
+
+template <typename T>
+Status PrimitiveBuilder<T>::FinishInternal(std::shared_ptr<ArrayData>* out) {
+  RETURN_NOT_OK(TrimBuffer(BitUtil::BytesForBits(length_), null_bitmap_.get()));
+  RETURN_NOT_OK(TrimBuffer(TypeTraits<T>::bytes_required(length_), data_.get()));
+
+  *out = ArrayData::Make(type_, length_, {null_bitmap_, data_}, null_count_);
+
+  data_ = null_bitmap_ = nullptr;
+  capacity_ = length_ = null_count_ = 0;
+
+  return Status::OK();
+}
+
+template class PrimitiveBuilder<UInt8Type>;
+template class PrimitiveBuilder<UInt16Type>;
+template class PrimitiveBuilder<UInt32Type>;
+template class PrimitiveBuilder<UInt64Type>;
+template class PrimitiveBuilder<Int8Type>;
+template class PrimitiveBuilder<Int16Type>;
+template class PrimitiveBuilder<Int32Type>;
+template class PrimitiveBuilder<Int64Type>;
+template class PrimitiveBuilder<Date32Type>;
+template class PrimitiveBuilder<Date64Type>;
+template class PrimitiveBuilder<Time32Type>;
+template class PrimitiveBuilder<Time64Type>;
+template class PrimitiveBuilder<TimestampType>;
+template class PrimitiveBuilder<HalfFloatType>;
+template class PrimitiveBuilder<FloatType>;
+template class PrimitiveBuilder<DoubleType>;
+
+BooleanBuilder::BooleanBuilder(MemoryPool* pool)
+    : ArrayBuilder(boolean(), pool), data_(nullptr), raw_data_(nullptr) {}
+
+BooleanBuilder::BooleanBuilder(const std::shared_ptr<DataType>& type, MemoryPool* pool)
+    : BooleanBuilder(pool) {
+  DCHECK_EQ(Type::BOOL, type->id());
+}
+
+void BooleanBuilder::Reset() {
+  ArrayBuilder::Reset();
+  data_.reset();
+  raw_data_ = nullptr;
+}
+
+Status BooleanBuilder::Resize(int64_t capacity) {
+  RETURN_NOT_OK(CheckCapacity(capacity, capacity_));
+  capacity = std::max(capacity, kMinBuilderCapacity);
+
+  const int64_t new_bitmap_size = BitUtil::BytesForBits(capacity);
+  if (capacity_ == 0) {
+    RETURN_NOT_OK(AllocateResizableBuffer(pool_, new_bitmap_size, &data_));
+    raw_data_ = reinterpret_cast<uint8_t*>(data_->mutable_data());
+
+    // We zero the memory for booleans to keep things simple; for some reason if
+    // we do not, even though we may write every bit (through in-place | or &),
+    // valgrind will still show a warning. If we do not zero the bytes here, we
+    // will have to be careful to zero them in AppendNull and AppendNulls. Also,
+    // zeroing the bits results in deterministic bits when each byte may have a
+    // mix of nulls and not nulls.
+    //
+    // We only zero up to new_bitmap_size because the padding was zeroed by
+    // AllocateResizableBuffer
+    memset(raw_data_, 0, static_cast<size_t>(new_bitmap_size));
+  } else {
+    const int64_t old_bitmap_capacity = data_->capacity();
+    RETURN_NOT_OK(data_->Resize(new_bitmap_size));
+    const int64_t new_bitmap_capacity = data_->capacity();
+    raw_data_ = reinterpret_cast<uint8_t*>(data_->mutable_data());
+
+    // See comment above about why we zero memory for booleans
+    memset(raw_data_ + old_bitmap_capacity, 0,
+           static_cast<size_t>(new_bitmap_capacity - old_bitmap_capacity));
+  }
+
+  return ArrayBuilder::Resize(capacity);
+}
+
+Status BooleanBuilder::FinishInternal(std::shared_ptr<ArrayData>* out) {
+  int64_t bit_offset = length_ % 8;
+  if (bit_offset > 0) {
+    // Adjust last byte
+    data_->mutable_data()[length_ / 8] &= BitUtil::kPrecedingBitmask[bit_offset];
+  }
+
+  RETURN_NOT_OK(TrimBuffer(BitUtil::BytesForBits(length_), null_bitmap_.get()));
+  RETURN_NOT_OK(TrimBuffer(BitUtil::BytesForBits(length_), data_.get()));
+
+  *out = ArrayData::Make(boolean(), length_, {null_bitmap_, data_}, null_count_);
+
+  data_ = null_bitmap_ = nullptr;
+  capacity_ = length_ = null_count_ = 0;
+  return Status::OK();
+}
+
+Status BooleanBuilder::AppendValues(const uint8_t* values, int64_t length,
+                                    const uint8_t* valid_bytes) {
+  RETURN_NOT_OK(Reserve(length));
+
+  int64_t i = 0;
+  internal::GenerateBitsUnrolled(raw_data_, length_, length,
+                                 [values, &i]() -> bool { return values[i++] != 0; });
+
+  // this updates length_
+  ArrayBuilder::UnsafeAppendToBitmap(valid_bytes, length);
+  return Status::OK();
+}
+
+Status BooleanBuilder::AppendValues(const uint8_t* values, int64_t length,
+                                    const std::vector<bool>& is_valid) {
+  RETURN_NOT_OK(Reserve(length));
+  DCHECK_EQ(length, static_cast<int64_t>(is_valid.size()));
+
+  int64_t i = 0;
+  internal::GenerateBitsUnrolled(raw_data_, length_, length,
+                                 [values, &i]() -> bool { return values[i++]; });
+
+  // this updates length_
+  ArrayBuilder::UnsafeAppendToBitmap(is_valid);
+  return Status::OK();
+}
+
+Status BooleanBuilder::AppendValues(const std::vector<uint8_t>& values,
+                                    const std::vector<bool>& is_valid) {
+  return AppendValues(values.data(), static_cast<int64_t>(values.size()), is_valid);
+}
+
+Status BooleanBuilder::AppendValues(const std::vector<uint8_t>& values) {
+  return AppendValues(values.data(), static_cast<int64_t>(values.size()));
+}
+
+Status BooleanBuilder::AppendValues(const std::vector<bool>& values,
+                                    const std::vector<bool>& is_valid) {
+  const int64_t length = static_cast<int64_t>(values.size());
+  RETURN_NOT_OK(Reserve(length));
+  DCHECK_EQ(length, static_cast<int64_t>(is_valid.size()));
+
+  int64_t i = 0;
+  internal::GenerateBitsUnrolled(raw_data_, length_, length,
+                                 [&values, &i]() -> bool { return values[i++]; });
+
+  // this updates length_
+  ArrayBuilder::UnsafeAppendToBitmap(is_valid);
+  return Status::OK();
+}
+
+Status BooleanBuilder::AppendValues(const std::vector<bool>& values) {
+  const int64_t length = static_cast<int64_t>(values.size());
+  RETURN_NOT_OK(Reserve(length));
+
+  int64_t i = 0;
+  internal::GenerateBitsUnrolled(raw_data_, length_, length,
+                                 [&values, &i]() -> bool { return values[i++]; });
+
+  // this updates length_
+  ArrayBuilder::UnsafeSetNotNull(length);
+  return Status::OK();
+}
+
+}  // namespace arrow

cpp/src/arrow/columnar/builder_primitive.h

@@ -0,0 +1,403 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#pragma once
+
+#include <algorithm>
+#include <memory>
+#include <vector>
+
+#include "arrow/columnar/builder_base.h"
+#include "arrow/type.h"
+
+namespace arrow {
+
+class ARROW_EXPORT NullBuilder : public ArrayBuilder {
+ public:
+  explicit NullBuilder(MemoryPool* pool ARROW_MEMORY_POOL_DEFAULT)
+      : ArrayBuilder(null(), pool) {}
+
+  Status AppendNull() {
+    ++null_count_;
+    ++length_;
+    return Status::OK();
+  }
+
+  Status Append(std::nullptr_t value) { return AppendNull(); }
+
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+};
+
+
+template <typename Type>
+class ARROW_EXPORT PrimitiveBuilder : public ArrayBuilder {
+ public:
+  using value_type = typename Type::c_type;
+
+  explicit PrimitiveBuilder(const std::shared_ptr<DataType>& type, MemoryPool* pool)
+      : ArrayBuilder(type, pool), data_(NULLPTR), raw_data_(NULLPTR) {}
+
+  using ArrayBuilder::Advance;
+
+  /// Write nulls as uint8_t* (0 value indicates null) into pre-allocated memory
+  /// The memory at the corresponding data slot is set to 0 to prevent uninitialized
+  /// memory access
+  Status AppendNulls(const uint8_t* valid_bytes, int64_t length) {
+    ARROW_RETURN_NOT_OK(Reserve(length));
+    memset(raw_data_ + length_, 0,
+           static_cast<size_t>(TypeTraits<Type>::bytes_required(length)));
+    UnsafeAppendToBitmap(valid_bytes, length);
+    return Status::OK();
+  }
+
+  /// \brief Append a single null element
+  Status AppendNull() {
+    ARROW_RETURN_NOT_OK(Reserve(1));
+    memset(raw_data_ + length_, 0, sizeof(value_type));
+    UnsafeAppendToBitmap(false);
+    return Status::OK();
+  }
+
+  value_type GetValue(int64_t index) const {
+    return reinterpret_cast<const value_type*>(data_->data())[index];
+  }
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values a contiguous C array of values
+  /// \param[in] length the number of values to append
+  /// \param[in] valid_bytes an optional sequence of bytes where non-zero
+  /// indicates a valid (non-null) value
+  /// \return Status
+  Status AppendValues(const value_type* values, int64_t length,
+                      const uint8_t* valid_bytes = NULLPTR);
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values a contiguous C array of values
+  /// \param[in] length the number of values to append
+  /// \param[in] is_valid an std::vector<bool> indicating valid (1) or null
+  /// (0). Equal in length to values
+  /// \return Status
+  Status AppendValues(const value_type* values, int64_t length,
+                      const std::vector<bool>& is_valid);
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values a std::vector of values
+  /// \param[in] is_valid an std::vector<bool> indicating valid (1) or null
+  /// (0). Equal in length to values
+  /// \return Status
+  Status AppendValues(const std::vector<value_type>& values,
+                      const std::vector<bool>& is_valid);
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values a std::vector of values
+  /// \return Status
+  Status AppendValues(const std::vector<value_type>& values);
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values_begin InputIterator to the beginning of the values
+  /// \param[in] values_end InputIterator pointing to the end of the values
+  /// \return Status
+
+  template <typename ValuesIter>
+  Status AppendValues(ValuesIter values_begin, ValuesIter values_end) {
+    int64_t length = static_cast<int64_t>(std::distance(values_begin, values_end));
+    ARROW_RETURN_NOT_OK(Reserve(length));
+
+    std::copy(values_begin, values_end, raw_data_ + length_);
+
+    // this updates the length_
+    UnsafeSetNotNull(length);
+    return Status::OK();
+  }
+
+  /// \brief Append a sequence of elements in one shot, with a specified nullmap
+  /// \param[in] values_begin InputIterator to the beginning of the values
+  /// \param[in] values_end InputIterator pointing to the end of the values
+  /// \param[in] valid_begin InputIterator with elements indication valid(1)
+  ///  or null(0) values.
+  /// \return Status
+  template <typename ValuesIter, typename ValidIter>
+  typename std::enable_if<!std::is_pointer<ValidIter>::value, Status>::type AppendValues(
+      ValuesIter values_begin, ValuesIter values_end, ValidIter valid_begin) {
+    static_assert(!internal::is_null_pointer<ValidIter>::value,
+                  "Don't pass a NULLPTR directly as valid_begin, use the 2-argument "
+                  "version instead");
+    int64_t length = static_cast<int64_t>(std::distance(values_begin, values_end));
+    ARROW_RETURN_NOT_OK(Reserve(length));
+
+    std::copy(values_begin, values_end, raw_data_ + length_);
+
+    // this updates the length_
+    UnsafeAppendToBitmap(valid_begin, std::next(valid_begin, length));
+    return Status::OK();
+  }
+
+  // Same as above, with a pointer type ValidIter
+  template <typename ValuesIter, typename ValidIter>
+  typename std::enable_if<std::is_pointer<ValidIter>::value, Status>::type AppendValues(
+      ValuesIter values_begin, ValuesIter values_end, ValidIter valid_begin) {
+    int64_t length = static_cast<int64_t>(std::distance(values_begin, values_end));
+    ARROW_RETURN_NOT_OK(Reserve(length));
+
+    std::copy(values_begin, values_end, raw_data_ + length_);
+
+    // this updates the length_
+    if (valid_begin == NULLPTR) {
+      UnsafeSetNotNull(length);
+    } else {
+      UnsafeAppendToBitmap(valid_begin, std::next(valid_begin, length));
+    }
+
+    return Status::OK();
+  }
+
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+  void Reset() override;
+
+  Status Resize(int64_t capacity) override;
+
+ protected:
+  std::shared_ptr<ResizableBuffer> data_;
+  value_type* raw_data_;
+};
+
+/// Base class for all Builders that emit an Array of a scalar numerical type.
+template <typename T>
+class ARROW_EXPORT NumericBuilder : public PrimitiveBuilder<T> {
+ public:
+  using typename PrimitiveBuilder<T>::value_type;
+  using PrimitiveBuilder<T>::PrimitiveBuilder;
+
+  template <typename T1 = T>
+  explicit NumericBuilder(
+      typename std::enable_if<TypeTraits<T1>::is_parameter_free, MemoryPool*>::type pool
+          ARROW_MEMORY_POOL_DEFAULT)
+      : PrimitiveBuilder<T1>(TypeTraits<T1>::type_singleton(), pool) {}
+
+  using ArrayBuilder::UnsafeAppendNull;
+  using PrimitiveBuilder<T>::AppendValues;
+  using PrimitiveBuilder<T>::Resize;
+  using PrimitiveBuilder<T>::Reserve;
+
+  /// Append a single scalar and increase the size if necessary.
+  Status Append(const value_type val) {
+    ARROW_RETURN_NOT_OK(ArrayBuilder::Reserve(1));
+    UnsafeAppend(val);
+    return Status::OK();
+  }
+
+  /// Append a single scalar under the assumption that the underlying Buffer is
+  /// large enough.
+  ///
+  /// This method does not capacity-check; make sure to call Reserve
+  /// beforehand.
+  void UnsafeAppend(const value_type val) {
+    BitUtil::SetBit(null_bitmap_data_, length_);
+    raw_data_[length_++] = val;
+  }
+
+ protected:
+  using PrimitiveBuilder<T>::length_;
+  using PrimitiveBuilder<T>::null_bitmap_data_;
+  using PrimitiveBuilder<T>::raw_data_;
+};
+
+// Builders
+
+using UInt8Builder = NumericBuilder<UInt8Type>;
+using UInt16Builder = NumericBuilder<UInt16Type>;
+using UInt32Builder = NumericBuilder<UInt32Type>;
+using UInt64Builder = NumericBuilder<UInt64Type>;
+
+using Int8Builder = NumericBuilder<Int8Type>;
+using Int16Builder = NumericBuilder<Int16Type>;
+using Int32Builder = NumericBuilder<Int32Type>;
+using Int64Builder = NumericBuilder<Int64Type>;
+using TimestampBuilder = NumericBuilder<TimestampType>;
+using Time32Builder = NumericBuilder<Time32Type>;
+using Time64Builder = NumericBuilder<Time64Type>;
+using Date32Builder = NumericBuilder<Date32Type>;
+using Date64Builder = NumericBuilder<Date64Type>;
+
+using HalfFloatBuilder = NumericBuilder<HalfFloatType>;
+using FloatBuilder = NumericBuilder<FloatType>;
+using DoubleBuilder = NumericBuilder<DoubleType>;
+
+
+class ARROW_EXPORT BooleanBuilder : public ArrayBuilder {
+ public:
+  using value_type = bool;
+  explicit BooleanBuilder(MemoryPool* pool ARROW_MEMORY_POOL_DEFAULT);
+
+  explicit BooleanBuilder(const std::shared_ptr<DataType>& type, MemoryPool* pool);
+
+  using ArrayBuilder::Advance;
+  using ArrayBuilder::UnsafeAppendNull;
+
+  /// Write nulls as uint8_t* (0 value indicates null) into pre-allocated memory
+  Status AppendNulls(const uint8_t* valid_bytes, int64_t length) {
+    ARROW_RETURN_NOT_OK(Reserve(length));
+    UnsafeAppendToBitmap(valid_bytes, length);
+
+    return Status::OK();
+  }
+
+  Status AppendNull() {
+    ARROW_RETURN_NOT_OK(Reserve(1));
+    UnsafeAppendToBitmap(false);
+
+    return Status::OK();
+  }
+
+  /// Scalar append
+  Status Append(const bool val) {
+    ARROW_RETURN_NOT_OK(Reserve(1));
+    UnsafeAppend(val);
+    return Status::OK();
+  }
+
+  Status Append(const uint8_t val) { return Append(val != 0); }
+
+  /// Scalar append, without checking for capacity
+  void UnsafeAppend(const bool val) {
+    BitUtil::SetBit(null_bitmap_data_, length_);
+    if (val) {
+      BitUtil::SetBit(raw_data_, length_);
+    } else {
+      BitUtil::ClearBit(raw_data_, length_);
+    }
+    ++length_;
+  }
+
+  void UnsafeAppend(const uint8_t val) { UnsafeAppend(val != 0); }
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values a contiguous array of bytes (non-zero is 1)
+  /// \param[in] length the number of values to append
+  /// \param[in] valid_bytes an optional sequence of bytes where non-zero
+  /// indicates a valid (non-null) value
+  /// \return Status
+  Status AppendValues(const uint8_t* values, int64_t length,
+                      const uint8_t* valid_bytes = NULLPTR);
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values a contiguous C array of values
+  /// \param[in] length the number of values to append
+  /// \param[in] is_valid an std::vector<bool> indicating valid (1) or null
+  /// (0). Equal in length to values
+  /// \return Status
+  Status AppendValues(const uint8_t* values, int64_t length,
+                      const std::vector<bool>& is_valid);
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values a std::vector of bytes
+  /// \param[in] is_valid an std::vector<bool> indicating valid (1) or null
+  /// (0). Equal in length to values
+  /// \return Status
+  Status AppendValues(const std::vector<uint8_t>& values,
+                      const std::vector<bool>& is_valid);
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values a std::vector of bytes
+  /// \return Status
+  Status AppendValues(const std::vector<uint8_t>& values);
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values an std::vector<bool> indicating true (1) or false
+  /// \param[in] is_valid an std::vector<bool> indicating valid (1) or null
+  /// (0). Equal in length to values
+  /// \return Status
+  Status AppendValues(const std::vector<bool>& values, const std::vector<bool>& is_valid);
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values an std::vector<bool> indicating true (1) or false
+  /// \return Status
+  Status AppendValues(const std::vector<bool>& values);
+
+  /// \brief Append a sequence of elements in one shot
+  /// \param[in] values_begin InputIterator to the beginning of the values
+  /// \param[in] values_end InputIterator pointing to the end of the values
+  ///  or null(0) values
+  /// \return Status
+  template <typename ValuesIter>
+  Status AppendValues(ValuesIter values_begin, ValuesIter values_end) {
+    int64_t length = static_cast<int64_t>(std::distance(values_begin, values_end));
+    ARROW_RETURN_NOT_OK(Reserve(length));
+    auto iter = values_begin;
+    internal::GenerateBitsUnrolled(raw_data_, length_, length,
+                                   [&iter]() -> bool { return *(iter++); });
+
+    // this updates length_
+    UnsafeSetNotNull(length);
+    return Status::OK();
+  }
+
+  /// \brief Append a sequence of elements in one shot, with a specified nullmap
+  /// \param[in] values_begin InputIterator to the beginning of the values
+  /// \param[in] values_end InputIterator pointing to the end of the values
+  /// \param[in] valid_begin InputIterator with elements indication valid(1)
+  ///  or null(0) values
+  /// \return Status
+  template <typename ValuesIter, typename ValidIter>
+  typename std::enable_if<!std::is_pointer<ValidIter>::value, Status>::type AppendValues(
+      ValuesIter values_begin, ValuesIter values_end, ValidIter valid_begin) {
+    static_assert(!internal::is_null_pointer<ValidIter>::value,
+                  "Don't pass a NULLPTR directly as valid_begin, use the 2-argument "
+                  "version instead");
+    int64_t length = static_cast<int64_t>(std::distance(values_begin, values_end));
+    ARROW_RETURN_NOT_OK(Reserve(length));
+
+    auto iter = values_begin;
+    internal::GenerateBitsUnrolled(raw_data_, length_, length,
+                                   [&iter]() -> bool { return *(iter++); });
+
+    // this updates length_
+    ArrayBuilder::UnsafeAppendToBitmap(valid_begin, std::next(valid_begin, length));
+    return Status::OK();
+  }
+
+  // Same as above, for a pointer type ValidIter
+  template <typename ValuesIter, typename ValidIter>
+  typename std::enable_if<std::is_pointer<ValidIter>::value, Status>::type AppendValues(
+      ValuesIter values_begin, ValuesIter values_end, ValidIter valid_begin) {
+    int64_t length = static_cast<int64_t>(std::distance(values_begin, values_end));
+    ARROW_RETURN_NOT_OK(Reserve(length));
+
+    auto iter = values_begin;
+    internal::GenerateBitsUnrolled(raw_data_, length_, length,
+                                   [&iter]() -> bool { return *(iter++); });
+
+    // this updates the length_
+    if (valid_begin == NULLPTR) {
+      UnsafeSetNotNull(length);
+    } else {
+      UnsafeAppendToBitmap(valid_begin, std::next(valid_begin, length));
+    }
+
+    return Status::OK();
+  }
+
+  Status FinishInternal(std::shared_ptr<ArrayData>* out) override;
+  void Reset() override;
+  Status Resize(int64_t capacity) override;
+
+ protected:
+  std::shared_ptr<ResizableBuffer> data_;
+  uint8_t* raw_data_;
+};
+
+}  // namespace arrow

cpp/src/arrow/csv/column-builder.h

@@ -18,22 +18,29 @@
 #ifndef ARROW_CSV_COLUMN_BUILDER_H
 #define ARROW_CSV_COLUMN_BUILDER_H
 
+#include <cstdint>
 #include <memory>
-#include <vector>
 
 #include "arrow/array.h"
-#include "arrow/csv/converter.h"
-#include "arrow/csv/options.h"
-#include "arrow/memory_pool.h"
 #include "arrow/status.h"
-#include "arrow/table.h"
-#include "arrow/type.h"
-#include "arrow/util/task-group.h"
 #include "arrow/util/visibility.h"
 
 namespace arrow {
+
+class ChunkedArray;
+class DataType;
+
+namespace internal {
+
+class TaskGroup;
+
+}  // namespace internal
+
 namespace csv {
 
+class BlockParser;
+struct ConvertOptions;
+
 class ARROW_EXPORT ColumnBuilder {
  public:
   virtual ~ColumnBuilder() = default;

cpp/src/arrow/csv/converter.cc

@@ -20,6 +20,7 @@
 #include <cstring>
 #include <sstream>
 #include <string>
+#include <type_traits>
 
 #include "arrow/builder.h"
 #include "arrow/csv/parser.h"

cpp/src/arrow/csv/parser.h

@@ -18,6 +18,7 @@
 #ifndef ARROW_CSV_PARSER_H
 #define ARROW_CSV_PARSER_H
 
+#include <cstddef>
 #include <cstdint>
 #include <memory>
 #include <vector>

cpp/src/arrow/csv/reader.cc

@@ -23,6 +23,8 @@
 #include <memory>
 #include <sstream>
 #include <string>
+#include <unordered_map>
+#include <utility>
 #include <vector>
 
 #include "arrow/buffer.h"

cpp/src/arrow/io/buffered.cc

@@ -21,10 +21,10 @@
 #include <cstring>
 #include <memory>
 #include <mutex>
-#include <string>
 #include <utility>
 
 #include "arrow/buffer.h"
+#include "arrow/memory_pool.h"
 #include "arrow/status.h"
 #include "arrow/util/logging.h"
 #include "arrow/util/string_view.h"

cpp/src/arrow/io/buffered.h

@@ -29,6 +29,7 @@
 
 namespace arrow {
 
+class Buffer;
 class MemoryPool;
 class Status;
 

cpp/src/arrow/ipc/feather-test.cc

@@ -30,6 +30,7 @@
 #include "arrow/pretty_print.h"
 #include "arrow/record_batch.h"
 #include "arrow/status.h"
+#include "arrow/table.h"
 #include "arrow/test-util.h"
 #include "arrow/type.h"
 #include "arrow/util/checked_cast.h"

cpp/src/arrow/ipc/json-simple-test.cc

@@ -34,6 +34,7 @@
 #include "arrow/type.h"
 #include "arrow/type_traits.h"
 #include "arrow/util/checked_cast.h"
+#include "arrow/util/decimal.h"
 
 #if defined(_MSC_VER)
 // "warning C4307: '+': integral constant overflow"

cpp/src/arrow/memory_pool-test.h

@@ -16,6 +16,7 @@
 // under the License.
 
 #include <algorithm>
+#include <cstddef>
 #include <cstdint>
 #include <limits>
 

cpp/src/arrow/memory_pool.cc

@@ -17,18 +17,16 @@
 
 #include "arrow/memory_pool.h"
 
-#include <algorithm>
-#include <atomic>
-#include <cerrno>
-#include <cstdlib>
-#include <cstring>
-#include <iostream>
+#include <algorithm>  // IWYU pragma: keep
+#include <cstdlib>    // IWYU pragma: keep
+#include <cstring>    // IWYU pragma: keep
+#include <iostream>   // IWYU pragma: keep
 #include <limits>
 #include <memory>
 #include <sstream>  // IWYU pragma: keep
 
 #include "arrow/status.h"
-#include "arrow/util/logging.h"
+#include "arrow/util/logging.h"  // IWYU pragma: keep
 
 #ifdef ARROW_JEMALLOC
 // Needed to support jemalloc 3 and 4

cpp/src/arrow/pretty_print-test.cc

@@ -26,12 +26,10 @@
 
 #include "arrow/array.h"
 #include "arrow/builder.h"
-#include "arrow/memory_pool.h"
 #include "arrow/pretty_print.h"
 #include "arrow/table.h"
 #include "arrow/test-util.h"
 #include "arrow/type.h"
-#include "arrow/util/decimal.h"
 
 namespace arrow {
 

cpp/src/arrow/pretty_print.cc

@@ -19,7 +19,6 @@
 #include <cstdint>
 #include <iostream>
 #include <memory>
-#include <sstream>
 #include <string>
 #include <type_traits>
 #include <vector>

cpp/src/arrow/pretty_print.h

@@ -21,14 +21,17 @@
 #include <ostream>
 #include <string>
 
-#include "arrow/type_fwd.h"
 #include "arrow/util/visibility.h"
 
 namespace arrow {
 
 class Array;
+class Column;
 class ChunkedArray;
+class RecordBatch;
+class Schema;
 class Status;
+class Table;
 
 struct PrettyPrintOptions {
   PrettyPrintOptions(int indent_arg, int window_arg = 10, int indent_size_arg = 2,

cpp/src/arrow/python/numpy_to_arrow.cc

@@ -539,33 +539,31 @@ Status NumPyConverter::Visit(const BinaryType& type) {
 
   auto data = reinterpret_cast<const uint8_t*>(PyArray_DATA(arr_));
 
-  int item_length = 0;
+  auto AppendNotNull = [&builder, this](const uint8_t* data) {
+    // This is annoying. NumPy allows strings to have nul-terminators, so
+    // we must check for them here
+    int item_length = 0;
+    for (; item_length < itemsize_; ++item_length) {
+      if (data[item_length] == 0) {
+        break;
+      }
+    }
+    return builder.Append(data, item_length);
+  };
+
   if (mask_ != nullptr) {
     Ndarray1DIndexer<uint8_t> mask_values(mask_);
     for (int64_t i = 0; i < length_; ++i) {
       if (mask_values[i]) {
         RETURN_NOT_OK(builder.AppendNull());
       } else {
-        // This is annoying. NumPy allows strings to have nul-terminators, so
-        // we must check for them here
-        for (item_length = 0; item_length < itemsize_; ++item_length) {
-          if (data[item_length] == 0) {
-            break;
-          }
-        }
-        RETURN_NOT_OK(builder.Append(data, item_length));
+        RETURN_NOT_OK(AppendNotNull(data));
       }
       data += stride_;
     }
   } else {
     for (int64_t i = 0; i < length_; ++i) {
-      for (item_length = 0; item_length < itemsize_; ++item_length) {
-        // Look for nul-terminator
-        if (data[item_length] == 0) {
-          break;
-        }
-      }
-      RETURN_NOT_OK(builder.Append(data, item_length));
+      RETURN_NOT_OK(AppendNotNull(data));
       data += stride_;
     }
   }

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

@@ -25,6 +25,7 @@
 #include "arrow/builder.h"
 #include "arrow/table.h"
 #include "arrow/test-util.h"
+#include "arrow/util/decimal.h"
 
 #include "arrow/python/arrow_to_pandas.h"
 #include "arrow/python/decimal.h"

cpp/src/arrow/record_batch.h

@@ -32,6 +32,7 @@ namespace arrow {
 class Array;
 struct ArrayData;
 class Status;
+class Table;
 
 /// \class RecordBatch
 /// \brief Collection of equal-length arrays matching a particular Schema

cpp/src/arrow/tensor.cc

@@ -17,6 +17,7 @@
 
 #include "arrow/tensor.h"
 
+#include <cstddef>
 #include <cstdint>
 #include <functional>
 #include <memory>

cpp/src/arrow/test-util.cc

@@ -18,13 +18,12 @@
 #include "arrow/test-util.h"
 
 #ifndef _WIN32
-#include <sys/stat.h>
-#include <sys/wait.h>
-#include <unistd.h>
+#include <sys/stat.h>  // IWYU pragma: keep
+#include <sys/wait.h>  // IWYU pragma: keep
+#include <unistd.h>    // IWYU pragma: keep
 #endif
 
 #include <algorithm>
-#include <chrono>
 #include <cstdint>
 #include <cstdlib>
 #include <iostream>
@@ -33,23 +32,17 @@
 #include <random>
 #include <sstream>
 #include <string>
-#include <thread>
 #include <vector>
 
 #include <gtest/gtest.h>
 
 #include "arrow/array.h"
 #include "arrow/buffer.h"
-#include "arrow/builder.h"
 #include "arrow/ipc/json-simple.h"
-#include "arrow/memory_pool.h"
 #include "arrow/pretty_print.h"
 #include "arrow/status.h"
 #include "arrow/table.h"
 #include "arrow/type.h"
-#include "arrow/type_traits.h"
-#include "arrow/util/bit-util.h"
-#include "arrow/util/decimal.h"
 #include "arrow/util/logging.h"
 
 namespace arrow {

cpp/src/arrow/test-util.h

@@ -17,23 +17,17 @@
 
 #pragma once
 
-#ifndef _WIN32
-#include <sys/stat.h>
-#include <sys/wait.h>
-#include <unistd.h>
-#endif
-
 #include <algorithm>
-#include <chrono>
 #include <cstdint>
 #include <cstdlib>
+#include <cstring>
 #include <iostream>
 #include <limits>
 #include <memory>
 #include <random>
 #include <sstream>
 #include <string>
-#include <thread>
+#include <type_traits>
 #include <vector>
 
 #include <gtest/gtest.h>
@@ -43,13 +37,13 @@
 #include "arrow/builder.h"
 #include "arrow/memory_pool.h"
 #include "arrow/pretty_print.h"
+#include "arrow/record_batch.h"
 #include "arrow/status.h"
-#include "arrow/table.h"
 #include "arrow/type.h"
 #include "arrow/type_traits.h"
 #include "arrow/util/bit-util.h"
-#include "arrow/util/decimal.h"
 #include "arrow/util/logging.h"
+#include "arrow/util/macros.h"
 #include "arrow/util/visibility.h"
 
 #define STRINGIFY(x) #x
@@ -102,6 +96,10 @@
 
 namespace arrow {
 
+class ChunkedArray;
+class Column;
+class Table;
+
 using ArrayVector = std::vector<std::shared_ptr<Array>>;
 
 #define ASSERT_ARRAYS_EQUAL(LEFT, RIGHT)                                               \

cpp/src/arrow/util/compression_lz4.cc

@@ -18,6 +18,7 @@
 #include "arrow/util/compression_lz4.h"
 
 #include <cstdint>
+#include <cstring>
 #include <sstream>
 
 #include <lz4.h>

cpp/src/arrow/util/int-util-test.cc

@@ -17,14 +17,12 @@
 
 #include <algorithm>
 #include <cstdint>
-#include <memory>
 #include <random>
 #include <utility>
 #include <vector>
 
 #include <gtest/gtest.h>
 
-#include "arrow/test-util.h"
 #include "arrow/util/int-util.h"
 
 namespace arrow {

cpp/src/arrow/util/string_view.h

@@ -18,7 +18,7 @@
 #ifndef ARROW_UTIL_STRING_VIEW_H
 #define ARROW_UTIL_STRING_VIEW_H
 
-#include "arrow/util/string_view/string_view.hpp"
+#include "arrow/util/string_view/string_view.hpp"  // IWYU pragma: export
 
 namespace arrow {
 namespace util {

cpp/src/parquet/arrow/reader.cc

@@ -1387,6 +1387,7 @@ Status PrimitiveImpl::NextBatch(int64_t records_to_read, std::shared_ptr<Array>*
     return ::arrow::Status::IOError(e.what());
   }
 
+  Datum result;
   switch (field_->type()->id()) {
     TRANSFER_CASE(BOOL, ::arrow::BooleanType, BooleanType)
     TRANSFER_CASE(UINT8, ::arrow::UInt8Type, Int32Type)

cpp/src/parquet/arrow/record_reader.cc

@@ -86,14 +86,6 @@ class RecordReader::RecordReaderImpl {
     valid_bits_ = AllocateBuffer(pool);
     def_levels_ = AllocateBuffer(pool);
     rep_levels_ = AllocateBuffer(pool);
-
-    if (descr->physical_type() == Type::BYTE_ARRAY) {
-      builder_.reset(new ::arrow::BinaryBuilder(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));
-    }
     Reset();
   }
 
@@ -229,8 +221,6 @@ class RecordReader::RecordReaderImpl {
     return result;
   }
 
-  ::arrow::ArrayBuilder* builder() { return builder_.get(); }
-
   // Process written repetition/definition levels to reach the end of
   // records. Process no more levels than necessary to delimit the indicated
   // number of logical records. Updates internal state of RecordReader
@@ -375,7 +365,7 @@ class RecordReader::RecordReaderImpl {
 
     records_read_ = 0;
 
-    // Calling Finish on the builders also resets them
+    // Call Finish on the binary builders to reset them
   }
 
   void ResetValues() {
@@ -391,6 +381,8 @@ class RecordReader::RecordReaderImpl {
 
   virtual void DebugPrintState() = 0;
 
+  virtual std::vector<std::shared_ptr<Array>> GetChunks() = 0;
+
  protected:
   virtual bool ReadNewPage() = 0;
 
@@ -434,9 +426,6 @@ class RecordReader::RecordReaderImpl {
   int64_t levels_position_;
   int64_t levels_capacity_;
 
-  // TODO(wesm): ByteArray / FixedLenByteArray types
-  std::unique_ptr<::arrow::ArrayBuilder> builder_;
-
   std::shared_ptr<::arrow::ResizableBuffer> values_;
 
   template <typename T>
@@ -449,13 +438,36 @@ class RecordReader::RecordReaderImpl {
   std::shared_ptr<::arrow::ResizableBuffer> rep_levels_;
 };
 
+template <typename DType>
+struct RecordReaderTraits {
+  using BuilderType = void;
+};
+
+template <>
+struct RecordReaderTraits<ByteArrayType> {
+  using BuilderType = ::arrow::ChunkedBinaryBuilder;
+};
+
 template <typename DType>
 class TypedRecordReader : public RecordReader::RecordReaderImpl {
  public:
-  typedef typename DType::c_type T;
+  using T = typename DType::c_type;
 
-  TypedRecordReader(const ColumnDescriptor* schema, ::arrow::MemoryPool* pool)
-      : RecordReader::RecordReaderImpl(schema, pool), current_decoder_(nullptr) {}
+  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));
+    }
+  }
 
   void ResetDecoders() override { decoders_.clear(); }
 
@@ -546,6 +558,10 @@ class TypedRecordReader : public RecordReader::RecordReaderImpl {
     std::cout << std::endl;
   }
 
+  virtual std::vector<std::shared_ptr<Array>> GetChunks() override {
+    throw ParquetException("GetChunks only implemented for binary types");
+  }
+
  private:
   typedef Decoder<DType> DecoderType;
 
@@ -554,6 +570,8 @@ class TypedRecordReader : public RecordReader::RecordReaderImpl {
   // plain-encoded data.
   std::unordered_map<int, std::shared_ptr<DecoderType>> decoders_;
 
+  std::unique_ptr<BuilderType> builder_;
+
   DecoderType* current_decoder_;
 
   // Advance to the next data page
@@ -572,6 +590,20 @@ void TypedRecordReader<ByteArrayType>::DebugPrintState() {}
 template <>
 void TypedRecordReader<FLBAType>::DebugPrintState() {}
 
+template <>
+::arrow::ArrayVector TypedRecordReader<ByteArrayType>::GetChunks() {
+  ::arrow::ArrayVector chunks;
+  PARQUET_THROW_NOT_OK(builder_->Finish(&chunks));
+  return chunks;
+}
+
+template <>
+::arrow::ArrayVector TypedRecordReader<FLBAType>::GetChunks() {
+  std::shared_ptr<Array> chunk;
+  PARQUET_THROW_NOT_OK(builder_->Finish(&chunk));
+  return ::arrow::ArrayVector({chunk});
+}
+
 template <>
 inline void TypedRecordReader<ByteArrayType>::ReadValuesDense(int64_t values_to_read) {
   auto values = ValuesHead<ByteArray>();
@@ -579,10 +611,9 @@ inline void TypedRecordReader<ByteArrayType>::ReadValuesDense(int64_t values_to_
       current_decoder_->Decode(values, static_cast<int>(values_to_read));
   DCHECK_EQ(num_decoded, values_to_read);
 
-  auto builder = static_cast<::arrow::BinaryBuilder*>(builder_.get());
   for (int64_t i = 0; i < num_decoded; i++) {
     PARQUET_THROW_NOT_OK(
-        builder->Append(values[i].ptr, static_cast<int32_t>(values[i].len)));
+        builder_->Append(values[i].ptr, static_cast<int32_t>(values[i].len)));
   }
   ResetValues();
 }
@@ -594,9 +625,8 @@ inline void TypedRecordReader<FLBAType>::ReadValuesDense(int64_t values_to_read)
       current_decoder_->Decode(values, static_cast<int>(values_to_read));
   DCHECK_EQ(num_decoded, values_to_read);
 
-  auto builder = static_cast<::arrow::FixedSizeBinaryBuilder*>(builder_.get());
   for (int64_t i = 0; i < num_decoded; i++) {
-    PARQUET_THROW_NOT_OK(builder->Append(values[i].ptr));
+    PARQUET_THROW_NOT_OK(builder_->Append(values[i].ptr));
   }
   ResetValues();
 }
@@ -613,14 +643,12 @@ inline void TypedRecordReader<ByteArrayType>::ReadValuesSpaced(int64_t values_to
       valid_bits_offset);
   DCHECK_EQ(num_decoded, values_to_read);
 
-  auto builder = static_cast<::arrow::BinaryBuilder*>(builder_.get());
-
   for (int64_t i = 0; i < num_decoded; i++) {
     if (::arrow::BitUtil::GetBit(valid_bits, valid_bits_offset + i)) {
       PARQUET_THROW_NOT_OK(
-          builder->Append(values[i].ptr, static_cast<int32_t>(values[i].len)));
+          builder_->Append(values[i].ptr, static_cast<int32_t>(values[i].len)));
     } else {
-      PARQUET_THROW_NOT_OK(builder->AppendNull());
+      PARQUET_THROW_NOT_OK(builder_->AppendNull());
     }
   }
   ResetValues();
@@ -638,12 +666,11 @@ inline void TypedRecordReader<FLBAType>::ReadValuesSpaced(int64_t values_to_read
       valid_bits_offset);
   DCHECK_EQ(num_decoded, values_to_read);
 
-  auto builder = static_cast<::arrow::FixedSizeBinaryBuilder*>(builder_.get());
   for (int64_t i = 0; i < num_decoded; i++) {
     if (::arrow::BitUtil::GetBit(valid_bits, valid_bits_offset + i)) {
-      PARQUET_THROW_NOT_OK(builder->Append(values[i].ptr));
+      PARQUET_THROW_NOT_OK(builder_->Append(values[i].ptr));
     } else {
-      PARQUET_THROW_NOT_OK(builder->AppendNull());
+      PARQUET_THROW_NOT_OK(builder_->AppendNull());
     }
   }
   ResetValues();
@@ -845,8 +872,6 @@ std::shared_ptr<ResizableBuffer> RecordReader::ReleaseIsValid() {
   return impl_->ReleaseIsValid();
 }
 
-::arrow::ArrayBuilder* RecordReader::builder() { return impl_->builder(); }
-
 int64_t RecordReader::values_written() const { return impl_->values_written(); }
 
 int64_t RecordReader::levels_position() const { return impl_->levels_position(); }

python/pyarrow/tests/test_parquet.py

@@ -1959,6 +1959,31 @@ def test_large_table_int32_overflow():
     _write_table(table, f)
 
 
+@pytest.mark.large_memory
+def test_binary_array_overflow_to_chunked():
+    # ARROW-3762
+
+    # 2^31 + 1 bytes
+    values = [b'x'] + [
+        b'x' * (1 << 20)
+    ] * 2 * (1 << 10)
+    df = pd.DataFrame({'byte_col': values})
+
+    tbl = pa.Table.from_pandas(df, preserve_index=False)
+
+    buf = io.BytesIO()
+    _write_table(tbl, buf)
+    buf.seek(0)
+    read_tbl = _read_table(buf)
+    buf = None
+
+    col0_data = read_tbl[0].data
+    assert isinstance(col0_data, pa.ChunkedArray)
+    assert col0_data.num_chunks == 2
+
+    assert tbl.equals(read_tbl)
+
+
 def test_index_column_name_duplicate(tempdir):
     data = {
         'close': {