code refine && several minor bug fix

dbms/src/Flash/Coprocessor/DAGBlockOutputStream.cpp

@@ -42,6 +42,7 @@ void DAGBlockOutputStream::writeSuffix()
     if (current_chunk != nullptr && records_per_chunk > 0)
     {
         current_chunk->set_rows_data(current_ss.str());
+        dag_response.add_output_counts(records_per_chunk);
     }
 }
 
@@ -62,6 +63,7 @@ void DAGBlockOutputStream::write(const Block & block)
             {
                 // set the current ss to current chunk
                 current_chunk->set_rows_data(current_ss.str());
+                dag_response.add_output_counts(current_records_num);
             }
             current_chunk = dag_response.add_chunks();
             current_ss.str("");

dbms/src/Flash/Coprocessor/DAGDriver.cpp

@@ -78,7 +78,7 @@ try
         {
             if (auto * p_stream = dynamic_cast<IProfilingBlockInputStream *>(streamPtr.get()))
             {
-                time_processed_ns += p_stream->getProfileInfo().total_stopwatch.elapsed();
+                time_processed_ns = std::max(time_processed_ns, p_stream->getProfileInfo().total_stopwatch.elapsed());
                 num_produced_rows += p_stream->getProfileInfo().rows;
                 num_iterations += p_stream->getProfileInfo().blocks;
             }

dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.cpp

@@ -143,6 +143,15 @@ void DAGExpressionAnalyzer::appendOrderBy(ExpressionActionsChain & chain, const
 
 const NamesAndTypesList & DAGExpressionAnalyzer::getCurrentInputColumns() { return after_agg ? aggregated_columns : source_columns; }
 
+void DAGExpressionAnalyzer::appendFinalProject(ExpressionActionsChain & chain, const NamesWithAliases & final_project)
+{
+    initChain(chain, getCurrentInputColumns());
+    for (auto name : final_project)
+    {
+        chain.steps.back().required_output.push_back(name.first);
+    }
+}
+
 void DAGExpressionAnalyzer::appendAggSelect(ExpressionActionsChain & chain, const tipb::Aggregation & aggregation)
 {
     initChain(chain, getCurrentInputColumns());
@@ -199,11 +208,11 @@ void DAGExpressionAnalyzer::appendAggSelect(ExpressionActionsChain & chain, cons
 
 String DAGExpressionAnalyzer::appendCastIfNeeded(const tipb::Expr & expr, ExpressionActionsPtr & actions, const String & expr_name)
 {
-    if (!expr.has_field_type())
+    if (!expr.has_field_type() && context.getSettingsRef().dag_expr_field_type_strict_check)
     {
         throw Exception("Expression without field type", ErrorCodes::COP_BAD_DAG_REQUEST);
     }
-    if (isFunctionExpr(expr))
+    if (expr.has_field_type() && isFunctionExpr(expr))
     {
         DataTypePtr expected_type = getDataTypeByFieldType(expr.field_type());
         DataTypePtr actual_type = actions->getSampleBlock().getByName(expr_name).type;

dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.h

@@ -44,6 +44,7 @@ class DAGExpressionAnalyzer : private boost::noncopyable
             chain.steps.emplace_back(std::make_shared<ExpressionActions>(columns, settings));
         }
     }
+    void appendFinalProject(ExpressionActionsChain & chain, const NamesWithAliases & final_project);
     String getActions(const tipb::Expr & expr, ExpressionActionsPtr & actions);
     const NamesAndTypesList & getCurrentInputColumns();
 };

dbms/src/Flash/Coprocessor/DAGQuerySource.cpp

@@ -90,7 +90,7 @@ bool fillExecutorOutputFieldTypes(const tipb::Executor & executor, std::vector<t
     switch (executor.tp())
     {
         case tipb::ExecType::TypeTableScan:
-            for (auto ci : executor.tbl_scan().columns())
+            for (auto & ci : executor.tbl_scan().columns())
             {
                 field_type.set_tp(ci.tp());
                 field_type.set_flag(ci.flag());
@@ -136,7 +136,7 @@ std::vector<tipb::FieldType> DAGQuerySource::getResultFieldTypes() const
         throw Exception("Do not found result field type for current dag request", ErrorCodes::COP_BAD_DAG_REQUEST);
     }
     // tispark assumes that if there is a agg, the output offset is
-    // ignored and the request out put is the same as the agg's output.
+    // ignored and the request output is the same as the agg's output.
     // todo should always use output offset to re-construct the output field types
     if (hasAggregation())
     {

dbms/src/Flash/Coprocessor/InterpreterDAG.cpp

@@ -208,15 +208,8 @@ InterpreterDAG::AnalysisResult InterpreterDAG::analyzeExpressions()
         analyzer.appendOrderBy(chain, dag.getTopN(), res.order_column_names);
     }
     // Append final project results if needed.
-    // TODO: Refine this logic by an `analyzer.appendFinalProject()`-like call.
-    if (dag.hasSelection() || dag.hasAggregation() || dag.hasTopN())
-    {
-        for (auto & name : final_project)
-        {
-            chain.steps.back().required_output.push_back(name.first);
-        }
-        res.before_order_and_select = chain.getLastActions();
-    }
+    analyzer.appendFinalProject(chain, final_project);
+    res.before_order_and_select = chain.getLastActions();
     chain.finalize();
     chain.clear();
     //todo need call prependProjectInput??

dbms/src/Flash/Coprocessor/tests/cop_test.cpp

@@ -98,16 +98,11 @@ class FlashClient
 };
 
 using ClientPtr = std::shared_ptr<FlashClient>;
-grpc::Status rpcTest()
+
+void appendTS(tipb::DAGRequest & dag_request, size_t & result_field_num)
 {
-    ChannelPtr cp = grpc::CreateChannel("localhost:9093", grpc::InsecureChannelCredentials());
-    ClientPtr clientPtr = std::make_shared<FlashClient>(cp);
-    size_t result_field_num = 0;
-    // construct a dag request
-    tipb::DAGRequest dagRequest;
-    dagRequest.set_start_ts(18446744073709551615uL);
     // table scan: s,i
-    tipb::Executor * executor = dagRequest.add_executors();
+    tipb::Executor * executor = dag_request.add_executors();
     executor->set_tp(tipb::ExecType::TypeTableScan);
     tipb::TableScan * ts = executor->mutable_tbl_scan();
     ts->set_table_id(44);
@@ -119,13 +114,16 @@ grpc::Status rpcTest()
     ci->set_column_id(2);
     ci->set_tp(8);
     ci->set_flag(0);
-    dagRequest.add_output_offsets(1);
-    dagRequest.add_output_offsets(0);
-    dagRequest.add_output_offsets(1);
+    dag_request.add_output_offsets(1);
+    dag_request.add_output_offsets(0);
+    dag_request.add_output_offsets(1);
     result_field_num = 3;
+}
 
+void appendSelection(tipb::DAGRequest & dag_request)
+{
     // selection: less(i, 123)
-    executor = dagRequest.add_executors();
+    auto * executor = dag_request.add_executors();
     executor->set_tp(tipb::ExecType::TypeSelection);
     tipb::Selection * selection = executor->mutable_selection();
     tipb::Expr * expr = selection->add_conditions();
@@ -150,16 +148,19 @@ grpc::Status rpcTest()
     type = expr->mutable_field_type();
     type->set_tp(1);
     type->set_flag(1 << 5);
+}
 
+void appendAgg(tipb::DAGRequest & dag_request, size_t & result_field_num)
+{
     // agg: count(s) group by i;
-    executor = dagRequest.add_executors();
+    auto * executor = dag_request.add_executors();
     executor->set_tp(tipb::ExecType::TypeAggregation);
     auto agg = executor->mutable_aggregation();
     auto agg_func = agg->add_agg_func();
     agg_func->set_tp(tipb::ExprType::Count);
     auto child = agg_func->add_children();
     child->set_tp(tipb::ExprType::ColumnRef);
-    ss.str("");
+    std::stringstream ss;
     DB::EncodeNumber<Int64, TiDB::CodecFlagInt>(0, ss);
     child->set_val(ss.str());
     auto f_type = agg_func->mutable_field_type();
@@ -174,32 +175,56 @@ grpc::Status rpcTest()
     f_type->set_tp(8);
     f_type->set_flag(1);
     result_field_num = 2;
+}
 
-    // topn
-    /*
-    executor = dagRequest.add_executors();
+void appendTopN(tipb::DAGRequest & dag_request)
+{
+    auto * executor = dag_request.add_executors();
     executor->set_tp(tipb::ExecType::TypeTopN);
     tipb::TopN * topN = executor->mutable_topn();
     topN->set_limit(3);
     tipb::ByItem * byItem = topN->add_order_by();
     byItem->set_desc(false);
     tipb::Expr * expr1 = byItem->mutable_expr();
     expr1->set_tp(tipb::ExprType::ColumnRef);
-    ss.str("");
+    std::stringstream ss;
     DB::EncodeNumber<Int64, TiDB::CodecFlagInt>(1, ss);
     expr1->set_val(ss.str());
-    type = expr1->mutable_field_type();
+    auto * type = expr1->mutable_field_type();
     type->set_tp(8);
     type->set_tp(0);
-     */
-    // limit
-    /*
-    executor = dagRequest.add_executors();
+}
+
+void appendLimit(tipb::DAGRequest & dag_request)
+{
+    auto * executor = dag_request.add_executors();
     executor->set_tp(tipb::ExecType::TypeLimit);
-    tipb::Limit *limit = executor->mutable_limit();
+    tipb::Limit * limit = executor->mutable_limit();
     limit->set_limit(5);
-     */
+}
+
+grpc::Status rpcTest()
+{
+    ChannelPtr cp = grpc::CreateChannel("localhost:9093", grpc::InsecureChannelCredentials());
+    ClientPtr clientPtr = std::make_shared<FlashClient>(cp);
+    size_t result_field_num = 0;
+    bool has_selection = false;
+    bool has_agg = true;
+    bool has_topN = false;
+    bool has_limit = false;
+    // construct a dag request
+    tipb::DAGRequest dagRequest;
+    dagRequest.set_start_ts(18446744073709551615uL);
 
+    appendTS(dagRequest, result_field_num);
+    if (has_selection)
+        appendSelection(dagRequest);
+    if (has_agg)
+        appendAgg(dagRequest, result_field_num);
+    if (has_topN)
+        appendTopN(dagRequest);
+    if (has_limit)
+        appendLimit(dagRequest);
 
     // construct a coprocessor request
     coprocessor::Request request;

dbms/src/Flash/FlashService.cpp

@@ -96,8 +96,10 @@ std::tuple<Context, ::grpc::Status> FlashService::createDBContext(grpc::ServerCo
     {
         context.setSetting("dag_records_per_chunk", dag_records_per_chunk_str);
     }
-    std::string planner = getClientMetaVarWithDefault(grpc_context, "dag_planner", "sql");
+    std::string planner = getClientMetaVarWithDefault(grpc_context, "dag_planner", "optree");
     context.setSetting("dag_planner", planner);
+    std::string expr_field_type_check = getClientMetaVarWithDefault(grpc_context, "dag_expr_field_type_strict_check", "1");
+    context.setSetting("dag_expr_field_type_strict_check", expr_field_type_check);
 
     return std::make_tuple(context, ::grpc::Status::OK);
 }

dbms/src/Interpreters/Settings.h

@@ -31,6 +31,7 @@ struct Settings
     M(SettingUInt64, read_tso, DEFAULT_MAX_READ_TSO, "tmt read tso.") \
     M(SettingInt64, dag_records_per_chunk, DEFAULT_DAG_RECORDS_PER_CHUNK, "default chunk size of a DAG response.") \
     M(SettingString, dag_planner, "sql", "planner for DAG query, sql builds the SQL string, optree builds the internal operator(stream) tree.") \
+    M(SettingBool, dag_expr_field_type_strict_check, true, "when set to true, every expr in the dag request must provide field type, otherwise only the result expr will be checked.") \
     M(SettingInt64, schema_version, DEFAULT_UNSPECIFIED_SCHEMA_VERSION, "tmt schema version.") \
     M(SettingUInt64, min_compress_block_size, DEFAULT_MIN_COMPRESS_BLOCK_SIZE, "The actual size of the block to compress, if the uncompressed data less than max_compress_block_size is no less than this value and no less than the volume of data for one mark.") \
     M(SettingUInt64, max_compress_block_size, DEFAULT_MAX_COMPRESS_BLOCK_SIZE, "The maximum size of blocks of uncompressed data before compressing for writing to a table.") \

dbms/src/Storages/Transaction/Codec.h

@@ -324,6 +324,11 @@ inline void EncodeVarUInt(UInt64 num, std::stringstream & ss)
     TiKV::writeVarUInt(num, ss);
 }
 
+inline void EncodeNull(std::stringstream &ss)
+{
+    writeIntBinary(UInt8(TiDB::CodecFlagNil), ss);
+}
+
 inline void EncodeDecimal(const Decimal & dec, std::stringstream & ss)
 {
     writeIntBinary(UInt8(TiDB::CodecFlagDecimal), ss);
@@ -394,6 +399,9 @@ T getFieldValue(const Field & field)
 
 inline void EncodeDatum(const Field & field, TiDB::CodecFlag flag, std::stringstream & ss)
 {
+    if (field.isNull()) {
+        return EncodeNull(ss);
+    }
     switch (flag)
     {
         case TiDB::CodecFlagDecimal: