database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-07-29 08:21:15 +03:00
Code Activity

Merge pull request #1986 from tntnatbry/MCOL-1482

Browse Source 
MCOL-1482 An UPDATE operation on a non-ColumnStore table involving a cross-engine join
This commit is contained in:
Roman Nozdrin
2021-07-01 14:25:32 +03:00
committed by GitHub
parent ceb73cddcd 49255f5cbd
commit 325bb6c9e0
6 changed files with 180 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
dbcon/mysql/ha_mcs_execplan.cpp
View File

@ -6326,6 +6326,48 @@ bool isMCSTable(TABLE* table_ptr)
return false;
}
bool isForeignTableUpdate(THD* thd)
{
LEX* lex = thd->lex;
if (lex->sql_command != SQLCOM_UPDATE_MULTI)
return false;
Item_field* item;
List_iterator_fast<Item> field_it(lex->first_select_lex()->item_list);
while ((item = (Item_field*) field_it++))
{
if (item->field && item->field->table && !isMCSTable(item->field->table))
return true;
}
return false;
}
// This function is different from isForeignTableUpdate()
// above as it only checks if any of the tables involved
// in the multi-table update statement is a foreign table,
// irrespective of whether the update is performed on the
// foreign table or not, as in isForeignTableUpdate().
bool isUpdateHasForeignTable(THD* thd)
{
LEX* lex = thd->lex;
if (lex->sql_command != SQLCOM_UPDATE_MULTI)
return false;
TABLE_LIST* table_ptr = lex->first_select_lex()->get_table_list();
for (; table_ptr; table_ptr = table_ptr->next_local)
{
if (table_ptr->table && !isMCSTable(table_ptr->table))
return true;
}
return false;
}
/*@brief set some runtime params to run the query */
/***********************************************************
* DESCRIPTION:
@ -6577,10 +6619,7 @@ int processWhere(SELECT_LEX &select_lex,
else if (select_lex.where)
icp = select_lex.where;
}
else if (!join && ( ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE ) ||
((gwi.thd->lex)->sql_command == SQLCOM_DELETE ) ||
((gwi.thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) ||
((gwi.thd->lex)->sql_command == SQLCOM_DELETE_MULTI )))
else if (!join && isUpdateOrDeleteStatement(gwi.thd->lex->sql_command))
{
isUpdateDelete = true;
}
@ -7393,7 +7432,7 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex,
}
//@Bug 3030 Add error check for dml statement
if ( ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE ) || ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE_MULTI ) )
if (isUpdateOrDeleteStatement(gwi.thd->lex->sql_command))
{
if ( after_size - before_size != 0 )
{
@ -7427,7 +7466,7 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex,
case REAL_RESULT:
case TIME_RESULT:
{
if ( ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE ) || ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE_MULTI ))
if (isUpdateOrDeleteStatement(gwi.thd->lex->sql_command))
{ }
else
{
@ -7459,7 +7498,7 @@ int getSelectPlan(gp_walk_info& gwi, SELECT_LEX& select_lex,
case Item::NULL_ITEM:
{
if ( ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE ) || ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE_MULTI ) )
if (isUpdateOrDeleteStatement(gwi.thd->lex->sql_command))
{ }
else
{
@ -9259,7 +9298,7 @@ int getGroupPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, SCSEP& csep, cal_gro
}
//@Bug 3030 Add error check for dml statement
if ( ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE ) || ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE_MULTI ) )
if (isUpdateOrDeleteStatement(gwi.thd->lex->sql_command))
{
if ( after_size - before_size != 0 )
{
@ -9290,7 +9329,7 @@ int getGroupPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, SCSEP& csep, cal_gro
case REAL_RESULT:
case TIME_RESULT:
{
if ( ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE ) || ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE_MULTI ))
if (isUpdateOrDeleteStatement(gwi.thd->lex->sql_command))
{ }
else
{
@ -9322,7 +9361,7 @@ int getGroupPlan(gp_walk_info& gwi, SELECT_LEX& select_lex, SCSEP& csep, cal_gro
case Item::NULL_ITEM:
{
if ( ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE ) || ((gwi.thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) || ((gwi.thd->lex)->sql_command == SQLCOM_DELETE_MULTI ) )
if (isUpdateOrDeleteStatement(gwi.thd->lex->sql_command))
{ }
else
{

79
dbcon/mysql/ha_mcs_impl.cpp
View File

@ -854,7 +854,7 @@ uint32_t doUpdateDelete(THD* thd, gp_walk_info& gwi, const std::vector<COND*>& c
{
Message::Args args;
if (((thd->lex)->sql_command == SQLCOM_UPDATE) || ((thd->lex)->sql_command == SQLCOM_UPDATE_MULTI))
if (isUpdateStatement(thd->lex->sql_command))
args.add("Update");
#if 0
else if (thd->rgi_slave && thd->rgi_slave->m_table_map.count() != 0)
@ -909,7 +909,7 @@ uint32_t doUpdateDelete(THD* thd, gp_walk_info& gwi, const std::vector<COND*>& c
updateCP->isDML(true);
//@Bug 2753. the memory already freed by destructor of UpdateSqlStatement
if (((thd->lex)->sql_command == SQLCOM_UPDATE) || ((thd->lex)->sql_command == SQLCOM_UPDATE_MULTI))
if (isUpdateStatement(thd->lex->sql_command))
{
ColumnAssignment* columnAssignmentPtr;
Item_field* item;
@ -1202,8 +1202,7 @@ uint32_t doUpdateDelete(THD* thd, gp_walk_info& gwi, const std::vector<COND*>& c
// Exit early if there is nothing to update
if (colAssignmentListPtr->empty() &&
(((thd->lex)->sql_command == SQLCOM_UPDATE) ||
((thd->lex)->sql_command == SQLCOM_UPDATE_MULTI)))
isUpdateStatement(thd->lex->sql_command))
{
ci->affectedRows = 0;
return 0;
@ -1216,7 +1215,7 @@ uint32_t doUpdateDelete(THD* thd, gp_walk_info& gwi, const std::vector<COND*>& c
CalpontSystemCatalog::TableName aTableName;
TABLE_LIST* first_table = 0;
if (( (thd->lex)->sql_command == SQLCOM_UPDATE ) || ( (thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) )
if (isUpdateStatement(thd->lex->sql_command))
{
aTableName.schema = schemaName;
aTableName.table = tableName;
@ -1238,10 +1237,10 @@ uint32_t doUpdateDelete(THD* thd, gp_walk_info& gwi, const std::vector<COND*>& c
IDEBUG( cout << "STMT: " << dmlStmt << " and sessionID " << thd->thread_id << endl );
VendorDMLStatement dmlStatement(dmlStmt, sessionID);
if (( (thd->lex)->sql_command == SQLCOM_UPDATE ) || ( (thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) )
dmlStatement.set_DMLStatementType( DML_UPDATE );
if (isUpdateStatement(thd->lex->sql_command))
dmlStatement.set_DMLStatementType(DML_UPDATE);
else
dmlStatement.set_DMLStatementType( DML_DELETE );
dmlStatement.set_DMLStatementType(DML_DELETE);
TableName* qualifiedTablName = new TableName();
@ -1249,7 +1248,7 @@ uint32_t doUpdateDelete(THD* thd, gp_walk_info& gwi, const std::vector<COND*>& c
//@Bug 2753. To make sure the momory is freed.
updateStmt.fColAssignmentListPtr = colAssignmentListPtr;
if (( (thd->lex)->sql_command == SQLCOM_UPDATE ) || ( (thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) )
if (isUpdateStatement(thd->lex->sql_command))
{
qualifiedTablName->fName = tableName;
qualifiedTablName->fSchema = schemaName;
@ -1343,7 +1342,7 @@ uint32_t doUpdateDelete(THD* thd, gp_walk_info& gwi, const std::vector<COND*>& c
List<Item> items;
SELECT_LEX select_lex;
if (( (thd->lex)->sql_command == SQLCOM_UPDATE ) || ( (thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) )
if (isUpdateStatement(thd->lex->sql_command))
{
items = (thd->lex->first_select_lex()->item_list);
thd->lex->first_select_lex()->item_list = thd->lex->value_list;
@ -1492,7 +1491,7 @@ uint32_t doUpdateDelete(THD* thd, gp_walk_info& gwi, const std::vector<COND*>& c
returnedCols.push_back((updateCP->columnMap()).begin()->second);
//@Bug 6123. get the correct returned columnlist
if (( (thd->lex)->sql_command == SQLCOM_DELETE ) || ( (thd->lex)->sql_command == SQLCOM_DELETE_MULTI ) )
if (isDeleteStatement(thd->lex->sql_command))
{
returnedCols.clear();
//choose the smallest column to project
@ -1543,7 +1542,7 @@ uint32_t doUpdateDelete(THD* thd, gp_walk_info& gwi, const std::vector<COND*>& c
updateCP->returnedCols( returnedCols );
if (( (thd->lex)->sql_command == SQLCOM_UPDATE ) || ( (thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) )
if (isUpdateStatement(thd->lex->sql_command))
{
const ParseTree* ptsub = updateCP->filters();
@ -1561,8 +1560,7 @@ uint32_t doUpdateDelete(THD* thd, gp_walk_info& gwi, const std::vector<COND*>& c
ptsub->walk(makeUpdateSemiJoin, &tbList[0] );
}
if (( (thd->lex)->sql_command == SQLCOM_UPDATE ) || ( (thd->lex)->sql_command == SQLCOM_UPDATE_MULTI ) )
thd->lex->first_select_lex()->item_list = items;
thd->lex->first_select_lex()->item_list = items;
}
}
@ -2039,10 +2037,7 @@ int ha_mcs::impl_rnd_init(TABLE* table, const std::vector<COND*>& condStack)
UPDATE innotab1 SET a=100 WHERE a NOT IN (SELECT a FROM cstab1 WHERE a=1);
*/
if (!isReadOnly() && // make sure the current table is being modified
(thd->lex->sql_command == SQLCOM_UPDATE ||
thd->lex->sql_command == SQLCOM_DELETE ||
thd->lex->sql_command == SQLCOM_DELETE_MULTI ||
thd->lex->sql_command == SQLCOM_UPDATE_MULTI))
isUpdateOrDeleteStatement(thd->lex->sql_command))
return doUpdateDelete(thd, gwi, condStack);
uint32_t sessionID = tid2sid(thd->thread_id);
@ -2372,8 +2367,10 @@ int ha_mcs_impl_rnd_next(uchar* buf, TABLE* table)
thd->lex->sql_command == SQLCOM_LOAD))
return HA_ERR_END_OF_FILE;
if (((thd->lex)->sql_command == SQLCOM_UPDATE) || ((thd->lex)->sql_command == SQLCOM_DELETE) ||
((thd->lex)->sql_command == SQLCOM_DELETE_MULTI) || ((thd->lex)->sql_command == SQLCOM_UPDATE_MULTI))
if ((thd->lex->sql_command == SQLCOM_UPDATE) ||
(thd->lex->sql_command == SQLCOM_DELETE) ||
(thd->lex->sql_command == SQLCOM_DELETE_MULTI) ||
((thd->lex->sql_command == SQLCOM_UPDATE_MULTI) && !isForeignTableUpdate(thd)))
return HA_ERR_END_OF_FILE;
// @bug 2547
@ -2461,22 +2458,15 @@ int ha_mcs_impl_rnd_end(TABLE* table, bool is_pushdown_hand)
thd->lex->sql_command == SQLCOM_LOAD))
return 0;
// MCOL-2178 isUnion member only assigned, never used
// if (is_pushdown_hand)
// {
// MIGR::infinidb_vtable.isUnion = false;
// }
if (get_fe_conn_info_ptr() != NULL)
ci = reinterpret_cast<cal_connection_info*>(get_fe_conn_info_ptr());
if ( (thd->lex)->sql_command == SQLCOM_ALTER_TABLE )
return rc;
if (((thd->lex)->sql_command == SQLCOM_UPDATE) ||
((thd->lex)->sql_command == SQLCOM_DELETE) ||
((thd->lex)->sql_command == SQLCOM_DELETE_MULTI) ||
((thd->lex)->sql_command == SQLCOM_UPDATE_MULTI))
if ((thd->lex->sql_command == SQLCOM_UPDATE) ||
(thd->lex->sql_command == SQLCOM_DELETE) ||
(thd->lex->sql_command == SQLCOM_DELETE_MULTI) ||
((thd->lex->sql_command == SQLCOM_UPDATE_MULTI) && !isForeignTableUpdate(thd)))
return rc;
if (!ci)
@ -3668,10 +3658,7 @@ COND* ha_mcs_impl_cond_push(COND* cond, TABLE* table, std::vector<COND*>& condSt
{
THD* thd = current_thd;
if (((thd->lex)->sql_command == SQLCOM_UPDATE) ||
((thd->lex)->sql_command == SQLCOM_UPDATE_MULTI) ||
((thd->lex)->sql_command == SQLCOM_DELETE) ||
((thd->lex)->sql_command == SQLCOM_DELETE_MULTI))
if (isUpdateOrDeleteStatement(thd->lex->sql_command))
{
condStack.push_back(cond);
return nullptr;
@ -3921,7 +3908,6 @@ int ha_mcs_impl_group_by_init(mcs_handler_info *handler_info, TABLE* table)
idbassert(ci != 0);
if (thd->killed == KILL_QUERY || thd->killed == KILL_QUERY_HARD)
{
force_close_fep_conn(thd, ci);
@ -4326,15 +4312,12 @@ int ha_mcs_impl_group_by_next(TABLE* table)
{
THD* thd = current_thd;
if (((thd->lex)->sql_command == SQLCOM_UPDATE) || ((thd->lex)->sql_command == SQLCOM_DELETE) ||
((thd->lex)->sql_command == SQLCOM_DELETE_MULTI) || ((thd->lex)->sql_command == SQLCOM_UPDATE_MULTI))
if ((thd->lex->sql_command == SQLCOM_UPDATE) ||
(thd->lex->sql_command == SQLCOM_DELETE) ||
(thd->lex->sql_command == SQLCOM_DELETE_MULTI) ||
(thd->lex->sql_command == SQLCOM_UPDATE_MULTI && !isForeignTableUpdate(thd)))
return HA_ERR_END_OF_FILE;
// @bug 2547
// MCOL-2178
// if (MIGR::infinidb_vtable.impossibleWhereOnUnion)
// return HA_ERR_END_OF_FILE;
if (get_fe_conn_info_ptr() == nullptr)
set_fe_conn_info_ptr((void*)new cal_connection_info());
@ -4426,9 +4409,6 @@ int ha_mcs_impl_group_by_end(TABLE* table)
thd->lex->sql_command == SQLCOM_LOAD))
return 0;
// MCOL-2178 isUnion member only assigned, never used
// MIGR::infinidb_vtable.isUnion = false;
if (get_fe_conn_info_ptr() != NULL)
ci = reinterpret_cast<cal_connection_info*>(get_fe_conn_info_ptr());
@ -4638,8 +4618,8 @@ int ha_mcs_impl_pushdown_init(mcs_handler_info* handler_info, TABLE* table)
return 0;
// MCOL-4023 We need to test this code path.
//Update and delete code
if ( ((thd->lex)->sql_command == SQLCOM_UPDATE) || ((thd->lex)->sql_command == SQLCOM_DELETE) || ((thd->lex)->sql_command == SQLCOM_DELETE_MULTI) || ((thd->lex)->sql_command == SQLCOM_UPDATE_MULTI))
// Update and delete code
if (isUpdateOrDeleteStatement(thd->lex->sql_command))
return doUpdateDelete(thd, gwi, std::vector<COND*>());
uint32_t sessionID = tid2sid(thd->thread_id);
@ -5059,8 +5039,7 @@ int ha_mcs_impl_select_next(uchar* buf, TABLE* table)
thd->lex->sql_command == SQLCOM_LOAD))
return HA_ERR_END_OF_FILE;
if (((thd->lex)->sql_command == SQLCOM_UPDATE) || ((thd->lex)->sql_command == SQLCOM_DELETE) ||
((thd->lex)->sql_command == SQLCOM_DELETE_MULTI) || ((thd->lex)->sql_command == SQLCOM_UPDATE_MULTI))
if (isUpdateOrDeleteStatement(thd->lex->sql_command))
return rc;
// @bug 2547

20
dbcon/mysql/ha_mcs_impl_if.h
View File

@ -366,6 +366,8 @@ void gp_walk(const Item* item, void* arg);
void parse_item (Item* item, std::vector<Item_field*>& field_vec, bool& hasNonSupportItem, uint16& parseInfo, gp_walk_info* gwip = NULL);
const std::string bestTableName(const Item_field* ifp);
bool isMCSTable(TABLE* table_ptr);
bool isForeignTableUpdate(THD* thd);
bool isUpdateHasForeignTable(THD* thd);
// execution plan util functions prototypes
execplan::ReturnedColumn* buildReturnedColumn(Item* item, gp_walk_info& gwi, bool& nonSupport, bool isRefItem = false);
@ -409,6 +411,24 @@ bool buildEqualityPredicate(execplan::ReturnedColumn* lhs,
const std::vector<Item*>& itemList,
bool isInSubs = false);
inline bool isUpdateStatement(const enum_sql_command& command)
{
return (command == SQLCOM_UPDATE) ||
(command == SQLCOM_UPDATE_MULTI);
}
inline bool isDeleteStatement(const enum_sql_command& command)
{
return (command == SQLCOM_DELETE) ||
(command == SQLCOM_DELETE_MULTI);
}
inline bool isUpdateOrDeleteStatement(const enum_sql_command& command)
{
return isUpdateStatement(command) ||
isDeleteStatement(command);
}
#ifdef DEBUG_WALK_COND
void debug_walk(const Item* item, void* arg);
#endif

15
dbcon/mysql/ha_mcs_pushdown.cpp
View File

@ -498,6 +498,13 @@ create_columnstore_derived_handler(THD* thd, TABLE_LIST *table_ptr)
if (thd->stmt_arena && thd->stmt_arena->is_stmt_execute())
return handler;
// MCOL-1482 Disable derived_handler if the multi-table update
// statement contains a non-columnstore table.
if (cal_impl_if::isUpdateHasForeignTable(thd))
{
return handler;
}
SELECT_LEX_UNIT *unit = table_ptr->derived;
SELECT_LEX *sl = unit->first_select();
@ -754,6 +761,14 @@ create_columnstore_select_handler(THD* thd, SELECT_LEX* select_lex)
return nullptr;
}
// MCOL-1482 Disable select_handler for a multi-table update
// with a non-columnstore table as the target table of the update
// operation.
if (cal_impl_if::isForeignTableUpdate(thd))
{
return nullptr;
}
// Flag to indicate if this is a prepared statement
bool isPS = thd->stmt_arena && thd->stmt_arena->is_stmt_execute();