perf(flow-php/etl): native int/float fast-path in sum() with opt-in exact mode

src/core/etl/src/Flow/ETL/Config.php

@@ -4,6 +4,7 @@
 
 namespace Flow\ETL;
 
+use Flow\Calculator\Calculator;
 use Flow\ETL\Config\Cache\CacheConfig;
 use Flow\ETL\Config\ConfigBuilder;
 use Flow\ETL\Config\Sort\SortConfig;
@@ -36,6 +37,7 @@ public function __construct(
         public SortConfig $sort,
         private ?Analyze $analyze,
         public TelemetryConfig $telemetry,
+        private Calculator $calculator = new Calculator(),
     ) {}
 
     public static function builder(): ConfigBuilder
@@ -53,6 +55,11 @@ public function analyze(): ?Analyze
         return $this->analyze;
     }
 
+    public function calculator(): Calculator
+    {
+        return $this->calculator;
+    }
+
     public function clock(): ClockInterface
     {
         return $this->clock;

src/core/etl/src/Flow/ETL/DSL/functions.php

@@ -1824,9 +1824,9 @@ function window(): Window
 }
 
 #[DocumentationDSL(module: Module::CORE, type: DSLType::AGGREGATING_FUNCTION)]
-function sum(EntryReference|string $ref): Sum
+function sum(EntryReference|string $ref, ScalarFunction|bool $exact = false): Sum
 {
-    return new Sum(is_string($ref) ? ref($ref) : $ref);
+    return new Sum(is_string($ref) ? ref($ref) : $ref, $exact);
 }
 
 #[DocumentationDSL(module: Module::CORE, type: DSLType::AGGREGATING_FUNCTION)]

src/core/etl/src/Flow/ETL/FlowContext.php

@@ -4,6 +4,7 @@
 
 namespace Flow\ETL;
 
+use Flow\Calculator\Calculator;
 use Flow\ETL\Config\Telemetry\TelemetryContext;
 use Flow\ETL\ErrorHandler\ThrowError;
 use Flow\ETL\Filesystem\FilesystemStreams;
@@ -37,6 +38,11 @@ public function cache(): Cache
         return $this->config->cache->cache;
     }
 
+    public function calculator(): Calculator
+    {
+        return $this->config->calculator();
+    }
+
     public function entryFactory(): EntryFactory
     {
         return $this->config->entryFactory();

src/core/etl/src/Flow/ETL/Function/Average.php

@@ -47,7 +47,7 @@ public function aggregate(Row $row, FlowContext $context): void
 
             if (is_numeric($value)) {
                 // @mago-ignore analysis:possibly-invalid-argument
-                $this->sum = (new Calculator())->add($this->sum, $value);
+                $this->sum = $context->calculator()->add($this->sum, $value);
                 $this->count++;
             }
         } catch (InvalidArgumentException $e) {
@@ -67,7 +67,7 @@ public function apply(Row $row, Rows $partition, FlowContext $context): mixed
 
                 if (is_numeric($value)) {
                     // @mago-ignore analysis:possibly-invalid-argument
-                    $sum = (new Calculator())->add($sum, $value);
+                    $sum = $context->calculator()->add($sum, $value);
                     $count++;
                 }
             } catch (InvalidArgumentException $e) {
@@ -79,7 +79,7 @@ public function apply(Row $row, Rows $partition, FlowContext $context): mixed
             }
         }
 
-        return (new Calculator())->divide($sum, $count, $this->scale, $this->rounding);
+        return $context->calculator()->divide($sum, $count, $this->scale, $this->rounding);
     }
 
     public function over(Window $window): WindowFunction

src/core/etl/src/Flow/ETL/Function/Sum.php

@@ -4,7 +4,6 @@
 
 namespace Flow\ETL\Function;
 
-use Flow\Calculator\Calculator;
 use Flow\ETL\Exception\InvalidArgumentException;
 use Flow\ETL\Exception\RuntimeException;
 use Flow\ETL\FlowContext;
@@ -27,6 +26,7 @@ final class Sum implements AggregatingFunction, WindowFunction
 
     public function __construct(
         private readonly Reference $ref,
+        private readonly ScalarFunction|bool $exact = false,
     ) {
         $this->sum = 0;
         $this->window = null;
@@ -38,7 +38,7 @@ public function aggregate(Row $row, FlowContext $context): void
             $value = $row->valueOf($this->ref);
 
             if (is_int($value) || is_float($value) || is_string($value) && is_numeric($value)) {
-                $this->sum = (new Calculator())->add($this->sum, $value);
+                $this->sum = $this->add($this->sum, $value, $row, $context);
             }
         } catch (InvalidArgumentException $e) {
             $context->functions()->invalidResult(new InvalidArgumentException('Sum error: ' . $e->getMessage()));
@@ -54,7 +54,7 @@ public function apply(Row $row, Rows $partition, FlowContext $context): mixed
                 $value = $partitionRow->valueOf($this->ref);
 
                 if (is_int($value) || is_float($value) || is_string($value) && is_numeric($value)) {
-                    $sum = (new Calculator())->add($sum, $value);
+                    $sum = $this->add($sum, $value, $partitionRow, $context);
                 }
             } catch (InvalidArgumentException $e) {
                 $context
@@ -104,4 +104,36 @@ public function window(): Window
 
         return $this->window;
     }
