From b36d20ef6e5b24edf32bc74e59e91e6e84b6cbf5 Mon Sep 17 00:00:00 2001
From: Russ Frank <rf@oden.io>
Date: Thu, 4 Feb 2021 00:11:43 -0500
Subject: [PATCH] record seen first and last

---
 .../AggregateFunctionDeltaSum.h               | 21 +++++++++++++------
 1 file changed, 15 insertions(+), 6 deletions(-)

diff --git a/src/AggregateFunctions/AggregateFunctionDeltaSum.h b/src/AggregateFunctions/AggregateFunctionDeltaSum.h
index 95f97d079fd..e83ff6ea140 100644
--- a/src/AggregateFunctions/AggregateFunctionDeltaSum.h
+++ b/src/AggregateFunctions/AggregateFunctionDeltaSum.h
@@ -19,9 +19,11 @@ namespace DB
 template <typename T>
 struct AggregationFunctionDeltaSumData
 {
-    T sum{};
-    T last{};
-    T first{};
+    T sum = 0;
+    bool seen_last = false;
+    T last;
+    bool seen_first = false;
+    T first;
 };
 
 template <typename T>
@@ -50,26 +52,29 @@ public:
     {
         const T & value = (*columns[0])[row_num].get<T>();
 
-        if (this->data(place).last < value) {
+        if (this->data(place).last < value && this->data(place).seen_last) {
             this->data(place).sum += (value - this->data(place).last);
         }
 
         this->data(place).last = value;
+        this->data(place).seen_last = true;
 
-        if (this->data(place).first == 0) {
+        if (this->data(place).seen_first == false) {
             this->data(place).first = value;
+            this->data(place).seen_first = true;
         }
     }
 
     void ALWAYS_INLINE merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena *) const override
     {
-        if (this->data(place).last < this->data(rhs).first) {
+        if ((this->data(place).last < this->data(rhs).first) && this->data(place).seen_last && this->data(place).seen_first) {
             this->data(place).sum += this->data(rhs).sum + (this->data(rhs).first - this->data(place).last);
         } else {
             this->data(place).sum += this->data(rhs).sum;
         }
 
         this->data(place).last = this->data(rhs).last;
+        this->data(place).first = this->data(rhs).first;
     }
 
         void serialize(ConstAggregateDataPtr place, WriteBuffer & buf) const override
@@ -77,6 +82,8 @@ public:
         writeIntBinary(this->data(place).sum, buf);
         writeIntBinary(this->data(place).first, buf);
         writeIntBinary(this->data(place).last, buf);
+        writePODBinary<bool>(this->data(place).seen_first, buf);
+        writePODBinary<bool>(this->data(place).seen_last, buf);
     }
 
     void deserialize(AggregateDataPtr place, ReadBuffer & buf, Arena *) const override
@@ -84,6 +91,8 @@ public:
         readIntBinary(this->data(place).sum, buf);
         readIntBinary(this->data(place).first, buf);
         readIntBinary(this->data(place).last, buf);
+        readPODBinary<bool>(this->data(place).seen_first, buf);
+        readPODBinary<bool>(this->data(place).seen_last, buf);
     }
 
     void insertResultInto(AggregateDataPtr place, IColumn & to, Arena *) const override