[C++][Acero] materializeColumn for boolean type overflow

[waruto210]

Describe the bug, including details regarding any error messages, version, and platform.

branch: main
commit sha: fcfb33d0b6f7d368bcd00bb661d35c71885c5140

/// file: cpp/src/arrow/acero/unmaterialized_table_internal.h:175

 template <class Type, class Builder = typename TypeTraits<Type>::BuilderType>
 enable_if_boolean<Type, Status> static BuilderAppend(
 Builder& builder, const std::shared_ptr<ArrayData>& source, uint64_t row) {
 if (source->IsNull(row)) {
 builder.UnsafeAppendNull();
 return Status::OK();
 }
 builder.UnsafeAppend(bit_util::GetBit(source->template GetValues<uint8_t>(1), row));
 return Status::OK();
 }

/// cpp/src/arrow/array/data.h:271
 template <typename T>
 inline const T* GetValues(int i) const {
 return GetValues<T>(i, offset);
 }

Here, uint8_t type is used for GetValues, and the underlying buffer applies a logical offset. However, since the boolean type is actually stored bitwise, this may lead to a segmentation fault or incorrect sorting results.
I added the following unit test, which can verify that the result of sortd_merge is incorrect.

TEST(SortedMergeNode, TestSortedMergeTwoInputsWithBool) {
 const int64_t row_count = (16 << 10); // 16k rows per input

 // Create schema with int column A and bool column B
 auto test_schema = arrow::schema(
 {arrow::field("col_a", arrow::int32()), arrow::field("col_b", arrow::boolean())});

 // Helper lambda to create table with specific pattern
 auto create_test_source = [&](int64_t cnt, int offset) -> arrow::Result<Declaration> {
 // Create column A (int) - values from offset to offset+cnt-1
 arrow::Int32Builder col_a_builder;
 std::vector<int32_t> col_a_values;
 col_a_values.reserve(cnt);
 for (int64_t i = 0; i < cnt; ++i) {
 col_a_values.push_back(static_cast<int32_t>(offset + i));
 }
 ARROW_RETURN_NOT_OK(col_a_builder.AppendValues(col_a_values));
 std::shared_ptr<arrow::Array> col_a_arr;
 ARROW_RETURN_NOT_OK(col_a_builder.Finish(&col_a_arr));

 // Create column B (bool) - pattern: true if col_a % 5 == 0, false otherwise
 arrow::BooleanBuilder col_b_builder;
 for (int64_t i = 0; i < cnt; ++i) {
 int32_t a_value = offset + i;
 bool b_value = (a_value % 5 == 0);
 ARROW_RETURN_NOT_OK(col_b_builder.Append(b_value));
 }
 std::shared_ptr<arrow::Array> col_b_arr;
 ARROW_RETURN_NOT_OK(col_b_builder.Finish(&col_b_arr));

 auto table = arrow::Table::Make(test_schema, {col_a_arr, col_b_arr});
 auto table_source =
 Declaration("table_source", TableSourceNodeOptions(table, row_count / 16));
 return table_source;
 };

 ASSERT_OK_AND_ASSIGN(auto source1, create_test_source(row_count, 0));
 ASSERT_OK_AND_ASSIGN(auto source2, create_test_source(row_count, 8192));

 // Create sorted merge by column A
 auto ops = OrderByNodeOptions(compute::Ordering({compute::SortKey("col_a")}));
 Declaration sorted_merge{"sorted_merge", {source1, source2}, ops};

 // Execute plan and collect result
 ASSERT_OK_AND_ASSIGN(auto result_table,
 arrow::acero::DeclarationToTable(sorted_merge, false));

 ASSERT_TRUE(result_table != nullptr);

 // Verify results
 auto col_a = result_table->GetColumnByName("col_a");
 auto col_b = result_table->GetColumnByName("col_b");
 ASSERT_TRUE(col_a != nullptr);
 ASSERT_TRUE(col_b != nullptr);

 // Verify sorting and bool values
 int32_t last_a_value = std::numeric_limits<int32_t>::min();
 int64_t total_rows_checked = 0;

 for (int i = 0; i < col_a->num_chunks(); i++) {
 auto a_chunk = std::static_pointer_cast<arrow::Int32Array>(col_a->chunk(i));
 auto b_chunk = std::static_pointer_cast<arrow::BooleanArray>(col_b->chunk(i));

 ASSERT_EQ(a_chunk->length(), b_chunk->length())
 << "Column A and B must have same length in chunk " << i;

 for (int64_t j = 0; j < a_chunk->length(); j++) {
 ASSERT_FALSE(a_chunk->IsNull(j)) << "Column A should not have null values";
 ASSERT_FALSE(b_chunk->IsNull(j)) << "Column B should not have null values";

 int32_t a_value = a_chunk->Value(j);
 bool b_value = b_chunk->Value(j);

 // Verify sorting by column A
 ASSERT_GE(a_value, last_a_value)
 << "Values not sorted at chunk " << i << ", row " << j
 << ": current=" << a_value << ", last=" << last_a_value;
 last_a_value = a_value;

 // Verify bool value correctness: should be true if a_value % 3 == 0
 bool expected_b_value = (a_value % 5 == 0);
 ASSERT_EQ(b_value, expected_b_value)
 << "Bool value incorrect at chunk " << i << ", row " << j
 << ": col_a=" << a_value << ", col_b=" << b_value
 << ", expected=" << expected_b_value;
 total_rows_checked++;
 }
 }

 ASSERT_EQ(last_a_value, 24575);

 ASSERT_EQ(total_rows_checked, row_count * 2)
 << "Expected " << row_count << " unique rows after merge";
}

Component(s)

C++