Optimized windowed partitioned aggregators.

Author: RyanLamansky

SqlServerSimulator.Tests/WindowFrameTests.cs

@@ -405,4 +405,102 @@ insert d values (1), (3), (5)
         // Running averages: 1/1=1, 4/2=2, 9/3=3.
         CollectionAssert.AreEqual(new[] { (1, 1), (3, 2), (5, 3) }, values);
     }
+
+    // === Sliding frames exercise the incremental Remove path per aggregator ===
+
+    [TestMethod]
+    public void Sum_RowsBetweenCurrentRowAndUnboundedFollowing_ReverseRunningTotal()
+    {
+        // Start advances (CURRENT ROW) with the end pinned at the partition's
+        // last row — every step removes the leaving row from the slider.
+        using var connection = SeededTies();
+        using var reader = connection.CreateCommand(
+            "select id, sum(v) over(partition by grp order by id rows between current row and unbounded following) from t order by grp, id").ExecuteReader();
+        var byId = new Dictionary<int, int>();
+        while (reader.Read())
+            byId[reader.GetInt32(0)] = reader.GetInt32(1);
+        // grp 1 v=10,20,20,30 → suffix sums 80,70,50,30.
+        AreEqual(80, byId[1]);
+        AreEqual(70, byId[2]);
+        AreEqual(50, byId[3]);
+        AreEqual(30, byId[4]);
+        // grp 2 v=5,5,50 → 60,55,50.
+        AreEqual(60, byId[5]);
+        AreEqual(55, byId[6]);
+        AreEqual(50, byId[7]);
+    }
+
+    [TestMethod]
+    public void Min_SlidingFrame_DropsLeavingExtreme()
+    {
+        // Ascending values with a trailing 2-PRECEDING frame: the current
+        // minimum repeatedly leaves the window, so the removable multiset must
+        // surface the next-smallest survivor rather than a stale extreme.
+        using var connection = new Simulation().CreateOpenConnection();
+        _ = connection.CreateCommand("""
+            create table m (id int, v int);
+            insert m values (1,10), (2,20), (3,30), (4,40), (5,50)
+            """).ExecuteNonQuery();
+        using var reader = connection.CreateCommand(
+            "select id, min(v) over(order by id rows between 2 preceding and current row) from m order by id").ExecuteReader();
+        var values = new List<int>();
+        while (reader.Read())
+            values.Add(reader.GetInt32(1));
+        // Frames {10},{10,20},{10,20,30},{20,30,40},{30,40,50} → mins 10,10,10,20,30.
+        CollectionAssert.AreEqual(new[] { 10, 10, 10, 20, 30 }, values);
+    }
+
+    [TestMethod]
+    public void Avg_SlidingFrame_RemovesLeavingRows()
+    {
+        using var connection = new Simulation().CreateOpenConnection();
+        _ = connection.CreateCommand("""
+            create table a (id int, v int);
+            insert a values (1,10), (2,20), (3,30), (4,40)
+            """).ExecuteNonQuery();
+        using var reader = connection.CreateCommand(
+            "select id, avg(v) over(order by id rows between 1 preceding and current row) from a order by id").ExecuteReader();
+        var values = new List<int>();
+        while (reader.Read())
+            values.Add(reader.GetInt32(1));
+        // Pairs {10}=10, {10,20}=15, {20,30}=25, {30,40}=35.
+        CollectionAssert.AreEqual(new[] { 10, 15, 25, 35 }, values);
+    }
+
+    [TestMethod]
+    public void VarP_SlidingFrame_SubtractsMoments()
+    {
+        using var connection = new Simulation().CreateOpenConnection();
+        _ = connection.CreateCommand("""
+            create table s (id int, v int);
+            insert s values (1,2), (2,4), (3,6), (4,8)
+            """).ExecuteNonQuery();
+        using var reader = connection.CreateCommand(
+            "select id, varp(v) over(order by id rows between 1 preceding and current row) from s order by id").ExecuteReader();
+        var values = new List<double>();
+        while (reader.Read())
+            values.Add(reader.GetDouble(1));
+        // {2}→0, {2,4}→1, {4,6}→1, {6,8}→1 (sum / sum-of-squares moments subtract).
+        CollectionAssert.AreEqual(new[] { 0.0, 1.0, 1.0, 1.0 }, values);
+    }
+
+    [TestMethod]
+    public void ChecksumAgg_SlidingFrame_MatchesDirectAggregate()
+    {
+        // XOR is its own inverse: removing the leaving row must leave exactly
+        // the state of aggregating the survivors directly.
+        using var connection = new Simulation().CreateOpenConnection();
+        _ = connection.CreateCommand("""
+            create table c (id int, v int);
+            insert c values (1,11), (2,22), (3,33), (4,44)
+            """).ExecuteNonQuery();
+        using var reader = connection.CreateCommand(
+            "select id, checksum_agg(v) over(order by id rows between 1 preceding and current row) from c order by id").ExecuteReader();
+        var byId = new Dictionary<int, int>();
+        while (reader.Read())
+            byId[reader.GetInt32(0)] = reader.GetInt32(1);
+        // Row 3's frame is {22, 33}; the slider reached it by removing 11.
+        var expected = (int)connection.CreateCommand("select checksum_agg(v) from c where id in (2,3)").ExecuteScalar()!;
+        AreEqual(expected, byId[3]);
+    }
 }

