GH-45876: [C++][Acero] Make NormalizeTime order-preserving

cpp/src/arrow/acero/CMakeLists.txt

@@ -187,7 +187,8 @@ add_arrow_acero_test(sorted_merge_node_test SOURCES sorted_merge_node_test.cc)
 add_arrow_acero_test(tpch_node_test SOURCES tpch_node_test.cc)
 add_arrow_acero_test(union_node_test SOURCES union_node_test.cc)
 add_arrow_acero_test(aggregate_node_test SOURCES aggregate_node_test.cc)
-add_arrow_acero_test(util_test SOURCES util_test.cc task_util_test.cc)
+add_arrow_acero_test(util_test SOURCES util_test.cc task_util_test.cc
+                     time_series_util_test.cc)
 add_arrow_acero_test(hash_aggregate_test SOURCES hash_aggregate_test.cc)
 
 add_arrow_acero_test(test_util_internal_test SOURCES test_util_internal_test.cc)

cpp/src/arrow/acero/asof_join_node_test.cc

@@ -1824,5 +1824,65 @@ TEST(AsofJoinTest, OneSideTsAllGreaterThanTheOther) {
   }
 }
 
+// GH-45876: NormalizeTime must preserve order across the epoch.
+//
+// A time column whose values cross zero used to appear to jump to ~2^63 and
+// then fall back to small values, because the negative half of the domain was
+// folded onto the non-negative half. The plan failed with an out-of-order
+// error.
+TEST(AsofJoinTest, TimesStraddlingEpochAreOrdered) {
+  auto left_schema = arrow::schema({field("on", int64())});
+  auto right_schema = arrow::schema({field("on", int64()), field("v", int64())});
+
+  auto l_on = ArrayFromJSON(int64(), "[-1000, 0, 1000]");
+  auto r_on = ArrayFromJSON(int64(), "[-1000, 0, 1000]");
+  auto r_v = ArrayFromJSON(int64(), "[1, 2, 3]");
+
+  ExecBatch left_batch({l_on}, l_on->length());
+  ExecBatch right_batch({r_on, r_v}, r_on->length());
+  ExecBatch exp_batch({l_on, r_v}, l_on->length());
+
+  AsofJoinNodeOptions opts({{{"on"}, {}}, {{"on"}, {}}}, /*tolerance=*/0);
+  auto left = Declaration("exec_batch_source",
+                          ExecBatchSourceNodeOptions(left_schema, {left_batch}));
+  auto right = Declaration("exec_batch_source",
+                           ExecBatchSourceNodeOptions(right_schema, {right_batch}));
+  auto asof_join = Declaration{"asofjoin", {left, right}, opts};
+  ASSERT_OK_AND_ASSIGN(auto result, DeclarationToExecBatches(std::move(asof_join)));
+  AssertExecBatchesEqualIgnoringOrder(result.schema, {exp_batch}, result.batches);
+}
+
+// GH-45876: a tolerance window that straddles the epoch must accept the same
+// rows it would accept anywhere else.
+//
+// This is the quieter half of the bug. TolType::Accepts compares differences of
+// normalized values, and under the old encoding differences were exact only
+// when both operands had the same sign. So a right-side row within tolerance
+// of a left-side row was silently dropped when the pair straddled zero, with
+// no error raised. Each side is monotone on its own here, so the ordering
+// check never fires and only the tolerance arithmetic is under test.
+TEST(AsofJoinTest, ToleranceWindowStraddlingEpoch) {
+  auto left_schema = arrow::schema({field("on", int64())});
+  auto right_schema = arrow::schema({field("on", int64()), field("v", int64())});
+
+  // 60 apart, on opposite sides of the epoch, with a backward tolerance of 60.
+  auto l_on = ArrayFromJSON(int64(), "[30]");
+  auto r_on = ArrayFromJSON(int64(), "[-30]");
+  auto r_v = ArrayFromJSON(int64(), "[7]");
+
+  ExecBatch left_batch({l_on}, l_on->length());
+  ExecBatch right_batch({r_on, r_v}, r_on->length());
+  ExecBatch exp_batch({l_on, r_v}, l_on->length());
+
+  AsofJoinNodeOptions opts({{{"on"}, {}}, {{"on"}, {}}}, /*tolerance=*/-60);
+  auto left = Declaration("exec_batch_source",
+                          ExecBatchSourceNodeOptions(left_schema, {left_batch}));
+  auto right = Declaration("exec_batch_source",
+                           ExecBatchSourceNodeOptions(right_schema, {right_batch}));
+  auto asof_join = Declaration{"asofjoin", {left, right}, opts};
+  ASSERT_OK_AND_ASSIGN(auto result, DeclarationToExecBatches(std::move(asof_join)));
+  AssertExecBatchesEqualIgnoringOrder(result.schema, {exp_batch}, result.batches);
+}
+
 }  // namespace acero
 }  // namespace arrow

