support fromUnixTime and dateFormat (#629) (#634)

pingcap · Apr 14, 2020 · 364b0cf · 364b0cf
1 parent a7f253a
commit 364b0cf
Show file tree

Hide file tree

Showing 17 changed files with 853 additions and 137 deletions.
diff --git a/dbms/src/Common/MyTime.cpp b/dbms/src/Common/MyTime.cpp
diff --git a/dbms/src/Common/MyTime.h b/dbms/src/Common/MyTime.h
@@ -9,7 +9,16 @@ namespace DB
 struct MyTimeBase
 {
 
-    static const UInt64 YMD_MASK = ~((1ull << 41) -1);
+    static const UInt64 YMD_MASK = ~((1ull << 41) - 1);
+
+    // weekBehaviourMondayFirst set Monday as first day of week; otherwise Sunday is first day of week
+    static const UInt32 WEEK_BEHAVIOR_MONDAY_FIRST = 1;
+    // If set, Week is in range 1-53, otherwise Week is in range 0-53.
+    // Note that this flag is only relevant if WEEK_JANUARY is not set
+    static const UInt32 WEEK_BEHAVIOR_YEAR = 2;
+    // If not set, Weeks are numbered according to ISO 8601:1988.
+    // If set, the week that contains the first 'first-day-of-week' is week 1.
+    static const UInt32 WEEK_BEHAVIOR_FIRST_WEEKDAY = 4;
 
     enum MyTimeType : UInt8
     {
@@ -39,6 +48,15 @@ struct MyTimeBase
     // See http://dev.mysql.com/doc/refman/5.7/en/date-and-time-functions.html#function_date-format
     String dateFormat(const String & layout) const;
 
+    // returns the week day of current date(0 as sunday)
+    int weekDay() const;
+    // the following methods are port from TiDB
+    int yearDay() const;
+
+    int week(UInt32 mode) const;
+
+    std::tuple<int, int> calcWeek(UInt32 mode) const;
+
 protected:
     void convertDateFormat(char c, String & result) const;
 };
@@ -69,4 +87,9 @@ void convertTimeZone(UInt64 from_time, UInt64 & to_time, const DateLUTImpl & tim
 
 void convertTimeZoneByOffset(UInt64 from_time, UInt64 & to_time, Int64 offset, const DateLUTImpl & time_zone);
 
+int calcDayNum(int year, int month, int day);
+
+size_t maxFormattedDateTimeStringLength(const String & format);
+
+
 } // namespace DB
diff --git a/dbms/src/Debug/dbgFuncCoprocessor.cpp b/dbms/src/Debug/dbgFuncCoprocessor.cpp
@@ -265,6 +265,28 @@ void compileExpr(const DAGSchema & input, ASTPtr ast, tipb::Expr * expr, std::un
             }
             return;
         }
+        else if (func_name_lowercase == "from_unixtime")
+        {
+            if (func->arguments->children.size() == 1)
+            {
+                expr->set_sig(tipb::ScalarFuncSig::FromUnixTime1Arg);
+                auto * ft = expr->mutable_field_type();
+                ft->set_tp(TiDB::TypeDatetime);
+                ft->set_decimal(6);
+            }
+            else
+            {
+                expr->set_sig(tipb::ScalarFuncSig::FromUnixTime2Arg);
+                auto * ft = expr->mutable_field_type();
+                ft->set_tp(TiDB::TypeString);
+            }
+        }
+        else if (func_name_lowercase == "date_format")
+        {
+            expr->set_sig(tipb::ScalarFuncSig::DateFormatSig);
+            auto * ft = expr->mutable_field_type();
+            ft->set_tp(TiDB::TypeString);
+        }
         else
         {
             throw Exception("Unsupported function: " + func_name_lowercase, ErrorCodes::LOGICAL_ERROR);

diff --git a/dbms/src/Flash/Coprocessor/DAGDriver.cpp b/dbms/src/Flash/Coprocessor/DAGDriver.cpp
@@ -38,6 +38,7 @@ DAGDriver::DAGDriver(Context & context_, const tipb::DAGRequest & dag_request_,
     if (schema_ver)
         // schema_ver being 0 means TiDB/TiSpark hasn't specified schema version.
         context.setSetting("schema_version", schema_ver);
+    context.getTimezoneInfo().resetByDAGRequest(dag_request);
 }
 
 void DAGDriver::execute()

diff --git a/dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.cpp b/dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.cpp
@@ -379,21 +379,12 @@ void DAGExpressionAnalyzer::appendFinalProject(ExpressionActionsChain & chain, c
     }
 }
 
-void constructTZExpr(tipb::Expr & tz_expr, const tipb::DAGRequest & rqst, bool from_utc)
+void constructTZExpr(tipb::Expr & tz_expr, const TimezoneInfo & dag_timezone_info, bool from_utc)
 {
-    if (rqst.has_time_zone_name() && rqst.time_zone_name().length() > 0)
-        constructStringLiteralTiExpr(tz_expr, rqst.time_zone_name());
+    if (dag_timezone_info.is_name_based)
+        constructStringLiteralTiExpr(tz_expr, dag_timezone_info.timezone_name);
     else
-        constructInt64LiteralTiExpr(tz_expr, from_utc ? rqst.time_zone_offset() : -rqst.time_zone_offset());
-}
-
-bool hasMeaningfulTZInfo(const tipb::DAGRequest & rqst)
-{
-    if (rqst.has_time_zone_name() && rqst.time_zone_name().length() > 0)
-        return rqst.time_zone_name() != "UTC";
-    if (rqst.has_time_zone_offset())
-        return rqst.has_time_zone_offset() != 0;
-    return false;
+        constructInt64LiteralTiExpr(tz_expr, from_utc ? dag_timezone_info.timezone_offset : -dag_timezone_info.timezone_offset);
 }
 
 String DAGExpressionAnalyzer::appendTimeZoneCast(
@@ -414,20 +405,18 @@ String DAGExpressionAnalyzer::appendTimeZoneCast(
 // useless casts to all the timestamp columns, in order to avoid redundant cast, when cast the ts
 // column to the columns with session-level timezone info, the original ts columns with UTC timezone
 // are still kept, and the InterpreterDAG will choose the correct column based on encode type
-bool DAGExpressionAnalyzer::appendTimeZoneCastsAfterTS(
-    ExpressionActionsChain & chain, std::vector<bool> is_ts_column, const tipb::DAGRequest & rqst)
+bool DAGExpressionAnalyzer::appendTimeZoneCastsAfterTS(ExpressionActionsChain & chain, std::vector<bool> is_ts_column)
 {
-    if (!hasMeaningfulTZInfo(rqst))
+    if (context.getTimezoneInfo().is_utc_timezone)
         return false;
 
     bool ret = false;
     initChain(chain, getCurrentInputColumns());
     ExpressionActionsPtr actions = chain.getLastActions();
     tipb::Expr tz_expr;
-    constructTZExpr(tz_expr, rqst, true);
+    constructTZExpr(tz_expr, context.getTimezoneInfo(), true);
     String tz_col;
-    String func_name
-        = rqst.has_time_zone_name() && rqst.time_zone_name().length() > 0 ? "ConvertTimeZoneFromUTC" : "ConvertTimeZoneByOffset";
+    String func_name = context.getTimezoneInfo().is_name_based ? "ConvertTimeZoneFromUTC" : "ConvertTimeZoneByOffset";
     for (size_t i = 0; i < is_ts_column.size(); i++)
     {
         if (is_ts_column[i])
@@ -444,19 +433,18 @@ bool DAGExpressionAnalyzer::appendTimeZoneCastsAfterTS(
 }
 
 void DAGExpressionAnalyzer::appendAggSelect(
-    ExpressionActionsChain & chain, const tipb::Aggregation & aggregation, const tipb::DAGRequest & rqst, bool keep_session_timezone_info)
+    ExpressionActionsChain & chain, const tipb::Aggregation & aggregation, bool keep_session_timezone_info)
 {
     initChain(chain, getCurrentInputColumns());
     bool need_update_aggregated_columns = false;
     std::vector<NameAndTypePair> updated_aggregated_columns;
     ExpressionActionsChain::Step step = chain.steps.back();
-    bool need_append_timezone_cast = !keep_session_timezone_info && hasMeaningfulTZInfo(rqst);
+    bool need_append_timezone_cast = !keep_session_timezone_info && !context.getTimezoneInfo().is_utc_timezone;
     tipb::Expr tz_expr;
     if (need_append_timezone_cast)
-        constructTZExpr(tz_expr, rqst, false);
+        constructTZExpr(tz_expr, context.getTimezoneInfo(), false);
     String tz_col;
-    String tz_cast_func_name
-        = rqst.has_time_zone_name() && rqst.time_zone_name().length() > 0 ? "ConvertTimeZoneToUTC" : "ConvertTimeZoneByOffset";
+    String tz_cast_func_name = context.getTimezoneInfo().is_name_based ? "ConvertTimeZoneToUTC" : "ConvertTimeZoneByOffset";
     for (Int32 i = 0; i < aggregation.agg_func_size(); i++)
     {
         String & name = aggregated_columns[i].name;

diff --git a/dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.h b/dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.h
@@ -42,8 +42,7 @@ class DAGExpressionAnalyzer : private boost::noncopyable
     void appendOrderBy(ExpressionActionsChain & chain, const tipb::TopN & topN, Strings & order_column_names);
     void appendAggregation(ExpressionActionsChain & chain, const tipb::Aggregation & agg, Names & aggregate_keys,
         AggregateDescriptions & aggregate_descriptions);
-    void appendAggSelect(
-        ExpressionActionsChain & chain, const tipb::Aggregation & agg, const tipb::DAGRequest & rqst, bool keep_session_timezone_info);
+    void appendAggSelect(ExpressionActionsChain & chain, const tipb::Aggregation & agg, bool keep_session_timezone_info);
     String appendCastIfNeeded(const tipb::Expr & expr, ExpressionActionsPtr & actions, const String & expr_name, bool explicit_cast);
     String alignReturnType(const tipb::Expr & expr, ExpressionActionsPtr & actions, const String & expr_name, bool force_uint8);
     void initChain(ExpressionActionsChain & chain, const std::vector<NameAndTypePair> & columns) const
@@ -66,7 +65,7 @@ class DAGExpressionAnalyzer : private boost::noncopyable
     void makeExplicitSetForIndex(const tipb::Expr & expr, const ManageableStoragePtr & storage);
     String applyFunction(const String & func_name, const Names & arg_names, ExpressionActionsPtr & actions);
     Int32 getImplicitCastCount() { return implicit_cast_count; };
-    bool appendTimeZoneCastsAfterTS(ExpressionActionsChain & chain, std::vector<bool> is_ts_column, const tipb::DAGRequest & rqst);
+    bool appendTimeZoneCastsAfterTS(ExpressionActionsChain & chain, std::vector<bool> is_ts_column);
     String appendTimeZoneCast(const String & tz_col, const String & ts_col, const String & func_name, ExpressionActionsPtr & actions);
     DAGPreparedSets & getPreparedSets() { return prepared_sets; }
     String convertToUInt8(ExpressionActionsPtr & actions, const String & column_name);

diff --git a/dbms/src/Flash/Coprocessor/DAGUtils.cpp b/dbms/src/Flash/Coprocessor/DAGUtils.cpp
@@ -674,7 +674,7 @@ std::unordered_map<tipb::ScalarFuncSig, String> scalar_func_map({
     //{tipb::ScalarFuncSig::JsonKeys2ArgsSig, "cast"},
     //{tipb::ScalarFuncSig::JsonValidStringSig, "cast"},
 
-    //{tipb::ScalarFuncSig::DateFormatSig, "cast"},
+    {tipb::ScalarFuncSig::DateFormatSig, "dateFormat"},
     //{tipb::ScalarFuncSig::DateLiteral, "cast"},
     //{tipb::ScalarFuncSig::DateDiff, "cast"},
     //{tipb::ScalarFuncSig::NullTimeDiff, "cast"},
@@ -791,8 +791,7 @@ std::unordered_map<tipb::ScalarFuncSig, String> scalar_func_map({
     //{tipb::ScalarFuncSig::StrToDateDate, "cast"},
     //{tipb::ScalarFuncSig::StrToDateDatetime, "cast"},
     //{tipb::ScalarFuncSig::StrToDateDuration, "cast"},
-    //{tipb::ScalarFuncSig::FromUnixTime1Arg, "cast"},
-    //{tipb::ScalarFuncSig::FromUnixTime2Arg, "cast"},
+    {tipb::ScalarFuncSig::FromUnixTime1Arg, "fromUnixTime"}, {tipb::ScalarFuncSig::FromUnixTime2Arg, "fromUnixTime"},
     //{tipb::ScalarFuncSig::ExtractDatetime, "cast"},
     //{tipb::ScalarFuncSig::ExtractDuration, "cast"},
 

diff --git a/dbms/src/Flash/Coprocessor/InterpreterDAG.cpp b/dbms/src/Flash/Coprocessor/InterpreterDAG.cpp
@@ -505,7 +505,7 @@ bool InterpreterDAG::addTimeZoneCastAfterTS(std::vector<bool> & is_ts_column, Pi
         return false;
 
     ExpressionActionsChain chain;
-    if (analyzer->appendTimeZoneCastsAfterTS(chain, is_ts_column, dag.getDAGRequest()))
+    if (analyzer->appendTimeZoneCastsAfterTS(chain, is_ts_column))
     {
         pipeline.transform([&](auto & stream) { stream = std::make_shared<ExpressionBlockInputStream>(stream, chain.getLastActions()); });
         return true;
@@ -536,7 +536,7 @@ InterpreterDAG::AnalysisResult InterpreterDAG::analyzeExpressions()
         chain.clear();
 
         // add cast if type is not match
-        analyzer->appendAggSelect(chain, dag.getAggregation(), dag.getDAGRequest(), keep_session_timezone_info);
+        analyzer->appendAggSelect(chain, dag.getAggregation(), keep_session_timezone_info);
         //todo use output_offset to reconstruct the final project columns
         for (auto & element : analyzer->getCurrentInputColumns())
         {

diff --git a/dbms/src/Functions/FunctionsConversion.cpp b/dbms/src/Functions/FunctionsConversion.cpp
@@ -88,6 +88,9 @@ void registerFunctionsConversion(FunctionFactory & factory)
     factory.registerFunction<FunctionConvert<DataTypeInterval, NameToIntervalWeek, PositiveMonotonicity>>();
     factory.registerFunction<FunctionConvert<DataTypeInterval, NameToIntervalMonth, PositiveMonotonicity>>();
     factory.registerFunction<FunctionConvert<DataTypeInterval, NameToIntervalYear, PositiveMonotonicity>>();
+
+    factory.registerFunction<FunctionFromUnixTime>();
+    factory.registerFunction<FunctionDateFormat>();
 }
 
 }