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

Fix date/datetime/bit encode error #200

Merged
merged 40 commits into from
Aug 26, 2019
Merged
Changes from 1 commit
Commits
Show all changes
40 commits
Select commit Hold shift + click to select a range
309ab6b
Enhance dbg invoke and add dag as schemaful function
zanmato1984 Aug 7, 2019
31d83c4
Add basic sql parse to dag
zanmato1984 Aug 8, 2019
3c4c508
Merge cop
zanmato1984 Aug 8, 2019
2de7311
Merge branch 'cop' into cop-ruoxi
zanmato1984 Aug 8, 2019
63a5800
Column id starts from 1
zanmato1984 Aug 8, 2019
232e7de
Fix value to ref
zanmato1984 Aug 8, 2019
1b14a12
Add basic dag test
zanmato1984 Aug 8, 2019
25eb831
Fix dag bugs and pass 1st mock test
zanmato1984 Aug 8, 2019
80f9fc6
Make dag go normal routine and add mock dag
zanmato1984 Aug 8, 2019
a1173e1
Add todo
zanmato1984 Aug 8, 2019
c8109f6
Add comment
zanmato1984 Aug 8, 2019
7dc0397
Fix gcc compile error
zanmato1984 Aug 8, 2019
66d9e8a
Enhance dag test
zanmato1984 Aug 8, 2019
36d1117
Address comments
zanmato1984 Aug 9, 2019
8aea5aa
Merge branch 'cop' into cop-ruoxi
zanmato1984 Aug 9, 2019
f62b318
Merge branch 'cop' into cop-ruoxi
zanmato1984 Aug 12, 2019
a9fe9f9
Enhance mock sql -> dag compiler and add project test
zanmato1984 Aug 12, 2019
1372262
Mock sql dag compiler support more expression types and add filter test
zanmato1984 Aug 13, 2019
e2f9a02
Add topn and limit test
zanmato1984 Aug 13, 2019
8cea243
Add agg for sql -> dag parser and agg test
zanmato1984 Aug 13, 2019
5008a7a
Merge branch 'cop' of github.com:pingcap/tics into cop-ruoxi
zanmato1984 Aug 14, 2019
8fb4d52
Add dag specific codec
zanmato1984 Aug 15, 2019
c77310e
Merge branch 'cop' of github.com:pingcap/tics into cop-ruoxi
zanmato1984 Aug 15, 2019
0c8e3a5
type
zanmato1984 Aug 15, 2019
76b5444
Merge branch 'cop' of github.com:pingcap/tics into cop-ruoxi
zanmato1984 Aug 15, 2019
41d2b4f
Update codec accordingly
zanmato1984 Aug 15, 2019
17111f5
Remove cop-test
zanmato1984 Aug 15, 2019
f3adf8e
Merge cop
zanmato1984 Aug 19, 2019
31684fa
Merge branch 'cop' of github.com:pingcap/tics into cop-ruoxi
zanmato1984 Aug 20, 2019
7a74ffb
Merge branch 'cop' of github.com:pingcap/tics into cop-ruoxi
zanmato1984 Aug 23, 2019
0c29e49
Merge branch 'cop' of github.com:pingcap/tics into cop-ruoxi
zanmato1984 Aug 24, 2019
d6eefa7
Pass tests after merging master
zanmato1984 Aug 24, 2019
916f5bb
Merge branch 'cop' of github.com:pingcap/tics into cop-ruoxi
zanmato1984 Aug 25, 2019
50873d6
Copy some changes from xufei
zanmato1984 Aug 25, 2019
96de82e
Enable date/datetime test
zanmato1984 Aug 25, 2019
ac37a78
Enable date/datetime test
zanmato1984 Aug 25, 2019
2153142
Refine code
zanmato1984 Aug 25, 2019
1485066
Adjust date/datetime tiflash rep to UInt
zanmato1984 Aug 26, 2019
d01d0b6
Fix datetime to Int
zanmato1984 Aug 26, 2019
68b4955
Typo
zanmato1984 Aug 26, 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
Enhance mock sql -> dag compiler and add project test
zanmato1984 committed Aug 12, 2019
commit a9fe9f923da32450fe038d42370345b675d504ca
223 changes: 189 additions & 34 deletions dbms/src/Debug/dbgFuncCoprocessor.cpp
Original file line number Diff line number Diff line change
@@ -3,6 +3,8 @@
#include <Debug/MockTiDB.h>
#include <Debug/dbgFuncCoprocessor.h>
#include <Flash/Coprocessor/DAGDriver.h>
#include <Parsers/ASTAsterisk.h>
#include <Parsers/ASTFunction.h>
#include <Parsers/ASTIdentifier.h>
#include <Parsers/ASTLiteral.h>
#include <Parsers/ASTSelectQuery.h>
@@ -21,6 +23,7 @@ namespace DB
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
extern const int LOGICA_ERROR;
} // namespace ErrorCodes

