1berry/MariaDB-server
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

MDEV-25581 Allow user thread to do InnoDB fts cache sync

Browse Source 
Problem:
========
  InnoDB FTS requesting the fts sync of the table once the fts
cache size reaches 1/10 of innodb_ft_cache_size. But fts_sync()
releases cache lock when writing the word. By doing this, InnoDB
insert thread increases the innodb fts cache memory and
SYNC operation will take more time to complete.

Solution:
=========
  Remove the fts sync operation(FTS_MSG_SYNC_TABLE) from
the fts optimize background thread. Instead of that,
allow user thread to sync the InnoDB fts cache when
the cache size exceeds 512 kb. User thread holds
cache lock while doing cache syncing, it make sure that
other threads doesn't add the docs into the cache.

Removed FTS_MSG_SYNC_TABLE and its related function
because we do remove the FTS_MSG_SYNC_TABLE message
itself.

Removed fts_sync_index_check() and all related
function because other threads doesn't add while
cache operation going on.
This commit is contained in:
Thirunarayanan Balathandayuthapani 2022-06-07 10:07:13 +05:30
parent 51ca5d517e
commit 1fd7d3a9ad
11 changed files with 82 additions and 395 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
mysql-test/suite/innodb_fts/r/concurrent_insert.result
View File

@ -19,7 +19,7 @@ INSERT INTO t2 VALUES('mariadb');
connection default; connection default;
SET @saved_dbug = @@GLOBAL.debug_dbug; SET @saved_dbug = @@GLOBAL.debug_dbug;
SET GLOBAL debug_dbug ='+d,fts_instrument_sync_request,ib_optimize_wq_hang'; SET GLOBAL debug_dbug ='+d,fts_instrument_sync_request,ib_optimize_wq_hang';
SET DEBUG_SYNC= 'fts_instrument_sync_request SET DEBUG_SYNC= 'fts_sync_end
SIGNAL drop_index_start WAIT_FOR sync_op'; SIGNAL drop_index_start WAIT_FOR sync_op';
INSERT INTO t1 VALUES('Keyword'); INSERT INTO t1 VALUES('Keyword');
connect con1,localhost,root,,,; connect con1,localhost,root,,,;

16
mysql-test/suite/innodb_fts/r/sync.result
View File

@ -11,19 +11,19 @@ INSERT INTO t1(title) VALUES('database');
connection con1; connection con1;
SET @old_dbug = @@SESSION.debug_dbug; SET @old_dbug = @@SESSION.debug_dbug;
SET debug_dbug = '+d,fts_instrument_sync_debug'; SET debug_dbug = '+d,fts_instrument_sync_debug';
SET DEBUG_SYNC= 'fts_write_node SIGNAL written WAIT_FOR selected'; SET DEBUG_SYNC= 'fts_sync_end SIGNAL written WAIT_FOR selected';
INSERT INTO t1(title) VALUES('mysql database'); INSERT INTO t1(title) VALUES('mysql database');
connection default; connection default;
SET DEBUG_SYNC= 'now WAIT_FOR written'; SET DEBUG_SYNC= 'now WAIT_FOR written';
SET GLOBAL innodb_ft_aux_table="test/t1"; SET GLOBAL innodb_ft_aux_table="test/t1";
SELECT * FROM INFORMATION_SCHEMA.INNODB_FT_INDEX_CACHE; SELECT * FROM INFORMATION_SCHEMA.INNODB_FT_INDEX_CACHE;
WORD FIRST_DOC_ID LAST_DOC_ID DOC_COUNT DOC_ID POSITION WORD FIRST_DOC_ID LAST_DOC_ID DOC_COUNT DOC_ID POSITION
SELECT * FROM INFORMATION_SCHEMA.INNODB_FT_INDEX_TABLE;
WORD FIRST_DOC_ID LAST_DOC_ID DOC_COUNT DOC_ID POSITION
database 2 3 2 2 0 database 2 3 2 2 0
database 2 3 2 3 6 database 2 3 2 3 6
mysql 1 3 2 1 0 mysql 1 3 2 1 0
mysql 1 3 2 3 0 mysql 1 3 2 3 0
SELECT * FROM INFORMATION_SCHEMA.INNODB_FT_INDEX_TABLE;
WORD FIRST_DOC_ID LAST_DOC_ID DOC_COUNT DOC_ID POSITION
SET GLOBAL innodb_ft_aux_table=default; SET GLOBAL innodb_ft_aux_table=default;
SELECT * FROM t1 WHERE MATCH(title) AGAINST('mysql database'); SELECT * FROM t1 WHERE MATCH(title) AGAINST('mysql database');
FTS_DOC_ID title FTS_DOC_ID title
@ -59,7 +59,7 @@ INSERT INTO t1(title) VALUES('mysql');
INSERT INTO t1(title) VALUES('database'); INSERT INTO t1(title) VALUES('database');
connection con1; connection con1;
SET debug_dbug = '+d,fts_instrument_sync_debug'; SET debug_dbug = '+d,fts_instrument_sync_debug';
SET DEBUG_SYNC= 'fts_write_node SIGNAL written WAIT_FOR inserted'; SET DEBUG_SYNC= 'fts_sync_end SIGNAL written WAIT_FOR inserted';
INSERT INTO t1(title) VALUES('mysql database'); INSERT INTO t1(title) VALUES('mysql database');
connection default; connection default;
SET DEBUG_SYNC= 'now WAIT_FOR written'; SET DEBUG_SYNC= 'now WAIT_FOR written';
@ -70,14 +70,14 @@ SET debug_dbug = @old_dbug;
SET GLOBAL innodb_ft_aux_table="test/t1"; SET GLOBAL innodb_ft_aux_table="test/t1";
SELECT * FROM INFORMATION_SCHEMA.INNODB_FT_INDEX_CACHE; SELECT * FROM INFORMATION_SCHEMA.INNODB_FT_INDEX_CACHE;
WORD FIRST_DOC_ID LAST_DOC_ID DOC_COUNT DOC_ID POSITION WORD FIRST_DOC_ID LAST_DOC_ID DOC_COUNT DOC_ID POSITION
database 4 4 1 4 6
mysql 4 4 1 4 0
SELECT * FROM INFORMATION_SCHEMA.INNODB_FT_INDEX_TABLE; SELECT * FROM INFORMATION_SCHEMA.INNODB_FT_INDEX_TABLE;
WORD FIRST_DOC_ID LAST_DOC_ID DOC_COUNT DOC_ID POSITION WORD FIRST_DOC_ID LAST_DOC_ID DOC_COUNT DOC_ID POSITION
database 2 3 2 2 0 database 2 3 2 2 0
database 2 3 2 3 6 database 2 3 2 3 6
database 4 4 1 4 6 mysql 1 3 2 1 0
mysql 1 4 3 1 0 mysql 1 3 2 3 0
mysql 1 4 3 3 0
mysql 1 4 3 4 0
SET GLOBAL innodb_ft_aux_table=default; SET GLOBAL innodb_ft_aux_table=default;
SELECT * FROM t1 WHERE MATCH(title) AGAINST('mysql database'); SELECT * FROM t1 WHERE MATCH(title) AGAINST('mysql database');
FTS_DOC_ID title FTS_DOC_ID title

