Tiflash crash when scanning partition table with time column (#4563)

close #4526

dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.cpp

@@ -616,11 +616,11 @@ bool DAGExpressionAnalyzer::buildExtraCastsAfterTS(
 bool DAGExpressionAnalyzer::appendExtraCastsAfterTS(
     ExpressionActionsChain & chain,
     const std::vector<ExtraCastAfterTSMode> & need_cast_column,
-    const tipb::TableScan & table_scan)
+    const TiDBTableScan & table_scan)
 {
     auto & step = initAndGetLastStep(chain);
 
-    bool has_cast = buildExtraCastsAfterTS(step.actions, need_cast_column, table_scan.columns());
+    bool has_cast = buildExtraCastsAfterTS(step.actions, need_cast_column, table_scan.getColumns());
 
     for (auto & col : source_columns)
         step.required_output.push_back(col.name);

dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.h

@@ -23,6 +23,7 @@
 #include <Flash/Coprocessor/DAGQueryBlock.h>
 #include <Flash/Coprocessor/DAGSet.h>
 #include <Flash/Coprocessor/DAGUtils.h>
+#include <Flash/Coprocessor/TiDBTableScan.h>
 #include <Interpreters/AggregateDescription.h>
 #include <Interpreters/ExpressionActions.h>
 #include <Interpreters/ExpressionAnalyzer.h>
@@ -122,7 +123,7 @@ class DAGExpressionAnalyzer : private boost::noncopyable
     bool appendExtraCastsAfterTS(
         ExpressionActionsChain & chain,
         const std::vector<ExtraCastAfterTSMode> & need_cast_column,
-        const tipb::TableScan & table_scan);
+        const TiDBTableScan & table_scan);
 
     /// return true if some actions is needed
     bool appendJoinKeyAndJoinFilters(

dbms/src/Flash/Coprocessor/DAGQueryBlockInterpreter.cpp

@@ -93,7 +93,7 @@ bool addExtraCastsAfterTs(
     DAGExpressionAnalyzer & analyzer,
     const std::vector<ExtraCastAfterTSMode> & need_cast_column,
     ExpressionActionsChain & chain,
-    const tipb::TableScan & table_scan)
+    const TiDBTableScan & table_scan)
 {
     bool has_need_cast_column = false;
     for (auto b : need_cast_column)
@@ -309,7 +309,11 @@ void DAGQueryBlockInterpreter::handleTableScan(const TiDBTableScan & table_scan,
     FAIL_POINT_PAUSE(FailPoints::pause_after_copr_streams_acquired);
 
     /// handle timezone/duration cast for local and remote table scan.
-    executeCastAfterTableScan(storage_interpreter.is_need_add_cast_column, remote_read_streams_start_index, pipeline);
+    executeCastAfterTableScan(
+        table_scan,
+        storage_interpreter.is_need_add_cast_column,
+        remote_read_streams_start_index,
+        pipeline);
     recordProfileStreams(pipeline, query_block.source_name);
 
     /// handle pushed down filter for local and remote table scan.
@@ -356,6 +360,7 @@ void DAGQueryBlockInterpreter::executePushedDownFilter(
 }
 
 void DAGQueryBlockInterpreter::executeCastAfterTableScan(
+    const TiDBTableScan & table_scan,
     const std::vector<ExtraCastAfterTSMode> & is_need_add_cast_column,
     size_t remote_read_streams_start_index,
     DAGPipeline & pipeline)
@@ -366,7 +371,7 @@ void DAGQueryBlockInterpreter::executeCastAfterTableScan(
     analyzer->initChain(chain, original_source_columns);
 
     // execute timezone cast or duration cast if needed for local table scan
-    if (addExtraCastsAfterTs(*analyzer, is_need_add_cast_column, chain, query_block.source->tbl_scan()))
+    if (addExtraCastsAfterTs(*analyzer, is_need_add_cast_column, chain, table_scan))
     {
         ExpressionActionsPtr extra_cast = chain.getLastActions();
         chain.finalize();

dbms/src/Flash/Coprocessor/DAGQueryBlockInterpreter.h

@@ -58,7 +58,11 @@ class DAGQueryBlockInterpreter
 private:
     void executeImpl(DAGPipeline & pipeline);
     void handleTableScan(const TiDBTableScan & table_scan, DAGPipeline & pipeline);
-    void executeCastAfterTableScan(const std::vector<ExtraCastAfterTSMode> & is_need_add_cast_column, size_t remote_read_streams_start_index, DAGPipeline & pipeline);
+    void executeCastAfterTableScan(
+        const TiDBTableScan & table_scan,
+        const std::vector<ExtraCastAfterTSMode> & is_need_add_cast_column,
+        size_t remote_read_streams_start_index,
+        DAGPipeline & pipeline);
     void executePushedDownFilter(const std::vector<const tipb::Expr *> & conditions, size_t remote_read_streams_start_index, DAGPipeline & pipeline);
     void handleJoin(const tipb::Join & join, DAGPipeline & pipeline, SubqueryForSet & right_query);
     void prepareJoin(

tests/delta-merge-test/query/mpp/partition_table.test

@@ -24,7 +24,7 @@
 => DBGInvoke __set_flush_threshold(1000000, 1000000)
 
 # Data.
-=> DBGInvoke __mock_tidb_table(default, test, 'col_1 String, col_2 Int64')
+=> DBGInvoke __mock_tidb_table(default, test, 'col_1 String, col_2 Int64, col_3 MyDatetime')
 => DBGInvoke __mock_tidb_partition(default, test, 9997)
 => DBGInvoke __mock_tidb_partition(default, test, 9998)
 => DBGInvoke __mock_tidb_partition(default, test, 9999)
@@ -38,21 +38,22 @@
 
 # query on empty table
 => DBGInvoke tidb_query('select count(col_1) from default.test group by col_2', 4,'mpp_query:true,mpp_partition_num:3')
+=> DBGInvoke tidb_query('select count(col_1) from default.test group by col_3', 4,'mpp_query:true,mpp_partition_num:3')
 
-=> DBGInvoke __raft_insert_row(default, test, 1, 50, 'test1', 666)
-=> DBGInvoke __raft_insert_row(default, test, 1, 51, 'test2', 666)
-=> DBGInvoke __raft_insert_row(default, test, 1, 52, 'test3', 777)
-=> DBGInvoke __raft_insert_row(default, test, 1, 53, 'test4', 888)
+=> DBGInvoke __raft_insert_row(default, test, 1, 50, 'test1', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 1, 51, 'test2', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 1, 52, 'test3', 777, '2022-03-31 11:22:22')
+=> DBGInvoke __raft_insert_row(default, test, 1, 53, 'test4', 888, '2022-03-31 11:33:33')
 
-=> DBGInvoke __raft_insert_row(default, test, 4, 50, 'test1', 666)
-=> DBGInvoke __raft_insert_row(default, test, 4, 51, 'test2', 666)
-=> DBGInvoke __raft_insert_row(default, test, 4, 52, 'test3', 777)
-=> DBGInvoke __raft_insert_row(default, test, 4, 53, 'test4', 888)
+=> DBGInvoke __raft_insert_row(default, test, 4, 50, 'test1', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 4, 51, 'test2', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 4, 52, 'test3', 777, '2022-03-31 11:22:22')
+=> DBGInvoke __raft_insert_row(default, test, 4, 53, 'test4', 888, '2022-03-31 11:33:33')
 
-=> DBGInvoke __raft_insert_row(default, test, 7, 50, 'test1', 666)
-=> DBGInvoke __raft_insert_row(default, test, 7, 51, 'test2', 666)
-=> DBGInvoke __raft_insert_row(default, test, 7, 52, 'test3', 777)
-=> DBGInvoke __raft_insert_row(default, test, 7, 53, 'test4', 888)
+=> DBGInvoke __raft_insert_row(default, test, 7, 50, 'test1', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 7, 51, 'test2', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 7, 52, 'test3', 777, '2022-03-31 11:22:22')
+=> DBGInvoke __raft_insert_row(default, test, 7, 53, 'test4', 888, '2022-03-31 11:33:33')
 
 # query on table that some partition does not contains region
 => DBGInvoke tidb_query('select count(col_1), col_2 from default.test group by col_2', 4,'mpp_query:true,mpp_partition_num:3')
@@ -62,6 +63,13 @@
 │                   3 │                 888 │
 └─────────────────────┴─────────────────────┘
 
+=> DBGInvoke tidb_query('select count(col_1), col_3 from default.test group by col_3', 4,'mpp_query:true,mpp_partition_num:3')
+┌─exchange_receiver_0─┬─exchange_receiver_1─┐
+│                   6 │ 2022-03-31 11:11:11 │
+│                   3 │ 2022-03-31 11:22:22 │
+│                   3 │ 2022-03-31 11:33:33 │
+└─────────────────────┴─────────────────────┘
+
 # add more regions
 => DBGInvoke __put_region(2, 100, 200, default, test, 9997)
 => DBGInvoke __put_region(3, 200, 300, default, test, 9997)
@@ -72,32 +80,32 @@
 => DBGInvoke __put_region(8, 100, 200, default, test, 9999)
 => DBGInvoke __put_region(9, 200, 300, default, test, 9999)
 
-=> DBGInvoke __raft_insert_row(default, test, 2, 150, 'test1', 666)
-=> DBGInvoke __raft_insert_row(default, test, 2, 151, 'test2', 666)
-=> DBGInvoke __raft_insert_row(default, test, 2, 152, 'test3', 777)
-=> DBGInvoke __raft_insert_row(default, test, 2, 153, 'test4', 888)
-=> DBGInvoke __raft_insert_row(default, test, 3, 250, 'test1', 666)
-=> DBGInvoke __raft_insert_row(default, test, 3, 251, 'test2', 666)
-=> DBGInvoke __raft_insert_row(default, test, 3, 252, 'test3', 777)
-=> DBGInvoke __raft_insert_row(default, test, 3, 253, 'test4', 888)
-
-=> DBGInvoke __raft_insert_row(default, test, 5, 150, 'test1', 666)
-=> DBGInvoke __raft_insert_row(default, test, 5, 151, 'test2', 666)
-=> DBGInvoke __raft_insert_row(default, test, 5, 152, 'test3', 777)
-=> DBGInvoke __raft_insert_row(default, test, 5, 153, 'test4', 888)
-=> DBGInvoke __raft_insert_row(default, test, 6, 250, 'test1', 666)
-=> DBGInvoke __raft_insert_row(default, test, 6, 251, 'test2', 666)
-=> DBGInvoke __raft_insert_row(default, test, 6, 252, 'test3', 777)
-=> DBGInvoke __raft_insert_row(default, test, 6, 253, 'test4', 888)
-
-=> DBGInvoke __raft_insert_row(default, test, 8, 150, 'test1', 666)
-=> DBGInvoke __raft_insert_row(default, test, 8, 151, 'test2', 666)
-=> DBGInvoke __raft_insert_row(default, test, 8, 152, 'test3', 777)
-=> DBGInvoke __raft_insert_row(default, test, 8, 153, 'test4', 888)
-=> DBGInvoke __raft_insert_row(default, test, 9, 250, 'test1', 666)
-=> DBGInvoke __raft_insert_row(default, test, 9, 251, 'test2', 666)
-=> DBGInvoke __raft_insert_row(default, test, 9, 252, 'test3', 777)
-=> DBGInvoke __raft_insert_row(default, test, 9, 253, 'test4', 888)
+=> DBGInvoke __raft_insert_row(default, test, 2, 150, 'test1', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 2, 151, 'test2', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 2, 152, 'test3', 777, '2022-03-31 11:22:22')
+=> DBGInvoke __raft_insert_row(default, test, 2, 153, 'test4', 888, '2022-03-31 11:33:33')
+=> DBGInvoke __raft_insert_row(default, test, 3, 250, 'test1', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 3, 251, 'test2', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 3, 252, 'test3', 777, '2022-03-31 11:22:22')
+=> DBGInvoke __raft_insert_row(default, test, 3, 253, 'test4', 888, '2022-03-31 11:33:33')
+
+=> DBGInvoke __raft_insert_row(default, test, 5, 150, 'test1', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 5, 151, 'test2', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 5, 152, 'test3', 777, '2022-03-31 11:22:22')
+=> DBGInvoke __raft_insert_row(default, test, 5, 153, 'test4', 888, '2022-03-31 11:33:33')
+=> DBGInvoke __raft_insert_row(default, test, 6, 250, 'test1', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 6, 251, 'test2', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 6, 252, 'test3', 777, '2022-03-31 11:22:22')
+=> DBGInvoke __raft_insert_row(default, test, 6, 253, 'test4', 888, '2022-03-31 11:33:33')
+
+=> DBGInvoke __raft_insert_row(default, test, 8, 150, 'test1', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 8, 151, 'test2', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 8, 152, 'test3', 777, '2022-03-31 11:22:22')
+=> DBGInvoke __raft_insert_row(default, test, 8, 153, 'test4', 888, '2022-03-31 11:33:33')
+=> DBGInvoke __raft_insert_row(default, test, 9, 250, 'test1', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 9, 251, 'test2', 666, '2022-03-31 11:11:11')
+=> DBGInvoke __raft_insert_row(default, test, 9, 252, 'test3', 777, '2022-03-31 11:22:22')
+=> DBGInvoke __raft_insert_row(default, test, 9, 253, 'test4', 888, '2022-03-31 11:33:33')
 
 # query on table that every partition contains region
 => DBGInvoke tidb_query('select count(col_1), col_2 from default.test group by col_2', 4,'mpp_query:true,mpp_partition_num:3')
@@ -107,6 +115,13 @@
 │                   9 │                 888 │
 └─────────────────────┴─────────────────────┘
 
+=> DBGInvoke tidb_query('select count(col_1), col_3 from default.test group by col_3', 4,'mpp_query:true,mpp_partition_num:3')
+┌─exchange_receiver_0─┬─exchange_receiver_1─┐
+│                  18 │ 2022-03-31 11:11:11 │
+│                   9 │ 2022-03-31 11:22:22 │
+│                   9 │ 2022-03-31 11:33:33 │
+└─────────────────────┴─────────────────────┘
+
 # Clean up.
 => DBGInvoke __drop_tidb_table(default, test)
 => drop table if exists default.test

tests/fullstack-test/mpp/partition_table_with_time.test

@@ -0,0 +1,35 @@
+# Copyright 2022 PingCAP, Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Preparation.
+=> DBGInvoke __init_fail_point()
+
+mysql> drop table if exists test.t;
+mysql> create table test.t (col1 smallint(28) unsigned DEFAULT 0, col2 time  DEFAULT NULL) PARTITION BY RANGE (col1) ( PARTITION p0 VALUES LESS THAN (2), PARTITION p3 VALUES LESS THAN MAXVALUE);
+mysql> insert into test.t values (1, 111111), (2, 222222), (3, 333333);
+mysql> alter table test.t set tiflash replica 1;
+
+func> wait_table test t
+
+mysql> set tidb_allow_mpp=1; set tidb_enforce_mpp=1; set tidb_isolation_read_engines='tiflash'; set tidb_partition_prune_mode= dynamic; select col2 from test.t order by col1;
++----------+
+| col2     |
++----------+
+| 11:11:11 |
+| 22:22:22 |
+| 33:33:33 |
++----------+
+
+# Clean up.
+mysql> drop table if exists test.t