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.

Sign up for GitHub

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

Jump to bottom

add more coprocessor mock tests #185

Merged
merged 9 commits into from
Aug 20, 2019
Merged

add more coprocessor mock tests #185

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
283f19d
check validation of dag exprs field type
windtalker Aug 16, 2019
22c7ddd
format code
windtalker Aug 16, 2019
0c998ac
address comments
windtalker Aug 16, 2019
d9df281
add more filter test
windtalker Aug 19, 2019
59cff30
Merge branch 'cop' of https://github.com/pingcap/tics into xf_cop
windtalker Aug 19, 2019
a560044
add data type tests
windtalker Aug 20, 2019
dc8b237
Merge branch 'cop' of https://github.com/pingcap/tics into xf_cop
windtalker Aug 20, 2019
db1b66d
remove useless comment
windtalker Aug 20, 2019
97b159a
disable decimal test
windtalker Aug 20, 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
9 changes: 9 additions & 0 deletions dbms/src/Debug/MockTiDB.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,15 @@ ColumnInfo getColumnInfoFromColumn(const NameAndTypePair & column, ColumnID id)
column_info.flen = decimal_type->getPrec();
column_info.decimal = decimal_type->getScale();
}
if (checkDataType<DataTypeEnum16>(nested_type))
{
auto enum16_type = checkAndGetDataType<DataTypeEnum16>(nested_type);
column_info.tp = TiDB::TypeEnum;
for (auto & element : enum16_type->getValues())
{
column_info.elems.emplace_back(element.first, element.second);
}
}

#ifdef M
#error "Please undefine macro M first."
Expand Down
91 changes: 69 additions & 22 deletions dbms/src/Debug/dbgFuncCoprocessor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,8 @@ extern const int BAD_ARGUMENTS;
extern const int LOGICA_ERROR;
} // namespace ErrorCodes

using DAGField = std::pair<String, tipb::FieldType>;
using DAGSchema = std::vector<DAGField>;
using DAGColumnInfo = std::pair<String, ColumnInfo>;
using DAGSchema = std::vector<DAGColumnInfo>;
using SchemaFetcher = std::function<TiDB::TableInfo(const String &, const String &)>;
std::tuple<TableID, DAGSchema, tipb::DAGRequest> compileQuery(
Context & context, const String & query, SchemaFetcher schema_fetcher, Timestamp start_ts);
Expand All @@ -49,7 +49,8 @@ BlockInputStreamPtr dbgFuncDAG(Context & context, const ASTs & args)
region_id = safeGet<RegionID>(typeid_cast<const ASTLiteral &>(*args[1]).value);
Timestamp start_ts = context.getTMTContext().getPDClient()->getTS();

