Finish the decimal-plan

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<Float64>` to `Int64` /
-`Series<Int64>`. 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

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;

src/runtime/aggregate_chunked.cpp

@@ -5565,6 +5565,14 @@ class DecimalAwareAggregateOperator final : public Operator {
         if (!first->has_value()) {
             rest = std::make_unique<ExhaustedOperator>();
         } 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<Chunk>{std::move(*result)};
+            }
             auto table = materialize_operator(
                 std::make_unique<PrependChunkOperator>(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<Decimal>>(*column.column);
+        });
+    }
+
+    [[nodiscard]] auto stream_decimal_sum(const Chunk& first) -> std::expected<Chunk, std::string> {
+        const auto& agg = aggregations_->front();
+        std::optional<DecimalType> input_type;
+        Int128 sum = 0;
+        std::int64_t valid_count = 0;
+        const auto consume = [&](const Chunk& chunk) -> std::expected<void, std::string> {
+            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<Column<Decimal>>(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<Decimal> result = make_decimal_column(decimal::sum_result_type(type));
+            result.push_back(Decimal{sum});
+            std::optional<ValidityBitmap> 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<double> result;
+            result.push_back(
+                valid_count > 0 ? decimal::divide_to_double(sum, type.scale, valid_count) : 0.0);
+            std::optional<ValidityBitmap> 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) {