SqlServerSimulator/Parser/Aggregator.cs

@@ -23,24 +23,52 @@ internal abstract class Aggregator
 
     public abstract SqlValue Result();
 
+    /// <summary>
+    /// Whether this instance supports <see cref="Remove"/> — i.e. can undo an
+    /// earlier <see cref="Add"/> so a sliding window frame can be maintained
+    /// incrementally. Defaults to <c>false</c>; the window executor only calls
+    /// <see cref="Remove"/> after checking this, and a removable instance is
+    /// requested via <see cref="Create"/>'s <c>removable</c> flag (some
+    /// aggregators need a heavier removal-capable representation).
+    /// </summary>
+    public virtual bool CanRemove => false;
+
+    /// <summary>
+    /// Reverses one prior <see cref="Add"/> of <paramref name="value"/>. Only
+    /// valid when <see cref="CanRemove"/>; the value must have been added
+    /// before. NULL inputs are no-ops on the matching <see cref="Add"/>, so
+    /// removing a NULL is likewise a no-op.
+    /// </summary>
+    public virtual void Remove(SqlValue value) =>
+        throw new NotSupportedException($"{this.GetType().Name} does not support incremental removal.");
+
     /// <summary>
     /// Builds a fresh aggregator for the given expression. Caller supplies
     /// the operand's resolved type (some aggregators use it to choose an
     /// accumulator) and the aggregate's overall result type (used to size
     /// the returned <see cref="SqlValue"/>). The Selection executor calls
     /// this once per group; aggregators don't outlive a single group.
+    /// <para>
+    /// <paramref name="removable"/> asks for a representation that supports
+    /// <see cref="Remove"/> (set by the window executor for sliding frames
+    /// whose start advances). It only changes the shape of aggregators that
+    /// would otherwise be cheaper without removal support — notably MIN / MAX,
+    /// which keeps a single running extreme unless asked to track a removable
+    /// multiset. Aggregators that are intrinsically removable (or never) ignore
+    /// it.
+    /// </para>
     /// </summary>
-    public static Aggregator Create(AggregateExpression aggregate, SqlType operandType, SqlType resultType) => aggregate.Kind switch
+    public static Aggregator Create(AggregateExpression aggregate, SqlType operandType, SqlType resultType, bool removable = false) => aggregate.Kind switch
     {
         AggregateKind.Count => new CountAggregator(isStar: aggregate.Operand is null, isBigCount: false, distinct: aggregate.Distinct),
         AggregateKind.CountBig => new CountAggregator(isStar: aggregate.Operand is null, isBigCount: true, distinct: aggregate.Distinct),
         AggregateKind.ApproxCountDistinct => new CountAggregator(isStar: false, isBigCount: true, distinct: true),
         AggregateKind.Max => operandType.IsLob || operandType is BitSqlType
             ? throw SimulatedSqlException.OperandDataTypeInvalid(operandType, "max")
-            : new MinMaxAggregator(resultType, isMax: true),
+            : new MinMaxAggregator(resultType, isMax: true, removable),
         AggregateKind.Min => operandType.IsLob || operandType is BitSqlType
             ? throw SimulatedSqlException.OperandDataTypeInvalid(operandType, "min")
-            : new MinMaxAggregator(resultType, isMax: false),
+            : new MinMaxAggregator(resultType, isMax: false, removable),
         AggregateKind.Sum => SumAggregator.Create(resultType, aggregate.Distinct),
         AggregateKind.Avg => AverageAggregator.Create(resultType, aggregate.Distinct),
         AggregateKind.Stdev or AggregateKind.StdevP or AggregateKind.Var or AggregateKind.VarP => new StatisticalAggregator(aggregate.Kind),

SqlServerSimulator/Parser/Aggregators/ChecksumAggAggregator.cs

@@ -26,5 +26,17 @@ public override void Add(SqlValue value)
         this.folded ^= value.GetHashCode();
     }
 
+    // XOR is its own inverse, so re-folding a value removes it — the window
+    // frame slides incrementally. (Two equal values XOR-cancel each other, which
+    // is exactly the multiset removal semantic.)
+    public override bool CanRemove => true;
+
+    public override void Remove(SqlValue value)
+    {
+        if (value.IsNull)
+            return;
+        this.folded ^= value.GetHashCode();
+    }
+
     public override SqlValue Result() => SqlValue.FromInt32(this.folded);
 }

SqlServerSimulator/Parser/Aggregators/CountAggregator.cs

@@ -39,6 +39,19 @@ public override void Add(SqlValue value)
         }
     }
 
