update cache put

This commit is contained in:
yihaoDeng 2021-12-18 23:06:08 +08:00
parent c51780622f
commit f074a4b3a5
7 changed files with 116 additions and 40 deletions

View File

@ -57,7 +57,7 @@ option(
) )
option( option(
USE_INVERTEDINDEX BUILD_WITH_INVERTEDINDEX
"If use invertedIndex" "If use invertedIndex"
ON ON
) )

View File

@ -51,7 +51,7 @@ int indexMultiTermQueryAdd(SIndexMultiTermQuery *pQuery, SIndexTerm
* @param: * @param:
* @param: * @param:
*/ */
SIndex* indexOpen(SIndexOpts *opt, const char *path); int indexOpen(SIndexOpts *opt, const char *path, SIndex **index);
void indexClose(SIndex *index); void indexClose(SIndex *index);
int indexPut(SIndex *index, SIndexMultiTerm *terms, int uid); int indexPut(SIndex *index, SIndexMultiTerm *terms, int uid);
int indexDelete(SIndex *index, SIndexMultiTermQuery *query); int indexDelete(SIndex *index, SIndexMultiTermQuery *query);

View File

@ -22,9 +22,13 @@ if (${BUILD_WITH_LUCENE})
index index
PUBLIC lucene++ PUBLIC lucene++
) )
endif(${BUILD_WITH_LUCENE}) endif(${BUILD_WITH_LUCENE})
if (${BUILD_WITH_INVERTEDINDEX})
add_definitions(-DUSE_INVERTED_INDEX)
endif(${BUILD_WITH_INVERTEDINDEX})
if (${BUILD_TEST}) if (${BUILD_TEST})
add_subdirectory(test) add_subdirectory(test)
endif(${BUILD_TEST}) endif(${BUILD_TEST})

View File

@ -37,10 +37,10 @@ struct SIndex {
#endif #endif
void *cache; void *cache;
void *tindex; void *tindex;
SHashObj *fieldObj;// < field name, field id> SHashObj *colObj;// < field name, field id>
int64_t suid; // current super table id, -1 is normal table int64_t suid; // current super table id, -1 is normal table
int fieldId; // field id allocated to cache int colId; // field id allocated to cache
int32_t cVersion; // current version allocated to cache int32_t cVersion; // current version allocated to cache
pthread_mutex_t mtx; pthread_mutex_t mtx;
}; };

View File