35
mysql-test/suite/innodb_fts/r/sync_block.result
View File

@ -30,6 +30,7 @@ connection con1;
connection con2; connection con2;
/* conneciton con2 */ SELECT * FROM t1 WHERE MATCH(title) AGAINST('mysql database'); /* conneciton con2 */ SELECT * FROM t1 WHERE MATCH(title) AGAINST('mysql database');
FTS_DOC_ID title FTS_DOC_ID title
1 mysql database
connection default; connection default;
# make con1 & con2 show up in mysql.slow_log # make con1 & con2 show up in mysql.slow_log
SELECT SLEEP(2); SELECT SLEEP(2);
@ -39,41 +40,11 @@ SLEEP(2)
SELECT sql_text FROM mysql.slow_log WHERE query_time >= '00:00:02'; SELECT sql_text FROM mysql.slow_log WHERE query_time >= '00:00:02';
sql_text sql_text
INSERT INTO t1(title) VALUES('mysql database') INSERT INTO t1(title) VALUES('mysql database')
SELECT * FROM t1 WHERE MATCH(title) AGAINST('mysql database')
SET GLOBAL debug_dbug = @old_debug; SET GLOBAL debug_dbug = @old_debug;
TRUNCATE TABLE mysql.slow_log; TRUNCATE TABLE mysql.slow_log;
DROP TABLE t1; DROP TABLE t1;
# Case 2: Sync blocks DML(insert) on other tables. SET DEBUG_SYNC=RESET;
CREATE TABLE t1 (
FTS_DOC_ID BIGINT UNSIGNED AUTO_INCREMENT NOT NULL PRIMARY KEY,
title VARCHAR(200),
FULLTEXT(title)
) ENGINE = InnoDB;
CREATE TABLE t2(id INT);
connection con1;
SET GLOBAL debug_dbug='+d,fts_instrument_sync_request,fts_instrument_sync_sleep';
SET DEBUG_SYNC= 'fts_instrument_sync_request SIGNAL begin WAIT_FOR continue';
INSERT INTO t1(title) VALUES('mysql database');
connection con2;
SET DEBUG_SYNC= 'now WAIT_FOR begin';
INSERT INTO t2 VALUES(1);
connection default;
SET DEBUG_SYNC= 'now SIGNAL continue';
connection con1;
/* connection con1 */ INSERT INTO t1(title) VALUES('mysql database');
connection con2;
/* conneciton con2 */ INSERT INTO t2 VALUES(1);
connection default;
SET DEBUG_SYNC = 'RESET';
# make con1 & con2 show up in mysql.slow_log
SELECT SLEEP(2);
SLEEP(2)
0
# slow log results should be empty here.
SELECT sql_text FROM mysql.slow_log WHERE query_time >= '00:00:02';
sql_text
SET GLOBAL debug_dbug = @old_debug;
TRUNCATE TABLE mysql.slow_log;
DROP TABLE t1,t2;
disconnect con1; disconnect con1;
disconnect con2; disconnect con2;
# Restore slow log settings. # Restore slow log settings.

2
mysql-test/suite/innodb_fts/t/concurrent_insert.test
View File