+
+    /**
+     * @param float|int|numeric-string $value
+     */
+    private function add(float|int $sum, float|int|string $value, Row $row, FlowContext $context): float|int
+    {
+        if ($this->isExact($row, $context)) {
+            return $context->calculator()->add($sum, $value);
+        }
+
+        $result = $sum + $value;
+
+        if (
+            is_float($result)
+            && floor($result) === $result
+            && $result >= (float) PHP_INT_MIN
+            && $result < (float) PHP_INT_MAX
+        ) {
+            return (int) $result;
+        }
+
+        return $result;
+    }
+
+    private function isExact(Row $row, FlowContext $context): bool
+    {
+        if (is_bool($this->exact)) {
+            return $this->exact;
+        }
+
+        return (new Parameter($this->exact))->asBoolean($row, $context);
+    }
 }

src/core/etl/tests/Flow/ETL/Tests/Unit/FlowContextTest.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Tests\Unit;
+
+use Flow\Calculator\Calculator;
+use Flow\ETL\Tests\FlowTestCase;
+
+use function Flow\ETL\DSL\config;
+use function Flow\ETL\DSL\flow_context;
+
+final class FlowContextTest extends FlowTestCase
+{
+    public function test_provides_shared_calculator_instance_from_config(): void
+    {
+        $context = flow_context(config());
+
+        static::assertInstanceOf(Calculator::class, $context->calculator());
+        static::assertSame($context->calculator(), $context->calculator());
+    }
+}

src/core/etl/tests/Flow/ETL/Tests/Unit/Function/SumTest.php

@@ -8,10 +8,12 @@
 use Flow\ETL\Function\ExecutionMode;
 use Flow\ETL\Tests\FlowTestCase;
 
+use function Flow\ETL\DSL\bool_entry;
 use function Flow\ETL\DSL\config;
 use function Flow\ETL\DSL\float_entry;
 use function Flow\ETL\DSL\flow_context;
 use function Flow\ETL\DSL\int_entry;
+use function Flow\ETL\DSL\lit;
 use function Flow\ETL\DSL\ref;
 use function Flow\ETL\DSL\row;
 use function Flow\ETL\DSL\rows;
@@ -58,6 +60,48 @@ public function test_aggregation_sum_with_float_result(): void
         static::assertSame(360.25, $aggregator->result(flow_context(config())->entryFactory())->value());
     }
 
+    public function test_aggregation_sum_of_decimal_fractions(): void
+    {
+        $aggregator = sum(ref('value'));
+
+        $aggregator->aggregate(row(float_entry('value', 0.1)), flow_context());
+        $aggregator->aggregate(row(float_entry('value', 0.2)), flow_context());
+
+        static::assertSame(0.3, $aggregator->result(flow_context(config())->entryFactory())->value());
+    }
+
+    public function test_window_function_sum_of_decimal_fractions_uses_float_arithmetic_by_default(): void
+    {
+        $rows = rows(
+            $row1 = row(int_entry('id', 1), float_entry('value', 0.1)),
+            row(int_entry('id', 2), float_entry('value', 0.2)),
+        );
+
+        $sum = sum(ref('value'))->over(window()->orderBy(ref('id')->desc()));
+
+        static::assertSame(0.1 + 0.2, $sum->apply($row1, $rows, flow_context()));
+    }
+
+    public function test_aggregation_sum_of_floats_returns_int_when_sum_is_whole(): void
+    {
+        $aggregator = sum(ref('value'));
+
+        $aggregator->aggregate(row(float_entry('value', 2.5)), flow_context());
+        $aggregator->aggregate(row(float_entry('value', 2.5)), flow_context());
+
+        static::assertSame(5, $aggregator->result(flow_context(config())->entryFactory())->value());
+    }
+
+    public function test_exact_aggregation_sum_of_decimal_fractions(): void
+    {
+        $aggregator = sum(ref('value'), exact: true);
+
+        $aggregator->aggregate(row(float_entry('value', 0.1)), flow_context());
+        $aggregator->aggregate(row(float_entry('value', 0.2)), flow_context());
+
+        static::assertSame(0.3, $aggregator->result(flow_context(config())->entryFactory())->value());
+    }
+
     public function test_window_function_sum_on_partitioned_rows(): void
     {
         $rows = rows(
@@ -73,6 +117,42 @@ public function test_window_function_sum_on_partitioned_rows(): void
         static::assertSame(15, $sum->apply($row1, $rows, flow_context()));
     }
 
+    public function test_window_function_sum_of_decimal_fractions_in_exact_mode(): void
+    {
+        $rows = rows(
+            $row1 = row(int_entry('id', 1), float_entry('value', 0.1)),
+            row(int_entry('id', 2), float_entry('value', 0.2)),
+        );
+
+        $sum = sum(ref('value'), exact: true)->over(window()->orderBy(ref('id')->desc()));
+
+        static::assertSame(0.3, $sum->apply($row1, $rows, flow_context()));
+    }
+
+    public function test_window_function_sum_with_exact_mode_from_column(): void
+    {
+        $rows = rows(
+            $row1 = row(int_entry('id', 1), float_entry('value', 0.1), bool_entry('is_exact', true)),
+            row(int_entry('id', 2), float_entry('value', 0.2), bool_entry('is_exact', true)),
+        );
+
+        $sum = sum(ref('value'), exact: ref('is_exact'))->over(window()->orderBy(ref('id')->desc()));
+
+        static::assertSame(0.3, $sum->apply($row1, $rows, flow_context()));
+    }
+
+    public function test_window_function_sum_with_exact_mode_from_literal(): void
+    {
+        $rows = rows(
+            $row1 = row(int_entry('id', 1), float_entry('value', 0.1)),
+            row(int_entry('id', 2), float_entry('value', 0.2)),
+        );
+
+        $sum = sum(ref('value'), exact: lit(true))->over(window()->orderBy(ref('id')->desc()));
+
+        static::assertSame(0.3, $sum->apply($row1, $rows, flow_context()));
+    }
+
     public function test_window_function_sum_with_missing_reference_in_strict_mode(): void
     {
         $this->expectException(InvalidArgumentException::class);