AVRO-4228: [c++] Fix BinaryDecoder::arrayNext() to handle negative block counts (#3646)

* AVRO-4228: Fix BinaryDecoder::arrayNext() to handle negative block counts

* AVRO-4228: Add test for arrayNext() with negative block counts

* AVRO-4228: Move negative block count to second block in test

* AVRO-4228: Avoid undefined behavior when negating INT64_MIN in doDecodeItemCount

---------

Co-authored-by: Gabriel Feyer <gabriel.feyer@indexexchange.com>

Author: gfeyer

lang/c++/impl/BinaryDecoder.cc

@@ -164,13 +164,13 @@ size_t BinaryDecoder::doDecodeItemCount() {
     auto result = doDecodeLong();
     if (result < 0) {
         doDecodeLong();
-        return static_cast<size_t>(-result);
+        return static_cast<size_t>(-(result + 1)) + 1;
     }
     return static_cast<size_t>(result);
 }
 
 size_t BinaryDecoder::arrayNext() {
-    return static_cast<size_t>(doDecodeLong());
+    return doDecodeItemCount();
 }
 
 size_t BinaryDecoder::skipArray() {

lang/c++/test/CodecTests.cc

@@ -2100,6 +2100,40 @@ static void testJsonCodecReinit() {
     }
 }
 
+static void testArrayNegativeBlockCount() {
+    // Array of ints [10, 20, 30, 40, 50] encoded with a negative block count
+    // in the second block, which exercises arrayNext().
+    // Per the Avro spec, a negative count means: abs(count) items follow,
+    // preceded by a long byte-size of the block.
+    //
+    // Block 1: count=2, items: 10, 20              (read by arrayStart)
+    // Block 2: count=-3, bytesize=3, items: 30, 40, 50  (read by arrayNext)
+    // Terminal: count=0
+    const uint8_t data[] = {
+        0x04,                    // zigzag(2) = 4: block count = 2
+        0x14, 0x28,             // zigzag ints: 10, 20
+        0x05,                    // zigzag(-3) = 5: block count = -3
+        0x06,                    // zigzag(3) = 6: byte-size of block
+        0x3c, 0x50, 0x64,       // zigzag ints: 30, 40, 50
+        0x00                     // terminal
+    };
+
+    InputStreamPtr is = memoryInputStream(data, sizeof(data));
+    DecoderPtr d = binaryDecoder();
+    d->init(*is);
+
+    std::vector<int32_t> result;
+    for (size_t n = d->arrayStart(); n != 0; n = d->arrayNext()) {
+        for (size_t i = 0; i < n; ++i) {
+            result.push_back(d->decodeInt());
+        }
+    }
+
+    const std::vector<int32_t> expected = {10, 20, 30, 40, 50};
+    BOOST_CHECK_EQUAL_COLLECTIONS(result.begin(), result.end(),
+                                  expected.begin(), expected.end());
+}
+
 static void testByteCount() {
     OutputStreamPtr os1 = memoryOutputStream();
     EncoderPtr e1 = binaryEncoder();
@@ -2125,6 +2159,7 @@ init_unit_test_suite(int, char *[]) {
     ts->add(BOOST_PARAM_TEST_CASE(&avro::testJson, avro::jsonData,
                                   ENDOF(avro::jsonData)));
     ts->add(BOOST_TEST_CASE(avro::testJsonCodecReinit));
+    ts->add(BOOST_TEST_CASE(avro::testArrayNegativeBlockCount));
     ts->add(BOOST_TEST_CASE(avro::testByteCount));
 
     return ts;