using DAGField = std::pair<String, tipb::FieldType>;
@@ -98,6 +101,42 @@ BlockInputStreamPtr dbgFuncMockDAG(Context & context, const ASTs & args)
return outputDAGResponse(context, schema, dag_response);
}

struct ExecutorCtx
{
tipb::Executor * input;
DAGSchema output;
std::unordered_map<String, tipb::Expr *> col_ref_map;
};

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)
{
if (ASTIdentifier * id = typeid_cast<ASTIdentifier *>(ast.get()))
{
auto ft = std::find_if(input.begin(), input.end(), [&](const auto & field) { return field.first == id->getColumnName(); });
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;

referred_columns.emplace((*ft).first);
col_ref_map.emplace((*ft).first, expr);
}
// else if (ASTFunction * func = typeid_cast<ASTFunction *>(ast.get()))
// {
// }
else
{
throw DB::Exception("Unsupported expression " + ast->getColumnName(), ErrorCodes::LOGICAL_ERROR);
}

for (const auto & child_ast : ast->children)
{
tipb::Expr * child = expr->add_children();
compileExpr(input, child_ast, child, referred_columns, col_ref_map);
}
};

std::tuple<TableID, DAGSchema, tipb::DAGRequest> compileQuery(
Context & context, const String & query, SchemaFetcher schema_fetcher, Timestamp start_ts)
{
@@ -110,59 +149,175 @@ std::tuple<TableID, DAGSchema, tipb::DAGRequest> compileQuery(
ASTPtr ast = parseQuery(parser, query.data(), query.data() + query.size(), "from DAG compiler", 0);
ASTSelectQuery & ast_query = typeid_cast<ASTSelectQuery &>(*ast);

String database_name, table_name;
auto query_database = ast_query.database();
auto query_table = ast_query.table();
if (query_database)
database_name = typeid_cast<ASTIdentifier &>(*query_database).name;
if (query_table)
table_name = typeid_cast<ASTIdentifier &>(*query_table).name;
if (!query_table)
/// Get table metadata.
TiDB::TableInfo table_info;
{
String database_name, table_name;
auto query_database = ast_query.database();
auto query_table = ast_query.table();
if (query_database)
database_name = typeid_cast<ASTIdentifier &>(*query_database).name;
if (query_table)
table_name = typeid_cast<ASTIdentifier &>(*query_table).name;
if (!query_table)
{
database_name = "system";
table_name = "one";
}
else if (!query_database)
{
database_name = context.getCurrentDatabase();
}

table_info = schema_fetcher(database_name, table_name);
}

std::map<tipb::Executor *, ExecutorCtx> executor_ctx_map;
std::unordered_set<String> referred_columns;
tipb::TableScan * ts = nullptr;
tipb::Executor * last_executor = nullptr;

/// Table scan.
{
tipb::Executor * ts_exec = dag_request.add_executors();
ts_exec->set_tp(tipb::ExecType::TypeTableScan);
ts = ts_exec->mutable_tbl_scan();
ts->set_table_id(table_info.id);
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)));
}
executor_ctx_map.emplace(ts_exec, ExecutorCtx{nullptr, std::move(ts_output), std::unordered_map<String, tipb::Expr *>{}});
last_executor = ts_exec;
}

