From c1aba6025c531eea181c3b7653b35621d60886dd Mon Sep 17 00:00:00 2001 From: Bob Jansen Date: Tue, 22 Sep 2026 23:01:52 +0200 Subject: [PATCH] Finish the decimal-plan --- SPEC.md | 24 ++-- libs/adbc/adbc.cpp | 4 + plans/decimal-plan.md | 155 -------------------- src/runtime/aggregate_chunked.cpp | 93 ++++++++++++ src/runtime/decimal_aggregate.cpp | 72 ++++++++-- src/runtime/decimal_ops.hpp | 78 +++++++--- src/runtime/expr.cpp | 208 +++++++++++++++++++++------ src/runtime/interpreter_internal.hpp | 2 +- src/runtime/sort.cpp | 37 +++++ src/runtime/window.cpp | 62 +++++++- tests/test_chunked_sources.cpp | 25 ++++ tests/test_decimal_e2e.cpp | 96 ++++++++++++- 12 files changed, 616 insertions(+), 240 deletions(-) delete mode 100644 plans/decimal-plan.md diff --git a/SPEC.md b/SPEC.md index 90579299..1f0ca0bc 100644 --- a/SPEC.md +++ b/SPEC.md @@ -3093,6 +3093,9 @@ window argument (below); `lag` and `lead` do not. Rolling functions are **aggregate-like**: they produce one scalar per row (evaluated over the window) and are valid in both `select` and `update`. +For Decimal inputs, `rolling_sum` returns `Decimal(38, s)`; `rolling_min`, +`rolling_max`, `rolling_first`, and `rolling_last` preserve `Decimal(p, s)`. +The statistical rolling functions still require an explicit `Float64` cast. **Window bounds.** `window_start()` and `window_end()` return the *nominal* bounds of the window a row belongs to (as `Timestamp`, or `Date` for a Date time @@ -3175,9 +3178,11 @@ For `TimeFrame`, the current row order is the time-index ordering. | `cumsum(col)` | Running sum: result[i] = col[0] + col[1] + ... + col[i] | | `cumprod(col)` | Running product: result[i] = col[0] * col[1] * ... * col[i] | -Both functions accept `Int` or `Float` columns and return the same type as the -input. They are valid in both `select` and `update` blocks (DataFrame or -TimeFrame), with or without a `window` clause. +`cumsum` accepts `Int`, `Float`, and Decimal columns. Decimal cumulative sums +return `Decimal(38, s)` and check every prefix for overflow. `cumprod` accepts +`Int` and `Float`; Decimal multiplication changes scale at every step, so +Decimal `cumprod` is not defined. The functions are valid in both `select` and +`update` blocks (DataFrame or TimeFrame), with or without a `window` clause. ``` df[select { cs = cumsum(price) }] @@ -3770,7 +3775,7 @@ extern implementations. The recommended path for custom scalar logic is | `hour(t)` | `Timestamp -> Int32` | | `minute(t)` | `Timestamp -> Int32` | | `second(t)` | `Timestamp -> Int32` | -| `round(x, mode)`| `Float -> Int64` | +| `round(x, mode)`| `Float64/Int64 -> Int64`; `Decimal(p,s) -> Decimal(p',0)` | These scalar functions, the cast constructors of Section 3.1.1 (`Int64`/`Float64`/…), and `round` are row-wise: they may be used uniformly in @@ -3792,8 +3797,11 @@ null). The null-handling exceptions are `is_null`/`is_not_null`, `coalesce`, and the fill/clean functions of Section 3.5 (`fill_null`, `null_if_nan`, `null_if_not_finite`), whose purpose is to consume null. -`round(x, mode)` converts a `Float64` scalar or `Series` to `Int64` / -`Series`. The mode is a bare identifier (not a string): +`round(x, mode)` converts a `Float64` scalar or series to `Int64`; an `Int64` +value is returned unchanged. For Decimal input it returns an exact scale-zero +Decimal; the result precision is `min(38, max(1, p - s + 1))`. Decimal rounding +applies to the scaled integer units without converting through Float64. The +mode is a bare identifier (not a string): | Mode | Behaviour | C++ equivalent | |-----------|--------------------------------------------|------------------------| @@ -3813,8 +3821,8 @@ for contexts such as `select { hi = max(price) }, by symbol`. Row-wise `pmin` / `pmax` require comparable arguments of one type; `Int64` and `Float64` may be mixed and widen to `Float64`. -Passing an `Int` or `Int` column is a type error. An unknown mode identifier -is a runtime error. +Passing a non-numeric value is a type error. An unknown mode identifier is a +runtime error. ``` round(3.7, nearest) // → 4 diff --git a/libs/adbc/adbc.cpp b/libs/adbc/adbc.cpp index 0bfff5d3..ebbc5a3f 100644 --- a/libs/adbc/adbc.cpp +++ b/libs/adbc/adbc.cpp @@ -163,6 +163,10 @@ class AdbcSourceOperator final : public ibex::runtime::Operator { auto batch_guard = std::unique_ptr<::ArrowArray, void (*)(::ArrowArray*)>( &batch, ibex::interop::release_arrow_array); + // ADBC record batches use the same Arrow C Data importer as direct + // Arrow input, including its zero-copy decimal128 `d:p,s` mapping. + // Keep decimal type interpretation in that shared boundary so the + // ADBC path cannot drift from Arrow C Data or Parquet semantics. auto imported = ibex::interop::adopt_table_from_arrow(&batch, schema_); if (!imported) { finished_ = true; diff --git a/plans/decimal-plan.md b/plans/decimal-plan.md deleted file mode 100644 index cfb0c129..00000000 --- a/plans/decimal-plan.md +++ /dev/null @@ -1,155 +0,0 @@ -# Decimal(precision, scale) — semantics and plan - -Status: in progress on branch `decimal` (2026-09-14). This is the foundation the -ADBC work builds on: exact values must survive ingestion → computation → export. - -## Scope - -In: `Decimal(p, s)` with `1 <= p <= 38`, `0 <= s <= p`, stored as a signed -128-bit count of `10^-s` units. Out (deliberately): arbitrary precision, -Decimal256, negative scale, optimized 64-bit storage for small `p`. - -## Representation - -- **Column element** `ibex::Decimal { Int128 units; }` — 16 bytes, no scale. - Precision and scale are *column-level*: they live in `ColumnMeta::decimal`, - which already travels with a column through every copy/gather/slice - (`with_meta_of`). A per-row scale would double the width and make every - kernel re-check what the type already guarantees. -- **Scalar** `ibex::DecimalValue { Int128 units; DecimalType type; }` — a - scalar has no column to hang metadata on, so it carries its own type. This is - the alternative in `ScalarValue`, `ExprValue` and literals. -- `Int128` is `__int128` on GCC/Clang and `std::_Signed128` on MSVC. Every - checked operation is written as a comparison against `±(10^38 - 1)` rather - than a compiler overflow builtin, so it is portable and every stored value is - always inside decimal128's range — no intermediate can overflow the int128. -- IR/parser type: `ColumnType::Decimal` / `ScalarType::Decimal` plus an - optional `DecimalType` on the schema field, so ascriptions check `p`/`s`. - -## Literals and text - -- `decimal"12.30"` is an exact literal: scale = digits after the point (2), - precision = significant integer digits + scale, minimum 1. Leading sign and - an exponent (`1.5e3`) are accepted; `p > 38` is a parse error. -- Formatting always prints exactly `s` fractional digits (`12.30`, `-0.05`). - That is the text CSV writes and the REPL prints, so text round-trips. - -## Type rules (result types are static; values are checked at runtime) - -| Expression | Result type | -|---|---| -| `a + b`, `a - b` | `s = max(s1, s2)`, `p = min(38, max(p1 - s1, p2 - s2) + s + 1)` | -| `a * b` | `s = s1 + s2`, `p = min(38, p1 + p2)`; error if `s1 + s2 > 38` | -| `a / b` | `Float64` — `/` has one result type in Ibex (SPEC 3.1); cast back with `Decimal(x, p, s)` | -| `a % b` | rejected | -| `-a` | same type | -| `Decimal ⊕ Int64` | the Int64 side is `Decimal(19, 0)`, exactly | -| `Decimal ⊕ Float64` column | rejected: silent float contamination is the bug this type exists to prevent | -| `Decimal ⊕ float literal` | the literal is read exactly from its shortest round-trip text (`10.5` is `10.5`) | -| comparisons | exact, across scales (scale-aligned), also against Int64 and float literals | - -Overflow is checked: any result whose magnitude exceeds `10^p - 1` of its -result type is a runtime error (`decimal overflow`), never a wrap or a silent -null. Because `p` is capped at 38 the static rules above only bind at the cap. - -## Rounding - -One rule everywhere a value loses scale: **half away from zero** (`1.005 → -1.01`, `-1.005 → -1.01`), the rule of PostgreSQL, DuckDB and SQL Server -`CAST`. It applies to `Decimal(x, p, s)` casts, parsing text with more -fractional digits than the target scale (CSV), and float→decimal casts. - -## Casts - -- `Decimal(x, p, s)` from Int64 (exact), Decimal (rescale + round), String - (exact parse + round; not-a-number is an error), Float64 (shortest - round-trip text, then round). NaN/inf and values exceeding `p` are errors. -- `Float64(d)` — correctly rounded nearest double. -- `Int64(d)` — like Float→Int: succeeds only for whole values. - -## Aggregates - -| Aggregate | Result | -|---|---| -| `sum` | `Decimal(38, s)`, checked | -| `min`, `max`, `first`, `last` | input type | -| `mean` | `Float64`: the exact sum divided in decimal, with fractional digits beyond scale 38 when needed, then converted | -| `count`, `count_distinct` | `Int64` | -| `median`, `std`, `var`, `quantile`, … | rejected (cast to Float64 explicitly) | - -Nulls follow the existing rules (ignored by aggregates, null in → null out). - -## Relational operations - -Filter, sort, group-by, distinct and joins operate on the units, which are -exact within a column. Join keys whose scales differ are **refused** with an -error naming both types and the cast that fixes it. Aligning them silently -would put a per-join rescale (and a possible overflow) out of sight, and -`Decimal(x, p, s)` on one side states the same thing explicitly. Sorting keys -that fit int64 flatten directly; wider ones use dense ordinal ranks, which are -exact. - -Aggregates with a Decimal input run in `decimal_aggregate.cpp`: the chunked -hash aggregate's first chunk decides, and a Decimal input materializes and -takes that path. Non-Decimal aggregates in the same query are delegated back -to `aggregate_table` keyed on the same group ids, so they keep their usual -semantics exactly. - -## I/O - -- Arrow C Data: format `d:p,s` / `d:p,s,128` (import and export, zero-copy - import). `d:p,s,32`/`,64` import by widening. `,256`, `p > 38`, negative - scale: rejected with an explicit error. -- Parquet: read via Arrow's decimal128 (any physical encoding Arrow decodes); - write as decimal128 with `p`/`s` preserved. -- CSV: schema hint `decimal(p,s)` parses exactly (no double on the path); - write_csv formats exactly. - -## Step checklist - -1. [x] Core `decimal.hpp`: types, checked arithmetic, rounding, parse/format — - `tests/test_decimal.cpp`. -2. [x] Storage: `Column` in `ColumnValue`, `DecimalValue` in scalars, - `ColumnMeta::decimal`; every exhaustive visit handled. -3. [x] Language: `decimal"…"` literal, `Decimal(p, s)` type and - `Decimal(x, p, s)` cast, lowering, IR schema, codegen (parity case - `tests/parity/cases/decimal_money.ibex`). -4. [x] Runtime ops: compare/filter/sort/group/join/arith/aggregate — - `tests/test_decimal_e2e.cpp`. -5. [x] I/O: Arrow C Data (zero-copy decimal128), Parquet (every physical - encoding; cross-checked with pyarrow both ways), CSV `decimal(p,s)`. -6. [x] Validate: full suite + parity gate, SPEC + docs + `examples/decimal_money.ibex`; - release benchmarks: `ibex_bench --suite decimal` and an A/B of the - existing suites against HEAD. - -## Measured cost (2026-09-14, release, 4M rows, `taskset -c 0-7`) - -`ibex_bench --suite decimal`, the same values as Int64 cents / Float64 / -Decimal(18,2); min of 7 iterations after 3 warmups: - -| Kernel | Int64 | Float64 | Decimal(18,2) | vs Int64 | -|---|---|---|---|---| -| sum | 0.25 ms | 0.54 ms | 9.9 ms (Decimal(38,2): 9.4) | ~40× | -| grouped sum, 1000 groups | 16.1 ms | 12.0 ms | 35.0 ms | 2.2× | -| filter `> 250` | 1.23 ms | 1.33 ms | 2.12 ms | 1.7× | -| `x * 3` | 1.11 ms | 1.11 ms | 35.2 ms | ~32× | -| `x + x` | 1.12 ms | 0.99 ms | 29.2 ms | ~26× | -| sort | 146 ms | 118 ms | 151 ms (Decimal(38,2): 737) | 1.04× | - -Where the cost is: arithmetic and `sum` are serial, checked int128 loops (the -Int64/Float64 kernels are fused, vectorized and parallel); an ungrouped `sum` -also materializes its input for the decimal aggregate path. Before routing -Decimal `update` arithmetic to the vectorized kernel it went through the -per-row evaluator at ~150× Int64. Sorting keys that fit int64 flattens like -Int64; wider ones pay for dense ordinal ranks (5×). - -## Follow-ups (not in this slice) - -- Performance: parallel/morselized decimal arithmetic and sum; a streaming - decimal sum in the chunked aggregate instead of materializing; a direct - radix key for wide decimal sorts. - -- `abs`, `round(d, mode)` and the rolling/cumulative kernels over Decimal. -- Scalar-context aggregates (`sum(x)` inside an `update` broadcast, REPL - `aggregate_series`) have not been exercised over Decimal. -- ADBC: map `d:p,s` through the Arrow C Data path this slice established. diff --git a/src/runtime/aggregate_chunked.cpp b/src/runtime/aggregate_chunked.cpp index b4c94586..17a7efb8 100644 --- a/src/runtime/aggregate_chunked.cpp +++ b/src/runtime/aggregate_chunked.cpp @@ -5565,6 +5565,14 @@ class DecimalAwareAggregateOperator final : public Operator { if (!first->has_value()) { rest = std::make_unique(); } else if (has_decimal_input(**first)) { + if (can_stream_decimal_sum(**first)) { + auto result = stream_decimal_sum(**first); + if (!result) { + return std::unexpected(result.error()); + } + done_ = true; + return std::optional{std::move(*result)}; + } auto table = materialize_operator( std::make_unique(std::move(**first), std::move(child_))); if (!table.has_value()) { @@ -5586,6 +5594,91 @@ class DecimalAwareAggregateOperator final : public Operator { } private: + [[nodiscard]] auto can_stream_decimal_sum(const Chunk& chunk) const -> bool { + if (!group_by_->empty() || aggregations_->size() != 1) { + return false; + } + const auto& agg = aggregations_->front(); + if (agg.func != ir::AggFunc::Sum && agg.func != ir::AggFunc::Mean) { + return false; + } + return std::ranges::any_of(chunk.columns, [&](const ColumnEntry& column) { + return column.name == agg.column.name && + std::holds_alternative>(*column.column); + }); + } + + [[nodiscard]] auto stream_decimal_sum(const Chunk& first) -> std::expected { + const auto& agg = aggregations_->front(); + std::optional input_type; + Int128 sum = 0; + std::int64_t valid_count = 0; + const auto consume = [&](const Chunk& chunk) -> std::expected { + const auto it = std::ranges::find(chunk.columns, agg.column.name, &ColumnEntry::name); + if (it == chunk.columns.end() || it->column == nullptr) { + return std::unexpected("aggregate column not found: " + agg.column.name); + } + const auto* decimal_col = std::get_if>(it->column.get()); + if (decimal_col == nullptr) { + return std::unexpected("Decimal aggregate input changed type between chunks"); + } + const DecimalType current_type = decimal_type_of(*decimal_col); + if (input_type.has_value() && *input_type != current_type) { + return std::unexpected("Decimal aggregate type changed between chunks"); + } + input_type = current_type; + const Decimal* values = decimal_col->data(); + for (std::size_t row = 0; row < decimal_col->size(); ++row) { + if (it->validity.has_value() && !(*it->validity)[row]) { + continue; + } + if (!decimal::checked_add(sum, values[row].units, sum)) { + return std::unexpected("decimal overflow: aggregate sum exceeds Decimal(38)"); + } + ++valid_count; + } + return {}; + }; + if (auto consumed = consume(first); !consumed) { + return std::unexpected(consumed.error()); + } + while (true) { + auto next = child_->next(); + if (!next) { + return std::unexpected(next.error()); + } + if (!next->has_value()) { + break; + } + if (auto consumed = consume(**next); !consumed) { + return std::unexpected(consumed.error()); + } + } + const DecimalType type = input_type.value_or(DecimalType{}); + Table output; + if (agg.func == ir::AggFunc::Sum) { + Column result = make_decimal_column(decimal::sum_result_type(type)); + result.push_back(Decimal{sum}); + std::optional validity; + if (valid_count == 0) { + validity.emplace(1, false); + } + output.add_column(agg.alias, ColumnValue{std::move(result)}); + output.columns.back().validity = std::move(validity); + } else { + Column result; + result.push_back( + valid_count > 0 ? decimal::divide_to_double(sum, type.scale, valid_count) : 0.0); + std::optional validity; + if (valid_count == 0) { + validity.emplace(1, false); + } + output.add_column(agg.alias, ColumnValue{std::move(result)}); + output.columns.back().validity = std::move(validity); + } + return table_to_chunk(std::move(output)); + } + [[nodiscard]] auto has_decimal_input(const Chunk& chunk) const -> bool { for (const auto& agg : *aggregations_) { if (agg.func == ir::AggFunc::Count) { diff --git a/src/runtime/decimal_aggregate.cpp b/src/runtime/decimal_aggregate.cpp index 719c179e..1dd8cb5c 100644 --- a/src/runtime/decimal_aggregate.cpp +++ b/src/runtime/decimal_aggregate.cpp @@ -16,7 +16,9 @@ #include #include #include +#include +#include #include #include #include @@ -210,17 +212,69 @@ auto aggregate_decimal_column(const ir::AggSpec& agg, const ColumnEntry& entry, // overflow is past 38 digits -- an error, never a wrap. std::vector sums(groups, 0); std::vector counts(groups, 0); - for (std::size_t row = 0; row < rows; ++row) { - if (!present(row)) { - continue; + constexpr std::size_t kMinRowsPerWorker = 65536; + const bool parallel = rows >= kMinRowsPerWorker && !on_worker_pool_thread(); + WorkerPool* pool = parallel ? &process_worker_pool() : nullptr; + const std::size_t workers = + parallel ? std::min(pool->size(), rows / kMinRowsPerWorker) : 1; + // A private group array per worker avoids atomics in the row loop. + // Bound the scratch size; very high-cardinality inputs keep the + // serial path rather than multiplying an already large state table. + const bool use_parallel = workers > 1 && groups <= (1U << 20) / workers; + if (use_parallel) { + std::vector> partial_sums(workers, + std::vector(groups, 0)); + std::vector> partial_counts( + workers, std::vector(groups, 0)); + std::vector overflow(workers, 0); + auto batch = pool->submit(workers, [&](std::size_t worker) { + const std::size_t begin = rows * worker / workers; + const std::size_t end = rows * (worker + 1) / workers; + auto& local_sums = partial_sums[worker]; + auto& local_counts = partial_counts[worker]; + for (std::size_t row = begin; row < end; ++row) { + if (!present(row)) { + continue; + } + const std::uint32_t g = gids[row]; + if (!decimal::checked_add(local_sums[g], data[row].units, local_sums[g])) { + overflow[worker] = 1; + return; + } + ++local_counts[g]; + } + }); + batch.wait(); + for (std::size_t worker = 0; worker < workers; ++worker) { + if (overflow[worker] != 0) { + return std::unexpected( + std::string(agg_name(agg.func)) + "(" + agg.column.name + + "): " + decimal_overflow(decimal::sum_result_type(in_type))); + } } - const std::uint32_t g = gids[row]; - if (!decimal::checked_add(sums[g], data[row].units, sums[g])) { - return std::unexpected( - std::string(agg_name(agg.func)) + "(" + agg.column.name + - "): " + decimal_overflow(decimal::sum_result_type(in_type))); + for (std::size_t worker = 0; worker < workers; ++worker) { + for (std::size_t g = 0; g < groups; ++g) { + if (!decimal::checked_add(sums[g], partial_sums[worker][g], sums[g])) { + return std::unexpected( + std::string(agg_name(agg.func)) + "(" + agg.column.name + + "): " + decimal_overflow(decimal::sum_result_type(in_type))); + } + counts[g] += partial_counts[worker][g]; + } + } + } else { + for (std::size_t row = 0; row < rows; ++row) { + if (!present(row)) { + continue; + } + const std::uint32_t g = gids[row]; + if (!decimal::checked_add(sums[g], data[row].units, sums[g])) { + return std::unexpected( + std::string(agg_name(agg.func)) + "(" + agg.column.name + + "): " + decimal_overflow(decimal::sum_result_type(in_type))); + } + ++counts[g]; } - ++counts[g]; } std::vector seen(groups, 0); for (std::size_t g = 0; g < groups; ++g) { diff --git a/src/runtime/decimal_ops.hpp b/src/runtime/decimal_ops.hpp index e9b96fbe..cf22b4bb 100644 --- a/src/runtime/decimal_ops.hpp +++ b/src/runtime/decimal_ops.hpp @@ -10,7 +10,9 @@ #include #include +#include +#include #include #include #include @@ -179,6 +181,9 @@ struct DecimalColumnView { } }; +template +[[nodiscard]] inline auto for_each_decimal_range(std::size_t rows, F&& body) -> std::size_t; + [[nodiscard]] inline auto decimal_column_view(const ColumnValue& col, std::size_t off) -> std::optional { if (const auto* d = std::get_if>(&col)) { @@ -216,14 +221,18 @@ struct DecimalColumnView { Column out; out.resize(n); double* op_out = out.data(); - for (std::size_t i = 0; i < n; ++i) { - if (!is_row_valid(lv, lhs_off, i) || !is_row_valid(rv, rhs_off, i)) { - op_out[i] = 0.0; - continue; + const auto run = [&](std::size_t begin, std::size_t end) { + for (std::size_t i = begin; i < end; ++i) { + if (!is_row_valid(lv, lhs_off, i) || !is_row_valid(rv, rhs_off, i)) { + op_out[i] = 0.0; + continue; + } + op_out[i] = decimal::to_double(a->units(i), a->type.scale) / + decimal::to_double(b->units(i), b->type.scale); } - op_out[i] = decimal::to_double(a->units(i), a->type.scale) / - decimal::to_double(b->units(i), b->type.scale); - } + return n; + }; + (void)for_each_decimal_range(n, run); return ColumnValue{std::move(out)}; } auto rt = decimal_arith_type(op, a->type, b->type); @@ -233,21 +242,56 @@ struct DecimalColumnView { Column out = make_decimal_column(*rt); out.resize(n); Decimal* dst = out.data(); - for (std::size_t i = 0; i < n; ++i) { - if (!is_row_valid(lv, lhs_off, i) || !is_row_valid(rv, rhs_off, i)) { - dst[i] = Decimal{}; - continue; - } - Int128 r = 0; - if (!decimal_arith_units(op, a->units(i), a->type.scale, b->units(i), b->type.scale, *rt, - r)) { - return std::unexpected(decimal_overflow(*rt)); + const auto run = [&](std::size_t begin, std::size_t end) { + for (std::size_t i = begin; i < end; ++i) { + if (!is_row_valid(lv, lhs_off, i) || !is_row_valid(rv, rhs_off, i)) { + dst[i] = Decimal{}; + continue; + } + Int128 r = 0; + if (!decimal_arith_units(op, a->units(i), a->type.scale, b->units(i), b->type.scale, + *rt, r)) { + return i; + } + dst[i] = Decimal{r}; } - dst[i] = Decimal{r}; + return n; + }; + const std::size_t failed = for_each_decimal_range(n, run); + if (failed != n) { + return std::unexpected(decimal_overflow(*rt)); } return ColumnValue{std::move(out)}; } +template +[[nodiscard]] inline auto for_each_decimal_range(std::size_t rows, F&& body) -> std::size_t { + constexpr std::size_t kMinRowsPerWorker = 65536; + if (rows < kMinRowsPerWorker || on_worker_pool_thread()) { + return body(0, rows); + } + auto& pool = process_worker_pool(); + const std::size_t workers = + std::min(pool.size(), std::max(1, rows / kMinRowsPerWorker)); + if (workers < 2) { + return body(0, rows); + } + std::atomic first_failure{rows}; + auto batch = pool.submit(workers, [&](std::size_t worker) { + const std::size_t begin = rows * worker / workers; + const std::size_t end = rows * (worker + 1) / workers; + const std::size_t failure = body(begin, end); + if (failure < rows) { + std::size_t seen = first_failure.load(std::memory_order_relaxed); + while (failure < seen && + !first_failure.compare_exchange_weak(seen, failure, std::memory_order_relaxed)) { + } + } + }); + batch.wait(); + return first_failure.load(std::memory_order_relaxed); +} + /// A join/semi-join key pair that cannot be matched on raw units: two Decimal /// columns at different scales. Rescaling one side per join would hide a cost /// and an overflow; asking for the cast keeps both visible and the semantics diff --git a/src/runtime/expr.cpp b/src/runtime/expr.cpp index 91fb7a70..9f8f8af7 100644 --- a/src/runtime/expr.cpp +++ b/src/runtime/expr.cpp @@ -940,26 +940,32 @@ const robin_hood::unordered_map& builtins() { robin_hood::unordered_map m; // abs: numeric -> same numeric type. - m.emplace("abs", BuiltinFn{ - .min_args = 1, - .max_args = 1, - .infer = [](std::string_view, const std::vector& a) -> IT { - if (a[0] == ExprType::Int || a[0] == ExprType::Double) { - return a[0]; - } - return std::unexpected("abs: argument must be numeric"); - }, - .exec = ScalarExec{.eval = [](std::string_view, - const std::vector& a) -> IV { - if (const auto* i = std::get_if(a.data())) { - return ExprValue{std::int64_t{std::abs(*i)}}; - } - if (const auto* d = std::get_if(a.data())) { - return ExprValue{std::abs(*d)}; - } - return std::unexpected("abs: argument must be numeric"); - }}, - }); + m.emplace("abs", + BuiltinFn{ + .min_args = 1, + .max_args = 1, + .infer = [](std::string_view, const std::vector& a) -> IT { + if (a[0] == ExprType::Int || a[0] == ExprType::Double || + a[0] == ExprType::Decimal) { + return a[0]; + } + return std::unexpected("abs: argument must be numeric"); + }, + .exec = ScalarExec{.eval = [](std::string_view, + const std::vector& a) -> IV { + if (const auto* i = std::get_if(a.data())) { + return ExprValue{std::int64_t{std::abs(*i)}}; + } + if (const auto* d = std::get_if(a.data())) { + return ExprValue{std::abs(*d)}; + } + if (const auto* d = std::get_if(a.data())) { + return ExprValue{DecimalValue{ + .units = d->units < 0 ? -d->units : d->units, .type = d->type}}; + } + return std::unexpected("abs: argument must be numeric"); + }}, + }); // __interp: string interpolation, generated by the parser for backtick // template strings (`a=${a} b=${b}`). Variadic; every argument is @@ -1607,28 +1613,28 @@ const robin_hood::unordered_map& builtins() { }}, }); - m.emplace( - "cumsum", - BuiltinFn{ - .min_args = 0, - .max_args = -1, - .infer = [](std::string_view name, const std::vector& a) -> IT { - if (a.empty()) { - return std::unexpected(std::string(name) + ": expected 1 argument"); - } - return a[0]; - }, - .exec = TransformExec{.column_eval = [](const ir::CallExpr& call, - const Table& input, std::size_t, - const ColumnEvalCtx&) - -> std::expected { - auto col = eval_cumsum_cumprod_column(call, input, call.callee == "cumprod"); - if (!col) { - return std::unexpected(col.error()); - } - return ComputedColumn{.column = std::move(*col), .validity = std::nullopt}; - }}, - }); + m.emplace("cumsum", + BuiltinFn{ + .min_args = 0, + .max_args = -1, + .infer = [](std::string_view name, const std::vector& a) -> IT { + if (a.empty()) { + return std::unexpected(std::string(name) + ": expected 1 argument"); + } + return a[0]; + }, + .exec = TransformExec{.column_eval = [](const ir::CallExpr& call, + const Table& input, std::size_t, + const ColumnEvalCtx&) + -> std::expected { + auto col = + eval_cumsum_cumprod_column(call, input, call.callee == "cumprod"); + if (!col) { + return std::unexpected(col.error()); + } + return std::move(*col); + }}, + }); m.emplace("cumprod", m.at("cumsum")); // fill_forward / fill_backward: genuinely ordered (LOCF / NOCB read @@ -1950,6 +1956,36 @@ auto apply_round(double v, std::string_view mode) -> std::int64_t { return static_cast(std::trunc(v)); // trunc } +auto apply_decimal_round(const DecimalValue& value, std::string_view mode) + -> std::expected { + const int scale = value.type.scale; + const Int128 divisor = decimal::pow10(scale); + Int128 quotient = value.units / divisor; + const Int128 remainder = value.units % divisor; + const bool has_remainder = remainder != 0; + bool increment = false; + if (mode == "floor") { + increment = value.units < 0 && has_remainder; + } else if (mode == "ceil") { + increment = value.units > 0 && has_remainder; + } else if (mode == "nearest" || mode == "bankers") { + const Int128 magnitude = remainder < 0 ? -remainder : remainder; + const Int128 half = divisor / 2; + increment = + magnitude > half || (magnitude == half && (mode == "nearest" || (quotient % 2) != 0)); + } + if (increment) { + quotient += value.units < 0 ? -1 : 1; + } + const auto precision = static_cast( + std::min(decimal::kMaxPrecision, std::max(1, value.type.precision - scale + 1))); + const DecimalType type{.precision = precision, .scale = 0}; + if (!decimal::fits(quotient, precision)) { + return std::unexpected(decimal_overflow(type)); + } + return DecimalValue{.units = quotient, .type = type}; +} + } // namespace auto infer_expr_type(const ir::Expr& expr, const Table& input, const ScalarRegistry* scalars, @@ -2166,10 +2202,11 @@ auto infer_expr_type(const ir::Expr& expr, const Table& input, const ScalarRegis if (!arg_type) { return arg_type; } - if (*arg_type != ExprType::Int && *arg_type != ExprType::Double) { + if (*arg_type != ExprType::Int && *arg_type != ExprType::Double && + *arg_type != ExprType::Decimal) { return std::unexpected("round: first argument must be numeric"); } - return ExprType::Int; + return *arg_type == ExprType::Decimal ? ExprType::Decimal : ExprType::Int; } // Transforms (rolling/cum/lag/lead/fill/null_if_*) infer through the // registry entries above, same as Scalars and Generators. @@ -2278,6 +2315,32 @@ auto infer_decimal_type(const ir::Expr& expr, const Table& input, const ScalarRe return decimal_arith_type(bin->op, *left, *right); } if (const auto* call = std::get_if(&expr.node)) { + if ((call->callee == "abs" || call->callee == "rolling_min" || + call->callee == "rolling_max" || call->callee == "rolling_first" || + call->callee == "rolling_last") && + !call->args.empty()) { + return infer_decimal_type(*call->args[0], input, scalars); + } + if ((call->callee == "cumsum" || call->callee == "rolling_sum") && !call->args.empty()) { + auto input_type = infer_decimal_type(*call->args[0], input, scalars); + if (!input_type) { + return input_type; + } + return decimal::sum_result_type(*input_type); + } + if ((call->callee == "abs" || call->callee == "round") && !call->args.empty()) { + auto input_type = infer_decimal_type(*call->args[0], input, scalars); + if (!input_type) { + return input_type; + } + if (call->callee == "abs") { + return *input_type; + } + const auto precision = static_cast( + std::min(decimal::kMaxPrecision, + std::max(1, input_type->precision - input_type->scale + 1))); + return DecimalType{.precision = precision, .scale = 0}; + } if (call->callee == "Decimal" && call->args.size() == 3) { const auto* p = std::get_if(&call->args[1]->node); const auto* s = std::get_if(&call->args[2]->node); @@ -2512,6 +2575,16 @@ auto eval_expr(const ir::Expr& expr, const Table& input, std::size_t row, if (std::holds_alternative(arg.value())) { return ExprValue{Null{}}; } + if (const auto* decimal_value = std::get_if(&arg.value())) { + auto rounded = apply_decimal_round(*decimal_value, *mode); + if (!rounded) { + return std::unexpected(rounded.error()); + } + return ExprValue{*rounded}; + } + if (const auto* integer = std::get_if(&arg.value())) { + return ExprValue{*integer}; + } auto d = expr_value_to_double(arg.value()); if (!d) { return std::unexpected("round: first argument must be numeric"); @@ -2963,7 +3036,7 @@ auto eval_lag_lead_column(const ir::CallExpr& call, const Table& input, bool is_ // cumprod(col)[i] = col[0] * col[1] * ... * col[i] // Only valid for numeric (Int / Float) columns. auto eval_cumsum_cumprod_column(const ir::CallExpr& call, const Table& input, bool is_prod) - -> std::expected { + -> std::expected { const std::string fname = is_prod ? "cumprod" : "cumsum"; if (call.args.size() != 1) { return std::unexpected(fname + ": expected 1 argument"); @@ -2977,8 +3050,48 @@ auto eval_cumsum_cumprod_column(const ir::CallExpr& call, const Table& input, bo return std::unexpected(fname + ": unknown column '" + col_ref->name + "'"); } std::size_t rows = input.rows(); + if (const auto* decimal_col = std::get_if>(src)) { + if (is_prod) { + return std::unexpected("cumprod: Decimal result precision and scale are not fixed"); + } + const DecimalType input_type = decimal_type_of(*decimal_col); + const DecimalType result_type = decimal::sum_result_type(input_type); + Column result = make_decimal_column(result_type); + result.resize(rows); + const auto* entry = input.find_entry(col_ref->name); + const ValidityBitmap* source_validity = + entry != nullptr && entry->validity.has_value() ? &*entry->validity : nullptr; + std::optional result_validity; + if (source_validity != nullptr) { + result_validity.emplace(rows, true); + } + Int128 sum = 0; + const Decimal* values = decimal_col->data(); + Decimal* output = result.data(); + for (std::size_t i = 0; i < rows; ++i) { + if (source_validity != nullptr && !(*source_validity)[i]) { + output[i] = Decimal{}; + if (result_validity.has_value()) { + result_validity->set(i, false); + } + continue; + } + Int128 next = 0; + if (!decimal::checked_add(sum, values[i].units, next) || + !decimal::fits(next, result_type.precision)) { + return std::unexpected("cumsum(" + col_ref->name + + "): decimal overflow: " + "result does not fit " + + decimal::type_name(result_type)); + } + sum = next; + output[i] = Decimal{sum}; + } + return ComputedColumn{.column = ColumnValue{std::move(result)}, + .validity = std::move(result_validity)}; + } return std::visit( - [&](const auto& col) -> std::expected { + [&](const auto& col) -> std::expected { using ColT = std::decay_t; using T = ColT::value_type; if constexpr (std::is_same_v || std::is_same_v) { @@ -3003,7 +3116,8 @@ auto eval_cumsum_cumprod_column(const ir::CallExpr& call, const Table& input, bo } out[i] = acc; } - return result; + return ComputedColumn{.column = ColumnValue{std::move(result)}, + .validity = std::nullopt}; } else { return std::unexpected(fname + ": column must be numeric (Int or Float)"); } diff --git a/src/runtime/interpreter_internal.hpp b/src/runtime/interpreter_internal.hpp index 37241fd9..8c1958cb 100644 --- a/src/runtime/interpreter_internal.hpp +++ b/src/runtime/interpreter_internal.hpp @@ -1796,7 +1796,7 @@ enum class FloatCleanMode : std::uint8_t { [[nodiscard]] auto eval_cumsum_cumprod_column(const ir::CallExpr& call, const Table& input, bool is_prod) - -> std::expected; + -> std::expected; [[nodiscard]] auto eval_fill_null(const ir::CallExpr& call, const Table& input) -> std::expected; [[nodiscard]] auto eval_fill_forward(const ir::CallExpr& call, const Table& input) diff --git a/src/runtime/sort.cpp b/src/runtime/sort.cpp index b020e195..bf1f6c2f 100644 --- a/src/runtime/sort.cpp +++ b/src/runtime/sort.cpp @@ -518,6 +518,43 @@ auto order_table_resolved(const Table& input, const std::vector& r FlatKey fk; fk.ascending = key.ascending; fk.validity = entry != nullptr && entry->validity.has_value() ? &*entry->validity : nullptr; + if (const auto* decimal_col = std::get_if>(column)) { + const Decimal* values = decimal_col->data(); + const bool fits64 = std::all_of(values, values + rows, [](const Decimal& value) { + return value.units >= Int128{INT64_MIN} && value.units <= Int128{INT64_MAX}; + }); + if (fits64) { + fk.kind = FlatKind::I64; + fk.u64 = decimal_order_keys(*decimal_col, rows); + flat_keys.push_back(std::move(fk)); + } else { + // Encode the signed 128-bit unit count as two order-preserving + // u64 keys. Splitting after flipping the sign bit preserves + // full signed order; the existing stable multi-key radix sorts + // the low half first and the high half second. Dense ordinal + // ranking sorted all rows once before sorting them again. + constexpr Int128 kWord = Int128{1} << 64; + FlatKey high; + high.kind = FlatKind::I64; + high.ascending = key.ascending; + high.validity = fk.validity; + high.u64.reserve(rows); + fk.u64.reserve(rows); + for (const Decimal& value : *decimal_col) { + Int128 upper = value.units / kWord; + const Int128 lower = value.units % kWord; + if (lower < 0) { + --upper; + } + high.u64.push_back(static_cast(upper) ^ kSignFlip); + fk.u64.push_back(static_cast(value.units)); + } + // Both halves must be complemented for descending order. + flat_keys.push_back(std::move(high)); + flat_keys.push_back(std::move(fk)); + } + continue; + } std::visit( [&](const auto& col) { using ColT = std::decay_t; diff --git a/src/runtime/window.cpp b/src/runtime/window.cpp index 25dce685..609c0528 100644 --- a/src/runtime/window.cpp +++ b/src/runtime/window.cpp @@ -6,6 +6,7 @@ // Split out of interpreter.cpp; shared declarations live in interpreter_internal.hpp. #include +#include #include #include #include @@ -609,6 +610,7 @@ auto window_bound_column(const Table& table, ir::Duration duration, bool aligned // pair in the output expression, which may be sampling skid -- the next step is // a cycle-accurate profile, which this box cannot give (WSL2 has no hardware // counters). +// NOLINTNEXTLINE(readability-function-size) auto apply_rolling_func(const ir::CallExpr& call, const Table& table, WindowSpec spec, bool aligned) -> std::expected { std::size_t rows = table.rows(); @@ -739,7 +741,7 @@ auto apply_rolling_func(const ir::CallExpr& call, const Table& table, WindowSpec } else { const auto* values = col.data(); Column result; - result.resize_for_overwrite(rows); + result.resize(rows); auto* result_values = result.data(); if (sv == nullptr) { double sum = 0.0; @@ -834,7 +836,57 @@ auto apply_rolling_func(const ir::CallExpr& call, const Table& table, WindowSpec [&](const auto& col) -> std::expected { using ColT = std::decay_t; using T = ColT::value_type; - if constexpr (!std::is_same_v && !std::is_same_v) { + if constexpr (std::is_same_v) { + const DecimalType input_type = decimal_type_of(col); + const DecimalType result_type = decimal::sum_result_type(input_type); + Column result = make_decimal_column(result_type); + result.resize(rows); + Decimal* output = result.data(); + const Decimal* values = col.data(); + std::optional out_valid; + Int128 sum = 0; + std::size_t valid_count = 0; + std::size_t lo = 0; + for (std::size_t i = 0; i < rows; ++i) { + while (lo < i && should_drop(lo, i)) { + if (valid_at(lo)) { + Int128 next = 0; + if (!decimal::checked_sub(sum, values[lo].units, next)) { + return std::unexpected( + call.callee + ": decimal overflow: result does not fit " + + decimal::type_name(result_type)); + } + sum = next; + --valid_count; + } + ++lo; + } + if (valid_at(i)) { + Int128 next = 0; + if (!decimal::checked_add(sum, values[i].units, next) || + !decimal::fits(next, result_type.precision)) { + return std::unexpected(call.callee + + ": decimal overflow: result " + "does not fit " + + decimal::type_name(result_type)); + } + sum = next; + ++valid_count; + } + if (valid_count == 0) { + output[i] = Decimal{}; + if (!out_valid) { + out_valid.emplace(rows, true); + } + out_valid->set(i, false); + } else { + output[i] = Decimal{sum}; + } + } + return ComputedColumn{.column = std::move(result), + .validity = std::move(out_valid)}; + } else if constexpr (!std::is_same_v && + !std::is_same_v) { return std::unexpected("rolling_sum: column must be numeric (Int or Float)"); } else { ColT result; @@ -1479,6 +1531,9 @@ auto apply_rolling_func(const ir::CallExpr& call, const Table& table, WindowSpec return std::unexpected(call.callee + ": categorical columns are not supported"); } else { ColT result; + if constexpr (std::is_same_v>) { + result.set_meta(col.meta()); + } std::optional out_valid; if constexpr (!std::is_same_v>) { result.resize(rows); @@ -1519,6 +1574,9 @@ auto apply_rolling_func(const ir::CallExpr& call, const Table& table, WindowSpec } else { using T = ColT::value_type; ColT result; + if constexpr (std::is_same_v>) { + result.set_meta(col.meta()); + } // Unlike the other rolling kernels this one also instantiates for // Date/Timestamp/bool, which have no uninitialised resize. if constexpr (requires { result.resize_for_overwrite(rows); }) { diff --git a/tests/test_chunked_sources.cpp b/tests/test_chunked_sources.cpp index 87807de1..30471165 100644 --- a/tests/test_chunked_sources.cpp +++ b/tests/test_chunked_sources.cpp @@ -16,6 +16,7 @@ // Both were invisible at one chunk and wrong at two. #include +#include #include #include #include @@ -791,6 +792,30 @@ TEST_CASE("chunked aggregate: moment aggregates agree serially and in parallel", } } +TEST_CASE("chunked Decimal sum and mean stream without materializing input", + "[runtime][chunked][aggregate][decimal]") { + constexpr std::size_t kRows = 200'000; + const DecimalType type{.precision = 8, .scale = 2}; + auto value = runtime::make_decimal_column(type); + value.resize(kRows, Decimal{1}); // 0.01 per row + runtime::Table table; + table.add_column("x", std::move(value)); + runtime::TableRegistry registry; + registry.emplace("t", std::move(table)); + + const ChunkGrainGuard guard{"1024"}; + const auto sum = run("t[select { total = sum(x) }];", registry); + const auto* total = std::get_if>(sum.find("total")); + REQUIRE(total != nullptr); + CHECK(runtime::decimal_type_of(*total) == (DecimalType{.precision = 38, .scale = 2})); + CHECK((*total)[0].units == 200'000); + + const auto mean = run("t[select { average = mean(x) }];", registry); + const auto* average = std::get_if>(mean.find("average")); + REQUIRE(average != nullptr); + CHECK((*average)[0] == 0.01); +} + TEST_CASE("chunked aggregate: output emission agrees serially and in parallel", "[runtime][chunked][aggregate]") { // `build_output_chunk` emits as `(output column x group range)` tasks. diff --git a/tests/test_decimal_e2e.cpp b/tests/test_decimal_e2e.cpp index 33c9be23..e2416c44 100644 --- a/tests/test_decimal_e2e.cpp +++ b/tests/test_decimal_e2e.cpp @@ -205,6 +205,85 @@ TEST_CASE("Decimal casts round half away from zero at the boundary", "[decimal][ CHECK(contains(run_err("t[update { x = Decimal(\"abc\", 10, 2) }];", tables), "abc")); } +TEST_CASE("Decimal abs and round preserve exact scale semantics", "[decimal][e2e]") { + runtime::Table t; + t.add_column("x", dec_col(dec_type(6, 2), {"1.50", "-1.50", "1.25", "-1.25", "0.50"})); + runtime::TableRegistry tables; + tables.emplace("t", std::move(t)); + auto out = run_ok( + "t[update { magnitude = abs(x), nearest = round(x, nearest), " + "bankers = round(x, bankers), low = round(x, floor), high = round(x, ceil), " + "toward_zero = round(x, trunc) }];", + tables); + CHECK(texts(out, "magnitude") == + std::vector{"1.50", "1.50", "1.25", "1.25", "0.50"}); + CHECK(type_of(out, "magnitude") == dec_type(6, 2)); + CHECK(texts(out, "nearest") == std::vector{"2", "-2", "1", "-1", "1"}); + CHECK(texts(out, "bankers") == std::vector{"2", "-2", "1", "-1", "0"}); + CHECK(texts(out, "low") == std::vector{"1", "-2", "1", "-2", "0"}); + CHECK(texts(out, "high") == std::vector{"2", "-1", "2", "-1", "1"}); + CHECK(texts(out, "toward_zero") == std::vector{"1", "-1", "1", "-1", "0"}); + CHECK(type_of(out, "nearest") == dec_type(5, 0)); +} + +TEST_CASE("Decimal cumulative and rolling kernels preserve exact values and types", + "[decimal][e2e][rolling]") { + runtime::Table t; + t.add_column("x", dec_col(dec_type(5, 2), {"1.20", "-0.05", "2.30", "0.10"})); + runtime::TableRegistry tables; + tables.emplace("t", std::move(t)); + auto out = run_ok( + "t[update { cumulative = cumsum(x), rs = rolling_sum(x, 2), " + "rmin = rolling_min(x, 2), rmax = rolling_max(x, 2), " + "first = rolling_first(x, 2), last = rolling_last(x, 2) }];", + tables); + CHECK(texts(out, "cumulative") == std::vector{"1.20", "1.15", "3.45", "3.55"}); + CHECK(type_of(out, "cumulative") == dec_type(38, 2)); + CHECK(texts(out, "rs") == std::vector{"1.20", "1.15", "2.25", "2.40"}); + CHECK(type_of(out, "rs") == dec_type(38, 2)); + CHECK(texts(out, "rmin") == std::vector{"1.20", "-0.05", "-0.05", "0.10"}); + CHECK(texts(out, "rmax") == std::vector{"1.20", "1.20", "2.30", "2.30"}); + CHECK(texts(out, "first") == std::vector{"1.20", "1.20", "-0.05", "2.30"}); + CHECK(texts(out, "last") == std::vector{"1.20", "-0.05", "2.30", "0.10"}); + for (const auto* name : {"rmin", "rmax", "first", "last"}) { + CHECK(type_of(out, name) == dec_type(5, 2)); + } +} + +TEST_CASE("Decimal cumulative and rolling kernels skip null payloads safely", + "[decimal][e2e][rolling][null]") { + runtime::Table t; + t.add_column("x", dec_col(dec_type(4, 2), {"1.00", "9.99", "2.00"}), + runtime::ValidityBitmap{true, false, true}); + runtime::TableRegistry tables; + tables.emplace("t", std::move(t)); + const auto out = + run_ok("t[update { cumulative = cumsum(x), rs = rolling_sum(x, 2) }];", tables); + CHECK(texts(out, "cumulative") == std::vector{"1.00", "null", "3.00"}); + CHECK(texts(out, "rs") == std::vector{"1.00", "1.00", "2.00"}); +} + +TEST_CASE("Decimal aggregates work in broadcasts and scalar series reduction", + "[decimal][e2e][aggregate]") { + runtime::Table t; + t.add_column("g", Column{1, 1, 2, 2}); + auto amounts = dec_col(dec_type(8, 2), {"0.10", "0.20", "1.25", "-0.25"}); + t.add_column("x", amounts); + auto sum = runtime::aggregate_series("sum", runtime::ColumnValue{amounts}); + REQUIRE(sum.has_value()); + const auto* scalar_sum = std::get_if(&*sum); + REQUIRE(scalar_sum != nullptr); + CHECK(scalar_sum->type == dec_type(38, 2)); + CHECK(decimal::to_string(*scalar_sum) == "1.30"); + + runtime::TableRegistry tables; + tables.emplace("t", std::move(t)); + auto out = run_ok("t[update { total = sum(x), centered = x - sum(x) }, by g];", tables); + CHECK(texts(out, "total") == std::vector{"0.30", "0.30", "1.00", "1.00"}); + CHECK(type_of(out, "total") == dec_type(38, 2)); + CHECK(texts(out, "centered") == std::vector{"-0.20", "-0.10", "0.25", "-1.25"}); +} + TEST_CASE("Decimal converts to Float64 and whole values to Int64", "[decimal][e2e]") { const auto tables = prices(); auto out = run_ok("t[update { d = Float64(price) }];", tables); @@ -260,16 +339,31 @@ TEST_CASE("Decimal sorts by value, including negatives and wide values", "[decim CHECK(texts(run_ok("t[order { price desc }];", tables), "price") == std::vector{"100.00", "10.50", "10.49", "-3.25"}); - // More than 18 digits: units no longer fit int64, so the ordinal path. + // More than 18 digits: units no longer fit int64, so sorting must use + // both halves of the direct decimal128 radix key. runtime::Table wide; wide.add_column( "v", dec_col(dec_type(38, 0), {"12345678901234567890123", "-99999999999999999999", "5", "12345678901234567890122"})); + wide.add_column("g", Column{2, 1, 1, 2}); runtime::TableRegistry wide_tables; wide_tables.emplace("t", std::move(wide)); CHECK(texts(run_ok("t[order v];", wide_tables), "v") == std::vector{"-99999999999999999999", "5", "12345678901234567890122", "12345678901234567890123"}); + CHECK(texts(run_ok("t[order { v desc }];", wide_tables), "v") == + std::vector{"12345678901234567890123", "12345678901234567890122", "5", + "-99999999999999999999"}); + auto multi = run_ok("t[order { g, v }];", wide_tables); + const auto* groups = std::get_if>(multi.find("g")); + REQUIRE(groups != nullptr); + CHECK((*groups)[0] == 1); + CHECK((*groups)[1] == 1); + CHECK((*groups)[2] == 2); + CHECK((*groups)[3] == 2); + CHECK(texts(multi, "v") == std::vector{"-99999999999999999999", "5", + "12345678901234567890122", + "12345678901234567890123"}); } TEST_CASE("Decimal grouping and aggregates", "[decimal][e2e]") {