ARROW-17798: [C++][Parquet] Add DELTA_BINARY_PACKED encoder to Parquet writer

cpp/src/arrow/util/bit_stream_utils.h

@@ -203,9 +203,10 @@ class BitReader {
 };
 
 inline bool BitWriter::PutValue(uint64_t v, int num_bits) {
-  // TODO: revisit this limit if necessary (can be raised to 64 by fixing some edge cases)
-  DCHECK_LE(num_bits, 32);
-  DCHECK_EQ(v >> num_bits, 0) << "v = " << v << ", num_bits = " << num_bits;
+  DCHECK_LE(num_bits, 64);
+  if (num_bits < 64) {
+    DCHECK_EQ(v >> num_bits, 0) << "v = " << v << ", num_bits = " << num_bits;
+  }
 
   if (ARROW_PREDICT_FALSE(byte_offset_ * 8 + bit_offset_ + num_bits > max_bytes_ * 8))
     return false;
@@ -220,7 +221,8 @@ inline bool BitWriter::PutValue(uint64_t v, int num_bits) {
     buffered_values_ = 0;
     byte_offset_ += 8;
     bit_offset_ -= 64;
-    buffered_values_ = v >> (num_bits - bit_offset_);
+    buffered_values_ =
+        (num_bits - bit_offset_ == 64) ? 0 : (v >> (num_bits - bit_offset_));
   }
   DCHECK_LT(bit_offset_, 64);
   return true;

cpp/src/arrow/util/rle_encoding_test.cc

@@ -173,6 +173,40 @@ TEST(BitArray, TestMixed) {
   }
 }
 
+// Write up to 'num_vals' values with width 'bit_width' and reads them back.
+static void TestPutValue(int bit_width, uint64_t num_vals) {
+  // The max value representable in `bit_width` bits.
+  const uint64_t max = std::numeric_limits<uint64_t>::max() >> (64 - bit_width);
+  num_vals = std::min(num_vals, max);
+  int len = static_cast<int>(bit_util::BytesForBits(bit_width * num_vals));
+  EXPECT_GT(len, 0);
+
+  std::vector<uint8_t> buffer(len);
+  bit_util::BitWriter writer(buffer.data(), len);
+  for (uint64_t i = max - num_vals; i < max; i++) {
+    bool result = writer.PutValue(i, bit_width);
+    EXPECT_TRUE(result);
+  }
+  writer.Flush();
+  EXPECT_EQ(writer.bytes_written(), len);
+
+  bit_util::BitReader reader(buffer.data(), len);
+  for (uint64_t i = max - num_vals; i < max; i++) {
+    int64_t val = 0;
+    bool result = reader.GetValue(bit_width, &val);
+    EXPECT_TRUE(result);
+    EXPECT_EQ(val, i);
+  }
+  EXPECT_EQ(reader.bytes_left(), 0);
+}
+
+TEST(BitUtil, RoundTripIntValues) {
+  for (int width = 1; width < 64; width++) {
+    TestPutValue(width, 1);
+    TestPutValue(width, 1024);
+  }
+}
+
 // Validates encoding of values by encoding and decoding them.  If
 // expected_encoding != NULL, also validates that the encoded buffer is
 // exactly 'expected_encoding'.

cpp/src/parquet/column_writer_test.cc

@@ -400,7 +400,8 @@ typedef ::testing::Types<Int32Type, Int64Type, Int96Type, FloatType, DoubleType,
 
 TYPED_TEST_SUITE(TestPrimitiveWriter, TestTypes);
 
-using TestNullValuesWriter = TestPrimitiveWriter<Int32Type>;
+using TestValuesWriterInt32Type = TestPrimitiveWriter<Int32Type>;
+using TestValuesWriterInt64Type = TestPrimitiveWriter<Int64Type>;
 
 TYPED_TEST(TestPrimitiveWriter, RequiredPlain) {
   this->TestRequiredWithEncoding(Encoding::PLAIN);
@@ -418,23 +419,29 @@ TYPED_TEST(TestPrimitiveWriter, RequiredRLE) {
 TYPED_TEST(TestPrimitiveWriter, RequiredBitPacked) {
   this->TestRequiredWithEncoding(Encoding::BIT_PACKED);
 }
+*/
+
+TEST_F(TestValuesWriterInt32Type, RequiredDeltaBinaryPacked) {
+  this->TestRequiredWithEncoding(Encoding::DELTA_BINARY_PACKED);
+}
 
-TYPED_TEST(TestPrimitiveWriter, RequiredDeltaBinaryPacked) {
+TEST_F(TestValuesWriterInt64Type, RequiredDeltaBinaryPacked) {
   this->TestRequiredWithEncoding(Encoding::DELTA_BINARY_PACKED);
 }
 
+/*
 TYPED_TEST(TestPrimitiveWriter, RequiredDeltaLengthByteArray) {
   this->TestRequiredWithEncoding(Encoding::DELTA_LENGTH_BYTE_ARRAY);
 }
 
 TYPED_TEST(TestPrimitiveWriter, RequiredDeltaByteArray) {
   this->TestRequiredWithEncoding(Encoding::DELTA_BYTE_ARRAY);
 }
+*/
 
 TYPED_TEST(TestPrimitiveWriter, RequiredRLEDictionary) {
   this->TestRequiredWithEncoding(Encoding::RLE_DICTIONARY);
 }
-*/
 
 TYPED_TEST(TestPrimitiveWriter, RequiredPlainWithStats) {
   this->TestRequiredWithSettings(Encoding::PLAIN, Compression::UNCOMPRESSED, false, true,
@@ -647,7 +654,7 @@ TEST(TestWriter, NullValuesBuffer) {
 
 // PARQUET-719
 // Test case for NULL values
-TEST_F(TestNullValuesWriter, OptionalNullValueChunk) {
+TEST_F(TestValuesWriterInt32Type, OptionalNullValueChunk) {
   this->SetUpSchema(Repetition::OPTIONAL);
 
   this->GenerateData(LARGE_SIZE);