/// Filter.
if (ast_query.where_expression)
{
database_name = "system";
table_name = "one";
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);
executor_ctx_map.emplace(filter_exec, ExecutorCtx{last_executor, executor_ctx_map[last_executor].output, std::move(col_ref_map)});
last_executor = filter_exec;
}
else if (!query_database)

/// TopN.
if (ast_query.order_expression_list && ast_query.limit_length)
{
database_name = context.getCurrentDatabase();
tipb::Executor * topn_exec = dag_request.add_executors();
topn_exec->set_tp(tipb::ExecType::TypeTopN);
tipb::TopN * topN = topn_exec->mutable_topn();
std::unordered_map<String, tipb::Expr *> col_ref_map;
for (const auto & child : ast_query.order_expression_list->children)
{
tipb::ByItem * by = topN->add_order_by();
tipb::Expr * expr = by->mutable_expr();
compileExpr(executor_ctx_map[last_executor].output, child, expr, referred_columns, col_ref_map);
}
auto limit = safeGet<UInt64>(typeid_cast<ASTLiteral &>(*ast_query.limit_length).value);
topN->set_limit(limit);
executor_ctx_map.emplace(topn_exec, ExecutorCtx{last_executor, executor_ctx_map[last_executor].output, std::move(col_ref_map)});
last_executor = topn_exec;
}
auto table_info = schema_fetcher(database_name, table_name);

tipb::Executor * executor = dag_request.add_executors();
executor->set_tp(tipb::ExecType::TypeTableScan);
tipb::TableScan * ts = executor->mutable_tbl_scan();
ts->set_table_id(table_info.id);
size_t i = 0;
for (const auto & column_info : table_info.columns)

/// Aggregation.
if (ast_query.group_expression_list) {}
if (ast_query.select_expression_list /* select_expression_list has agg*/) {}

/// Finalize.
if (!last_executor->has_aggregation())
{
tipb::ColumnInfo * ci = ts->add_columns();
ci->set_column_id(column_info.id);
ci->set_tp(column_info.tp);
ci->set_flag(column_info.flag);
std::vector<String> final_output;
for (const auto & expr : ast_query.select_expression_list->children)
{
if (ASTIdentifier * id = typeid_cast<ASTIdentifier *>(expr.get()))
{
referred_columns.emplace(id->getColumnName());
final_output.emplace_back(id->getColumnName());
}
else if (typeid_cast<ASTAsterisk *>(expr.get()))
{
const auto & last_output = executor_ctx_map[last_executor].output;
for (const auto & field : last_output)
{
referred_columns.emplace(field.first);
final_output.push_back(field.first);
}
}
else
{
throw DB::Exception("Unsupported expression type in select", ErrorCodes::LOGICAL_ERROR);
}
}

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);
schema.emplace_back(std::make_pair(column_info.name, std::move(field_type)));
std::function<void(ExecutorCtx &)> column_pruner = [&](ExecutorCtx & executor_ctx) {
if (!executor_ctx.input)
{
executor_ctx.output.erase(std::remove_if(executor_ctx.output.begin(), executor_ctx.output.end(),
[&](const auto & field) { return referred_columns.count(field.first) == 0; }),
executor_ctx.output.end());

dag_request.add_output_offsets(i);
for (const auto & field : 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());
}

i++;
}
return;
}
column_pruner(executor_ctx_map[executor_ctx.input]);
const auto & last_output = executor_ctx_map[executor_ctx.input].output;
for (const auto & pair : executor_ctx.col_ref_map)
{
auto iter
= std::find_if(last_output.begin(), last_output.end(), [&](const auto & field) { return field.first == pair.first; });
if (iter == last_output.end())
throw DB::Exception("Column not found when pruning: " + pair.first, ErrorCodes::LOGICAL_ERROR);
std::stringstream ss;
DB::EncodeNumber<Int64, TiDB::CodecFlagInt>(iter - last_output.begin(), ss);
pair.second->set_val(ss.str());
}
executor_ctx.output = last_output;
};
column_pruner(executor_ctx_map[last_executor]);

