diff --git a/cpp/src/arrow/acero/CMakeLists.txt b/cpp/src/arrow/acero/CMakeLists.txt index dc18afa97970..795801b9c1d8 100644 --- a/cpp/src/arrow/acero/CMakeLists.txt +++ b/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) diff --git a/cpp/src/arrow/acero/asof_join_node_test.cc b/cpp/src/arrow/acero/asof_join_node_test.cc index 59a9b4ebba12..579fa08c5206 100644 --- a/cpp/src/arrow/acero/asof_join_node_test.cc +++ b/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 diff --git a/cpp/src/arrow/acero/meson.build b/cpp/src/arrow/acero/meson.build index 9f34377024b4..de9bb91d2776 100644 --- a/cpp/src/arrow/acero/meson.build +++ b/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']}, } diff --git a/cpp/src/arrow/acero/time_series_util.cc b/cpp/src/arrow/acero/time_series_util.cc index 60f9044d7bf7..9fbf54666373 100644 --- a/cpp/src/arrow/acero/time_series_util.cc +++ b/cpp/src/arrow/acero/time_series_util.cc @@ -22,13 +22,6 @@ namespace arrow::acero { -template ::value, bool>> -inline uint64_t NormalizeTime(T t) { - uint64_t bias = - std::is_signed::value ? static_cast(1) << (8 * sizeof(T) - 1) : 0; - return t < 0 ? static_cast(t + bias) : static_cast(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: { \ diff --git a/cpp/src/arrow/acero/time_series_util.h b/cpp/src/arrow/acero/time_series_util.h index 97707f43bf20..96607ea7f73c 100644 --- a/cpp/src/arrow/acero/time_series_util.h +++ b/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 ::value, bool> = true> -uint64_t NormalizeTime(T t); +inline uint64_t NormalizeTime(T t) { + using Unsigned = typename std::make_unsigned::type; + auto normalized = static_cast(t); + if (std::is_signed::value) { + normalized = static_cast(normalized ^ + (static_cast(1) << (8 * sizeof(T) - 1))); + } + return static_cast(normalized); +} uint64_t GetTime(const RecordBatch* batch, Type::type time_type, int col, uint64_t row); diff --git a/cpp/src/arrow/acero/time_series_util_test.cc b/cpp/src/arrow/acero/time_series_util_test.cc new file mode 100644 index 000000000000..895e63e47cd6 --- /dev/null +++ b/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 +#include +#include +#include +#include +#include + +#include + +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 +void AssertOrderPreservedExhaustively() { + using Wide = int64_t; + constexpr Wide kMin = static_cast(std::numeric_limits::min()); + constexpr Wide kMax = static_cast(std::numeric_limits::max()); + + std::unordered_set seen; + uint64_t previous = 0; + bool first = true; + for (Wide value = kMin; value <= kMax; ++value) { + const T t = static_cast(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::value << ")"; + } + ASSERT_TRUE(seen.insert(normalized).second) + << "collision at " << value << " -> " << normalized; + previous = normalized; + first = false; + } + ASSERT_EQ(seen.size(), static_cast(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 +std::vector BoundarySweep() { + constexpr T kMin = std::numeric_limits::min(); + constexpr T kMax = std::numeric_limits::max(); + + std::vector values = {kMin, + static_cast(kMin + 1), + static_cast(kMin / 2), + static_cast(0), + static_cast(1), + static_cast(2), + static_cast(kMax / 2), + static_cast(kMax - 1), + kMax}; + if (std::is_signed::value) { + values.push_back(static_cast(-1)); + values.push_back(static_cast(-2)); + } + // Only representable if T is wide enough; int8_t would wrap. + if (static_cast(kMax) >= 1000) { + values.push_back(static_cast(1000)); + if (std::is_signed::value) values.push_back(static_cast(-1000)); + } + + std::sort(values.begin(), values.end()); + values.erase(std::unique(values.begin(), values.end()), values.end()); + return values; +} + +template +void AssertOrderPreservedAtBoundaries() { + const std::vector values = BoundarySweep(); + 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(values[i - 1]) + << " and " << static_cast(values[i]); + } +} + +} // namespace + +TEST(NormalizeTime, OrderPreservedExhaustively) { + AssertOrderPreservedExhaustively(); + AssertOrderPreservedExhaustively(); + AssertOrderPreservedExhaustively(); + AssertOrderPreservedExhaustively(); +} + +TEST(NormalizeTime, OrderPreservedAtBoundaries) { + AssertOrderPreservedAtBoundaries(); + AssertOrderPreservedAtBoundaries(); + AssertOrderPreservedAtBoundaries(); + AssertOrderPreservedAtBoundaries(); + AssertOrderPreservedAtBoundaries(); + AssertOrderPreservedAtBoundaries(); + AssertOrderPreservedAtBoundaries(); + AssertOrderPreservedAtBoundaries(); +} + +// 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::min()), 0u); + ASSERT_EQ(NormalizeTime(std::numeric_limits::max()), + std::numeric_limits::max()); + ASSERT_EQ(NormalizeTime(std::numeric_limits::min()), 0u); + ASSERT_EQ(NormalizeTime(std::numeric_limits::max()), + std::numeric_limits::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(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