Fix bugs in aggregate operator and null column (#615)

Fix bug where insert comparer gave wrong result and also where updating null column increased its counter

src/FlowtideDotNet.Core/ColumnStore/Column.cs

@@ -348,11 +348,8 @@ public void UpdateAt<T>(in int index, in T value)
             }
             else if (value.Type == ArrowTypeId.Null)
             {
-                if (_type == ArrowTypeId.Null)
-                {
-                    _nullCounter++;
-                }
-                else
+                // If it is null and being updated to null nothing needs to be updated
+                if (_type != ArrowTypeId.Null)
                 {
                     CheckNullInitialization();
                     if (_validityList.Get(index))

src/FlowtideDotNet.Core/Operators/Aggregate/Column/AggregateInsertComparer.cs

@@ -55,7 +55,7 @@ public int FindIndex(in ColumnRowReference key, in AggregateKeyStorageContainer
                 key.referenceBatch.Columns[i].GetValueAt(key.RowIndex, dataValueContainer, default);
                 var (low, high) = keyContainer._data.Columns[i].SearchBoundries(dataValueContainer, start, end, default);
 
-                if (low != 0)
+                if (low < 0)
                 {
                     return low;
                 }

tests/FlowtideDotNet.AcceptanceTests/AggregateTests.cs

@@ -103,6 +103,34 @@ GROUP BY userkey
                 .Select(x => new { Userkey = x.Key, Sum = (double)x.Sum(y => y.OrderKey), Count = x.Count() }));
         }
 
+        /// <summary>
+        /// Test case to solve bug when using multiple aggregates and group by's
+        /// </summary>
+        /// <returns></returns>
+        [Fact]
+        public async Task MultipleMaxAggregates()
+        {
+            GenerateData();
+            await StartStream(@"
+                INSERT INTO output 
+                SELECT 
+                    userkey, max(orderkey), max(orderkey)
+                FROM orders
+                GROUP BY userkey, orderkey
+                ");
+            await WaitForUpdate();
+
+            for (int i = 0; i < 10; i++)
+            {
+                GenerateData();
+                await WaitForUpdate();
+            }
+
+            AssertCurrentDataEqual(Orders
+                .GroupBy(x => $"{x.UserKey}:{x.OrderKey}")
+                .Select(x => new { Userkey = int.Parse(x.Key.Substring(0, x.Key.IndexOf(':'))), Max1 = (double)x.Max(y => y.OrderKey), Max2 = x.Max(y => y.OrderKey) }));
+        }
+
         [Fact]
         public async Task HavingSameAggregate()
         {

tests/FlowtideDotNet.Core.Tests/ColumnStore/NullColumnTests.cs

@@ -72,5 +72,17 @@ public void TestCopy()
 
             Assert.Equal(1000, copy.Count);
         }
+
+        [Fact]
+        public void TestUpdateNullColumn()
+        {
+            Column column = new Column(GlobalMemoryManager.Instance);
+
+            column.Add(NullValue.Instance);
+
+            column.UpdateAt(0, NullValue.Instance);
+
+            Assert.Single(column);
+        }
     }
 }