@ -31,7 +31,7 @@ INSERT INTO t2 VALUES('mariadb');
connection default; connection default;
SET @saved_dbug = @@GLOBAL.debug_dbug; SET @saved_dbug = @@GLOBAL.debug_dbug;
SET GLOBAL debug_dbug ='+d,fts_instrument_sync_request,ib_optimize_wq_hang'; SET GLOBAL debug_dbug ='+d,fts_instrument_sync_request,ib_optimize_wq_hang';
SET DEBUG_SYNC= 'fts_instrument_sync_request SET DEBUG_SYNC= 'fts_sync_end
SIGNAL drop_index_start WAIT_FOR sync_op'; SIGNAL drop_index_start WAIT_FOR sync_op';
send INSERT INTO t1 VALUES('Keyword'); send INSERT INTO t1 VALUES('Keyword');

4
mysql-test/suite/innodb_fts/t/sync.test
View File

@ -26,7 +26,7 @@ connection con1;
SET @old_dbug = @@SESSION.debug_dbug; SET @old_dbug = @@SESSION.debug_dbug;
SET debug_dbug = '+d,fts_instrument_sync_debug'; SET debug_dbug = '+d,fts_instrument_sync_debug';
SET DEBUG_SYNC= 'fts_write_node SIGNAL written WAIT_FOR selected'; SET DEBUG_SYNC= 'fts_sync_end SIGNAL written WAIT_FOR selected';
send INSERT INTO t1(title) VALUES('mysql database'); send INSERT INTO t1(title) VALUES('mysql database');
@ -73,7 +73,7 @@ connection con1;
SET debug_dbug = '+d,fts_instrument_sync_debug'; SET debug_dbug = '+d,fts_instrument_sync_debug';
SET DEBUG_SYNC= 'fts_write_node SIGNAL written WAIT_FOR inserted'; SET DEBUG_SYNC= 'fts_sync_end SIGNAL written WAIT_FOR inserted';
send INSERT INTO t1(title) VALUES('mysql database'); send INSERT INTO t1(title) VALUES('mysql database');

48
mysql-test/suite/innodb_fts/t/sync_block.test
View File

@ -65,53 +65,7 @@ SET GLOBAL debug_dbug = @old_debug;
TRUNCATE TABLE mysql.slow_log; TRUNCATE TABLE mysql.slow_log;
DROP TABLE t1; DROP TABLE t1;
SET DEBUG_SYNC=RESET;
--echo # Case 2: Sync blocks DML(insert) on other tables.
CREATE TABLE t1 (
FTS_DOC_ID BIGINT UNSIGNED AUTO_INCREMENT NOT NULL PRIMARY KEY,
title VARCHAR(200),
FULLTEXT(title)
) ENGINE = InnoDB;
CREATE TABLE t2(id INT);
connection con1;
SET GLOBAL debug_dbug='+d,fts_instrument_sync_request,fts_instrument_sync_sleep';
SET DEBUG_SYNC= 'fts_instrument_sync_request SIGNAL begin WAIT_FOR continue';
send INSERT INTO t1(title) VALUES('mysql database');
connection con2;
SET DEBUG_SYNC= 'now WAIT_FOR begin';
send INSERT INTO t2 VALUES(1);
connection default;
SET DEBUG_SYNC= 'now SIGNAL continue';
connection con1;
--echo /* connection con1 */ INSERT INTO t1(title) VALUES('mysql database');
--reap
connection con2;
--echo /* conneciton con2 */ INSERT INTO t2 VALUES(1);
--reap
connection default;
SET DEBUG_SYNC = 'RESET';
-- echo # make con1 & con2 show up in mysql.slow_log
SELECT SLEEP(2);
-- echo # slow log results should be empty here.
SELECT sql_text FROM mysql.slow_log WHERE query_time >= '00:00:02';
SET GLOBAL debug_dbug = @old_debug;
TRUNCATE TABLE mysql.slow_log;
DROP TABLE t1,t2;
disconnect con1; disconnect con1;
disconnect con2; disconnect con2;

265
storage/innobase/fts/fts0fts.cc
View File

