Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Flash-481 Arrow encode #279

Merged
merged 150 commits into from
Oct 25, 2019
Merged
Changes from 1 commit
Commits
Show all changes
150 commits
Select commit Hold shift + click to select a range
aa92f4e
basic framework for coprocessor support in tiflash
windtalker Jul 30, 2019
4f37218
basic support for InterpreterDagRequestV2
windtalker Jul 30, 2019
85bfd5c
code refine
windtalker Jul 30, 2019
e1700c3
tipb submodule use tipb master branch
windtalker Jul 31, 2019
0f82665
rewrite build flow in InterpreterDagRequest
windtalker Jul 31, 2019
a7655bc
rename Dag to DAG
windtalker Jul 31, 2019
f516f00
Update tipb submodule
zanmato1984 Aug 1, 2019
3b520c9
basic support for selection/limit/topn executor in InterpreterDAGRequest
windtalker Aug 2, 2019
9591d26
Merge branch 'cop' of https://github.com/pingcap/tics into cop
windtalker Aug 2, 2019
ead9609
basic support for selection/limit/topn executor in InterpreterDAGRequ…
windtalker Aug 2, 2019
bed0bd4
merge pingcap/cop branch
windtalker Aug 2, 2019
526cad9
Code reorg
zanmato1984 Aug 4, 2019
be4d80c
Format
zanmato1984 Aug 4, 2019
64a45a9
merge pingcap/cop
windtalker Aug 5, 2019
a76fdb3
merge pingcap/cop
windtalker Aug 5, 2019
0cfe045
Refine code
zanmato1984 Aug 5, 2019
e9b216c
Merge branch 'cop' of https://github.com/pingcap/tics into cop
windtalker Aug 5, 2019
3617a87
basic support for dag agg executor
windtalker Aug 5, 2019
cb55df4
Code refine
zanmato1984 Aug 5, 2019
ed41c93
Merge master into cop
zanmato1984 Aug 5, 2019
08b7142
Refine code
zanmato1984 Aug 5, 2019
bc25942
Another way of getting codec flag
zanmato1984 Aug 5, 2019
059f267
fix cop test regression (#157)
windtalker Aug 6, 2019
e59e8f3
fix npe during dag execute (#160)
windtalker Aug 6, 2019
a618cb5
Add tipb cpp gen in build script
zanmato1984 Aug 6, 2019
4f797fe
Merge branch 'master' into cop
zanmato1984 Aug 6, 2019
bb51749
Fix build error and adjust some formats
zanmato1984 Aug 6, 2019
da1cb0e
Fix build error
zanmato1984 Aug 6, 2019
816ef4b
Fix build error
zanmato1984 Aug 6, 2019
f18fcdd
Update flash configs
zanmato1984 Aug 6, 2019
2ade1cb
Format
zanmato1984 Aug 6, 2019
3870d93
Merge branch 'master' into cop
zanmato1984 Aug 7, 2019
7cb9e71
throw exception when meet error duing cop request handling (#162)
windtalker Aug 7, 2019
5fe66ee
Merge branch 'master' into cop
zanmato1984 Aug 8, 2019
0174b7e
add DAGContext so InterpreterDAG can exchange information with DAGDri…
windtalker Aug 8, 2019
9a1dd23
columnref index is based on executor output schema (#167)
windtalker Aug 8, 2019
26e20d5
Move flash/cop/dag to individual library
zanmato1984 Aug 8, 2019
bf67d9d
Merge cop lib
zanmato1984 Aug 8, 2019
62ced38
DAG planner fix and mock dag request (#169)
zanmato1984 Aug 9, 2019
b346a24
Merge branch 'master' into cop
zanmato1984 Aug 9, 2019
57cd382
Fix DAG get and lock storage
zanmato1984 Aug 9, 2019
4a76e91
handle error in cop request (#171)
windtalker Aug 12, 2019
2d093a8
code refine && several minor bug fix (#174)
windtalker Aug 12, 2019
c8cd3d7
Fix region id in mock dag
zanmato1984 Aug 12, 2019
0492af6
support udf in (#175)
windtalker Aug 14, 2019
4a6bad8
Merge branch 'master' into cop
zanmato1984 Aug 14, 2019
8713ff2
1. fix decode literal expr error, 2. add all scalar function sig in s…
windtalker Aug 14, 2019
7759af1
Merge branch 'master' into cop
zanmato1984 Aug 15, 2019
b25d1cc
some bug fix (#179)
windtalker Aug 15, 2019
3d38b7b
Support all DAG operator types in mock SQL -> DAG parser (#176)
zanmato1984 Aug 15, 2019
cbcfdb0
filter column must be uint8 in tiflash (#180)
windtalker Aug 16, 2019
d87e2d5
1. fix encode null error, 2. fix empty field type generated by TiFlas…
windtalker Aug 16, 2019
17f7fcb
Merge branch 'master' into cop
zanmato1984 Aug 16, 2019
5853b91
check validation of dag exprs field type (#183)
windtalker Aug 19, 2019
0a6767a
Merge branch 'master' into cop
zanmato1984 Aug 19, 2019
d53ca34
Merge branch 'master' into cop
zanmato1984 Aug 20, 2019
5de0ec6
add more coprocessor mock tests (#185)
windtalker Aug 20, 2019
6196171
add some log about implicit cast (#188)
windtalker Aug 21, 2019
960cc56
Merge branch 'master' into cop
zanmato1984 Aug 24, 2019
08bacd7
Pass DAG tests after merging master (#199)
zanmato1984 Aug 24, 2019
e8b4198
Fix date/datetime/bit encode error (#200)
zanmato1984 Aug 26, 2019
61cdc8f
improve dag execution time collection (#202)
windtalker Aug 26, 2019
53dcd1f
Merge branch 'master' into cop
zanmato1984 Aug 27, 2019
10e3883
column id in table scan operator may be -1 (#205)
windtalker Aug 27, 2019
39d1994
quick fix for decimal encode (#210)
windtalker Aug 30, 2019
8a0fb66
support udf like with 3 arguments (#212)
windtalker Sep 2, 2019
ff9a1de
Flash-473 optimize date and datetime comparison (#221)
windtalker Sep 5, 2019
17aacde
Merge master
zanmato1984 Sep 5, 2019
6b14b38
FLASH-479 select from empty table throw error in tiflash (#223)
windtalker Sep 6, 2019
548e519
Update flash service port
zanmato1984 Sep 6, 2019
a1b8444
fix bug in DAGBlockOutputStream
windtalker Sep 10, 2019
fce3676
fix bug in DAGBlockOutputStream (#230)
windtalker Sep 10, 2019
a9f9b48
FLASH-475: Support BATCH COMMANDS in flash service (#232)
zanmato1984 Sep 12, 2019
bdc7d57
init change for array encode
windtalker Sep 12, 2019
516d340
merge pingcap/tics/cop
windtalker Sep 12, 2019
1ccfbd4
Merge branch 'master' into cop
zhexuany Sep 12, 2019
df07939
FLASH-483: Combine raft service and flash service (#235)
zanmato1984 Sep 16, 2019
99f26c0
Merge master
zanmato1984 Sep 16, 2019
0bb7991
Fix build error
zanmato1984 Sep 16, 2019
f41f853
Fix test regression
zanmato1984 Sep 16, 2019
259ec77
Fix null value bug in datum
zanmato1984 Sep 17, 2019
ef65514
Merge branch 'master' into cop
zanmato1984 Sep 17, 2019
708d52f
FLASH-490: Fix table scan with -1 column ID and no agg (#240)
zanmato1984 Sep 23, 2019
3656a95
Merge branch 'master' into cop
zanmato1984 Sep 23, 2019
a4c1074
throw error if the cop request is not based on full region scan (#247)
windtalker Sep 24, 2019
b57656c
Merge branch 'master' into cop
zanmato1984 Sep 25, 2019
3a43942
FLASH-437 Support time zone in coprocessor (#259)
windtalker Sep 27, 2019
01caa55
Merge branch 'master' into cop
zanmato1984 Sep 27, 2019
8d2576e
Address comment
zanmato1984 Sep 29, 2019
8ec5380
Merge branch 'cop' of https://github.com/pingcap/tics into array_encode
windtalker Sep 29, 2019
2e3b1c1
use the new date implementation
windtalker Sep 29, 2019
d33a278
FLASH-489 support key condition for coprocessor query (#261)
windtalker Sep 30, 2019
087faee
Merge branch 'master' into cop
zanmato1984 Sep 30, 2019
4aa2b58
only return execute summaies if requested (#264)
windtalker Sep 30, 2019
aed5e84
Merge branch 'cop' of https://github.com/pingcap/tics into array_encode
windtalker Oct 8, 2019
8663811
refine code
windtalker Oct 8, 2019
80f6f35
Refine service init (#265)
zanmato1984 Oct 8, 2019
0b737dc
fix bug
windtalker Oct 9, 2019
d3af009
fix bug
windtalker Oct 9, 2019
004f7c5
Merge branch 'cop' of https://github.com/pingcap/tics into arrow_encode
windtalker Oct 9, 2019
f255362
FLASH-554 cop check range should be based on region range (#270)
windtalker Oct 10, 2019
170f652
add ut for arrow encode
windtalker Oct 11, 2019
c53e456
Merge branch 'cop' of https://github.com/pingcap/tics into arrow_encode
windtalker Oct 11, 2019
7fc53ad
minor improve (#273)
windtalker Oct 11, 2019
22ad2d3
Merge branch 'master' into cop
zanmato1984 Oct 11, 2019
b01ccb3
update tipb
windtalker Oct 11, 2019
a1304ae
Fix mutex on timezone retrieval (#276)
ilovesoup2000 Oct 11, 2019
687dcbe
Fix race condition of batch command handling (#277)
zanmato1984 Oct 12, 2019
4dd5e1e
Merge branch 'cop' of https://github.com/pingcap/tics into arrow_encode
windtalker Oct 12, 2019
80c20b2
update tipb version
windtalker Oct 12, 2019
7c5bea6
set default record_per_chunk to 1024
windtalker Oct 13, 2019
939b8cf
address comment
windtalker Oct 14, 2019
d25dadc
address comments
windtalker Oct 14, 2019
512fa8e
refine code
windtalker Oct 14, 2019
ff9bf8f
Merge branch 'cop' of https://github.com/pingcap/tics into arrow_encode
windtalker Oct 14, 2019
a6f6dda
refine code
windtalker Oct 14, 2019
a943e8d
add mock_dag test
windtalker Oct 14, 2019
41272da
code refine
windtalker Oct 14, 2019
00dac75
code refine
windtalker Oct 14, 2019
4080fba
address comments
windtalker Oct 14, 2019
1188e69
Merge branch 'cop' of https://github.com/pingcap/tics into arrow_encode
windtalker Oct 14, 2019
d2890e3
Fix NULL order for dag (#281)
zanmato1984 Oct 14, 2019
bc075c5
refine get actions in DAGExpressionAnalyzer, fix bug in dbgFuncCoproc…
windtalker Oct 15, 2019
4dbff78
Merge branch 'cop' of https://github.com/pingcap/tics into arrow_encode
windtalker Oct 15, 2019
fbcbdc0
remove duplicate agg funcs (#283)
windtalker Oct 15, 2019
8f2bfaf
Merge branch 'cop' of https://github.com/pingcap/tics into arrow_encode
windtalker Oct 16, 2019
3716b98
refine code
windtalker Oct 16, 2019
fa42c69
remove useless code
windtalker Oct 16, 2019
7bbe8c0
address comments
windtalker Oct 16, 2019
31973bf
remove uselss include
windtalker Oct 16, 2019
d968c09
address comments
windtalker Oct 16, 2019
edf32d4
Merge branch 'cop' of https://github.com/pingcap/tics into arrow_encode
windtalker Oct 16, 2019
f1256bd
refine code
windtalker Oct 17, 2019
73befbd
address comments
windtalker Oct 17, 2019
3188c07
format code
windtalker Oct 17, 2019
87955d1
fix typo
windtalker Oct 17, 2019
4f58878
Update dbms/src/Flash/BatchCommandsHandler.cpp
zanmato1984 Oct 17, 2019
92c16c2
revert unnecessary changes
windtalker Oct 17, 2019
0f6f0a6
Merge branch 'cop' of https://github.com/pingcap/tics into arrow_encode
windtalker Oct 17, 2019
d550644
refine code
windtalker Oct 17, 2019
bac7951
fix build error
windtalker Oct 17, 2019
4a251b0
refine code
windtalker Oct 17, 2019
e8b92b4
Merge branch 'master' into cop
zanmato1984 Oct 17, 2019
48dd7bd
Merge master
zanmato1984 Oct 18, 2019
a8cba5f
Merge remote-tracking branch 'origin/cop' into arrow_encode_2
windtalker Oct 18, 2019
e3232af
Merge branch 'master' of https://github.com/pingcap/tics into arrow_e…
windtalker Oct 21, 2019
4d5e5d4
address comments
windtalker Oct 21, 2019
c7d8d4e
refine code
windtalker Oct 22, 2019
0b1ed77
address comments
windtalker Oct 25, 2019
683e7e0
Merge branch 'master' into arrow_encode
zanmato1984 Oct 25, 2019
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
support udf in (#175)
* fix cop test regression

* address comments

* format code

* fix npe for dag execute

* format code

* address comment

* add some comments

* throw exception when meet error duing cop request handling

* address comments

* add error code

* throw exception when meet error duing cop request handling

* address comments

* add DAGContext so InterpreterDAG can exchange information with DAGDriver

* fix bug

* 1. refine code, 2. address comments

* update comments

* columnref index is based on executor output schema

* handle error in coprocessor request

* refine code

* use Clear to clear a protobuf message completely

* refine code

* code refine && several minor bug fix

* address comments

* address comments

* support udf in

* refine code

* address comments

* address comments
windtalker authored and zanmato1984 committed Aug 14, 2019
commit 0492af6a5ac0076ced5ef53f17efad04f9f54c25
60 changes: 52 additions & 8 deletions dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.cpp
Original file line number Diff line number Diff line change
@@ -1,10 +1,13 @@
#include <Flash/Coprocessor/DAGExpressionAnalyzer.h>

#include <AggregateFunctions/AggregateFunctionFactory.h>
#include <Columns/ColumnSet.h>
#include <DataTypes/DataTypeSet.h>
#include <DataTypes/FieldToDataType.h>
#include <Flash/Coprocessor/DAGUtils.h>
#include <Functions/FunctionFactory.h>
#include <Interpreters/Context.h>
#include <Interpreters/Set.h>
#include <Interpreters/convertFieldToType.h>
#include <Storages/Transaction/Codec.h>
#include <Storages/Transaction/TypeMapping.h>
@@ -251,6 +254,32 @@ String DAGExpressionAnalyzer::appendCastIfNeeded(const tipb::Expr & expr, Expres
return expr_name;
}

void DAGExpressionAnalyzer::makeExplicitSet(
const tipb::Expr & expr, const Block & sample_block, bool create_ordered_set, const String & left_arg_name)
{
if (prepared_sets.count(&expr))
{
return;
}
DataTypes set_element_types;
// todo support tuple in, i.e. (a,b) in ((1,2), (3,4)), currently TiDB convert tuple in into a series of or/and/eq exprs
// which means tuple in is never be pushed to coprocessor, but it is quite in-efficient
set_element_types.push_back(sample_block.getByName(left_arg_name).type);

// todo if this is a single value in, then convert it to equal expr
SetPtr set = std::make_shared<Set>(SizeLimits(settings.max_rows_in_set, settings.max_bytes_in_set, settings.set_overflow_mode));
set->createFromDAGExpr(set_element_types, expr, create_ordered_set);
prepared_sets[&expr] = std::move(set);
}

static String getUniqueName(const Block & block, const String & prefix)
{
int i = 1;
while (block.has(prefix + toString(i)))
++i;
return prefix + toString(i);
}

String DAGExpressionAnalyzer::getActions(const tipb::Expr & expr, ExpressionActionsPtr & actions)
{
String expr_name = getName(expr, getCurrentInputColumns());
@@ -288,20 +317,35 @@ String DAGExpressionAnalyzer::getActions(const tipb::Expr & expr, ExpressionActi
throw Exception("agg function is not supported yet", ErrorCodes::UNSUPPORTED_METHOD);
}
const String & func_name = getFunctionName(expr);
if (func_name == "in" || func_name == "notIn" || func_name == "globalIn" || func_name == "globalNotIn")
{
// todo support in
throw Exception(func_name + " is not supported yet", ErrorCodes::UNSUPPORTED_METHOD);
}

const FunctionBuilderPtr & function_builder = FunctionFactory::instance().get(func_name, context);
Names argument_names;
DataTypes argument_types;
for (auto & child : expr.children())

if (isInOrGlobalInOperator(func_name))
{
String name = getActions(child, actions);
String name = getActions(expr.children(0), actions);
argument_names.push_back(name);
argument_types.push_back(actions->getSampleBlock().getByName(name).type);
makeExplicitSet(expr, actions->getSampleBlock(), false, name);
ColumnWithTypeAndName column;
column.type = std::make_shared<DataTypeSet>();

const SetPtr & set = prepared_sets[&expr];

column.name = getUniqueName(actions->getSampleBlock(), "___set");
column.column = ColumnSet::create(1, set);
actions->add(ExpressionAction::addColumn(column));
argument_names.push_back(column.name);
argument_types.push_back(column.type);
}
else
{
for (auto & child : expr.children())
{
String name = getActions(child, actions);
argument_names.push_back(name);
argument_types.push_back(actions->getSampleBlock().getByName(name).type);
}
}

// re-construct expr_name, because expr_name generated previously is based on expr tree,
6 changes: 6 additions & 0 deletions dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.h
Original file line number Diff line number Diff line change
@@ -13,6 +13,10 @@
namespace DB
{

class Set;
using SetPtr = std::shared_ptr<Set>;
using DAGPreparedSets = std::unordered_map<const tipb::Expr *, SetPtr>;

/** Transforms an expression from DAG expression into a sequence of actions to execute it.
*
*/
@@ -24,6 +28,7 @@ class DAGExpressionAnalyzer : private boost::noncopyable
NamesAndTypesList source_columns;
// all columns after aggregation
NamesAndTypesList aggregated_columns;
DAGPreparedSets prepared_sets;
Settings settings;
const Context & context;
bool after_agg;
@@ -47,6 +52,7 @@ class DAGExpressionAnalyzer : private boost::noncopyable
void appendFinalProject(ExpressionActionsChain & chain, const NamesWithAliases & final_project);
String getActions(const tipb::Expr & expr, ExpressionActionsPtr & actions);
const NamesAndTypesList & getCurrentInputColumns();
void makeExplicitSet(const tipb::Expr & expr, const Block & sample_block, bool create_ordered_set, const String & left_arg_name);
};

} // namespace DB
2 changes: 1 addition & 1 deletion dbms/src/Flash/Coprocessor/DAGStringConverter.cpp
Original file line number Diff line number Diff line change
@@ -58,7 +58,7 @@ void DAGStringConverter::buildTSString(const tipb::TableScan & ts, std::stringst
String name = merge_tree->getTableInfo().columns[cid - 1].name;
output_from_ts.push_back(std::move(name));
}
ss << "FROM " << merge_tree->getTableInfo().db_name << "." << merge_tree->getTableInfo().name << " ";
ss << "FROM " << storage->getDatabaseName() << "." << storage->getTableName() << " ";
}

void DAGStringConverter::buildSelString(const tipb::Selection & sel, std::stringstream & ss)
61 changes: 33 additions & 28 deletions dbms/src/Flash/Coprocessor/DAGUtils.cpp
Original file line number Diff line number Diff line change
@@ -9,6 +9,12 @@
namespace DB
{

namespace ErrorCodes
{
extern const int COP_BAD_DAG_REQUEST;
extern const int UNSUPPORTED_METHOD;
} // namespace ErrorCodes

bool isFunctionExpr(const tipb::Expr & expr)
{
switch (expr.tp())
@@ -43,7 +49,7 @@ const String & getAggFunctionName(const tipb::Expr & expr)
{
if (!aggFunMap.count(expr.tp()))
{
throw Exception(tipb::ExprType_Name(expr.tp()) + " is not supported.");
throw Exception(tipb::ExprType_Name(expr.tp()) + " is not supported.", ErrorCodes::UNSUPPORTED_METHOD);
}
return aggFunMap[expr.tp()];
}
@@ -54,21 +60,21 @@ const String & getFunctionName(const tipb::Expr & expr)
{
if (!aggFunMap.count(expr.tp()))
{
throw Exception(tipb::ExprType_Name(expr.tp()) + " is not supported.");
throw Exception(tipb::ExprType_Name(expr.tp()) + " is not supported.", ErrorCodes::UNSUPPORTED_METHOD);
}
return aggFunMap[expr.tp()];
}
else
{
if (!scalarFunMap.count(expr.sig()))
{
throw Exception(tipb::ScalarFuncSig_Name(expr.sig()) + " is not supported.");
throw Exception(tipb::ScalarFuncSig_Name(expr.sig()) + " is not supported.", ErrorCodes::UNSUPPORTED_METHOD);
}
return scalarFunMap[expr.sig()];
}
}

String exprToString(const tipb::Expr & expr, const NamesAndTypesList & input_col)
String exprToString(const tipb::Expr & expr, const NamesAndTypesList & input_col, bool for_parser)
{
std::stringstream ss;
size_t cursor = 1;
@@ -94,7 +100,7 @@ String exprToString(const tipb::Expr & expr, const NamesAndTypesList & input_col
columnId = DecodeInt<Int64>(cursor, expr.val());
if (columnId < 0 || columnId >= (ColumnID)input_col.size())
{
throw Exception("out of bound");
throw Exception("Column id out of bound", ErrorCodes::COP_BAD_DAG_REQUEST);
}
return input_col.getNames()[columnId];
case tipb::ExprType::Count:
@@ -105,53 +111,50 @@ String exprToString(const tipb::Expr & expr, const NamesAndTypesList & input_col
case tipb::ExprType::First:
if (!aggFunMap.count(expr.tp()))
{
throw Exception("not supported");
throw Exception(tipb::ExprType_Name(expr.tp()) + "not supported", ErrorCodes::UNSUPPORTED_METHOD);
}
func_name = aggFunMap.find(expr.tp())->second;
break;
case tipb::ExprType::ScalarFunc:
if (!scalarFunMap.count(expr.sig()))
{
throw Exception("not supported");
throw Exception(tipb::ScalarFuncSig_Name(expr.sig()) + "not supported", ErrorCodes::UNSUPPORTED_METHOD);
}
func_name = scalarFunMap.find(expr.sig())->second;
break;
default:
throw Exception("not supported");
throw Exception(tipb::ExprType_Name(expr.tp()) + "not supported", ErrorCodes::UNSUPPORTED_METHOD);
}
// build function expr
if (func_name == "in")
if (isInOrGlobalInOperator(func_name) && for_parser)
{
// for in, we could not represent the function expr using func_name(param1, param2, ...)
throw Exception("not supported");
throw Exception("Function " + func_name + " not supported", ErrorCodes::UNSUPPORTED_METHOD);
}
else
ss << func_name << "(";
bool first = true;
for (const tipb::Expr & child : expr.children())
{
ss << func_name << "(";
bool first = true;
for (const tipb::Expr & child : expr.children())
String s = exprToString(child, input_col, for_parser);
if (first)
{
String s = exprToString(child, input_col);
if (first)
{
first = false;
}
else
{
ss << ", ";
}
ss << s;
first = false;
}
ss << ") ";
return ss.str();
else
{
ss << ", ";
}
ss << s;
}
ss << ") ";
return ss.str();
}

const String & getTypeName(const tipb::Expr & expr) { return tipb::ExprType_Name(expr.tp()); }

String getName(const tipb::Expr & expr, const NamesAndTypesList & current_input_columns)
{
return exprToString(expr, current_input_columns);
return exprToString(expr, current_input_columns, false);
}

bool isAggFunctionExpr(const tipb::Expr & expr)
@@ -225,7 +228,7 @@ Field decodeLiteral(const tipb::Expr & expr)
case tipb::ExprType::MysqlTime:
case tipb::ExprType::MysqlJson:
case tipb::ExprType::ValueList:
throw Exception("mysql type literal is not supported yet");
throw Exception(tipb::ExprType_Name(expr.tp()) + "is not supported yet", ErrorCodes::UNSUPPORTED_METHOD);
default:
return DecodeDatum(cursor, expr.val());
}
@@ -237,6 +240,8 @@ ColumnID getColumnID(const tipb::Expr & expr)
return DecodeInt<Int64>(cursor, expr.val());
}

bool isInOrGlobalInOperator(const String & name) { return name == "in" || name == "notIn" || name == "globalIn" || name == "globalNotIn"; }

std::unordered_map<tipb::ExprType, String> aggFunMap({
{tipb::ExprType::Count, "count"}, {tipb::ExprType::Sum, "sum"}, {tipb::ExprType::Avg, "avg"}, {tipb::ExprType::Min, "min"},
{tipb::ExprType::Max, "max"}, {tipb::ExprType::First, "any"},
3 changes: 2 additions & 1 deletion dbms/src/Flash/Coprocessor/DAGUtils.h
Original file line number Diff line number Diff line change
@@ -24,7 +24,8 @@ bool isColumnExpr(const tipb::Expr & expr);
ColumnID getColumnID(const tipb::Expr & expr);
String getName(const tipb::Expr & expr, const NamesAndTypesList & current_input_columns);
const String & getTypeName(const tipb::Expr & expr);
String exprToString(const tipb::Expr & expr, const NamesAndTypesList & input_col);
String exprToString(const tipb::Expr & expr, const NamesAndTypesList & input_col, bool for_parser = true);
bool isInOrGlobalInOperator(const String & name);
extern std::unordered_map<tipb::ExprType, String> aggFunMap;
extern std::unordered_map<tipb::ScalarFuncSig, String> scalarFunMap;

15 changes: 8 additions & 7 deletions dbms/src/Flash/Coprocessor/InterpreterDAG.cpp
Original file line number Diff line number Diff line change
@@ -82,6 +82,8 @@ void InterpreterDAG::executeTS(const tipb::TableScan & ts, Pipeline & pipeline)
throw Exception("No column is selected in table scan executor", ErrorCodes::COP_BAD_DAG_REQUEST);
}

analyzer = std::make_unique<DAGExpressionAnalyzer>(source_columns, context);

if (!dag.hasAggregation())
{
// if the dag request does not contain agg, then the final output is
@@ -175,28 +177,27 @@ InterpreterDAG::AnalysisResult InterpreterDAG::analyzeExpressions()
{
AnalysisResult res;
ExpressionActionsChain chain;
DAGExpressionAnalyzer analyzer(source_columns, context);
if (dag.hasSelection())
{
analyzer.appendWhere(chain, dag.getSelection(), res.filter_column_name);
analyzer->appendWhere(chain, dag.getSelection(), res.filter_column_name);
res.has_where = true;
res.before_where = chain.getLastActions();
chain.addStep();
}
// There will be either Agg...
if (dag.hasAggregation())
{
analyzer.appendAggregation(chain, dag.getAggregation(), res.aggregation_keys, res.aggregate_descriptions);
analyzer->appendAggregation(chain, dag.getAggregation(), res.aggregation_keys, res.aggregate_descriptions);
res.need_aggregate = true;
res.before_aggregation = chain.getLastActions();

chain.finalize();
chain.clear();

// add cast if type is not match
analyzer.appendAggSelect(chain, dag.getAggregation());
analyzer->appendAggSelect(chain, dag.getAggregation());
//todo use output_offset to reconstruct the final project columns
for (auto element : analyzer.getCurrentInputColumns())
for (auto element : analyzer->getCurrentInputColumns())
{
final_project.emplace_back(element.name, "");
}
@@ -205,10 +206,10 @@ InterpreterDAG::AnalysisResult InterpreterDAG::analyzeExpressions()
if (dag.hasTopN())
{
res.has_order_by = true;
analyzer.appendOrderBy(chain, dag.getTopN(), res.order_column_names);
analyzer->appendOrderBy(chain, dag.getTopN(), res.order_column_names);
}
// Append final project results if needed.
analyzer.appendFinalProject(chain, final_project);
analyzer->appendFinalProject(chain, final_project);
res.before_order_and_select = chain.getLastActions();
chain.finalize();
chain.clear();
3 changes: 3 additions & 0 deletions dbms/src/Flash/Coprocessor/InterpreterDAG.h
Original file line number Diff line number Diff line change
@@ -7,6 +7,7 @@
#pragma GCC diagnostic pop

#include <DataStreams/BlockIO.h>
#include <Flash/Coprocessor/DAGExpressionAnalyzer.h>
#include <Flash/Coprocessor/DAGQuerySource.h>
#include <Flash/Coprocessor/DAGUtils.h>
#include <Interpreters/AggregateDescription.h>
@@ -98,6 +99,8 @@ class InterpreterDAG : public IInterpreter
TMTStoragePtr storage;
TableStructureReadLockPtr table_lock;

std::unique_ptr<DAGExpressionAnalyzer> analyzer;

Poco::Logger * log;
};
} // namespace DB
73 changes: 70 additions & 3 deletions dbms/src/Flash/Coprocessor/tests/cop_test.cpp
Original file line number Diff line number Diff line change
@@ -56,6 +56,7 @@ class FlashClient
grpc::ClientContext clientContext;
clientContext.AddMetadata("user_name", "");
clientContext.AddMetadata("dag_planner", "optree");
clientContext.AddMetadata("dag_expr_field_type_strict_check", "0");
coprocessor::Response response;
grpc::Status status = sp->Coprocessor(&clientContext, *rqst, &response);
if (status.ok())
@@ -64,6 +65,12 @@ class FlashClient
tipb::SelectResponse selectResponse;
if (selectResponse.ParseFromString(response.data()))
{
if (selectResponse.has_error())
{
std::cout << "Coprocessor request failed, error code " << selectResponse.error().code() << " error msg "
<< selectResponse.error().msg();
return status;
}
for (const tipb::Chunk & chunk : selectResponse.chunks())
{
size_t cursor = 0;
@@ -148,6 +155,66 @@ void appendSelection(tipb::DAGRequest & dag_request)
type = expr->mutable_field_type();
type->set_tp(1);
type->set_flag(1 << 5);

// selection i in (5,10,11)
selection->clear_conditions();
expr = selection->add_conditions();
expr->set_tp(tipb::ExprType::ScalarFunc);
expr->set_sig(tipb::ScalarFuncSig::InInt);
col = expr->add_children();
col->set_tp(tipb::ExprType::ColumnRef);
ss.str("");
DB::EncodeNumber<Int64, TiDB::CodecFlagInt>(1, ss);
col->set_val(ss.str());
type = col->mutable_field_type();
type->set_tp(8);
type->set_flag(0);
value = expr->add_children();
value->set_tp(tipb::ExprType::Int64);
ss.str("");
DB::EncodeNumber<Int64, TiDB::CodecFlagInt>(10, ss);
value->set_val(std::string(ss.str()));
type = value->mutable_field_type();
type->set_tp(8);
type->set_flag(1);
type = expr->mutable_field_type();
type->set_tp(1);
type->set_flag(1 << 5);
value = expr->add_children();
value->set_tp(tipb::ExprType::Int64);
ss.str("");
DB::EncodeNumber<Int64, TiDB::CodecFlagInt>(5, ss);
value->set_val(std::string(ss.str()));
type = value->mutable_field_type();
type->set_tp(8);
type->set_flag(1);
type = expr->mutable_field_type();
type->set_tp(1);
type->set_flag(1 << 5);
value = expr->add_children();
value->set_tp(tipb::ExprType::Int64);
ss.str("");
DB::EncodeNumber<Int64, TiDB::CodecFlagInt>(11, ss);
value->set_val(std::string(ss.str()));
type = value->mutable_field_type();
type->set_tp(8);
type->set_flag(1);
type = expr->mutable_field_type();
type->set_tp(1);
type->set_flag(1 << 5);

// selection i is null
/*
selection->clear_conditions();
expr = selection->add_conditions();
expr->set_tp(tipb::ExprType::ScalarFunc);
expr->set_sig(tipb::ScalarFuncSig::IntIsNull);
col = expr->add_children();
col->set_tp(tipb::ExprType::ColumnRef);
ss.str("");
DB::EncodeNumber<Int64, TiDB::CodecFlagInt>(1, ss);
col->set_val(ss.str());
*/
}

void appendAgg(tipb::DAGRequest & dag_request, size_t & result_field_num)
@@ -208,9 +275,9 @@ 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_selection = true;
bool has_agg = false;
bool has_topN = true;
bool has_limit = false;
// construct a dag request
tipb::DAGRequest dagRequest;
39 changes: 39 additions & 0 deletions dbms/src/Interpreters/Set.cpp
Original file line number Diff line number Diff line change
@@ -12,6 +12,8 @@
#include <DataTypes/DataTypeTuple.h>
#include <DataTypes/DataTypeNullable.h>

#include <Flash/Coprocessor/DAGUtils.h>

#include <Parsers/ASTExpressionList.h>
#include <Parsers/ASTFunction.h>
#include <Parsers/ASTLiteral.h>
@@ -22,6 +24,7 @@
#include <Interpreters/NullableUtils.h>

#include <Storages/MergeTree/KeyCondition.h>
#include <Storages/Transaction/TypeMapping.h>


namespace DB
@@ -34,6 +37,7 @@ namespace ErrorCodes
extern const int TYPE_MISMATCH;
extern const int INCORRECT_ELEMENT_OF_SET;
extern const int NUMBER_OF_COLUMNS_DOESNT_MATCH;
extern const int COP_BAD_DAG_REQUEST;
}


@@ -256,6 +260,41 @@ void Set::createFromAST(const DataTypes & types, ASTPtr node, const Context & co
insertFromBlock(block, fill_set_elements);
}

void Set::createFromDAGExpr(const DataTypes & types, const tipb::Expr & expr, bool fill_set_elements)
{
/// Will form a block with values from the set.

Block header;
size_t num_columns = types.size();
if (num_columns != 1)
{
throw Exception("Incorrect element of set, tuple in is not supported yet", ErrorCodes::INCORRECT_ELEMENT_OF_SET);
}
for (size_t i = 0; i < num_columns; ++i)
header.insert(ColumnWithTypeAndName(types[i]->createColumn(), types[i], "_" + toString(i)));
setHeader(header);

MutableColumns columns = header.cloneEmptyColumns();

for (int i = 1; i < expr.children_size(); i++)
{
auto & child = expr.children(i);
// todo support constant expression by constant folding
if (!isLiteralExpr(child))
{
throw Exception("Only literal is supported in children of expr `in`", ErrorCodes::COP_BAD_DAG_REQUEST);
}
Field value = decodeLiteral(child);
DataTypePtr type = child.has_field_type() ? getDataTypeByFieldType(child.field_type()) : types[0];
value = convertFieldToType(value, *type);

if (!value.isNull())
columns[0]->insert(value);
}

Block block = header.cloneWithColumns(std::move(columns));
insertFromBlock(block, fill_set_elements);
}

ColumnPtr Set::execute(const Block & block, bool negative) const
{
10 changes: 10 additions & 0 deletions dbms/src/Interpreters/Set.h
Original file line number Diff line number Diff line change
@@ -1,5 +1,10 @@
#pragma once

#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wunused-parameter"
#include <tipb/expression.pb.h>
#pragma GCC diagnostic pop

#include <shared_mutex>
#include <Core/Block.h>
#include <DataStreams/SizeLimits.h>
@@ -48,6 +53,11 @@ class Set
*/
void createFromAST(const DataTypes & types, ASTPtr node, const Context & context, bool fill_set_elements);

/**
* Create a Set from DAG Expr, used when processing DAG Request
*/
void createFromDAGExpr(const DataTypes & types, const tipb::Expr & expr, bool fill_set_elements);

/** Create a Set from stream.
* Call setHeader, then call insertFromBlock for each block.
*/