From ff9a1dea21cc0983f3b3b96fc22360ca145db901 Mon Sep 17 00:00:00 2001
From: xufei <xufei@pingcap.com>
Date: Thu, 5 Sep 2019 14:05:24 +0800
Subject: [PATCH] Flash-473 optimize date and datetime comparison (#221)

* support udf like with 3 arguments

* address comments

* add some comments

* Flash-473 optimize date and datetime comparison

* address comments
---
 dbms/src/Functions/FunctionsComparison.h | 79 +++++++++++++++++-------
 1 file changed, 57 insertions(+), 22 deletions(-)

diff --git a/dbms/src/Functions/FunctionsComparison.h b/dbms/src/Functions/FunctionsComparison.h
index f94fb6f6a2a..4ddef605914 100644
--- a/dbms/src/Functions/FunctionsComparison.h
+++ b/dbms/src/Functions/FunctionsComparison.h
@@ -133,6 +133,16 @@ inline time_t dateToDateTime(UInt32 date_data)
     return DateLUT::instance().makeDateTime(local_date.year(), local_date.month(), local_date.day(), 0, 0, 0);
 }
 
+inline std::tuple<DayNum_t, bool> dateTimeToDate(time_t time_data)
+{
+    // todo use timezone info
+    auto & date_lut = DateLUT::instance();
+    auto truncated = date_lut.toHour(time_data) != 0 || date_lut.toMinute(time_data) != 0 || date_lut.toSecond(time_data) != 0;
+    auto values = date_lut.getValues(time_data);
+    auto day_num = date_lut.makeDayNum(values.year, values.month, values.day_of_month);
+    return std::make_tuple(day_num, truncated);
+}
+
 
 template <typename A, typename B, template <typename, typename> class Op, bool is_left_date>
 struct DateDateTimeComparisonImpl
@@ -175,18 +185,31 @@ struct DateDateTimeComparisonImpl
         }
         else
         {
-            using OpType = B;
-            size_t size = a.size();
-            const A * a_pos = &a[0];
-            UInt8 * c_pos = &c[0];
-            const A * a_end = a_pos + size;
-
-            while (a_pos < a_end)
+            // date vector with datetime constant
+            // first check if datetime constant can be convert to date constant
+            bool truncated;
+            DayNum_t date_num;
+            std::tie(date_num, truncated) = dateTimeToDate((time_t) b);
+            if (!truncated)
             {
-                time_t date_time = dateToDateTime(*a_pos);
-                *c_pos = Op<OpType, OpType>::apply((OpType)date_time, b);
-                ++a_pos;
-                ++c_pos;
+                using OpType = A;
+                NumComparisonImpl<OpType, OpType, Op<OpType, OpType>>::vector_constant(a, (OpType) date_num, c);
+            }
+            else
+            {
+                using OpType = B;
+                size_t size = a.size();
+                const A *a_pos = &a[0];
+                UInt8 *c_pos = &c[0];
+                const A *a_end = a_pos + size;
+
+                while (a_pos < a_end)
+                {
+                    time_t date_time = dateToDateTime(*a_pos);
+                    *c_pos = Op<OpType, OpType>::apply((OpType) date_time, b);
+                    ++a_pos;
+                    ++c_pos;
+                }
             }
         }
     }
@@ -202,18 +225,30 @@ struct DateDateTimeComparisonImpl
         }
         else
         {
-            using OpType = A;
-            size_t size = b.size();
-            const B * b_pos = &b[0];
-            UInt8 * c_pos = &c[0];
-            const B * b_end = b_pos + size;
-
-            while (b_pos < b_end)
+            // datetime constant with date vector
+            bool truncated;
+            DayNum_t date_num;
+            std::tie(date_num, truncated) = dateTimeToDate((time_t) a);
+            if (!truncated)
             {
-                time_t date_time = dateToDateTime(*b_pos);
-                *c_pos = Op<OpType, OpType>::apply(a, (OpType)date_time);
-                ++b_pos;
-                ++c_pos;
+                using OpType = B;
+                NumComparisonImpl<OpType, OpType, Op<OpType, OpType>>::vector_constant((OpType)a, date_num, c);
+            }
+            else
+            {
+                using OpType = A;
+                size_t size = b.size();
+                const B *b_pos = &b[0];
+                UInt8 *c_pos = &c[0];
+                const B *b_end = b_pos + size;
+
+                while (b_pos < b_end)
+                {
+                    time_t date_time = dateToDateTime(*b_pos);
+                    *c_pos = Op<OpType, OpType>::apply(a, (OpType) date_time);
+                    ++b_pos;
+                    ++c_pos;
+                }
             }
         }
     }