SqliteMetaImpl.cpp 58.6 KB
Newer Older
J
jinhai 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

S
starlord 已提交
18
#include "db/meta/SqliteMetaImpl.h"
Y
youny626 已提交
19
#include "MetaConsts.h"
S
starlord 已提交
20 21
#include "db/IDGenerator.h"
#include "db/Utils.h"
22
#include "metrics/Metrics.h"
Y
youny626 已提交
23 24
#include "utils/Exception.h"
#include "utils/Log.h"
25
#include "utils/StringHelpFunctions.h"
X
Xu Peng 已提交
26

Y
youny626 已提交
27
#include <sqlite_orm.h>
X
Xu Peng 已提交
28
#include <unistd.h>
X
Xu Peng 已提交
29
#include <boost/filesystem.hpp>
30
#include <chrono>
X
Xu Peng 已提交
31
#include <fstream>
Y
youny626 已提交
32
#include <iostream>
S
starlord 已提交
33
#include <map>
Y
youny626 已提交
34
#include <memory>
S
starlord 已提交
35
#include <set>
Y
youny626 已提交
36
#include <sstream>
S
starlord 已提交
37

J
jinhai 已提交
38
namespace milvus {
X
Xu Peng 已提交
39
namespace engine {
40
namespace meta {
X
Xu Peng 已提交
41

X
Xu Peng 已提交
42 43
using namespace sqlite_orm;

G
groot 已提交
44 45
namespace {

S
starlord 已提交
46
Status
Y
youny626 已提交
47
HandleException(const std::string& desc, const char* what = nullptr) {
S
starlord 已提交
48
    if (what == nullptr) {
S
starlord 已提交
49 50 51 52 53 54 55
        ENGINE_LOG_ERROR << desc;
        return Status(DB_META_TRANSACTION_FAILED, desc);
    } else {
        std::string msg = desc + ":" + what;
        ENGINE_LOG_ERROR << msg;
        return Status(DB_META_TRANSACTION_FAILED, msg);
    }
G
groot 已提交
56 57
}

Y
youny626 已提交
58
}  // namespace
G
groot 已提交
59

S
starlord 已提交
60
inline auto
G
groot 已提交
61
StoragePrototype(const std::string& path) {
G
groot 已提交
62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87
    return make_storage(path,
                        make_table(META_TABLES,
                                   make_column("id", &TableSchema::id_, primary_key()),
                                   make_column("table_id", &TableSchema::table_id_, unique()),
                                   make_column("state", &TableSchema::state_),
                                   make_column("dimension", &TableSchema::dimension_),
                                   make_column("created_on", &TableSchema::created_on_),
                                   make_column("flag", &TableSchema::flag_, default_value(0)),
                                   make_column("index_file_size", &TableSchema::index_file_size_),
                                   make_column("engine_type", &TableSchema::engine_type_),
                                   make_column("nlist", &TableSchema::nlist_),
                                   make_column("metric_type", &TableSchema::metric_type_),
                                   make_column("owner_table", &TableSchema::owner_table_, default_value("")),
                                   make_column("partition_tag", &TableSchema::partition_tag_, default_value("")),
                                   make_column("version", &TableSchema::version_, default_value(CURRENT_VERSION))),
                        make_table(META_TABLEFILES,
                                   make_column("id", &TableFileSchema::id_, primary_key()),
                                   make_column("table_id", &TableFileSchema::table_id_),
                                   make_column("engine_type", &TableFileSchema::engine_type_),
                                   make_column("file_id", &TableFileSchema::file_id_),
                                   make_column("file_type", &TableFileSchema::file_type_),
                                   make_column("file_size", &TableFileSchema::file_size_, default_value(0)),
                                   make_column("row_count", &TableFileSchema::row_count_, default_value(0)),
                                   make_column("updated_time", &TableFileSchema::updated_time_),
                                   make_column("created_on", &TableFileSchema::created_on_),
                                   make_column("date", &TableFileSchema::date_)));
X
Xu Peng 已提交
88 89
}

X
Xu Peng 已提交
90
using ConnectorT = decltype(StoragePrototype(""));
X
Xu Peng 已提交
91 92
static std::unique_ptr<ConnectorT> ConnectorPtr;

Y
youny626 已提交
93
SqliteMetaImpl::SqliteMetaImpl(const DBMetaOptions& options) : options_(options) {
94 95 96 97 98 99
    Initialize();
}

SqliteMetaImpl::~SqliteMetaImpl() {
}

S
starlord 已提交
100
Status
Y
youny626 已提交
101
SqliteMetaImpl::NextTableId(std::string& table_id) {
G
groot 已提交
102
    std::lock_guard<std::mutex> lock(genid_mutex_);  // avoid duplicated id
103 104
    std::stringstream ss;
    SimpleIDGenerator g;
105
    ss << g.GetNextIDNumber();
106
    table_id = ss.str();
107 108 109
    return Status::OK();
}

S
starlord 已提交
110
Status
Y
youny626 已提交
111
SqliteMetaImpl::NextFileId(std::string& file_id) {
G
groot 已提交
112
    std::lock_guard<std::mutex> lock(genid_mutex_);  // avoid duplicated id
X
Xu Peng 已提交
113 114
    std::stringstream ss;
    SimpleIDGenerator g;
115
    ss << g.GetNextIDNumber();
X
Xu Peng 已提交
116 117 118 119
    file_id = ss.str();
    return Status::OK();
}

S
starlord 已提交
120 121 122
void
SqliteMetaImpl::ValidateMetaSchema() {
    if (ConnectorPtr == nullptr) {
123 124 125
        return;
    }

Y
youny626 已提交
126
    // old meta could be recreated since schema changed, throw exception if meta schema is not compatible
127
    auto ret = ConnectorPtr->sync_schema_simulate();
Y
youny626 已提交
128 129
    if (ret.find(META_TABLES) != ret.end() &&
        sqlite_orm::sync_schema_result::dropped_and_recreated == ret[META_TABLES]) {
130 131
        throw Exception(DB_INCOMPATIB_META, "Meta Tables schema is created by Milvus old version");
    }
Y
youny626 已提交
132 133
    if (ret.find(META_TABLEFILES) != ret.end() &&
        sqlite_orm::sync_schema_result::dropped_and_recreated == ret[META_TABLEFILES]) {
134 135 136 137
        throw Exception(DB_INCOMPATIB_META, "Meta TableFiles schema is created by Milvus old version");
    }
}

S
starlord 已提交
138 139
Status
SqliteMetaImpl::Initialize() {
S
starlord 已提交
140 141
    if (!boost::filesystem::is_directory(options_.path_)) {
        auto ret = boost::filesystem::create_directory(options_.path_);
142
        if (!ret) {
S
starlord 已提交
143
            std::string msg = "Failed to create db directory " + options_.path_;
S
starlord 已提交
144 145
            ENGINE_LOG_ERROR << msg;
            return Status(DB_INVALID_PATH, msg);
146
        }
X
Xu Peng 已提交
147
    }
X
Xu Peng 已提交
148

S
starlord 已提交
149
    ConnectorPtr = std::make_unique<ConnectorT>(StoragePrototype(options_.path_ + "/meta.sqlite"));
X
Xu Peng 已提交
150

151
    ValidateMetaSchema();
152

X
Xu Peng 已提交
153
    ConnectorPtr->sync_schema();
Y
youny626 已提交
154 155
    ConnectorPtr->open_forever();                          // thread safe option
    ConnectorPtr->pragma.journal_mode(journal_mode::WAL);  // WAL => write ahead log
X
Xu Peng 已提交
156

157
    CleanUp();
X
Xu Peng 已提交
158

X
Xu Peng 已提交
159
    return Status::OK();
X
Xu Peng 已提交
160 161
}

G
groot 已提交
162
Status
G
groot 已提交
163
SqliteMetaImpl::CreateTable(TableSchema& table_schema) {
G
groot 已提交
164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190
    try {
        server::MetricCollector metric;

        //multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

        if (table_schema.table_id_ == "") {
            NextTableId(table_schema.table_id_);
        } else {
            auto table = ConnectorPtr->select(columns(&TableSchema::state_),
                                              where(c(&TableSchema::table_id_) == table_schema.table_id_));
            if (table.size() == 1) {
                if (TableSchema::TO_DELETE == std::get<0>(table[0])) {
                    return Status(DB_ERROR, "Table already exists and it is in delete state, please wait a second");
                } else {
                    // Change from no error to already exist.
                    return Status(DB_ALREADY_EXIST, "Table already exists");
                }
            }
        }

        table_schema.id_ = -1;
        table_schema.created_on_ = utils::GetMicroSecTimeStamp();

        try {
            auto id = ConnectorPtr->insert(table_schema);
            table_schema.id_ = id;
G
groot 已提交
191
        } catch (std::exception& e) {
G
groot 已提交
192 193 194 195 196 197
            return HandleException("Encounter exception when create table", e.what());
        }

        ENGINE_LOG_DEBUG << "Successfully create table: " << table_schema.table_id_;

        return utils::CreateTablePath(options_, table_schema.table_id_);
G
groot 已提交
198
    } catch (std::exception& e) {
G
groot 已提交
199 200 201 202 203
        return HandleException("Encounter exception when create table", e.what());
    }
}

Status
G
groot 已提交
204
SqliteMetaImpl::DescribeTable(TableSchema& table_schema) {
G
groot 已提交
205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220
    try {
        server::MetricCollector metric;

        auto groups = ConnectorPtr->select(columns(&TableSchema::id_,
                                                   &TableSchema::state_,
                                                   &TableSchema::dimension_,
                                                   &TableSchema::created_on_,
                                                   &TableSchema::flag_,
                                                   &TableSchema::index_file_size_,
                                                   &TableSchema::engine_type_,
                                                   &TableSchema::nlist_,
                                                   &TableSchema::metric_type_,
                                                   &TableSchema::owner_table_,
                                                   &TableSchema::partition_tag_,
                                                   &TableSchema::version_),
                                           where(c(&TableSchema::table_id_) == table_schema.table_id_
G
groot 已提交
221
                                                 and c(&TableSchema::state_) != (int)TableSchema::TO_DELETE));
G
groot 已提交
222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238

        if (groups.size() == 1) {
            table_schema.id_ = std::get<0>(groups[0]);
            table_schema.state_ = std::get<1>(groups[0]);
            table_schema.dimension_ = std::get<2>(groups[0]);
            table_schema.created_on_ = std::get<3>(groups[0]);
            table_schema.flag_ = std::get<4>(groups[0]);
            table_schema.index_file_size_ = std::get<5>(groups[0]);
            table_schema.engine_type_ = std::get<6>(groups[0]);
            table_schema.nlist_ = std::get<7>(groups[0]);
            table_schema.metric_type_ = std::get<8>(groups[0]);
            table_schema.owner_table_ = std::get<9>(groups[0]);
            table_schema.partition_tag_ = std::get<10>(groups[0]);
            table_schema.version_ = std::get<11>(groups[0]);
        } else {
            return Status(DB_NOT_FOUND, "Table " + table_schema.table_id_ + " not found");
        }
G
groot 已提交
239
    } catch (std::exception& e) {
G
groot 已提交
240 241 242 243 244 245 246
        return HandleException("Encounter exception when describe table", e.what());
    }

    return Status::OK();
}

Status
G
groot 已提交
247
SqliteMetaImpl::HasTable(const std::string& table_id, bool& has_or_not) {
G
groot 已提交
248 249 250 251 252 253
    has_or_not = false;

    try {
        server::MetricCollector metric;
        auto tables = ConnectorPtr->select(columns(&TableSchema::id_),
                                           where(c(&TableSchema::table_id_) == table_id
G
groot 已提交
254
                                                 and c(&TableSchema::state_) != (int)TableSchema::TO_DELETE));
G
groot 已提交
255 256 257 258 259
        if (tables.size() == 1) {
            has_or_not = true;
        } else {
            has_or_not = false;
        }
G
groot 已提交
260
    } catch (std::exception& e) {
G
groot 已提交
261 262 263 264 265 266 267
        return HandleException("Encounter exception when lookup table", e.what());
    }

    return Status::OK();
}

Status
G
groot 已提交
268
SqliteMetaImpl::AllTables(std::vector<TableSchema>& table_schema_array) {
G
groot 已提交
269 270 271 272 273 274 275 276 277 278 279 280 281 282 283
    try {
        server::MetricCollector metric;

        auto selected = ConnectorPtr->select(columns(&TableSchema::id_,
                                                     &TableSchema::table_id_,
                                                     &TableSchema::dimension_,
                                                     &TableSchema::created_on_,
                                                     &TableSchema::flag_,
                                                     &TableSchema::index_file_size_,
                                                     &TableSchema::engine_type_,
                                                     &TableSchema::nlist_,
                                                     &TableSchema::metric_type_,
                                                     &TableSchema::owner_table_,
                                                     &TableSchema::partition_tag_,
                                                     &TableSchema::version_),
G
groot 已提交
284 285
                                             where(c(&TableSchema::state_) != (int)TableSchema::TO_DELETE));
        for (auto& table : selected) {
G
groot 已提交
286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301
            TableSchema schema;
            schema.id_ = std::get<0>(table);
            schema.table_id_ = std::get<1>(table);
            schema.dimension_ = std::get<2>(table);
            schema.created_on_ = std::get<3>(table);
            schema.flag_ = std::get<4>(table);
            schema.index_file_size_ = std::get<5>(table);
            schema.engine_type_ = std::get<6>(table);
            schema.nlist_ = std::get<7>(table);
            schema.metric_type_ = std::get<8>(table);
            schema.owner_table_ = std::get<9>(table);
            schema.partition_tag_ = std::get<10>(table);
            schema.version_ = std::get<11>(table);

            table_schema_array.emplace_back(schema);
        }
G
groot 已提交
302
    } catch (std::exception& e) {
G
groot 已提交
303 304 305 306 307 308 309
        return HandleException("Encounter exception when lookup all tables", e.what());
    }

    return Status::OK();
}

Status
G
groot 已提交
310
SqliteMetaImpl::DropTable(const std::string& table_id) {
G
groot 已提交
311 312 313 314 315 316 317 318 319
    try {
        server::MetricCollector metric;

        //multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

        //soft delete table
        ConnectorPtr->update_all(
            set(
G
groot 已提交
320
                c(&TableSchema::state_) = (int)TableSchema::TO_DELETE),
G
groot 已提交
321 322
            where(
                c(&TableSchema::table_id_) == table_id and
G
groot 已提交
323
                c(&TableSchema::state_) != (int)TableSchema::TO_DELETE));
G
groot 已提交
324 325

        ENGINE_LOG_DEBUG << "Successfully delete table, table id = " << table_id;
G
groot 已提交
326
    } catch (std::exception& e) {
G
groot 已提交
327 328 329 330 331 332 333
        return HandleException("Encounter exception when delete table", e.what());
    }

    return Status::OK();
}

Status
G
groot 已提交
334
SqliteMetaImpl::DeleteTableFiles(const std::string& table_id) {
G
groot 已提交
335 336 337 338 339 340 341 342 343
    try {
        server::MetricCollector metric;

        //multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

        //soft delete table files
        ConnectorPtr->update_all(
            set(
G
groot 已提交
344
                c(&TableFileSchema::file_type_) = (int)TableFileSchema::TO_DELETE,
G
groot 已提交
345 346 347
                c(&TableFileSchema::updated_time_) = utils::GetMicroSecTimeStamp()),
            where(
                c(&TableFileSchema::table_id_) == table_id and
G
groot 已提交
348
                c(&TableFileSchema::file_type_) != (int)TableFileSchema::TO_DELETE));
G
groot 已提交
349 350

        ENGINE_LOG_DEBUG << "Successfully delete table files, table id = " << table_id;
G
groot 已提交
351
    } catch (std::exception& e) {
G
groot 已提交
352 353 354 355 356 357 358
        return HandleException("Encounter exception when delete table files", e.what());
    }

    return Status::OK();
}

Status
G
groot 已提交
359
SqliteMetaImpl::CreateTableFile(TableFileSchema& file_schema) {
G
groot 已提交
360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391
    if (file_schema.date_ == EmptyDate) {
        file_schema.date_ = utils::GetDate();
    }
    TableSchema table_schema;
    table_schema.table_id_ = file_schema.table_id_;
    auto status = DescribeTable(table_schema);
    if (!status.ok()) {
        return status;
    }

    try {
        server::MetricCollector metric;

        NextFileId(file_schema.file_id_);
        file_schema.dimension_ = table_schema.dimension_;
        file_schema.file_size_ = 0;
        file_schema.row_count_ = 0;
        file_schema.created_on_ = utils::GetMicroSecTimeStamp();
        file_schema.updated_time_ = file_schema.created_on_;
        file_schema.index_file_size_ = table_schema.index_file_size_;
        file_schema.engine_type_ = table_schema.engine_type_;
        file_schema.nlist_ = table_schema.nlist_;
        file_schema.metric_type_ = table_schema.metric_type_;

        //multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

        auto id = ConnectorPtr->insert(file_schema);
        file_schema.id_ = id;

        ENGINE_LOG_DEBUG << "Successfully create table file, file id = " << file_schema.file_id_;
        return utils::CreateTableFilePath(options_, file_schema);
G
groot 已提交
392
    } catch (std::exception& e) {
G
groot 已提交
393 394 395 396 397 398
        return HandleException("Encounter exception when create table file", e.what());
    }

    return Status::OK();
}

S
starlord 已提交
399
// TODO(myh): Delete single vecotor by id
S
starlord 已提交
400
Status
G
groot 已提交
401 402
SqliteMetaImpl::DropDataByDate(const std::string& table_id,
                               const DatesT& dates) {
403
    if (dates.empty()) {
X
Xu Peng 已提交
404 405 406
        return Status::OK();
    }

407
    TableSchema table_schema;
G
groot 已提交
408
    table_schema.table_id_ = table_id;
X
Xu Peng 已提交
409
    auto status = DescribeTable(table_schema);
X
Xu Peng 已提交
410 411 412 413
    if (!status.ok()) {
        return status;
    }

G
groot 已提交
414
    try {
Y
youny626 已提交
415 416
        // sqlite_orm has a bug, 'in' statement cannot handle too many elements
        // so we split one query into multi-queries, this is a work-around!!
417 418 419
        std::vector<DatesT> split_dates;
        split_dates.push_back(DatesT());
        const size_t batch_size = 30;
Y
youny626 已提交
420
        for (DateT date : dates) {
421 422
            DatesT& last_batch = *split_dates.rbegin();
            last_batch.push_back(date);
Y
youny626 已提交
423
            if (last_batch.size() > batch_size) {
424 425 426 427
                split_dates.push_back(DatesT());
            }
        }

Y
youny626 已提交
428
        // multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
429 430
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

Y
youny626 已提交
431 432
        for (auto& batch_dates : split_dates) {
            if (batch_dates.empty()) {
433 434 435 436
                continue;
            }

            ConnectorPtr->update_all(
Y
youny626 已提交
437
                set(c(&TableFileSchema::file_type_) = (int)TableFileSchema::TO_DELETE,
438
                    c(&TableFileSchema::updated_time_) = utils::GetMicroSecTimeStamp()),
Y
youny626 已提交
439
                where(c(&TableFileSchema::table_id_) == table_id and in(&TableFileSchema::date_, batch_dates)));
440
        }
441

G
groot 已提交
442
        ENGINE_LOG_DEBUG << "Successfully drop data by date, table id = " << table_schema.table_id_;
G
groot 已提交
443
    } catch (std::exception& e) {
S
starlord 已提交
444
        return HandleException("Encounter exception when drop partition", e.what());
X
Xu Peng 已提交
445
    }
G
groot 已提交
446

X
Xu Peng 已提交
447 448 449
    return Status::OK();
}

S
starlord 已提交
450
Status
G
groot 已提交
451 452 453
SqliteMetaImpl::GetTableFiles(const std::string& table_id,
                              const std::vector<size_t>& ids,
                              TableFilesSchema& table_files) {
G
groot 已提交
454
    try {
G
groot 已提交
455 456 457 458 459 460 461 462 463 464 465
        table_files.clear();
        auto files = ConnectorPtr->select(columns(&TableFileSchema::id_,
                                                  &TableFileSchema::file_id_,
                                                  &TableFileSchema::file_type_,
                                                  &TableFileSchema::file_size_,
                                                  &TableFileSchema::row_count_,
                                                  &TableFileSchema::date_,
                                                  &TableFileSchema::engine_type_,
                                                  &TableFileSchema::created_on_),
                                          where(c(&TableFileSchema::table_id_) == table_id and
                                                in(&TableFileSchema::id_, ids) and
G
groot 已提交
466
                                                c(&TableFileSchema::file_type_) != (int)TableFileSchema::TO_DELETE));
G
groot 已提交
467 468 469 470 471
        TableSchema table_schema;
        table_schema.table_id_ = table_id;
        auto status = DescribeTable(table_schema);
        if (!status.ok()) {
            return status;
G
groot 已提交
472
        }
G
groot 已提交
473

G
groot 已提交
474
        Status result;
G
groot 已提交
475
        for (auto& file : files) {
G
groot 已提交
476 477 478 479 480 481 482 483 484 485 486 487 488 489
            TableFileSchema file_schema;
            file_schema.table_id_ = table_id;
            file_schema.id_ = std::get<0>(file);
            file_schema.file_id_ = std::get<1>(file);
            file_schema.file_type_ = std::get<2>(file);
            file_schema.file_size_ = std::get<3>(file);
            file_schema.row_count_ = std::get<4>(file);
            file_schema.date_ = std::get<5>(file);
            file_schema.engine_type_ = std::get<6>(file);
            file_schema.created_on_ = std::get<7>(file);
            file_schema.dimension_ = table_schema.dimension_;
            file_schema.index_file_size_ = table_schema.index_file_size_;
            file_schema.nlist_ = table_schema.nlist_;
            file_schema.metric_type_ = table_schema.metric_type_;
G
groot 已提交
490

G
groot 已提交
491
            utils::GetTableFilePath(options_, file_schema);
492

G
groot 已提交
493 494
            table_files.emplace_back(file_schema);
        }
495

G
groot 已提交
496 497
        ENGINE_LOG_DEBUG << "Get table files by id";
        return result;
G
groot 已提交
498
    } catch (std::exception& e) {
G
groot 已提交
499
        return HandleException("Encounter exception when lookup table files", e.what());
500
    }
X
Xu Peng 已提交
501 502
}

S
starlord 已提交
503
Status
G
groot 已提交
504
SqliteMetaImpl::UpdateTableFlag(const std::string& table_id, int64_t flag) {
G
groot 已提交
505
    try {
Y
Yu Kun 已提交
506
        server::MetricCollector metric;
G
groot 已提交
507

G
groot 已提交
508
        //set all backup file to raw
S
starlord 已提交
509
        ConnectorPtr->update_all(
G
groot 已提交
510 511 512 513 514
            set(
                c(&TableSchema::flag_) = flag),
            where(
                c(&TableSchema::table_id_) == table_id));
        ENGINE_LOG_DEBUG << "Successfully update table flag, table id = " << table_id;
G
groot 已提交
515
    } catch (std::exception& e) {
G
groot 已提交
516 517
        std::string msg = "Encounter exception when update table flag: table_id = " + table_id;
        return HandleException(msg, e.what());
G
groot 已提交
518 519 520 521 522
    }

    return Status::OK();
}

S
starlord 已提交
523
Status
G
groot 已提交
524
SqliteMetaImpl::UpdateTableFile(TableFileSchema& file_schema) {
G
groot 已提交
525
    file_schema.updated_time_ = utils::GetMicroSecTimeStamp();
G
groot 已提交
526
    try {
Y
Yu Kun 已提交
527
        server::MetricCollector metric;
G
groot 已提交
528

Y
youny626 已提交
529
        // multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
530 531
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

G
groot 已提交
532 533
        auto tables = ConnectorPtr->select(columns(&TableSchema::state_),
                                           where(c(&TableSchema::table_id_) == file_schema.table_id_));
G
groot 已提交
534

G
groot 已提交
535 536
        //if the table has been deleted, just mark the table file as TO_DELETE
        //clean thread will delete the file later
G
groot 已提交
537
        if (tables.size() < 1 || std::get<0>(tables[0]) == (int)TableSchema::TO_DELETE) {
G
groot 已提交
538 539
            file_schema.file_type_ = TableFileSchema::TO_DELETE;
        }
G
groot 已提交
540

G
groot 已提交
541 542 543
        ConnectorPtr->update(file_schema);

        ENGINE_LOG_DEBUG << "Update single table file, file id = " << file_schema.file_id_;
G
groot 已提交
544
    } catch (std::exception& e) {
G
groot 已提交
545 546 547 548
        std::string msg = "Exception update table file: table_id = " + file_schema.table_id_
                          + " file_id = " + file_schema.file_id_;
        return HandleException(msg, e.what());
    }
G
groot 已提交
549 550 551
    return Status::OK();
}

S
starlord 已提交
552
Status
G
groot 已提交
553
SqliteMetaImpl::UpdateTableFiles(TableFilesSchema& files) {
554
    try {
Y
Yu Kun 已提交
555
        server::MetricCollector metric;
G
groot 已提交
556

G
groot 已提交
557 558
        //multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);
G
groot 已提交
559

G
groot 已提交
560
        std::map<std::string, bool> has_tables;
G
groot 已提交
561
        for (auto& file : files) {
G
groot 已提交
562 563 564 565 566
            if (has_tables.find(file.table_id_) != has_tables.end()) {
                continue;
            }
            auto tables = ConnectorPtr->select(columns(&TableSchema::id_),
                                               where(c(&TableSchema::table_id_) == file.table_id_
G
groot 已提交
567
                                                     and c(&TableSchema::state_) != (int)TableSchema::TO_DELETE));
G
groot 已提交
568 569 570 571 572
            if (tables.size() >= 1) {
                has_tables[file.table_id_] = true;
            } else {
                has_tables[file.table_id_] = false;
            }
573
        }
P
peng.xu 已提交
574

G
groot 已提交
575
        auto commited = ConnectorPtr->transaction([&]() mutable {
G
groot 已提交
576
            for (auto& file : files) {
G
groot 已提交
577 578
                if (!has_tables[file.table_id_]) {
                    file.file_type_ = TableFileSchema::TO_DELETE;
579
                }
G
groot 已提交
580 581 582

                file.updated_time_ = utils::GetMicroSecTimeStamp();
                ConnectorPtr->update(file);
583
            }
G
groot 已提交
584 585
            return true;
        });
586

G
groot 已提交
587 588
        if (!commited) {
            return HandleException("UpdateTableFiles error: sqlite transaction failed");
P
peng.xu 已提交
589
        }
G
groot 已提交
590 591

        ENGINE_LOG_DEBUG << "Update " << files.size() << " table files";
G
groot 已提交
592
    } catch (std::exception& e) {
G
groot 已提交
593
        return HandleException("Encounter exception when update table files", e.what());
P
peng.xu 已提交
594 595 596 597
    }
    return Status::OK();
}

S
starlord 已提交
598
Status
Y
youny626 已提交
599
SqliteMetaImpl::UpdateTableIndex(const std::string& table_id, const TableIndex& index) {
600
    try {
Y
Yu Kun 已提交
601
        server::MetricCollector metric;
602

Y
youny626 已提交
603
        // multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
604 605
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

G
groot 已提交
606 607 608 609 610 611 612 613 614 615
        auto tables = ConnectorPtr->select(columns(&TableSchema::id_,
                                                   &TableSchema::state_,
                                                   &TableSchema::dimension_,
                                                   &TableSchema::created_on_,
                                                   &TableSchema::flag_,
                                                   &TableSchema::index_file_size_,
                                                   &TableSchema::owner_table_,
                                                   &TableSchema::partition_tag_,
                                                   &TableSchema::version_),
                                           where(c(&TableSchema::table_id_) == table_id
G
groot 已提交
616
                                                 and c(&TableSchema::state_) != (int)TableSchema::TO_DELETE));
617

S
starlord 已提交
618
        if (tables.size() > 0) {
619 620 621 622 623 624
            meta::TableSchema table_schema;
            table_schema.id_ = std::get<0>(tables[0]);
            table_schema.table_id_ = table_id;
            table_schema.state_ = std::get<1>(tables[0]);
            table_schema.dimension_ = std::get<2>(tables[0]);
            table_schema.created_on_ = std::get<3>(tables[0]);
S
starlord 已提交
625
            table_schema.flag_ = std::get<4>(tables[0]);
626
            table_schema.index_file_size_ = std::get<5>(tables[0]);
G
groot 已提交
627 628 629
            table_schema.owner_table_ = std::get<6>(tables[0]);
            table_schema.partition_tag_ = std::get<7>(tables[0]);
            table_schema.version_ = std::get<8>(tables[0]);
630
            table_schema.engine_type_ = index.engine_type_;
S
starlord 已提交
631 632
            table_schema.nlist_ = index.nlist_;
            table_schema.metric_type_ = index.metric_type_;
633 634 635

            ConnectorPtr->update(table_schema);
        } else {
S
starlord 已提交
636
            return Status(DB_NOT_FOUND, "Table " + table_id + " not found");
637 638
        }

G
groot 已提交
639 640 641
        //set all backup file to raw
        ConnectorPtr->update_all(
            set(
G
groot 已提交
642
                c(&TableFileSchema::file_type_) = (int)TableFileSchema::RAW,
G
groot 已提交
643 644 645
                c(&TableFileSchema::updated_time_) = utils::GetMicroSecTimeStamp()),
            where(
                c(&TableFileSchema::table_id_) == table_id and
G
groot 已提交
646
                c(&TableFileSchema::file_type_) == (int)TableFileSchema::BACKUP));
647

648
        ENGINE_LOG_DEBUG << "Successfully update table index, table id = " << table_id;
Y
youny626 已提交
649
    } catch (std::exception& e) {
650
        std::string msg = "Encounter exception when update table index: table_id = " + table_id;
S
starlord 已提交
651
        return HandleException(msg, e.what());
652
    }
S
starlord 已提交
653 654 655 656

    return Status::OK();
}

S
starlord 已提交
657
Status
G
groot 已提交
658
SqliteMetaImpl::UpdateTableFilesToIndex(const std::string& table_id) {
S
starlord 已提交
659
    try {
Y
Yu Kun 已提交
660
        server::MetricCollector metric;
S
starlord 已提交
661

G
groot 已提交
662 663 664 665 666
        //multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

        ConnectorPtr->update_all(
            set(
G
groot 已提交
667
                c(&TableFileSchema::file_type_) = (int)TableFileSchema::TO_INDEX),
G
groot 已提交
668 669
            where(
                c(&TableFileSchema::table_id_) == table_id and
G
groot 已提交
670 671
                c(&TableFileSchema::row_count_) >= meta::BUILD_INDEX_THRESHOLD and
                c(&TableFileSchema::file_type_) == (int)TableFileSchema::RAW));
G
groot 已提交
672 673

        ENGINE_LOG_DEBUG << "Update files to to_index, table id = " << table_id;
G
groot 已提交
674
    } catch (std::exception& e) {
G
groot 已提交
675
        return HandleException("Encounter exception when update table files to to_index", e.what());
S
starlord 已提交
676 677
    }

678 679 680
    return Status::OK();
}

S
starlord 已提交
681
Status
Y
youny626 已提交
682
SqliteMetaImpl::DescribeTableIndex(const std::string& table_id, TableIndex& index) {
683
    try {
Y
Yu Kun 已提交
684
        server::MetricCollector metric;
685

G
groot 已提交
686 687 688 689
        auto groups = ConnectorPtr->select(columns(&TableSchema::engine_type_,
                                                   &TableSchema::nlist_,
                                                   &TableSchema::metric_type_),
                                           where(c(&TableSchema::table_id_) == table_id
G
groot 已提交
690
                                                 and c(&TableSchema::state_) != (int)TableSchema::TO_DELETE));
691 692 693

        if (groups.size() == 1) {
            index.engine_type_ = std::get<0>(groups[0]);
S
starlord 已提交
694
            index.nlist_ = std::get<1>(groups[0]);
S
starlord 已提交
695
            index.metric_type_ = std::get<2>(groups[0]);
696
        } else {
S
starlord 已提交
697
            return Status(DB_NOT_FOUND, "Table " + table_id + " not found");
698
        }
Y
youny626 已提交
699
    } catch (std::exception& e) {
S
starlord 已提交
700
        return HandleException("Encounter exception when describe index", e.what());
701 702 703 704 705
    }

    return Status::OK();
}

S
starlord 已提交
706
Status
Y
youny626 已提交
707
SqliteMetaImpl::DropTableIndex(const std::string& table_id) {
708
    try {
Y
Yu Kun 已提交
709
        server::MetricCollector metric;
710

Y
youny626 已提交
711
        // multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
712 713
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

G
groot 已提交
714 715 716
        //soft delete index files
        ConnectorPtr->update_all(
            set(
G
groot 已提交
717
                c(&TableFileSchema::file_type_) = (int)TableFileSchema::TO_DELETE,
G
groot 已提交
718 719 720
                c(&TableFileSchema::updated_time_) = utils::GetMicroSecTimeStamp()),
            where(
                c(&TableFileSchema::table_id_) == table_id and
G
groot 已提交
721
                c(&TableFileSchema::file_type_) == (int)TableFileSchema::INDEX));
G
groot 已提交
722 723

        //set all backup file to raw
724
        ConnectorPtr->update_all(
G
groot 已提交
725
            set(
G
groot 已提交
726
                c(&TableFileSchema::file_type_) = (int)TableFileSchema::RAW,
G
groot 已提交
727 728 729
                c(&TableFileSchema::updated_time_) = utils::GetMicroSecTimeStamp()),
            where(
                c(&TableFileSchema::table_id_) == table_id and
G
groot 已提交
730
                c(&TableFileSchema::file_type_) == (int)TableFileSchema::BACKUP));
G
groot 已提交
731 732 733 734 735 736

        //set table index type to raw
        ConnectorPtr->update_all(
            set(
                c(&TableSchema::engine_type_) = DEFAULT_ENGINE_TYPE,
                c(&TableSchema::nlist_) = DEFAULT_NLIST,
S
starlord 已提交
737
                c(&TableSchema::metric_type_) = DEFAULT_METRIC_TYPE),
G
groot 已提交
738 739
            where(
                c(&TableSchema::table_id_) == table_id));
740

741
        ENGINE_LOG_DEBUG << "Successfully drop table index, table id = " << table_id;
G
groot 已提交
742
    } catch (std::exception& e) {
S
starlord 已提交
743
        return HandleException("Encounter exception when delete table index files", e.what());
744 745 746 747 748
    }

    return Status::OK();
}

S
starlord 已提交
749
Status
G
groot 已提交
750 751 752
SqliteMetaImpl::CreatePartition(const std::string& table_id,
                                const std::string& partition_name,
                                const std::string& tag) {
G
groot 已提交
753
    server::MetricCollector metric;
754

G
groot 已提交
755 756 757 758 759
    TableSchema table_schema;
    table_schema.table_id_ = table_id;
    auto status = DescribeTable(table_schema);
    if (!status.ok()) {
        return status;
G
groot 已提交
760
    }
G
groot 已提交
761

G
groot 已提交
762
    // not allow create partition under partition
G
groot 已提交
763
    if (!table_schema.owner_table_.empty()) {
G
groot 已提交
764
        return Status(DB_ERROR, "Nested partition is not allowed");
G
groot 已提交
765
    }
G
groot 已提交
766

767 768 769 770 771 772 773 774
    // trim side-blank of tag, only compare valid characters
    // for example: " ab cd " is treated as "ab cd"
    std::string valid_tag = tag;
    server::StringHelpFunctions::TrimStringBlank(valid_tag);

    // not allow duplicated partition
    std::string exist_partition;
    GetPartitionName(table_id, valid_tag, exist_partition);
G
groot 已提交
775
    if (!exist_partition.empty()) {
G
groot 已提交
776
        return Status(DB_ERROR, "Duplicate partition is not allowed");
777
    }
G
groot 已提交
778

779 780
    if (partition_name == "") {
        // generate unique partition name
G
groot 已提交
781 782 783
        NextTableId(table_schema.table_id_);
    } else {
        table_schema.table_id_ = partition_name;
X
Xu Peng 已提交
784
    }
G
groot 已提交
785

G
groot 已提交
786 787 788 789
    table_schema.id_ = -1;
    table_schema.flag_ = 0;
    table_schema.created_on_ = utils::GetMicroSecTimeStamp();
    table_schema.owner_table_ = table_id;
790 791 792 793 794 795
    table_schema.partition_tag_ = valid_tag;

    status = CreateTable(table_schema);
    if (status.code() == DB_ALREADY_EXIST) {
        return Status(DB_ALREADY_EXIST, "Partition already exists");
    }
G
groot 已提交
796

797
    return status;
X
Xu Peng 已提交
798 799
}

S
starlord 已提交
800
Status
G
groot 已提交
801 802 803
SqliteMetaImpl::DropPartition(const std::string& partition_name) {
    return DropTable(partition_name);
}
804

G
groot 已提交
805 806
Status
SqliteMetaImpl::ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partiton_schema_array) {
G
groot 已提交
807
    try {
Y
Yu Kun 已提交
808
        server::MetricCollector metric;
G
groot 已提交
809

G
groot 已提交
810
        auto partitions = ConnectorPtr->select(columns(&TableSchema::table_id_),
G
groot 已提交
811 812 813
                                               where(c(&TableSchema::owner_table_) == table_id
                                                     and c(&TableSchema::state_) != (int)TableSchema::TO_DELETE));
        for (size_t i = 0; i < partitions.size(); i++) {
G
groot 已提交
814 815 816 817 818 819
            std::string partition_name = std::get<0>(partitions[i]);
            meta::TableSchema partition_schema;
            partition_schema.table_id_ = partition_name;
            DescribeTable(partition_schema);
            partiton_schema_array.emplace_back(partition_schema);
        }
G
groot 已提交
820
    } catch (std::exception& e) {
G
groot 已提交
821
        return HandleException("Encounter exception when show partitions", e.what());
822 823
    }

X
Xu Peng 已提交
824
    return Status::OK();
X
Xu Peng 已提交
825 826
}

S
starlord 已提交
827
Status
G
groot 已提交
828
SqliteMetaImpl::GetPartitionName(const std::string& table_id, const std::string& tag, std::string& partition_name) {
829
    try {
Y
Yu Kun 已提交
830
        server::MetricCollector metric;
G
groot 已提交
831

832 833 834 835 836
        // trim side-blank of tag, only compare valid characters
        // for example: " ab cd " is treated as "ab cd"
        std::string valid_tag = tag;
        server::StringHelpFunctions::TrimStringBlank(valid_tag);

G
groot 已提交
837
        auto name = ConnectorPtr->select(columns(&TableSchema::table_id_),
G
groot 已提交
838 839
                                         where(c(&TableSchema::owner_table_) == table_id
                                               and c(&TableSchema::partition_tag_) == valid_tag));
G
groot 已提交
840 841 842
        if (name.size() > 0) {
            partition_name = std::get<0>(name[0]);
        } else {
843
            return Status(DB_NOT_FOUND, "Table " + table_id + "'s partition " + valid_tag + " not found");
844
        }
G
groot 已提交
845
    } catch (std::exception& e) {
G
groot 已提交
846
        return HandleException("Encounter exception when get partition name", e.what());
X
Xu Peng 已提交
847
    }
G
groot 已提交
848 849

    return Status::OK();
X
Xu Peng 已提交
850 851
}

S
starlord 已提交
852
Status
G
groot 已提交
853 854 855
SqliteMetaImpl::FilesToSearch(const std::string& table_id,
                              const std::vector<size_t>& ids,
                              const DatesT& dates,
Y
youny626 已提交
856
                              DatePartionedTableFilesSchema& files) {
X
xj.lin 已提交
857
    files.clear();
Y
Yu Kun 已提交
858
    server::MetricCollector metric;
X
xj.lin 已提交
859 860

    try {
Y
youny626 已提交
861 862 863 864
        auto select_columns =
            columns(&TableFileSchema::id_, &TableFileSchema::table_id_, &TableFileSchema::file_id_,
                    &TableFileSchema::file_type_, &TableFileSchema::file_size_, &TableFileSchema::row_count_,
                    &TableFileSchema::date_, &TableFileSchema::engine_type_);
X
xj.lin 已提交
865 866

        auto match_tableid = c(&TableFileSchema::table_id_) == table_id;
X
xj.lin 已提交
867

Y
youny626 已提交
868 869
        std::vector<int> file_types = {(int)TableFileSchema::RAW, (int)TableFileSchema::TO_INDEX,
                                       (int)TableFileSchema::INDEX};
S
starlord 已提交
870
        auto match_type = in(&TableFileSchema::file_type_, file_types);
X
xj.lin 已提交
871 872 873 874

        TableSchema table_schema;
        table_schema.table_id_ = table_id;
        auto status = DescribeTable(table_schema);
Y
youny626 已提交
875 876 877
        if (!status.ok()) {
            return status;
        }
X
xj.lin 已提交
878

Y
youny626 已提交
879 880
        // sqlite_orm has a bug, 'in' statement cannot handle too many elements
        // so we split one query into multi-queries, this is a work-around!!
881 882 883
        std::vector<DatesT> split_dates;
        split_dates.push_back(DatesT());
        const size_t batch_size = 30;
Y
youny626 已提交
884
        for (DateT date : dates) {
885 886
            DatesT& last_batch = *split_dates.rbegin();
            last_batch.push_back(date);
Y
youny626 已提交
887
            if (last_batch.size() > batch_size) {
888 889 890 891
                split_dates.push_back(DatesT());
            }
        }

Y
youny626 已提交
892
        // perform query
893
        decltype(ConnectorPtr->select(select_columns)) selected;
894
        if (dates.empty() && ids.empty()) {
X
xj.lin 已提交
895
            auto filter = where(match_tableid and match_type);
896
            selected = ConnectorPtr->select(select_columns, filter);
897
        } else if (dates.empty() && !ids.empty()) {
X
xj.lin 已提交
898
            auto match_fileid = in(&TableFileSchema::id_, ids);
X
xj.lin 已提交
899
            auto filter = where(match_tableid and match_fileid and match_type);
900
            selected = ConnectorPtr->select(select_columns, filter);
901
        } else if (!dates.empty() && ids.empty()) {
Y
youny626 已提交
902 903
            for (auto& batch_dates : split_dates) {
                if (batch_dates.empty()) {
904 905 906 907 908
                    continue;
                }
                auto match_date = in(&TableFileSchema::date_, batch_dates);
                auto filter = where(match_tableid and match_date and match_type);
                auto batch_selected = ConnectorPtr->select(select_columns, filter);
Y
youny626 已提交
909
                for (auto& file : batch_selected) {
910 911 912 913 914
                    selected.push_back(file);
                }
            }

        } else if (!dates.empty() && !ids.empty()) {
Y
youny626 已提交
915 916
            for (auto& batch_dates : split_dates) {
                if (batch_dates.empty()) {
917 918 919 920 921 922
                    continue;
                }
                auto match_fileid = in(&TableFileSchema::id_, ids);
                auto match_date = in(&TableFileSchema::date_, batch_dates);
                auto filter = where(match_tableid and match_fileid and match_date and match_type);
                auto batch_selected = ConnectorPtr->select(select_columns, filter);
Y
youny626 已提交
923
                for (auto& file : batch_selected) {
924 925 926
                    selected.push_back(file);
                }
            }
X
xj.lin 已提交
927 928
        }

S
starlord 已提交
929
        Status ret;
X
xj.lin 已提交
930
        TableFileSchema table_file;
Y
youny626 已提交
931
        for (auto& file : selected) {
X
xj.lin 已提交
932 933 934 935
            table_file.id_ = std::get<0>(file);
            table_file.table_id_ = std::get<1>(file);
            table_file.file_id_ = std::get<2>(file);
            table_file.file_type_ = std::get<3>(file);
S
starlord 已提交
936 937 938 939
            table_file.file_size_ = std::get<4>(file);
            table_file.row_count_ = std::get<5>(file);
            table_file.date_ = std::get<6>(file);
            table_file.engine_type_ = std::get<7>(file);
X
xj.lin 已提交
940
            table_file.dimension_ = table_schema.dimension_;
S
starlord 已提交
941
            table_file.index_file_size_ = table_schema.index_file_size_;
942
            table_file.nlist_ = table_schema.nlist_;
S
starlord 已提交
943 944
            table_file.metric_type_ = table_schema.metric_type_;

S
starlord 已提交
945
            auto status = utils::GetTableFilePath(options_, table_file);
S
starlord 已提交
946
            if (!status.ok()) {
S
starlord 已提交
947
                ret = status;
S
starlord 已提交
948 949
            }

X
xj.lin 已提交
950 951 952 953 954 955
            auto dateItr = files.find(table_file.date_);
            if (dateItr == files.end()) {
                files[table_file.date_] = TableFilesSchema();
            }
            files[table_file.date_].push_back(table_file);
        }
S
starlord 已提交
956
        if (files.empty()) {
S
starlord 已提交
957
            ENGINE_LOG_ERROR << "No file to search for table: " << table_id;
958
        }
S
starlord 已提交
959

S
starlord 已提交
960
        if (selected.size() > 0) {
961 962
            ENGINE_LOG_DEBUG << "Collect " << selected.size() << " to-search files";
        }
S
starlord 已提交
963
        return ret;
Y
youny626 已提交
964
    } catch (std::exception& e) {
S
starlord 已提交
965
        return HandleException("Encounter exception when iterate index files", e.what());
X
xj.lin 已提交
966 967 968
    }
}

S
starlord 已提交
969
Status
Y
youny626 已提交
970
SqliteMetaImpl::FilesToMerge(const std::string& table_id, DatePartionedTableFilesSchema& files) {
X
Xu Peng 已提交
971
    files.clear();
X
Xu Peng 已提交
972

973
    try {
Y
Yu Kun 已提交
974
        server::MetricCollector metric;
G
groot 已提交
975

Y
youny626 已提交
976
        // check table existence
S
starlord 已提交
977 978 979 980 981 982 983
        TableSchema table_schema;
        table_schema.table_id_ = table_id;
        auto status = DescribeTable(table_schema);
        if (!status.ok()) {
            return status;
        }

Y
youny626 已提交
984 985 986 987 988 989 990 991
        // get files to merge
        auto selected = ConnectorPtr->select(
            columns(&TableFileSchema::id_, &TableFileSchema::table_id_, &TableFileSchema::file_id_,
                    &TableFileSchema::file_type_, &TableFileSchema::file_size_, &TableFileSchema::row_count_,
                    &TableFileSchema::date_, &TableFileSchema::created_on_),
            where(c(&TableFileSchema::file_type_) == (int)TableFileSchema::RAW and
                  c(&TableFileSchema::table_id_) == table_id),
            order_by(&TableFileSchema::file_size_).desc());
G
groot 已提交
992

S
starlord 已提交
993
        Status result;
994
        int64_t to_merge_files = 0;
Y
youny626 已提交
995
        for (auto& file : selected) {
S
starlord 已提交
996 997
            TableFileSchema table_file;
            table_file.file_size_ = std::get<4>(file);
S
starlord 已提交
998
            if (table_file.file_size_ >= table_schema.index_file_size_) {
Y
youny626 已提交
999
                continue;  // skip large file
S
starlord 已提交
1000 1001
            }

G
groot 已提交
1002 1003 1004 1005
            table_file.id_ = std::get<0>(file);
            table_file.table_id_ = std::get<1>(file);
            table_file.file_id_ = std::get<2>(file);
            table_file.file_type_ = std::get<3>(file);
S
starlord 已提交
1006 1007 1008
            table_file.row_count_ = std::get<5>(file);
            table_file.date_ = std::get<6>(file);
            table_file.created_on_ = std::get<7>(file);
G
groot 已提交
1009
            table_file.dimension_ = table_schema.dimension_;
S
starlord 已提交
1010
            table_file.index_file_size_ = table_schema.index_file_size_;
1011
            table_file.nlist_ = table_schema.nlist_;
S
starlord 已提交
1012 1013
            table_file.metric_type_ = table_schema.metric_type_;

S
starlord 已提交
1014
            auto status = utils::GetTableFilePath(options_, table_file);
S
starlord 已提交
1015
            if (!status.ok()) {
S
starlord 已提交
1016
                result = status;
S
starlord 已提交
1017 1018
            }

G
groot 已提交
1019
            auto dateItr = files.find(table_file.date_);
1020
            if (dateItr == files.end()) {
G
groot 已提交
1021
                files[table_file.date_] = TableFilesSchema();
1022
            }
1023

G
groot 已提交
1024
            files[table_file.date_].push_back(table_file);
1025
            to_merge_files++;
X
Xu Peng 已提交
1026
        }
S
starlord 已提交
1027

1028 1029
        if (to_merge_files > 0) {
            ENGINE_LOG_TRACE << "Collect " << to_merge_files << " to-merge files";
1030
        }
G
groot 已提交
1031 1032 1033 1034 1035 1036 1037
        return result;
    } catch (std::exception& e) {
        return HandleException("Encounter exception when iterate merge files", e.what());
    }
}

Status
G
groot 已提交
1038
SqliteMetaImpl::FilesToIndex(TableFilesSchema& files) {
G
groot 已提交
1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053
    files.clear();

    try {
        server::MetricCollector metric;

        auto selected = ConnectorPtr->select(columns(&TableFileSchema::id_,
                                                     &TableFileSchema::table_id_,
                                                     &TableFileSchema::file_id_,
                                                     &TableFileSchema::file_type_,
                                                     &TableFileSchema::file_size_,
                                                     &TableFileSchema::row_count_,
                                                     &TableFileSchema::date_,
                                                     &TableFileSchema::engine_type_,
                                                     &TableFileSchema::created_on_),
                                             where(c(&TableFileSchema::file_type_)
G
groot 已提交
1054
                                                   == (int)TableFileSchema::TO_INDEX));
G
groot 已提交
1055 1056 1057 1058 1059

        std::map<std::string, TableSchema> groups;
        TableFileSchema table_file;

        Status ret;
G
groot 已提交
1060
        for (auto& file : selected) {
G
groot 已提交
1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095
            table_file.id_ = std::get<0>(file);
            table_file.table_id_ = std::get<1>(file);
            table_file.file_id_ = std::get<2>(file);
            table_file.file_type_ = std::get<3>(file);
            table_file.file_size_ = std::get<4>(file);
            table_file.row_count_ = std::get<5>(file);
            table_file.date_ = std::get<6>(file);
            table_file.engine_type_ = std::get<7>(file);
            table_file.created_on_ = std::get<8>(file);

            auto status = utils::GetTableFilePath(options_, table_file);
            if (!status.ok()) {
                ret = status;
            }
            auto groupItr = groups.find(table_file.table_id_);
            if (groupItr == groups.end()) {
                TableSchema table_schema;
                table_schema.table_id_ = table_file.table_id_;
                auto status = DescribeTable(table_schema);
                if (!status.ok()) {
                    return status;
                }
                groups[table_file.table_id_] = table_schema;
            }
            table_file.dimension_ = groups[table_file.table_id_].dimension_;
            table_file.index_file_size_ = groups[table_file.table_id_].index_file_size_;
            table_file.nlist_ = groups[table_file.table_id_].nlist_;
            table_file.metric_type_ = groups[table_file.table_id_].metric_type_;
            files.push_back(table_file);
        }

        if (selected.size() > 0) {
            ENGINE_LOG_DEBUG << "Collect " << selected.size() << " to-index files";
        }
        return ret;
G
groot 已提交
1096
    } catch (std::exception& e) {
G
groot 已提交
1097
        return HandleException("Encounter exception when iterate raw files", e.what());
X
Xu Peng 已提交
1098
    }
X
Xu Peng 已提交
1099 1100
}

S
starlord 已提交
1101
Status
G
groot 已提交
1102 1103 1104
SqliteMetaImpl::FilesByType(const std::string& table_id,
                            const std::vector<int>& file_types,
                            TableFilesSchema& table_files) {
G
groot 已提交
1105 1106 1107
    if (file_types.empty()) {
        return Status(DB_ERROR, "file types array is empty");
    }
1108

G
groot 已提交
1109
    try {
G
groot 已提交
1110 1111 1112 1113 1114 1115 1116 1117 1118
        table_files.clear();
        auto selected = ConnectorPtr->select(columns(&TableFileSchema::id_,
                                                     &TableFileSchema::file_id_,
                                                     &TableFileSchema::file_type_,
                                                     &TableFileSchema::file_size_,
                                                     &TableFileSchema::row_count_,
                                                     &TableFileSchema::date_,
                                                     &TableFileSchema::engine_type_,
                                                     &TableFileSchema::created_on_),
G
groot 已提交
1119 1120
                                             where(in(&TableFileSchema::file_type_, file_types)
                                                   and c(&TableFileSchema::table_id_) == table_id));
S
starlord 已提交
1121

G
groot 已提交
1122 1123 1124
        if (selected.size() >= 1) {
            int raw_count = 0, new_count = 0, new_merge_count = 0, new_index_count = 0;
            int to_index_count = 0, index_count = 0, backup_count = 0;
G
groot 已提交
1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138
            for (auto& file : selected) {
                TableFileSchema file_schema;
                file_schema.table_id_ = table_id;
                file_schema.id_ = std::get<0>(file);
                file_schema.file_id_ = std::get<1>(file);
                file_schema.file_type_ = std::get<2>(file);
                file_schema.file_size_ = std::get<3>(file);
                file_schema.row_count_ = std::get<4>(file);
                file_schema.date_ = std::get<5>(file);
                file_schema.engine_type_ = std::get<6>(file);
                file_schema.created_on_ = std::get<7>(file);

                switch (file_schema.file_type_) {
                    case (int)TableFileSchema::RAW:raw_count++;
G
groot 已提交
1139
                        break;
G
groot 已提交
1140
                    case (int)TableFileSchema::NEW:new_count++;
G
groot 已提交
1141
                        break;
G
groot 已提交
1142
                    case (int)TableFileSchema::NEW_MERGE:new_merge_count++;
G
groot 已提交
1143
                        break;
G
groot 已提交
1144
                    case (int)TableFileSchema::NEW_INDEX:new_index_count++;
G
groot 已提交
1145
                        break;
G
groot 已提交
1146
                    case (int)TableFileSchema::TO_INDEX:to_index_count++;
G
groot 已提交
1147
                        break;
G
groot 已提交
1148
                    case (int)TableFileSchema::INDEX:index_count++;
G
groot 已提交
1149
                        break;
G
groot 已提交
1150
                    case (int)TableFileSchema::BACKUP:backup_count++;
G
groot 已提交
1151 1152 1153
                        break;
                    default:break;
                }
G
groot 已提交
1154 1155

                table_files.emplace_back(file_schema);
G
groot 已提交
1156
            }
1157

G
groot 已提交
1158 1159 1160 1161
            ENGINE_LOG_DEBUG << "Table " << table_id << " currently has raw files:" << raw_count
                             << " new files:" << new_count << " new_merge files:" << new_merge_count
                             << " new_index files:" << new_index_count << " to_index files:" << to_index_count
                             << " index files:" << index_count << " backup files:" << backup_count;
X
Xu Peng 已提交
1162
        }
G
groot 已提交
1163
    } catch (std::exception& e) {
G
groot 已提交
1164
        return HandleException("Encounter exception when check non index files", e.what());
X
Xu Peng 已提交
1165
    }
G
groot 已提交
1166
    return Status::OK();
X
Xu Peng 已提交
1167 1168
}

S
starlord 已提交
1169
// TODO(myh): Support swap to cloud storage
S
starlord 已提交
1170 1171
Status
SqliteMetaImpl::Archive() {
Y
youny626 已提交
1172
    auto& criterias = options_.archive_conf_.GetCriterias();
X
Xu Peng 已提交
1173 1174 1175 1176 1177
    if (criterias.size() == 0) {
        return Status::OK();
    }

    for (auto kv : criterias) {
Y
youny626 已提交
1178 1179
        auto& criteria = kv.first;
        auto& limit = kv.second;
G
groot 已提交
1180
        if (criteria == engine::ARCHIVE_CONF_DAYS) {
S
starlord 已提交
1181 1182
            int64_t usecs = limit * D_SEC * US_PS;
            int64_t now = utils::GetMicroSecTimeStamp();
1183
            try {
Y
youny626 已提交
1184
                // multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
1185 1186
                std::lock_guard<std::mutex> meta_lock(meta_mutex_);

G
groot 已提交
1187 1188
                ConnectorPtr->update_all(
                    set(
G
groot 已提交
1189
                        c(&TableFileSchema::file_type_) = (int)TableFileSchema::TO_DELETE),
G
groot 已提交
1190
                    where(
G
groot 已提交
1191 1192 1193
                        c(&TableFileSchema::created_on_) < (int64_t)(now - usecs) and
                        c(&TableFileSchema::file_type_) != (int)TableFileSchema::TO_DELETE));
            } catch (std::exception& e) {
S
starlord 已提交
1194
                return HandleException("Encounter exception when update table files", e.what());
X
Xu Peng 已提交
1195
            }
1196 1197

            ENGINE_LOG_DEBUG << "Archive old files";
X
Xu Peng 已提交
1198
        }
G
groot 已提交
1199
        if (criteria == engine::ARCHIVE_CONF_DISK) {
G
groot 已提交
1200
            uint64_t sum = 0;
X
Xu Peng 已提交
1201
            Size(sum);
X
Xu Peng 已提交
1202

Y
youny626 已提交
1203
            int64_t to_delete = (int64_t)sum - limit * G;
X
Xu Peng 已提交
1204
            DiscardFiles(to_delete);
1205 1206

            ENGINE_LOG_DEBUG << "Archive files to free disk";
X
Xu Peng 已提交
1207 1208 1209 1210 1211 1212
        }
    }

    return Status::OK();
}

S
starlord 已提交
1213
Status
Y
youny626 已提交
1214
SqliteMetaImpl::Size(uint64_t& result) {
X
Xu Peng 已提交
1215
    result = 0;
X
Xu Peng 已提交
1216
    try {
1217
        auto selected = ConnectorPtr->select(columns(sum(&TableFileSchema::file_size_)),
Y
youny626 已提交
1218 1219
                                             where(c(&TableFileSchema::file_type_) != (int)TableFileSchema::TO_DELETE));
        for (auto& total_size : selected) {
1220 1221
            if (!std::get<0>(total_size)) {
                continue;
X
Xu Peng 已提交
1222
            }
Y
youny626 已提交
1223
            result += (uint64_t)(*std::get<0>(total_size));
X
Xu Peng 已提交
1224
        }
Y
youny626 已提交
1225
    } catch (std::exception& e) {
S
starlord 已提交
1226
        return HandleException("Encounter exception when calculte db size", e.what());
X
Xu Peng 已提交
1227 1228 1229 1230 1231
    }

    return Status::OK();
}

S
starlord 已提交
1232
Status
G
groot 已提交
1233
SqliteMetaImpl::CleanUp() {
X
Xu Peng 已提交
1234
    try {
Y
Yu Kun 已提交
1235
        server::MetricCollector metric;
G
groot 已提交
1236

Y
youny626 已提交
1237
        // multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
1238 1239
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

G
groot 已提交
1240
        std::vector<int> file_types = {
G
groot 已提交
1241 1242 1243
            (int)TableFileSchema::NEW,
            (int)TableFileSchema::NEW_INDEX,
            (int)TableFileSchema::NEW_MERGE
G
groot 已提交
1244 1245 1246
        };
        auto files =
            ConnectorPtr->select(columns(&TableFileSchema::id_), where(in(&TableFileSchema::file_type_, file_types)));
1247

G
groot 已提交
1248
        auto commited = ConnectorPtr->transaction([&]() mutable {
G
groot 已提交
1249
            for (auto& file : files) {
G
groot 已提交
1250 1251
                ENGINE_LOG_DEBUG << "Remove table file type as NEW";
                ConnectorPtr->remove<TableFileSchema>(std::get<0>(file));
G
groot 已提交
1252 1253 1254 1255 1256
            }
            return true;
        });

        if (!commited) {
G
groot 已提交
1257
            return HandleException("CleanUp error: sqlite transaction failed");
G
groot 已提交
1258
        }
X
Xu Peng 已提交
1259

G
groot 已提交
1260 1261
        if (files.size() > 0) {
            ENGINE_LOG_DEBUG << "Clean " << files.size() << " files";
G
groot 已提交
1262
        }
G
groot 已提交
1263
    } catch (std::exception& e) {
G
groot 已提交
1264
        return HandleException("Encounter exception when clean table file", e.what());
P
peng.xu 已提交
1265 1266 1267 1268 1269
    }

    return Status::OK();
}

S
starlord 已提交
1270 1271
Status
SqliteMetaImpl::CleanUpFilesWithTTL(uint16_t seconds) {
X
Xu Peng 已提交
1272
    auto now = utils::GetMicroSecTimeStamp();
S
starlord 已提交
1273 1274
    std::set<std::string> table_ids;

Y
youny626 已提交
1275
    // remove to_delete files
1276
    try {
Y
Yu Kun 已提交
1277
        server::MetricCollector metric;
1278

Y
youny626 已提交
1279
        // multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
1280 1281
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

G
groot 已提交
1282 1283 1284 1285 1286 1287
        auto files = ConnectorPtr->select(columns(&TableFileSchema::id_,
                                                  &TableFileSchema::table_id_,
                                                  &TableFileSchema::file_id_,
                                                  &TableFileSchema::date_),
                                          where(
                                              c(&TableFileSchema::file_type_) ==
G
groot 已提交
1288
                                              (int)TableFileSchema::TO_DELETE
G
groot 已提交
1289 1290 1291
                                              and
                                              c(&TableFileSchema::updated_time_)
                                              < now - seconds * US_PS));
1292

G
groot 已提交
1293 1294
        auto commited = ConnectorPtr->transaction([&]() mutable {
            TableFileSchema table_file;
Y
youny626 已提交
1295
            for (auto& file : files) {
G
groot 已提交
1296 1297 1298 1299 1300
                table_file.id_ = std::get<0>(file);
                table_file.table_id_ = std::get<1>(file);
                table_file.file_id_ = std::get<2>(file);
                table_file.date_ = std::get<3>(file);

S
starlord 已提交
1301
                utils::DeleteTableFilePath(options_, table_file);
1302
                ENGINE_LOG_DEBUG << "Removing file id:" << table_file.file_id_ << " location:" << table_file.location_;
G
groot 已提交
1303 1304
                ConnectorPtr->remove<TableFileSchema>(table_file.id_);

S
starlord 已提交
1305
                table_ids.insert(table_file.table_id_);
1306
            }
G
groot 已提交
1307 1308 1309 1310
            return true;
        });

        if (!commited) {
S
starlord 已提交
1311
            return HandleException("CleanUpFilesWithTTL error: sqlite transaction failed");
G
groot 已提交
1312 1313
        }

S
starlord 已提交
1314
        if (files.size() > 0) {
1315 1316
            ENGINE_LOG_DEBUG << "Clean " << files.size() << " files deleted in " << seconds << " seconds";
        }
Y
youny626 已提交
1317
    } catch (std::exception& e) {
S
starlord 已提交
1318
        return HandleException("Encounter exception when clean table files", e.what());
G
groot 已提交
1319 1320
    }

Y
youny626 已提交
1321
    // remove to_delete tables
G
groot 已提交
1322
    try {
Y
Yu Kun 已提交
1323
        server::MetricCollector metric;
G
groot 已提交
1324

Y
youny626 已提交
1325
        // multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
1326 1327
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

Y
youny626 已提交
1328 1329
        auto tables = ConnectorPtr->select(columns(&TableSchema::id_, &TableSchema::table_id_),
                                           where(c(&TableSchema::state_) == (int)TableSchema::TO_DELETE));
G
groot 已提交
1330 1331

        auto commited = ConnectorPtr->transaction([&]() mutable {
Y
youny626 已提交
1332 1333
            for (auto& table : tables) {
                utils::DeleteTablePath(options_, std::get<1>(table), false);  // only delete empty folder
G
groot 已提交
1334
                ConnectorPtr->remove<TableSchema>(std::get<0>(table));
1335
            }
G
groot 已提交
1336 1337 1338 1339 1340

            return true;
        });

        if (!commited) {
S
starlord 已提交
1341
            return HandleException("CleanUpFilesWithTTL error: sqlite transaction failed");
X
Xu Peng 已提交
1342
        }
G
groot 已提交
1343

S
starlord 已提交
1344
        if (tables.size() > 0) {
1345 1346
            ENGINE_LOG_DEBUG << "Remove " << tables.size() << " tables from meta";
        }
Y
youny626 已提交
1347
    } catch (std::exception& e) {
S
starlord 已提交
1348
        return HandleException("Encounter exception when clean table files", e.what());
X
Xu Peng 已提交
1349 1350
    }

Y
youny626 已提交
1351 1352
    // remove deleted table folder
    // don't remove table folder until all its files has been deleted
S
starlord 已提交
1353
    try {
Y
Yu Kun 已提交
1354
        server::MetricCollector metric;
S
starlord 已提交
1355

1356
        int64_t remove_tables = 0;
Y
youny626 已提交
1357
        for (auto& table_id : table_ids) {
S
starlord 已提交
1358 1359
            auto selected = ConnectorPtr->select(columns(&TableFileSchema::file_id_),
                                                 where(c(&TableFileSchema::table_id_) == table_id));
S
starlord 已提交
1360
            if (selected.size() == 0) {
S
starlord 已提交
1361
                utils::DeleteTablePath(options_, table_id);
1362
                remove_tables++;
S
starlord 已提交
1363 1364 1365
            }
        }

1366 1367
        if (remove_tables) {
            ENGINE_LOG_DEBUG << "Remove " << remove_tables << " tables folder";
1368
        }
Y
youny626 已提交
1369
    } catch (std::exception& e) {
S
starlord 已提交
1370
        return HandleException("Encounter exception when delete table folder", e.what());
S
starlord 已提交
1371 1372
    }

X
Xu Peng 已提交
1373 1374 1375
    return Status::OK();
}

S
starlord 已提交
1376
Status
G
groot 已提交
1377
SqliteMetaImpl::Count(const std::string& table_id, uint64_t& result) {
1378
    try {
Y
Yu Kun 已提交
1379
        server::MetricCollector metric;
1380

Y
youny626 已提交
1381 1382 1383 1384 1385
        std::vector<int> file_types = {(int)TableFileSchema::RAW, (int)TableFileSchema::TO_INDEX,
                                       (int)TableFileSchema::INDEX};
        auto selected = ConnectorPtr->select(
            columns(&TableFileSchema::row_count_),
            where(in(&TableFileSchema::file_type_, file_types) and c(&TableFileSchema::table_id_) == table_id));
1386

1387
        TableSchema table_schema;
G
groot 已提交
1388
        table_schema.table_id_ = table_id;
X
Xu Peng 已提交
1389
        auto status = DescribeTable(table_schema);
1390

1391 1392 1393 1394 1395
        if (!status.ok()) {
            return status;
        }

        result = 0;
Y
youny626 已提交
1396
        for (auto& file : selected) {
1397 1398
            result += std::get<0>(file);
        }
Y
youny626 已提交
1399
    } catch (std::exception& e) {
S
starlord 已提交
1400
        return HandleException("Encounter exception when calculate table file size", e.what());
X
Xu Peng 已提交
1401 1402 1403 1404
    }
    return Status::OK();
}

S
starlord 已提交
1405 1406
Status
SqliteMetaImpl::DropAll() {
S
starlord 已提交
1407 1408 1409
    ENGINE_LOG_DEBUG << "Drop all sqlite meta";

    try {
1410 1411
        ConnectorPtr->drop_table(META_TABLES);
        ConnectorPtr->drop_table(META_TABLEFILES);
Y
youny626 已提交
1412
    } catch (std::exception& e) {
S
starlord 已提交
1413
        return HandleException("Encounter exception when drop all meta", e.what());
S
starlord 已提交
1414
    }
S
starlord 已提交
1415

X
Xu Peng 已提交
1416 1417 1418
    return Status::OK();
}

G
groot 已提交
1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436
Status
SqliteMetaImpl::DiscardFiles(int64_t to_discard_size) {
    if (to_discard_size <= 0) {
        return Status::OK();
    }

    ENGINE_LOG_DEBUG << "About to discard size=" << to_discard_size;

    try {
        server::MetricCollector metric;

        //multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
        std::lock_guard<std::mutex> meta_lock(meta_mutex_);

        auto commited = ConnectorPtr->transaction([&]() mutable {
            auto selected = ConnectorPtr->select(columns(&TableFileSchema::id_,
                                                         &TableFileSchema::file_size_),
                                                 where(c(&TableFileSchema::file_type_)
G
groot 已提交
1437
                                                       != (int)TableFileSchema::TO_DELETE),
G
groot 已提交
1438 1439 1440 1441 1442 1443
                                                 order_by(&TableFileSchema::id_),
                                                 limit(10));

            std::vector<int> ids;
            TableFileSchema table_file;

G
groot 已提交
1444
            for (auto& file : selected) {
G
groot 已提交
1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459
                if (to_discard_size <= 0) break;
                table_file.id_ = std::get<0>(file);
                table_file.file_size_ = std::get<1>(file);
                ids.push_back(table_file.id_);
                ENGINE_LOG_DEBUG << "Discard table_file.id=" << table_file.file_id_
                                 << " table_file.size=" << table_file.file_size_;
                to_discard_size -= table_file.file_size_;
            }

            if (ids.size() == 0) {
                return true;
            }

            ConnectorPtr->update_all(
                set(
G
groot 已提交
1460
                    c(&TableFileSchema::file_type_) = (int)TableFileSchema::TO_DELETE,
G
groot 已提交
1461 1462 1463 1464 1465 1466 1467 1468 1469 1470
                    c(&TableFileSchema::updated_time_) = utils::GetMicroSecTimeStamp()),
                where(
                    in(&TableFileSchema::id_, ids)));

            return true;
        });

        if (!commited) {
            return HandleException("DiscardFiles error: sqlite transaction failed");
        }
G
groot 已提交
1471
    } catch (std::exception& e) {
G
groot 已提交
1472 1473 1474 1475 1476 1477 1478 1479 1480 1481
        return HandleException("Encounter exception when discard table file", e.what());
    }

    return DiscardFiles(to_discard_size);
}

} // namespace meta
} // namespace engine
} // namespace milvus