in storage/innobase/row/row0merge.cc [1313:2001]
static MY_ATTRIBUTE((nonnull(1,2,3,4,6,9,10,16), warn_unused_result))
dberr_t
row_merge_read_clustered_index(
trx_t* trx,
struct TABLE* table,
const dict_table_t* old_table,
const dict_table_t* new_table,
bool online,
dict_index_t** index,
dict_index_t* fts_sort_idx,
fts_psort_t* psort_info,
merge_file_t* files,
const ulint* key_numbers,
ulint n_index,
const dtuple_t* add_cols,
const ulint* col_map,
ulint add_autoinc,
ib_sequence_t& sequence,
row_merge_block_t* block,
int* tmpfd)
{
dict_index_t* clust_index; /* Clustered index */
mem_heap_t* row_heap; /* Heap memory to create
clustered index tuples */
row_merge_buf_t** merge_buf; /* Temporary list for records*/
btr_pcur_t pcur; /* Cursor on the clustered
index */
mtr_t mtr; /* Mini transaction */
dberr_t err = DB_SUCCESS;/* Return code */
ulint n_nonnull = 0; /* number of columns
changed to NOT NULL */
ulint* nonnull = NULL; /* NOT NULL columns */
dict_index_t* fts_index = NULL;/* FTS index */
doc_id_t doc_id = 0;
doc_id_t max_doc_id = 0;
ibool add_doc_id = FALSE;
os_event_t fts_parallel_sort_event = NULL;
ibool fts_pll_sort = FALSE;
ib_int64_t sig_count = 0;
mem_heap_t* conv_heap = NULL;
DBUG_ENTER("row_merge_read_clustered_index");
ut_ad((old_table == new_table) == !col_map);
ut_ad(!add_cols || col_map);
trx->op_info = "reading clustered index";
#ifdef FTS_INTERNAL_DIAG_PRINT
DEBUG_FTS_SORT_PRINT("FTS_SORT: Start Create Index\n");
#endif
ut_ad(trx->mysql_thd != NULL);
const char* path = thd_innodb_tmpdir(trx->mysql_thd);
/* Create and initialize memory for record buffers */
merge_buf = static_cast<row_merge_buf_t**>(
mem_alloc(n_index * sizeof *merge_buf));
for (ulint i = 0; i < n_index; i++) {
if (index[i]->type & DICT_FTS) {
/* We are building a FT index, make sure
we have the temporary 'fts_sort_idx' */
ut_a(fts_sort_idx);
fts_index = index[i];
merge_buf[i] = row_merge_buf_create(fts_sort_idx);
add_doc_id = DICT_TF2_FLAG_IS_SET(
new_table, DICT_TF2_FTS_ADD_DOC_ID);
/* If Doc ID does not exist in the table itself,
fetch the first FTS Doc ID */
if (add_doc_id) {
fts_get_next_doc_id(
(dict_table_t*) new_table,
&doc_id);
ut_ad(doc_id > 0);
}
fts_pll_sort = TRUE;
row_fts_start_psort(psort_info);
fts_parallel_sort_event =
psort_info[0].psort_common->sort_event;
} else {
merge_buf[i] = row_merge_buf_create(index[i]);
}
}
mtr_start(&mtr);
/* Find the clustered index and create a persistent cursor
based on that. */
clust_index = dict_table_get_first_index(old_table);
btr_pcur_open_at_index_side(
true, clust_index, BTR_SEARCH_LEAF, &pcur, true, 0, &mtr);
if (old_table != new_table) {
/* The table is being rebuilt. Identify the columns
that were flagged NOT NULL in the new table, so that
we can quickly check that the records in the old table
do not violate the added NOT NULL constraints. */
nonnull = static_cast<ulint*>(
mem_alloc(dict_table_get_n_cols(new_table)
* sizeof *nonnull));
for (ulint i = 0; i < dict_table_get_n_cols(old_table); i++) {
if (dict_table_get_nth_col(old_table, i)->prtype
& DATA_NOT_NULL) {
continue;
}
const ulint j = col_map[i];
if (j == ULINT_UNDEFINED) {
/* The column was dropped. */
continue;
}
if (dict_table_get_nth_col(new_table, j)->prtype
& DATA_NOT_NULL) {
nonnull[n_nonnull++] = j;
}
}
if (!n_nonnull) {
mem_free(nonnull);
nonnull = NULL;
}
}
row_heap = mem_heap_create(sizeof(mrec_buf_t));
if (dict_table_is_comp(old_table)
&& !dict_table_is_comp(new_table)) {
conv_heap = mem_heap_create(sizeof(mrec_buf_t));
}
/* Scan the clustered index. */
for (;;) {
const rec_t* rec;
ulint* offsets;
const dtuple_t* row;
row_ext_t* ext;
page_cur_t* cur = btr_pcur_get_page_cur(&pcur);
page_cur_move_to_next(cur);
if (page_cur_is_after_last(cur)) {
if (UNIV_UNLIKELY(trx_is_interrupted(trx))) {
err = DB_INTERRUPTED;
trx->error_key_num = 0;
goto func_exit;
}
if (online && old_table != new_table) {
err = row_log_table_get_error(clust_index);
if (err != DB_SUCCESS) {
trx->error_key_num = 0;
goto func_exit;
}
}
#ifdef DBUG_OFF
# define dbug_run_purge false
#else /* DBUG_OFF */
bool dbug_run_purge = false;
#endif /* DBUG_OFF */
DBUG_EXECUTE_IF(
"ib_purge_on_create_index_page_switch",
dbug_run_purge = true;);
if (dbug_run_purge
|| rw_lock_get_waiters(
dict_index_get_lock(clust_index))) {
/* There are waiters on the clustered
index tree lock, likely the purge
thread. Store and restore the cursor
position, and yield so that scanning a
large table will not starve other
threads. */
/* Store the cursor position on the last user
record on the page. */
btr_pcur_move_to_prev_on_page(&pcur);
/* Leaf pages must never be empty, unless
this is the only page in the index tree. */
ut_ad(btr_pcur_is_on_user_rec(&pcur)
|| buf_block_get_page_no(
btr_pcur_get_block(&pcur))
== clust_index->page);
btr_pcur_store_position(&pcur, &mtr);
mtr_commit(&mtr);
if (dbug_run_purge) {
/* This is for testing
purposes only (see
DBUG_EXECUTE_IF above). We
signal the purge thread and
hope that the purge batch will
complete before we execute
btr_pcur_restore_position(). */
trx_purge_run();
os_thread_sleep(1000000);
}
/* Give the waiters a chance to proceed. */
os_thread_yield();
mtr_start(&mtr);
/* Restore position on the record, or its
predecessor if the record was purged
meanwhile. */
btr_pcur_restore_position(
BTR_SEARCH_LEAF, &pcur, &mtr);
/* Move to the successor of the
original record. */
if (!btr_pcur_move_to_next_user_rec(
&pcur, &mtr)) {
end_of_index:
row = NULL;
mtr_commit(&mtr);
mem_heap_free(row_heap);
if (nonnull) {
mem_free(nonnull);
}
goto write_buffers;
}
} else {
ulint next_page_no;
buf_block_t* block;
next_page_no = btr_page_get_next(
page_cur_get_page(cur), &mtr);
if (next_page_no == FIL_NULL) {
goto end_of_index;
}
block = page_cur_get_block(cur);
block = btr_block_get(
buf_block_get_space(block),
buf_block_get_zip_size(block),
next_page_no, BTR_SEARCH_LEAF,
clust_index, &mtr);
btr_leaf_page_release(page_cur_get_block(cur),
BTR_SEARCH_LEAF, &mtr);
page_cur_set_before_first(block, cur);
page_cur_move_to_next(cur);
ut_ad(!page_cur_is_after_last(cur));
}
}
rec = page_cur_get_rec(cur);
offsets = rec_get_offsets(rec, clust_index, NULL,
ULINT_UNDEFINED, &row_heap);
if (online) {
/* Perform a REPEATABLE READ.
When rebuilding the table online,
row_log_table_apply() must not see a newer
state of the table when applying the log.
This is mainly to prevent false duplicate key
errors, because the log will identify records
by the PRIMARY KEY, and also to prevent unsafe
BLOB access.
When creating a secondary index online, this
table scan must not see records that have only
been inserted to the clustered index, but have
not been written to the online_log of
index[]. If we performed READ UNCOMMITTED, it
could happen that the ADD INDEX reaches
ONLINE_INDEX_COMPLETE state between the time
the DML thread has updated the clustered index
but has not yet accessed secondary index. */
ut_ad(trx->read_view);
if (!read_view_sees_trx_id(
trx->read_view,
row_get_rec_trx_id(
rec, clust_index, offsets))) {
rec_t* old_vers;
row_vers_build_for_consistent_read(
rec, &mtr, clust_index, &offsets,
trx->read_view, &row_heap,
row_heap, &old_vers);
rec = old_vers;
if (!rec) {
continue;
}
}
if (rec_get_deleted_flag(
rec,
dict_table_is_comp(old_table))) {
/* This record was deleted in the latest
committed version, or it was deleted and
then reinserted-by-update before purge
kicked in. Skip it. */
continue;
}
ut_ad(!rec_offs_any_null_extern(rec, offsets));
} else if (rec_get_deleted_flag(
rec, dict_table_is_comp(old_table))) {
/* Skip delete-marked records.
Skipping delete-marked records will make the
created indexes unuseable for transactions
whose read views were created before the index
creation completed, but preserving the history
would make it tricky to detect duplicate
keys. */
continue;
}
/* When !online, we are holding a lock on old_table, preventing
any inserts that could have written a record 'stub' before
writing out off-page columns. */
ut_ad(!rec_offs_any_null_extern(rec, offsets));
/* Build a row based on the clustered index. */
row = row_build(ROW_COPY_POINTERS, clust_index,
rec, offsets, new_table,
add_cols, col_map, &ext, row_heap);
ut_ad(row);
for (ulint i = 0; i < n_nonnull; i++) {
const dfield_t* field = &row->fields[nonnull[i]];
ut_ad(dfield_get_type(field)->prtype & DATA_NOT_NULL);
if (dfield_is_null(field)) {
err = DB_INVALID_NULL;
trx->error_key_num = 0;
goto func_exit;
}
}
/* Get the next Doc ID */
if (add_doc_id) {
doc_id++;
} else {
doc_id = 0;
}
if (add_autoinc != ULINT_UNDEFINED) {
ut_ad(add_autoinc
< dict_table_get_n_user_cols(new_table));
const dfield_t* dfield;
dfield = dtuple_get_nth_field(row, add_autoinc);
if (dfield_is_null(dfield)) {
goto write_buffers;
}
const dtype_t* dtype = dfield_get_type(dfield);
byte* b = static_cast<byte*>(dfield_get_data(dfield));
if (sequence.eof()) {
err = DB_ERROR;
trx->error_key_num = 0;
ib_errf(trx->mysql_thd, IB_LOG_LEVEL_ERROR,
ER_AUTOINC_READ_FAILED, "[NULL]");
goto func_exit;
}
ulonglong value = sequence++;
switch (dtype_get_mtype(dtype)) {
case DATA_INT: {
ibool usign;
ulint len = dfield_get_len(dfield);
usign = dtype_get_prtype(dtype) & DATA_UNSIGNED;
mach_write_ulonglong(b, value, len, usign);
break;
}
case DATA_FLOAT:
mach_float_write(
b, static_cast<float>(value));
break;
case DATA_DOUBLE:
mach_double_write(
b, static_cast<double>(value));
break;
default:
ut_ad(0);
}
}
write_buffers:
/* Build all entries for all the indexes to be created
in a single scan of the clustered index. */
for (ulint i = 0; i < n_index; i++) {
row_merge_buf_t* buf = merge_buf[i];
merge_file_t* file = &files[i];
ulint rows_added = 0;
bool exceed_page = false;
if (UNIV_LIKELY
(row && (rows_added = row_merge_buf_add(
buf, fts_index, old_table,
psort_info, row, ext, &doc_id,
conv_heap, &exceed_page)))) {
/* If we are creating FTS index,
a single row can generate more
records for tokenized word */
file->n_rec += rows_added;
if (exceed_page) {
err = DB_TOO_BIG_RECORD;
break;
}
if (doc_id > max_doc_id) {
max_doc_id = doc_id;
}
if (buf->index->type & DICT_FTS) {
/* Check if error occurs in child thread */
for (ulint j = 0; j < fts_sort_pll_degree; j++) {
if (psort_info[j].error != DB_SUCCESS) {
err = psort_info[j].error;
trx->error_key_num = i;
break;
}
}
if (err != DB_SUCCESS) {
break;
}
}
continue;
}
if (buf->index->type & DICT_FTS) {
if (!row || !doc_id) {
continue;
}
}
/* The buffer must be sufficiently large
to hold at least one record. It may only
be empty when we reach the end of the
clustered index. row_merge_buf_add()
must not have been called in this loop. */
ut_ad(buf->n_tuples || row == NULL);
/* We have enough data tuples to form a block.
Sort them and write to disk. */
if (buf->n_tuples) {
if (dict_index_is_unique(buf->index)) {
row_merge_dup_t dup = {
buf->index, table, col_map, 0};
row_merge_buf_sort(buf, &dup);
if (dup.n_dup) {
err = DB_DUPLICATE_KEY;
trx->error_key_num
= key_numbers[i];
break;
}
} else {
row_merge_buf_sort(buf, NULL);
}
} else if (online && new_table == old_table) {
/* Note the newest transaction that
modified this index when the scan was
completed. We prevent older readers
from accessing this index, to ensure
read consistency. */
trx_id_t max_trx_id;
ut_a(row == NULL);
rw_lock_x_lock(
dict_index_get_lock(buf->index));
ut_a(dict_index_get_online_status(buf->index)
== ONLINE_INDEX_CREATION);
max_trx_id = row_log_get_max_trx(buf->index);
if (max_trx_id > buf->index->trx_id) {
buf->index->trx_id = max_trx_id;
}
rw_lock_x_unlock(
dict_index_get_lock(buf->index));
}
if (buf->n_tuples > 0) {
if (row_merge_file_create_if_needed(
file, tmpfd, buf->n_tuples, path) < 0) {
err = DB_OUT_OF_MEMORY;
trx->error_key_num = i;
break;
}
ut_ad(file->n_rec > 0);
row_merge_buf_write(buf, file, block);
if (!row_merge_write(file->fd, file->offset++,
block)) {
err = DB_TEMP_FILE_WRITE_FAILURE;
trx->error_key_num = i;
break;
}
}
UNIV_MEM_INVALID(&block[0], srv_sort_buf_size);
merge_buf[i] = row_merge_buf_empty(buf);
if (UNIV_LIKELY(row != NULL)) {
/* Try writing the record again, now
that the buffer has been written out
and emptied. */
if (UNIV_UNLIKELY
(!(rows_added = row_merge_buf_add(
buf, fts_index, old_table,
psort_info, row, ext,
&doc_id, conv_heap,
&exceed_page)))) {
/* An empty buffer should have enough
room for at least one record. */
ut_error;
}
if (exceed_page) {
err = DB_TOO_BIG_RECORD;
break;
}
file->n_rec += rows_added;
}
}
if (row == NULL) {
goto all_done;
}
if (err != DB_SUCCESS) {
goto func_exit;
}
mem_heap_empty(row_heap);
}
func_exit:
mtr_commit(&mtr);
mem_heap_free(row_heap);
if (nonnull) {
mem_free(nonnull);
}
all_done:
if (conv_heap != NULL) {
mem_heap_free(conv_heap);
}
#ifdef FTS_INTERNAL_DIAG_PRINT
DEBUG_FTS_SORT_PRINT("FTS_SORT: Complete Scan Table\n");
#endif
if (fts_pll_sort) {
bool all_exit = false;
ulint trial_count = 0;
const ulint max_trial_count = 10000;
wait_again:
/* Check if error occurs in child thread */
for (ulint j = 0; j < fts_sort_pll_degree; j++) {
if (psort_info[j].error != DB_SUCCESS) {
err = psort_info[j].error;
trx->error_key_num = j;
break;
}
}
/* Tell all children that parent has done scanning */
for (ulint i = 0; i < fts_sort_pll_degree; i++) {
if (err == DB_SUCCESS) {
psort_info[i].state = FTS_PARENT_COMPLETE;
} else {
psort_info[i].state = FTS_PARENT_EXITING;
}
}
/* Now wait all children to report back to be completed */
os_event_wait_time_low(fts_parallel_sort_event,
1000000, sig_count);
for (ulint i = 0; i < fts_sort_pll_degree; i++) {
if (psort_info[i].child_status != FTS_CHILD_COMPLETE
&& psort_info[i].child_status != FTS_CHILD_EXITING) {
sig_count = os_event_reset(
fts_parallel_sort_event);
goto wait_again;
}
}
/* Now all children should complete, wait a bit until
they all finish setting the event, before we free everything.
This has a 10 second timeout */
do {
all_exit = true;
for (ulint j = 0; j < fts_sort_pll_degree; j++) {
if (psort_info[j].child_status
!= FTS_CHILD_EXITING) {
all_exit = false;
os_thread_sleep(1000);
break;
}
}
trial_count++;
} while (!all_exit && trial_count < max_trial_count);
if (!all_exit) {
ut_ad(0);
ib_logf(IB_LOG_LEVEL_FATAL,
"Not all child sort threads exited"
" when creating FTS index '%s'",
fts_sort_idx->name);
}
}
#ifdef FTS_INTERNAL_DIAG_PRINT
DEBUG_FTS_SORT_PRINT("FTS_SORT: Complete Tokenization\n");
#endif
for (ulint i = 0; i < n_index; i++) {
row_merge_buf_free(merge_buf[i]);
}
row_fts_free_pll_merge_buf(psort_info);
mem_free(merge_buf);
btr_pcur_close(&pcur);
/* Update the next Doc ID we used. Table should be locked, so
no concurrent DML */
if (max_doc_id && err == DB_SUCCESS) {
/* Sync fts cache for other fts indexes to keep all
fts indexes consistent in sync_doc_id. */
err = fts_sync_table(const_cast<dict_table_t*>(new_table),
false, true, false);
if (err == DB_SUCCESS) {
fts_update_next_doc_id(
0, new_table, old_table->name, max_doc_id);
}
}
trx->op_info = "";
DBUG_RETURN(err);
}