revision-id: 5dc4ca6554c9bb4685b64c35f345c06d28c8d3f9 (mariadb-10.3.6-130-g5dc4ca6) parent(s): 171fbbb968ed52dc7e2bbd33a6f8f72bbc6f5e88 author: Igor Babaev committer: Igor Babaev timestamp: 2018-10-09 02:36:09 -0700 message: MDEV-17096 Pushdown of simple derived tables to storage engines Interface + Proof of Concept for federatedx. --- libmysqld/CMakeLists.txt | 2 +- sql/CMakeLists.txt | 2 +- sql/derived_handler.cc | 72 ++++++++++++++++++++ sql/derived_handler.h | 59 +++++++++++++++++ sql/handler.h | 10 +++ sql/sql_derived.cc | 78 +++++++++++++++++++++- sql/sql_explain.cc | 22 ++++-- sql/sql_explain.h | 1 + sql/sql_lex.cc | 6 +- sql/sql_select.cc | 8 ++- sql/sql_select.h | 18 ++++- sql/table.h | 7 ++ storage/federatedx/ha_federatedx.cc | 129 +++++++++++++++++++++++++++++++++++- storage/federatedx/ha_federatedx.h | 24 +++++++ 14 files changed, 424 insertions(+), 14 deletions(-) diff --git a/libmysqld/CMakeLists.txt b/libmysqld/CMakeLists.txt index 99b6208..dc3d190 100644 --- a/libmysqld/CMakeLists.txt +++ b/libmysqld/CMakeLists.txt @@ -77,7 +77,7 @@ SET(SQL_EMBEDDED_SOURCES emb_qcache.cc libmysqld.c lib_sql.cc ../sql/debug_sync.cc ../sql/opt_table_elimination.cc ../sql/sql_prepare.cc ../sql/sql_rename.cc ../sql/sql_repl.cc ../sql/sql_select.cc ../sql/sql_servers.cc - ../sql/group_by_handler.cc + ../sql/group_by_handler.cc ../sql/derived_handler.cc ../sql/sql_show.cc ../sql/sql_state.c ../sql/sql_statistics.cc ../sql/sql_string.cc ../sql/sql_tablespace.cc ../sql/sql_table.cc ../sql/sql_test.cc diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt index 708c36a..f76753a 100644 --- a/sql/CMakeLists.txt +++ b/sql/CMakeLists.txt @@ -96,7 +96,7 @@ SET (SQL_SOURCE sql_partition.cc sql_plugin.cc sql_prepare.cc sql_rename.cc debug_sync.cc sql_repl.cc sql_select.cc sql_show.cc sql_state.c - group_by_handler.cc + group_by_handler.cc derived_handler.cc sql_statistics.cc sql_string.cc lex_string.h sql_table.cc sql_test.cc sql_trigger.cc sql_udf.cc sql_union.cc sql_update.cc sql_view.cc strfunc.cc table.cc thr_malloc.cc diff --git a/sql/derived_handler.cc b/sql/derived_handler.cc new file mode 100644 index 0000000..dd017fd --- /dev/null +++ b/sql/derived_handler.cc @@ -0,0 +1,72 @@ +#include "mariadb.h" +#include "sql_priv.h" +#include "sql_select.h" +#include "derived_handler.h" + +void derived_handler::set_derived(TABLE_LIST *tbl) +{ + derived= tbl; + table= tbl->table; + unit= tbl->derived; + select= unit->first_select(); + tmp_table_param= select->next_select() ? + ((select_unit *)(unit->result))->get_tmp_table_param() : + &select->join->tmp_table_param; +} + +Pushdown_derived::~Pushdown_derived() +{ + delete handler; +} + +int Pushdown_derived::execute() +{ + int err; + THD *thd= handler->thd; + TABLE *table= handler->table; + TMP_TABLE_PARAM *tmp_table_param= handler->tmp_table_param; + + DBUG_ENTER("Pushdown_query::execute"); + + if ((err= handler->init_scan())) + goto error; + + while (!(err= handler->next_row())) + { + if (unlikely(thd->check_killed())) + { + handler->end_scan(); + DBUG_RETURN(-1); + } + + if ((err= table->file->ha_write_tmp_row(table->record[0]))) + { + bool is_duplicate; + if (likely(!table->file->is_fatal_error(err, HA_CHECK_DUP))) + continue; // Distinct elimination + + if (create_internal_tmp_table_from_heap(thd, table, + tmp_table_param->start_recinfo, + &tmp_table_param->recinfo, + err, 1, &is_duplicate)) + DBUG_RETURN(1); + if (is_duplicate) + continue; + } + } + + if (err != 0 && err != HA_ERR_END_OF_FILE) + goto error; + + if ((err= handler->end_scan())) + goto error_2; + + DBUG_RETURN(0); + +error: + handler->end_scan(); +error_2: + handler->print_error(err, MYF(0)); + DBUG_RETURN(-1); // Error not sent to client +} + diff --git a/sql/derived_handler.h b/sql/derived_handler.h new file mode 100644 index 0000000..a7b1294 --- /dev/null +++ b/sql/derived_handler.h @@ -0,0 +1,59 @@ +#ifndef DERIVED_HANDLER_INCLUDED +#define DERIVED_HANDLER_INCLUDED + +#include "mariadb.h" +#include "sql_priv.h" + +class TMP_TABLE_PARAM; + +class derived_handler +{ +public: + THD *thd; + handlerton *ht; + + TABLE_LIST *derived; + + /* + Temporary table where all results should be stored in record[0] + The table has a field for every item from the select list of + the specification of derived. + */ + TABLE *table; + + TMP_TABLE_PARAM *tmp_table_param; + + struct st_select_lex_unit *unit; + + struct st_select_lex *select; + + derived_handler(THD *thd_arg, handlerton *ht_arg) + : thd(thd_arg), ht(ht_arg), derived(0),table(0), tmp_table_param(0), + unit(0), select(0) {} + virtual ~derived_handler() {} + + /* + Functions to scan data. All these returns 0 if ok, error code in case + of error + */ + + /* Initialize the process of producing rows of the derived table */ + virtual int init_scan()= 0; + + /* + Put the next produced row of the derived in table->record[0] and return 0. + Return HA_ERR_END_OF_FILE if there are no more rows, return other error + number in case of fatal error. + */ + virtual int next_row()= 0; + + /* End prodicing rows */ + virtual int end_scan()=0; + + /* Report errors */ + virtual void print_error(int error, myf errflag)=0; + + void set_derived(TABLE_LIST *tbl); +}; + +#endif /* DERIVED_HANDLER_INCLUDED */ diff --git a/sql/handler.h b/sql/handler.h index 68a54cc..ce6dd35 100644 --- a/sql/handler.h +++ b/sql/handler.h @@ -1183,6 +1183,7 @@ struct handler_iterator { class handler; class group_by_handler; +class derived_handler; struct Query; typedef class st_select_lex SELECT_LEX; typedef struct st_order ORDER; @@ -1502,6 +1503,15 @@ struct handlerton */ group_by_handler *(*create_group_by)(THD *thd, Query *query); + /* + Create and return a derived_handler if the storage engine can execute + the derived table 'derived', otherwise return NULL. + In a general case 'derived' may contain tables not from the engine. + If the engine cannot handle or does not want to handle such pushed derived + the function create_group_by has to return NULL. + */ + derived_handler *(*create_derived)(THD *thd, TABLE_LIST *derived); + /********************************************************************* Table discovery API. It allows the server to "discover" tables that exist in the storage diff --git a/sql/sql_derived.cc b/sql/sql_derived.cc index d65969d..7e7ac02 100644 --- a/sql/sql_derived.cc +++ b/sql/sql_derived.cc @@ -27,6 +27,7 @@ #include "unireg.h" #include "sql_derived.h" #include "sql_select.h" +#include "derived_handler.h" #include "sql_base.h" #include "sql_view.h" // check_duplicate_names #include "sql_acl.h" // SELECT_ACL @@ -384,9 +385,16 @@ bool mysql_derived_merge(THD *thd, LEX *lex, TABLE_LIST *derived) DBUG_RETURN(FALSE); } - if (thd->lex->sql_command == SQLCOM_UPDATE_MULTI || - thd->lex->sql_command == SQLCOM_DELETE_MULTI) - thd->save_prep_leaf_list= TRUE; + if ((derived->dt_handler= derived->find_derived_handler(thd))) + { + derived->change_refs_to_fields(); + derived->set_materialized_derived(); + DBUG_RETURN(FALSE); + } + + if (thd->lex->sql_command == SQLCOM_UPDATE_MULTI || + thd->lex->sql_command == SQLCOM_DELETE_MULTI) + thd->save_prep_leaf_list= TRUE; arena= thd->activate_stmt_arena_if_needed(&backup); // For easier test @@ -904,6 +912,15 @@ bool mysql_derived_optimize(THD *thd, LEX *lex, TABLE_LIST *derived) DBUG_RETURN(FALSE); } + if (derived->is_materialized_derived() && !derived->dt_handler) + derived->dt_handler= derived->find_derived_handler(thd); + if (derived->dt_handler) + { + if (!(derived->pushdown_derived= + new (thd->mem_root) Pushdown_derived(derived, derived->dt_handler))) + DBUG_RETURN(1); + } + lex->current_select= first_select; if (unit->is_unit_op()) @@ -1108,6 +1125,17 @@ bool mysql_derived_fill(THD *thd, LEX *lex, TABLE_LIST *derived) select_unit *derived_result= derived->derived_result; SELECT_LEX *save_current_select= lex->current_select; + if (derived->pushdown_derived) + { + int res; + if (unit->executed) + DBUG_RETURN(FALSE); + res= derived->pushdown_derived->execute(); + unit->executed= true; + delete derived->pushdown_derived; + DBUG_RETURN(res); + } + if (unit->executed && !derived_is_recursive && (unit->uncacheable & UNCACHEABLE_DEPENDENT)) { @@ -1404,3 +1432,47 @@ bool pushdown_cond_for_derived(THD *thd, Item *cond, TABLE_LIST *derived) thd->lex->current_select= save_curr_select; DBUG_RETURN(false); } + + +derived_handler *TABLE_LIST::find_derived_handler(THD *thd) +{ + if (!derived || is_recursive_with_table()) + return 0; + for (SELECT_LEX *sl= derived->first_select(); sl; sl= sl->next_select()) + { + if (!(sl->join)) + continue; + for (TABLE_LIST *tbl= sl->join->tables_list; tbl; tbl= tbl->next_local) + { + if (!tbl->table) + continue; + handlerton *ht= tbl->table->file->partition_ht(); + if (!ht->create_derived) + continue; + derived_handler *dh= ht->create_derived(thd, this); + if (dh) + { + dh->set_derived(this); + return dh; + } + } + } + return 0; +} + + +TABLE_LIST *TABLE_LIST::get_first_table() +{ + for (SELECT_LEX *sl= derived->first_select(); sl; sl= sl->next_select()) + { + if (!(sl->join)) + continue; + for (TABLE_LIST *tbl= sl->join->tables_list; tbl; tbl= tbl->next_local) + { + if (!tbl->table) + continue; + return tbl; + } + } + return 0; +} diff --git a/sql/sql_explain.cc b/sql/sql_explain.cc index 1c45b05..3f1dc80 100644 --- a/sql/sql_explain.cc +++ b/sql/sql_explain.cc @@ -34,6 +34,8 @@ const char *unit_operation_text[4]= "UNIT RESULT","UNION RESULT","INTERSECT RESULT","EXCEPT RESULT" }; +const char *pushed_derived_text= "PUSHED DERIVED"; + static void write_item(Json_writer *writer, Item *item); static void append_item_to_str(String *out, Item *item); @@ -334,6 +336,9 @@ int print_explain_row(select_result_sink *result, List<Item> item_list; Item *item; + if (!select_type[0]) + return 0; + item_list.push_back(new (mem_root) Item_int(thd, (int32) select_number), mem_root); item_list.push_back(new (mem_root) Item_string_sys(thd, select_type), @@ -746,7 +751,15 @@ int Explain_select::print_explain(Explain_query *query, THD *thd= output->thd; MEM_ROOT *mem_root= thd->mem_root; - if (message) + if (select_type == pushed_derived_text) + { + print_explain_message_line(output, explain_flags, is_analyze, + select_id /*select number*/, + select_type, + NULL, /* rows */ + NULL); + } + else if (message) { List<Item> item_list; Item *item_null= new (mem_root) Item_null(thd); @@ -869,14 +882,15 @@ void Explain_select::print_explain_json(Explain_query *query, bool started_cache= print_explain_json_cache(writer, is_analyze); - if (message) + if (message || select_type == pushed_derived_text) { writer->add_member("query_block").start_object(); writer->add_member("select_id").add_ll(select_id); add_linkage(writer); writer->add_member("table").start_object(); - writer->add_member("message").add_str(message); + writer->add_member("message").add_str(select_type == pushed_derived_text ? + "Pushed derived" : message); writer->end_object(); print_explain_json_for_children(query, writer, is_analyze); @@ -1205,7 +1219,7 @@ int Explain_table_access::print_explain(select_result_sink *output, uint8 explai { THD *thd= output->thd; MEM_ROOT *mem_root= thd->mem_root; - + List<Item> item_list; Item *item_null= new (mem_root) Item_null(thd); diff --git a/sql/sql_explain.h b/sql/sql_explain.h index 38250cc..549b085 100644 --- a/sql/sql_explain.h +++ b/sql/sql_explain.h @@ -328,6 +328,7 @@ class Explain_aggr_window_funcs : public Explain_aggr_node ///////////////////////////////////////////////////////////////////////////// extern const char *unit_operation_text[4]; +extern const char *pushed_derived_text; /* Explain structure for a UNION. diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc index bff6dfb..b309a0a 100644 --- a/sql/sql_lex.cc +++ b/sql/sql_lex.cc @@ -4659,7 +4659,11 @@ void st_select_lex::set_explain_type(bool on_the_fly) /* If we're a direct child of a UNION, we're the first sibling there */ if (linkage == DERIVED_TABLE_TYPE) { - if (is_uncacheable & UNCACHEABLE_DEPENDENT) + bool is_pushed_master_unit= master_unit()->derived && + master_unit()->derived->pushdown_derived; + if (is_pushed_master_unit) + type= pushed_derived_text; + else if (is_uncacheable & UNCACHEABLE_DEPENDENT) type= "LATERAL DERIVED"; else type= "DERIVED"; diff --git a/sql/sql_select.cc b/sql/sql_select.cc index 3b92751..d0acbef 100644 --- a/sql/sql_select.cc +++ b/sql/sql_select.cc @@ -25681,6 +25681,7 @@ bool mysql_explain_union(THD *thd, SELECT_LEX_UNIT *unit, select_result *result) DBUG_ENTER("mysql_explain_union"); bool res= 0; SELECT_LEX *first= unit->first_select(); + bool is_pushed_union= unit->derived && unit->derived->pushdown_derived; for (SELECT_LEX *sl= first; sl; sl= sl->next_select()) { @@ -25698,9 +25699,12 @@ bool mysql_explain_union(THD *thd, SELECT_LEX_UNIT *unit, select_result *result) } if (!(res= unit->prepare(unit->derived, result, SELECT_NO_UNLOCK | SELECT_DESCRIBE))) - res= unit->exec(); + { + if (!is_pushed_union) + res= unit->exec(); + } } - else + else { thd->lex->current_select= first; unit->set_limit(unit->global_parameters()); diff --git a/sql/sql_select.h b/sql/sql_select.h index 4140a02..e51367c 100644 --- a/sql/sql_select.h +++ b/sql/sql_select.h @@ -2442,7 +2442,23 @@ class Pushdown_query: public Sql_alloc ~Pushdown_query() { delete handler; } /* Function that calls the above scan functions */ - int execute(JOIN *join); + int execute(JOIN *); +}; + +class derived_handler; + +class Pushdown_derived: public Sql_alloc +{ +public: + TABLE_LIST *derived; + derived_handler *handler; + + Pushdown_derived(TABLE_LIST *tbl, derived_handler *h) + : derived(tbl), handler(h) {} + + ~Pushdown_derived(); + + int execute(); }; bool test_if_order_compatible(SQL_I_List<ORDER> &a, SQL_I_List<ORDER> &b); diff --git a/sql/table.h b/sql/table.h index b75fa90..33cf23b 100644 --- a/sql/table.h +++ b/sql/table.h @@ -55,6 +55,8 @@ class Virtual_column_info; class Table_triggers_list; class TMP_TABLE_PARAM; class SEQUENCE; +class derived_handler; +class Pushdown_derived; /* Used to identify NESTED_JOIN structures within a join (applicable only to @@ -2118,6 +2120,8 @@ struct TABLE_LIST TABLE_LIST * next_with_rec_ref; bool is_derived_with_recursive_reference; bool block_handle_derived; + derived_handler *dt_handler; + Pushdown_derived *pushdown_derived; ST_SCHEMA_TABLE *schema_table; /* Information_schema table */ st_select_lex *schema_select_lex; /* @@ -2584,6 +2588,9 @@ struct TABLE_LIST } void set_lock_type(THD* thd, enum thr_lock_type lock); + derived_handler *find_derived_handler(THD *thd); + TABLE_LIST *get_first_table(); + private: bool prep_check_option(THD *thd, uint8 check_opt_type); bool prep_where(THD *thd, Item **conds, bool no_where_clause); diff --git a/storage/federatedx/ha_federatedx.cc b/storage/federatedx/ha_federatedx.cc index 74d547c..45cd14e 100644 --- a/storage/federatedx/ha_federatedx.cc +++ b/storage/federatedx/ha_federatedx.cc @@ -319,6 +319,7 @@ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. #include "sql_analyse.h" // append_escaped() #include "sql_show.h" // append_identifier() #include "tztime.h" // my_tz_find() +#include "sql_select.h" #ifdef I_AM_PARANOID #define MIN_PORT 1023 @@ -401,6 +402,10 @@ static void init_federated_psi_keys(void) #define init_federated_psi_keys() /* no-op */ #endif /* HAVE_PSI_INTERFACE */ +handlerton* federatedx_hton; + +static derived_handler* +create_federatedx_derived_handler(THD* thd, TABLE_LIST *derived); /* Initialize the federatedx handler. @@ -418,7 +423,7 @@ int federatedx_db_init(void *p) { DBUG_ENTER("federatedx_db_init"); init_federated_psi_keys(); - handlerton *federatedx_hton= (handlerton *)p; + federatedx_hton= (handlerton *)p; federatedx_hton->state= SHOW_OPTION_YES; /* Needed to work with old .frm files */ federatedx_hton->db_type= DB_TYPE_FEDERATED_DB; @@ -432,6 +437,7 @@ int federatedx_db_init(void *p) federatedx_hton->discover_table_structure= ha_federatedx::discover_assisted; federatedx_hton->create= federatedx_create_handler; federatedx_hton->flags= HTON_ALTER_NOT_SUPPORTED; + federatedx_hton->create_derived= create_federatedx_derived_handler; if (mysql_mutex_init(fe_key_mutex_federatedx, &federatedx_mutex, MY_MUTEX_INIT_FAST)) @@ -3668,6 +3674,126 @@ int ha_federatedx::discover_assisted(handlerton *hton, THD* thd, return error; } +static derived_handler* +create_federatedx_derived_handler(THD* thd, TABLE_LIST *derived) +{ + ha_federatedx_derived_handler* handler = NULL; + handlerton *ht= 0; + + SELECT_LEX_UNIT *unit= derived->derived; + + for (SELECT_LEX *sl= unit->first_select(); sl; sl= sl->next_select()) + { + if (!(sl->join)) + return 0; + for (TABLE_LIST *tbl= sl->join->tables_list; tbl; tbl= tbl->next_local) + { + if (!tbl->table) + return 0; + if (!ht) + ht= tbl->table->file->partition_ht(); + else if (ht != tbl->table->file->partition_ht()) + return 0; + } + } + + handler= new ha_federatedx_derived_handler(thd, derived); + + return handler; +} + + +ha_federatedx_derived_handler::ha_federatedx_derived_handler(THD *thd, + TABLE_LIST *dt) + : derived_handler(thd, federatedx_hton) +{ + derived= dt; +} + +ha_federatedx_derived_handler::~ha_federatedx_derived_handler() {} + +int ha_federatedx_derived_handler::init_scan() +{ + char query_buff[4096]; + THD *thd; + int rc= 0; + + DBUG_ENTER("ha_federatedx_derived_handler::init_scan"); + + TABLE *table= derived->get_first_table()->table; + ha_federatedx *h= (ha_federatedx *) table->file; + io= h->io; + share= get_share(table->s->table_name.str, table); + thd= table->in_use; + txn= h->get_txn(thd); + if ((rc= txn->acquire(share, thd, TRUE, &io))) + DBUG_RETURN(rc); + + String derived_query(query_buff, sizeof(query_buff), thd->charset()); + derived_query.length(0); + derived->derived->print(&derived_query, QT_ORDINARY); + + // if (stored_result) + // (void) free_result(); + + if (io->query(derived_query.ptr(), derived_query.length())) + goto err; + + stored_result= io->store_result(); + if (!stored_result) + goto err; + + DBUG_RETURN(0); + +err: + DBUG_RETURN(HA_FEDERATEDX_ERROR_WITH_REMOTE_SYSTEM); +} + +int ha_federatedx_derived_handler::next_row() +{ + int rc; + FEDERATEDX_IO_ROW *row; + ulong *lengths; + Field **field; + int column= 0; + Time_zone *saved_time_zone= table->in_use->variables.time_zone; + DBUG_ENTER("ha_federatedx_derived_handler::next_row"); + + if ((rc= txn->acquire(share, table->in_use, TRUE, &io))) + DBUG_RETURN(rc); + + if (!(row= io->fetch_row(stored_result))) + DBUG_RETURN(HA_ERR_END_OF_FILE); + + /* Convert row to internal format */ + table->in_use->variables.time_zone= UTC; + lengths= io->fetch_lengths(stored_result); + + for (field= table->field; *field; field++, column++) + { + if (io->is_column_null(row, column)) + (*field)->set_null(); + else + { + (*field)->set_notnull(); + (*field)->store(io->get_column_data(row, column), + lengths[column], &my_charset_bin); + } + } + table->in_use->variables.time_zone= saved_time_zone; + + DBUG_RETURN(rc); +} + +int ha_federatedx_derived_handler::end_scan() +{ + DBUG_ENTER("ha_federatedx_derived_handler::end_scan"); + DBUG_RETURN(0); +} + +void ha_federatedx_derived_handler::print_error(int, unsigned long) +{ +} struct st_mysql_storage_engine federatedx_storage_engine= { MYSQL_HANDLERTON_INTERFACE_VERSION }; @@ -3689,3 +3815,4 @@ maria_declare_plugin(federatedx) MariaDB_PLUGIN_MATURITY_STABLE /* maturity */ } maria_declare_plugin_end; + diff --git a/storage/federatedx/ha_federatedx.h b/storage/federatedx/ha_federatedx.h index 16a1944..61c7029 100644 --- a/storage/federatedx/ha_federatedx.h +++ b/storage/federatedx/ha_federatedx.h @@ -1,3 +1,5 @@ +#ifndef HA_FEDERATEDX_INCLUDED +#define HA_FEDERATEDX_INCLUDED /* Copyright (c) 2008, Patrick Galbraith All rights reserved. @@ -40,6 +42,7 @@ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. #include <my_global.h> #include <thr_lock.h> #include "handler.h" +#include "derived_handler.h" class federatedx_io; @@ -445,6 +448,8 @@ class ha_federatedx: public handler int external_lock(THD *thd, int lock_type); int reset(void); int free_result(void); + + friend class ha_federatedx_derived_handler; }; extern const char ident_quote_char; // Character for quoting @@ -460,3 +465,22 @@ extern federatedx_io *instantiate_io_mysql(MEM_ROOT *server_root, FEDERATEDX_SERVER *server); extern federatedx_io *instantiate_io_null(MEM_ROOT *server_root, FEDERATEDX_SERVER *server); + +class ha_federatedx_derived_handler: public derived_handler +{ +private: + FEDERATEDX_SHARE *share; + federatedx_txn *txn; + federatedx_io *io; + FEDERATEDX_IO_RESULT *stored_result; + +public: + ha_federatedx_derived_handler(THD* thd_arg, TABLE_LIST *tbl); + ~ha_federatedx_derived_handler(); + int init_scan(); + int next_row(); + int end_scan(); + void print_error(int, unsigned long); +}; + +#endif /* HA_FEDERATEDX_INCLUDED */