+    // A plain row/value count decrements cleanly; the DISTINCT form can't,
+    // since the dedup set doesn't carry per-value multiplicity. (COUNT(DISTINCT)
+    // / APPROX_COUNT_DISTINCT are illegal with OVER, so the window path never
+    // requests a removable distinct count.)
+    public override bool CanRemove => this.seen is null;
+
+    public override void Remove(SqlValue value)
+    {
+        if (!isStar && value.IsNull)
+            return;
+        this.count--;
+    }
+
     public override SqlValue Result() => isBigCount
         ? SqlValue.FromInt64(this.count)
         : this.count > int.MaxValue

SqlServerSimulator/Parser/Aggregators/MinMaxAggregator.cs

@@ -8,26 +8,80 @@ namespace SqlServerSimulator.Parser.Aggregators;
 /// returns NULL of the operand type. The DISTINCT keyword is honored at
 /// parse time but doesn't affect the result (DISTINCT extremes equal plain
 /// extremes; SQL Server accepts the keyword as a no-op here).
+/// <para>
+/// A single running extreme can't be un-done (dropping the current extreme
+/// leaves the next one unknown), so the removable mode requested for sliding
+/// window frames keeps a directional multiset instead — ordered so the wanted
+/// extreme is always the first key, with per-value multiplicity for removal.
+/// GROUP BY and forward-cumulative windows never remove, so they keep the
+/// cheaper two-field running-extreme path.
+/// </para>
 /// </summary>
-internal sealed class MinMaxAggregator(SqlType resultType, bool isMax) : Aggregator
+internal sealed class MinMaxAggregator : Aggregator
 {
-    private SqlValue current = SqlValue.Null(resultType);
+    private readonly SqlType resultType;
+    private readonly bool isMax;
+    private readonly SortedDictionary<SqlValue, int>? multiset;
+
+    private SqlValue current;
     private bool sawAny;
 
+    public MinMaxAggregator(SqlType resultType, bool isMax, bool removable = false)
+    {
+        this.resultType = resultType;
+        this.isMax = isMax;
+        this.current = SqlValue.Null(resultType);
+        if (removable)
+        {
+            this.multiset = new SortedDictionary<SqlValue, int>(
+                Comparer<SqlValue>.Create(isMax ? static (a, b) => b.CompareTo(a) : static (a, b) => a.CompareTo(b)));
+        }
+    }
+
     public override void Add(SqlValue value)
     {
         if (value.IsNull)
             return;
+        if (this.multiset is { } bag)
+        {
+            _ = bag.TryGetValue(value, out var n);
+            bag[value] = n + 1;
+            return;
+        }
         if (!this.sawAny)
         {
             this.current = value;
             this.sawAny = true;
             return;
         }
         var cmp = value.CompareTo(this.current);
-        if ((isMax && cmp > 0) || (!isMax && cmp < 0))
+        if ((this.isMax && cmp > 0) || (!this.isMax && cmp < 0))
             this.current = value;
     }
 
-    public override SqlValue Result() => this.sawAny ? this.current : SqlValue.Null(resultType);
+    public override bool CanRemove => this.multiset is not null;
+
+    public override void Remove(SqlValue value)
+    {
+        if (value.IsNull)
+            return;
+        var bag = this.multiset!;
+        var n = bag[value];
+        if (n == 1)
+            _ = bag.Remove(value);
+        else
+            bag[value] = n - 1;
+    }
+
+    public override SqlValue Result()
+    {
+        if (this.multiset is { } bag)
+        {
+            foreach (var pair in bag)
+                return pair.Key;
+            return SqlValue.Null(this.resultType);
+        }
+
+        return this.sawAny ? this.current : SqlValue.Null(this.resultType);
+    }
 }

SqlServerSimulator/Parser/Aggregators/NumericAggregator.cs

@@ -50,6 +50,20 @@ public sealed override void Add(SqlValue value)
         this.Count++;
     }
 
+    // A running sum subtracts cleanly, so SUM / AVG slide incrementally — but
+    // only without DISTINCT, whose dedup set can't tell whether a removed value
+    // still has surviving duplicates. (DISTINCT is illegal with OVER anyway, so
+    // the window path never requests it.)
+    public sealed override bool CanRemove => !this.distinct;
+
+    public sealed override void Remove(SqlValue value)
+    {
+        if (value.IsNull)
+            return;
+        this.Accumulator -= Extract(value.CoerceTo(this.ResultType));
+        this.Count--;
+    }
+
     public sealed override SqlValue Result()
     {
         if (this.Count == 0)

SqlServerSimulator/Parser/Aggregators/StatisticalAggregator.cs

@@ -29,6 +29,20 @@ public override void Add(SqlValue value)
         this.sumOfSquares += x * x;
     }
 
+    // The classical sum / sum-of-squares moments subtract directly, so the
+    // statistical aggregates slide incrementally over a window frame.
+    public override bool CanRemove => true;
+
+    public override void Remove(SqlValue value)
+    {
+        if (value.IsNull)
+            return;
+        var x = value.CoerceTo(SqlType.Float).AsDouble;
+        this.count--;
+        this.sum -= x;
+        this.sumOfSquares -= x * x;
+    }
+
     public override SqlValue Result()
     {
         var isPopulation = kind is AggregateKind.StdevP or AggregateKind.VarP;