cpp/src/arrow/acero/meson.build

@@ -115,7 +115,9 @@ arrow_acero_tests = {
     'tpch-node-test': {'sources': ['tpch_node_test.cc']},
     'union-node-test': {'sources': ['union_node_test.cc']},
     'aggregate-node-test': {'sources': ['aggregate_node_test.cc']},
-    'util-test': {'sources': ['util_test.cc', 'task_util_test.cc']},
+    'util-test': {
+        'sources': ['util_test.cc', 'task_util_test.cc', 'time_series_util_test.cc'],
+    },
     'hash-aggregate-test': {'sources': ['hash_aggregate_test.cc']},
     'test-util-internal-test': {'sources': ['test_util_internal_test.cc']},
 }

cpp/src/arrow/acero/time_series_util.cc

@@ -22,13 +22,6 @@
 
 namespace arrow::acero {
 
-template <typename T, enable_if_t<std::is_integral<T>::value, bool>>
-inline uint64_t NormalizeTime(T t) {
-  uint64_t bias =
-      std::is_signed<T>::value ? static_cast<uint64_t>(1) << (8 * sizeof(T) - 1) : 0;
-  return t < 0 ? static_cast<uint64_t>(t + bias) : static_cast<uint64_t>(t);
-}
-
 uint64_t GetTime(const RecordBatch* batch, Type::type time_type, int col, uint64_t row) {
 #define LATEST_VAL_CASE(id, val)                     \
   case Type::id: {                                   \

cpp/src/arrow/acero/time_series_util.h

@@ -22,9 +22,26 @@
 
 namespace arrow::acero {
 
-// normalize the value to unsigned 64-bits while preserving ordering of values
+// Map an integral time value onto uint64_t, preserving order.
+//
+// For a signed T this flips the sign bit within T's own width, which is a
+// strictly increasing bijection from T onto the unsigned type of the same
+// width, and then zero-extends. Biasing after conversion to uint64_t would
+// not work: a negative t is sign-extended first, so `t + (1 << (W - 1))`
+// wraps back into [0, 2^(W-1)) and collides with the non-negative values.
+//
+// Order is preserved exactly, so differences within a type are exact too,
+// which is what the as-of join tolerance comparisons rely on.
 template <typename T, enable_if_t<std::is_integral<T>::value, bool> = true>
-uint64_t NormalizeTime(T t);
+inline uint64_t NormalizeTime(T t) {
+  using Unsigned = typename std::make_unsigned<T>::type;
+  auto normalized = static_cast<Unsigned>(t);
+  if (std::is_signed<T>::value) {
+    normalized = static_cast<Unsigned>(normalized ^
+                                       (static_cast<Unsigned>(1) << (8 * sizeof(T) - 1)));
+  }
+  return static_cast<uint64_t>(normalized);
+}
 
 uint64_t GetTime(const RecordBatch* batch, Type::type time_type, int col, uint64_t row);
 

cpp/src/arrow/acero/time_series_util_test.cc

@@ -0,0 +1,155 @@
+// 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/acero/time_series_util.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <limits>
+#include <type_traits>
+#include <unordered_set>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+namespace arrow {
+namespace acero {
+namespace {
+
+// Sweep every value of T in increasing order and assert that NormalizeTime is
+// strictly increasing and injective over the whole domain. Only usable for
+// types narrow enough to enumerate.
+template <typename T>
+void AssertOrderPreservedExhaustively() {
+  using Wide = int64_t;
+  constexpr Wide kMin = static_cast<Wide>(std::numeric_limits<T>::min());
+  constexpr Wide kMax = static_cast<Wide>(std::numeric_limits<T>::max());
+
+  std::unordered_set<uint64_t> seen;
+  uint64_t previous = 0;
+  bool first = true;
+  for (Wide value = kMin; value <= kMax; ++value) {
+    const T t = static_cast<T>(value);
+    const uint64_t normalized = NormalizeTime(t);
+    if (!first) {
+      ASSERT_GT(normalized, previous)
+          << "not strictly increasing at " << value << " (width " << 8 * sizeof(T)
+          << ", signed " << std::is_signed<T>::value << ")";
+    }
+    ASSERT_TRUE(seen.insert(normalized).second)
+        << "collision at " << value << " -> " << normalized;
+    previous = normalized;
+    first = false;
+  }
+  ASSERT_EQ(seen.size(), static_cast<size_t>(kMax - kMin + 1));
+}
+
+// Values around the extremes and around zero, sorted and deduplicated. Zero
+// is the interesting one: it is where the sign bit changes, and where the
+// previous implementation folded the negative half of the domain back onto the
+// non-negative half. Candidates outside T's range are skipped rather than
+// wrapped, so the sweep is strictly increasing by construction for every T.
+template <typename T>
+std::vector<T> BoundarySweep() {
+  constexpr T kMin = std::numeric_limits<T>::min();
+  constexpr T kMax = std::numeric_limits<T>::max();
+
+  std::vector<T> values = {kMin,
+                           static_cast<T>(kMin + 1),
+                           static_cast<T>(kMin / 2),
+                           static_cast<T>(0),
+                           static_cast<T>(1),
+                           static_cast<T>(2),
+                           static_cast<T>(kMax / 2),
+                           static_cast<T>(kMax - 1),
+                           kMax};
+  if (std::is_signed<T>::value) {
+    values.push_back(static_cast<T>(-1));
+    values.push_back(static_cast<T>(-2));
+  }
+  // Only representable if T is wide enough; int8_t would wrap.
+  if (static_cast<int64_t>(kMax) >= 1000) {
+    values.push_back(static_cast<T>(1000));
+    if (std::is_signed<T>::value) values.push_back(static_cast<T>(-1000));
+  }
+
+  std::sort(values.begin(), values.end());
+  values.erase(std::unique(values.begin(), values.end()), values.end());
+  return values;
+}
+
+template <typename T>
+void AssertOrderPreservedAtBoundaries() {
+  const std::vector<T> values = BoundarySweep<T>();
+  ASSERT_GE(values.size(), 2u);
+  for (size_t i = 1; i < values.size(); ++i) {
+    ASSERT_LT(values[i - 1], values[i]) << "test data is not sorted";
+    ASSERT_LT(NormalizeTime(values[i - 1]), NormalizeTime(values[i]))
+        << "order not preserved between " << static_cast<int64_t>(values[i - 1])
+        << " and " << static_cast<int64_t>(values[i]);
+  }
+}
+
+}  // namespace
+
+TEST(NormalizeTime, OrderPreservedExhaustively) {
+  AssertOrderPreservedExhaustively<int8_t>();
+  AssertOrderPreservedExhaustively<uint8_t>();
+  AssertOrderPreservedExhaustively<int16_t>();
+  AssertOrderPreservedExhaustively<uint16_t>();
+}
+
+TEST(NormalizeTime, OrderPreservedAtBoundaries) {
+  AssertOrderPreservedAtBoundaries<int8_t>();
+  AssertOrderPreservedAtBoundaries<uint8_t>();
+  AssertOrderPreservedAtBoundaries<int16_t>();
+  AssertOrderPreservedAtBoundaries<uint16_t>();
+  AssertOrderPreservedAtBoundaries<int32_t>();
+  AssertOrderPreservedAtBoundaries<uint32_t>();
+  AssertOrderPreservedAtBoundaries<int64_t>();
+  AssertOrderPreservedAtBoundaries<uint64_t>();
+}
+
+// The most negative value must land at zero and the most positive at the top
+// of the range, so that the full width of uint64_t is used for 64-bit inputs
+// and TolType's saturation at kMinValue/kMaxValue keeps working.
+TEST(NormalizeTime, EndpointsMapToEndpoints) {
+  ASSERT_EQ(NormalizeTime(std::numeric_limits<int64_t>::min()), 0u);
+  ASSERT_EQ(NormalizeTime(std::numeric_limits<int64_t>::max()),
+            std::numeric_limits<uint64_t>::max());
+  ASSERT_EQ(NormalizeTime(std::numeric_limits<uint64_t>::min()), 0u);
+  ASSERT_EQ(NormalizeTime(std::numeric_limits<uint64_t>::max()),
+            std::numeric_limits<uint64_t>::max());
+}
+
+// TolType::Accepts compares differences of normalized values, so a difference
+// that spans zero has to come out exact. Under the previous implementation the
+// negative and non-negative halves were both mapped into [0, 2^(W-1)), which
+// made differences across zero wrong by about 2^(W-1).
+TEST(NormalizeTime, DifferencesAcrossZeroAreExact) {
+  for (int64_t delta : {int64_t{1}, int64_t{2}, int64_t{1000}, int64_t{1} << 32}) {
+    const int64_t left = -delta / 2;
+    const int64_t right = left + delta;
+    ASSERT_EQ(NormalizeTime(right) - NormalizeTime(left), static_cast<uint64_t>(delta))
+        << "difference across zero is wrong for delta " << delta;
+  }
+  // Same property for a narrower type, where the fold was easier to hit.
+  ASSERT_EQ(NormalizeTime(int32_t{500}) - NormalizeTime(int32_t{-500}), 1000u);
+}
+
+}  // namespace acero
+}  // namespace arrow