revision-id: 2fef5d8d8e5b5781a71ff6305c0eb7a3de555509 (mariadb-10.4.0-38-g2fef5d8d8e5) parent(s): 2c4844c9e76427525e8c39a2d72686085efe89c3 author: Oleksandr Byelkin committer: Oleksandr Byelkin timestamp: 2018-11-27 14:02:49 +0100 message: MDEV-11419: Report all INSERT ID for bulk operation INSERT Send all Insert IDs of the buld operation to client (JDBC need it) # Conflicts: # sql/sql_class.cc # sql/sql_class.h # sql/sql_insert.cc # sql/sql_prepare.cc --- include/mysql.h.pp | 9 +++- include/mysql_com.h | 10 ++++- sql/protocol.cc | 22 +++++++--- sql/protocol.h | 6 +++ sql/sql_class.cc | 117 ++++++++++++++++++++++++++++++++++++++++++++++++++++ sql/sql_class.h | 8 ++++ sql/sql_insert.cc | 11 ++++- sql/sql_prepare.cc | 23 +++++++++-- 8 files changed, 193 insertions(+), 13 deletions(-) diff --git a/include/mysql.h.pp b/include/mysql.h.pp index e36131b93d2..4da8a5be491 100644 --- a/include/mysql.h.pp +++ b/include/mysql.h.pp @@ -92,7 +92,14 @@ enum enum_cursor_type CURSOR_TYPE_NO_CURSOR= 0, CURSOR_TYPE_READ_ONLY= 1, CURSOR_TYPE_FOR_UPDATE= 2, - CURSOR_TYPE_SCROLLABLE= 4 + CURSOR_TYPE_SCROLLABLE= 4, +}; +enum stmt_flags_type +{ + STMTFLG_CURSOR_TYPE_READ_ONLY= CURSOR_TYPE_READ_ONLY, + STMTFLG_CURSOR_TYPE_FOR_UPDATE= CURSOR_TYPE_FOR_UPDATE, + STMTFLG_CURSOR_TYPE_SCROLLABLE= CURSOR_TYPE_FOR_UPDATE, + STMTFLG_INSERT_ID_REQUEST= 128 }; enum enum_mysql_set_option { diff --git a/include/mysql_com.h b/include/mysql_com.h index 902c0ff2706..4217d6ca0e5 100644 --- a/include/mysql_com.h +++ b/include/mysql_com.h @@ -578,7 +578,15 @@ enum enum_cursor_type CURSOR_TYPE_NO_CURSOR= 0, CURSOR_TYPE_READ_ONLY= 1, CURSOR_TYPE_FOR_UPDATE= 2, - CURSOR_TYPE_SCROLLABLE= 4 + CURSOR_TYPE_SCROLLABLE= 4, +}; +/* first values should be the same as enum_cursor_type */ +enum stmt_flags_type +{ + STMTFLG_CURSOR_TYPE_READ_ONLY= CURSOR_TYPE_READ_ONLY, + STMTFLG_CURSOR_TYPE_FOR_UPDATE= CURSOR_TYPE_FOR_UPDATE, + STMTFLG_CURSOR_TYPE_SCROLLABLE= CURSOR_TYPE_FOR_UPDATE, + STMTFLG_INSERT_ID_REQUEST= 128 }; diff --git a/sql/protocol.cc b/sql/protocol.cc index 7eee9283989..a68159a20cd 100644 --- a/sql/protocol.cc +++ b/sql/protocol.cc @@ -563,6 +563,7 @@ void Protocol::end_statement() switch (thd->get_stmt_da()->status()) { case Diagnostics_area::DA_ERROR: + thd->stop_collecting_insert_id(); /* The query failed, send error to log and abort bootstrap. */ error= send_error(thd->get_stmt_da()->sql_errno(), thd->get_stmt_da()->message(), @@ -574,12 +575,21 @@ void Protocol::end_statement() break; case Diagnostics_area::DA_OK: case Diagnostics_area::DA_OK_BULK: - error= send_ok(thd->server_status, - thd->get_stmt_da()->statement_warn_count(), - thd->get_stmt_da()->affected_rows(), - thd->get_stmt_da()->last_insert_id(), - thd->get_stmt_da()->message(), - thd->get_stmt_da()->skip_flush()); + if (thd->report_collected_insert_id()) + if (thd->is_error()) + error= send_error(thd->get_stmt_da()->sql_errno(), + thd->get_stmt_da()->message(), + thd->get_stmt_da()->get_sqlstate()); + else + error= send_eof(thd->server_status, + thd->get_stmt_da()->statement_warn_count()); + else + error= send_ok(thd->server_status, + thd->get_stmt_da()->statement_warn_count(), + thd->get_stmt_da()->affected_rows(), + thd->get_stmt_da()->last_insert_id(), + thd->get_stmt_da()->message(), + thd->get_stmt_da()->skip_flush()); break; case Diagnostics_area::DA_DISABLED: break; diff --git a/sql/protocol.h b/sql/protocol.h index 6397e3dd5e6..bf74f52fa98 100644 --- a/sql/protocol.h +++ b/sql/protocol.h @@ -30,6 +30,12 @@ class Item_param; typedef struct st_mysql_field MYSQL_FIELD; typedef struct st_mysql_rows MYSQL_ROWS; +struct insert_id_desc +{ + ulonglong first_id; + ulonglong sequence; +}; + class Protocol { protected: diff --git a/sql/sql_class.cc b/sql/sql_class.cc index 52ebc186b1a..87b83e60bd9 100644 --- a/sql/sql_class.cc +++ b/sql/sql_class.cc @@ -1261,6 +1261,7 @@ void THD::init(bool skip_lock) #endif //EMBEDDED_LIBRARY apc_target.init(&LOCK_thd_kill); + insert_ids= NULL; DBUG_VOID_RETURN; } @@ -7692,6 +7693,122 @@ bool Qualified_column_ident::append_to(THD *thd, String *str) const } +bool THD::init_collecting_insert_id() +{ + if (!insert_ids) + { + void *buff; + if (!(my_multi_malloc(MYF(MY_WME), &insert_ids, sizeof(DYNAMIC_ARRAY), + &buff, sizeof(insert_id_desc) * 10, + NullS)) || + my_init_dynamic_array2(insert_ids, sizeof(insert_id_desc), + buff, 10, 100, MYF(MY_WME))) + { + if (insert_ids) + my_free(insert_ids); + insert_ids= NULL; + return TRUE; + } + collect_auto_increment_increment= variables.auto_increment_increment; + } + return FALSE; +} + +void THD::stop_collecting_insert_id() +{ + if (insert_ids) + { + delete_dynamic(insert_ids); + my_free(insert_ids); + insert_ids= NULL; + } +} + +bool THD::collect_insert_id(ulonglong id) +{ + if (insert_ids) + { + if (insert_ids->elements) + { + insert_id_desc *last= + (insert_id_desc *)dynamic_array_ptr(insert_ids, + insert_ids->elements - 1); + if (id == last->first_id) + { + return FALSE; // no new insert id + } + if (id == last->first_id + (last->sequence * + collect_auto_increment_increment)) + { + last->sequence++; + return FALSE; + } + } + insert_id_desc el; + el.first_id= id; + el.sequence= 1; + if (insert_dynamic(insert_ids, &el)) + { + return TRUE; + } + } + return FALSE; +} + + +bool THD::report_collected_insert_id() +{ + if (insert_ids) + { + List<Item> field_list; + MEM_ROOT tmp_mem_root; + Query_arena arena(&tmp_mem_root, Query_arena::STMT_INITIALIZED), backup; + + init_alloc_root(arena.mem_root, " THD::report_collected_insert_i", + 2048, 4096, MYF(MY_THREAD_SPECIFIC)); + set_n_backup_active_arena(&arena, &backup); + DBUG_ASSERT(mem_root == &tmp_mem_root); + + field_list.push_back(new (mem_root) + Item_int(this, "Id", 0, MY_INT64_NUM_DECIMAL_DIGITS), + mem_root); + field_list.push_back(new (mem_root) + Item_int(this, "Len", 0, MY_INT64_NUM_DECIMAL_DIGITS), + mem_root); + field_list.push_back(new (mem_root) + Item_return_int(this, "Inc", 0, MYSQL_TYPE_LONG), + mem_root); + + if (protocol_binary.send_result_set_metadata(&field_list, + Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF)) + goto error; + + for (ulonglong i= 0; i < insert_ids->elements; i++) + { + insert_id_desc *last= + (insert_id_desc *)dynamic_array_ptr(insert_ids, i); + if (insert_ids->elements == 1 && last->first_id == 0 && + get_stmt_da()->affected_rows() != 1) + continue; // No insert IDs + protocol_binary.prepare_for_resend(); + protocol_binary.store_longlong(last->first_id, TRUE); + protocol_binary.store_longlong(last->sequence, TRUE); + protocol_binary.store_long(collect_auto_increment_increment); + if (protocol_binary.write()) + goto error; + } +error: + restore_active_arena(&arena, &backup); + DBUG_ASSERT(arena.mem_root == &tmp_mem_root); + // no need free Items because they was only constants + free_root(arena.mem_root, MYF(0)); + stop_collecting_insert_id(); + return TRUE; + } + return FALSE; + +} + #endif /* !defined(MYSQL_CLIENT) */ diff --git a/sql/sql_class.h b/sql/sql_class.h index 8174a8b313b..5fb6a6c575e 100644 --- a/sql/sql_class.h +++ b/sql/sql_class.h @@ -4901,6 +4901,14 @@ class THD :public Statement, LOG_SLOW_DISABLE_ADMIN); query_plan_flags|= QPLAN_ADMIN; } + + /* Data and methods for buld INSERT IDs reporting */ + DYNAMIC_ARRAY *insert_ids; + ulong collect_auto_increment_increment; + bool init_collecting_insert_id(); + bool collect_insert_id(ulonglong id); + bool report_collected_insert_id(); + void stop_collecting_insert_id(); }; inline void add_to_active_threads(THD *thd) diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc index 127b4b10eb4..349ac2646fe 100644 --- a/sql/sql_insert.cc +++ b/sql/sql_insert.cc @@ -1067,6 +1067,12 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list, } its.rewind(); iteration++; + + if (!error && thd->bulk_param) + { + thd->collect_insert_id(table->file->insert_id_for_cur_row); + } + } while (bulk_parameters_iterations(thd)); values_loop_end: @@ -1217,8 +1223,9 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list, retval= thd->lex->explain->send_explain(thd); goto abort; } - if ((iteration * values_list.elements) == 1 && (!(thd->variables.option_bits & OPTION_WARNINGS) || - !thd->cuted_fields)) + if ((iteration * values_list.elements) == 1 && + (!(thd->variables.option_bits & OPTION_WARNINGS) || + !thd->cuted_fields)) { my_ok(thd, info.copied + info.deleted + ((thd->client_capabilities & CLIENT_FOUND_ROWS) ? diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc index cb822fc2e98..5346af194ef 100644 --- a/sql/sql_prepare.cc +++ b/sql/sql_prepare.cc @@ -200,7 +200,8 @@ class Prepared_statement: public Statement uchar *packet_arg, uchar *packet_end_arg); bool execute_bulk_loop(String *expanded_query, bool open_cursor, - uchar *packet_arg, uchar *packet_end_arg); + uchar *packet_arg, uchar *packet_end_arg, + bool insert_id_request); bool execute_server_runnable(Server_runnable *server_runnable); my_bool set_bulk_parameters(bool reset); bool bulk_iterations() { return iterations; }; @@ -3214,10 +3215,19 @@ static void mysql_stmt_execute_common(THD *thd, open_cursor= MY_TEST(cursor_flags & (ulong) CURSOR_TYPE_READ_ONLY); thd->protocol= &thd->protocol_binary; + + if (!(thd->client_capabilities & MARIADB_CLIENT_STMT_BULK_OPERATIONS)) + { + DBUG_PRINT("info", + ("There is no bulk capability so reset iteration counter")); + bulk_op= FALSE; + } if (!bulk_op) stmt->execute_loop(&expanded_query, open_cursor, packet, packet_end); else - stmt->execute_bulk_loop(&expanded_query, open_cursor, packet, packet_end); + stmt->execute_bulk_loop(&expanded_query, open_cursor, packet, packet_end, + MY_TEST((cursor_flags & + (ulong) STMTFLG_INSERT_ID_REQUEST))); thd->protocol= save_protocol; sp_cache_enforce_limit(thd->sp_proc_cache, stored_program_cache_size); @@ -4310,7 +4320,8 @@ bool Prepared_statement::execute_bulk_loop(String *expanded_query, bool open_cursor, uchar *packet_arg, - uchar *packet_end_arg) + uchar *packet_end_arg, + bool insert_id_request) { Reprepare_observer reprepare_observer; bool error= 0; @@ -4345,6 +4356,12 @@ Prepared_statement::execute_bulk_loop(String *expanded_query, thd->set_bulk_execution(0); return TRUE; } + if (lex->sql_command == SQLCOM_INSERT && insert_id_request && + thd->init_collecting_insert_id()) + { + thd->set_bulk_execution(0); + return TRUE; + } #ifndef EMBEDDED_LIBRARY if (read_types &&