GrpcRequestTask.cpp 25.7 KB
Newer Older
K
kun yu 已提交
1
/*******************************************************************************
Y
Yu Kun 已提交
2 3 4 5
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
Y
Yu Kun 已提交
6
#include "GrpcRequestTask.h"
K
kun yu 已提交
7
#include "../ServerConfig.h"
K
kun yu 已提交
8 9 10 11
#include "utils/CommonUtil.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
K
kun yu 已提交
12
#include "../DBWrapper.h"
K
kun yu 已提交
13
#include "version.h"
Y
Yu Kun 已提交
14
#include "GrpcMilvusServer.h"
K
kun yu 已提交
15

K
kun yu 已提交
16
#include "src/server/Server.h"
K
kun yu 已提交
17 18 19 20

namespace zilliz {
namespace milvus {
namespace server {
Y
Yu Kun 已提交
21 22 23 24 25
namespace grpc {

static const char *DQL_TASK_GROUP = "dql";
static const char *DDL_DML_TASK_GROUP = "ddl_dml";
static const char *PING_TASK_GROUP = "ping";
K
kun yu 已提交
26 27 28 29 30 31 32 33 34 35 36 37 38

using DB_META = zilliz::milvus::engine::meta::Meta;
using DB_DATE = zilliz::milvus::engine::meta::DateT;

namespace {
    engine::EngineType EngineType(int type) {
        static std::map<int, engine::EngineType> map_type = {
                {0, engine::EngineType::INVALID},
                {1, engine::EngineType::FAISS_IDMAP},
                {2, engine::EngineType::FAISS_IVFFLAT},
                {3, engine::EngineType::FAISS_IVFSQ8},
        };

Y
Yu Kun 已提交
39
        if (map_type.find(type) == map_type.end()) {
K
kun yu 已提交
40 41 42 43 44 45 46 47
            return engine::EngineType::INVALID;
        }

        return map_type[type];
    }

    int IndexType(engine::EngineType type) {
        static std::map<engine::EngineType, int> map_type = {
Y
Yu Kun 已提交
48 49
                {engine::EngineType::INVALID,       0},
                {engine::EngineType::FAISS_IDMAP,   1},
K
kun yu 已提交
50
                {engine::EngineType::FAISS_IVFFLAT, 2},
Y
Yu Kun 已提交
51
                {engine::EngineType::FAISS_IVFSQ8,  3},
K
kun yu 已提交
52 53
        };

Y
Yu Kun 已提交
54
        if (map_type.find(type) == map_type.end()) {
K
kun yu 已提交
55 56 57 58 59 60
            return 0;
        }

        return map_type[type];
    }

K
kun yu 已提交
61
    constexpr long DAY_SECONDS = 24 * 60 * 60;
K
kun yu 已提交
62 63 64

    void
    ConvertTimeRangeToDBDates(const std::vector<::milvus::grpc::Range> &range_array,
Y
Yu Kun 已提交
65 66 67
                              std::vector<DB_DATE> &dates,
                              ServerError &error_code,
                              std::string &error_msg) {
K
kun yu 已提交
68
        dates.clear();
Y
Yu Kun 已提交
69
        for (auto &range : range_array) {
K
kun yu 已提交
70 71
            time_t tt_start, tt_end;
            tm tm_start, tm_end;
Y
Yu Kun 已提交
72
            if (!CommonUtil::TimeStrToTime(range.start_value(), tt_start, tm_start)) {
K
kun yu 已提交
73 74 75 76 77
                error_code = SERVER_INVALID_TIME_RANGE;
                error_msg = "Invalid time range: " + range.start_value();
                return;
            }

Y
Yu Kun 已提交
78
            if (!CommonUtil::TimeStrToTime(range.end_value(), tt_end, tm_end)) {
K
kun yu 已提交
79 80 81 82 83
                error_code = SERVER_INVALID_TIME_RANGE;
                error_msg = "Invalid time range: " + range.start_value();
                return;
            }

Y
Yu Kun 已提交
84 85 86
            long days = (tt_end > tt_start) ? (tt_end - tt_start) / DAY_SECONDS : (tt_start - tt_end) /
                                                                                  DAY_SECONDS;
            if (days == 0) {
K
kun yu 已提交
87 88
                error_code = SERVER_INVALID_TIME_RANGE;
                error_msg = "Invalid time range: " + range.start_value() + " to " + range.end_value();
Y
Yu Kun 已提交
89
                return;
K
kun yu 已提交
90 91
            }

Y
Yu Kun 已提交
92 93
            for (long i = 0; i < days; i++) {
                time_t tt_day = tt_start + DAY_SECONDS * i;
K
kun yu 已提交
94 95 96
                tm tm_day;
                CommonUtil::ConvertTime(tt_day, tm_day);

Y
Yu Kun 已提交
97 98
                long date = tm_day.tm_year * 10000 + tm_day.tm_mon * 100 +
                            tm_day.tm_mday;//according to db logic
K
kun yu 已提交
99 100 101 102 103 104 105
                dates.push_back(date);
            }
        }
    }
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Y
Yu Kun 已提交
106
CreateTableTask::CreateTableTask(const ::milvus::grpc::TableSchema &schema)
Y
Yu Kun 已提交
107
        : GrpcBaseTask(DDL_DML_TASK_GROUP),
K
kun yu 已提交
108 109 110 111
          schema_(schema) {

}

Y
Yu Kun 已提交
112
BaseTaskPtr
Y
Yu Kun 已提交
113
CreateTableTask::Create(const ::milvus::grpc::TableSchema &schema) {
Y
Yu Kun 已提交
114
    return std::shared_ptr<GrpcBaseTask>(new CreateTableTask(schema));
K
kun yu 已提交
115 116
}

Y
Yu Kun 已提交
117 118
ServerError
CreateTableTask::OnExecute() {
K
kun yu 已提交
119 120 121 122
    TimeRecorder rc("CreateTableTask");

    try {
        //step 1: check arguments
K
kun yu 已提交
123
        ServerError res = ValidationUtil::ValidateTableName(schema_.table_name().table_name());
Y
Yu Kun 已提交
124
        if (res != SERVER_SUCCESS) {
K
kun yu 已提交
125 126 127
            return SetError(res, "Invalid table name: " + schema_.table_name().table_name());
        }

K
kun yu 已提交
128
        res = ValidationUtil::ValidateTableDimension(schema_.dimension());
Y
Yu Kun 已提交
129
        if (res != SERVER_SUCCESS) {
K
kun yu 已提交
130 131 132
            return SetError(res, "Invalid table dimension: " + std::to_string(schema_.dimension()));
        }

K
kun yu 已提交
133
        res = ValidationUtil::ValidateTableIndexType(schema_.index_type());
Y
Yu Kun 已提交
134
        if (res != SERVER_SUCCESS) {
K
kun yu 已提交
135 136 137 138 139
            return SetError(res, "Invalid index type: " + std::to_string(schema_.index_type()));
        }

        //step 2: construct table schema
        engine::meta::TableSchema table_info;
Y
Yu Kun 已提交
140
        table_info.dimension_ = (uint16_t) schema_.dimension();
K
kun yu 已提交
141
        table_info.table_id_ = schema_.table_name().table_name();
Y
Yu Kun 已提交
142
        table_info.engine_type_ = (int) EngineType(schema_.index_type());
K
kun yu 已提交
143 144 145 146
        table_info.store_raw_data_ = schema_.store_raw_vector();

        //step 3: create table
        engine::Status stat = DBWrapper::DB()->CreateTable(table_info);
Y
Yu Kun 已提交
147
        if (!stat.ok()) {
K
kun yu 已提交
148
            //table could exist
K
kun yu 已提交
149 150 151
            return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
        }

Y
Yu Kun 已提交
152
    } catch (std::exception &ex) {
K
kun yu 已提交
153 154 155
        return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
    }

K
kun yu 已提交
156
    rc.ElapseFromBegin("totally cost");
K
kun yu 已提交
157 158 159 160 161

    return SERVER_SUCCESS;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Y
Yu Kun 已提交
162
DescribeTableTask::DescribeTableTask(const std::string &table_name, ::milvus::grpc::TableSchema &schema)
Y
Yu Kun 已提交
163
        : GrpcBaseTask(DDL_DML_TASK_GROUP),
K
kun yu 已提交
164 165 166 167
          table_name_(table_name),
          schema_(schema) {
}

Y
Yu Kun 已提交
168
BaseTaskPtr
Y
Yu Kun 已提交
169
DescribeTableTask::Create(const std::string &table_name, ::milvus::grpc::TableSchema &schema) {
Y
Yu Kun 已提交
170
    return std::shared_ptr<GrpcBaseTask>(new DescribeTableTask(table_name, schema));
K
kun yu 已提交
171 172
}

Y
Yu Kun 已提交
173 174
ServerError
DescribeTableTask::OnExecute() {
K
kun yu 已提交
175 176 177 178
    TimeRecorder rc("DescribeTableTask");

    try {
        //step 1: check arguments
K
kun yu 已提交
179
        ServerError res = ValidationUtil::ValidateTableName(table_name_);
Y
Yu Kun 已提交
180
        if (res != SERVER_SUCCESS) {
K
kun yu 已提交
181 182 183 184 185 186 187
            return SetError(res, "Invalid table name: " + table_name_);
        }

        //step 2: get table info
        engine::meta::TableSchema table_info;
        table_info.table_id_ = table_name_;
        engine::Status stat = DBWrapper::DB()->DescribeTable(table_info);
Y
Yu Kun 已提交
188
        if (!stat.ok()) {
K
kun yu 已提交
189 190 191
            return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
        }

K
kun yu 已提交
192
        schema_.mutable_table_name()->set_table_name(table_info.table_id_);
K
kun yu 已提交
193

Y
Yu Kun 已提交
194
        schema_.set_index_type(IndexType((engine::EngineType) table_info.engine_type_));
K
kun yu 已提交
195 196
        schema_.set_dimension(table_info.dimension_);
        schema_.set_store_raw_vector(table_info.store_raw_data_);
K
kun yu 已提交
197

Y
Yu Kun 已提交
198
    } catch (std::exception &ex) {
K
kun yu 已提交
199 200 201
        return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
    }

K
kun yu 已提交
202
    rc.ElapseFromBegin("totally cost");
K
kun yu 已提交
203 204 205 206 207

    return SERVER_SUCCESS;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Y
Yu Kun 已提交
208
BuildIndexTask::BuildIndexTask(const std::string &table_name)
Y
Yu Kun 已提交
209
        : GrpcBaseTask(DDL_DML_TASK_GROUP),
K
kun yu 已提交
210 211 212
          table_name_(table_name) {
}

Y
Yu Kun 已提交
213
BaseTaskPtr
Y
Yu Kun 已提交
214
BuildIndexTask::Create(const std::string &table_name) {
Y
Yu Kun 已提交
215
    return std::shared_ptr<GrpcBaseTask>(new BuildIndexTask(table_name));
K
kun yu 已提交
216 217
}

Y
Yu Kun 已提交
218 219
ServerError
BuildIndexTask::OnExecute() {
K
kun yu 已提交
220 221 222 223
    try {
        TimeRecorder rc("BuildIndexTask");

        //step 1: check arguments
K
kun yu 已提交
224
        ServerError res = ValidationUtil::ValidateTableName(table_name_);
Y
Yu Kun 已提交
225
        if (res != SERVER_SUCCESS) {
K
kun yu 已提交
226 227 228 229 230
            return SetError(res, "Invalid table name: " + table_name_);
        }

        bool has_table = false;
        engine::Status stat = DBWrapper::DB()->HasTable(table_name_, has_table);
Y
Yu Kun 已提交
231
        if (!stat.ok()) {
K
kun yu 已提交
232 233 234
            return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
        }

Y
Yu Kun 已提交
235
        if (!has_table) {
K
kun yu 已提交
236 237 238 239 240
            return SetError(SERVER_TABLE_NOT_EXIST, "Table " + table_name_ + " not exists");
        }

        //step 2: check table existence
        stat = DBWrapper::DB()->BuildIndex(table_name_);
Y
Yu Kun 已提交
241
        if (!stat.ok()) {
K
kun yu 已提交
242 243 244
            return SetError(SERVER_BUILD_INDEX_ERROR, "Engine failed: " + stat.ToString());
        }

K
kun yu 已提交
245
        rc.ElapseFromBegin("totally cost");
Y
Yu Kun 已提交
246
    } catch (std::exception &ex) {
K
kun yu 已提交
247 248 249 250 251 252 253
        return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
    }

    return SERVER_SUCCESS;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Y
Yu Kun 已提交
254
HasTableTask::HasTableTask(const std::string &table_name, bool &has_table)
Y
Yu Kun 已提交
255
        : GrpcBaseTask(DDL_DML_TASK_GROUP),
K
kun yu 已提交
256 257 258 259 260
          table_name_(table_name),
          has_table_(has_table) {

}

Y
Yu Kun 已提交
261
BaseTaskPtr
Y
Yu Kun 已提交
262
HasTableTask::Create(const std::string &table_name, bool &has_table) {
Y
Yu Kun 已提交
263
    return std::shared_ptr<GrpcBaseTask>(new HasTableTask(table_name, has_table));
K
kun yu 已提交
264 265
}

Y
Yu Kun 已提交
266 267
ServerError
HasTableTask::OnExecute() {
K
kun yu 已提交
268 269 270 271
    try {
        TimeRecorder rc("HasTableTask");

        //step 1: check arguments
K
kun yu 已提交
272
        ServerError res = ValidationUtil::ValidateTableName(table_name_);
Y
Yu Kun 已提交
273
        if (res != SERVER_SUCCESS) {
K
kun yu 已提交
274 275 276 277 278
            return SetError(res, "Invalid table name: " + table_name_);
        }

        //step 2: check table existence
        engine::Status stat = DBWrapper::DB()->HasTable(table_name_, has_table_);
Y
Yu Kun 已提交
279
        if (!stat.ok()) {
K
kun yu 已提交
280 281 282
            return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
        }

K
kun yu 已提交
283
        rc.ElapseFromBegin("totally cost");
Y
Yu Kun 已提交
284
    } catch (std::exception &ex) {
K
kun yu 已提交
285 286 287 288 289 290 291
        return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
    }

    return SERVER_SUCCESS;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Y
Yu Kun 已提交
292
DropTableTask::DropTableTask(const std::string &table_name)
Y
Yu Kun 已提交
293
        : GrpcBaseTask(DDL_DML_TASK_GROUP),
K
kun yu 已提交
294 295 296 297
          table_name_(table_name) {

}

Y
Yu Kun 已提交
298
BaseTaskPtr
Y
Yu Kun 已提交
299
DropTableTask::Create(const std::string &table_name) {
Y
Yu Kun 已提交
300
    return std::shared_ptr<GrpcBaseTask>(new DropTableTask(table_name));
K
kun yu 已提交
301 302
}

Y
Yu Kun 已提交
303 304
ServerError
DropTableTask::OnExecute() {
K
kun yu 已提交
305 306 307 308
    try {
        TimeRecorder rc("DropTableTask");

        //step 1: check arguments
K
kun yu 已提交
309
        ServerError res = ValidationUtil::ValidateTableName(table_name_);
Y
Yu Kun 已提交
310
        if (res != SERVER_SUCCESS) {
K
kun yu 已提交
311 312 313 314 315 316 317
            return SetError(res, "Invalid table name: " + table_name_);
        }

        //step 2: check table existence
        engine::meta::TableSchema table_info;
        table_info.table_id_ = table_name_;
        engine::Status stat = DBWrapper::DB()->DescribeTable(table_info);
Y
Yu Kun 已提交
318 319
        if (!stat.ok()) {
            if (stat.IsNotFound()) {
K
kun yu 已提交
320 321 322 323 324 325
                return SetError(SERVER_TABLE_NOT_EXIST, "Table " + table_name_ + " not exists");
            } else {
                return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
            }
        }

K
kun yu 已提交
326
        rc.ElapseFromBegin("check validation");
K
kun yu 已提交
327 328 329 330

        //step 3: Drop table
        std::vector<DB_DATE> dates;
        stat = DBWrapper::DB()->DeleteTable(table_name_, dates);
Y
Yu Kun 已提交
331
        if (!stat.ok()) {
K
kun yu 已提交
332 333 334
            return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
        }

K
kun yu 已提交
335
        rc.ElapseFromBegin("total cost");
Y
Yu Kun 已提交
336
    } catch (std::exception &ex) {
K
kun yu 已提交
337 338 339 340 341 342 343
        return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
    }

    return SERVER_SUCCESS;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Y
Yu Kun 已提交
344
ShowTablesTask::ShowTablesTask(::grpc::ServerWriter<::milvus::grpc::TableName> &writer)
Y
Yu Kun 已提交
345
        : GrpcBaseTask(DDL_DML_TASK_GROUP),
K
kun yu 已提交
346 347 348 349
          writer_(writer) {

}

Y
Yu Kun 已提交
350
BaseTaskPtr
Y
Yu Kun 已提交
351
ShowTablesTask::Create(::grpc::ServerWriter<::milvus::grpc::TableName> &writer) {
Y
Yu Kun 已提交
352
    return std::shared_ptr<GrpcBaseTask>(new ShowTablesTask(writer));
K
kun yu 已提交
353 354
}

Y
Yu Kun 已提交
355 356
ServerError
ShowTablesTask::OnExecute() {
K
kun yu 已提交
357 358
    std::vector<engine::meta::TableSchema> schema_array;
    engine::Status stat = DBWrapper::DB()->AllTables(schema_array);
Y
Yu Kun 已提交
359
    if (!stat.ok()) {
K
kun yu 已提交
360 361 362
        return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
    }

Y
Yu Kun 已提交
363
    for (auto &schema : schema_array) {
K
kun yu 已提交
364 365
        ::milvus::grpc::TableName tableName;
        tableName.set_table_name(schema.table_id_);
K
kun yu 已提交
366 367 368
        if (!writer_.Write(tableName)) {
            return SetError(SERVER_WRITE_ERROR, "Write table name failed!");
        }
K
kun yu 已提交
369 370 371 372 373
    }
    return SERVER_SUCCESS;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Y
Yu Kun 已提交
374 375
InsertVectorTask::InsertVectorTask(const ::milvus::grpc::InsertInfos &insert_infos,
                                   ::milvus::grpc::VectorIds &record_ids)
Y
Yu Kun 已提交
376
        : GrpcBaseTask(DDL_DML_TASK_GROUP),
K
kun yu 已提交
377 378 379 380 381
          insert_infos_(insert_infos),
          record_ids_(record_ids) {
    record_ids_.Clear();
}

Y
Yu Kun 已提交
382
BaseTaskPtr
Y
Yu Kun 已提交
383 384
InsertVectorTask::Create(const ::milvus::grpc::InsertInfos &insert_infos,
                         ::milvus::grpc::VectorIds &record_ids) {
Y
Yu Kun 已提交
385
    return std::shared_ptr<GrpcBaseTask>(new InsertVectorTask(insert_infos, record_ids));
K
kun yu 已提交
386 387
}

Y
Yu Kun 已提交
388 389
ServerError
InsertVectorTask::OnExecute() {
K
kun yu 已提交
390 391 392 393
    try {
        TimeRecorder rc("InsertVectorTask");

        //step 1: check arguments
K
kun yu 已提交
394
        ServerError res = ValidationUtil::ValidateTableName(insert_infos_.table_name());
Y
Yu Kun 已提交
395
        if (res != SERVER_SUCCESS) {
K
kun yu 已提交
396 397
            return SetError(res, "Invalid table name: " + insert_infos_.table_name());
        }
Y
Yu Kun 已提交
398
        if (insert_infos_.row_record_array().empty()) {
K
kun yu 已提交
399 400 401 402 403 404 405
            return SetError(SERVER_INVALID_ROWRECORD_ARRAY, "Row record array is empty");
        }

        //step 2: check table existence
        engine::meta::TableSchema table_info;
        table_info.table_id_ = insert_infos_.table_name();
        engine::Status stat = DBWrapper::DB()->DescribeTable(table_info);
Y
Yu Kun 已提交
406 407 408 409
        if (!stat.ok()) {
            if (stat.IsNotFound()) {
                return SetError(SERVER_TABLE_NOT_EXIST,
                                "Table " + insert_infos_.table_name() + " not exists");
K
kun yu 已提交
410 411 412 413 414
            } else {
                return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
            }
        }

K
kun yu 已提交
415 416 417 418 419 420 421
        rc.RecordSection("check validation");

#ifdef MILVUS_ENABLE_PROFILING
        std::string fname = "/tmp/insert_" + std::to_string(this->record_array_.size()) +
                            "_" + GetCurrTimeStr() + ".profiling";
        ProfilerStart(fname.c_str());
#endif
K
kun yu 已提交
422 423

        //step 3: prepare float data
K
kun yu 已提交
424
        std::vector<float> vec_f(insert_infos_.row_record_array_size() * table_info.dimension_, 0);
K
kun yu 已提交
425

K
kun yu 已提交
426
        // TODO: change to one dimension array in protobuf or use multiple-thread to copy the data
K
kun yu 已提交
427 428
        for (size_t i = 0; i < insert_infos_.row_record_array_size(); i++) {
            for (size_t j = 0; j < table_info.dimension_; j++) {
K
kun yu 已提交
429 430 431 432 433 434 435
                if (insert_infos_.row_record_array(i).vector_data().empty()) {
                    return SetError(SERVER_INVALID_ROWRECORD_ARRAY, "Row record float array is empty");
                }
                uint64_t vec_dim = insert_infos_.row_record_array(i).vector_data().size();
                if (vec_dim != table_info.dimension_) {
                    ServerError error_code = SERVER_INVALID_VECTOR_DIMENSION;
                    std::string error_msg = "Invalid rowrecord dimension: " + std::to_string(vec_dim)
Y
Yu Kun 已提交
436 437
                                            + " vs. table dimension:" +
                                            std::to_string(table_info.dimension_);
K
kun yu 已提交
438 439
                    return SetError(error_code, error_msg);
                }
K
kun yu 已提交
440 441 442 443
                vec_f[i * table_info.dimension_ + j] = insert_infos_.row_record_array(i).vector_data(j);
            }
        }

K
kun yu 已提交
444
        rc.ElapseFromBegin("prepare vectors data");
K
kun yu 已提交
445 446

        //step 4: insert vectors
Y
Yu Kun 已提交
447
        auto vec_count = (uint64_t) insert_infos_.row_record_array_size();
K
kun yu 已提交
448
        std::vector<int64_t> vec_ids(record_ids_.vector_id_array_size(), 0);
K
kun yu 已提交
449

Y
Yu Kun 已提交
450 451
        stat = DBWrapper::DB()->InsertVectors(insert_infos_.table_name(), vec_count, vec_f.data(),
                                              vec_ids);
K
kun yu 已提交
452
        rc.ElapseFromBegin("add vectors to engine");
Y
Yu Kun 已提交
453
        if (!stat.ok()) {
K
kun yu 已提交
454 455
            return SetError(SERVER_CACHE_ERROR, "Cache error: " + stat.ToString());
        }
K
kun yu 已提交
456 457
        for (int64_t id : vec_ids) {
            record_ids_.add_vector_id_array(id);
K
kun yu 已提交
458 459
        }

K
kun yu 已提交
460
        auto ids_size = record_ids_.vector_id_array_size();
Y
Yu Kun 已提交
461
        if (ids_size != vec_count) {
K
kun yu 已提交
462
            std::string msg = "Add " + std::to_string(vec_count) + " vectors but only return "
K
kun yu 已提交
463
                              + std::to_string(ids_size) + " id";
K
kun yu 已提交
464 465 466
            return SetError(SERVER_ILLEGAL_VECTOR_ID, msg);
        }

K
kun yu 已提交
467 468 469 470 471 472
#ifdef MILVUS_ENABLE_PROFILING
        ProfilerStop();
#endif

        rc.RecordSection("add vectors to engine");
        rc.ElapseFromBegin("total cost");
K
kun yu 已提交
473

Y
Yu Kun 已提交
474
    } catch (std::exception &ex) {
K
kun yu 已提交
475 476 477 478 479 480 481
        return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
    }

    return SERVER_SUCCESS;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Y
Yu Kun 已提交
482 483 484
SearchVectorTask::SearchVectorTask(const ::milvus::grpc::SearchVectorInfos &search_vector_infos,
                                   const std::vector<std::string> &file_id_array,
                                   ::grpc::ServerWriter<::milvus::grpc::TopKQueryResult> &writer)
Y
Yu Kun 已提交
485
        : GrpcBaseTask(DQL_TASK_GROUP),
K
kun yu 已提交
486
          search_vector_infos_(search_vector_infos),
K
kun yu 已提交
487 488 489 490 491
          file_id_array_(file_id_array),
          writer_(writer) {

}

Y
Yu Kun 已提交
492
BaseTaskPtr
Y
Yu Kun 已提交
493 494 495
SearchVectorTask::Create(const ::milvus::grpc::SearchVectorInfos &search_vector_infos,
                         const std::vector<std::string> &file_id_array,
                         ::grpc::ServerWriter<::milvus::grpc::TopKQueryResult> &writer) {
Y
Yu Kun 已提交
496
    return std::shared_ptr<GrpcBaseTask>(new SearchVectorTask(search_vector_infos, file_id_array,
Y
Yu Kun 已提交
497
                                                              writer));
K
kun yu 已提交
498 499
}

Y
Yu Kun 已提交
500 501
ServerError
SearchVectorTask::OnExecute() {
K
kun yu 已提交
502 503 504 505
    try {
        TimeRecorder rc("SearchVectorTask");

        //step 1: check arguments
K
kun yu 已提交
506
        std::string table_name_ = search_vector_infos_.table_name();
K
kun yu 已提交
507
        ServerError res = ValidationUtil::ValidateTableName(table_name_);
Y
Yu Kun 已提交
508
        if (res != SERVER_SUCCESS) {
K
kun yu 已提交
509 510 511
            return SetError(res, "Invalid table name: " + table_name_);
        }

K
kun yu 已提交
512
        int top_k_ = search_vector_infos_.topk();
K
kun yu 已提交
513

Y
Yu Kun 已提交
514
        if (top_k_ <= 0 || top_k_ > 1024) {
K
kun yu 已提交
515 516 517
            return SetError(SERVER_INVALID_TOPK, "Invalid topk: " + std::to_string(
                    top_k_));
        }
Y
Yu Kun 已提交
518
        if (search_vector_infos_.query_record_array().empty()) {
K
kun yu 已提交
519 520 521 522 523 524 525
            return SetError(SERVER_INVALID_ROWRECORD_ARRAY, "Row record array is empty");
        }

        //step 2: check table existence
        engine::meta::TableSchema table_info;
        table_info.table_id_ = table_name_;
        engine::Status stat = DBWrapper::DB()->DescribeTable(table_info);
Y
Yu Kun 已提交
526 527
        if (!stat.ok()) {
            if (stat.IsNotFound()) {
K
kun yu 已提交
528 529 530 531 532 533 534 535 536 537 538 539
                return SetError(SERVER_TABLE_NOT_EXIST, "Table " + table_name_ + " not exists");
            } else {
                return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
            }
        }

        //step 3: check date range, and convert to db dates
        std::vector<DB_DATE> dates;
        ServerError error_code = SERVER_SUCCESS;
        std::string error_msg;

        std::vector<::milvus::grpc::Range> range_array;
K
kun yu 已提交
540 541
        for (size_t i = 0; i < search_vector_infos_.query_range_array_size(); i++) {
            range_array.emplace_back(search_vector_infos_.query_range_array(i));
K
kun yu 已提交
542 543
        }
        ConvertTimeRangeToDBDates(range_array, dates, error_code, error_msg);
Y
Yu Kun 已提交
544
        if (error_code != SERVER_SUCCESS) {
K
kun yu 已提交
545 546 547
            return SetError(error_code, error_msg);
        }

K
kun yu 已提交
548 549 550 551 552 553 554 555
        double span_check = rc.RecordSection("check validation");

#ifdef MILVUS_ENABLE_PROFILING
        std::string fname = "/tmp/search_nq_" + std::to_string(this->record_array_.size()) +
                            "_top_" + std::to_string(this->top_k_) + "_" +
                            GetCurrTimeStr() + ".profiling";
        ProfilerStart(fname.c_str());
#endif
K
kun yu 已提交
556 557

        //step 3: prepare float data
K
kun yu 已提交
558 559 560
        auto record_array_size = search_vector_infos_.query_record_array_size();
        std::vector<float> vec_f(record_array_size * table_info.dimension_, 0);
        for (size_t i = 0; i < record_array_size; i++) {
K
kun yu 已提交
561
            for (size_t j = 0; j < table_info.dimension_; j++) {
K
kun yu 已提交
562
                if (search_vector_infos_.query_record_array(i).vector_data().empty()) {
Y
Yu Kun 已提交
563 564
                    return SetError(SERVER_INVALID_ROWRECORD_ARRAY,
                                    "Query record float array is empty");
K
kun yu 已提交
565
                }
Y
Yu Kun 已提交
566 567
                uint64_t query_vec_dim = search_vector_infos_.query_record_array(
                        i).vector_data().size();
K
kun yu 已提交
568 569
                if (query_vec_dim != table_info.dimension_) {
                    ServerError error_code = SERVER_INVALID_VECTOR_DIMENSION;
Y
Yu Kun 已提交
570 571 572
                    std::string error_msg =
                            "Invalid rowrecord dimension: " + std::to_string(query_vec_dim)
                            + " vs. table dimension:" + std::to_string(table_info.dimension_);
K
kun yu 已提交
573 574
                    return SetError(error_code, error_msg);
                }
Y
Yu Kun 已提交
575 576
                vec_f[i * table_info.dimension_ + j] = search_vector_infos_.query_record_array(
                        i).vector_data(j);
K
kun yu 已提交
577 578
            }
        }
K
kun yu 已提交
579
        rc.ElapseFromBegin("prepare vector data");
K
kun yu 已提交
580 581 582

        //step 4: search vectors
        engine::QueryResults results;
Y
Yu Kun 已提交
583
        auto record_count = (uint64_t) search_vector_infos_.query_record_array().size();
K
kun yu 已提交
584

Y
Yu Kun 已提交
585 586 587
        if (file_id_array_.empty()) {
            stat = DBWrapper::DB()->Query(table_name_, (size_t) top_k_, record_count, vec_f.data(),
                                          dates, results);
K
kun yu 已提交
588 589
        } else {
            stat = DBWrapper::DB()->Query(table_name_, file_id_array_,
Y
Yu Kun 已提交
590
                                          (size_t) top_k_, record_count, vec_f.data(), dates, results);
K
kun yu 已提交
591 592
        }

K
kun yu 已提交
593
        rc.ElapseFromBegin("search vectors from engine");
Y
Yu Kun 已提交
594
        if (!stat.ok()) {
K
kun yu 已提交
595 596 597
            return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
        }

Y
Yu Kun 已提交
598
        if (results.empty()) {
K
kun yu 已提交
599 600 601
            return SERVER_SUCCESS; //empty table
        }

Y
Yu Kun 已提交
602
        if (results.size() != record_count) {
K
kun yu 已提交
603 604 605 606 607
            std::string msg = "Search " + std::to_string(record_count) + " vectors but only return "
                              + std::to_string(results.size()) + " results";
            return SetError(SERVER_ILLEGAL_SEARCH_RESULT, msg);
        }

K
kun yu 已提交
608
        rc.ElapseFromBegin("do search");
K
kun yu 已提交
609 610

        //step 5: construct result array
Y
Yu Kun 已提交
611 612
        for (uint64_t i = 0; i < record_count; i++) {
            auto &result = results[i];
K
kun yu 已提交
613
            const auto &record = search_vector_infos_.query_record_array(i);
K
kun yu 已提交
614
            ::milvus::grpc::TopKQueryResult grpc_topk_result;
Y
Yu Kun 已提交
615
            for (auto &pair : result) {
K
kun yu 已提交
616 617 618 619
                ::milvus::grpc::QueryResult *grpc_result = grpc_topk_result.add_query_result_arrays();
                grpc_result->set_id(pair.first);
                grpc_result->set_distance(pair.second);
            }
K
kun yu 已提交
620 621 622
            if (!writer_.Write(grpc_topk_result)) {
                return SetError(SERVER_WRITE_ERROR, "Write topk result failed!");
            }
K
kun yu 已提交
623
        }
K
kun yu 已提交
624 625 626 627 628 629 630 631 632

#ifdef MILVUS_ENABLE_PROFILING
        ProfilerStop();
#endif

        double span_result = rc.RecordSection("construct result");
        rc.ElapseFromBegin("totally cost");

        //step 6: print time cost percent
K
kun yu 已提交
633

Y
Yu Kun 已提交
634
    } catch (std::exception &ex) {
K
kun yu 已提交
635 636 637 638 639 640 641
        return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
    }

    return SERVER_SUCCESS;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Y
Yu Kun 已提交
642
GetTableRowCountTask::GetTableRowCountTask(const std::string &table_name, int64_t &row_count)
Y
Yu Kun 已提交
643
        : GrpcBaseTask(DDL_DML_TASK_GROUP),
K
kun yu 已提交
644 645 646 647 648
          table_name_(table_name),
          row_count_(row_count) {

}

Y
Yu Kun 已提交
649
BaseTaskPtr
Y
Yu Kun 已提交
650
GetTableRowCountTask::Create(const std::string &table_name, int64_t &row_count) {
Y
Yu Kun 已提交
651
    return std::shared_ptr<GrpcBaseTask>(new GetTableRowCountTask(table_name, row_count));
K
kun yu 已提交
652 653
}

Y
Yu Kun 已提交
654 655
ServerError
GetTableRowCountTask::OnExecute() {
K
kun yu 已提交
656 657 658 659 660
    try {
        TimeRecorder rc("GetTableRowCountTask");

        //step 1: check arguments
        ServerError res = SERVER_SUCCESS;
K
kun yu 已提交
661
        res = ValidationUtil::ValidateTableName(table_name_);
Y
Yu Kun 已提交
662
        if (res != SERVER_SUCCESS) {
K
kun yu 已提交
663 664 665 666 667 668 669 670 671 672 673 674
            return SetError(res, "Invalid table name: " + table_name_);
        }

        //step 2: get row count
        uint64_t row_count = 0;
        engine::Status stat = DBWrapper::DB()->GetTableRowCount(table_name_, row_count);
        if (!stat.ok()) {
            return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
        }

        row_count_ = (int64_t) row_count;

K
kun yu 已提交
675
        rc.ElapseFromBegin("total cost");
K
kun yu 已提交
676

Y
Yu Kun 已提交
677
    } catch (std::exception &ex) {
K
kun yu 已提交
678 679 680 681 682 683 684
        return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
    }

    return SERVER_SUCCESS;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Y
Yu Kun 已提交
685
PingTask::PingTask(const std::string &cmd, std::string &result)
Y
Yu Kun 已提交
686
        : GrpcBaseTask(PING_TASK_GROUP),
K
kun yu 已提交
687 688 689 690 691
          cmd_(cmd),
          result_(result) {

}

Y
Yu Kun 已提交
692
BaseTaskPtr
Y
Yu Kun 已提交
693
PingTask::Create(const std::string &cmd, std::string &result) {
Y
Yu Kun 已提交
694
    return std::shared_ptr<GrpcBaseTask>(new PingTask(cmd, result));
K
kun yu 已提交
695 696
}

Y
Yu Kun 已提交
697 698
ServerError
PingTask::OnExecute() {
Y
Yu Kun 已提交
699
    if (cmd_ == "version") {
K
kun yu 已提交
700
        result_ = MILVUS_VERSION;
K
kun yu 已提交
701 702
    } else {
        result_ = "OK";
K
kun yu 已提交
703 704 705 706 707
    }

    return SERVER_SUCCESS;
}

Y
Yu Kun 已提交
708
}
K
kun yu 已提交
709 710 711
}
}
}