diff options
Diffstat (limited to 'storage/innobase/row/row0ins.cc')
| -rw-r--r-- | storage/innobase/row/row0ins.cc | 1387 |
1 files changed, 921 insertions, 466 deletions
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc index e3817ea4de7..05d2378b81e 100644 --- a/storage/innobase/row/row0ins.cc +++ b/storage/innobase/row/row0ins.cc @@ -23,13 +23,14 @@ Insert into a table Created 4/20/1996 Heikki Tuuri *******************************************************/ +#include "ha_prototypes.h" + #include "row0ins.h" #ifdef UNIV_NONINL #include "row0ins.ic" #endif -#include "ha_prototypes.h" #include "dict0dict.h" #include "dict0boot.h" #include "trx0rec.h" @@ -52,6 +53,7 @@ Created 4/20/1996 Heikki Tuuri #include "fts0fts.h" #include "fts0types.h" #include "m_string.h" +#include "gis0geo.h" /************************************************************************* IMPORTANT NOTE: Any operation that generates redo MUST check that there @@ -65,8 +67,7 @@ introduced where a call to log_free_check() is bypassed. */ /*********************************************************************//** Creates an insert node struct. -@return own: insert node struct */ -UNIV_INTERN +@return own: insert node struct */ ins_node_t* ins_node_create( /*============*/ @@ -91,6 +92,7 @@ ins_node_create( node->select = NULL; node->trx_id = 0; + node->duplicate = NULL; node->entry_sys_heap = mem_heap_create(128); @@ -112,7 +114,7 @@ ins_node_create_entry_list( ut_ad(node->entry_sys_heap); - UT_LIST_INIT(node->entry_list); + UT_LIST_INIT(node->entry_list, &dtuple_t::tuple_list); /* We will include all indexes (include those corrupted secondary indexes) in the entry list. Filteration of @@ -122,10 +124,11 @@ ins_node_create_entry_list( index != 0; index = dict_table_get_next_index(index)) { - entry = row_build_index_entry( - node->row, NULL, index, node->entry_sys_heap); + entry = row_build_index_entry_low( + node->row, NULL, index, node->entry_sys_heap, + ROW_BUILD_FOR_INSERT); - UT_LIST_ADD_LAST(tuple_list, node->entry_list, entry); + UT_LIST_ADD_LAST(node->entry_list, entry); } } @@ -152,7 +155,10 @@ row_ins_alloc_sys_fields( ut_ad(dtuple_get_n_fields(row) == dict_table_get_n_cols(table)); /* allocate buffer to hold the needed system created hidden columns. */ - uint len = DATA_ROW_ID_LEN + DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN; + uint len = DATA_ROW_ID_LEN + DATA_TRX_ID_LEN; + if (!dict_table_is_intrinsic(table)) { + len += DATA_ROLL_PTR_LEN; + } ptr = static_cast<byte*>(mem_heap_zalloc(heap, len)); /* 1. Populate row-id */ @@ -177,20 +183,19 @@ row_ins_alloc_sys_fields( ptr += DATA_TRX_ID_LEN; - /* 3. Populate roll ptr */ - - col = dict_table_get_sys_col(table, DATA_ROLL_PTR); + if (!dict_table_is_intrinsic(table)) { + col = dict_table_get_sys_col(table, DATA_ROLL_PTR); - dfield = dtuple_get_nth_field(row, dict_col_get_no(col)); + dfield = dtuple_get_nth_field(row, dict_col_get_no(col)); - dfield_set_data(dfield, ptr, DATA_ROLL_PTR_LEN); + dfield_set_data(dfield, ptr, DATA_ROLL_PTR_LEN); + } } /*********************************************************************//** Sets a new row to insert for an INS_DIRECT node. This function is only used if we have constructed the row separately, which is a rare case; this function is quite slow. */ -UNIV_INTERN void ins_node_set_new_row( /*=================*/ @@ -200,6 +205,7 @@ ins_node_set_new_row( node->state = INS_NODE_SET_IX_LOCK; node->index = NULL; node->entry = NULL; + node->duplicate = NULL; node->row = row; @@ -223,7 +229,7 @@ ins_node_set_new_row( Does an insert operation by updating a delete-marked existing record in the index. This situation can occur if the delete-marked record is kept in the index for consistent reads. -@return DB_SUCCESS or error code */ +@return DB_SUCCESS or error code */ static MY_ATTRIBUTE((nonnull, warn_unused_result)) dberr_t row_ins_sec_index_entry_by_modify( @@ -270,11 +276,10 @@ row_ins_sec_index_entry_by_modify( case, the change would already be there. The CREATE INDEX should be waiting for a MySQL meta-data lock upgrade at least until this INSERT or UPDATE - returns. After that point, the TEMP_INDEX_PREFIX - would be dropped from the index name in - commit_inplace_alter_table(). */ + returns. After that point, set_committed(true) + would be invoked in commit_inplace_alter_table(). */ ut_a(update->n_fields == 0); - ut_a(*cursor->index->name == TEMP_INDEX_PREFIX); + ut_a(!cursor->index->is_committed()); ut_ad(!dict_index_is_online_ddl(cursor->index)); return(DB_SUCCESS); } @@ -318,37 +323,34 @@ row_ins_sec_index_entry_by_modify( Does an insert operation by delete unmarking and updating a delete marked existing record in the index. This situation can occur if the delete marked record is kept in the index for consistent reads. -@return DB_SUCCESS, DB_FAIL, or error code */ +@return DB_SUCCESS, DB_FAIL, or error code */ static MY_ATTRIBUTE((nonnull, warn_unused_result)) dberr_t row_ins_clust_index_entry_by_modify( /*================================*/ + btr_pcur_t* pcur, /*!< in/out: a persistent cursor pointing + to the clust_rec that is being modified. */ ulint flags, /*!< in: undo logging and locking flags */ ulint mode, /*!< in: BTR_MODIFY_LEAF or BTR_MODIFY_TREE, depending on whether mtr holds just a leaf latch or also a tree latch */ - btr_cur_t* cursor, /*!< in: B-tree cursor */ ulint** offsets,/*!< out: offsets on cursor->page_cur.rec */ mem_heap_t** offsets_heap, /*!< in/out: pointer to memory heap that can be emptied, or NULL */ mem_heap_t* heap, /*!< in/out: memory heap */ - big_rec_t** big_rec,/*!< out: possible big rec vector of fields - which have to be stored externally by the - caller */ const dtuple_t* entry, /*!< in: index entry to insert */ que_thr_t* thr, /*!< in: query thread */ mtr_t* mtr) /*!< in: mtr; must be committed before latching any further pages */ { const rec_t* rec; - const upd_t* update; + upd_t* update; dberr_t err; + btr_cur_t* cursor = btr_pcur_get_btr_cur(pcur); ut_ad(dict_index_is_clust(cursor->index)); - *big_rec = NULL; - rec = btr_cur_get_rec(cursor); ut_ad(rec_get_deleted_flag(rec, @@ -384,10 +386,24 @@ row_ins_clust_index_entry_by_modify( return(DB_LOCK_TABLE_FULL); } + + big_rec_t* big_rec = NULL; + err = btr_cur_pessimistic_update( flags | BTR_KEEP_POS_FLAG, cursor, offsets, offsets_heap, heap, - big_rec, update, 0, thr, thr_get_trx(thr)->id, mtr); + &big_rec, update, 0, thr, thr_get_trx(thr)->id, mtr); + + if (big_rec) { + ut_a(err == DB_SUCCESS); + + DEBUG_SYNC_C("before_row_ins_upd_extern"); + err = btr_store_big_rec_extern_fields( + pcur, update, *offsets, big_rec, mtr, + BTR_STORE_INSERT_UPDATE); + DEBUG_SYNC_C("after_row_ins_upd_extern"); + dtuple_big_rec_free(big_rec); + } } return(err); @@ -396,7 +412,7 @@ row_ins_clust_index_entry_by_modify( /*********************************************************************//** Returns TRUE if in a cascaded update/delete an ancestor node of node updates (not DELETE, but UPDATE) table. -@return TRUE if an ancestor updates table */ +@return TRUE if an ancestor updates table */ static ibool row_ins_cascade_ancestor_updates_table( @@ -426,7 +442,7 @@ row_ins_cascade_ancestor_updates_table( /*********************************************************************//** Returns the number of ancestor UPDATE or DELETE nodes of a cascaded update/delete node. -@return number of ancestors */ +@return number of ancestors */ static MY_ATTRIBUTE((nonnull, warn_unused_result)) ulint row_ins_cascade_n_ancestors( @@ -464,9 +480,10 @@ row_ins_cascade_calc_update_vec( mem_heap_t* heap, /*!< in: memory heap to use as temporary storage */ trx_t* trx, /*!< in: update transaction */ - ibool* fts_col_affected)/*!< out: is FTS column affected */ + ibool* fts_col_affected, + /*!< out: is FTS column affected */ + upd_node_t* cascade) /*!< in: cascade update node */ { - upd_node_t* cascade = node->cascade_node; dict_table_t* table = foreign->foreign_table; dict_index_t* index = foreign->foreign_index; upd_t* update; @@ -480,6 +497,7 @@ row_ins_cascade_calc_update_vec( ibool doc_id_updated = FALSE; ulint doc_id_pos = 0; doc_id_t new_doc_id = FTS_NULL_DOC_ID; + ulint prefix_col; ut_a(node); ut_a(foreign); @@ -508,14 +526,15 @@ row_ins_cascade_calc_update_vec( if (table->fts) { doc_id_pos = dict_table_get_nth_col_pos( - table, table->fts->doc_col); + table, table->fts->doc_col, &prefix_col); } for (i = 0; i < foreign->n_fields; i++) { parent_field_no = dict_table_get_nth_col_pos( parent_table, - dict_index_get_nth_col_no(parent_index, i)); + dict_index_get_nth_col_no(parent_index, i), + &prefix_col); for (j = 0; j < parent_update->n_fields; j++) { const upd_field_t* parent_ufield @@ -538,7 +557,8 @@ row_ins_cascade_calc_update_vec( ufield->field_no = dict_table_get_nth_col_pos( - table, dict_col_get_no(col)); + table, dict_col_get_no(col), + &prefix_col); ufield->orig_len = 0; ufield->exp = NULL; @@ -629,7 +649,8 @@ row_ins_cascade_calc_update_vec( if (table->fts && dict_table_is_fts_column( table->fts->indexes, - dict_col_get_no(col)) + dict_col_get_no(col), + dict_col_is_virtual(col)) != ULINT_UNDEFINED) { *fts_col_affected = TRUE; } @@ -649,25 +670,19 @@ row_ins_cascade_calc_update_vec( &ufield->new_val))); if (new_doc_id <= 0) { - fprintf(stderr, - "InnoDB: FTS Doc ID " - "must be larger than " - "0 \n"); + ib::error() << "FTS Doc ID" + " must be larger than" + " 0"; return(ULINT_UNDEFINED); } if (new_doc_id < n_doc_id) { - fprintf(stderr, - "InnoDB: FTS Doc ID " - "must be larger than " - IB_ID_FMT" for table", - n_doc_id -1); - - ut_print_name(stderr, trx, - TRUE, - table->name); + ib::error() << "FTS Doc ID" + " must be larger than " + << n_doc_id - 1 + << " for table " + << table->name; - putc('\n', stderr); return(ULINT_UNDEFINED); } @@ -684,26 +699,27 @@ row_ins_cascade_calc_update_vec( if (table->fts && *fts_col_affected) { if (DICT_TF2_FLAG_IS_SET(table, DICT_TF2_FTS_HAS_DOC_ID)) { doc_id_t doc_id; - upd_field_t* ufield; + doc_id_t* next_doc_id; + upd_field_t* ufield; + + next_doc_id = static_cast<doc_id_t*>(mem_heap_alloc( + heap, sizeof(doc_id_t))); ut_ad(!doc_id_updated); ufield = update->fields + n_fields_updated; - fts_get_next_doc_id(table, &trx->fts_next_doc_id); - doc_id = fts_update_doc_id(table, ufield, - &trx->fts_next_doc_id); + fts_get_next_doc_id(table, next_doc_id); + doc_id = fts_update_doc_id(table, ufield, next_doc_id); n_fields_updated++; - fts_trx_add_op(trx, table, doc_id, FTS_INSERT, NULL); + cascade->fts_next_doc_id = doc_id; } else { if (doc_id_updated) { ut_ad(new_doc_id); - fts_trx_add_op(trx, table, new_doc_id, - FTS_INSERT, NULL); + cascade->fts_next_doc_id = new_doc_id; } else { - fprintf(stderr, "InnoDB: FTS Doc ID must be " - "updated along with FTS indexed " - "column for table "); - ut_print_name(stderr, trx, TRUE, table->name); - putc('\n', stderr); + cascade->fts_next_doc_id = FTS_NULL_DOC_ID; + ib::error() << "FTS Doc ID must be updated" + " along with FTS indexed column for" + " table " << table->name; return(ULINT_UNDEFINED); } } @@ -730,10 +746,9 @@ row_ins_set_detailed( rewind(srv_misc_tmpfile); if (os_file_set_eof(srv_misc_tmpfile)) { - std::string fk_str; - ut_print_name(srv_misc_tmpfile, trx, TRUE, + ut_print_name(srv_misc_tmpfile, trx, foreign->foreign_table_name); - fk_str = dict_print_info_on_foreign_key_in_create_format( + std::string fk_str = dict_print_info_on_foreign_key_in_create_format( trx, foreign, FALSE); fputs(fk_str.c_str(), srv_misc_tmpfile); trx_set_detailed_error_from_file(trx, srv_misc_tmpfile); @@ -768,7 +783,7 @@ row_ins_foreign_trx_print( heap_size = mem_heap_get_size(trx->lock.lock_heap); lock_mutex_exit(); - mutex_enter(&trx_sys->mutex); + trx_sys_mutex_enter(); mutex_enter(&dict_foreign_err_mutex); rewind(dict_foreign_err_file); @@ -778,7 +793,7 @@ row_ins_foreign_trx_print( trx_print_low(dict_foreign_err_file, trx, 600, n_rec_locks, n_trx_locks, heap_size); - mutex_exit(&trx_sys->mutex); + trx_sys_mutex_exit(); ut_ad(mutex_own(&dict_foreign_err_mutex)); } @@ -814,23 +829,22 @@ row_ins_foreign_report_err( row_ins_foreign_trx_print(trx); fputs("Foreign key constraint fails for table ", ef); - ut_print_name(ef, trx, TRUE, foreign->foreign_table_name); + ut_print_name(ef, trx, foreign->foreign_table_name); fputs(":\n", ef); fk_str = dict_print_info_on_foreign_key_in_create_format(trx, foreign, TRUE); fputs(fk_str.c_str(), ef); putc('\n', ef); fputs(errstr, ef); - fputs(" in parent table, in index ", ef); - ut_print_name(ef, trx, FALSE, foreign->referenced_index->name); + fprintf(ef, " in parent table, in index %s", + foreign->referenced_index->name()); if (entry) { fputs(" tuple:\n", ef); dtuple_print(ef, entry); } fputs("\nBut in child table ", ef); - ut_print_name(ef, trx, TRUE, foreign->foreign_table_name); - fputs(", in index ", ef); - ut_print_name(ef, trx, FALSE, foreign->foreign_index->name); + ut_print_name(ef, trx, foreign->foreign_table_name); + fprintf(ef, ", in index %s", foreign->foreign_index->name()); if (rec) { fputs(", there is a record:\n", ef); rec_print(ef, rec, foreign->foreign_index); @@ -871,13 +885,13 @@ row_ins_foreign_report_add_err( row_ins_foreign_trx_print(trx); fputs("Foreign key constraint fails for table ", ef); - ut_print_name(ef, trx, TRUE, foreign->foreign_table_name); + ut_print_name(ef, trx, foreign->foreign_table_name); fputs(":\n", ef); fk_str = dict_print_info_on_foreign_key_in_create_format(trx, foreign, TRUE); fputs(fk_str.c_str(), ef); - fputs("\nTrying to add in child table, in index ", ef); - ut_print_name(ef, trx, FALSE, foreign->foreign_index->name); + fprintf(ef, " in parent table, in index %s", + foreign->foreign_index->name()); if (entry) { fputs(" tuple:\n", ef); /* TODO: DB_TRX_ID and DB_ROLL_PTR may be uninitialized. @@ -885,10 +899,10 @@ row_ins_foreign_report_add_err( dtuple_print(ef, entry); } fputs("\nBut in parent table ", ef); - ut_print_name(ef, trx, TRUE, foreign->referenced_table_name); - fputs(", in index ", ef); - ut_print_name(ef, trx, FALSE, foreign->referenced_index->name); - fputs(",\nthe closest match we can find is record:\n", ef); + ut_print_name(ef, trx, foreign->referenced_table_name); + fprintf(ef, ", in index %s,\n" + "the closest match we can find is record:\n", + foreign->referenced_index->name()); if (rec && page_rec_is_supremum(rec)) { /* If the cursor ended on a supremum record, it is better to report the previous record in the error message, so that @@ -915,18 +929,8 @@ row_ins_invalidate_query_cache( const char* name) /*!< in: table name prefixed with database name and a '/' character */ { - char* buf; - char* ptr; ulint len = strlen(name) + 1; - - buf = mem_strdupl(name, len); - - ptr = strchr(buf, '/'); - ut_a(ptr); - *ptr = '\0'; - - innobase_invalidate_query_cache(thr_get_trx(thr), buf, len); - mem_free(buf); + innobase_invalidate_query_cache(thr_get_trx(thr), name, len); } #ifdef WITH_WSREP dberr_t wsrep_append_foreign_key(trx_t *trx, @@ -941,7 +945,7 @@ dberr_t wsrep_append_foreign_key(trx_t *trx, Perform referential actions or checks when a parent row is deleted or updated and the constraint had an ON DELETE or ON UPDATE condition which was not RESTRICT. -@return DB_SUCCESS, DB_LOCK_WAIT, or error code */ +@return DB_SUCCESS, DB_LOCK_WAIT, or error code */ static MY_ATTRIBUTE((nonnull, warn_unused_result)) dberr_t row_ins_foreign_check_on_constraint( @@ -963,7 +967,6 @@ row_ins_foreign_check_on_constraint( dict_index_t* index; dict_index_t* clust_index; dtuple_t* ref; - mem_heap_t* upd_vec_heap = NULL; const rec_t* rec; const rec_t* clust_rec; const buf_block_t* clust_block; @@ -976,6 +979,7 @@ row_ins_foreign_check_on_constraint( doc_id_t doc_id = FTS_NULL_DOC_ID; ibool fts_col_affacted = FALSE; + DBUG_ENTER("row_ins_foreign_check_on_constraint"); ut_a(thr); ut_a(foreign); ut_a(pcur); @@ -986,10 +990,10 @@ row_ins_foreign_check_on_constraint( /* Since we are going to delete or update a row, we have to invalidate the MySQL query cache for table. A deadlock of threads is not possible here because the caller of this function does not hold any latches with - the sync0sync.h rank above the lock_sys_t::mutex. The query cache mutex - has a rank just above the lock_sys_t::mutex. */ + the mutex rank above the lock_sys_t::mutex. The query cache mutex + has a rank just above the lock_sys_t::mutex. */ - row_ins_invalidate_query_cache(thr, table->name); + row_ins_invalidate_query_cache(thr, table->name.m_name); node = static_cast<upd_node_t*>(thr->run_node); @@ -1001,7 +1005,7 @@ row_ins_foreign_check_on_constraint( thr, foreign, btr_pcur_get_rec(pcur), entry); - return(DB_ROW_IS_REFERENCED); + DBUG_RETURN(DB_ROW_IS_REFERENCED); } if (!node->is_delete && 0 == (foreign->type @@ -1014,26 +1018,20 @@ row_ins_foreign_check_on_constraint( thr, foreign, btr_pcur_get_rec(pcur), entry); - return(DB_ROW_IS_REFERENCED); - } - - if (node->cascade_node == NULL) { - /* Extend our query graph by creating a child to current - update node. The child is used in the cascade or set null - operation. */ - - node->cascade_heap = mem_heap_create(128); - node->cascade_node = row_create_update_node_for_mysql( - table, node->cascade_heap); - que_node_set_parent(node->cascade_node, node); + DBUG_RETURN(DB_ROW_IS_REFERENCED); } - /* Initialize cascade_node to do the operation we want. Note that we - use the SAME cascade node to do all foreign key operations of the - SQL DELETE: the table of the cascade node may change if there are - several child tables to the table where the delete is done! */ + cascade = row_create_update_node_for_mysql(table, node->cascade_heap); + que_node_set_parent(cascade, node); - cascade = node->cascade_node; + /* For the cascaded operation, all the update nodes are allocated in + the same heap. All the update nodes will point to the same heap. + This heap is owned by the first update node. And it must be freed + only in the first update node */ + cascade->cascade_heap = node->cascade_heap; + cascade->cascade_upd_nodes = node->cascade_upd_nodes; + cascade->new_upd_nodes = node->new_upd_nodes; + cascade->processed_cascades = node->processed_cascades; cascade->table = table; @@ -1079,8 +1077,8 @@ row_ins_foreign_check_on_constraint( goto nonstandard_exit_func; } - if (row_ins_cascade_n_ancestors(cascade) >= 15) { - err = DB_ROW_IS_REFERENCED; + if (row_ins_cascade_n_ancestors(cascade) >= FK_MAX_CASCADE_DEL) { + err = DB_FOREIGN_EXCEED_MAX_CASCADE; row_ins_foreign_report_err( "Trying a too deep cascaded delete or update\n", @@ -1123,12 +1121,11 @@ row_ins_foreign_check_on_constraint( || btr_pcur_get_low_match(cascade->pcur) < dict_index_get_n_unique(clust_index)) { - fputs("InnoDB: error in cascade of a foreign key op\n" - "InnoDB: ", stderr); - dict_index_name_print(stderr, trx, index); + ib::error() << "In cascade of a foreign key op index " + << index->name + << " of table " << index->table->name; - fputs("\n" - "InnoDB: record ", stderr); + fputs("InnoDB: record ", stderr); rec_print(stderr, rec, index); fputs("\n" "InnoDB: clustered record ", stderr); @@ -1171,8 +1168,10 @@ row_ins_foreign_check_on_constraint( goto nonstandard_exit_func; } + if (table->fts) { - doc_id = fts_get_doc_id_from_rec(table, clust_rec, tmp_heap); + doc_id = fts_get_doc_id_from_rec(table, clust_rec, + clust_index, tmp_heap); } if (node->is_delete @@ -1181,6 +1180,14 @@ row_ins_foreign_check_on_constraint( /* Build the appropriate update vector which sets foreign->n_fields first fields in rec to SQL NULL */ + if (table->fts) { + + /* For the clause ON DELETE SET NULL, the cascade + operation is actually an update operation with the new + values being null. For FTS, this means that the old + values be deleted and no new values to be added.*/ + cascade->fts_next_doc_id = FTS_NULL_DOC_ID; + } update = cascade->update; @@ -1191,38 +1198,48 @@ row_ins_foreign_check_on_constraint( for (i = 0; i < foreign->n_fields; i++) { upd_field_t* ufield = &update->fields[i]; + ulint col_no = dict_index_get_nth_col_no( + index, i); + ulint prefix_col; ufield->field_no = dict_table_get_nth_col_pos( - table, - dict_index_get_nth_col_no(index, i)); + table, col_no, &prefix_col); + dict_col_t* col = dict_table_get_nth_col( + table, col_no); + dict_col_copy_type(col, dfield_get_type(&ufield->new_val)); + ufield->orig_len = 0; ufield->exp = NULL; dfield_set_null(&ufield->new_val); if (table->fts && dict_table_is_fts_column( table->fts->indexes, - dict_index_get_nth_col_no(index, i)) - != ULINT_UNDEFINED) { + dict_index_get_nth_col_no(index, i), + dict_col_is_virtual( + dict_index_get_nth_col(index, i))) + != ULINT_UNDEFINED) { fts_col_affacted = TRUE; } } if (fts_col_affacted) { - fts_trx_add_op(trx, table, doc_id, FTS_DELETE, NULL); + cascade->fts_doc_id = doc_id; } } else if (table->fts && cascade->is_delete) { /* DICT_FOREIGN_ON_DELETE_CASCADE case */ for (i = 0; i < foreign->n_fields; i++) { if (table->fts && dict_table_is_fts_column( table->fts->indexes, - dict_index_get_nth_col_no(index, i)) - != ULINT_UNDEFINED) { + dict_index_get_nth_col_no(index, i), + dict_col_is_virtual( + dict_index_get_nth_col(index, i))) + != ULINT_UNDEFINED) { fts_col_affacted = TRUE; } } if (fts_col_affacted) { - fts_trx_add_op(trx, table, doc_id, FTS_DELETE, NULL); + cascade->fts_doc_id = doc_id; } } @@ -1232,10 +1249,9 @@ row_ins_foreign_check_on_constraint( /* Build the appropriate update vector which sets changing foreign->n_fields first fields in rec to new values */ - upd_vec_heap = mem_heap_create(256); - n_to_update = row_ins_cascade_calc_update_vec( - node, foreign, upd_vec_heap, trx, &fts_col_affacted); + node, foreign, cascade->cascade_heap, + trx, &fts_col_affacted, cascade); if (n_to_update == ULINT_UNDEFINED) { err = DB_ROW_IS_REFERENCED; @@ -1266,7 +1282,7 @@ row_ins_foreign_check_on_constraint( /* Mark the old Doc ID as deleted */ if (fts_col_affacted) { ut_ad(table->fts); - fts_trx_add_op(trx, table, doc_id, FTS_DELETE, NULL); + cascade->fts_doc_id = doc_id; } } @@ -1299,16 +1315,11 @@ row_ins_foreign_check_on_constraint( "WSREP: foreign key append failed: %d\n", err); } else #endif /* WITH_WSREP */ - err = row_update_cascade_for_mysql(thr, cascade, - foreign->foreign_table); + node->new_upd_nodes->push_back(cascade); - if (foreign->foreign_table->n_foreign_key_checks_running == 0) { - fprintf(stderr, - "InnoDB: error: table %s has the counter 0" - " though there is\n" - "InnoDB: a FOREIGN KEY check running on it.\n", - foreign->foreign_table->name); - } + os_atomic_increment_ulint(&table->n_foreign_key_checks_running, 1); + + ut_ad(foreign->foreign_table->n_foreign_key_checks_running > 0); /* Release the data dictionary latch for a while, so that we do not starve other threads from doing CREATE TABLE etc. if we have a huge @@ -1322,7 +1333,7 @@ row_ins_foreign_check_on_constraint( row_mysql_freeze_data_dictionary(thr_get_trx(thr)); - mtr_start_trx(mtr, trx); + mtr_start(mtr); /* Restore pcur position */ @@ -1332,35 +1343,29 @@ row_ins_foreign_check_on_constraint( mem_heap_free(tmp_heap); } - if (upd_vec_heap) { - mem_heap_free(upd_vec_heap); - } - - return(err); + DBUG_RETURN(err); nonstandard_exit_func: + que_graph_free_recursive(cascade); + if (tmp_heap) { mem_heap_free(tmp_heap); } - if (upd_vec_heap) { - mem_heap_free(upd_vec_heap); - } - btr_pcur_store_position(pcur, mtr); mtr_commit(mtr); - mtr_start_trx(mtr, trx); + mtr_start(mtr); btr_pcur_restore_position(BTR_SEARCH_LEAF, pcur, mtr); - return(err); + DBUG_RETURN(err); } /*********************************************************************//** Sets a shared lock on a record. Used in locking possible duplicate key records and also in checking foreign key constraints. -@return DB_SUCCESS, DB_SUCCESS_LOCKED_REC, or error code */ +@return DB_SUCCESS, DB_SUCCESS_LOCKED_REC, or error code */ static dberr_t row_ins_set_shared_rec_lock( @@ -1391,7 +1396,7 @@ row_ins_set_shared_rec_lock( /*********************************************************************//** Sets a exclusive lock on a record. Used in locking possible duplicate key records -@return DB_SUCCESS, DB_SUCCESS_LOCKED_REC, or error code */ +@return DB_SUCCESS, DB_SUCCESS_LOCKED_REC, or error code */ static dberr_t row_ins_set_exclusive_rec_lock( @@ -1419,12 +1424,22 @@ row_ins_set_exclusive_rec_lock( return(err); } +/* Decrement a counter in the destructor. */ +class ib_dec_in_dtor { +public: + ib_dec_in_dtor(ulint& c): counter(c) {} + ~ib_dec_in_dtor() { + os_atomic_decrement_ulint(&counter, 1); + } +private: + ulint& counter; +}; + /***************************************************************//** Checks if foreign key constraint fails for an index entry. Sets shared locks which lock either the success or the failure of the constraint. NOTE that the caller must have a shared latch on dict_operation_lock. -@return DB_SUCCESS, DB_NO_REFERENCED_ROW, or DB_ROW_IS_REFERENCED */ -UNIV_INTERN +@return DB_SUCCESS, DB_NO_REFERENCED_ROW, or DB_ROW_IS_REFERENCED */ dberr_t row_ins_check_foreign_constraint( /*=============================*/ @@ -1446,18 +1461,17 @@ row_ins_check_foreign_constraint( ulint n_fields_cmp; btr_pcur_t pcur; int cmp; - ulint i; mtr_t mtr; trx_t* trx = thr_get_trx(thr); mem_heap_t* heap = NULL; ulint offsets_[REC_OFFS_NORMAL_SIZE]; ulint* offsets = offsets_; + + DBUG_ENTER("row_ins_check_foreign_constraint"); + rec_offs_init(offsets_); -run_again: -#ifdef UNIV_SYNC_DEBUG - ut_ad(rw_lock_own(&dict_operation_lock, RW_LOCK_SHARED)); -#endif /* UNIV_SYNC_DEBUG */ + ut_ad(rw_lock_own(dict_operation_lock, RW_LOCK_S)); err = DB_SUCCESS; @@ -1470,11 +1484,8 @@ run_again: /* If any of the foreign key fields in entry is SQL NULL, we suppress the foreign key check: this is compatible with Oracle, for example */ - - for (i = 0; i < foreign->n_fields; i++) { - if (UNIV_SQL_NULL == dfield_get_len( - dtuple_get_nth_field(entry, i))) { - + for (ulint i = 0; i < foreign->n_fields; i++) { + if (dfield_is_null(dtuple_get_nth_field(entry, i))) { goto exit_func; } } @@ -1524,19 +1535,17 @@ run_again: row_ins_foreign_trx_print(trx); fputs("Foreign key constraint fails for table ", ef); - ut_print_name(ef, trx, TRUE, + ut_print_name(ef, trx, foreign->foreign_table_name); fputs(":\n", ef); fk_str = dict_print_info_on_foreign_key_in_create_format( trx, foreign, TRUE); fputs(fk_str.c_str(), ef); - fputs("\nTrying to add to index ", ef); - ut_print_name(ef, trx, FALSE, - foreign->foreign_index->name); - fputs(" tuple:\n", ef); + fprintf(ef, "\nTrying to add to index %s tuple:\n", + foreign->foreign_index->name()); dtuple_print(ef, entry); fputs("\nBut the parent table ", ef); - ut_print_name(ef, trx, TRUE, + ut_print_name(ef, trx, foreign->referenced_table_name); fputs("\nor its .ibd file does" " not currently exist!\n", ef); @@ -1560,7 +1569,7 @@ run_again: } } - mtr_start_trx(&mtr, trx); + mtr_start(&mtr); /* Store old value on n_fields_cmp */ @@ -1726,55 +1735,43 @@ end_scan: do_possible_lock_wait: if (err == DB_LOCK_WAIT) { - bool verified = false; + /* An object that will correctly decrement the FK check counter + when it goes out of this scope. */ + ib_dec_in_dtor dec(check_table->n_foreign_key_checks_running); trx->error_state = err; que_thr_stop_for_mysql(thr); + thr->lock_state = QUE_THR_LOCK_ROW; + + /* To avoid check_table being dropped, increment counter */ + os_atomic_increment_ulint( + &check_table->n_foreign_key_checks_running, 1); + lock_wait_suspend_thread(thr); + thr->lock_state = QUE_THR_LOCK_NOLOCK; + + DBUG_PRINT("to_be_dropped", + ("table: %s", check_table->name.m_name)); if (check_table->to_be_dropped) { /* The table is being dropped. We shall timeout this operation */ err = DB_LOCK_WAIT_TIMEOUT; - goto exit_func; - } - /* We had temporarily released dict_operation_lock in - above lock sleep wait, now we have the lock again, and - we will need to re-check whether the foreign key has been - dropped. We only need to verify if the table is referenced - table case (check_ref == 0), since MDL lock will prevent - concurrent DDL and DML on the same table */ - if (!check_ref) { - for (dict_foreign_set::iterator it - = table->referenced_set.begin(); - it != table->referenced_set.end(); - ++it) { - if (*it == foreign) { - verified = true; - break; - } - } - } else { - verified = true; + goto exit_func; } - if (!verified) { - err = DB_DICT_CHANGED; - } else if (trx->error_state == DB_SUCCESS) { - goto run_again; - } else { - err = trx->error_state; - } } + exit_func: - if (UNIV_LIKELY_NULL(heap)) { + if (heap != NULL) { mem_heap_free(heap); } - return(err); + + DBUG_RETURN(err); } /***************************************************************//** @@ -1783,7 +1780,7 @@ is not mentioned in any constraint, this function does nothing, Otherwise does searches to the indexes of referenced tables and sets shared locks which lock either the success or the failure of a constraint. -@return DB_SUCCESS or error code */ +@return DB_SUCCESS or error code */ static MY_ATTRIBUTE((nonnull, warn_unused_result)) dberr_t row_ins_check_foreign_constraints( @@ -1811,7 +1808,6 @@ row_ins_check_foreign_constraints( if (foreign->foreign_index == index) { dict_table_t* ref_table = NULL; - dict_table_t* foreign_table = foreign->foreign_table; dict_table_t* referenced_table = foreign->referenced_table; @@ -1828,12 +1824,6 @@ row_ins_check_foreign_constraints( row_mysql_freeze_data_dictionary(trx); } - if (referenced_table) { - os_inc_counter(dict_sys->mutex, - foreign_table - ->n_foreign_key_checks_running); - } - /* NOTE that if the thread ends up waiting for a lock we will release dict_operation_lock temporarily! But the counter on the table protects the referenced @@ -1842,15 +1832,6 @@ row_ins_check_foreign_constraints( err = row_ins_check_foreign_constraint( TRUE, foreign, table, entry, thr); - DBUG_EXECUTE_IF("row_ins_dict_change_err", - err = DB_DICT_CHANGED;); - - if (referenced_table) { - os_dec_counter(dict_sys->mutex, - foreign_table - ->n_foreign_key_checks_running); - } - if (got_s_lock) { row_mysql_unfreeze_data_dictionary(trx); } @@ -1872,7 +1853,7 @@ row_ins_check_foreign_constraints( /***************************************************************//** Checks if a unique key violation to rec would occur at the index entry insert. -@return TRUE if error */ +@return TRUE if error */ static ibool row_ins_dupl_error_with_rec( @@ -1885,7 +1866,6 @@ row_ins_dupl_error_with_rec( const ulint* offsets)/*!< in: rec_get_offsets(rec, index) */ { ulint matched_fields; - ulint matched_bytes; ulint n_unique; ulint i; @@ -1894,10 +1874,8 @@ row_ins_dupl_error_with_rec( n_unique = dict_index_get_n_unique(index); matched_fields = 0; - matched_bytes = 0; - cmp_dtuple_rec_with_match(entry, rec, offsets, - &matched_fields, &matched_bytes); + cmp_dtuple_rec_with_match(entry, rec, offsets, &matched_fields); if (matched_fields < n_unique) { @@ -1907,7 +1885,7 @@ row_ins_dupl_error_with_rec( /* In a unique secondary index we allow equal key values if they contain SQL NULLs */ - if (!dict_index_is_clust(index)) { + if (!dict_index_is_clust(index) && !index->nulls_equal) { for (i = 0; i < n_unique; i++) { if (dfield_is_null(dtuple_get_nth_field(entry, i))) { @@ -1924,7 +1902,7 @@ row_ins_dupl_error_with_rec( Scans a unique non-clustered index at a given index entry to determine whether a uniqueness violation has occurred for the key value of the entry. Set shared locks on possible duplicate records. -@return DB_SUCCESS, DB_DUPLICATE_KEY, or DB_LOCK_WAIT */ +@return DB_SUCCESS, DB_DUPLICATE_KEY, or DB_LOCK_WAIT */ static MY_ATTRIBUTE((nonnull, warn_unused_result)) dberr_t row_ins_scan_sec_index_for_duplicate( @@ -1945,10 +1923,11 @@ row_ins_scan_sec_index_for_duplicate( dberr_t err = DB_SUCCESS; ulint allow_duplicates; ulint* offsets = NULL; + DBUG_ENTER("row_ins_scan_sec_index_for_duplicate"); -#ifdef UNIV_SYNC_DEBUG - ut_ad(s_latch == rw_lock_own(&index->lock, RW_LOCK_SHARED)); -#endif /* UNIV_SYNC_DEBUG */ + + ut_ad(s_latch == rw_lock_own_flagged( + &index->lock, RW_LOCK_FLAG_S | RW_LOCK_FLAG_SX)); n_unique = dict_index_get_n_unique(index); @@ -1956,11 +1935,13 @@ row_ins_scan_sec_index_for_duplicate( n_unique first fields is NULL, a unique key violation cannot occur, since we define NULL != NULL in this case */ - for (ulint i = 0; i < n_unique; i++) { - if (UNIV_SQL_NULL == dfield_get_len( - dtuple_get_nth_field(entry, i))) { + if (!index->nulls_equal) { + for (ulint i = 0; i < n_unique; i++) { + if (UNIV_SQL_NULL == dfield_get_len( + dtuple_get_nth_field(entry, i))) { - return(DB_SUCCESS); + DBUG_RETURN(DB_SUCCESS); + } } } @@ -2027,7 +2008,7 @@ row_ins_scan_sec_index_for_duplicate( cmp = cmp_dtuple_rec(entry, rec, offsets); - if (cmp == 0) { + if (cmp == 0 && !index->allow_duplicates) { if (row_ins_dupl_error_with_rec(rec, entry, index, offsets)) { err = DB_DUPLICATE_KEY; @@ -2036,21 +2017,20 @@ row_ins_scan_sec_index_for_duplicate( /* If the duplicate is on hidden FTS_DOC_ID, state so in the error log */ - if (DICT_TF2_FLAG_IS_SET( + if (index == index->table->fts_doc_id_index + && DICT_TF2_FLAG_IS_SET( index->table, - DICT_TF2_FTS_HAS_DOC_ID) - && strcmp(index->name, - FTS_DOC_ID_INDEX_NAME) == 0) { - ib_logf(IB_LOG_LEVEL_ERROR, - "Duplicate FTS_DOC_ID value" - " on table %s", - index->table->name); + DICT_TF2_FTS_HAS_DOC_ID)) { + + ib::error() << "Duplicate FTS_DOC_ID" + " value on table " + << index->table->name; } goto end_scan; } } else { - ut_a(cmp < 0); + ut_a(cmp < 0 || index->allow_duplicates); goto end_scan; } } while (btr_pcur_move_to_next(&pcur, mtr)); @@ -2059,14 +2039,14 @@ end_scan: /* Restore old value */ dtuple_set_n_fields_cmp(entry, n_fields_cmp); - return(err); + DBUG_RETURN(err); } /** Checks for a duplicate when the table is being rebuilt online. -@retval DB_SUCCESS when no duplicate is detected -@retval DB_SUCCESS_LOCKED_REC when rec is an exact match of entry or +@retval DB_SUCCESS when no duplicate is detected +@retval DB_SUCCESS_LOCKED_REC when rec is an exact match of entry or a newer version of entry (the entry should not be inserted) -@retval DB_DUPLICATE_KEY when entry is a duplicate of rec */ +@retval DB_DUPLICATE_KEY when entry is a duplicate of rec */ static MY_ATTRIBUTE((nonnull, warn_unused_result)) dberr_t row_ins_duplicate_online( @@ -2077,7 +2057,6 @@ row_ins_duplicate_online( ulint* offsets)/*!< in/out: rec_get_offsets(rec) */ { ulint fields = 0; - ulint bytes = 0; /* During rebuild, there should not be any delete-marked rows in the new table. */ @@ -2087,7 +2066,7 @@ row_ins_duplicate_online( /* Compare the PRIMARY KEY fields and the DB_TRX_ID, DB_ROLL_PTR. */ cmp_dtuple_rec_with_match_low( - entry, rec, offsets, n_uniq + 2, &fields, &bytes); + entry, rec, offsets, n_uniq + 2, &fields); if (fields < n_uniq) { /* Not a duplicate. */ @@ -2096,7 +2075,6 @@ row_ins_duplicate_online( if (fields == n_uniq + 2) { /* rec is an exact match of entry. */ - ut_ad(bytes == 0); return(DB_SUCCESS_LOCKED_REC); } @@ -2104,10 +2082,10 @@ row_ins_duplicate_online( } /** Checks for a duplicate when the table is being rebuilt online. -@retval DB_SUCCESS when no duplicate is detected -@retval DB_SUCCESS_LOCKED_REC when rec is an exact match of entry or +@retval DB_SUCCESS when no duplicate is detected +@retval DB_SUCCESS_LOCKED_REC when rec is an exact match of entry or a newer version of entry (the entry should not be inserted) -@retval DB_DUPLICATE_KEY when entry is a duplicate of rec */ +@retval DB_DUPLICATE_KEY when entry is a duplicate of rec */ static MY_ATTRIBUTE((nonnull, warn_unused_result)) dberr_t row_ins_duplicate_error_in_clust_online( @@ -2197,12 +2175,21 @@ row_ins_duplicate_error_in_clust( offsets = rec_get_offsets(rec, cursor->index, offsets, ULINT_UNDEFINED, &heap); + ulint lock_type; + + lock_type = + trx->isolation_level <= TRX_ISO_READ_COMMITTED + ? LOCK_REC_NOT_GAP : LOCK_ORDINARY; + /* We set a lock on the possible duplicate: this is needed in logical logging of MySQL to make sure that in roll-forward we get the same duplicate errors as in original execution */ - if (trx->duplicates) { + if (flags & BTR_NO_LOCKING_FLAG) { + /* Do nothing if no-locking is set */ + err = DB_SUCCESS; + } else if (trx->duplicates) { /* If the SQL-query will update or replace duplicate key we will take X-lock for @@ -2210,13 +2197,13 @@ row_ins_duplicate_error_in_clust( INSERT ON DUPLICATE KEY UPDATE). */ err = row_ins_set_exclusive_rec_lock( - LOCK_REC_NOT_GAP, + lock_type, btr_cur_get_block(cursor), rec, cursor->index, offsets, thr); } else { err = row_ins_set_shared_rec_lock( - LOCK_REC_NOT_GAP, + lock_type, btr_cur_get_block(cursor), rec, cursor->index, offsets, thr); } @@ -2332,7 +2319,6 @@ the delete marked record. @retval DB_LOCK_WAIT on lock wait when !(flags & BTR_NO_LOCKING_FLAG) @retval DB_FAIL if retry with BTR_MODIFY_TREE is needed @return error code */ -UNIV_INTERN dberr_t row_ins_clust_index_entry_low( /*==========================*/ @@ -2344,57 +2330,84 @@ row_ins_clust_index_entry_low( ulint n_uniq, /*!< in: 0 or index->n_uniq */ dtuple_t* entry, /*!< in/out: index entry to insert */ ulint n_ext, /*!< in: number of externally stored columns */ - que_thr_t* thr) /*!< in: query thread */ + que_thr_t* thr, /*!< in: query thread */ + bool dup_chk_only) + /*!< in: if true, just do duplicate check + and return. don't execute actual insert. */ { - btr_cur_t cursor; - ulint* offsets = NULL; + btr_pcur_t pcur; + btr_cur_t* cursor; dberr_t err = DB_SUCCESS; big_rec_t* big_rec = NULL; mtr_t mtr; mem_heap_t* offsets_heap = NULL; + ulint offsets_[REC_OFFS_NORMAL_SIZE]; + ulint* offsets = offsets_; + rec_offs_init(offsets_); + + DBUG_ENTER("row_ins_clust_index_entry_low"); ut_ad(dict_index_is_clust(index)); ut_ad(!dict_index_is_unique(index) || n_uniq == dict_index_get_n_unique(index)); ut_ad(!n_uniq || n_uniq == dict_index_get_n_unique(index)); + ut_ad(!thr_get_trx(thr)->in_rollback); + + mtr_start(&mtr); + mtr.set_named_space(index->space); + + if (dict_table_is_temporary(index->table)) { + /* Disable REDO logging as the lifetime of temp-tables is + limited to server or connection lifetime and so REDO + information is not needed on restart for recovery. + Disable locking as temp-tables are local to a connection. */ - mtr_start_trx(&mtr, thr_get_trx(thr)); + ut_ad(flags & BTR_NO_LOCKING_FLAG); + ut_ad(!dict_table_is_intrinsic(index->table) + || (flags & BTR_NO_UNDO_LOG_FLAG)); + + mtr.set_log_mode(MTR_LOG_NO_REDO); + } if (mode == BTR_MODIFY_LEAF && dict_index_is_online_ddl(index)) { mode = BTR_MODIFY_LEAF | BTR_ALREADY_S_LATCHED; mtr_s_lock(dict_index_get_lock(index), &mtr); } - cursor.thr = thr; - /* Note that we use PAGE_CUR_LE as the search mode, because then the function will return in both low_match and up_match of the cursor sensible values */ + btr_pcur_open(index, entry, PAGE_CUR_LE, mode, &pcur, &mtr); + cursor = btr_pcur_get_btr_cur(&pcur); - err = btr_cur_search_to_nth_level(index, 0, entry, PAGE_CUR_LE, mode, - &cursor, 0, __FILE__, __LINE__, &mtr); - - if (err != DB_SUCCESS) { - index->table->is_encrypted = true; - index->table->ibd_file_missing = true; - mtr_commit(&mtr); - goto func_exit; + if (cursor) { + cursor->thr = thr; } + ut_ad(!dict_table_is_intrinsic(index->table) + || cursor->page_cur.block->made_dirty_with_no_latch); + #ifdef UNIV_DEBUG { - page_t* page = btr_cur_get_page(&cursor); + page_t* page = btr_cur_get_page(cursor); rec_t* first_rec = page_rec_get_next( page_get_infimum_rec(page)); ut_ad(page_rec_is_supremum(first_rec) - || rec_get_n_fields(first_rec, index) - == dtuple_get_n_fields(entry)); + || rec_n_fields_is_sane(index, first_rec, entry)); } -#endif +#endif /* UNIV_DEBUG */ - if (n_uniq && (cursor.up_match >= n_uniq - || cursor.low_match >= n_uniq)) { + /* Allowing duplicates in clustered index is currently enabled + only for intrinsic table and caller understand the limited + operation that can be done in this case. */ + ut_ad(!index->allow_duplicates + || (index->allow_duplicates + && dict_table_is_intrinsic(index->table))); + + if (!index->allow_duplicates + && n_uniq + && (cursor->up_match >= n_uniq || cursor->low_match >= n_uniq)) { if (flags == (BTR_CREATE_FLAG | BTR_NO_LOCKING_FLAG @@ -2402,7 +2415,7 @@ row_ins_clust_index_entry_low( /* Set no locks when applying log in online table rebuild. Only check for duplicates. */ err = row_ins_duplicate_error_in_clust_online( - n_uniq, entry, &cursor, + n_uniq, entry, cursor, &offsets, &offsets_heap); switch (err) { @@ -2413,14 +2426,14 @@ row_ins_clust_index_entry_low( /* fall through */ case DB_SUCCESS_LOCKED_REC: case DB_DUPLICATE_KEY: - thr_get_trx(thr)->error_info = cursor.index; + thr_get_trx(thr)->error_info = cursor->index; } } else { /* Note that the following may return also DB_LOCK_WAIT */ err = row_ins_duplicate_error_in_clust( - flags, &cursor, entry, thr, &mtr); + flags, cursor, entry, thr, &mtr); } if (err != DB_SUCCESS) { @@ -2430,77 +2443,33 @@ err_exit: } } - if (row_ins_must_modify_rec(&cursor)) { + if (dup_chk_only) { + mtr_commit(&mtr); + goto func_exit; + } + + /* Note: Allowing duplicates would qualify for modification of + an existing record as the new entry is exactly same as old entry. + Avoid this check if allow duplicates is enabled. */ + if (!index->allow_duplicates && row_ins_must_modify_rec(cursor)) { /* There is already an index entry with a long enough common prefix, we must convert the insert into a modify of an existing record */ mem_heap_t* entry_heap = mem_heap_create(1024); - err = row_ins_clust_index_entry_by_modify( - flags, mode, &cursor, &offsets, &offsets_heap, - entry_heap, &big_rec, entry, thr, &mtr); - - rec_t* rec = btr_cur_get_rec(&cursor); + /* If the existing record is being modified and the new record + doesn't fit the provided slot then existing record is added + to free list and new record is inserted. This also means + cursor that we have cached for SELECT is now invalid. */ + index->last_sel_cur->invalid = true; - if (big_rec) { - ut_a(err == DB_SUCCESS); - /* Write out the externally stored - columns while still x-latching - index->lock and block->lock. Allocate - pages for big_rec in the mtr that - modified the B-tree, but be sure to skip - any pages that were freed in mtr. We will - write out the big_rec pages before - committing the B-tree mini-transaction. If - the system crashes so that crash recovery - will not replay the mtr_commit(&mtr), the - big_rec pages will be left orphaned until - the pages are allocated for something else. - - TODO: If the allocation extends the - tablespace, it will not be redo - logged, in either mini-transaction. - Tablespace extension should be - redo-logged in the big_rec - mini-transaction, so that recovery - will not fail when the big_rec was - written to the extended portion of the - file, in case the file was somehow - truncated in the crash. */ - - DEBUG_SYNC_C_IF_THD( - thr_get_trx(thr)->mysql_thd, - "before_row_ins_upd_extern"); - err = btr_store_big_rec_extern_fields( - index, btr_cur_get_block(&cursor), - rec, offsets, big_rec, &mtr, - BTR_STORE_INSERT_UPDATE); - DEBUG_SYNC_C_IF_THD( - thr_get_trx(thr)->mysql_thd, - "after_row_ins_upd_extern"); - /* If writing big_rec fails (for - example, because of DB_OUT_OF_FILE_SPACE), - the record will be corrupted. Even if - we did not update any externally - stored columns, our update could cause - the record to grow so that a - non-updated column was selected for - external storage. This non-update - would not have been written to the - undo log, and thus the record cannot - be rolled back. - - However, because we have not executed - mtr_commit(mtr) yet, the update will - not be replayed in crash recovery, and - the following assertion failure will - effectively "roll back" the operation. */ - ut_a(err == DB_SUCCESS); - dtuple_big_rec_free(big_rec); - } + err = row_ins_clust_index_entry_by_modify( + &pcur, flags, mode, &offsets, &offsets_heap, + entry_heap, entry, thr, &mtr); if (err == DB_SUCCESS && dict_index_is_online_ddl(index)) { - row_log_table_insert(rec, index, offsets); + row_log_table_insert(btr_cur_get_rec(cursor), entry, + index, offsets); } mtr_commit(&mtr); @@ -2512,7 +2481,7 @@ err_exit: ut_ad((mode & ~BTR_ALREADY_S_LATCHED) == BTR_MODIFY_LEAF); err = btr_cur_optimistic_insert( - flags, &cursor, &offsets, &offsets_heap, + flags, cursor, &offsets, &offsets_heap, entry, &insert_rec, &big_rec, n_ext, thr, &mtr); } else { @@ -2522,22 +2491,24 @@ err_exit: goto err_exit; } + DEBUG_SYNC_C("before_insert_pessimitic_row_ins_clust"); + err = btr_cur_optimistic_insert( - flags, &cursor, + flags, cursor, &offsets, &offsets_heap, entry, &insert_rec, &big_rec, n_ext, thr, &mtr); if (err == DB_FAIL) { err = btr_cur_pessimistic_insert( - flags, &cursor, + flags, cursor, &offsets, &offsets_heap, entry, &insert_rec, &big_rec, n_ext, thr, &mtr); } } - if (UNIV_LIKELY_NULL(big_rec)) { + if (big_rec != NULL) { mtr_commit(&mtr); /* Online table rebuild could read (and @@ -2558,7 +2529,7 @@ err_exit: if (err == DB_SUCCESS && dict_index_is_online_ddl(index)) { row_log_table_insert( - insert_rec, index, offsets); + insert_rec, entry, index, offsets); } mtr_commit(&mtr); @@ -2566,30 +2537,178 @@ err_exit: } func_exit: - if (offsets_heap) { + if (offsets_heap != NULL) { mem_heap_free(offsets_heap); } - return(err); + btr_pcur_close(&pcur); + + DBUG_RETURN(err); } -/***************************************************************//** -Starts a mini-transaction and checks if the index will be dropped. +/** This is a specialized function meant for direct insertion to +auto-generated clustered index based on cached position from +last successful insert. To be used when data is sorted. + +@param[in] mode BTR_MODIFY_LEAF or BTR_MODIFY_TREE. + depending on whether we wish optimistic or + pessimistic descent down the index tree +@param[in,out] index clustered index +@param[in,out] entry index entry to insert +@param[in] thr query thread + +@return error code */ +static +dberr_t +row_ins_sorted_clust_index_entry( + ulint mode, + dict_index_t* index, + dtuple_t* entry, + ulint n_ext, + que_thr_t* thr) +{ + dberr_t err = DB_SUCCESS; + mtr_t* mtr; + const bool commit_mtr = mode == BTR_MODIFY_TREE; + + mem_heap_t* offsets_heap = NULL; + ulint offsets_[REC_OFFS_NORMAL_SIZE]; + ulint* offsets = offsets_; + rec_offs_init(offsets_); + + DBUG_ENTER("row_ins_sorted_clust_index_entry"); + + ut_ad(index->last_ins_cur != NULL); + ut_ad(dict_index_is_clust(index)); + ut_ad(dict_table_is_intrinsic(index->table)); + ut_ad(dict_index_is_auto_gen_clust(index)); + + btr_cur_t cursor; + cursor.thr = thr; + mtr = &index->last_ins_cur->mtr; + + /* Search for position if tree needs to be split or if last position + is not cached. */ + if (mode == BTR_MODIFY_TREE + || index->last_ins_cur->rec == NULL + || index->last_ins_cur->disable_caching) { + + /* Commit the previous mtr. */ + index->last_ins_cur->release(); + + mtr_start(mtr); + mtr_set_log_mode(mtr, MTR_LOG_NO_REDO); + + err = btr_cur_search_to_nth_level_with_no_latch( + index, 0, entry, PAGE_CUR_LE, &cursor, + __FILE__, __LINE__, mtr); + ut_ad(cursor.page_cur.block != NULL); + ut_ad(cursor.page_cur.block->made_dirty_with_no_latch); + } else { + cursor.index = index; + + cursor.page_cur.index = index; + + cursor.page_cur.rec = index->last_ins_cur->rec; + + cursor.page_cur.block = index->last_ins_cur->block; + } + + const ulint flags = BTR_NO_LOCKING_FLAG | BTR_NO_UNDO_LOG_FLAG; + + for (;;) { + rec_t* insert_rec; + big_rec_t* big_rec = NULL; + + if (mode != BTR_MODIFY_TREE) { + ut_ad((mode & ~BTR_ALREADY_S_LATCHED) + == BTR_MODIFY_LEAF); + + err = btr_cur_optimistic_insert( + flags, &cursor, &offsets, &offsets_heap, entry, + &insert_rec, &big_rec, n_ext, thr, mtr); + if (err != DB_SUCCESS) { + break; + } + } else { + /* TODO: Check if this is needed for intrinsic table. */ + if (buf_LRU_buf_pool_running_out()) { + err = DB_LOCK_TABLE_FULL; + break; + } + + err = btr_cur_optimistic_insert( + flags, &cursor, &offsets, &offsets_heap, entry, + &insert_rec, &big_rec, n_ext, thr, mtr); + + if (err == DB_FAIL) { + err = btr_cur_pessimistic_insert( + flags, &cursor, &offsets, &offsets_heap, + entry, &insert_rec, &big_rec, n_ext, + thr, mtr); + } + } + + if (big_rec != NULL) { + /* If index involves big-record optimization is + turned-off. */ + index->last_ins_cur->release(); + index->last_ins_cur->disable_caching = true; + + err = row_ins_index_entry_big_rec( + entry, big_rec, offsets, &offsets_heap, index, + thr_get_trx(thr)->mysql_thd, __FILE__, __LINE__); + + dtuple_convert_back_big_rec(index, entry, big_rec); + + } else if (err == DB_SUCCESS ) { + if (!commit_mtr + && !index->last_ins_cur->disable_caching) { + index->last_ins_cur->rec = insert_rec; + + index->last_ins_cur->block + = cursor.page_cur.block; + } else { + index->last_ins_cur->release(); + } + } + + break; + } + + if (err != DB_SUCCESS) { + index->last_ins_cur->release(); + } + + if (offsets_heap != NULL) { + mem_heap_free(offsets_heap); + } + + DBUG_RETURN(err); +} + +/** Start a mini-transaction and check if the index will be dropped. +@param[in,out] mtr mini-transaction +@param[in,out] index secondary index +@param[in] check whether to check +@param[in] search_mode flags @return true if the index is to be dropped */ static MY_ATTRIBUTE((nonnull, warn_unused_result)) bool -row_ins_sec_mtr_start_trx_and_check_if_aborted( -/*=======================================*/ - mtr_t* mtr, /*!< out: mini-transaction */ - trx_t* trx, /*!< in: transaction handle */ - dict_index_t* index, /*!< in/out: secondary index */ - bool check, /*!< in: whether to check */ +row_ins_sec_mtr_start_and_check_if_aborted( + mtr_t* mtr, + dict_index_t* index, + bool check, ulint search_mode) - /*!< in: flags */ { ut_ad(!dict_index_is_clust(index)); + ut_ad(mtr->is_named_space(index->space)); + + const mtr_log_t log_mode = mtr->get_log_mode(); - mtr_start_trx(mtr, trx); + mtr_start(mtr); + mtr->set_named_space(index->space); + mtr->set_log_mode(log_mode); if (!check) { return(false); @@ -2598,13 +2717,13 @@ row_ins_sec_mtr_start_trx_and_check_if_aborted( if (search_mode & BTR_ALREADY_S_LATCHED) { mtr_s_lock(dict_index_get_lock(index), mtr); } else { - mtr_x_lock(dict_index_get_lock(index), mtr); + mtr_sx_lock(dict_index_get_lock(index), mtr); } switch (index->online_status) { case ONLINE_INDEX_ABORTED: case ONLINE_INDEX_ABORTED_DROPPED: - ut_ad(*index->name == TEMP_INDEX_PREFIX); + ut_ad(!index->is_committed()); return(true); case ONLINE_INDEX_COMPLETE: return(false); @@ -2624,7 +2743,6 @@ It is then unmarked. Otherwise, the entry is just inserted to the index. @retval DB_LOCK_WAIT on lock wait when !(flags & BTR_NO_LOCKING_FLAG) @retval DB_FAIL if retry with BTR_MODIFY_TREE is needed @return error code */ -UNIV_INTERN dberr_t row_ins_sec_index_entry_low( /*========================*/ @@ -2639,22 +2757,50 @@ row_ins_sec_index_entry_low( dtuple_t* entry, /*!< in/out: index entry to insert */ trx_id_t trx_id, /*!< in: PAGE_MAX_TRX_ID during row_log_table_apply(), or 0 */ - que_thr_t* thr) /*!< in: query thread */ + que_thr_t* thr, /*!< in: query thread */ + bool dup_chk_only) + /*!< in: if true, just do duplicate check + and return. don't execute actual insert. */ { + DBUG_ENTER("row_ins_sec_index_entry_low"); + btr_cur_t cursor; - ulint search_mode = mode | BTR_INSERT; + ulint search_mode = mode; dberr_t err = DB_SUCCESS; ulint n_unique; mtr_t mtr; - ulint* offsets = NULL; - trx_t* trx = thr_get_trx(thr); + ulint offsets_[REC_OFFS_NORMAL_SIZE]; + ulint* offsets = offsets_; + rec_offs_init(offsets_); + rtr_info_t rtr_info; ut_ad(!dict_index_is_clust(index)); ut_ad(mode == BTR_MODIFY_LEAF || mode == BTR_MODIFY_TREE); cursor.thr = thr; - ut_ad(thr_get_trx(thr)->id); - mtr_start_trx(&mtr, trx); + cursor.rtr_info = NULL; + ut_ad(thr_get_trx(thr)->id != 0 + || dict_table_is_intrinsic(index->table)); + + mtr_start(&mtr); + mtr.set_named_space(index->space); + + if (dict_table_is_temporary(index->table)) { + /* Disable REDO logging as the lifetime of temp-tables is + limited to server or connection lifetime and so REDO + information is not needed on restart for recovery. + Disable locking as temp-tables are local to a connection. */ + + ut_ad(flags & BTR_NO_LOCKING_FLAG); + ut_ad(!dict_table_is_intrinsic(index->table) + || (flags & BTR_NO_UNDO_LOG_FLAG)); + + mtr.set_log_mode(MTR_LOG_NO_REDO); + } else if (!dict_index_is_spatial(index)) { + /* Enable insert buffering if it's neither temp-table + nor spatial index. */ + search_mode |= BTR_INSERT; + } /* Ensure that we acquire index->lock when inserting into an index with index->online_status == ONLINE_INDEX_COMPLETE, but @@ -2662,14 +2808,14 @@ row_ins_sec_index_entry_low( This prevents a concurrent change of index->online_status. The memory object cannot be freed as long as we have an open reference to the table, or index->table->n_ref_count > 0. */ - const bool check = *index->name == TEMP_INDEX_PREFIX; + const bool check = !index->is_committed(); if (check) { DEBUG_SYNC_C("row_ins_sec_index_enter"); if (mode == BTR_MODIFY_LEAF) { search_mode |= BTR_ALREADY_S_LATCHED; mtr_s_lock(dict_index_get_lock(index), &mtr); } else { - mtr_x_lock(dict_index_get_lock(index), &mtr); + mtr_sx_lock(dict_index_get_lock(index), &mtr); } if (row_log_online_op_try( @@ -2686,11 +2832,55 @@ row_ins_sec_index_entry_low( search_mode |= BTR_IGNORE_SEC_UNIQUE; } - err = btr_cur_search_to_nth_level(index, 0, entry, PAGE_CUR_LE, - search_mode, - &cursor, 0, __FILE__, __LINE__, &mtr); + if (dict_index_is_spatial(index)) { + cursor.index = index; + rtr_init_rtr_info(&rtr_info, false, &cursor, index, false); + rtr_info_update_btr(&cursor, &rtr_info); + + btr_cur_search_to_nth_level( + index, 0, entry, PAGE_CUR_RTREE_INSERT, + search_mode, + &cursor, 0, __FILE__, __LINE__, &mtr); + + if (mode == BTR_MODIFY_LEAF && rtr_info.mbr_adj) { + mtr_commit(&mtr); + rtr_clean_rtr_info(&rtr_info, true); + rtr_init_rtr_info(&rtr_info, false, &cursor, + index, false); + rtr_info_update_btr(&cursor, &rtr_info); + mtr_start(&mtr); + mtr.set_named_space(index->space); + search_mode &= ~BTR_MODIFY_LEAF; + search_mode |= BTR_MODIFY_TREE; + btr_cur_search_to_nth_level( + index, 0, entry, PAGE_CUR_RTREE_INSERT, + search_mode, + &cursor, 0, __FILE__, __LINE__, &mtr); + mode = BTR_MODIFY_TREE; + } + + DBUG_EXECUTE_IF( + "rtree_test_check_count", { + goto func_exit;}); + + } else { + if (dict_table_is_intrinsic(index->table)) { + err = btr_cur_search_to_nth_level_with_no_latch( + index, 0, entry, PAGE_CUR_LE, &cursor, + __FILE__, __LINE__, &mtr); + ut_ad(cursor.page_cur.block != NULL); + ut_ad(cursor.page_cur.block->made_dirty_with_no_latch); + } else { + btr_cur_search_to_nth_level( + index, 0, entry, PAGE_CUR_LE, + search_mode, + &cursor, 0, __FILE__, __LINE__, &mtr); + } + } if (err != DB_SUCCESS) { + trx_t* trx = thr_get_trx(thr); + if (err == DB_DECRYPTION_FAILED) { ib_push_warning(trx->mysql_thd, DB_DECRYPTION_FAILED, @@ -2704,6 +2894,7 @@ row_ins_sec_index_entry_low( } if (cursor.flag == BTR_CUR_INSERT_TO_IBUF) { + ut_ad(!dict_index_is_spatial(index)); /* The insert was buffered during the search: we are done */ goto func_exit; } @@ -2715,10 +2906,9 @@ row_ins_sec_index_entry_low( page_get_infimum_rec(page)); ut_ad(page_rec_is_supremum(first_rec) - || rec_get_n_fields(first_rec, index) - == dtuple_get_n_fields(entry)); + || rec_n_fields_is_sane(index, first_rec, entry)); } -#endif +#endif /* UNIV_DEBUG */ n_unique = dict_index_get_n_unique(index); @@ -2728,8 +2918,8 @@ row_ins_sec_index_entry_low( DEBUG_SYNC_C("row_ins_sec_index_unique"); - if (row_ins_sec_mtr_start_trx_and_check_if_aborted( - &mtr, trx, index, check, search_mode)) { + if (row_ins_sec_mtr_start_and_check_if_aborted( + &mtr, index, check, search_mode)) { goto func_exit; } @@ -2742,12 +2932,11 @@ row_ins_sec_index_entry_low( case DB_SUCCESS: break; case DB_DUPLICATE_KEY: - if (*index->name == TEMP_INDEX_PREFIX) { + if (!index->is_committed()) { ut_ad(!thr_get_trx(thr) ->dict_operation_lock_mode); mutex_enter(&dict_sys->mutex); - dict_set_corrupted_index_cache_only( - index, index->table); + dict_set_corrupted_index_cache_only(index); mutex_exit(&dict_sys->mutex); /* Do not return any error to the caller. The duplicate will be reported @@ -2760,11 +2949,14 @@ row_ins_sec_index_entry_low( } /* fall through */ default: - return(err); + if (dict_index_is_spatial(index)) { + rtr_clean_rtr_info(&rtr_info, true); + } + DBUG_RETURN(err); } - if (row_ins_sec_mtr_start_trx_and_check_if_aborted( - &mtr, trx, index, check, search_mode)) { + if (row_ins_sec_mtr_start_and_check_if_aborted( + &mtr, index, check, search_mode)) { goto func_exit; } @@ -2775,14 +2967,70 @@ row_ins_sec_index_entry_low( prevent any insertion of a duplicate by another transaction. Let us now reposition the cursor and continue the insertion. */ + if (dict_table_is_intrinsic(index->table)) { + err = btr_cur_search_to_nth_level_with_no_latch( + index, 0, entry, PAGE_CUR_LE, &cursor, + __FILE__, __LINE__, &mtr); + ut_ad(cursor.page_cur.block != NULL); + ut_ad(cursor.page_cur.block->made_dirty_with_no_latch); + } else { + btr_cur_search_to_nth_level( + index, 0, entry, PAGE_CUR_LE, + (search_mode + & ~(BTR_INSERT | BTR_IGNORE_SEC_UNIQUE)), + &cursor, 0, __FILE__, __LINE__, &mtr); + } + } - btr_cur_search_to_nth_level( - index, 0, entry, PAGE_CUR_LE, - search_mode & ~(BTR_INSERT | BTR_IGNORE_SEC_UNIQUE), - &cursor, 0, __FILE__, __LINE__, &mtr); + if (!(flags & BTR_NO_LOCKING_FLAG) + && dict_index_is_unique(index) + && thr_get_trx(thr)->duplicates + && thr_get_trx(thr)->isolation_level >= TRX_ISO_REPEATABLE_READ) { + + /* When using the REPLACE statement or ON DUPLICATE clause, a + gap lock is taken on the position of the to-be-inserted record, + to avoid other concurrent transactions from inserting the same + record. */ + + dberr_t err; + const rec_t* rec = page_rec_get_next_const( + btr_cur_get_rec(&cursor)); + + ut_ad(!page_rec_is_infimum(rec)); + + offsets = rec_get_offsets(rec, index, offsets, + ULINT_UNDEFINED, &offsets_heap); + + err = row_ins_set_exclusive_rec_lock( + LOCK_GAP, btr_cur_get_block(&cursor), rec, + index, offsets, thr); + + switch (err) { + case DB_SUCCESS: + case DB_SUCCESS_LOCKED_REC: + if (thr_get_trx(thr)->error_state != DB_DUPLICATE_KEY) { + break; + } + /* Fall through (skip actual insert) after we have + successfully acquired the gap lock. */ + default: + goto func_exit; + } + } + + ut_ad(thr_get_trx(thr)->error_state == DB_SUCCESS); + + if (dup_chk_only) { + goto func_exit; } if (row_ins_must_modify_rec(&cursor)) { + /* If the existing record is being modified and the new record + is doesn't fit the provided slot then existing record is added + to free list and new record is inserted. This also means + cursor that we have cached for SELECT is now invalid. */ + index->last_sel_cur->invalid = true; + /* There is already an index entry with a long enough common prefix, we must convert the insert into a modify of an existing record */ @@ -2793,6 +3041,11 @@ row_ins_sec_index_entry_low( err = row_ins_sec_index_entry_by_modify( flags, mode, &cursor, &offsets, offsets_heap, heap, entry, thr, &mtr); + + if (err == DB_SUCCESS && dict_index_is_spatial(index) + && rtr_info.mbr_adj) { + err = rtr_ins_enlarge_mbr(&cursor, thr, &mtr); + } } else { rec_t* insert_rec; big_rec_t* big_rec; @@ -2802,6 +3055,11 @@ row_ins_sec_index_entry_low( flags, &cursor, &offsets, &offsets_heap, entry, &insert_rec, &big_rec, 0, thr, &mtr); + if (err == DB_SUCCESS + && dict_index_is_spatial(index) + && rtr_info.mbr_adj) { + err = rtr_ins_enlarge_mbr(&cursor, thr, &mtr); + } } else { ut_ad(mode == BTR_MODIFY_TREE); if (buf_LRU_buf_pool_running_out()) { @@ -2822,6 +3080,11 @@ row_ins_sec_index_entry_low( entry, &insert_rec, &big_rec, 0, thr, &mtr); } + if (err == DB_SUCCESS + && dict_index_is_spatial(index) + && rtr_info.mbr_adj) { + err = rtr_ins_enlarge_mbr(&cursor, thr, &mtr); + } } if (err == DB_SUCCESS && trx_id) { @@ -2835,15 +3098,18 @@ row_ins_sec_index_entry_low( } func_exit: + if (dict_index_is_spatial(index)) { + rtr_clean_rtr_info(&rtr_info, true); + } + mtr_commit(&mtr); - return(err); + DBUG_RETURN(err); } /***************************************************************//** Tries to insert the externally stored fields (off-page columns) of a clustered index entry. @return DB_SUCCESS or DB_OUT_OF_FILE_SPACE */ -UNIV_INTERN dberr_t row_ins_index_entry_big_rec_func( /*=============================*/ @@ -2854,12 +3120,12 @@ row_ins_index_entry_big_rec_func( dict_index_t* index, /*!< in: index */ const char* file, /*!< in: file name of caller */ #ifndef DBUG_OFF - const void* thd, /*!< in: connection, or NULL */ + const void* thd, /*!< in: connection, or NULL */ #endif /* DBUG_OFF */ ulint line) /*!< in: line number of caller */ { mtr_t mtr; - btr_cur_t cursor; + btr_pcur_t pcur; rec_t* rec; dberr_t error; @@ -2868,26 +3134,30 @@ row_ins_index_entry_big_rec_func( DEBUG_SYNC_C_IF_THD(thd, "before_row_ins_extern_latch"); mtr_start(&mtr); - btr_cur_search_to_nth_level(index, 0, entry, PAGE_CUR_LE, - BTR_MODIFY_TREE, &cursor, 0, - file, line, &mtr); - rec = btr_cur_get_rec(&cursor); + mtr.set_named_space(index->space); + dict_disable_redo_if_temporary(index->table, &mtr); + + btr_pcur_open(index, entry, PAGE_CUR_LE, BTR_MODIFY_TREE, + &pcur, &mtr); + rec = btr_pcur_get_rec(&pcur); offsets = rec_get_offsets(rec, index, offsets, ULINT_UNDEFINED, heap); DEBUG_SYNC_C_IF_THD(thd, "before_row_ins_extern"); error = btr_store_big_rec_extern_fields( - index, btr_cur_get_block(&cursor), - rec, offsets, big_rec, &mtr, BTR_STORE_INSERT); + &pcur, 0, offsets, big_rec, &mtr, BTR_STORE_INSERT); DEBUG_SYNC_C_IF_THD(thd, "after_row_ins_extern"); if (error == DB_SUCCESS && dict_index_is_online_ddl(index)) { - row_log_table_insert(rec, index, offsets); + row_log_table_insert(btr_pcur_get_rec(&pcur), entry, + index, offsets); } mtr_commit(&mtr); + btr_pcur_close(&pcur); + return(error); } @@ -2896,57 +3166,83 @@ Inserts an entry into a clustered index. Tries first optimistic, then pessimistic descent down the tree. If the entry matches enough to a delete marked record, performs the insert by updating or delete unmarking the delete marked record. -@return DB_SUCCESS, DB_LOCK_WAIT, DB_DUPLICATE_KEY, or some other error code */ -UNIV_INTERN +@return DB_SUCCESS, DB_LOCK_WAIT, DB_DUPLICATE_KEY, or some other error code */ dberr_t row_ins_clust_index_entry( /*======================*/ dict_index_t* index, /*!< in: clustered index */ dtuple_t* entry, /*!< in/out: index entry to insert */ que_thr_t* thr, /*!< in: query thread */ - ulint n_ext) /*!< in: number of externally stored columns */ + ulint n_ext, /*!< in: number of externally stored columns */ + bool dup_chk_only) + /*!< in: if true, just do duplicate check + and return. don't execute actual insert. */ { dberr_t err; ulint n_uniq; + DBUG_ENTER("row_ins_clust_index_entry"); + if (!index->table->foreign_set.empty()) { err = row_ins_check_foreign_constraints( index->table, index, entry, thr); if (err != DB_SUCCESS) { - return(err); + DBUG_RETURN(err); } } n_uniq = dict_index_is_unique(index) ? index->n_uniq : 0; /* Try first optimistic descent to the B-tree */ + ulint flags; - log_free_check(); - - err = row_ins_clust_index_entry_low( - 0, BTR_MODIFY_LEAF, index, n_uniq, entry, n_ext, thr); + if (!dict_table_is_intrinsic(index->table)) { + log_free_check(); + flags = dict_table_is_temporary(index->table) + ? BTR_NO_LOCKING_FLAG + : 0; + } else { + flags = BTR_NO_LOCKING_FLAG | BTR_NO_UNDO_LOG_FLAG; + } -#ifdef UNIV_DEBUG - /* Work around Bug#14626800 ASSERTION FAILURE IN DEBUG_SYNC(). - Once it is fixed, remove the 'ifdef', 'if' and this comment. */ - if (!thr_get_trx(thr)->ddl) { - DEBUG_SYNC_C_IF_THD(thr_get_trx(thr)->mysql_thd, - "after_row_ins_clust_index_entry_leaf"); + if (dict_table_is_intrinsic(index->table) + && dict_index_is_auto_gen_clust(index)) { + err = row_ins_sorted_clust_index_entry( + BTR_MODIFY_LEAF, index, entry, n_ext, thr); + } else { + err = row_ins_clust_index_entry_low( + flags, BTR_MODIFY_LEAF, index, n_uniq, entry, + n_ext, thr, dup_chk_only); } -#endif /* UNIV_DEBUG */ + + + DEBUG_SYNC_C_IF_THD(thr_get_trx(thr)->mysql_thd, + "after_row_ins_clust_index_entry_leaf"); if (err != DB_FAIL) { DEBUG_SYNC_C("row_ins_clust_index_entry_leaf_after"); - return(err); + DBUG_RETURN(err); } /* Try then pessimistic descent to the B-tree */ + if (!dict_table_is_intrinsic(index->table)) { + log_free_check(); + } else { + index->last_sel_cur->invalid = true; + } - log_free_check(); + if (dict_table_is_intrinsic(index->table) + && dict_index_is_auto_gen_clust(index)) { + err = row_ins_sorted_clust_index_entry( + BTR_MODIFY_TREE, index, entry, n_ext, thr); + } else { + err = row_ins_clust_index_entry_low( + flags, BTR_MODIFY_TREE, index, n_uniq, entry, + n_ext, thr, dup_chk_only); + } - return(row_ins_clust_index_entry_low( - 0, BTR_MODIFY_TREE, index, n_uniq, entry, n_ext, thr)); + DBUG_RETURN(err); } /***************************************************************//** @@ -2954,19 +3250,25 @@ Inserts an entry into a secondary index. Tries first optimistic, then pessimistic descent down the tree. If the entry matches enough to a delete marked record, performs the insert by updating or delete unmarking the delete marked record. -@return DB_SUCCESS, DB_LOCK_WAIT, DB_DUPLICATE_KEY, or some other error code */ -UNIV_INTERN +@return DB_SUCCESS, DB_LOCK_WAIT, DB_DUPLICATE_KEY, or some other error code */ dberr_t row_ins_sec_index_entry( /*====================*/ dict_index_t* index, /*!< in: secondary index */ dtuple_t* entry, /*!< in/out: index entry to insert */ - que_thr_t* thr) /*!< in: query thread */ + que_thr_t* thr, /*!< in: query thread */ + bool dup_chk_only) + /*!< in: if true, just do duplicate check + and return. don't execute actual insert. */ { dberr_t err; mem_heap_t* offsets_heap; mem_heap_t* heap; + DBUG_EXECUTE_IF("row_ins_sec_index_entry_timeout", { + DBUG_SET("-d,row_ins_sec_index_entry_timeout"); + return(DB_LOCK_WAIT);}); + if (!index->table->foreign_set.empty()) { err = row_ins_check_foreign_constraints(index->table, index, entry, thr); @@ -2976,27 +3278,43 @@ row_ins_sec_index_entry( } } - ut_ad(thr_get_trx(thr)->id); + ut_ad(thr_get_trx(thr)->id != 0 + || dict_table_is_intrinsic(index->table)); offsets_heap = mem_heap_create(1024); heap = mem_heap_create(1024); /* Try first optimistic descent to the B-tree */ - log_free_check(); + ulint flags; + + if (!dict_table_is_intrinsic(index->table)) { + log_free_check(); + flags = dict_table_is_temporary(index->table) + ? BTR_NO_LOCKING_FLAG + : 0; + } else { + flags = BTR_NO_LOCKING_FLAG | BTR_NO_UNDO_LOG_FLAG; + } err = row_ins_sec_index_entry_low( - 0, BTR_MODIFY_LEAF, index, offsets_heap, heap, entry, 0, thr); + flags, BTR_MODIFY_LEAF, index, offsets_heap, heap, entry, + 0, thr, dup_chk_only); if (err == DB_FAIL) { mem_heap_empty(heap); /* Try then pessimistic descent to the B-tree */ - log_free_check(); + if (!dict_table_is_intrinsic(index->table)) { + log_free_check(); + } else { + index->last_sel_cur->invalid = true; + } err = row_ins_sec_index_entry_low( - 0, BTR_MODIFY_TREE, index, - offsets_heap, heap, entry, 0, thr); + flags, BTR_MODIFY_TREE, index, + offsets_heap, heap, entry, 0, thr, + dup_chk_only); } mem_heap_free(heap); @@ -3009,7 +3327,7 @@ Inserts an index entry to index. Tries first optimistic, then pessimistic descent down the tree. If the entry matches enough to a delete marked record, performs the insert by updating or delete unmarking the delete marked record. -@return DB_SUCCESS, DB_LOCK_WAIT, DB_DUPLICATE_KEY, or some other error code */ +@return DB_SUCCESS, DB_LOCK_WAIT, DB_DUPLICATE_KEY, or some other error code */ static dberr_t row_ins_index_entry( @@ -3018,46 +3336,101 @@ row_ins_index_entry( dtuple_t* entry, /*!< in/out: index entry to insert */ que_thr_t* thr) /*!< in: query thread */ { + ut_ad(thr_get_trx(thr)->id != 0); + DBUG_EXECUTE_IF("row_ins_index_entry_timeout", { DBUG_SET("-d,row_ins_index_entry_timeout"); return(DB_LOCK_WAIT);}); if (dict_index_is_clust(index)) { - return(row_ins_clust_index_entry(index, entry, thr, 0)); + return(row_ins_clust_index_entry(index, entry, thr, 0, false)); } else { - return(row_ins_sec_index_entry(index, entry, thr)); + return(row_ins_sec_index_entry(index, entry, thr, false)); } } -/***********************************************************//** -Sets the values of the dtuple fields in entry from the values of appropriate -columns in row. */ -static MY_ATTRIBUTE((nonnull)) + +/*****************************************************************//** +This function generate MBR (Minimum Bounding Box) for spatial objects +and set it to spatial index field. */ +static void +row_ins_spatial_index_entry_set_mbr_field( +/*======================================*/ + dfield_t* field, /*!< in/out: mbr field */ + const dfield_t* row_field) /*!< in: row field */ +{ + uchar* dptr = NULL; + ulint dlen = 0; + double mbr[SPDIMS * 2]; + + /* This must be a GEOMETRY datatype */ + ut_ad(DATA_GEOMETRY_MTYPE(field->type.mtype)); + + dptr = static_cast<uchar*>(dfield_get_data(row_field)); + dlen = dfield_get_len(row_field); + + /* obtain the MBR */ + rtree_mbr_from_wkb(dptr + GEO_DATA_HEADER_SIZE, + static_cast<uint>(dlen - GEO_DATA_HEADER_SIZE), + SPDIMS, mbr); + + /* Set mbr as index entry data */ + dfield_write_mbr(field, mbr); +} + +/** Sets the values of the dtuple fields in entry from the values of appropriate +columns in row. +@param[in] index index handler +@param[out] entry index entry to make +@param[in] row row + +@return DB_SUCCESS if the set is successful */ +dberr_t row_ins_index_entry_set_vals( -/*=========================*/ - dict_index_t* index, /*!< in: index */ - dtuple_t* entry, /*!< in: index entry to make */ - const dtuple_t* row) /*!< in: row */ + const dict_index_t* index, + dtuple_t* entry, + const dtuple_t* row) { ulint n_fields; ulint i; + ulint num_v = dtuple_get_n_v_fields(entry); n_fields = dtuple_get_n_fields(entry); - for (i = 0; i < n_fields; i++) { - dict_field_t* ind_field; + for (i = 0; i < n_fields + num_v; i++) { + dict_field_t* ind_field = NULL; dfield_t* field; const dfield_t* row_field; ulint len; + dict_col_t* col; + + if (i >= n_fields) { + /* This is virtual field */ + field = dtuple_get_nth_v_field(entry, i - n_fields); + col = &dict_table_get_nth_v_col( + index->table, i - n_fields)->m_col; + } else { + field = dtuple_get_nth_field(entry, i); + ind_field = dict_index_get_nth_field(index, i); + col = ind_field->col; + } + + if (dict_col_is_virtual(col)) { + const dict_v_col_t* v_col + = reinterpret_cast<const dict_v_col_t*>(col); + ut_ad(dtuple_get_n_fields(row) + == dict_table_get_n_cols(index->table)); + row_field = dtuple_get_nth_v_field(row, v_col->v_pos); + } else { + row_field = dtuple_get_nth_field( + row, ind_field->col->ind); + } - field = dtuple_get_nth_field(entry, i); - ind_field = dict_index_get_nth_field(index, i); - row_field = dtuple_get_nth_field(row, ind_field->col->ind); len = dfield_get_len(row_field); /* Check column prefix indexes */ - if (ind_field->prefix_len > 0 + if (ind_field != NULL && ind_field->prefix_len > 0 && dfield_get_len(row_field) != UNIV_SQL_NULL) { const dict_col_t* col @@ -3073,12 +3446,26 @@ row_ins_index_entry_set_vals( ut_ad(!dfield_is_ext(row_field)); } + /* Handle spatial index. For the first field, replace + the data with its MBR (Minimum Bounding Box). */ + if ((i == 0) && dict_index_is_spatial(index)) { + if (!row_field->data + || row_field->len < GEO_DATA_HEADER_SIZE) { + return(DB_CANT_CREATE_GEOMETRY_OBJECT); + } + row_ins_spatial_index_entry_set_mbr_field( + field, row_field); + continue; + } + dfield_set_data(field, dfield_get_data(row_field), len); if (dfield_is_ext(row_field)) { ut_ad(dict_index_is_clust(index)); dfield_set_ext(field); } } + + return(DB_SUCCESS); } /***********************************************************//** @@ -3094,24 +3481,24 @@ row_ins_index_entry_step( { dberr_t err; + DBUG_ENTER("row_ins_index_entry_step"); + ut_ad(dtuple_check_typed(node->row)); - row_ins_index_entry_set_vals(node->index, node->entry, node->row); + err = row_ins_index_entry_set_vals(node->index, node->entry, node->row); + + if (err != DB_SUCCESS) { + DBUG_RETURN(err); + } ut_ad(dtuple_check_typed(node->entry)); err = row_ins_index_entry(node->index, node->entry, thr); -#ifdef UNIV_DEBUG - /* Work around Bug#14626800 ASSERTION FAILURE IN DEBUG_SYNC(). - Once it is fixed, remove the 'ifdef', 'if' and this comment. */ - if (!thr_get_trx(thr)->ddl) { - DEBUG_SYNC_C_IF_THD(thr_get_trx(thr)->mysql_thd, - "after_row_ins_index_entry_step"); - } -#endif /* UNIV_DEBUG */ + DEBUG_SYNC_C_IF_THD(thr_get_trx(thr)->mysql_thd, + "after_row_ins_index_entry_step"); - return(err); + DBUG_RETURN(err); } /***********************************************************//** @@ -3217,6 +3604,14 @@ row_ins( { dberr_t err; + DBUG_ENTER("row_ins"); + + DBUG_PRINT("row_ins", ("table: %s", node->table->name.m_name)); + + if (node->duplicate) { + thr_get_trx(thr)->error_state = DB_DUPLICATE_KEY; + } + if (node->state == INS_NODE_ALLOC_ROW_ID) { row_ins_alloc_row_id_step(node); @@ -3242,12 +3637,52 @@ row_ins( if (node->index->type != DICT_FTS) { err = row_ins_index_entry_step(node, thr); - if (err != DB_SUCCESS) { - - return(err); + switch (err) { + case DB_SUCCESS: + break; + case DB_DUPLICATE_KEY: + ut_ad(dict_index_is_unique(node->index)); + + if (thr_get_trx(thr)->isolation_level + >= TRX_ISO_REPEATABLE_READ + && thr_get_trx(thr)->duplicates) { + + /* When we are in REPLACE statement or + INSERT .. ON DUPLICATE UPDATE + statement, we process all the + unique secondary indexes, even after we + encounter a duplicate error. This is + done to take necessary gap locks in + secondary indexes to block concurrent + transactions from inserting the + searched records. */ + if (!node->duplicate) { + /* Save 1st dup error. Ignore + subsequent dup errors. */ + node->duplicate = node->index; + thr_get_trx(thr)->error_state + = DB_DUPLICATE_KEY; + } + break; + } + // fall through + default: + DBUG_RETURN(err); } } + if (node->duplicate && dict_table_is_temporary(node->table)) { + ut_ad(thr_get_trx(thr)->error_state + == DB_DUPLICATE_KEY); + /* For TEMPORARY TABLE, we won't lock anything, + so we can simply break here instead of requiring + GAP locks for other unique secondary indexes, + pretending we have consumed all indexes. */ + node->index = NULL; + node->entry = NULL; + break; + } + node->index = dict_table_get_next_index(node->index); node->entry = UT_LIST_GET_NEXT(tuple_list, node->entry); @@ -3261,20 +3696,36 @@ row_ins( node->index = dict_table_get_next_index(node->index); node->entry = UT_LIST_GET_NEXT(tuple_list, node->entry); } + + /* After encountering a duplicate key error, we process + remaining indexes just to place gap locks and no actual + insertion will take place. These gap locks are needed + only for unique indexes. So skipping non-unique indexes. */ + if (node->duplicate) { + while (node->index + && !dict_index_is_unique(node->index)) { + + node->index = dict_table_get_next_index( + node->index); + node->entry = UT_LIST_GET_NEXT(tuple_list, + node->entry); + } + thr_get_trx(thr)->error_state = DB_DUPLICATE_KEY; + } } ut_ad(node->entry == NULL); + thr_get_trx(thr)->error_info = node->duplicate; node->state = INS_NODE_ALLOC_ROW_ID; - return(DB_SUCCESS); + DBUG_RETURN(node->duplicate ? DB_DUPLICATE_KEY : DB_SUCCESS); } /***********************************************************//** Inserts a row to a table. This is a high-level function used in SQL execution graphs. -@return query thread to run next or NULL */ -UNIV_INTERN +@return query thread to run next or NULL */ que_thr_t* row_ins_step( /*=========*/ @@ -3288,13 +3739,16 @@ row_ins_step( ut_ad(thr); + DEBUG_SYNC_C("innodb_row_ins_step_enter"); + trx = thr_get_trx(thr); - trx_start_if_not_started_xa(trx); + trx_start_if_not_started_xa(trx, true); node = static_cast<ins_node_t*>(thr->run_node); ut_ad(que_node_get_type(node) == QUE_NODE_INSERT); + ut_ad(!dict_table_is_intrinsic(node->table)); parent = que_node_get_parent(node); sel_node = node->select; @@ -3313,6 +3767,7 @@ row_ins_step( table during the search operation, and there is no need to set it again here. But we must write trx->id to node->trx_id_buf. */ + memset(node->trx_id_buf, 0, DATA_TRX_ID_LEN); trx_write_trx_id(node->trx_id_buf, trx->id); if (node->state == INS_NODE_SET_IX_LOCK) { |