database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-08-01 06:46:55 +03:00
Code Activity

feat(CSEP): CSEP printer with indentations to simplify reading + rewriter skeleton + some test binary to describe minimalistic CSEP localy

Browse Source
This commit is contained in:
drrtuy
2025-06-05 10:06:54 +00:00
parent 3a91cded27
commit 79008f4f69
7 changed files with 1046 additions and 843 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
dbcon/execplan/CMakeLists.txt
View File

@ -47,3 +47,11 @@ set(execplan_LIB_SRCS
columnstore_library(execplan ${execplan_LIB_SRCS})
columnstore_link(execplan messageqcpp ${NETSNMP_LIBRARIES} ${ENGINE_DT_LIB} pron loggingcpp)
columnstore_executable(csep_rewrite csep_rewrite.cpp)
columnstore_link(csep_rewrite
${ENGINE_LDFLAGS}
${NETSNMP_LIBRARIES}
${ENGINE_WRITE_LIBS}
loggingcpp
)

96
dbcon/execplan/calpontselectexecutionplan.cpp
View File

@ -193,51 +193,64 @@ void CalpontSelectExecutionPlan::havingTokenList(const FilterTokenList& havingTo
having(parser.parse(tokens.begin(), tokens.end()));
}
string CalpontSelectExecutionPlan::toString() const
std::string endlWithIndent(const size_t ident)
{
ostringstream output;
output << endl;
output << std::string(ident, ' ');
return output.str();
}
string CalpontSelectExecutionPlan::toString(const size_t ident) const
{
ostringstream output;
output << ">SELECT ";
output << std::string(ident, ' ') << "SELECT ";
if (distinct())
{
output << "DISTINCT ";
}
output << "limit: " << limitStart() << " - " << limitNum() << endl;
output << "limit: " << limitStart() << " - " << limitNum() << endlWithIndent(ident);
switch (location())
{
case CalpontSelectExecutionPlan::MAIN: output << "MAIN" << endl; break;
case CalpontSelectExecutionPlan::MAIN: output << "MAIN" << endlWithIndent(ident); break;
case CalpontSelectExecutionPlan::FROM: output << "FROM" << endl; break;
case CalpontSelectExecutionPlan::FROM: output << "FROM" << endlWithIndent(ident); break;
case CalpontSelectExecutionPlan::WHERE: output << "WHERE" << endl; break;
case CalpontSelectExecutionPlan::WHERE: output << "WHERE" << endlWithIndent(ident); break;
case CalpontSelectExecutionPlan::HAVING: output << "HAVING" << endl; break;
case CalpontSelectExecutionPlan::HAVING: output << "HAVING" << endlWithIndent(ident); break;
}
// Returned Column
CalpontSelectExecutionPlan::ReturnedColumnList retCols = returnedCols();
output << ">>Returned Columns" << endl;
output << ">>Returned Columns" << endlWithIndent(ident);
output << std::string(ident, ' ');
uint32_t seq = 0;
for (unsigned int i = 0; i < retCols.size(); i++)
{
output << *retCols[i] << endl;
output << *retCols[i] << endlWithIndent(ident);
if (retCols[i]->colSource() & SELECT_SUB)
{
output << "select sub -- " << endl;
output << "select sub -- " << endlWithIndent(ident);
CalpontSelectExecutionPlan* plan =
dynamic_cast<CalpontSelectExecutionPlan*>(fSelectSubList[seq++].get());
if (plan)
output << "{" << *plan << "}" << endl;
output << "{" << plan->toString(ident + 2) << "}" << endlWithIndent(ident);
}
}
// From Clause
CalpontSelectExecutionPlan::TableList tables = tableList();
output << ">>From Tables" << endl;
output << ">>From Tables" << endlWithIndent(ident);
output << std::string(ident, ' ');
seq = 0;
for (unsigned int i = 0; i < tables.size(); i++)
@ -245,42 +258,55 @@ string CalpontSelectExecutionPlan::toString() const
// derived table
if (tables[i].schema.length() == 0 && tables[i].table.length() == 0)
{
output << "derived table - " << tables[i].alias << endl;
output << "derived table - " << tables[i].alias << endlWithIndent(ident);
CalpontSelectExecutionPlan* plan =
dynamic_cast<CalpontSelectExecutionPlan*>(fDerivedTableList[seq++].get());
if (plan)
output << "{" << *plan << "}" << endl;
{
output << "{" << plan->toString(ident + 2) << "}" << endlWithIndent(ident);
}
output << std::string(ident, ' ');
}
else
{
output << tables[i] << endl;
output << std::string(ident, ' ');
}
}
// Filters
output << ">>Filters" << endl;
output << ">>Filters" << endlWithIndent(ident);
output << std::string(ident, ' ');
if (filters() != nullptr)
filters()->walk(ParseTree::print, output);
else
output << "empty filter tree" << endl;
output << "empty filter tree" << endlWithIndent(ident);
output << std::string(ident, ' ');
// Group by columns
const CalpontSelectExecutionPlan::GroupByColumnList& gbc = groupByCols();
if (gbc.size() > 0)
{
output << ">>Group By Columns" << endl;
output << ">>Group By Columns" << endlWithIndent(ident);
output << std::string(ident, ' ');
for (unsigned int i = 0; i < gbc.size(); i++)
output << *gbc[i] << endl;
{
output << *gbc[i] << endlWithIndent(ident);
}
output << std::string(ident, ' ');
}
// Having
if (having() != nullptr)
{
output << ">>Having" << endl;
output << ">>Having" << endlWithIndent(ident);
output << std::string(ident, ' ');
having()->walk(ParseTree::print, output);
}
@ -289,39 +315,39 @@ string CalpontSelectExecutionPlan::toString() const
if (obc.size() > 0)
{
output << ">>Order By Columns" << endl;
output << ">>Order By Columns" << endlWithIndent(ident);
for (unsigned int i = 0; i < obc.size(); i++)
output << *obc[i] << endl;
output << *obc[i] << endlWithIndent(ident);
}
output << "SessionID: " << fSessionID << endl;
output << "TxnID: " << fTxnID << endl;
output << "VerID: " << fVerID << endl;
output << "TraceFlags: " << fTraceFlags << endl;
output << "StatementID: " << fStatementID << endl;
output << "DistUnionNum: " << (int)fDistinctUnionNum << endl;
output << "Limit: " << fLimitStart << " - " << fLimitNum << endl;
output << "String table threshold: " << fStringTableThreshold << endl;
output << "SessionID: " << fSessionID << endlWithIndent(ident);
output << "TxnID: " << fTxnID << endlWithIndent(ident);
output << "VerID: " << fVerID << endlWithIndent(ident);
output << "TraceFlags: " << fTraceFlags << endlWithIndent(ident);
output << "StatementID: " << fStatementID << endlWithIndent(ident);
output << "DistUnionNum: " << (int)fDistinctUnionNum << endlWithIndent(ident);
output << "Limit: " << fLimitStart << " - " << fLimitNum << endlWithIndent(ident);
output << "String table threshold: " << fStringTableThreshold << endlWithIndent(ident);
output << "--- Column Map ---" << endl;
output << "--- Column Map ---" << endlWithIndent(ident);
CalpontSelectExecutionPlan::ColumnMap::const_iterator iter;
for (iter = columnMap().begin(); iter != columnMap().end(); iter++)
output << (*iter).first << " : " << (*iter).second << endl;
output << (*iter).first << " : " << (*iter).second << endlWithIndent(ident);
output << "UUID: " << fUuid << endl;
output << "QueryType: " << queryType() << endl;
output << "UUID: " << fUuid << endlWithIndent(ident);
output << "QueryType: " << queryType() << endlWithIndent(ident);
if (!unionVec().empty())
output << "\n--- Union Unit ---" << endl;
output << "\n--- Union Unit ---" << endlWithIndent(ident);
for (unsigned i = 0; i < unionVec().size(); i++)
{
CalpontSelectExecutionPlan* plan = dynamic_cast<CalpontSelectExecutionPlan*>(unionVec()[i].get());
if (plan)
output << "{" << *plan << "}\n" << endl;
output << "{" << *plan << "}\n" << endlWithIndent(ident);
}
return output.str();

2
dbcon/execplan/calpontselectexecutionplan.h
View File

@ -797,7 +797,7 @@ class CalpontSelectExecutionPlan : public CalpontExecutionPlan
* Return a string rep of the CSEP
* @return a string
*/
virtual std::string toString() const;
virtual std::string toString(const size_t ident = 0) const;
/** @brief Is this an internal query?
*

5
dbcon/execplan/calpontsystemcatalog.h
View File

@ -458,8 +458,13 @@ class CalpontSystemCatalog : public datatypes::SystemCatalog
{
return !(*this == rhs);
}
bool isColumnstore() const
{
return fisColumnStore;
}
void serialize(messageqcpp::ByteStream&) const;
void unserialize(messageqcpp::ByteStream&);
friend std::ostream& operator<<(std::ostream& os, const TableAliasName& rhs);
};

87
dbcon/execplan/csep_rewrite.cpp Normal file
View File

@ -0,0 +1,87 @@
#include "execplan/calpontselectexecutionplan.h"
#include "simplecolumn.h"
#include "execplan/calpontsystemcatalog.h"
#include "simplefilter.h"
#include "constantcolumn.h"
using namespace execplan;
const SOP opeq(new Operator("="));
int main()
{
CalpontSelectExecutionPlan csep;
CalpontSelectExecutionPlan::ReturnedColumnList returnedColumnList;
CalpontSelectExecutionPlan::ColumnMap colMap;
string columnlength = CALPONT_SCHEMA + "." + SYSCOLUMN_TABLE + "." + COLUMNLEN_COL;
SimpleColumn* col[1];
col[0] = new SimpleColumn(columnlength, 22222222);
SRCP srcp;
srcp.reset(col[0]);
colMap.insert({columnlength, srcp});
csep.columnMapNonStatic(colMap);
srcp.reset(col[0]->clone());
returnedColumnList.push_back(srcp);
csep.returnedCols(returnedColumnList);
{
SCSEP csepDerived(new CalpontSelectExecutionPlan());
CalpontSelectExecutionPlan::ReturnedColumnList returnedColumnListLocal;
CalpontSelectExecutionPlan::ColumnMap colMapLocal;
string columnlength = CALPONT_SCHEMA + "." + SYSCOLUMN_TABLE + "." + COLUMNLEN_COL;
SimpleColumn* col[1];
col[0] = new SimpleColumn(columnlength, 22222222);
SRCP srcpLocal;
srcpLocal.reset(col[0]);
colMapLocal.insert({columnlength, srcpLocal});
csepDerived->columnMapNonStatic(colMapLocal);
srcp.reset(col[0]->clone());
returnedColumnListLocal.push_back(srcpLocal);
csepDerived->returnedCols(returnedColumnList);
CalpontSelectExecutionPlan::SelectList derivedTables;
derivedTables.push_back(csepDerived);
csep.derivedTableList(derivedTables);
}
CalpontSelectExecutionPlan::TableList tableList = {execplan::CalpontSystemCatalog::TableAliasName("", "", "alias")};
csep.tableList(tableList);
CalpontSelectExecutionPlan::SelectList unionVec;
for (size_t i = 0; i < 3; ++i)
{
SCSEP plan(new CalpontSelectExecutionPlan());
CalpontSelectExecutionPlan::ReturnedColumnList returnedColumnListLocal;
CalpontSelectExecutionPlan::ColumnMap colMapLocal;
SRCP srcpLocal;
srcpLocal.reset(col[0]);
colMapLocal.insert({columnlength, srcpLocal});
plan->columnMapNonStatic(colMapLocal);
srcpLocal.reset(col[0]->clone());
returnedColumnListLocal.push_back(srcpLocal);
plan->returnedCols(returnedColumnListLocal);
plan->txnID(csep.txnID());
plan->verID(csep.verID());
plan->sessionID(csep.sessionID());
plan->columnMapNonStatic(colMapLocal);
plan->returnedCols(returnedColumnListLocal);
unionVec.push_back(plan);
// std::cout << plan->toString() << std::endl;
}
csep.unionVec(unionVec);
std::cout << csep.toString() << std::endl;
}

3
dbcon/joblist/subquerytransformer.cpp
View File

@ -79,9 +79,6 @@ SubQueryTransformer::~SubQueryTransformer()
SJSTEP& SubQueryTransformer::makeSubQueryStep(execplan::CalpontSelectExecutionPlan* csep,
bool subInFromClause)
{
if (fOutJobInfo->trace)
cout << (*csep) << endl;
// Setup job info, job list and error status relation.
fSubJobInfo = new JobInfo(fOutJobInfo->rm);
fSubJobInfo->sessionId = fOutJobInfo->sessionId;

124
dbcon/mysql/ha_mcs_execplan.cpp
View File

@ -6598,8 +6598,6 @@ void gp_walk(const Item* item, void* arg)
gwip->hasSubSelect = true;
gwip->subQuery = existsSub;
gwip->ptWorkStack.push(existsSub->transform());
// MCOL-2178 isUnion member only assigned, never used
// MIGR::infinidb_vtable.isUnion = true; // only temp. bypass the 2nd phase.
// recover original
gwip->subQuery = orig;
gwip->lastSub = existsSub;
@ -6985,8 +6983,6 @@ int processFrom(bool& isUnion, SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP&
gwi.tbList.push_back(tn);
CalpontSystemCatalog::TableAliasName tan = make_aliastable("", alias, alias);
gwi.tableMap[tan] = make_pair(0, table_ptr);
// MCOL-2178 isUnion member only assigned, never used
// MIGR::infinidb_vtable.isUnion = true; //by-pass the 2nd pass of rnd_init
}
else if (table_ptr->view)
{
@ -7086,8 +7082,6 @@ int processFrom(bool& isUnion, SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP&
if (!isUnion && (!isSelectHandlerTop || isSelectLexUnit) && select_lex.master_unit()->is_unit_op())
{
// MCOL-2178 isUnion member only assigned, never used
// MIGR::infinidb_vtable.isUnion = true;
CalpontSelectExecutionPlan::SelectList unionVec;
SELECT_LEX* select_cursor = select_lex.master_unit()->first_select();
unionSel = true;
@ -7242,7 +7236,8 @@ int processFrom(bool& isUnion, SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP&
* RETURNS
* error id as an int
***********************************************************/
int processHaving(SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP& csep, std::unique_ptr<ParseTree>& havingFilter)
int processHaving(SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP& csep,
std::unique_ptr<ParseTree>& havingFilter)
{
gwi.havingDespiteSelect = true;
clearStacks(gwi, false, true);
@ -8086,8 +8081,6 @@ int processLimitAndOffset(SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP& csep
return 0;
}
/*@brief Process SELECT part of a query or sub-query */
/***********************************************************
* DESCRIPTION:
@ -8095,8 +8088,7 @@ int processLimitAndOffset(SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP& csep
* RETURNS
* error id as an int
***********************************************************/
int processSelect(SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP& csep,
vector<Item_field*>& funcFieldVec,
int processSelect(SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP& csep, vector<Item_field*>& funcFieldVec,
CalpontSelectExecutionPlan::SelectList& selectSubList)
{
gwi.select_lex = &select_lex;
@ -8158,7 +8150,6 @@ int processLimitAndOffset(SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP& csep
scalar->returnedColPos(gwi.additionalRetCols.size());
}
while ((item = it++))
{
string itemAlias = (item->name.length ? item->name.str : "<NULL>");
@ -8419,8 +8410,6 @@ int processLimitAndOffset(SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP& csep
break;
}
// MCOL-2178 This switch doesn't handl
// ROW_
default:
{
IDEBUG(cerr << "Warning unsupported cmp_type() in projection" << endl);
@ -8602,12 +8591,8 @@ int processLimitAndOffset(SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP& csep
* RETURNS
* error id as an int
***********************************************************/
int processOrderBy(SELECT_LEX& select_lex,
gp_walk_info& gwi,
SCSEP& csep,
boost::shared_ptr<CalpontSystemCatalog>& csc,
SRCP& minSc,
const bool isUnion,
int processOrderBy(SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP& csep,
boost::shared_ptr<CalpontSystemCatalog>& csc, SRCP& minSc, const bool isUnion,
const bool unionSel)
{
SQL_I_List<ORDER> order_list = select_lex.order_list;
@ -9069,8 +9054,7 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, SCSEP& csep, bool i
gwi.select_lex = originalSelectLex;
// append additionalRetCols to returnedCols
gwi.returnedCols.insert(gwi.returnedCols.begin(), gwi.additionalRetCols.begin(),
gwi.additionalRetCols.end(
));
gwi.additionalRetCols.end());
csep->groupByCols(gwi.groupByCols);
csep->withRollup(withRollup);
@ -9222,6 +9206,88 @@ int cs_get_derived_plan(ha_columnstore_derived_handler* handler, THD* /*thd*/, S
return 0;
}
bool matchParallelCES(CalpontSelectExecutionPlan& csep)
{
std::cout << csep.toString() << std::endl;
auto tables = csep.tableList();
// This is leaf and there are no other tables at this level.
for (auto& table : tables)
{
if (!table.isColumnstore())
{
return true;
}
}
return false;
}
// CalpontSelectExecutionPlan tableIntoUnion(CalpontSelectExecutionPlan& table, CalpontSelectExecutionPlan& csep)
// {
// auto* unionCSEP = {new CalpontSelectExecutionPlan()};
// CalpontSelectExecutionPlan::ReturnedColumnList returnedColumnList;
// CalpontSelectExecutionPlan::ColumnMap colMap;
// unionSCEP.unionVec({csep});
// return unionSCEP;
// }
void applyParallelCES(CalpontSelectExecutionPlan& csep)
{
auto tables = csep.tableList();
for (auto it = tables.begin(); it != tables.end(); ++it)
{
if (!it->isColumnstore())
{
// auto unionSCEP = tableIntoUnion(*it, csep);
// tables.erase(it);
// csep.unionVec().push_back(unionSCEP);
}
}
}
struct Rule
{
Rule(std::string&& name, bool (*match)(CalpontSelectExecutionPlan&),
void (*apply)(CalpontSelectExecutionPlan&))
: name(name), match(match), apply(apply) {};
std::string name;
bool (*match)(CalpontSelectExecutionPlan&);
void (*apply)(CalpontSelectExecutionPlan&);
bool walk(CalpontSelectExecutionPlan& csep)
{
for (auto& table : csep.derivedTableList())
{
auto csepLocal = *dynamic_cast<execplan::CalpontSelectExecutionPlan*>(table.get());
if (match(csepLocal))
{
apply(csepLocal);
return true;
}
}
for (auto& unionUnit : csep.unionVec())
{
auto unionUnitLocal = *dynamic_cast<execplan::CalpontSelectExecutionPlan*>(unionUnit.get());
if (match(unionUnitLocal))
{
apply(unionUnitLocal);
return true;
}
}
if (match(csep))
{
apply(csep);
return true;
}
return false;
}
};
int cs_get_select_plan(ha_columnstore_select_handler* handler, THD* /*thd*/, SCSEP& csep, gp_walk_info& gwi,
bool isSelectLexUnit)
{
@ -9252,6 +9318,20 @@ int cs_get_select_plan(ha_columnstore_select_handler* handler, THD* /*thd*/, SCS
// Derived table projection and filter optimization.
derivedTableOptimization(&gwi, csep);
// static const Rule rules[] = {
// {"paralliseCES", matchparalliseCES, applyParalliseCES},
// /* add more here */
// };
// auto matchParallelCES = [](CalpontSelectExecutionPlan&){return false;};
// auto applyParallelCES = [](CalpontSelectExecutionPlan&){ return;};
Rule parallelCES{"parallelCES", matchParallelCES, applyParallelCES};
{
parallelCES.walk(*csep);
}
return 0;
}