From 6207d6137db2308d4e71b8853cce98d9703be796 Mon Sep 17 00:00:00 2001
From: Gagan Goel <gagan.nith@gmail.com>
Date: Tue, 8 Oct 2019 10:23:56 -0400
Subject: [PATCH] Handle comparison of two negative TIME values properly in the
 ORDER BY clause.

---
 utils/windowfunction/idborderby.cpp | 57 ++++++++++++++++++++++++++++-
 utils/windowfunction/idborderby.h   |  9 +++++
 2 files changed, 65 insertions(+), 1 deletion(-)

diff --git a/utils/windowfunction/idborderby.cpp b/utils/windowfunction/idborderby.cpp
index 302241047..f6e88bd7e 100644
--- a/utils/windowfunction/idborderby.cpp
+++ b/utils/windowfunction/idborderby.cpp
@@ -237,6 +237,55 @@ int LongDoubleCompare::operator()(IdbCompare* l, Row::Pointer r1, Row::Pointer r
 }
 
 
+int TimeCompare::operator()(IdbCompare* l, Row::Pointer r1, Row::Pointer r2)
+{
+    l->row1().setData(r1);
+    l->row2().setData(r2);
+
+    bool b1 = l->row1().isNullValue(fSpec.fIndex);
+    bool b2 = l->row2().isNullValue(fSpec.fIndex);
+
+    int ret = 0;
+
+    if (b1 == true || b2 == true)
+    {
+        if (b1 == false && b2 == true)
+            ret = fSpec.fNf;
+        else if (b1 == true && b2 == false)
+            ret = -fSpec.fNf;
+    }
+    else
+    {
+        int64_t v1 = l->row1().getIntField(fSpec.fIndex);
+        int64_t v2 = l->row2().getIntField(fSpec.fIndex);
+
+        // ((int64_t) -00:00:26) > ((int64_t) -00:00:25)
+        // i.e. For 2 negative TIME values, we invert the order of
+        // comparison operations to force "-00:00:26" to appear before
+        // "-00:00:25" in ascending order.
+        if (v1 < 0 && v2 < 0)
+        {
+            // Unset the MSB.
+            v1 &= ~(1ULL << 63);
+            v2 &= ~(1ULL << 63);
+            if (v1 < v2)
+                ret = fSpec.fAsc;
+            else if (v1 > v2)
+                ret = -fSpec.fAsc;
+        }
+        else
+        {
+            if (v1 > v2)
+                ret = fSpec.fAsc;
+            else if (v1 < v2)
+                ret = -fSpec.fAsc;
+        }
+    }
+
+    return ret;
+}
+
+
 bool CompareRule::less(Row::Pointer r1, Row::Pointer r2)
 {
     for (vector<Compare*>::iterator i = fCompares.begin(); i != fCompares.end(); i++)
@@ -268,7 +317,6 @@ void CompareRule::compileRules(const std::vector<IdbSortSpec>& spec, const rowgr
             case CalpontSystemCatalog::BIGINT:
             case CalpontSystemCatalog::DECIMAL:
             case CalpontSystemCatalog::UDECIMAL:
-            case CalpontSystemCatalog::TIME:
             {
                 Compare* c = new IntCompare(*i);
                 fCompares.push_back(c);
@@ -326,6 +374,13 @@ void CompareRule::compileRules(const std::vector<IdbSortSpec>& spec, const rowgr
                 break;
             }
 
+            case CalpontSystemCatalog::TIME:
+            {
+                Compare* c = new TimeCompare(*i);
+                fCompares.push_back(c);
+                break;
+            }
+
             default:
             {
                 break;
diff --git a/utils/windowfunction/idborderby.h b/utils/windowfunction/idborderby.h
index 4453828ba..b11b9c44e 100644
--- a/utils/windowfunction/idborderby.h
+++ b/utils/windowfunction/idborderby.h
@@ -138,6 +138,15 @@ public:
 };
 
 
+class TimeCompare : public Compare
+{
+public:
+    TimeCompare(const IdbSortSpec& spec) : Compare(spec) {}
+
+    int operator()(IdbCompare*, rowgroup::Row::Pointer, rowgroup::Row::Pointer);
+};
+
+
 class CompareRule
 {
 public: