MySQLMetaImpl.cpp 80.7 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/MySQLMetaImpl.h"
S
starlord 已提交
19
#include "MetaConsts.h"
S
starlord 已提交
20 21
#include "db/IDGenerator.h"
#include "db/Utils.h"
Z
update  
zhiru 已提交
22
#include "metrics/Metrics.h"
S
starlord 已提交
23 24
#include "utils/Exception.h"
#include "utils/Log.h"
Z
update  
zhiru 已提交
25

S
starlord 已提交
26 27
#include <mysql++/mysql++.h>
#include <string.h>
Z
update  
zhiru 已提交
28
#include <unistd.h>
S
starlord 已提交
29
#include <boost/filesystem.hpp>
Z
update  
zhiru 已提交
30 31
#include <chrono>
#include <fstream>
S
starlord 已提交
32 33 34
#include <iostream>
#include <map>
#include <mutex>
Z
update  
zhiru 已提交
35
#include <regex>
S
starlord 已提交
36 37
#include <set>
#include <sstream>
Z
update  
zhiru 已提交
38
#include <string>
Z
zhiru 已提交
39
#include <thread>
Z
update  
zhiru 已提交
40 41 42 43 44

namespace milvus {
namespace engine {
namespace meta {

45
namespace {
Z
update  
zhiru 已提交
46

S
starlord 已提交
47
Status
S
starlord 已提交
48
HandleException(const std::string& desc, const char* what = nullptr) {
S
starlord 已提交
49
    if (what == nullptr) {
S
starlord 已提交
50 51 52
        ENGINE_LOG_ERROR << desc;
        return Status(DB_META_TRANSACTION_FAILED, desc);
    }
S
starlord 已提交
53 54 55 56

    std::string msg = desc + ":" + what;
    ENGINE_LOG_ERROR << msg;
    return Status(DB_META_TRANSACTION_FAILED, msg);
57
}
Z
update  
zhiru 已提交
58

59
class MetaField {
S
starlord 已提交
60
 public:
S
starlord 已提交
61 62
    MetaField(const std::string& name, const std::string& type, const std::string& setting)
        : name_(name), type_(type), setting_(setting) {
63 64
    }

S
starlord 已提交
65 66
    std::string
    name() const {
67 68 69
        return name_;
    }

S
starlord 已提交
70 71
    std::string
    ToString() const {
72 73 74 75 76
        return name_ + " " + type_ + " " + setting_;
    }

    // mysql field type has additional information. for instance, a filed type is defined as 'BIGINT'
    // we get the type from sql is 'bigint(20)', so we need to ignore the '(20)'
S
starlord 已提交
77 78
    bool
    IsEqual(const MetaField& field) const {
79 80 81
        size_t name_len_min = field.name_.length() > name_.length() ? name_.length() : field.name_.length();
        size_t type_len_min = field.type_.length() > type_.length() ? type_.length() : field.type_.length();
        return strncasecmp(field.name_.c_str(), name_.c_str(), name_len_min) == 0 &&
S
starlord 已提交
82
               strncasecmp(field.type_.c_str(), type_.c_str(), type_len_min) == 0;
83 84
    }

S
starlord 已提交
85
 private:
86 87 88 89 90 91 92
    std::string name_;
    std::string type_;
    std::string setting_;
};

using MetaFields = std::vector<MetaField>;
class MetaSchema {
S
starlord 已提交
93
 public:
S
starlord 已提交
94
    MetaSchema(const std::string& name, const MetaFields& fields) : name_(name), fields_(fields) {
95 96
    }

S
starlord 已提交
97 98
    std::string
    name() const {
99 100 101
        return name_;
    }

S
starlord 已提交
102 103
    std::string
    ToString() const {
104
        std::string result;
S
starlord 已提交
105
        for (auto& field : fields_) {
S
starlord 已提交
106
            if (!result.empty()) {
107 108 109 110 111 112 113
                result += ",";
            }
            result += field.ToString();
        }
        return result;
    }

S
starlord 已提交
114 115 116 117
    // if the outer fields contains all this MetaSchema fields, return true
    // otherwise return false
    bool
    IsEqual(const MetaFields& fields) const {
118
        std::vector<std::string> found_field;
S
starlord 已提交
119 120
        for (const auto& this_field : fields_) {
            for (const auto& outer_field : fields) {
S
starlord 已提交
121
                if (this_field.IsEqual(outer_field)) {
122 123 124 125 126 127 128 129 130
                    found_field.push_back(this_field.name());
                    break;
                }
            }
        }

        return found_field.size() == fields_.size();
    }

S
starlord 已提交
131
 private:
132 133 134 135
    std::string name_;
    MetaFields fields_;
};

S
starlord 已提交
136
// Tables schema
137
static const MetaSchema TABLES_SCHEMA(META_TABLES, {
S
starlord 已提交
138 139 140 141 142 143 144 145 146 147
                                                       MetaField("id", "BIGINT", "PRIMARY KEY AUTO_INCREMENT"),
                                                       MetaField("table_id", "VARCHAR(255)", "UNIQUE NOT NULL"),
                                                       MetaField("state", "INT", "NOT NULL"),
                                                       MetaField("dimension", "SMALLINT", "NOT NULL"),
                                                       MetaField("created_on", "BIGINT", "NOT NULL"),
                                                       MetaField("flag", "BIGINT", "DEFAULT 0 NOT NULL"),
                                                       MetaField("index_file_size", "BIGINT", "DEFAULT 1024 NOT NULL"),
                                                       MetaField("engine_type", "INT", "DEFAULT 1 NOT NULL"),
                                                       MetaField("nlist", "INT", "DEFAULT 16384 NOT NULL"),
                                                       MetaField("metric_type", "INT", "DEFAULT 1 NOT NULL"),
G
groot 已提交
148 149 150 151
                                                       MetaField("owner_table", "VARCHAR(255)", "NOT NULL"),
                                                       MetaField("partition_tag", "VARCHAR(255)", "NOT NULL"),
                                                       MetaField("version", "VARCHAR(64)",
                                                                 std::string("DEFAULT '") + CURRENT_VERSION + "'"),
S
starlord 已提交
152 153 154
                                                   });

// TableFiles schema
J
JinHai-CN 已提交
155 156 157 158 159 160 161 162 163 164 165 166
static const MetaSchema TABLEFILES_SCHEMA(META_TABLEFILES, {
                                                               MetaField("id", "BIGINT", "PRIMARY KEY AUTO_INCREMENT"),
                                                               MetaField("table_id", "VARCHAR(255)", "NOT NULL"),
                                                               MetaField("engine_type", "INT", "DEFAULT 1 NOT NULL"),
                                                               MetaField("file_id", "VARCHAR(255)", "NOT NULL"),
                                                               MetaField("file_type", "INT", "DEFAULT 0 NOT NULL"),
                                                               MetaField("file_size", "BIGINT", "DEFAULT 0 NOT NULL"),
                                                               MetaField("row_count", "BIGINT", "DEFAULT 0 NOT NULL"),
                                                               MetaField("updated_time", "BIGINT", "NOT NULL"),
                                                               MetaField("created_on", "BIGINT", "NOT NULL"),
                                                               MetaField("date", "INT", "DEFAULT -1 NOT NULL"),
                                                           });
S
starlord 已提交
167 168

}  // namespace
169

170
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
S
starlord 已提交
171
MySQLMetaImpl::MySQLMetaImpl(const DBMetaOptions& options, const int& mode) : options_(options), mode_(mode) {
172 173 174 175 176 177
    Initialize();
}

MySQLMetaImpl::~MySQLMetaImpl() {
}

S
starlord 已提交
178
Status
S
starlord 已提交
179
MySQLMetaImpl::NextTableId(std::string& table_id) {
G
groot 已提交
180
    std::lock_guard<std::mutex> lock(genid_mutex_);  // avoid duplicated id
181 182 183 184 185 186 187
    std::stringstream ss;
    SimpleIDGenerator g;
    ss << g.GetNextIDNumber();
    table_id = ss.str();
    return Status::OK();
}

S
starlord 已提交
188
Status
S
starlord 已提交
189
MySQLMetaImpl::NextFileId(std::string& file_id) {
G
groot 已提交
190
    std::lock_guard<std::mutex> lock(genid_mutex_);  // avoid duplicated id
191 192 193 194 195 196 197
    std::stringstream ss;
    SimpleIDGenerator g;
    ss << g.GetNextIDNumber();
    file_id = ss.str();
    return Status::OK();
}

S
starlord 已提交
198 199 200
void
MySQLMetaImpl::ValidateMetaSchema() {
    if (nullptr == mysql_connection_pool_) {
201 202 203
        return;
    }

S
starlord 已提交
204
    mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
205 206 207 208
    if (connectionPtr == nullptr) {
        return;
    }

S
starlord 已提交
209
    auto validate_func = [&](const MetaSchema& schema) {
S
starlord 已提交
210
        mysqlpp::Query query_statement = connectionPtr->query();
211 212 213 214 215
        query_statement << "DESC " << schema.name() << ";";

        MetaFields exist_fields;

        try {
S
starlord 已提交
216
            mysqlpp::StoreQueryResult res = query_statement.store();
217
            for (size_t i = 0; i < res.num_rows(); i++) {
S
starlord 已提交
218
                const mysqlpp::Row& row = res[i];
219 220 221 222 223 224
                std::string name, type;
                row["Field"].to_string(name);
                row["Type"].to_string(type);

                exist_fields.push_back(MetaField(name, type, ""));
            }
S
starlord 已提交
225
        } catch (std::exception& e) {
226 227 228
            ENGINE_LOG_DEBUG << "Meta table '" << schema.name() << "' not exist and will be created";
        }

S
starlord 已提交
229
        if (exist_fields.empty()) {
230 231 232 233 234 235
            return true;
        }

        return schema.IsEqual(exist_fields);
    };

S
starlord 已提交
236
    // verify Tables
237 238 239 240
    if (!validate_func(TABLES_SCHEMA)) {
        throw Exception(DB_INCOMPATIB_META, "Meta Tables schema is created by Milvus old version");
    }

S
starlord 已提交
241
    // verufy TableFiles
242 243 244 245 246
    if (!validate_func(TABLEFILES_SCHEMA)) {
        throw Exception(DB_INCOMPATIB_META, "Meta TableFiles schema is created by Milvus old version");
    }
}

S
starlord 已提交
247 248
Status
MySQLMetaImpl::Initialize() {
S
starlord 已提交
249
    // step 1: create db root path
S
starlord 已提交
250 251
    if (!boost::filesystem::is_directory(options_.path_)) {
        auto ret = boost::filesystem::create_directory(options_.path_);
252
        if (!ret) {
S
starlord 已提交
253
            std::string msg = "Failed to create db directory " + options_.path_;
S
starlord 已提交
254 255
            ENGINE_LOG_ERROR << msg;
            return Status(DB_META_TRANSACTION_FAILED, msg);
Z
update  
zhiru 已提交
256 257 258
        }
    }

S
starlord 已提交
259
    std::string uri = options_.backend_uri_;
260

S
starlord 已提交
261
    // step 2: parse and check meta uri
262 263
    utils::MetaUriInfo uri_info;
    auto status = utils::ParseMetaUri(uri, uri_info);
S
starlord 已提交
264
    if (!status.ok()) {
265 266 267 268 269 270 271 272 273 274 275
        std::string msg = "Wrong URI format: " + uri;
        ENGINE_LOG_ERROR << msg;
        throw Exception(DB_INVALID_META_URI, msg);
    }

    if (strcasecmp(uri_info.dialect_.c_str(), "mysql") != 0) {
        std::string msg = "URI's dialect is not MySQL";
        ENGINE_LOG_ERROR << msg;
        throw Exception(DB_INVALID_META_URI, msg);
    }

S
starlord 已提交
276
    // step 3: connect mysql
277 278 279 280 281 282 283
    int thread_hint = std::thread::hardware_concurrency();
    int max_pool_size = (thread_hint == 0) ? 8 : thread_hint;
    unsigned int port = 0;
    if (!uri_info.port_.empty()) {
        port = std::stoi(uri_info.port_);
    }

S
starlord 已提交
284 285
    mysql_connection_pool_ = std::make_shared<MySQLConnectionPool>(
        uri_info.db_name_, uri_info.username_, uri_info.password_, uri_info.host_, port, max_pool_size);
286 287
    ENGINE_LOG_DEBUG << "MySQL connection pool: maximum pool size = " << std::to_string(max_pool_size);

S
starlord 已提交
288
    // step 4: validate to avoid open old version schema
289 290
    ValidateMetaSchema();

S
starlord 已提交
291
    // step 5: create meta tables
292
    try {
Y
yudong.cai 已提交
293
        if (mode_ != DBOptions::MODE::CLUSTER_READONLY) {
294
            CleanUp();
295 296
        }

297
        {
S
starlord 已提交
298
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
299

300
            if (connectionPtr == nullptr) {
G
groot 已提交
301
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
302
            }
303

304 305 306
            if (!connectionPtr->thread_aware()) {
                ENGINE_LOG_ERROR << "MySQL++ wasn't built with thread awareness! Can't run without it.";
                return Status(DB_ERROR, "MySQL++ wasn't built with thread awareness! Can't run without it.");
307
            }
S
starlord 已提交
308
            mysqlpp::Query InitializeQuery = connectionPtr->query();
309

S
starlord 已提交
310 311
            InitializeQuery << "CREATE TABLE IF NOT EXISTS " << TABLES_SCHEMA.name() << " ("
                            << TABLES_SCHEMA.ToString() + ");";
312

313
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::Initialize: " << InitializeQuery.str();
Z
update  
zhiru 已提交
314

315 316 317
            if (!InitializeQuery.exec()) {
                return HandleException("Initialization Error", InitializeQuery.error());
            }
318

S
starlord 已提交
319 320
            InitializeQuery << "CREATE TABLE IF NOT EXISTS " << TABLEFILES_SCHEMA.name() << " ("
                            << TABLEFILES_SCHEMA.ToString() + ");";
Z
update  
zhiru 已提交
321

322
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::Initialize: " << InitializeQuery.str();
Z
update  
zhiru 已提交
323

324 325 326
            if (!InitializeQuery.exec()) {
                return HandleException("Initialization Error", InitializeQuery.error());
            }
S
starlord 已提交
327 328
        }  // Scoped Connection
    } catch (std::exception& e) {
329
        return HandleException("GENERAL ERROR DURING INITIALIZATION", e.what());
Z
update  
zhiru 已提交
330
    }
S
starlord 已提交
331 332

    return Status::OK();
333 334
}

S
starlord 已提交
335
Status
S
starlord 已提交
336
MySQLMetaImpl::CreateTable(TableSchema& table_schema) {
337
    try {
Y
Yu Kun 已提交
338
        server::MetricCollector metric;
339
        {
S
starlord 已提交
340
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
341

342
            if (connectionPtr == nullptr) {
G
groot 已提交
343
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
344
            }
Z
update  
zhiru 已提交
345

S
starlord 已提交
346
            mysqlpp::Query createTableQuery = connectionPtr->query();
Z
update  
zhiru 已提交
347

348 349 350
            if (table_schema.table_id_.empty()) {
                NextTableId(table_schema.table_id_);
            } else {
G
groot 已提交
351 352
                createTableQuery << "SELECT state FROM " << META_TABLES << " WHERE table_id = " << mysqlpp::quote
                                 << table_schema.table_id_ << ";";
Z
zhiru 已提交
353

354
                ENGINE_LOG_DEBUG << "MySQLMetaImpl::CreateTable: " << createTableQuery.str();
Z
update  
zhiru 已提交
355

S
starlord 已提交
356
                mysqlpp::StoreQueryResult res = createTableQuery.store();
357

358 359 360
                if (res.num_rows() == 1) {
                    int state = res[0]["state"];
                    if (TableSchema::TO_DELETE == state) {
S
starlord 已提交
361
                        return Status(DB_ERROR, "Table already exists and it is in delete state, please wait a second");
362
                    } else {
S
starlord 已提交
363
                        return Status(DB_ALREADY_EXIST, "Table already exists");
364 365 366
                    }
                }
            }
367

368 369
            table_schema.id_ = -1;
            table_schema.created_on_ = utils::GetMicroSecTimeStamp();
Z
update  
zhiru 已提交
370

S
starlord 已提交
371
            std::string id = "NULL";  // auto-increment
G
groot 已提交
372
            std::string& table_id = table_schema.table_id_;
373 374 375
            std::string state = std::to_string(table_schema.state_);
            std::string dimension = std::to_string(table_schema.dimension_);
            std::string created_on = std::to_string(table_schema.created_on_);
S
starlord 已提交
376 377
            std::string flag = std::to_string(table_schema.flag_);
            std::string index_file_size = std::to_string(table_schema.index_file_size_);
378
            std::string engine_type = std::to_string(table_schema.engine_type_);
S
starlord 已提交
379 380
            std::string nlist = std::to_string(table_schema.nlist_);
            std::string metric_type = std::to_string(table_schema.metric_type_);
G
groot 已提交
381 382 383
            std::string& owner_table = table_schema.owner_table_;
            std::string& partition_tag = table_schema.partition_tag_;
            std::string& version = table_schema.version_;
Z
update  
zhiru 已提交
384

G
groot 已提交
385 386 387 388 389
            createTableQuery << "INSERT INTO " << META_TABLES << " VALUES(" << id << ", " << mysqlpp::quote << table_id
                             << ", " << state << ", " << dimension << ", " << created_on << ", " << flag << ", "
                             << index_file_size << ", " << engine_type << ", " << nlist << ", " << metric_type << ", "
                             << mysqlpp::quote << owner_table << ", " << mysqlpp::quote << partition_tag << ", "
                             << mysqlpp::quote << version << ");";
Z
update  
zhiru 已提交
390

391
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::CreateTable: " << createTableQuery.str();
392

S
starlord 已提交
393
            if (mysqlpp::SimpleResult res = createTableQuery.execute()) {
S
starlord 已提交
394
                table_schema.id_ = res.insert_id();  // Might need to use SELECT LAST_INSERT_ID()?
395

S
starlord 已提交
396
                // Consume all results to avoid "Commands out of sync" error
397
            } else {
S
starlord 已提交
398
                return HandleException("Add Table Error", createTableQuery.error());
399
            }
S
starlord 已提交
400
        }  // Scoped Connection
401

402
        ENGINE_LOG_DEBUG << "Successfully create table: " << table_schema.table_id_;
403
        return utils::CreateTablePath(options_, table_schema.table_id_);
S
starlord 已提交
404
    } catch (std::exception& e) {
S
starlord 已提交
405
        return HandleException("GENERAL ERROR WHEN CREATING TABLE", e.what());
406 407
    }
}
408

S
starlord 已提交
409
Status
G
groot 已提交
410
MySQLMetaImpl::DescribeTable(TableSchema& table_schema) {
Z
zhiru 已提交
411
    try {
G
groot 已提交
412
        server::MetricCollector metric;
S
starlord 已提交
413
        mysqlpp::StoreQueryResult res;
Z
zhiru 已提交
414
        {
S
starlord 已提交
415
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
Z
zhiru 已提交
416 417

            if (connectionPtr == nullptr) {
G
groot 已提交
418
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
419 420
            }

G
groot 已提交
421 422 423 424 425 426
            mysqlpp::Query describeTableQuery = connectionPtr->query();
            describeTableQuery
                << "SELECT id, state, dimension, created_on, flag, index_file_size, engine_type, nlist, metric_type"
                << " ,owner_table, partition_tag, version"
                << " FROM " << META_TABLES << " WHERE table_id = " << mysqlpp::quote << table_schema.table_id_
                << " AND state <> " << std::to_string(TableSchema::TO_DELETE) << ";";
Z
zhiru 已提交
427

G
groot 已提交
428
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::DescribeTable: " << describeTableQuery.str();
Z
zhiru 已提交
429

G
groot 已提交
430
            res = describeTableQuery.store();
S
starlord 已提交
431
        }  // Scoped Connection
Z
zhiru 已提交
432

G
groot 已提交
433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448
        if (res.num_rows() == 1) {
            const mysqlpp::Row& resRow = res[0];
            table_schema.id_ = resRow["id"];  // implicit conversion
            table_schema.state_ = resRow["state"];
            table_schema.dimension_ = resRow["dimension"];
            table_schema.created_on_ = resRow["created_on"];
            table_schema.flag_ = resRow["flag"];
            table_schema.index_file_size_ = resRow["index_file_size"];
            table_schema.engine_type_ = resRow["engine_type"];
            table_schema.nlist_ = resRow["nlist"];
            table_schema.metric_type_ = resRow["metric_type"];
            resRow["owner_table"].to_string(table_schema.owner_table_);
            resRow["partition_tag"].to_string(table_schema.partition_tag_);
            resRow["version"].to_string(table_schema.version_);
        } else {
            return Status(DB_NOT_FOUND, "Table " + table_schema.table_id_ + " not found");
449
        }
S
starlord 已提交
450
    } catch (std::exception& e) {
G
groot 已提交
451
        return HandleException("GENERAL ERROR WHEN DESCRIBING TABLE", e.what());
Z
zhiru 已提交
452 453
    }

454 455
    return Status::OK();
}
456

S
starlord 已提交
457
Status
G
groot 已提交
458
MySQLMetaImpl::HasTable(const std::string& table_id, bool& has_or_not) {
S
starlord 已提交
459
    try {
Y
Yu Kun 已提交
460
        server::MetricCollector metric;
G
groot 已提交
461
        mysqlpp::StoreQueryResult res;
S
starlord 已提交
462
        {
S
starlord 已提交
463
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
S
starlord 已提交
464 465

            if (connectionPtr == nullptr) {
G
groot 已提交
466
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
S
starlord 已提交
467 468
            }

G
groot 已提交
469 470 471 472 473 474 475
            mysqlpp::Query hasTableQuery = connectionPtr->query();
            // since table_id is a unique column we just need to check whether it exists or not
            hasTableQuery << "SELECT EXISTS"
                          << " (SELECT 1 FROM " << META_TABLES << " WHERE table_id = " << mysqlpp::quote << table_id
                          << " AND state <> " << std::to_string(TableSchema::TO_DELETE) << ")"
                          << " AS " << mysqlpp::quote << "check"
                          << ";";
S
starlord 已提交
476

G
groot 已提交
477
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::HasTable: " << hasTableQuery.str();
S
starlord 已提交
478

G
groot 已提交
479
            res = hasTableQuery.store();
S
starlord 已提交
480
        }  // Scoped Connection
S
starlord 已提交
481

G
groot 已提交
482 483
        int check = res[0]["check"];
        has_or_not = (check == 1);
S
starlord 已提交
484
    } catch (std::exception& e) {
G
groot 已提交
485
        return HandleException("GENERAL ERROR WHEN CHECKING IF TABLE EXISTS", e.what());
S
starlord 已提交
486 487
    }

488 489 490
    return Status::OK();
}

S
starlord 已提交
491
Status
G
groot 已提交
492
MySQLMetaImpl::AllTables(std::vector<TableSchema>& table_schema_array) {
S
starlord 已提交
493
    try {
Y
Yu Kun 已提交
494
        server::MetricCollector metric;
G
groot 已提交
495
        mysqlpp::StoreQueryResult res;
S
starlord 已提交
496
        {
S
starlord 已提交
497
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
S
starlord 已提交
498 499

            if (connectionPtr == nullptr) {
G
groot 已提交
500
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
S
starlord 已提交
501 502
            }

G
groot 已提交
503 504 505 506 507
            mysqlpp::Query allTablesQuery = connectionPtr->query();
            allTablesQuery << "SELECT id, table_id, dimension, engine_type, nlist, index_file_size, metric_type"
                           << " ,owner_table, partition_tag, version"
                           << " FROM " << META_TABLES << " WHERE state <> " << std::to_string(TableSchema::TO_DELETE)
                           << ";";
S
starlord 已提交
508

G
groot 已提交
509
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::AllTables: " << allTablesQuery.str();
S
starlord 已提交
510

G
groot 已提交
511
            res = allTablesQuery.store();
S
starlord 已提交
512
        }  // Scoped Connection
S
starlord 已提交
513

G
groot 已提交
514 515 516 517 518 519 520 521 522 523 524 525 526 527 528
        for (auto& resRow : res) {
            TableSchema table_schema;
            table_schema.id_ = resRow["id"];  // implicit conversion
            resRow["table_id"].to_string(table_schema.table_id_);
            table_schema.dimension_ = resRow["dimension"];
            table_schema.index_file_size_ = resRow["index_file_size"];
            table_schema.engine_type_ = resRow["engine_type"];
            table_schema.nlist_ = resRow["nlist"];
            table_schema.metric_type_ = resRow["metric_type"];
            resRow["owner_table"].to_string(table_schema.owner_table_);
            resRow["partition_tag"].to_string(table_schema.partition_tag_);
            resRow["version"].to_string(table_schema.version_);

            table_schema_array.emplace_back(table_schema);
        }
S
starlord 已提交
529
    } catch (std::exception& e) {
G
groot 已提交
530
        return HandleException("GENERAL ERROR WHEN DESCRIBING ALL TABLES", e.what());
S
starlord 已提交
531 532 533 534 535
    }

    return Status::OK();
}

S
starlord 已提交
536
Status
G
groot 已提交
537
MySQLMetaImpl::DropTable(const std::string& table_id) {
S
starlord 已提交
538
    try {
Y
Yu Kun 已提交
539
        server::MetricCollector metric;
S
starlord 已提交
540
        {
S
starlord 已提交
541
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
542

S
starlord 已提交
543
            if (connectionPtr == nullptr) {
G
groot 已提交
544
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
S
starlord 已提交
545 546
            }

G
groot 已提交
547 548 549 550 551
            // soft delete table
            mysqlpp::Query deleteTableQuery = connectionPtr->query();
            //
            deleteTableQuery << "UPDATE " << META_TABLES << " SET state = " << std::to_string(TableSchema::TO_DELETE)
                             << " WHERE table_id = " << mysqlpp::quote << table_id << ";";
S
starlord 已提交
552

G
groot 已提交
553
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::DeleteTable: " << deleteTableQuery.str();
S
starlord 已提交
554

G
groot 已提交
555 556
            if (!deleteTableQuery.exec()) {
                return HandleException("QUERY ERROR WHEN DELETING TABLE", deleteTableQuery.error());
S
starlord 已提交
557
            }
S
starlord 已提交
558
        }  // Scoped Connection
G
groot 已提交
559 560 561 562 563 564

        if (mode_ == DBOptions::MODE::CLUSTER_WRITABLE) {
            DeleteTableFiles(table_id);
        }

        ENGINE_LOG_DEBUG << "Successfully delete table, table id = " << table_id;
S
starlord 已提交
565
    } catch (std::exception& e) {
G
groot 已提交
566
        return HandleException("GENERAL ERROR WHEN DELETING TABLE", e.what());
S
starlord 已提交
567 568 569 570 571
    }

    return Status::OK();
}

S
starlord 已提交
572
Status
G
groot 已提交
573
MySQLMetaImpl::DeleteTableFiles(const std::string& table_id) {
S
starlord 已提交
574
    try {
Y
Yu Kun 已提交
575
        server::MetricCollector metric;
576
        {
S
starlord 已提交
577
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
578

579
            if (connectionPtr == nullptr) {
G
groot 已提交
580
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
581 582
            }

G
groot 已提交
583 584 585 586 587 588 589 590
            // soft delete table files
            mysqlpp::Query deleteTableFilesQuery = connectionPtr->query();
            //
            deleteTableFilesQuery << "UPDATE " << META_TABLEFILES
                                  << " SET file_type = " << std::to_string(TableFileSchema::TO_DELETE)
                                  << " ,updated_time = " << std::to_string(utils::GetMicroSecTimeStamp())
                                  << " WHERE table_id = " << mysqlpp::quote << table_id << " AND file_type <> "
                                  << std::to_string(TableFileSchema::TO_DELETE) << ";";
591

G
groot 已提交
592
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::DeleteTableFiles: " << deleteTableFilesQuery.str();
593

G
groot 已提交
594 595
            if (!deleteTableFilesQuery.exec()) {
                return HandleException("QUERY ERROR WHEN DELETING TABLE FILES", deleteTableFilesQuery.error());
S
starlord 已提交
596
            }
S
starlord 已提交
597
        }  // Scoped Connection
S
starlord 已提交
598

G
groot 已提交
599
        ENGINE_LOG_DEBUG << "Successfully delete table files, table id = " << table_id;
S
starlord 已提交
600
    } catch (std::exception& e) {
G
groot 已提交
601
        return HandleException("GENERAL ERROR WHEN DELETING TABLE FILES", e.what());
S
starlord 已提交
602
    }
603

S
starlord 已提交
604 605
    return Status::OK();
}
Z
update  
zhiru 已提交
606

S
starlord 已提交
607
Status
G
groot 已提交
608 609 610 611 612 613 614 615 616 617
MySQLMetaImpl::CreateTableFile(TableFileSchema& file_schema) {
    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;
    }
Z
zhiru 已提交
618

G
groot 已提交
619 620
    try {
        server::MetricCollector metric;
Z
update  
zhiru 已提交
621

G
groot 已提交
622 623 624 625 626 627 628 629 630 631
        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_;
632

G
groot 已提交
633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648
        std::string id = "NULL";  // auto-increment
        std::string table_id = file_schema.table_id_;
        std::string engine_type = std::to_string(file_schema.engine_type_);
        std::string file_id = file_schema.file_id_;
        std::string file_type = std::to_string(file_schema.file_type_);
        std::string file_size = std::to_string(file_schema.file_size_);
        std::string row_count = std::to_string(file_schema.row_count_);
        std::string updated_time = std::to_string(file_schema.updated_time_);
        std::string created_on = std::to_string(file_schema.created_on_);
        std::string date = std::to_string(file_schema.date_);

        {
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);

            if (connectionPtr == nullptr) {
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
649
            }
Z
zhiru 已提交
650

G
groot 已提交
651
            mysqlpp::Query createTableFileQuery = connectionPtr->query();
Z
update  
zhiru 已提交
652

G
groot 已提交
653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670
            createTableFileQuery << "INSERT INTO " << META_TABLEFILES << " VALUES(" << id << ", " << mysqlpp::quote
                                 << table_id << ", " << engine_type << ", " << mysqlpp::quote << file_id << ", "
                                 << file_type << ", " << file_size << ", " << row_count << ", " << updated_time << ", "
                                 << created_on << ", " << date << ");";

            ENGINE_LOG_DEBUG << "MySQLMetaImpl::CreateTableFile: " << createTableFileQuery.str();

            if (mysqlpp::SimpleResult res = createTableFileQuery.execute()) {
                file_schema.id_ = res.insert_id();  // Might need to use SELECT LAST_INSERT_ID()?

                // Consume all results to avoid "Commands out of sync" error
            } else {
                return HandleException("QUERY ERROR WHEN CREATING TABLE FILE", createTableFileQuery.error());
            }
        }  // Scoped Connection

        ENGINE_LOG_DEBUG << "Successfully create table file, file id = " << file_schema.file_id_;
        return utils::CreateTableFilePath(options_, file_schema);
S
starlord 已提交
671
    } catch (std::exception& e) {
G
groot 已提交
672
        return HandleException("GENERAL ERROR WHEN CREATING TABLE FILE", e.what());
673 674
    }
}
Z
update  
zhiru 已提交
675

G
groot 已提交
676
// TODO(myh): Delete single vecotor by id
S
starlord 已提交
677
Status
G
groot 已提交
678 679 680 681 682 683 684 685 686 687 688 689
MySQLMetaImpl::DropDataByDate(const std::string& table_id, const DatesT& dates) {
    if (dates.empty()) {
        return Status::OK();
    }

    TableSchema table_schema;
    table_schema.table_id_ = table_id;
    auto status = DescribeTable(table_schema);
    if (!status.ok()) {
        return status;
    }

690
    try {
G
groot 已提交
691 692 693 694 695 696 697
        std::stringstream dateListSS;
        for (auto& date : dates) {
            dateListSS << std::to_string(date) << ", ";
        }
        std::string dateListStr = dateListSS.str();
        dateListStr = dateListStr.substr(0, dateListStr.size() - 2);  // remove the last ", "

698
        {
S
starlord 已提交
699
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
700 701

            if (connectionPtr == nullptr) {
G
groot 已提交
702
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
703
            }
704

G
groot 已提交
705
            mysqlpp::Query dropPartitionsByDatesQuery = connectionPtr->query();
706

G
groot 已提交
707 708 709 710 711
            dropPartitionsByDatesQuery << "UPDATE " << META_TABLEFILES
                                       << " SET file_type = " << std::to_string(TableFileSchema::TO_DELETE)
                                       << " ,updated_time = " << utils::GetMicroSecTimeStamp()
                                       << " WHERE table_id = " << mysqlpp::quote << table_id << " AND date in ("
                                       << dateListStr << ");";
712

G
groot 已提交
713 714 715 716 717
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::DropDataByDate: " << dropPartitionsByDatesQuery.str();

            if (!dropPartitionsByDatesQuery.exec()) {
                return HandleException("QUERY ERROR WHEN DROPPING PARTITIONS BY DATES",
                                       dropPartitionsByDatesQuery.error());
718
            }
S
starlord 已提交
719
        }  // Scoped Connection
720

G
groot 已提交
721
        ENGINE_LOG_DEBUG << "Successfully drop data by date, table id = " << table_schema.table_id_;
S
starlord 已提交
722
    } catch (std::exception& e) {
G
groot 已提交
723
        return HandleException("GENERAL ERROR WHEN DROPPING PARTITIONS BY DATES", e.what());
Z
update  
zhiru 已提交
724
    }
725 726
    return Status::OK();
}
Z
zhiru 已提交
727

S
starlord 已提交
728
Status
G
groot 已提交
729 730 731 732 733 734 735 736 737 738 739 740 741
MySQLMetaImpl::GetTableFiles(const std::string& table_id, const std::vector<size_t>& ids,
                             TableFilesSchema& table_files) {
    if (ids.empty()) {
        return Status::OK();
    }

    std::stringstream idSS;
    for (auto& id : ids) {
        idSS << "id = " << std::to_string(id) << " OR ";
    }
    std::string idStr = idSS.str();
    idStr = idStr.substr(0, idStr.size() - 4);  // remove the last " OR "

742
    try {
S
starlord 已提交
743
        mysqlpp::StoreQueryResult res;
744
        {
S
starlord 已提交
745
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
746

747
            if (connectionPtr == nullptr) {
G
groot 已提交
748
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
749
            }
Z
zhiru 已提交
750

G
groot 已提交
751 752 753 754 755
            mysqlpp::Query getTableFileQuery = connectionPtr->query();
            getTableFileQuery << "SELECT id, engine_type, file_id, file_type, file_size, row_count, date, created_on"
                              << " FROM " << META_TABLEFILES << " WHERE table_id = " << mysqlpp::quote << table_id
                              << " AND (" << idStr << ")"
                              << " AND file_type <> " << std::to_string(TableFileSchema::TO_DELETE) << ";";
756

G
groot 已提交
757
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::GetTableFiles: " << getTableFileQuery.str();
Z
update  
zhiru 已提交
758

G
groot 已提交
759
            res = getTableFileQuery.store();
S
starlord 已提交
760
        }  // Scoped Connection
761

G
groot 已提交
762 763 764
        TableSchema table_schema;
        table_schema.table_id_ = table_id;
        DescribeTable(table_schema);
S
starlord 已提交
765

G
groot 已提交
766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781
        Status ret;
        for (auto& resRow : res) {
            TableFileSchema file_schema;
            file_schema.id_ = resRow["id"];
            file_schema.table_id_ = table_id;
            file_schema.index_file_size_ = table_schema.index_file_size_;
            file_schema.engine_type_ = resRow["engine_type"];
            file_schema.nlist_ = table_schema.nlist_;
            file_schema.metric_type_ = table_schema.metric_type_;
            resRow["file_id"].to_string(file_schema.file_id_);
            file_schema.file_type_ = resRow["file_type"];
            file_schema.file_size_ = resRow["file_size"];
            file_schema.row_count_ = resRow["row_count"];
            file_schema.date_ = resRow["date"];
            file_schema.created_on_ = resRow["created_on"];
            file_schema.dimension_ = table_schema.dimension_;
S
starlord 已提交
782

G
groot 已提交
783 784
            utils::GetTableFilePath(options_, file_schema);
            table_files.emplace_back(file_schema);
785
        }
G
groot 已提交
786 787 788

        ENGINE_LOG_DEBUG << "Get table files by id";
        return ret;
S
starlord 已提交
789
    } catch (std::exception& e) {
G
groot 已提交
790
        return HandleException("GENERAL ERROR WHEN RETRIEVING TABLE FILES", e.what());
Z
update  
zhiru 已提交
791
    }
792
}
Z
zhiru 已提交
793

S
starlord 已提交
794
Status
G
groot 已提交
795
MySQLMetaImpl::UpdateTableIndex(const std::string& table_id, const TableIndex& index) {
796
    try {
Y
Yu Kun 已提交
797
        server::MetricCollector metric;
G
groot 已提交
798

799
        {
S
starlord 已提交
800
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
Z
update  
zhiru 已提交
801

802
            if (connectionPtr == nullptr) {
G
groot 已提交
803
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
804
            }
Z
update  
zhiru 已提交
805

G
groot 已提交
806 807 808 809
            mysqlpp::Query updateTableIndexParamQuery = connectionPtr->query();
            updateTableIndexParamQuery << "SELECT id, state, dimension, created_on"
                                       << " FROM " << META_TABLES << " WHERE table_id = " << mysqlpp::quote << table_id
                                       << " AND state <> " << std::to_string(TableSchema::TO_DELETE) << ";";
Z
update  
zhiru 已提交
810

G
groot 已提交
811
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::UpdateTableIndex: " << updateTableIndexParamQuery.str();
812

G
groot 已提交
813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837
            mysqlpp::StoreQueryResult res = updateTableIndexParamQuery.store();

            if (res.num_rows() == 1) {
                const mysqlpp::Row& resRow = res[0];

                size_t id = resRow["id"];
                int32_t state = resRow["state"];
                uint16_t dimension = resRow["dimension"];
                int64_t created_on = resRow["created_on"];

                updateTableIndexParamQuery << "UPDATE " << META_TABLES << " SET id = " << id << " ,state = " << state
                                           << " ,dimension = " << dimension << " ,created_on = " << created_on
                                           << " ,engine_type = " << index.engine_type_ << " ,nlist = " << index.nlist_
                                           << " ,metric_type = " << index.metric_type_
                                           << " WHERE table_id = " << mysqlpp::quote << table_id << ";";

                ENGINE_LOG_DEBUG << "MySQLMetaImpl::UpdateTableIndex: " << updateTableIndexParamQuery.str();

                if (!updateTableIndexParamQuery.exec()) {
                    return HandleException("QUERY ERROR WHEN UPDATING TABLE INDEX PARAM",
                                           updateTableIndexParamQuery.error());
                }
            } else {
                return Status(DB_NOT_FOUND, "Table " + table_id + " not found");
            }
S
starlord 已提交
838
        }  // Scoped Connection
839

G
groot 已提交
840
        ENGINE_LOG_DEBUG << "Successfully update table index, table id = " << table_id;
S
starlord 已提交
841
    } catch (std::exception& e) {
G
groot 已提交
842
        return HandleException("GENERAL ERROR WHEN UPDATING TABLE INDEX PARAM", e.what());
843
    }
844

845 846
    return Status::OK();
}
847

S
starlord 已提交
848
Status
G
groot 已提交
849
MySQLMetaImpl::UpdateTableFlag(const std::string& table_id, int64_t flag) {
850
    try {
Y
Yu Kun 已提交
851
        server::MetricCollector metric;
G
groot 已提交
852

853
        {
S
starlord 已提交
854
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
855

856
            if (connectionPtr == nullptr) {
G
groot 已提交
857
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
858
            }
Z
update  
zhiru 已提交
859

G
groot 已提交
860 861 862
            mysqlpp::Query updateTableFlagQuery = connectionPtr->query();
            updateTableFlagQuery << "UPDATE " << META_TABLES << " SET flag = " << flag
                                 << " WHERE table_id = " << mysqlpp::quote << table_id << ";";
Z
zhiru 已提交
863

G
groot 已提交
864
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::UpdateTableFlag: " << updateTableFlagQuery.str();
Z
zhiru 已提交
865

G
groot 已提交
866 867 868
            if (!updateTableFlagQuery.exec()) {
                return HandleException("QUERY ERROR WHEN UPDATING TABLE FLAG", updateTableFlagQuery.error());
            }
S
starlord 已提交
869
        }  // Scoped Connection
870

G
groot 已提交
871
        ENGINE_LOG_DEBUG << "Successfully update table flag, table id = " << table_id;
S
starlord 已提交
872
    } catch (std::exception& e) {
G
groot 已提交
873
        return HandleException("GENERAL ERROR WHEN UPDATING TABLE FLAG", e.what());
874
    }
Z
update  
zhiru 已提交
875

876 877
    return Status::OK();
}
878

G
groot 已提交
879
// ZR: this function assumes all fields in file_schema have value
S
starlord 已提交
880
Status
G
groot 已提交
881 882
MySQLMetaImpl::UpdateTableFile(TableFileSchema& file_schema) {
    file_schema.updated_time_ = utils::GetMicroSecTimeStamp();
883

884
    try {
Y
Yu Kun 已提交
885
        server::MetricCollector metric;
G
groot 已提交
886 887
        {
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
Z
update  
zhiru 已提交
888

889
            if (connectionPtr == nullptr) {
G
groot 已提交
890
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
891
            }
Z
update  
zhiru 已提交
892

G
groot 已提交
893
            mysqlpp::Query updateTableFileQuery = connectionPtr->query();
894

G
groot 已提交
895 896 897 898
            // if the table has been deleted, just mark the table file as TO_DELETE
            // clean thread will delete the file later
            updateTableFileQuery << "SELECT state FROM " << META_TABLES << " WHERE table_id = " << mysqlpp::quote
                                 << file_schema.table_id_ << ";";
899

G
groot 已提交
900
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::UpdateTableFile: " << updateTableFileQuery.str();
901

G
groot 已提交
902
            mysqlpp::StoreQueryResult res = updateTableFileQuery.store();
903

G
groot 已提交
904 905 906 907 908
            if (res.num_rows() == 1) {
                int state = res[0]["state"];
                if (state == TableSchema::TO_DELETE) {
                    file_schema.file_type_ = TableFileSchema::TO_DELETE;
                }
909
            } else {
G
groot 已提交
910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935
                file_schema.file_type_ = TableFileSchema::TO_DELETE;
            }

            std::string id = std::to_string(file_schema.id_);
            std::string table_id = file_schema.table_id_;
            std::string engine_type = std::to_string(file_schema.engine_type_);
            std::string file_id = file_schema.file_id_;
            std::string file_type = std::to_string(file_schema.file_type_);
            std::string file_size = std::to_string(file_schema.file_size_);
            std::string row_count = std::to_string(file_schema.row_count_);
            std::string updated_time = std::to_string(file_schema.updated_time_);
            std::string created_on = std::to_string(file_schema.created_on_);
            std::string date = std::to_string(file_schema.date_);

            updateTableFileQuery << "UPDATE " << META_TABLEFILES << " SET table_id = " << mysqlpp::quote << table_id
                                 << " ,engine_type = " << engine_type << " ,file_id = " << mysqlpp::quote << file_id
                                 << " ,file_type = " << file_type << " ,file_size = " << file_size
                                 << " ,row_count = " << row_count << " ,updated_time = " << updated_time
                                 << " ,created_on = " << created_on << " ,date = " << date << " WHERE id = " << id
                                 << ";";

            ENGINE_LOG_DEBUG << "MySQLMetaImpl::UpdateTableFile: " << updateTableFileQuery.str();

            if (!updateTableFileQuery.exec()) {
                ENGINE_LOG_DEBUG << "table_id= " << file_schema.table_id_ << " file_id=" << file_schema.file_id_;
                return HandleException("QUERY ERROR WHEN UPDATING TABLE FILE", updateTableFileQuery.error());
936
            }
S
starlord 已提交
937
        }  // Scoped Connection
938

G
groot 已提交
939
        ENGINE_LOG_DEBUG << "Update single table file, file id = " << file_schema.file_id_;
S
starlord 已提交
940
    } catch (std::exception& e) {
G
groot 已提交
941
        return HandleException("GENERAL ERROR WHEN UPDATING TABLE FILE", e.what());
942
    }
G
groot 已提交
943 944

    return Status::OK();
945
}
946

S
starlord 已提交
947
Status
G
groot 已提交
948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976
MySQLMetaImpl::UpdateTableFilesToIndex(const std::string& table_id) {
    try {
        mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);

        if (connectionPtr == nullptr) {
            return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
        }

        mysqlpp::Query updateTableFilesToIndexQuery = connectionPtr->query();

        updateTableFilesToIndexQuery << "UPDATE " << META_TABLEFILES
                                     << " SET file_type = " << std::to_string(TableFileSchema::TO_INDEX)
                                     << " WHERE table_id = " << mysqlpp::quote << table_id
                                     << " AND file_type = " << std::to_string(TableFileSchema::RAW) << ";";

        ENGINE_LOG_DEBUG << "MySQLMetaImpl::UpdateTableFilesToIndex: " << updateTableFilesToIndexQuery.str();

        if (!updateTableFilesToIndexQuery.exec()) {
            return HandleException("QUERY ERROR WHEN UPDATING TABLE FILE TO INDEX",
                                   updateTableFilesToIndexQuery.error());
        }

        ENGINE_LOG_DEBUG << "Update files to to_index, table id = " << table_id;
    } catch (std::exception& e) {
        return HandleException("GENERAL ERROR WHEN UPDATING TABLE FILES TO INDEX", e.what());
    }

    return Status::OK();
}
977

G
groot 已提交
978 979
Status
MySQLMetaImpl::UpdateTableFiles(TableFilesSchema& files) {
980
    try {
Y
Yu Kun 已提交
981
        server::MetricCollector metric;
982
        {
S
starlord 已提交
983
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
984 985

            if (connectionPtr == nullptr) {
G
groot 已提交
986
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
987
            }
988

G
groot 已提交
989
            mysqlpp::Query updateTableFilesQuery = connectionPtr->query();
990

G
groot 已提交
991 992 993 994 995
            std::map<std::string, bool> has_tables;
            for (auto& file_schema : files) {
                if (has_tables.find(file_schema.table_id_) != has_tables.end()) {
                    continue;
                }
996

G
groot 已提交
997 998 999 1000 1001 1002
                updateTableFilesQuery << "SELECT EXISTS"
                                      << " (SELECT 1 FROM " << META_TABLES << " WHERE table_id = " << mysqlpp::quote
                                      << file_schema.table_id_ << " AND state <> "
                                      << std::to_string(TableSchema::TO_DELETE) << ")"
                                      << " AS " << mysqlpp::quote << "check"
                                      << ";";
1003

G
groot 已提交
1004
                ENGINE_LOG_DEBUG << "MySQLMetaImpl::UpdateTableFiles: " << updateTableFilesQuery.str();
1005

G
groot 已提交
1006
                mysqlpp::StoreQueryResult res = updateTableFilesQuery.store();
1007

G
groot 已提交
1008 1009 1010
                int check = res[0]["check"];
                has_tables[file_schema.table_id_] = (check == 1);
            }
1011

G
groot 已提交
1012 1013 1014 1015 1016
            for (auto& file_schema : files) {
                if (!has_tables[file_schema.table_id_]) {
                    file_schema.file_type_ = TableFileSchema::TO_DELETE;
                }
                file_schema.updated_time_ = utils::GetMicroSecTimeStamp();
S
starlord 已提交
1017

G
groot 已提交
1018 1019 1020 1021 1022 1023 1024 1025 1026 1027
                std::string id = std::to_string(file_schema.id_);
                std::string& table_id = file_schema.table_id_;
                std::string engine_type = std::to_string(file_schema.engine_type_);
                std::string& file_id = file_schema.file_id_;
                std::string file_type = std::to_string(file_schema.file_type_);
                std::string file_size = std::to_string(file_schema.file_size_);
                std::string row_count = std::to_string(file_schema.row_count_);
                std::string updated_time = std::to_string(file_schema.updated_time_);
                std::string created_on = std::to_string(file_schema.created_on_);
                std::string date = std::to_string(file_schema.date_);
1028

G
groot 已提交
1029 1030 1031 1032 1033 1034
                updateTableFilesQuery << "UPDATE " << META_TABLEFILES << " SET table_id = " << mysqlpp::quote
                                      << table_id << " ,engine_type = " << engine_type
                                      << " ,file_id = " << mysqlpp::quote << file_id << " ,file_type = " << file_type
                                      << " ,file_size = " << file_size << " ,row_count = " << row_count
                                      << " ,updated_time = " << updated_time << " ,created_on = " << created_on
                                      << " ,date = " << date << " WHERE id = " << id << ";";
1035

G
groot 已提交
1036
                ENGINE_LOG_DEBUG << "MySQLMetaImpl::UpdateTableFiles: " << updateTableFilesQuery.str();
S
starlord 已提交
1037

G
groot 已提交
1038 1039
                if (!updateTableFilesQuery.exec()) {
                    return HandleException("QUERY ERROR WHEN UPDATING TABLE FILES", updateTableFilesQuery.error());
1040
                }
S
starlord 已提交
1041
            }
G
groot 已提交
1042
        }  // Scoped Connection
1043

G
groot 已提交
1044
        ENGINE_LOG_DEBUG << "Update " << files.size() << " table files";
S
starlord 已提交
1045
    } catch (std::exception& e) {
G
groot 已提交
1046
        return HandleException("GENERAL ERROR WHEN UPDATING TABLE FILES", e.what());
Z
update  
zhiru 已提交
1047
    }
G
groot 已提交
1048 1049

    return Status::OK();
1050 1051
}

S
starlord 已提交
1052
Status
G
groot 已提交
1053
MySQLMetaImpl::DescribeTableIndex(const std::string& table_id, TableIndex& index) {
X
xj.lin 已提交
1054
    try {
Y
Yu Kun 已提交
1055
        server::MetricCollector metric;
G
groot 已提交
1056

X
xj.lin 已提交
1057
        {
S
starlord 已提交
1058
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
X
xj.lin 已提交
1059 1060

            if (connectionPtr == nullptr) {
G
groot 已提交
1061
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
X
xj.lin 已提交
1062 1063
            }

G
groot 已提交
1064 1065 1066 1067
            mysqlpp::Query describeTableIndexQuery = connectionPtr->query();
            describeTableIndexQuery << "SELECT engine_type, nlist, index_file_size, metric_type"
                                    << " FROM " << META_TABLES << " WHERE table_id = " << mysqlpp::quote << table_id
                                    << " AND state <> " << std::to_string(TableSchema::TO_DELETE) << ";";
X
xj.lin 已提交
1068

G
groot 已提交
1069
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::DescribeTableIndex: " << describeTableIndexQuery.str();
X
xj.lin 已提交
1070

G
groot 已提交
1071
            mysqlpp::StoreQueryResult res = describeTableIndexQuery.store();
X
xj.lin 已提交
1072

G
groot 已提交
1073 1074
            if (res.num_rows() == 1) {
                const mysqlpp::Row& resRow = res[0];
X
xj.lin 已提交
1075

G
groot 已提交
1076 1077 1078 1079 1080
                index.engine_type_ = resRow["engine_type"];
                index.nlist_ = resRow["nlist"];
                index.metric_type_ = resRow["metric_type"];
            } else {
                return Status(DB_NOT_FOUND, "Table " + table_id + " not found");
X
xj.lin 已提交
1081
            }
S
starlord 已提交
1082
        }  // Scoped Connection
G
groot 已提交
1083 1084 1085
    } catch (std::exception& e) {
        return HandleException("GENERAL ERROR WHEN UPDATING TABLE FLAG", e.what());
    }
X
xj.lin 已提交
1086

G
groot 已提交
1087 1088
    return Status::OK();
}
X
xj.lin 已提交
1089

G
groot 已提交
1090 1091 1092 1093
Status
MySQLMetaImpl::DropTableIndex(const std::string& table_id) {
    try {
        server::MetricCollector metric;
X
xj.lin 已提交
1094

G
groot 已提交
1095 1096
        {
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
X
xj.lin 已提交
1097

G
groot 已提交
1098 1099 1100
            if (connectionPtr == nullptr) {
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
            }
1101

G
groot 已提交
1102
            mysqlpp::Query dropTableIndexQuery = connectionPtr->query();
X
xj.lin 已提交
1103

G
groot 已提交
1104 1105 1106 1107 1108 1109
            // soft delete index files
            dropTableIndexQuery << "UPDATE " << META_TABLEFILES
                                << " SET file_type = " << std::to_string(TableFileSchema::TO_DELETE)
                                << " ,updated_time = " << utils::GetMicroSecTimeStamp()
                                << " WHERE table_id = " << mysqlpp::quote << table_id
                                << " AND file_type = " << std::to_string(TableFileSchema::INDEX) << ";";
S
starlord 已提交
1110

G
groot 已提交
1111
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::DropTableIndex: " << dropTableIndexQuery.str();
S
starlord 已提交
1112

G
groot 已提交
1113 1114 1115
            if (!dropTableIndexQuery.exec()) {
                return HandleException("QUERY ERROR WHEN DROPPING TABLE INDEX", dropTableIndexQuery.error());
            }
X
xj.lin 已提交
1116

G
groot 已提交
1117 1118 1119 1120 1121 1122
            // set all backup file to raw
            dropTableIndexQuery << "UPDATE " << META_TABLEFILES
                                << " SET file_type = " << std::to_string(TableFileSchema::RAW)
                                << " ,updated_time = " << utils::GetMicroSecTimeStamp()
                                << " WHERE table_id = " << mysqlpp::quote << table_id
                                << " AND file_type = " << std::to_string(TableFileSchema::BACKUP) << ";";
X
xj.lin 已提交
1123

G
groot 已提交
1124
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::DropTableIndex: " << dropTableIndexQuery.str();
S
starlord 已提交
1125

G
groot 已提交
1126 1127 1128
            if (!dropTableIndexQuery.exec()) {
                return HandleException("QUERY ERROR WHEN DROPPING TABLE INDEX", dropTableIndexQuery.error());
            }
X
xj.lin 已提交
1129

G
groot 已提交
1130 1131 1132 1133 1134 1135
            // set table index type to raw
            dropTableIndexQuery << "UPDATE " << META_TABLES
                                << " SET engine_type = " << std::to_string(DEFAULT_ENGINE_TYPE)
                                << " ,nlist = " << std::to_string(DEFAULT_NLIST)
                                << " ,metric_type = " << std::to_string(DEFAULT_METRIC_TYPE)
                                << " WHERE table_id = " << mysqlpp::quote << table_id << ";";
X
xj.lin 已提交
1136

G
groot 已提交
1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::DropTableIndex: " << dropTableIndexQuery.str();

            if (!dropTableIndexQuery.exec()) {
                return HandleException("QUERY ERROR WHEN DROPPING TABLE INDEX", dropTableIndexQuery.error());
            }
        }  // Scoped Connection

        ENGINE_LOG_DEBUG << "Successfully drop table index, table id = " << table_id;
    } catch (std::exception& e) {
        return HandleException("GENERAL ERROR WHEN DROPPING TABLE INDEX", e.what());
    }

    return Status::OK();
}

Status
MySQLMetaImpl::CreatePartition(const std::string& table_id, const std::string& partition_name, const std::string& tag) {
    server::MetricCollector metric;

    TableSchema table_schema;
    table_schema.table_id_ = table_id;
    auto status = DescribeTable(table_schema);
    if (!status.ok()) {
        return status;
    }

    // not allow create partition under partition
    if (!table_schema.owner_table_.empty()) {
        return Status(DB_ERROR, "Nested partition is not allow");
    }

    if (partition_name == "") {
        // not allow duplicated partition
        std::string exist_partition;
        GetPartitionName(table_id, tag, exist_partition);
        if (!exist_partition.empty()) {
            return Status(DB_ERROR, "Duplicated partition is not allow");
        }

        NextTableId(table_schema.table_id_);
    } else {
        table_schema.table_id_ = partition_name;
    }

    table_schema.id_ = -1;
    table_schema.flag_ = 0;
    table_schema.created_on_ = utils::GetMicroSecTimeStamp();
    table_schema.owner_table_ = table_id;
    table_schema.partition_tag_ = tag;

    return CreateTable(table_schema);
}

Status
MySQLMetaImpl::DropPartition(const std::string& partition_name) {
    return DropTable(partition_name);
}

Status
MySQLMetaImpl::ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partiton_schema_array) {
    try {
        server::MetricCollector metric;
        mysqlpp::StoreQueryResult res;
        {
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);

            if (connectionPtr == nullptr) {
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
            }

            mysqlpp::Query allPartitionsQuery = connectionPtr->query();
            allPartitionsQuery << "SELECT table_id FROM " << META_TABLES << " WHERE owner_table = " << mysqlpp::quote
                               << table_id << " AND state <> " << std::to_string(TableSchema::TO_DELETE) << ";";

            ENGINE_LOG_DEBUG << "MySQLMetaImpl::AllTables: " << allPartitionsQuery.str();

            res = allPartitionsQuery.store();
        }  // Scoped Connection

        for (auto& resRow : res) {
            meta::TableSchema partition_schema;
            resRow["table_id"].to_string(partition_schema.table_id_);
            DescribeTable(partition_schema);
            partiton_schema_array.emplace_back(partition_schema);
        }
    } catch (std::exception& e) {
        return HandleException("GENERAL ERROR WHEN SHOW PARTITIONS", e.what());
    }

    return Status::OK();
}

Status
MySQLMetaImpl::GetPartitionName(const std::string& table_id, const std::string& tag, std::string& partition_name) {
    try {
        server::MetricCollector metric;
        mysqlpp::StoreQueryResult res;
        {
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);

            if (connectionPtr == nullptr) {
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
            }

            mysqlpp::Query allPartitionsQuery = connectionPtr->query();
            allPartitionsQuery << "SELECT table_id FROM " << META_TABLES << " WHERE owner_table = " << mysqlpp::quote
                               << table_id << " AND partition_tag = " << mysqlpp::quote << tag << " AND state <> "
                               << std::to_string(TableSchema::TO_DELETE) << ";";

            ENGINE_LOG_DEBUG << "MySQLMetaImpl::AllTables: " << allPartitionsQuery.str();

            res = allPartitionsQuery.store();
        }  // Scoped Connection

        if (res.num_rows() > 0) {
            const mysqlpp::Row& resRow = res[0];
            resRow["table_id"].to_string(partition_name);
        } else {
            return Status(DB_NOT_FOUND, "Partition " + tag + " of table " + table_id + " not found");
        }
    } catch (std::exception& e) {
        return HandleException("GENERAL ERROR WHEN GET PARTITION NAME", e.what());
    }

    return Status::OK();
}

Status
MySQLMetaImpl::FilesToSearch(const std::string& table_id, const std::vector<size_t>& ids, const DatesT& dates,
                             DatePartionedTableFilesSchema& files) {
    files.clear();

    try {
        server::MetricCollector metric;
        mysqlpp::StoreQueryResult res;
        {
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);

            if (connectionPtr == nullptr) {
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
            }

            mysqlpp::Query filesToSearchQuery = connectionPtr->query();
            filesToSearchQuery << "SELECT id, table_id, engine_type, file_id, file_type, file_size, row_count, date"
                               << " FROM " << META_TABLEFILES << " WHERE table_id = " << mysqlpp::quote << table_id;

            if (!dates.empty()) {
                std::stringstream partitionListSS;
                for (auto& date : dates) {
                    partitionListSS << std::to_string(date) << ", ";
                }
                std::string partitionListStr = partitionListSS.str();

                partitionListStr = partitionListStr.substr(0, partitionListStr.size() - 2);  // remove the last ", "
                filesToSearchQuery << " AND date IN (" << partitionListStr << ")";
            }

            if (!ids.empty()) {
                std::stringstream idSS;
                for (auto& id : ids) {
                    idSS << "id = " << std::to_string(id) << " OR ";
                }
                std::string idStr = idSS.str();
                idStr = idStr.substr(0, idStr.size() - 4);  // remove the last " OR "

                filesToSearchQuery << " AND (" << idStr << ")";
            }
            // End
            filesToSearchQuery << " AND"
                               << " (file_type = " << std::to_string(TableFileSchema::RAW)
                               << " OR file_type = " << std::to_string(TableFileSchema::TO_INDEX)
                               << " OR file_type = " << std::to_string(TableFileSchema::INDEX) << ");";

            ENGINE_LOG_DEBUG << "MySQLMetaImpl::FilesToSearch: " << filesToSearchQuery.str();

            res = filesToSearchQuery.store();
        }  // Scoped Connection

        TableSchema table_schema;
        table_schema.table_id_ = table_id;
        auto status = DescribeTable(table_schema);
        if (!status.ok()) {
            return status;
        }

        Status ret;
        TableFileSchema table_file;
        for (auto& resRow : res) {
            table_file.id_ = resRow["id"];  // implicit conversion
            resRow["table_id"].to_string(table_file.table_id_);
            table_file.index_file_size_ = table_schema.index_file_size_;
            table_file.engine_type_ = resRow["engine_type"];
            table_file.nlist_ = table_schema.nlist_;
            table_file.metric_type_ = table_schema.metric_type_;
            resRow["file_id"].to_string(table_file.file_id_);
            table_file.file_type_ = resRow["file_type"];
            table_file.file_size_ = resRow["file_size"];
            table_file.row_count_ = resRow["row_count"];
            table_file.date_ = resRow["date"];
            table_file.dimension_ = table_schema.dimension_;
X
xj.lin 已提交
1337

S
starlord 已提交
1338
            auto status = utils::GetTableFilePath(options_, table_file);
S
starlord 已提交
1339
            if (!status.ok()) {
S
starlord 已提交
1340
                ret = status;
S
starlord 已提交
1341
            }
1342

1343 1344 1345
            auto dateItr = files.find(table_file.date_);
            if (dateItr == files.end()) {
                files[table_file.date_] = TableFilesSchema();
1346 1347
            }

1348
            files[table_file.date_].push_back(table_file);
1349
        }
S
starlord 已提交
1350

S
starlord 已提交
1351
        if (res.size() > 0) {
1352 1353
            ENGINE_LOG_DEBUG << "Collect " << res.size() << " to-search files";
        }
S
starlord 已提交
1354
        return ret;
S
starlord 已提交
1355
    } catch (std::exception& e) {
S
starlord 已提交
1356
        return HandleException("GENERAL ERROR WHEN FINDING TABLE FILES TO SEARCH", e.what());
Z
update  
zhiru 已提交
1357
    }
1358
}
Z
update  
zhiru 已提交
1359

S
starlord 已提交
1360
Status
S
starlord 已提交
1361
MySQLMetaImpl::FilesToMerge(const std::string& table_id, DatePartionedTableFilesSchema& files) {
1362
    files.clear();
Z
update  
zhiru 已提交
1363

1364
    try {
Y
Yu Kun 已提交
1365
        server::MetricCollector metric;
S
starlord 已提交
1366

S
starlord 已提交
1367
        // check table existence
S
starlord 已提交
1368 1369 1370 1371 1372 1373 1374
        TableSchema table_schema;
        table_schema.table_id_ = table_id;
        auto status = DescribeTable(table_schema);
        if (!status.ok()) {
            return status;
        }

S
starlord 已提交
1375
        mysqlpp::StoreQueryResult res;
1376
        {
S
starlord 已提交
1377
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
Z
update  
zhiru 已提交
1378

1379
            if (connectionPtr == nullptr) {
G
groot 已提交
1380
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
1381
            }
Z
update  
zhiru 已提交
1382

S
starlord 已提交
1383 1384
            mysqlpp::Query filesToMergeQuery = connectionPtr->query();
            filesToMergeQuery
G
groot 已提交
1385 1386 1387
                << "SELECT id, table_id, file_id, file_type, file_size, row_count, date, engine_type, created_on"
                << " FROM " << META_TABLEFILES << " WHERE table_id = " << mysqlpp::quote << table_id
                << " AND file_type = " << std::to_string(TableFileSchema::RAW) << " ORDER BY row_count DESC;";
Z
update  
zhiru 已提交
1388

1389
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::FilesToMerge: " << filesToMergeQuery.str();
1390

1391
            res = filesToMergeQuery.store();
S
starlord 已提交
1392
        }  // Scoped Connection
1393

S
starlord 已提交
1394
        Status ret;
S
starlord 已提交
1395
        for (auto& resRow : res) {
S
starlord 已提交
1396 1397
            TableFileSchema table_file;
            table_file.file_size_ = resRow["file_size"];
S
starlord 已提交
1398
            if (table_file.file_size_ >= table_schema.index_file_size_) {
S
starlord 已提交
1399
                continue;  // skip large file
S
starlord 已提交
1400
            }
Z
update  
zhiru 已提交
1401

S
starlord 已提交
1402
            table_file.id_ = resRow["id"];  // implicit conversion
G
groot 已提交
1403 1404
            resRow["table_id"].to_string(table_file.table_id_);
            resRow["file_id"].to_string(table_file.file_id_);
1405
            table_file.file_type_ = resRow["file_type"];
S
starlord 已提交
1406
            table_file.row_count_ = resRow["row_count"];
1407
            table_file.date_ = resRow["date"];
1408
            table_file.index_file_size_ = table_schema.index_file_size_;
S
starlord 已提交
1409
            table_file.engine_type_ = resRow["engine_type"];
S
starlord 已提交
1410
            table_file.nlist_ = table_schema.nlist_;
S
starlord 已提交
1411
            table_file.metric_type_ = table_schema.metric_type_;
S
starlord 已提交
1412
            table_file.created_on_ = resRow["created_on"];
1413
            table_file.dimension_ = table_schema.dimension_;
Z
update  
zhiru 已提交
1414

S
starlord 已提交
1415
            auto status = utils::GetTableFilePath(options_, table_file);
S
starlord 已提交
1416
            if (!status.ok()) {
S
starlord 已提交
1417
                ret = status;
S
starlord 已提交
1418
            }
Z
update  
zhiru 已提交
1419

1420 1421 1422
            auto dateItr = files.find(table_file.date_);
            if (dateItr == files.end()) {
                files[table_file.date_] = TableFilesSchema();
1423
            }
1424 1425

            files[table_file.date_].push_back(table_file);
1426
        }
Z
update  
zhiru 已提交
1427

S
starlord 已提交
1428
        if (res.size() > 0) {
1429 1430
            ENGINE_LOG_DEBUG << "Collect " << res.size() << " to-merge files";
        }
S
starlord 已提交
1431
        return ret;
S
starlord 已提交
1432
    } catch (std::exception& e) {
S
starlord 已提交
1433
        return HandleException("GENERAL ERROR WHEN FINDING TABLE FILES TO MERGE", e.what());
1434 1435 1436
    }
}

S
starlord 已提交
1437
Status
G
groot 已提交
1438 1439
MySQLMetaImpl::FilesToIndex(TableFilesSchema& files) {
    files.clear();
Z
zhiru 已提交
1440

1441
    try {
G
groot 已提交
1442
        server::MetricCollector metric;
S
starlord 已提交
1443
        mysqlpp::StoreQueryResult res;
1444
        {
S
starlord 已提交
1445
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
1446 1447

            if (connectionPtr == nullptr) {
G
groot 已提交
1448
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
1449 1450
            }

G
groot 已提交
1451 1452 1453 1454 1455
            mysqlpp::Query filesToIndexQuery = connectionPtr->query();
            filesToIndexQuery
                << "SELECT id, table_id, engine_type, file_id, file_type, file_size, row_count, date, created_on"
                << " FROM " << META_TABLEFILES << " WHERE file_type = " << std::to_string(TableFileSchema::TO_INDEX)
                << ";";
1456

G
groot 已提交
1457
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::FilesToIndex: " << filesToIndexQuery.str();
1458

G
groot 已提交
1459
            res = filesToIndexQuery.store();
S
starlord 已提交
1460
        }  // Scoped Connection
1461

S
starlord 已提交
1462
        Status ret;
G
groot 已提交
1463 1464
        std::map<std::string, TableSchema> groups;
        TableFileSchema table_file;
S
starlord 已提交
1465
        for (auto& resRow : res) {
G
groot 已提交
1466 1467 1468 1469 1470 1471 1472 1473 1474
            table_file.id_ = resRow["id"];  // implicit conversion
            resRow["table_id"].to_string(table_file.table_id_);
            table_file.engine_type_ = resRow["engine_type"];
            resRow["file_id"].to_string(table_file.file_id_);
            table_file.file_type_ = resRow["file_type"];
            table_file.file_size_ = resRow["file_size"];
            table_file.row_count_ = resRow["row_count"];
            table_file.date_ = resRow["date"];
            table_file.created_on_ = resRow["created_on"];
Z
update  
zhiru 已提交
1475

G
groot 已提交
1476 1477 1478 1479 1480 1481 1482
            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;
1483
                }
G
groot 已提交
1484
                groups[table_file.table_id_] = table_schema;
Z
zhiru 已提交
1485
            }
G
groot 已提交
1486 1487 1488 1489
            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_;
1490

G
groot 已提交
1491 1492 1493
            auto status = utils::GetTableFilePath(options_, table_file);
            if (!status.ok()) {
                ret = status;
1494
            }
Z
zhiru 已提交
1495

G
groot 已提交
1496 1497
            files.push_back(table_file);
        }
Z
update  
zhiru 已提交
1498

G
groot 已提交
1499 1500
        if (res.size() > 0) {
            ENGINE_LOG_DEBUG << "Collect " << res.size() << " to-index files";
1501
        }
G
groot 已提交
1502
        return ret;
S
starlord 已提交
1503
    } catch (std::exception& e) {
G
groot 已提交
1504
        return HandleException("GENERAL ERROR WHEN FINDING TABLE FILES TO INDEX", e.what());
1505 1506
    }
}
1507

S
starlord 已提交
1508
Status
G
groot 已提交
1509 1510 1511 1512
MySQLMetaImpl::FilesByType(const std::string& table_id, const std::vector<int>& file_types,
                           std::vector<std::string>& file_ids) {
    if (file_types.empty()) {
        return Status(DB_ERROR, "file types array is empty");
Z
update  
zhiru 已提交
1513 1514
    }

1515
    try {
G
groot 已提交
1516 1517 1518
        file_ids.clear();

        mysqlpp::StoreQueryResult res;
1519
        {
S
starlord 已提交
1520
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
1521

1522
            if (connectionPtr == nullptr) {
G
groot 已提交
1523
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
1524
            }
Z
zhiru 已提交
1525

G
groot 已提交
1526 1527 1528 1529
            std::string types;
            for (auto type : file_types) {
                if (!types.empty()) {
                    types += ",";
Z
update  
zhiru 已提交
1530
                }
G
groot 已提交
1531
                types += std::to_string(type);
1532
            }
Z
update  
zhiru 已提交
1533

G
groot 已提交
1534 1535 1536 1537 1538
            mysqlpp::Query hasNonIndexFilesQuery = connectionPtr->query();
            // since table_id is a unique column we just need to check whether it exists or not
            hasNonIndexFilesQuery << "SELECT file_id, file_type"
                                  << " FROM " << META_TABLEFILES << " WHERE table_id = " << mysqlpp::quote << table_id
                                  << " AND file_type in (" << types << ");";
1539

G
groot 已提交
1540
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::FilesByType: " << hasNonIndexFilesQuery.str();
Z
update  
zhiru 已提交
1541

G
groot 已提交
1542
            res = hasNonIndexFilesQuery.store();
S
starlord 已提交
1543
        }  // Scoped Connection
1544

G
groot 已提交
1545 1546 1547 1548 1549 1550 1551
        if (res.num_rows() > 0) {
            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;
            for (auto& resRow : res) {
                std::string file_id;
                resRow["file_id"].to_string(file_id);
                file_ids.push_back(file_id);
1552

G
groot 已提交
1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577
                int32_t file_type = resRow["file_type"];
                switch (file_type) {
                    case (int)TableFileSchema::RAW:
                        raw_count++;
                        break;
                    case (int)TableFileSchema::NEW:
                        new_count++;
                        break;
                    case (int)TableFileSchema::NEW_MERGE:
                        new_merge_count++;
                        break;
                    case (int)TableFileSchema::NEW_INDEX:
                        new_index_count++;
                        break;
                    case (int)TableFileSchema::TO_INDEX:
                        to_index_count++;
                        break;
                    case (int)TableFileSchema::INDEX:
                        index_count++;
                        break;
                    case (int)TableFileSchema::BACKUP:
                        backup_count++;
                        break;
                    default:
                        break;
Z
update  
zhiru 已提交
1578
                }
1579 1580
            }

G
groot 已提交
1581 1582 1583 1584 1585
            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;
        }
S
starlord 已提交
1586
    } catch (std::exception& e) {
G
groot 已提交
1587
        return HandleException("GENERAL ERROR WHEN GET FILE BY TYPE", e.what());
1588
    }
S
starlord 已提交
1589

1590 1591
    return Status::OK();
}
1592

G
groot 已提交
1593
// TODO(myh): Support swap to cloud storage
S
starlord 已提交
1594
Status
G
groot 已提交
1595 1596 1597 1598 1599
MySQLMetaImpl::Archive() {
    auto& criterias = options_.archive_conf_.GetCriterias();
    if (criterias.empty()) {
        return Status::OK();
    }
1600

G
groot 已提交
1601 1602 1603 1604 1605 1606
    for (auto& kv : criterias) {
        auto& criteria = kv.first;
        auto& limit = kv.second;
        if (criteria == engine::ARCHIVE_CONF_DAYS) {
            size_t usecs = limit * D_SEC * US_PS;
            int64_t now = utils::GetMicroSecTimeStamp();
Z
update  
zhiru 已提交
1607

G
groot 已提交
1608 1609
            try {
                mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
Z
zhiru 已提交
1610

G
groot 已提交
1611 1612 1613
                if (connectionPtr == nullptr) {
                    return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
                }
1614

G
groot 已提交
1615 1616 1617 1618 1619
                mysqlpp::Query archiveQuery = connectionPtr->query();
                archiveQuery << "UPDATE " << META_TABLEFILES
                             << " SET file_type = " << std::to_string(TableFileSchema::TO_DELETE)
                             << " WHERE created_on < " << std::to_string(now - usecs) << " AND file_type <> "
                             << std::to_string(TableFileSchema::TO_DELETE) << ";";
Z
update  
zhiru 已提交
1620

G
groot 已提交
1621 1622 1623 1624 1625 1626 1627 1628 1629 1630
                ENGINE_LOG_DEBUG << "MySQLMetaImpl::Archive: " << archiveQuery.str();

                if (!archiveQuery.exec()) {
                    return HandleException("QUERY ERROR DURING ARCHIVE", archiveQuery.error());
                }

                ENGINE_LOG_DEBUG << "Archive old files";
            } catch (std::exception& e) {
                return HandleException("GENERAL ERROR WHEN DURING ARCHIVE", e.what());
            }
Z
fix  
zhiru 已提交
1631
        }
G
groot 已提交
1632 1633 1634
        if (criteria == engine::ARCHIVE_CONF_DISK) {
            uint64_t sum = 0;
            Size(sum);
Z
fix  
zhiru 已提交
1635

G
groot 已提交
1636 1637 1638 1639 1640
            auto to_delete = (sum - limit * G);
            DiscardFiles(to_delete);

            ENGINE_LOG_DEBUG << "Archive files to free disk";
        }
1641
    }
Z
update  
zhiru 已提交
1642

1643 1644
    return Status::OK();
}
Z
zhiru 已提交
1645

S
starlord 已提交
1646
Status
G
groot 已提交
1647 1648 1649
MySQLMetaImpl::Size(uint64_t& result) {
    result = 0;

1650
    try {
G
groot 已提交
1651
        mysqlpp::StoreQueryResult res;
1652
        {
S
starlord 已提交
1653
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
1654 1655

            if (connectionPtr == nullptr) {
G
groot 已提交
1656
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
1657
            }
Z
update  
zhiru 已提交
1658

G
groot 已提交
1659 1660 1661 1662
            mysqlpp::Query getSizeQuery = connectionPtr->query();
            getSizeQuery << "SELECT IFNULL(SUM(file_size),0) AS sum"
                         << " FROM " << META_TABLEFILES << " WHERE file_type <> "
                         << std::to_string(TableFileSchema::TO_DELETE) << ";";
1663

G
groot 已提交
1664
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::Size: " << getSizeQuery.str();
1665

G
groot 已提交
1666 1667
            res = getSizeQuery.store();
        }  // Scoped Connection
1668

G
groot 已提交
1669 1670 1671 1672 1673 1674 1675 1676
        if (res.empty()) {
            result = 0;
        } else {
            result = res[0]["sum"];
        }
    } catch (std::exception& e) {
        return HandleException("GENERAL ERROR WHEN RETRIEVING SIZE", e.what());
    }
1677

G
groot 已提交
1678 1679
    return Status::OK();
}
1680

G
groot 已提交
1681 1682 1683 1684
Status
MySQLMetaImpl::CleanUp() {
    try {
        mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
1685

G
groot 已提交
1686 1687 1688
        if (connectionPtr == nullptr) {
            return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
        }
1689

G
groot 已提交
1690 1691 1692 1693 1694
        mysqlpp::Query cleanUpQuery = connectionPtr->query();
        cleanUpQuery << "SELECT table_name"
                     << " FROM information_schema.tables"
                     << " WHERE table_schema = " << mysqlpp::quote << mysql_connection_pool_->getDB()
                     << " AND table_name = " << mysqlpp::quote << META_TABLEFILES << ";";
Z
fix  
zhiru 已提交
1695

G
groot 已提交
1696
        ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUp: " << cleanUpQuery.str();
1697

G
groot 已提交
1698
        mysqlpp::StoreQueryResult res = cleanUpQuery.store();
1699

G
groot 已提交
1700 1701 1702 1703 1704 1705 1706 1707 1708 1709
        if (!res.empty()) {
            ENGINE_LOG_DEBUG << "Remove table file type as NEW";
            cleanUpQuery << "DELETE FROM " << META_TABLEFILES << " WHERE file_type IN ("
                         << std::to_string(TableFileSchema::NEW) << "," << std::to_string(TableFileSchema::NEW_MERGE)
                         << "," << std::to_string(TableFileSchema::NEW_INDEX) << ");";

            ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUp: " << cleanUpQuery.str();

            if (!cleanUpQuery.exec()) {
                return HandleException("QUERY ERROR WHEN CLEANING UP FILES", cleanUpQuery.error());
1710
            }
G
groot 已提交
1711
        }
1712

G
groot 已提交
1713 1714 1715
        if (res.size() > 0) {
            ENGINE_LOG_DEBUG << "Clean " << res.size() << " files";
        }
S
starlord 已提交
1716
    } catch (std::exception& e) {
G
groot 已提交
1717
        return HandleException("GENERAL ERROR WHEN CLEANING UP FILES", e.what());
1718
    }
S
starlord 已提交
1719

1720 1721
    return Status::OK();
}
Z
fix  
zhiru 已提交
1722

S
starlord 已提交
1723 1724
Status
MySQLMetaImpl::CleanUpFilesWithTTL(uint16_t seconds) {
1725
    auto now = utils::GetMicroSecTimeStamp();
S
starlord 已提交
1726 1727
    std::set<std::string> table_ids;

S
starlord 已提交
1728
    // remove to_delete files
1729
    try {
Y
Yu Kun 已提交
1730
        server::MetricCollector metric;
1731

1732
        {
S
starlord 已提交
1733
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
Z
update  
zhiru 已提交
1734

1735
            if (connectionPtr == nullptr) {
G
groot 已提交
1736
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
1737
            }
Z
zhiru 已提交
1738

S
starlord 已提交
1739
            mysqlpp::Query cleanUpFilesWithTTLQuery = connectionPtr->query();
G
groot 已提交
1740 1741 1742 1743
            cleanUpFilesWithTTLQuery << "SELECT id, table_id, file_id, date"
                                     << " FROM " << META_TABLEFILES
                                     << " WHERE file_type = " << std::to_string(TableFileSchema::TO_DELETE)
                                     << " AND updated_time < " << std::to_string(now - seconds * US_PS) << ";";
Z
update  
zhiru 已提交
1744

1745
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << cleanUpFilesWithTTLQuery.str();
Z
update  
zhiru 已提交
1746

S
starlord 已提交
1747
            mysqlpp::StoreQueryResult res = cleanUpFilesWithTTLQuery.store();
Z
update  
zhiru 已提交
1748

1749 1750
            TableFileSchema table_file;
            std::vector<std::string> idsToDelete;
1751

S
starlord 已提交
1752 1753
            for (auto& resRow : res) {
                table_file.id_ = resRow["id"];  // implicit conversion
G
groot 已提交
1754 1755
                resRow["table_id"].to_string(table_file.table_id_);
                resRow["file_id"].to_string(table_file.file_id_);
1756
                table_file.date_ = resRow["date"];
Z
update  
zhiru 已提交
1757

1758 1759
                utils::DeleteTableFilePath(options_, table_file);

S
starlord 已提交
1760
                ENGINE_LOG_DEBUG << "Removing file id:" << table_file.id_ << " location:" << table_file.location_;
1761 1762

                idsToDelete.emplace_back(std::to_string(table_file.id_));
S
starlord 已提交
1763
                table_ids.insert(table_file.table_id_);
1764 1765 1766 1767
            }

            if (!idsToDelete.empty()) {
                std::stringstream idsToDeleteSS;
S
starlord 已提交
1768
                for (auto& id : idsToDelete) {
1769
                    idsToDeleteSS << "id = " << id << " OR ";
1770
                }
1771

1772
                std::string idsToDeleteStr = idsToDeleteSS.str();
S
starlord 已提交
1773
                idsToDeleteStr = idsToDeleteStr.substr(0, idsToDeleteStr.size() - 4);  // remove the last " OR "
G
groot 已提交
1774
                cleanUpFilesWithTTLQuery << "DELETE FROM " << META_TABLEFILES << " WHERE " << idsToDeleteStr << ";";
1775

1776 1777 1778
                ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << cleanUpFilesWithTTLQuery.str();

                if (!cleanUpFilesWithTTLQuery.exec()) {
S
starlord 已提交
1779 1780
                    return HandleException("QUERY ERROR WHEN CLEANING UP FILES WITH TTL",
                                           cleanUpFilesWithTTLQuery.error());
1781 1782
                }
            }
1783

S
starlord 已提交
1784
            if (res.size() > 0) {
1785 1786
                ENGINE_LOG_DEBUG << "Clean " << res.size() << " files deleted in " << seconds << " seconds";
            }
S
starlord 已提交
1787 1788
        }  // Scoped Connection
    } catch (std::exception& e) {
S
starlord 已提交
1789
        return HandleException("GENERAL ERROR WHEN CLEANING UP FILES WITH TTL", e.what());
Z
update  
zhiru 已提交
1790
    }
1791

S
starlord 已提交
1792
    // remove to_delete tables
1793
    try {
Y
Yu Kun 已提交
1794
        server::MetricCollector metric;
1795 1796

        {
S
starlord 已提交
1797
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
Z
zhiru 已提交
1798

Z
update  
zhiru 已提交
1799
            if (connectionPtr == nullptr) {
G
groot 已提交
1800
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
Z
update  
zhiru 已提交
1801 1802
            }

S
starlord 已提交
1803
            mysqlpp::Query cleanUpFilesWithTTLQuery = connectionPtr->query();
G
groot 已提交
1804 1805 1806
            cleanUpFilesWithTTLQuery << "SELECT id, table_id"
                                     << " FROM " << META_TABLES
                                     << " WHERE state = " << std::to_string(TableSchema::TO_DELETE) << ";";
1807

1808 1809
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << cleanUpFilesWithTTLQuery.str();

S
starlord 已提交
1810
            mysqlpp::StoreQueryResult res = cleanUpFilesWithTTLQuery.store();
Z
update  
zhiru 已提交
1811

Z
update  
zhiru 已提交
1812
            if (!res.empty()) {
1813
                std::stringstream idsToDeleteSS;
S
starlord 已提交
1814
                for (auto& resRow : res) {
1815 1816 1817
                    size_t id = resRow["id"];
                    std::string table_id;
                    resRow["table_id"].to_string(table_id);
Z
update  
zhiru 已提交
1818

S
starlord 已提交
1819
                    utils::DeleteTablePath(options_, table_id, false);  // only delete empty folder
1820 1821

                    idsToDeleteSS << "id = " << std::to_string(id) << " OR ";
Z
update  
zhiru 已提交
1822
                }
1823
                std::string idsToDeleteStr = idsToDeleteSS.str();
S
starlord 已提交
1824
                idsToDeleteStr = idsToDeleteStr.substr(0, idsToDeleteStr.size() - 4);  // remove the last " OR "
G
groot 已提交
1825
                cleanUpFilesWithTTLQuery << "DELETE FROM " << META_TABLES << " WHERE " << idsToDeleteStr << ";";
1826

1827
                ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << cleanUpFilesWithTTLQuery.str();
Z
update  
zhiru 已提交
1828

1829
                if (!cleanUpFilesWithTTLQuery.exec()) {
S
starlord 已提交
1830 1831
                    return HandleException("QUERY ERROR WHEN CLEANING UP TABLES WITH TTL",
                                           cleanUpFilesWithTTLQuery.error());
1832 1833
                }
            }
1834

S
starlord 已提交
1835
            if (res.size() > 0) {
1836 1837
                ENGINE_LOG_DEBUG << "Remove " << res.size() << " tables from meta";
            }
S
starlord 已提交
1838 1839
        }  // Scoped Connection
    } catch (std::exception& e) {
S
starlord 已提交
1840
        return HandleException("GENERAL ERROR WHEN CLEANING UP TABLES WITH TTL", e.what());
Z
update  
zhiru 已提交
1841 1842
    }

S
starlord 已提交
1843 1844
    // remove deleted table folder
    // don't remove table folder until all its files has been deleted
S
starlord 已提交
1845
    try {
Y
Yu Kun 已提交
1846
        server::MetricCollector metric;
S
starlord 已提交
1847 1848

        {
S
starlord 已提交
1849
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
S
starlord 已提交
1850 1851

            if (connectionPtr == nullptr) {
G
groot 已提交
1852
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
S
starlord 已提交
1853 1854
            }

S
starlord 已提交
1855
            for (auto& table_id : table_ids) {
S
starlord 已提交
1856
                mysqlpp::Query cleanUpFilesWithTTLQuery = connectionPtr->query();
G
groot 已提交
1857 1858 1859
                cleanUpFilesWithTTLQuery << "SELECT file_id"
                                         << " FROM " << META_TABLEFILES << " WHERE table_id = " << mysqlpp::quote
                                         << table_id << ";";
S
starlord 已提交
1860 1861 1862

                ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << cleanUpFilesWithTTLQuery.str();

S
starlord 已提交
1863
                mysqlpp::StoreQueryResult res = cleanUpFilesWithTTLQuery.store();
S
starlord 已提交
1864 1865 1866 1867 1868

                if (res.empty()) {
                    utils::DeleteTablePath(options_, table_id);
                }
            }
1869

S
starlord 已提交
1870
            if (table_ids.size() > 0) {
1871 1872
                ENGINE_LOG_DEBUG << "Remove " << table_ids.size() << " tables folder";
            }
S
starlord 已提交
1873
        }
S
starlord 已提交
1874
    } catch (std::exception& e) {
S
starlord 已提交
1875
        return HandleException("GENERAL ERROR WHEN CLEANING UP TABLES WITH TTL", e.what());
S
starlord 已提交
1876 1877
    }

1878 1879
    return Status::OK();
}
1880

S
starlord 已提交
1881
Status
S
starlord 已提交
1882
MySQLMetaImpl::Count(const std::string& table_id, uint64_t& result) {
1883
    try {
Y
Yu Kun 已提交
1884
        server::MetricCollector metric;
1885 1886 1887 1888 1889 1890 1891

        TableSchema table_schema;
        table_schema.table_id_ = table_id;
        auto status = DescribeTable(table_schema);

        if (!status.ok()) {
            return status;
1892
        }
Z
zhiru 已提交
1893

S
starlord 已提交
1894
        mysqlpp::StoreQueryResult res;
1895
        {
S
starlord 已提交
1896
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
Z
zhiru 已提交
1897

Z
update  
zhiru 已提交
1898
            if (connectionPtr == nullptr) {
G
groot 已提交
1899
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
Z
update  
zhiru 已提交
1900 1901
            }

S
starlord 已提交
1902
            mysqlpp::Query countQuery = connectionPtr->query();
G
groot 已提交
1903 1904 1905 1906 1907
            countQuery << "SELECT row_count"
                       << " FROM " << META_TABLEFILES << " WHERE table_id = " << mysqlpp::quote << table_id
                       << " AND (file_type = " << std::to_string(TableFileSchema::RAW)
                       << " OR file_type = " << std::to_string(TableFileSchema::TO_INDEX)
                       << " OR file_type = " << std::to_string(TableFileSchema::INDEX) << ");";
Z
update  
zhiru 已提交
1908

1909
            ENGINE_LOG_DEBUG << "MySQLMetaImpl::Count: " << countQuery.str();
Z
update  
zhiru 已提交
1910

1911
            res = countQuery.store();
S
starlord 已提交
1912
        }  // Scoped Connection
1913 1914

        result = 0;
S
starlord 已提交
1915
        for (auto& resRow : res) {
S
starlord 已提交
1916
            size_t size = resRow["row_count"];
1917
            result += size;
1918
        }
S
starlord 已提交
1919
    } catch (std::exception& e) {
S
starlord 已提交
1920
        return HandleException("GENERAL ERROR WHEN RETRIEVING COUNT", e.what());
Z
update  
zhiru 已提交
1921
    }
S
starlord 已提交
1922

1923 1924 1925
    return Status::OK();
}

S
starlord 已提交
1926 1927
Status
MySQLMetaImpl::DropAll() {
1928
    try {
S
starlord 已提交
1929
        ENGINE_LOG_DEBUG << "Drop all mysql meta";
S
starlord 已提交
1930
        mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
1931 1932

        if (connectionPtr == nullptr) {
G
groot 已提交
1933
            return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
Z
zhiru 已提交
1934
        }
1935

S
starlord 已提交
1936
        mysqlpp::Query dropTableQuery = connectionPtr->query();
1937
        dropTableQuery << "DROP TABLE IF EXISTS " << TABLES_SCHEMA.name() << ", " << TABLEFILES_SCHEMA.name() << ";";
1938 1939 1940 1941 1942 1943

        ENGINE_LOG_DEBUG << "MySQLMetaImpl::DropAll: " << dropTableQuery.str();

        if (dropTableQuery.exec()) {
            return Status::OK();
        }
S
starlord 已提交
1944
        return HandleException("QUERY ERROR WHEN DROPPING ALL", dropTableQuery.error());
S
starlord 已提交
1945
    } catch (std::exception& e) {
S
starlord 已提交
1946
        return HandleException("GENERAL ERROR WHEN DROPPING ALL", e.what());
1947 1948 1949
    }
}

G
groot 已提交
1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980 1981 1982 1983 1984 1985 1986 1987 1988 1989 1990 1991 1992 1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011 2012 2013 2014 2015
Status
MySQLMetaImpl::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;
        bool status;
        {
            mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);

            if (connectionPtr == nullptr) {
                return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
            }

            mysqlpp::Query discardFilesQuery = connectionPtr->query();
            discardFilesQuery << "SELECT id, file_size"
                              << " FROM " << META_TABLEFILES << " WHERE file_type <> "
                              << std::to_string(TableFileSchema::TO_DELETE) << " ORDER BY id ASC "
                              << " LIMIT 10;";

            ENGINE_LOG_DEBUG << "MySQLMetaImpl::DiscardFiles: " << discardFilesQuery.str();

            mysqlpp::StoreQueryResult res = discardFilesQuery.store();
            if (res.num_rows() == 0) {
                return Status::OK();
            }

            TableFileSchema table_file;
            std::stringstream idsToDiscardSS;
            for (auto& resRow : res) {
                if (to_discard_size <= 0) {
                    break;
                }
                table_file.id_ = resRow["id"];
                table_file.file_size_ = resRow["file_size"];
                idsToDiscardSS << "id = " << std::to_string(table_file.id_) << " OR ";
                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_;
            }

            std::string idsToDiscardStr = idsToDiscardSS.str();
            idsToDiscardStr = idsToDiscardStr.substr(0, idsToDiscardStr.size() - 4);  // remove the last " OR "

            discardFilesQuery << "UPDATE " << META_TABLEFILES
                              << " SET file_type = " << std::to_string(TableFileSchema::TO_DELETE)
                              << " ,updated_time = " << std::to_string(utils::GetMicroSecTimeStamp()) << " WHERE "
                              << idsToDiscardStr << ";";

            ENGINE_LOG_DEBUG << "MySQLMetaImpl::DiscardFiles: " << discardFilesQuery.str();

            status = discardFilesQuery.exec();
            if (!status) {
                return HandleException("QUERY ERROR WHEN DISCARDING FILES", discardFilesQuery.error());
            }
        }  // Scoped Connection

        return DiscardFiles(to_discard_size);
    } catch (std::exception& e) {
        return HandleException("GENERAL ERROR WHEN DISCARDING FILES", e.what());
    }
}

S
starlord 已提交
2016 2017 2018
}  // namespace meta
}  // namespace engine
}  // namespace milvus