auto [table_id, schema, dag_request] = compileQuery(context, query,
auto [table_id, schema, dag_request] = compileQuery(
context, query,
[&](const String & database_name, const String & table_name) {
auto storage = context.getTable(database_name, table_name);
auto mmt = std::dynamic_pointer_cast<StorageMergeTree>(storage);
Expand Down Expand Up @@ -91,7 +92,8 @@ BlockInputStreamPtr dbgFuncMockDAG(Context & context, const ASTs & args)
if (start_ts == 0)
start_ts = context.getTMTContext().getPDClient()->getTS();

auto [table_id, schema, dag_request] = compileQuery(context, query,
auto [table_id, schema, dag_request] = compileQuery(
context, query,
[&](const String & database_name, const String & table_name) {
return MockTiDB::instance().getTableByName(database_name, table_name)->table_info;
},
Expand All @@ -111,6 +113,26 @@ struct ExecutorCtx
std::unordered_map<String, tipb::Expr *> col_ref_map;
};

tipb::FieldType columnInfoToFieldType(const ColumnInfo & ci)
{
tipb::FieldType ret;
ret.set_tp(ci.tp);
ret.set_flag(ci.flag);
ret.set_flen(ci.flen);
ret.set_decimal(ci.decimal);
return ret;
}

ColumnInfo fieldTypeToColumnInfo(const tipb::FieldType & field_type)
{
ColumnInfo ret;
ret.tp = static_cast<TiDB::TP>(field_type.tp());
ret.flag = field_type.flag();
ret.flen = field_type.flen();
ret.decimal = field_type.decimal();
return ret;
}

void compileExpr(const DAGSchema & input, ASTPtr ast, tipb::Expr * expr, std::unordered_set<String> & referred_columns,
std::unordered_map<String, tipb::Expr *> & col_ref_map)
{
Expand All @@ -120,7 +142,7 @@ void compileExpr(const DAGSchema & input, ASTPtr ast, tipb::Expr * expr, std::un
if (ft == input.end())
throw DB::Exception("No such column " + id->getColumnName(), ErrorCodes::NO_SUCH_COLUMN_IN_TABLE);
expr->set_tp(tipb::ColumnRef);
*(expr->mutable_field_type()) = (*ft).second;
*(expr->mutable_field_type()) = columnInfoToFieldType((*ft).second);

referred_columns.emplace((*ft).first);
col_ref_map.emplace((*ft).first, expr);
Expand Down Expand Up @@ -208,6 +230,24 @@ void compileExpr(const DAGSchema & input, ASTPtr ast, tipb::Expr * expr, std::un
}
}

void compileFilter(const DAGSchema & input, ASTPtr ast, tipb::Selection * filter, std::unordered_set<String> & referred_columns,
std::unordered_map<String, tipb::Expr *> & col_ref_map)
{
if (auto * func = typeid_cast<ASTFunction *>(ast.get()))
{
if (func->name == "and")
{
for (auto & child : func->arguments->children)
{
compileFilter(input, child, filter, referred_columns, col_ref_map);
}
return;
}
}
tipb::Expr * cond = filter->add_conditions();
compileExpr(input, ast, cond, referred_columns, col_ref_map);
}

std::tuple<TableID, DAGSchema, tipb::DAGRequest> compileQuery(
Context & context, const String & query, SchemaFetcher schema_fetcher, Timestamp start_ts)
{
Expand Down Expand Up @@ -257,12 +297,13 @@ std::tuple<TableID, DAGSchema, tipb::DAGRequest> compileQuery(
DAGSchema ts_output;
for (const auto & column_info : table_info.columns)
{
tipb::FieldType field_type;
field_type.set_tp(column_info.tp);
field_type.set_flag(column_info.flag);
field_type.set_flen(column_info.flen);
field_type.set_decimal(column_info.decimal);
ts_output.emplace_back(std::make_pair(column_info.name, std::move(field_type)));
ColumnInfo ci;
ci.tp = column_info.tp;
ci.flag = column_info.flag;
ci.flen = column_info.flen;
ci.decimal = column_info.flen;
ci.elems = column_info.elems;
ts_output.emplace_back(std::make_pair(column_info.name, std::move(ci)));
}
executor_ctx_map.emplace(ts_exec, ExecutorCtx{nullptr, std::move(ts_output), std::unordered_map<String, tipb::Expr *>{}});
last_executor = ts_exec;
Expand All @@ -274,9 +315,8 @@ std::tuple<TableID, DAGSchema, tipb::DAGRequest> compileQuery(
tipb::Executor * filter_exec = dag_request.add_executors();
filter_exec->set_tp(tipb::ExecType::TypeSelection);
tipb::Selection * filter = filter_exec->mutable_selection();
tipb::Expr * cond = filter->add_conditions();
std::unordered_map<String, tipb::Expr *> col_ref_map;
compileExpr(executor_ctx_map[last_executor].output, ast_query.where_expression, cond, referred_columns, col_ref_map);
compileFilter(executor_ctx_map[last_executor].output, ast_query.where_expression, filter, referred_columns, col_ref_map);
executor_ctx_map.emplace(filter_exec, ExecutorCtx{last_executor, executor_ctx_map[last_executor].output, std::move(col_ref_map)});
last_executor = filter_exec;
}
Expand Down Expand Up @@ -323,14 +363,21 @@ std::tuple<TableID, DAGSchema, tipb::DAGRequest> compileQuery(
[&](const auto & field) { return referred_columns.count(field.first) == 0; }),
executor_ctx.output.end());

for (const auto & field : executor_ctx.output)
for (const auto & info : executor_ctx.output)
{
tipb::ColumnInfo * ci = ts->add_columns();
ci->set_column_id(table_info.getColumnID(field.first));
ci->set_tp(field.second.tp());
ci->set_flag(field.second.flag());
ci->set_columnlen(field.second.flen());
ci->set_decimal(field.second.decimal());
ci->set_column_id(table_info.getColumnID(info.first));
ci->set_tp(info.second.tp);
ci->set_flag(info.second.flag);
ci->set_columnlen(info.second.flen);
ci->set_decimal(info.second.decimal);
if (info.second.elems.size() != 0)
{
for (auto & pair : info.second.elems)
{
ci->add_elems(pair.first);
}
}
}

return;
Expand Down Expand Up @@ -399,7 +446,7 @@ std::tuple<TableID, DAGSchema, tipb::DAGRequest> compileQuery(
throw DB::Exception("Unsupported agg function " + func->name, ErrorCodes::LOGICAL_ERROR);
}

schema.emplace_back(std::make_pair(func->getColumnName(), agg_func->field_type()));
schema.emplace_back(std::make_pair(func->getColumnName(), fieldTypeToColumnInfo(agg_func->field_type())));
}

if (has_gby)
Expand All @@ -408,7 +455,7 @@ std::tuple<TableID, DAGSchema, tipb::DAGRequest> compileQuery(
{
tipb::Expr * gby = agg->add_group_by();
compileExpr(executor_ctx_map[last_executor].output, child, gby, referred_columns, col_ref_map);
schema.emplace_back(std::make_pair(child->getColumnName(), gby->field_type()));
schema.emplace_back(std::make_pair(child->getColumnName(), fieldTypeToColumnInfo(gby->field_type())));
}
}

Expand Down Expand Up @@ -501,7 +548,7 @@ BlockInputStreamPtr outputDAGResponse(Context &, const DAGSchema & schema, const
for (auto & field : schema)
{
const auto & name = field.first;
auto data_type = getDataTypeByFieldType(field.second);
auto data_type = getDataTypeByColumnInfo(field.second);
ColumnWithTypeAndName col(data_type, name);
col.column->assumeMutable()->reserve(rows.size());
columns.emplace_back(std::move(col));
Expand Down
Loading