GH-55503: [C++] Fix Float16 memo table keys built from raw bits

cpp/src/arrow/array/array_dict.cc

@@ -21,6 +21,7 @@
 #include <climits>
 #include <cstdint>
 #include <limits>
+#include <type_traits>
 #include <utility>
 #include <vector>
 
@@ -39,6 +40,7 @@
 #include "arrow/util/bit_util.h"
 #include "arrow/util/bitmap_ops.h"
 #include "arrow/util/checked_cast.h"
+#include "arrow/util/float16.h"
 #include "arrow/util/int_util.h"
 #include "arrow/util/logging_internal.h"
 #include "arrow/visit_type_inline.h"
@@ -358,18 +360,27 @@ class DictionaryUnifierImpl : public DictionaryUnifier {
                              dictionary.type()->ToString());
     }
     const ArrayType& values = checked_cast<const ArrayType&>(dictionary);
+    // The half float memo table is keyed on `util::Float16`, whose converting
+    // constructor would read the raw uint16_t storage value as a number.
+    auto get_key = [&](int64_t i) {
+      if constexpr (std::is_same_v<T, HalfFloatType>) {
+        return ::arrow::util::Float16::FromBits(values.GetView(i));
+      } else {
+        return values.GetView(i);
+      }
+    };
     if (out != nullptr) {
       ARROW_ASSIGN_OR_RAISE(auto result,
                             AllocateBuffer(dictionary.length() * sizeof(int32_t), pool_));
       auto result_raw = reinterpret_cast<int32_t*>(result->mutable_data());
       for (int64_t i = 0; i < values.length(); ++i) {
-        RETURN_NOT_OK(memo_table_.GetOrInsert(values.GetView(i), &result_raw[i]));
+        RETURN_NOT_OK(memo_table_.GetOrInsert(get_key(i), &result_raw[i]));
       }
       *out = std::move(result);
     } else {
       for (int64_t i = 0; i < values.length(); ++i) {
         int32_t unused_memo_index;
-        RETURN_NOT_OK(memo_table_.GetOrInsert(values.GetView(i), &unused_memo_index));
+        RETURN_NOT_OK(memo_table_.GetOrInsert(get_key(i), &unused_memo_index));
       }
     }
     return Status::OK();

cpp/src/arrow/array/array_dict_test.cc

@@ -1655,6 +1655,29 @@ TEST(TestDictionaryUnifier, Numeric) {
   CheckTransposeMap(*b3, {3, 5});
 }
 
+TEST(TestDictionaryUnifier, HalfFloat) {
+  // The unified dictionary must keep the values, not their bit patterns read as numbers
+  auto dict_ty = float16();
+  auto d1 = ArrayFromJSON(dict_ty, "[1.0, 2.5, 65504.0]");
+  auto d2 = ArrayFromJSON(dict_ty, "[0.5, 65504.0, 1.0]");
+  auto expected = dictionary(int8(), dict_ty);
+  auto expected_dict = ArrayFromJSON(dict_ty, "[1.0, 2.5, 65504.0, 0.5]");
+
+  ASSERT_OK_AND_ASSIGN(auto unifier, DictionaryUnifier::Make(dict_ty));
+  std::shared_ptr<Buffer> b1, b2;
+  ASSERT_OK(unifier->Unify(*d1, &b1));
+  ASSERT_OK(unifier->Unify(*d2, &b2));
+
+  std::shared_ptr<DataType> out_type;
+  std::shared_ptr<Array> out_dict;
+  ASSERT_OK(unifier->GetResult(&out_type, &out_dict));
+  ASSERT_TRUE(out_type->Equals(*expected));
+  AssertArraysEqual(*expected_dict, *out_dict, /*verbose=*/true);
+
+  CheckTransposeMap(*b1, {0, 1, 2});
+  CheckTransposeMap(*b2, {3, 2, 0});
+}
+
 TEST(TestDictionaryUnifier, String) {
   auto dict_ty = utf8();
 

cpp/src/arrow/compute/kernels/aggregate_basic.cc

@@ -172,7 +172,12 @@ struct CountDistinctImpl : public ScalarAggregator {
       auto visit_null = []() { return Status::OK(); };
       auto visit_value = [&](VisitorArgType arg) {
         int32_t y;
-        return memo_table_->GetOrInsert(arg, &y);
+        if constexpr (std::is_same_v<Type, HalfFloatType>) {
+          // `arg` holds the raw bits, don't let `Float16` read them as a number.
+          return memo_table_->GetOrInsert(::arrow::util::Float16::FromBits(arg), &y);
+        } else {
+          return memo_table_->GetOrInsert(arg, &y);
+        }
       };
       RETURN_NOT_OK(VisitArraySpanInline<Type>(arr, visit_value, visit_null));
     } else {

cpp/src/arrow/compute/kernels/aggregate_test.cc

@@ -1194,6 +1194,15 @@ TEST_F(TestCountDistinctKernel, AllScalarTypesWithNulls) {
   Check(decimal256(13, 3), sample);
 }
 
+TEST_F(TestCountDistinctKernel, HalfFloat) {
+  // 1.0, 1.0009765625 and 1.001953125 have the consecutive bit patterns
+  // 0x3C00, 0x3C01 and 0x3C02 and must be counted as distinct values.
+  Check(float16(), "[1.0, 1.0009765625, 1.001953125, 1.0, null]", 4);
+  Check(float16(), "[1.0, 1.0009765625, 1.001953125, 1.0]", 3, /*has_nulls=*/false);
+  CheckChunkedArr(float16(), {"[1.0, 1.0009765625]", "[1.001953125, null]"}, 4);
+  Check(float16(), "1.5");
+}
+
 TEST_F(TestCountDistinctKernel, Random) {
   UInt32Builder builder;
   std::unordered_set<uint32_t> memo;

cpp/src/arrow/compute/kernels/codegen_internal.h

@@ -432,7 +432,7 @@ template <>
 struct UnboxScalar<HalfFloatType> {
   using T = Float16;
   static T Unbox(const Scalar& val) {
-    return T(checked_cast<const HalfFloatScalar&>(val).value);
+    return T::FromBits(checked_cast<const HalfFloatScalar&>(val).value);
   }
 };