@ -22,11 +22,10 @@
#endif #endif
typedef struct SIdxFieldInfo { typedef struct SIdxColInfo {
int fieldId; // generated by index internal int colId; // generated by index internal
int cVersion; int cVersion;
int type; // field type } SIdxColInfo;
} SIdxFieldInfo;
static pthread_once_t isInit = PTHREAD_ONCE_INIT; static pthread_once_t isInit = PTHREAD_ONCE_INIT;
static void indexInit(); static void indexInit();
@ -38,9 +37,10 @@ static int indexMergeCacheIntoTindex(struct SIndex *sIdx) {
indexWarn("suid %" PRIu64 " merge cache into tindex", sIdx->suid); indexWarn("suid %" PRIu64 " merge cache into tindex", sIdx->suid);
return 0; return 0;
} }
SIndex *indexOpen(SIndexOpts *opts, const char *path) { int indexOpen(SIndexOpts *opts, const char *path, SIndex **index) {
pthread_once(&isInit, indexInit); pthread_once(&isInit, indexInit);
SIndex *sIdx = calloc(1, sizeof(SIndex)); SIndex *sIdx = calloc(1, sizeof(SIndex));
if (sIdx == NULL) { return -1; }
#ifdef USE_LUCENE #ifdef USE_LUCENE
index_t *index = index_open(path); index_t *index = index_open(path);
@ -49,11 +49,13 @@ SIndex *indexOpen(SIndexOpts *opts, const char *path) {
sIdx->cache = (void*)indexCacheCreate(); sIdx->cache = (void*)indexCacheCreate();
sIdx->tindex = NULL; sIdx->tindex = NULL;
sIdx->fieldObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK); sIdx->colObj = taosHashInit(8, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
sIdx->fieldId = 1; sIdx->colId = 1;
sIdx->cVersion = 1; sIdx->cVersion = 1;
pthread_mutex_init(&sIdx->mtx, NULL); pthread_mutex_init(&sIdx->mtx, NULL);
return sIdx;
*index = sIdx;
return 0;
} }
void indexClose(SIndex *sIdx) { void indexClose(SIndex *sIdx) {
@ -62,16 +64,16 @@ void indexClose(SIndex *sIdx) {
sIdx->index = NULL; sIdx->index = NULL;
#endif #endif
#ifdef USE_INVERTEDINDEX #ifdef USE_INVERTED_INDEX
indexCacheDestroy(sIdx->cache); indexCacheDestroy(sIdx->cache);
taosHashCleanup(sIdx->fieldObj); taosHashCleanup(sIdx->colObj);
pthread_mutex_destroy(&sIdx->mtx); pthread_mutex_destroy(&sIdx->mtx);
#endif #endif
free(sIdx); free(sIdx);
return; return;
} }
int indexPut(SIndex *index, SArray* fVals, int uid) { int indexPut(SIndex *index, SIndexMultiTerm * fVals, int uid) {
#ifdef USE_LUCENE #ifdef USE_LUCENE
index_document_t *doc = index_document_create(); index_document_t *doc = index_document_create();
@ -89,38 +91,38 @@ int indexPut(SIndex *index, SArray* fVals, int uid) {
index_document_destroy(doc); index_document_destroy(doc);
#endif #endif
#ifdef USE_INVERTEDINDEX #ifdef USE_INVERTED_INDEX
//TODO(yihao): reduce the lock range //TODO(yihao): reduce the lock range
pthread_mutex_lock(&index->mtx); pthread_mutex_lock(&index->mtx);
for (int i = 0; i < taosArrayGetSize(fVals); i++) { for (int i = 0; i < taosArrayGetSize(fVals); i++) {
SIndexTerm *p = taosArrayGetP(fVals, i); SIndexTerm *p = taosArrayGetP(fVals, i);
SIdxFieldInfo *fi = taosHashGet(index->fieldObj, p->key, p->nKey); SIdxColInfo *fi = taosHashGet(index->colObj, p->colName, p->nColName);
if (fi == NULL) { if (fi == NULL) {
SIdxFieldInfo tfi = {.fieldId = index->fieldId, .type = p->type}; SIdxColInfo tfi = {.colId = index->colId};
index->cVersion++; index->cVersion++;
index->fieldId++; index->colId++;
taosHashPut(index->fieldObj, p->key, p->nKey, &tfi, sizeof(tfi)); taosHashPut(index->colObj, p->colName, p->nColName, &tfi, sizeof(tfi));
} else { } else {
//TODO, del //TODO, del
} }
} }
pthread_mutex_unlock(&index->mtx);
for (int i = 0; i < taosArrayGetSize(fVals); i++) { for (int i = 0; i < taosArrayGetSize(fVals); i++) {
SIndexTerm *p = taosArrayGetP(fVals, i); SIndexTerm *p = taosArrayGetP(fVals, i);
SIdxFieldInfo *fi = taosHashGet(index->fieldObj, p->key, p->nKey); SIdxColInfo *fi = taosHashGet(index->colObj, p->colName, p->nColName);
assert(fi != NULL); assert(fi != NULL);
int32_t fieldId = fi->fieldId; int32_t colId = fi->colId;
int32_t fieldType = fi->type;
int32_t version = index->cVersion; int32_t version = index->cVersion;
int res = indexCachePut(index->cache, fieldId, fieldType, p->val, p->nVal, version, uid, p->operType); int ret = indexCachePut(index->cache, colId, p->colType, p->colVal, p->nColVal, version, uid, p->operType);
if (ret != 0) { if (ret != 0) {
return return ret;
} }
} }
pthread_mutex_unlock(&index->mtx);
#endif #endif
return 1; return 0;
} }
int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result) { int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result) {
#ifdef USE_LUCENE #ifdef USE_LUCENE
@ -159,7 +161,7 @@ int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result
free(types); free(types);
#endif #endif
#ifdef USE_INVERTEDINDEX #ifdef USE_INVERTED_INDEX
#endif #endif
return 1; return 1;
@ -167,13 +169,13 @@ int indexSearch(SIndex *index, SIndexMultiTermQuery *multiQuerys, SArray *result
int indexDelete(SIndex *index, SIndexMultiTermQuery *query) { int indexDelete(SIndex *index, SIndexMultiTermQuery *query) {
#ifdef USE_INVERTEDINDEX #ifdef USE_INVERTED_INDEX
#endif #endif
return 1; return 1;
} }
int indexRebuild(SIndex *index, SIndexOpts *opts) { int indexRebuild(SIndex *index, SIndexOpts *opts) {
#ifdef USE_INVERTEDINDEX #ifdef USE_INVERTED_INDEX
#endif #endif
} }

View File

@ -1,7 +1,7 @@
add_executable(indexTest "") add_executable(indexTest "")
target_sources(indexTest target_sources(indexTest
PRIVATE PRIVATE
"indexTests.cpp" "indexTests.cc"
) )
target_include_directories ( indexTest target_include_directories ( indexTest
PUBLIC PUBLIC

View File

@ -1,3 +1,17 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <gtest/gtest.h> #include <gtest/gtest.h>
#include <string> #include <string>
#include <iostream> #include <iostream>
@ -279,15 +293,71 @@ void validateFst() {
delete m; delete m;
} }
class IndexEnv : public ::testing::Test {
protected:
virtual void SetUp() {
taosRemoveDir(path);
opts = indexOptsCreate();
int ret = indexOpen(opts, path, &index);
assert(ret == 0);
}
virtual void TearDown() {
indexClose(index);
indexOptsDestroy(opts);
}
int main(int argc, char** argv) { const char *path = "/tmp/tindex";
checkFstPerf(); SIndexOpts *opts;
//checkFstPrefixSearch(); SIndex *index;
return 1; };
TEST_F(IndexEnv, testPut) {
// single index column
{
std::string colName("tag1"), colVal("Hello world");
SIndexTerm *term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
SIndexMultiTerm *terms = indexMultiTermCreate();
indexMultiTermAdd(terms, term);
for (size_t i = 0; i < 100; i++) {
int tableId = i;
int ret = indexPut(index, terms, tableId);
assert(ret == 0);
}
indexMultiTermDestroy(terms);
}
// multi index column
{
SIndexMultiTerm *terms = indexMultiTermCreate();
{
std::string colName("tag1"), colVal("Hello world");
SIndexTerm *term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
indexMultiTermAdd(terms, term);
}
{
std::string colName("tag2"), colVal("Hello world");
SIndexTerm *term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), colVal.c_str(), colVal.size());
indexMultiTermAdd(terms, term);
}
for (int i = 0; i < 100; i++) {
int tableId = i;
int ret = indexPut(index, terms, tableId);
assert(ret == 0);
}
indexMultiTermDestroy(terms);
}
//
} }
//TEST(IndexFstBuilder, IndexFstInput) { TEST_F(IndexEnv, testDel) {
//
//} }