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

Add a limit (as runtime value) for long in queries

Browse Source
This commit is contained in:
mariadb-AndreyPiskunov
2023-07-21 22:14:50 +03:00
parent 6ff121a91c
commit 05547f2342
8 changed files with 132 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
dbcon/execplan/rewrites.cpp
View File

@ -9,7 +9,9 @@
*/
#include "rewrites.h"
#include <cstdint>
#include <typeinfo>
#include "constantfilter.h"
#include "objectreader.h"
#include "installdir.h"
#include "parsetree.h"
@ -482,4 +484,19 @@ bool NodeSemanticComparator::operator()(execplan::ParseTree* left, execplan::Par
return left->data()->data() < right->data()->data();
}
bool checkFiltersLimit(execplan::ParseTree* tree, uint64_t limit)
{
uint64_t maxLimit = 0;
auto walker = [](const execplan::ParseTree* node, void* maxLimit){
auto maybe_cf = dynamic_cast<ConstantFilter*>(node->data());
if (maybe_cf != nullptr && (maybe_cf->op()->op() == OpType::OP_OR || maybe_cf->op()->op() == OpType::OP_IN))
{
*((uint64_t*)maxLimit) = std::max(maybe_cf->filterList().size(), *((uint64_t*)maxLimit));
}
};
tree->walk(walker, &maxLimit);
return maxLimit <= limit;
}
} // namespace execplan

2
dbcon/execplan/rewrites.h
View File

@ -36,5 +36,5 @@ struct NodeSemanticComparator
// Walk the tree and find out common conjuctions
template<bool stableSort = false>
execplan::ParseTree* extractCommonLeafConjunctionsToRoot(execplan::ParseTree* tree);
bool checkFiltersLimit(execplan::ParseTree* tree, uint64_t limit);
} // namespace execplan

69
dbcon/mysql/ha_mcs_execplan.cpp
View File