@ -38,6 +38,22 @@ Full Text Search interface
#include "dict0stats.h" #include "dict0stats.h"
#include "btr0pcur.h" #include "btr0pcur.h"
/** The SYNC state of the cache. There is one instance of this struct
associated with each ADD thread. */
struct fts_sync_t {
/** Transaction used for SYNCing the cache to disk */
trx_t *trx;
/** Table with FTS index(es) */
dict_table_t *table;
/** Max size in bytes of the cache */
ulint max_cache_size;
/** The doc id at which the cache was noted as being
full, we use this to set the upper_limit field */
doc_id_t max_doc_id;
/** SYNC start time; only used if fts_enable_diag_print */
time_t start_time;
};
static const ulint FTS_MAX_ID_LEN = 32; static const ulint FTS_MAX_ID_LEN = 32;
/** Column name from the FTS config table */ /** Column name from the FTS config table */
@ -185,15 +201,8 @@ struct fts_tokenize_param_t {
/** Run SYNC on the table, i.e., write out data from the cache to the /** Run SYNC on the table, i.e., write out data from the cache to the
FTS auxiliary INDEX table and clear the cache at the end. FTS auxiliary INDEX table and clear the cache at the end.
@param[in,out] sync sync state @param[in,out] sync sync state
@param[in] unlock_cache whether unlock cache lock when write node
@param[in] wait whether wait when a sync is in progress
@return DB_SUCCESS if all OK */ @return DB_SUCCESS if all OK */
static static dberr_t fts_sync(fts_sync_t *sync);
dberr_t
fts_sync(
fts_sync_t* sync,
bool unlock_cache,
bool wait);
/****************************************************************//** /****************************************************************//**
Release all resources help by the words rb tree e.g., the node ilist. */ Release all resources help by the words rb tree e.g., the node ilist. */
@ -266,7 +275,6 @@ fts_cache_destroy(fts_cache_t* cache)
mysql_mutex_destroy(&cache->init_lock); mysql_mutex_destroy(&cache->init_lock);
mysql_mutex_destroy(&cache->deleted_lock); mysql_mutex_destroy(&cache->deleted_lock);
mysql_mutex_destroy(&cache->doc_id_lock); mysql_mutex_destroy(&cache->doc_id_lock);
pthread_cond_destroy(&cache->sync->cond);
if (cache->stopword_info.cached_stopword) { if (cache->stopword_info.cached_stopword) {
rbt_free(cache->stopword_info.cached_stopword); rbt_free(cache->stopword_info.cached_stopword);
@ -540,7 +548,6 @@ fts_index_cache_init(
for (i = 0; i < FTS_NUM_AUX_INDEX; ++i) { for (i = 0; i < FTS_NUM_AUX_INDEX; ++i) {
ut_a(index_cache->ins_graph[i] == NULL); ut_a(index_cache->ins_graph[i] == NULL);
ut_a(index_cache->sel_graph[i] == NULL);
} }
} }
@ -610,7 +617,6 @@ fts_cache_create(
mem_heap_zalloc(heap, sizeof(fts_sync_t))); mem_heap_zalloc(heap, sizeof(fts_sync_t)));
cache->sync->table = table; cache->sync->table = table;
pthread_cond_init(&cache->sync->cond, nullptr);
/* Create the index cache vector that will hold the inverted indexes. */ /* Create the index cache vector that will hold the inverted indexes. */
cache->indexes = ib_vector_create( cache->indexes = ib_vector_create(
@ -935,10 +941,6 @@ fts_cache_index_cache_create(
mem_heap_zalloc(static_cast<mem_heap_t*>( mem_heap_zalloc(static_cast<mem_heap_t*>(
cache->self_heap->arg), n_bytes)); cache->self_heap->arg), n_bytes));
index_cache->sel_graph = static_cast<que_t**>(
mem_heap_zalloc(static_cast<mem_heap_t*>(
cache->self_heap->arg), n_bytes));
fts_index_cache_init(cache->sync_heap, index_cache); fts_index_cache_init(cache->sync_heap, index_cache);
if (cache->get_docs) { if (cache->get_docs) {
@ -1012,13 +1014,6 @@ fts_cache_clear(
index_cache->ins_graph[j] = NULL; index_cache->ins_graph[j] = NULL;
} }
if (index_cache->sel_graph[j] != NULL) {
que_graph_free(index_cache->sel_graph[j]);
index_cache->sel_graph[j] = NULL;
}
} }
index_cache->doc_stats = NULL; index_cache->doc_stats = NULL;
@ -1311,8 +1306,7 @@ fts_cache_add_doc(
ib_vector_last(word->nodes)); ib_vector_last(word->nodes));
} }
if (fts_node == NULL || fts_node->synced if (!fts_node || fts_node->ilist_size > FTS_ILIST_MAX_SIZE
|| fts_node->ilist_size > FTS_ILIST_MAX_SIZE
|| doc_id < fts_node->last_doc_id) { || doc_id < fts_node->last_doc_id) {
fts_node = static_cast<fts_node_t*>( fts_node = static_cast<fts_node_t*>(
@ -3284,7 +3278,7 @@ fts_add_doc_from_tuple(
if (cache->total_size > fts_max_cache_size / 5 if (cache->total_size > fts_max_cache_size / 5
|| fts_need_sync) { || fts_need_sync) {
fts_sync(cache->sync, true, false); fts_sync(cache->sync);
} }
mtr_start(&mtr); mtr_start(&mtr);
@ -3444,42 +3438,34 @@ fts_add_doc_by_id(
get_doc->index_cache, get_doc->index_cache,
doc_id, doc.tokens); doc_id, doc.tokens);
bool need_sync = !cache->sync->in_progress /** FTS cache sync should happen
&& (fts_need_sync frequently. Because user thread
|| (cache->total_size shouldn't hold the cache lock for
- cache->total_size_at_sync) longer time. So cache should sync
> fts_max_cache_size / 10); whenever cache size exceeds 512 KB */
if (need_sync) { bool need_sync =
cache->total_size_at_sync = cache->total_size > 512*1024;
cache->total_size;
}
mysql_mutex_unlock(&table->fts->cache->lock); mysql_mutex_unlock(&table->fts->cache->lock);
DBUG_EXECUTE_IF( DBUG_EXECUTE_IF(
"fts_instrument_sync", "fts_instrument_sync",
fts_optimize_request_sync_table(table); fts_sync_table(table);
mysql_mutex_lock(&cache->lock);
if (cache->sync->in_progress)
my_cond_wait(
&cache->sync->cond,
&cache->lock.m_mutex);
mysql_mutex_unlock(&cache->lock);
); );
DBUG_EXECUTE_IF( DBUG_EXECUTE_IF(
"fts_instrument_sync_debug", "fts_instrument_sync_debug",
fts_sync(cache->sync, true, true); fts_sync(cache->sync);
); );
DEBUG_SYNC_C("fts_instrument_sync_request"); DEBUG_SYNC_C("fts_instrument_sync_request");
DBUG_EXECUTE_IF( DBUG_EXECUTE_IF(
"fts_instrument_sync_request", "fts_instrument_sync_request",
fts_optimize_request_sync_table(table); need_sync= true;
); );
if (need_sync) { if (need_sync) {
fts_optimize_request_sync_table(table); fts_sync_table(table);
} }
mtr_start(&mtr); mtr_start(&mtr);
@ -3846,15 +3832,13 @@ static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t dberr_t
fts_sync_write_words( fts_sync_write_words(
trx_t* trx, trx_t* trx,
fts_index_cache_t* index_cache, fts_index_cache_t* index_cache)
bool unlock_cache)
{ {
fts_table_t fts_table; fts_table_t fts_table;
ulint n_nodes = 0; ulint n_nodes = 0;
ulint n_words = 0; ulint n_words = 0;
const ib_rbt_node_t* rbt_node; const ib_rbt_node_t* rbt_node;
dberr_t error = DB_SUCCESS; dberr_t error = DB_SUCCESS;
ibool print_error = FALSE;
dict_table_t* table = index_cache->index->table; dict_table_t* table = index_cache->index->table;
FTS_INIT_INDEX_TABLE( FTS_INIT_INDEX_TABLE(
@ -3885,53 +3869,35 @@ fts_sync_write_words(
fts_table.suffix = fts_get_suffix(selected); fts_table.suffix = fts_get_suffix(selected);
/* We iterate over all the nodes even if there was an error */
for (i = 0; i < ib_vector_size(word->nodes); ++i) { for (i = 0; i < ib_vector_size(word->nodes); ++i) {
fts_node_t* fts_node = static_cast<fts_node_t*>( fts_node_t* fts_node = static_cast<fts_node_t*>(
ib_vector_get(word->nodes, i)); ib_vector_get(word->nodes, i));
if (fts_node->synced) { error = fts_write_node(
continue; trx, &index_cache->ins_graph[selected],
} else { &fts_table, &word->text, fts_node);
fts_node->synced = true;
}
/*FIXME: we need to handle the error properly. */ DEBUG_SYNC_C("fts_write_node");
if (error == DB_SUCCESS) { DBUG_EXECUTE_IF("fts_write_node_crash",
if (unlock_cache) {
mysql_mutex_unlock(
&table->fts->cache->lock);
}
error = fts_write_node(
trx,
&index_cache->ins_graph[selected],
&fts_table, &word->text, fts_node);
DEBUG_SYNC_C("fts_write_node");
DBUG_EXECUTE_IF("fts_write_node_crash",
DBUG_SUICIDE();); DBUG_SUICIDE(););
DBUG_EXECUTE_IF( DBUG_EXECUTE_IF("fts_instrument_sync_sleep",
"fts_instrument_sync_sleep",
std::this_thread::sleep_for( std::this_thread::sleep_for(
std::chrono::seconds(1));); std::chrono::seconds(1)););
if (unlock_cache) { if (error != DB_SUCCESS) {
mysql_mutex_lock( goto err_exit;
&table->fts->cache->lock);
}
} }
} }
n_nodes += ib_vector_size(word->nodes); n_nodes += ib_vector_size(word->nodes);
if (UNIV_UNLIKELY(error != DB_SUCCESS) && !print_error) { if (UNIV_UNLIKELY(error != DB_SUCCESS)) {
err_exit:
ib::error() << "(" << error << ") writing" ib::error() << "(" << error << ") writing"
" word node to FTS auxiliary index table " " word node to FTS auxiliary index table "
<< table->name; << table->name;
print_error = TRUE;
} }
} }
@ -3990,58 +3956,7 @@ fts_sync_index(
ut_ad(rbt_validate(index_cache->words)); ut_ad(rbt_validate(index_cache->words));
return(fts_sync_write_words(trx, index_cache, sync->unlock_cache)); return(fts_sync_write_words(trx, index_cache));
}
/** Check if index cache has been synced completely
@param[in,out] index_cache index cache
@return true if index is synced, otherwise false. */
static
bool
fts_sync_index_check(
fts_index_cache_t* index_cache)
{
const ib_rbt_node_t* rbt_node;
for (rbt_node = rbt_first(index_cache->words);
rbt_node != NULL;
rbt_node = rbt_next(index_cache->words, rbt_node)) {
fts_tokenizer_word_t* word;
word = rbt_value(fts_tokenizer_word_t, rbt_node);
fts_node_t* fts_node;
fts_node = static_cast<fts_node_t*>(ib_vector_last(word->nodes));
if (!fts_node->synced) {
return(false);
}
}
return(true);
}
/** Reset synced flag in index cache when rollback
@param[in,out] index_cache index cache */
static
void
fts_sync_index_reset(
fts_index_cache_t* index_cache)
{
const ib_rbt_node_t* rbt_node;
for (rbt_node = rbt_first(index_cache->words);
rbt_node != NULL;
rbt_node = rbt_next(index_cache->words, rbt_node)) {
fts_tokenizer_word_t* word;
word = rbt_value(fts_tokenizer_word_t, rbt_node);
fts_node_t* fts_node;
fts_node = static_cast<fts_node_t*>(ib_vector_last(word->nodes));
fts_node->synced = false;
}
} }
/** Commit the SYNC, change state of processed doc ids etc. /** Commit the SYNC, change state of processed doc ids etc.
@ -4074,14 +3989,14 @@ fts_sync_commit(
sync, cache->deleted_doc_ids); sync, cache->deleted_doc_ids);
} }
/* We need to do this within the deleted lock since fts_delete() can
attempt to add a deleted doc id to the cache deleted id array. */
fts_cache_clear(cache);
DEBUG_SYNC_C("fts_deleted_doc_ids_clear");
fts_cache_init(cache);
mysql_mutex_unlock(&cache->lock);
if (UNIV_LIKELY(error == DB_SUCCESS)) { if (UNIV_LIKELY(error == DB_SUCCESS)) {
/* We need to do this within the deleted lock
since fts_delete() can attempt to add a deleted
doc id to the cache deleted id array. */
fts_cache_clear(cache);
DEBUG_SYNC_C("fts_deleted_doc_ids_clear");
fts_cache_init(cache);
mysql_mutex_unlock(&cache->lock);
fts_sql_commit(trx); fts_sql_commit(trx);
} else { } else {
fts_sql_rollback(trx); fts_sql_rollback(trx);
@ -4123,10 +4038,6 @@ fts_sync_rollback(
index_cache = static_cast<fts_index_cache_t*>( index_cache = static_cast<fts_index_cache_t*>(
ib_vector_get(cache->indexes, i)); ib_vector_get(cache->indexes, i));
/* Reset synced flag so nodes will not be skipped
in the next sync, see fts_sync_write_words(). */
fts_sync_index_reset(index_cache);
for (j = 0; fts_index_selector[j].value; ++j) { for (j = 0; fts_index_selector[j].value; ++j) {
if (index_cache->ins_graph[j] != NULL) { if (index_cache->ins_graph[j] != NULL) {
@ -4135,13 +4046,6 @@ fts_sync_rollback(
index_cache->ins_graph[j] = NULL; index_cache->ins_graph[j] = NULL;
} }
if (index_cache->sel_graph[j] != NULL) {
que_graph_free(index_cache->sel_graph[j]);
index_cache->sel_graph[j] = NULL;
}
} }
} }
@ -4160,12 +4064,7 @@ FTS auxiliary INDEX table and clear the cache at the end.
@param[in] unlock_cache whether unlock cache lock when write node @param[in] unlock_cache whether unlock cache lock when write node
@param[in] wait whether wait when a sync is in progress @param[in] wait whether wait when a sync is in progress
@return DB_SUCCESS if all OK */ @return DB_SUCCESS if all OK */
static static dberr_t fts_sync(fts_sync_t *sync)
dberr_t
fts_sync(
fts_sync_t* sync,
bool unlock_cache,
bool wait)
{ {
if (srv_read_only_mode) { if (srv_read_only_mode) {
return DB_READ_ONLY; return DB_READ_ONLY;
@ -4176,33 +4075,13 @@ fts_sync(
fts_cache_t* cache = sync->table->fts->cache; fts_cache_t* cache = sync->table->fts->cache;
mysql_mutex_lock(&cache->lock); mysql_mutex_lock(&cache->lock);
/* Check if cache is being synced.
Note: we release cache lock in fts_sync_write_words() to
avoid long wait for the lock by other threads. */
if (sync->in_progress) {
if (!wait) {
mysql_mutex_unlock(&cache->lock);
return(DB_SUCCESS);
}
do {
my_cond_wait(&sync->cond, &cache->lock.m_mutex);
} while (sync->in_progress);
}
sync->unlock_cache = unlock_cache;
sync->in_progress = true;
DEBUG_SYNC_C("fts_sync_begin"); DEBUG_SYNC_C("fts_sync_begin");
fts_sync_begin(sync); fts_sync_begin(sync);
begin_sync:
const size_t fts_cache_size= fts_max_cache_size; const size_t fts_cache_size= fts_max_cache_size;
if (cache->total_size > fts_cache_size) { if (cache->total_size > fts_cache_size) {
/* Avoid the case: sync never finish when /* Avoid the case: sync never finish when
insert/update keeps comming. */ insert/update keeps comming. */
ut_ad(sync->unlock_cache);
sync->unlock_cache = false;
ib::warn() << "Total InnoDB FTS size " ib::warn() << "Total InnoDB FTS size "
<< cache->total_size << " for the table " << cache->total_size << " for the table "
<< cache->sync->table->name << cache->sync->table->name
@ -4226,52 +4105,23 @@ begin_sync:
error = fts_sync_index(sync, index_cache); error = fts_sync_index(sync, index_cache);
if (error != DB_SUCCESS) { if (error != DB_SUCCESS) {
goto end_sync; goto err_exit;
}
if (!sync->unlock_cache
&& cache->total_size < fts_max_cache_size) {
/* Reset the unlock cache if the value
is less than innodb_ft_cache_size */
sync->unlock_cache = true;
} }
} }
DBUG_EXECUTE_IF("fts_instrument_sync_interrupted", DBUG_EXECUTE_IF("fts_instrument_sync_interrupted",
sync->interrupted = true;
error = DB_INTERRUPTED; error = DB_INTERRUPTED;
goto end_sync; goto err_exit;
); );
/* Make sure all the caches are synced. */ if (error == DB_SUCCESS) {
for (i = 0; i < ib_vector_size(cache->indexes); ++i) {
fts_index_cache_t* index_cache;
index_cache = static_cast<fts_index_cache_t*>(
ib_vector_get(cache->indexes, i));
if (index_cache->index->to_be_dropped
|| fts_sync_index_check(index_cache)) {
continue;
}
goto begin_sync;
}
end_sync:
if (error == DB_SUCCESS && !sync->interrupted) {
error = fts_sync_commit(sync); error = fts_sync_commit(sync);
} else { } else {
err_exit:
fts_sync_rollback(sync); fts_sync_rollback(sync);
return error;
} }
mysql_mutex_lock(&cache->lock);
ut_ad(sync->in_progress);
sync->interrupted = false;
sync->in_progress = false;
pthread_cond_broadcast(&sync->cond);
mysql_mutex_unlock(&cache->lock);
/* We need to check whether an optimize is required, for that /* We need to check whether an optimize is required, for that
we make copies of the two variables that control the trigger. These we make copies of the two variables that control the trigger. These
variables can change behind our back and we don't want to hold the variables can change behind our back and we don't want to hold the
@ -4283,6 +4133,7 @@ end_sync:
mysql_mutex_unlock(&cache->deleted_lock); mysql_mutex_unlock(&cache->deleted_lock);
DEBUG_SYNC_C("fts_sync_end");
return(error); return(error);
} }
@ -4291,12 +4142,12 @@ FTS auxiliary INDEX table and clear the cache at the end.
@param[in,out] table fts table @param[in,out] table fts table
@param[in] wait whether wait for existing sync to finish @param[in] wait whether wait for existing sync to finish
@return DB_SUCCESS on success, error code on failure. */ @return DB_SUCCESS on success, error code on failure. */
dberr_t fts_sync_table(dict_table_t* table, bool wait) dberr_t fts_sync_table(dict_table_t* table)
{ {
ut_ad(table->fts); ut_ad(table->fts);
return table->space && !table->corrupted && table->fts->cache return table->space && !table->corrupted && table->fts->cache
? fts_sync(table->fts->cache->sync, !wait, wait) ? fts_sync(table->fts->cache->sync)
: DB_SUCCESS; : DB_SUCCESS;
} }

56
storage/innobase/fts/fts0opt.cc
View File

@ -83,9 +83,8 @@ enum fts_msg_type_t {
FTS_MSG_ADD_TABLE, /*!< Add table to the optimize thread's FTS_MSG_ADD_TABLE, /*!< Add table to the optimize thread's
work queue */ work queue */
FTS_MSG_DEL_TABLE, /*!< Remove a table from the optimize FTS_MSG_DEL_TABLE /*!< Remove a table from the optimize
threads work queue */ threads work queue */
FTS_MSG_SYNC_TABLE /*!< Sync fts cache of a table */
}; };
/** Compressed list of words that have been read from FTS INDEX /** Compressed list of words that have been read from FTS INDEX
@ -2625,36 +2624,6 @@ fts_optimize_remove_table(
mysql_mutex_unlock(&fts_optimize_wq->mutex); mysql_mutex_unlock(&fts_optimize_wq->mutex);
} }
/** Send sync fts cache for the table.
@param[in] table table to sync */
void
fts_optimize_request_sync_table(
dict_table_t* table)
{
/* if the optimize system not yet initialized, return */
if (!fts_optimize_wq) {
return;
}
mysql_mutex_lock(&fts_optimize_wq->mutex);
/* FTS optimizer thread is already exited */
if (fts_opt_start_shutdown) {
ib::info() << "Try to sync table " << table->name
<< " after FTS optimize thread exiting.";
} else if (table->fts->sync_message) {
/* If the table already has SYNC message in
fts_optimize_wq queue then ignore it */
} else {
add_msg(fts_optimize_create_msg(FTS_MSG_SYNC_TABLE, table));
table->fts->sync_message = true;
DBUG_EXECUTE_IF("fts_optimize_wq_count_check",
DBUG_ASSERT(fts_optimize_wq->length <= 1000););
}
mysql_mutex_unlock(&fts_optimize_wq->mutex);
}
/** Add a table to fts_slots if it doesn't already exist. */ /** Add a table to fts_slots if it doesn't already exist. */
static bool fts_optimize_new_table(dict_table_t* table) static bool fts_optimize_new_table(dict_table_t* table)
{ {
@ -2796,7 +2765,8 @@ static void fts_optimize_sync_table(dict_table_t *table,
if (sync_table->fts && sync_table->fts->cache && sync_table->is_accessible()) if (sync_table->fts && sync_table->fts->cache && sync_table->is_accessible())
{ {
fts_sync_table(sync_table, false); fts_sync_table(sync_table);
if (process_message) if (process_message)
{ {
mysql_mutex_lock(&fts_optimize_wq->mutex); mysql_mutex_lock(&fts_optimize_wq->mutex);
@ -2896,24 +2866,6 @@ retry_later:
--n_tables; --n_tables;
} }
break; break;
case FTS_MSG_SYNC_TABLE:
if (UNIV_UNLIKELY(wsrep_sst_disable_writes)) {
add_msg(msg);
goto retry_later;
}
DBUG_EXECUTE_IF(
"fts_instrument_msg_sync_sleep",
std::this_thread::sleep_for(
std::chrono::milliseconds(
300)););
fts_optimize_sync_table(
static_cast<dict_table_t*>(msg->ptr),
true);
break;
default: default:
ut_error; ut_error;
} }
@ -3046,7 +2998,7 @@ void fts_sync_during_ddl(dict_table_t* table)
if (!sync_message) if (!sync_message)
return; return;
fts_sync_table(table, false); fts_sync_table(table);
mysql_mutex_lock(&fts_optimize_wq->mutex); mysql_mutex_lock(&fts_optimize_wq->mutex);
table->fts->sync_message = false; table->fts->sync_message = false;

8
storage/innobase/handler/handler0alter.cc
View File

@ -11415,12 +11415,8 @@ foreign_fail:
ut_d(dict_table_check_for_dup_indexes( ut_d(dict_table_check_for_dup_indexes(
ctx->new_table, CHECK_ABORTED_OK)); ctx->new_table, CHECK_ABORTED_OK));
#ifdef UNIV_DEBUG ut_ad(!ctx->new_table->fts
if (!(ctx->new_table->fts != NULL || fts_check_cached_index(ctx->new_table));
&& ctx->new_table->fts->cache->sync->in_progress)) {
ut_a(fts_check_cached_index(ctx->new_table));
}
#endif
} }
unlock_and_close_files(deleted, trx); unlock_and_close_files(deleted, trx);

9
storage/innobase/include/fts0fts.h
View File

@ -656,12 +656,6 @@ fts_optimize_remove_table(
void void
fts_optimize_shutdown(); fts_optimize_shutdown();
/** Send sync fts cache for the table.
@param[in] table table to sync */
void
fts_optimize_request_sync_table(
dict_table_t* table);
/**********************************************************************//** /**********************************************************************//**
Take a FTS savepoint. */ Take a FTS savepoint. */
void void
@ -716,9 +710,8 @@ fts_savepoint_rollback_last_stmt(
/** Run SYNC on the table, i.e., write out data from the cache to the /** Run SYNC on the table, i.e., write out data from the cache to the
FTS auxiliary INDEX table and clear the cache at the end. FTS auxiliary INDEX table and clear the cache at the end.
@param[in,out] table fts table @param[in,out] table fts table
@param[in] wait whether to wait for existing sync to finish
@return DB_SUCCESS on success, error code on failure. */ @return DB_SUCCESS on success, error code on failure. */
dberr_t fts_sync_table(dict_table_t* table, bool wait = true); dberr_t fts_sync_table(dict_table_t* table);
/****************************************************************//** /****************************************************************//**
Create an FTS index cache. */ Create an FTS index cache. */

32
storage/innobase/include/fts0types.h
View File

@ -75,7 +75,6 @@ struct fts_index_cache_t {
que_t** ins_graph; /*!< Insert query graphs */ que_t** ins_graph; /*!< Insert query graphs */
que_t** sel_graph; /*!< Select query graphs */
CHARSET_INFO* charset; /*!< charset */ CHARSET_INFO* charset; /*!< charset */
}; };
@ -87,35 +86,7 @@ struct fts_stopword_t {
CHARSET_INFO* charset; /*!< charset for stopword */ CHARSET_INFO* charset; /*!< charset for stopword */
}; };
/** The SYNC state of the cache. There is one instance of this struct struct fts_sync_t;
associated with each ADD thread. */
struct fts_sync_t {
trx_t* trx; /*!< The transaction used for SYNCing
the cache to disk */
dict_table_t* table; /*!< Table with FTS index(es) */
ulint max_cache_size; /*!< Max size in bytes of the cache */
ibool cache_full; /*!< flag, when true it indicates that
we need to sync the cache to disk */
ulint lower_index; /*!< the start index of the doc id
vector from where to start adding
documents to the FTS cache */
ulint upper_index; /*!< max index of the doc id vector to
add to the FTS cache */
ibool interrupted; /*!< TRUE if SYNC was interrupted */
doc_id_t min_doc_id; /*!< The smallest doc id added to the
cache. It should equal to
doc_ids[lower_index] */
doc_id_t max_doc_id; /*!< The doc id at which the cache was
noted as being full, we use this to
set the upper_limit field */
time_t start_time; /*!< SYNC start time; only used if
fts_enable_diag_print */
bool in_progress; /*!< flag whether sync is in progress.*/
bool unlock_cache; /*!< flag whether unlock cache when
write fts node */
/** condition variable for in_progress; used with table->fts->cache->lock */
pthread_cond_t cond;
};
/** The cache for the FTS system. It is a memory-based inverted index /** The cache for the FTS system. It is a memory-based inverted index
that new entries are added to, until it grows over the configured maximum that new entries are added to, until it grows over the configured maximum
@ -204,7 +175,6 @@ struct fts_node_t {
ulint ilist_size_alloc; ulint ilist_size_alloc;
/*!< Allocated size of ilist in /*!< Allocated size of ilist in
bytes */ bytes */
bool synced; /*!< flag whether the node is synced */
}; };
/** A tokenizer word. Contains information about one word. */ /** A tokenizer word. Contains information about one word. */