// TODO: Other operator compile.
const auto & last_output = executor_ctx_map[last_executor].output;
for (const auto & field : final_output)
{
auto iter
= std::find_if(last_output.begin(), last_output.end(), [&](const auto & last_field) { return last_field.first == field; });
if (iter == last_output.end())
throw DB::Exception("Column not found after pruning: " + field, ErrorCodes::LOGICAL_ERROR);
dag_request.add_output_offsets(iter - last_output.begin());
schema.push_back(*iter);
}
}

return std::make_tuple(table_info.id, std::move(schema), std::move(dag_request));
}

tipb::SelectResponse executeDAGRequest(
Context & context, const tipb::DAGRequest & dag_request, RegionID region_id, UInt64 region_version, UInt64 region_conf_version)
{
Logger * log = &Logger::get("MockDAG");
LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": Handling DAG request: " << dag_request.DebugString());
tipb::SelectResponse dag_response;
DAGDriver driver(context, dag_request, region_id, region_version, region_conf_version, dag_response, true);
driver.execute();
LOG_DEBUG(log, __PRETTY_FUNCTION__ << ": Handle DAG request done");
return dag_response;
}

41 changes: 41 additions & 0 deletions tests/mutable-test/txn_dag/project.test
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
# Preparation.
=> DBGInvoke __enable_schema_sync_service('true')

=> DBGInvoke __drop_tidb_table(default, test)
=> drop table if exists default.test

=> DBGInvoke __set_flush_threshold(1000000, 1000000)

# Data.
=> DBGInvoke __mock_tidb_table(default, test, 'col_1 String, col_2 Int64')
=> DBGInvoke __refresh_schemas()
=> DBGInvoke __put_region(4, 0, 100, default, test)
=> DBGInvoke __raft_insert_row(default, test, 4, 50, 'test1', 666)

# DAG read by not specifying region id, select *.
=> DBGInvoke dag('select * from default.test') " --dag_planner="optree
┌─col_1─┬─col_2─┐
│ test1 │ 666 │
└───────┴───────┘

# DAG read by not specifying region id, select col_1.
=> DBGInvoke dag('select col_1 from default.test') " --dag_planner="optree
┌─col_1─┐
│ test1 │
└───────┘

# DAG read by explicitly specifying region id, select col_2.
=> DBGInvoke dag('select col_2 from default.test', 4) " --dag_planner="optree
┌─col_2─┐
│ 666 │
└───────┘

# Mock DAG read, select col_2, col_1, col_2.
=> DBGInvoke mock_dag('select col_2, col_1, col_2 from default.test', 4) " --dag_planner="optree
┌─col_2─┬─col_1─┬─col_2─┐
│ 666 │ test1 │ 666 │
└───────┴───────┴───────┘

# Clean up.
=> DBGInvoke __drop_tidb_table(default, test)
=> drop table if exists default.test
2 changes: 1 addition & 1 deletion tests/mutable-test/txn_dag/table_scan.test
Original file line number Diff line number Diff line change
@@ -6,7 +6,7 @@

=> DBGInvoke __set_flush_threshold(1000000, 1000000)

# Data
# Data.
=> DBGInvoke __mock_tidb_table(default, test, 'col_1 String')
=> DBGInvoke __refresh_schemas()
=> DBGInvoke __put_region(4, 0, 100, default, test)