@ -32,6 +32,7 @@
#include <vector>
#include <map>
#include <limits>
#include "messagelog.h"
#include <string.h>
@ -56,6 +57,7 @@ using namespace logging;
#include "ha_subquery.h"
#include "ha_mcs_pushdown.h"
#include "ha_tzinfo.h"
#include "ha_mcs_logging.h"
using namespace cal_impl_if;
#include "aggregatecolumn.h"
@ -1881,7 +1883,8 @@ bool buildPredicateItem(Item_func* ifp, gp_walk_info* gwip)
}
}
if (get_fe_conn_info_ptr() == NULL) {
if (get_fe_conn_info_ptr() == NULL)
{
set_fe_conn_info_ptr((void*)new cal_connection_info());
thd_set_ha_data(current_thd, mcs_hton, get_fe_conn_info_ptr());
}
@ -2078,7 +2081,8 @@ bool buildPredicateItem(Item_func* ifp, gp_walk_info* gwip)
}
if (udf->result_type() == STRING_RESULT)
gwip->rcWorkStack.push(new ConstantColumn(buf.ptr())); // XXX: constantcolumn from string = can it be NULL?
gwip->rcWorkStack.push(
new ConstantColumn(buf.ptr())); // XXX: constantcolumn from string = can it be NULL?
else
{
gwip->rcWorkStack.push(new ConstantColumn(buf.ptr(), ConstantColumn::NUM));
@ -2796,7 +2800,8 @@ void setError(THD* thd, uint32_t errcode, string errmsg)
thd->raise_error_printf(errcode, errmsg.c_str());
// reset expressionID
if (get_fe_conn_info_ptr() == NULL) {
if (get_fe_conn_info_ptr() == NULL)
{
set_fe_conn_info_ptr((void*)new cal_connection_info());
thd_set_ha_data(current_thd, mcs_hton, get_fe_conn_info_ptr());
}
@ -3606,7 +3611,8 @@ ReturnedColumn* buildBooleanConstantColumn(Item* item, gp_walk_info& gwi, bool&
ArithmeticColumn* buildArithmeticColumn(Item_func* item, gp_walk_info& gwi, bool& nonSupport)
{
if (get_fe_conn_info_ptr() == NULL) {
if (get_fe_conn_info_ptr() == NULL)
{
set_fe_conn_info_ptr((void*)new cal_connection_info());
thd_set_ha_data(current_thd, mcs_hton, get_fe_conn_info_ptr());
}
@ -3867,7 +3873,8 @@ ArithmeticColumn* buildArithmeticColumn(Item_func* item, gp_walk_info& gwi, bool
ReturnedColumn* buildFunctionColumn(Item_func* ifp, gp_walk_info& gwi, bool& nonSupport, bool selectBetweenIn)
{
if (get_fe_conn_info_ptr() == NULL) {
if (get_fe_conn_info_ptr() == NULL)
{
set_fe_conn_info_ptr((void*)new cal_connection_info());
thd_set_ha_data(current_thd, mcs_hton, get_fe_conn_info_ptr());
}
@ -3960,7 +3967,8 @@ ReturnedColumn* buildFunctionColumn(Item_func* ifp, gp_walk_info& gwi, bool& non
nonSupport = true;
gwi.fatalParseError = true;
Message::Args args;
string info = funcName + " with argument count > " + std::to_string(std::numeric_limits<uint16_t>::max());
string info =
funcName + " with argument count > " + std::to_string(std::numeric_limits<uint16_t>::max());
args.add(info);
gwi.parseErrorText = IDBErrorInfo::instance()->errorMsg(ERR_NON_SUPPORTED_FUNCTION, args);
return NULL;
@ -4044,7 +4052,6 @@ ReturnedColumn* buildFunctionColumn(Item_func* ifp, gp_walk_info& gwi, bool& non
// @todo. merge this logic to buildParseTree().
if ((funcName == "if" && i == 0) || funcName == "xor")
{
// make sure the rcWorkStack is cleaned.
gwi.clauseType = WHERE;
sptp.reset(buildParseTree(ifp->arguments()[i], gwi, nonSupport));
@ -4445,7 +4452,8 @@ ReturnedColumn* buildFunctionColumn(Item_func* ifp, gp_walk_info& gwi, bool& non
FunctionColumn* buildCaseFunction(Item_func* item, gp_walk_info& gwi, bool& nonSupport)
{
if (get_fe_conn_info_ptr() == NULL) {
if (get_fe_conn_info_ptr() == NULL)
{
set_fe_conn_info_ptr((void*)new cal_connection_info());
thd_set_ha_data(current_thd, mcs_hton, get_fe_conn_info_ptr());
}
@ -4872,7 +4880,8 @@ ReturnedColumn* buildAggregateColumn(Item* item, gp_walk_info& gwi)
vector<SRCP> orderCols;
ConstArgParam constArgParam;
if (get_fe_conn_info_ptr() == NULL) {
if (get_fe_conn_info_ptr() == NULL)
{
set_fe_conn_info_ptr((void*)new cal_connection_info());
thd_set_ha_data(current_thd, mcs_hton, get_fe_conn_info_ptr());
}
@ -5533,8 +5542,6 @@ because it has multiple arguments.";
return ac;
}
void addIntervalArgs(gp_walk_info* gwip, Item_func* ifp, FunctionParm& functionParms)
{
string funcName = ifp->func_name();
@ -5700,7 +5707,7 @@ void gp_walk(const Item* item, void* arg)
{
case INT_RESULT:
{
Item* non_const_item = const_cast<Item*>(item);
Item* non_const_item = const_cast<Item*>(item);
gwip->rcWorkStack.push(buildReturnedColumn(non_const_item, *gwip, gwip->fatalParseError));
break;
}
@ -5902,7 +5909,10 @@ void gp_walk(const Item* item, void* arg)
// bug 3137. If filter constant like 1=0, put it to ptWorkStack
// MariaDB bug 750. Breaks if compare is an argument to a function.
// if ((int32_t)gwip->rcWorkStack.size() <= (gwip->rcBookMarkStack.empty() ? 0
// if ((int32_t)gwip->rcWorkStack.size() <=
//(gwip->rcBookMarkStack.empty()
//?
// 0
//: gwip->rcBookMarkStack.top())
// && isPredicateFunction(ifp, gwip))
if (isPredicateFunction(ifp, gwip))
@ -6178,12 +6188,13 @@ void gp_walk(const Item* item, void* arg)
if (operand)
{
gwip->rcWorkStack.push(operand);
if (i == 0 && gwip->scsp == NULL) // first item is the WHEN LHS
if (i == 0 && gwip->scsp == NULL) // first item is the WHEN LHS
{
SimpleColumn* sc = dynamic_cast<SimpleColumn*>(operand);
if (sc)
{
gwip->scsp.reset(sc->clone()); // We need to clone else sc gets double deleted. This code is rarely executed so the cost is acceptable.
gwip->scsp.reset(sc->clone()); // We need to clone else sc gets double deleted. This code is
// rarely executed so the cost is acceptable.
}
}
}
@ -7086,6 +7097,18 @@ int processWhere(SELECT_LEX& select_lex, gp_walk_info& gwi, SCSEP& csep, const s
filters = extractCommonLeafConjunctionsToRoot(filters);
}
uint64_t limit = get_max_in_limit_query_length(gwi.thd);
if (filters && !checkFiltersLimit(filters, limit))
{
gwi.fatalParseError = true;
setError(gwi.thd, ER_CHECK_NOT_IMPLEMENTED,
"long in clauses. Query exceeds max_in_limit_query_length threshold: consider changing the "
"value via SET @var_name := value;",
gwi);
return ER_CHECK_NOT_IMPLEMENTED;
}
// Append outer join filters at the end of inner join filters.
// JLF_ExecPlanToJobList::walkTree processes ParseTree::left
// before ParseTree::right which is what we intend to do in the
@ -7361,8 +7384,7 @@ void buildInToExistsFilter(gp_walk_info& gwi, SELECT_LEX& select_lex)
* error id as an int
***********************************************************/
int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, SCSEP& csep, bool isUnion,
bool isSelectHandlerTop, bool isSelectLexUnit,
const std::vector<COND*>& condStack)
bool isSelectHandlerTop, bool isSelectLexUnit, const std::vector<COND*>& condStack)
{
#ifdef DEBUG_WALK_COND
cerr << "getSelectPlan()" << endl;
@ -7390,8 +7412,7 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, SCSEP& csep, bool i
CalpontSelectExecutionPlan::SelectList derivedTbList;
// @bug 1796. Remember table order on the FROM list.
gwi.clauseType = FROM;
if ((rc = processFrom(isUnion, select_lex, gwi, csep, isSelectHandlerTop,
isSelectLexUnit)))
if ((rc = processFrom(isUnion, select_lex, gwi, csep, isSelectHandlerTop, isSelectLexUnit)))
{
return rc;
}
@ -7855,7 +7876,8 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, SCSEP& csep, bool i
gwi.returnedCols[i]->hasAggregate(true);
}
gwi.returnedCols[i]->resultType(CalpontSystemCatalog::ColType::convertUnionColType(coltypes, unionedTypeRc));
gwi.returnedCols[i]->resultType(
CalpontSystemCatalog::ColType::convertUnionColType(coltypes, unionedTypeRc));
if (unionedTypeRc != 0)
{
@ -9089,14 +9111,14 @@ int getGroupPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, SCSEP& csep, cal_gro
{
// MCOL-1052 The condition could be useless.
// MariaDB bug 624 - without the fix_fields call, delete with join may error with "No query step".
//#if MYSQL_VERSION_ID < 50172
// #if MYSQL_VERSION_ID < 50172
//@bug 3039. fix fields for constants
if (!icp->fixed())
{
icp->fix_fields(gwi.thd, (Item**)&icp);
}
//#endif
// #endif
gwi.fatalParseError = false;
#ifdef DEBUG_WALK_COND
cerr << "------------------ WHERE -----------------------" << endl;
@ -9690,7 +9712,8 @@ int getGroupPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, SCSEP& csep, cal_gro
gwi.returnedCols[i]->hasAggregate(true);
}
gwi.returnedCols[i]->resultType(CalpontSystemCatalog::ColType::convertUnionColType(coltypes, unionedTypeRc));
gwi.returnedCols[i]->resultType(
CalpontSystemCatalog::ColType::convertUnionColType(coltypes, unionedTypeRc));
if (unionedTypeRc != 0)
{

45
dbcon/mysql/ha_mcs_sysvars.cpp
View File

@ -39,8 +39,8 @@ static MYSQL_THDVAR_ENUM(compression_type, PLUGIN_VAR_RQCMDARG,
"SNAPPY segment files are Snappy compressed (default);"
#ifdef HAVE_LZ4
"LZ4 segment files are LZ4 compressed;",
# else
,
#else
,
#endif
NULL, // check
NULL, // update
@ -201,22 +201,27 @@ static MYSQL_THDVAR_ULONGLONG(cache_flush_threshold, PLUGIN_VAR_RQCMDARG,
"Threshold on the number of rows in the cache to trigger a flush", NULL, NULL,
500000, 1, 1000000000, 1);
static MYSQL_THDVAR_STR(cmapi_host, PLUGIN_VAR_NOCMDOPT|PLUGIN_VAR_MEMALLOC, "CMAPI host", NULL, NULL,
static MYSQL_THDVAR_STR(cmapi_host, PLUGIN_VAR_NOCMDOPT | PLUGIN_VAR_MEMALLOC, "CMAPI host", NULL, NULL,
"https://localhost");
static MYSQL_THDVAR_STR(cmapi_version, PLUGIN_VAR_NOCMDOPT|PLUGIN_VAR_MEMALLOC, "CMAPI version", NULL, NULL,
static MYSQL_THDVAR_STR(cmapi_version, PLUGIN_VAR_NOCMDOPT | PLUGIN_VAR_MEMALLOC, "CMAPI version", NULL, NULL,
"0.4.0");
static MYSQL_THDVAR_STR(cmapi_key, PLUGIN_VAR_NOCMDOPT|PLUGIN_VAR_MEMALLOC, "CMAPI key", NULL, NULL,
"");
static MYSQL_THDVAR_STR(pron, PLUGIN_VAR_NOCMDOPT|PLUGIN_VAR_MEMALLOC, "Debug options json dictionary", NULL, NULL, "");
static MYSQL_THDVAR_STR(cmapi_key, PLUGIN_VAR_NOCMDOPT | PLUGIN_VAR_MEMALLOC, "CMAPI key", NULL, NULL, "");
static MYSQL_THDVAR_STR(pron, PLUGIN_VAR_NOCMDOPT | PLUGIN_VAR_MEMALLOC, "Debug options json dictionary",
NULL, NULL, "");
static MYSQL_THDVAR_ULONGLONG(cmapi_port, PLUGIN_VAR_NOCMDOPT, "CMAPI port", NULL,
NULL, 8640, 100, 65356, 1);
static MYSQL_THDVAR_ULONGLONG(cmapi_port, PLUGIN_VAR_NOCMDOPT, "CMAPI port", NULL, NULL, 8640, 100, 65356, 1);
static MYSQL_THDVAR_STR(s3_key, PLUGIN_VAR_NOCMDOPT|PLUGIN_VAR_MEMALLOC, "S3 Authentication Key ", NULL, NULL, "");
static MYSQL_THDVAR_STR(s3_secret, PLUGIN_VAR_NOCMDOPT|PLUGIN_VAR_MEMALLOC, "S3 Authentication Secret", NULL, NULL, "");
static MYSQL_THDVAR_STR(s3_region, PLUGIN_VAR_NOCMDOPT|PLUGIN_VAR_MEMALLOC, "S3 region", NULL, NULL, "");
static MYSQL_THDVAR_STR(s3_key, PLUGIN_VAR_NOCMDOPT | PLUGIN_VAR_MEMALLOC, "S3 Authentication Key ", NULL,
NULL, "");
static MYSQL_THDVAR_STR(s3_secret, PLUGIN_VAR_NOCMDOPT | PLUGIN_VAR_MEMALLOC, "S3 Authentication Secret",
NULL, NULL, "");
static MYSQL_THDVAR_STR(s3_region, PLUGIN_VAR_NOCMDOPT | PLUGIN_VAR_MEMALLOC, "S3 region", NULL, NULL, "");
static MYSQL_THDVAR_ULONG(max_in_limit_query_length, PLUGIN_VAR_RQCMDARG,
"The maximum length of the entries in the IN uery clause.", NULL, NULL, 6000, 1,
~0U, 1);
st_mysql_sys_var* mcs_system_variables[] = {MYSQL_SYSVAR(compression_type),
MYSQL_SYSVAR(fe_conn_info_ptr),
@ -258,6 +263,7 @@ st_mysql_sys_var* mcs_system_variables[] = {MYSQL_SYSVAR(compression_type),
MYSQL_SYSVAR(s3_secret),
MYSQL_SYSVAR(s3_region),
MYSQL_SYSVAR(pron),
MYSQL_SYSVAR(max_in_limit_query_length),
NULL};
st_mysql_show_var mcs_status_variables[] = {{"columnstore_version", (char*)&cs_version, SHOW_CHAR},
@ -271,8 +277,10 @@ void* get_fe_conn_info_ptr(THD* thd)
void set_fe_conn_info_ptr(void* ptr, THD* thd)
{
if (thd == NULL) thd = current_thd;
if (thd == NULL) return;
if (thd == NULL)
thd = current_thd;
if (thd == NULL)
return;
THDVAR(thd, fe_conn_info_ptr) = (uint64_t)(ptr);
}
@ -651,3 +659,12 @@ void set_s3_region(THD* thd, char* value)
{
THDVAR(thd, s3_region) = value;
}
ulong get_max_in_limit_query_length(THD* thd)
{
return (thd == NULL) ? 0 : THDVAR(thd, max_in_limit_query_length);
}
void set_max_in_limit_query_length(THD* thd, ulong value)
{
THDVAR(thd, max_in_limit_query_length) = value;
}

3
dbcon/mysql/ha_mcs_sysvars.h
View File

@ -176,3 +176,6 @@ void set_s3_region(THD* thd, char* value);
const char* get_pron(THD* thd);
void set_pron(THD* thd, char* value);
ulong get_max_in_limit_query_length(THD* thd);
void set_max_in_limit_query_length(THD* thd, ulong value);

20
mysql-test/columnstore/bugfixes/mcol-5468.result
View File

File diff suppressed because one or more lines are too long

22
mysql-test/columnstore/bugfixes/mcol-5468.test
View File

File diff suppressed because one or more lines are too long

14
oam/etc/Columnstore.xml
View File

@ -91,7 +91,7 @@
<ModuleSwapCriticalThreshold2>90</ModuleSwapCriticalThreshold2>
<ModuleSwapMajorThreshold2>80</ModuleSwapMajorThreshold2>
<ModuleSwapMinorThreshold2>70</ModuleSwapMinorThreshold2>
<ModuleDiskMonitorFileSystem1-2>/</ModuleDiskMonitorFileSystem1-2>
<ModuleDiskMonitorFileSystem1-2>/</ModuleDiskMonitorFileSystem1-2>
<ModuleDBRootCount1-2>unassigned</ModuleDBRootCount1-2>
<ModuleDBRootID1-1-2>unassigned</ModuleDBRootID1-1-2>
<ModuleType3>pm</ModuleType3>
@ -195,17 +195,17 @@
<JobList>
<FlushInterval>16K</FlushInterval>
<FifoSize>16</FifoSize>
<RequestSize>1</RequestSize> <!-- Number of extents per request, should be
<RequestSize>1</RequestSize> <!-- Number of extents per request, should be
less than MaxOutstandingRequests. Otherwise, default value 1 is used. -->
<!-- ProcessorThreadsPerScan is the number of jobs issued to process
each extent. The default is 16. MaxOutstandingRequests is the size of
the window of work in terms of extents. A value of 20 means there
each extent. The default is 16. MaxOutstandingRequests is the size of
the window of work in terms of extents. A value of 20 means there
is 20 extents worth of work for the PMs to process at any given time.
ProcessorThreadsPerScan * MaxOutstandingRequests should be at least
as many threads are available across all PMs. -->
<!-- <ProcessorThreadsPerScan>16</ProcessorThreadsPerScan> -->
<!-- MaxOutstandingRequests is going to default to the num of cores available
across all performance modules * 4 divided by the ProcessorThreadsPerScan,
<!-- <ProcessorThreadsPerScan>16</ProcessorThreadsPerScan> -->
<!-- MaxOutstandingRequests is going to default to the num of cores available
across all performance modules * 4 divided by the ProcessorThreadsPerScan,
but will be lower bounded by 20 -->
<!-- <MaxOutstandingRequests>20</MaxOutstandingRequests> -->
<ThreadPoolSize>100</ThreadPoolSize>