milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

MS-413 Remove thrift dependency 

Former-commit-id: 00ef82d80e81b0076299436ec109025e6551e215
pull/191/head
starlord 2019-08-24 12:20:32 +08:00
parent 1766a316c9
commit 86ca848f2d
39 changed files with 48 additions and 12335 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cpp/CHANGELOG.md
View File

@ -47,6 +47,7 @@ Please mark all change in change log and use the ticket from JIRA.
- MS-407 - Reconstruct MetricsCollector
- MS-408 - Add device_id in resource construct function
- MS-409 - Using new scheduler
- MS-413 - Remove thrift dependency
## New Feature
- MS-343 - Implement ResourceMgr

4
cpp/CMakeLists.txt
View File

@ -71,10 +71,6 @@ set(CUDA_NVCC_FLAGS "${CUDA_NVCC_FLAGS} -O0 -g")
message("CUDA_TOOLKIT_ROOT_DIR=${CUDA_TOOLKIT_ROOT_DIR}")
message("CUDA_NVCC_FLAGS=${CUDA_NVCC_FLAGS}")
set(MILVUS_WITH_THRIFT "OFF")
set(GPU_VERSION "ON")
add_definitions("-DGPU_VERSION")
set(CMAKE_CXX_STANDARD 14)
set(CMAKE_CXX_STANDARD_REQUIRED on)

2
cpp/cmake/DefineOptions.cmake
View File

@ -90,8 +90,6 @@ define_option(MILVUS_WITH_SQLITE_ORM "Build with SQLite ORM library" ON)
define_option(MILVUS_WITH_MYSQLPP "Build with MySQL++" ON)
define_option(MILVUS_WITH_THRIFT "Build with Apache Thrift library" OFF)
define_option(MILVUS_WITH_YAMLCPP "Build with yaml-cpp library" ON)
define_option(MILVUS_WITH_ZLIB "Build with zlib compression" ON)

148
cpp/cmake/ThirdPartyPackages.cmake
View File

@ -84,8 +84,6 @@ macro(build_dependency DEPENDENCY_NAME)
build_sqlite()
elseif ("${DEPENDENCY_NAME}" STREQUAL "SQLite_ORM")
build_sqlite_orm()
elseif("${DEPENDENCY_NAME}" STREQUAL "Thrift")
build_thrift()
elseif("${DEPENDENCY_NAME}" STREQUAL "yaml-cpp")
build_yamlcpp()
elseif("${DEPENDENCY_NAME}" STREQUAL "ZLIB")
@ -403,14 +401,6 @@ else()
endif()
set(SQLITE_ORM_MD5 "ba9a405a8a1421c093aa8ce988ff8598")
if(DEFINED ENV{MILVUS_THRIFT_URL})
set(THRIFT_SOURCE_URL "$ENV{MILVUS_THRIFT_URL}")
else()
set(THRIFT_SOURCE_URL
"https://github.com/apache/thrift/archive/${THRIFT_VERSION}.tar.gz")
endif()
set(THRIFT_MD5 "ff9af01fec424b5a279fa8a3c9e95c0c")
if(DEFINED ENV{MILVUS_YAMLCPP_URL})
set(YAMLCPP_SOURCE_URL "$ENV{MILVUS_YAMLCPP_URL}")
else()
@ -1943,127 +1933,6 @@ if(MILVUS_WITH_SQLITE_ORM)
include_directories(SYSTEM "${SQLITE_ORM_INCLUDE_DIR}")
endif()
# ----------------------------------------------------------------------
# Thrift
macro(build_thrift)
message(STATUS "Building Apache Thrift-${THRIFT_VERSION} from source")
set(THRIFT_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/thrift_ep-prefix/src/thrift_ep")
set(THRIFT_INCLUDE_DIR "${THRIFT_PREFIX}/include")
set(THRIFT_COMPILER "${THRIFT_PREFIX}/bin/thrift")
set(THRIFT_CMAKE_ARGS
${EP_COMMON_CMAKE_ARGS}
"-DCMAKE_INSTALL_PREFIX=${THRIFT_PREFIX}"
"-DCMAKE_INSTALL_RPATH=${THRIFT_PREFIX}/lib"
-DBOOST_ROOT=${BOOST_PREFIX}
-DWITH_CPP=ON
-DWITH_STATIC_LIB=ON
-DBUILD_SHARED_LIBS=OFF
-DBUILD_TESTING=OFF
-DBUILD_EXAMPLES=OFF
-DBUILD_TUTORIALS=OFF
-DWITH_QT4=OFF
-DWITH_QT5=OFF
-DWITH_C_GLIB=OFF
-DWITH_JAVA=OFF
-DWITH_PYTHON=OFF
-DWITH_HASKELL=OFF
-DWITH_LIBEVENT=OFF
-DCMAKE_BUILD_TYPE=Release)
# Thrift also uses boost. Forward important boost settings if there were ones passed.
if(DEFINED BOOST_ROOT)
set(THRIFT_CMAKE_ARGS ${THRIFT_CMAKE_ARGS} "-DBOOST_ROOT=${BOOST_ROOT}")
endif()
if(DEFINED Boost_NAMESPACE)
set(THRIFT_CMAKE_ARGS ${THRIFT_CMAKE_ARGS} "-DBoost_NAMESPACE=${Boost_NAMESPACE}")
endif()
set(THRIFT_STATIC_LIB_NAME "${CMAKE_STATIC_LIBRARY_PREFIX}thrift")
set(THRIFT_STATIC_LIB
"${THRIFT_PREFIX}/lib/${THRIFT_STATIC_LIB_NAME}${CMAKE_STATIC_LIBRARY_SUFFIX}")
if(ZLIB_SHARED_LIB)
set(THRIFT_CMAKE_ARGS "-DZLIB_LIBRARY=${ZLIB_SHARED_LIB}" ${THRIFT_CMAKE_ARGS})
else()
set(THRIFT_CMAKE_ARGS "-DZLIB_LIBRARY=${ZLIB_STATIC_LIB}" ${THRIFT_CMAKE_ARGS})
endif()
set(THRIFT_DEPENDENCIES ${THRIFT_DEPENDENCIES} ${ZLIB_LIBRARY})
if(USE_JFROG_CACHE STREQUAL "ON")
string(MD5 THRIFT_COMBINE_MD5 "${THRIFT_MD5}${ZLIB_MD5}")
set(THRIFT_CACHE_PACKAGE_NAME "thrift_${THRIFT_COMBINE_MD5}.tar.gz")
set(THRIFT_CACHE_URL "${JFROG_ARTFACTORY_CACHE_URL}/${THRIFT_CACHE_PACKAGE_NAME}")
set(THRIFT_CACHE_PACKAGE_PATH "${THIRDPARTY_PACKAGE_CACHE}/${THRIFT_CACHE_PACKAGE_NAME}")
execute_process(COMMAND wget -q --method HEAD ${THRIFT_CACHE_URL} RESULT_VARIABLE return_code)
message(STATUS "Check the remote file ${THRIFT_CACHE_URL}. return code = ${return_code}")
if (NOT return_code EQUAL 0)
externalproject_add(thrift_ep
URL
${THRIFT_SOURCE_URL}
BUILD_BYPRODUCTS
"${THRIFT_STATIC_LIB}"
"${THRIFT_COMPILER}"
BUILD_COMMAND
${MAKE}
${MAKE_BUILD_ARGS}
CMAKE_ARGS
${THRIFT_CMAKE_ARGS}
INSTALL_COMMAND
${MAKE} install
DEPENDS
${THRIFT_DEPENDENCIES}
${EP_LOG_OPTIONS})
ExternalProject_Create_Cache(thrift_ep ${THRIFT_CACHE_PACKAGE_PATH} "${CMAKE_CURRENT_BINARY_DIR}/thrift_ep-prefix" ${JFROG_USER_NAME} ${JFROG_PASSWORD} ${THRIFT_CACHE_URL})
else()
file(DOWNLOAD ${THRIFT_CACHE_URL} ${THRIFT_CACHE_PACKAGE_PATH} STATUS status)
list(GET status 0 status_code)
message(STATUS "DOWNLOADING FROM ${THRIFT_CACHE_URL} TO ${THRIFT_CACHE_PACKAGE_PATH}. STATUS = ${status_code}")
if (status_code EQUAL 0)
ExternalProject_Use_Cache(thrift_ep ${THRIFT_CACHE_PACKAGE_PATH} ${CMAKE_CURRENT_BINARY_DIR})
endif()
endif()
else()
externalproject_add(thrift_ep
URL
${THRIFT_SOURCE_URL}
BUILD_BYPRODUCTS
"${THRIFT_STATIC_LIB}"
"${THRIFT_COMPILER}"
BUILD_COMMAND
${MAKE}
${MAKE_BUILD_ARGS}
CMAKE_ARGS
${THRIFT_CMAKE_ARGS}
INSTALL_COMMAND
${MAKE} install
DEPENDS
${THRIFT_DEPENDENCIES}
${EP_LOG_OPTIONS})
endif()
# The include directory must exist before it is referenced by a target.
file(MAKE_DIRECTORY "${THRIFT_INCLUDE_DIR}")
add_library(thrift STATIC IMPORTED)
set_target_properties(thrift
PROPERTIES IMPORTED_LOCATION "${THRIFT_STATIC_LIB}"
INTERFACE_INCLUDE_DIRECTORIES "${THRIFT_INCLUDE_DIR}")
add_dependencies(thrift thrift_ep)
endmacro()
if(MILVUS_WITH_THRIFT)
resolve_dependency(Thrift)
link_directories(SYSTEM ${THRIFT_PREFIX}/lib/)
link_directories(SYSTEM ${CMAKE_CURRENT_BINARY_DIR}/thrift_ep-prefix/src/thrift_ep-build/lib)
include_directories(SYSTEM ${THRIFT_INCLUDE_DIR})
include_directories(SYSTEM ${THRIFT_PREFIX}/lib/cpp/src)
include_directories(SYSTEM ${CMAKE_CURRENT_BINARY_DIR}/thrift_ep-prefix/src/thrift_ep-build)
endif()
# ----------------------------------------------------------------------
# yaml-cpp
@ -2683,15 +2552,12 @@ macro(build_grpc)
add_dependencies(grpc_protoc grpc_ep)
endmacro()
if(NOT MILVUS_WITH_THRIFT STREQUAL "ON")
resolve_dependency(GRPC)
resolve_dependency(GRPC)
get_target_property(GRPC_INCLUDE_DIR grpc INTERFACE_INCLUDE_DIRECTORIES)
include_directories(SYSTEM ${GRPC_INCLUDE_DIR})
link_directories(SYSTEM ${GRPC_PREFIX}/lib)
get_target_property(GRPC_INCLUDE_DIR grpc INTERFACE_INCLUDE_DIRECTORIES)
include_directories(SYSTEM ${GRPC_INCLUDE_DIR})
link_directories(SYSTEM ${GRPC_PREFIX}/lib)
set(GRPC_THIRD_PARTY_DIR ${CMAKE_CURRENT_BINARY_DIR}/grpc_ep-prefix/src/grpc_ep/third_party)
include_directories(SYSTEM ${GRPC_THIRD_PARTY_DIR}/protobuf/src)
link_directories(SYSTEM ${GRPC_PROTOBUF_LIB_DIR})
endif()
set(GRPC_THIRD_PARTY_DIR ${CMAKE_CURRENT_BINARY_DIR}/grpc_ep-prefix/src/grpc_ep/third_party)
include_directories(SYSTEM ${GRPC_THIRD_PARTY_DIR}/protobuf/src)
link_directories(SYSTEM ${GRPC_PROTOBUF_LIB_DIR})

1
cpp/conf/server_config.template
View File

@ -37,7 +37,6 @@ cache_config:
engine_config:
use_blas_threshold: 20
metric_type: L2 # compare vectors by euclidean distance(L2) or inner product(IP), optional: L2 or IP
omp_thread_num: 0 # how many compute threads be used by engine, 0 means use all cpu core to compute
resource_config:

98
cpp/src/CMakeLists.txt
View File

@ -8,7 +8,6 @@ aux_source_directory(cache cache_files)
aux_source_directory(config config_files)
aux_source_directory(server server_files)
aux_source_directory(server/grpc_impl grpcserver_files)
aux_source_directory(server/thrift_impl thriftserver_files)
aux_source_directory(utils utils_files)
aux_source_directory(db db_main_files)
aux_source_directory(db/engine db_engine_files)
@ -48,17 +47,6 @@ set(license_generator_files
license/LicenseLibrary.cpp
)
if (MILVUS_WITH_THRIFT STREQUAL "ON")
set(thrift_service_files
thrift/gen-cpp/MilvusService.cpp
thrift/gen-cpp/milvus_constants.cpp
thrift/gen-cpp/milvus_types.cpp
metrics/SystemInfo.cpp
metrics/SystemInfo.h
server/thrift_impl/ThreadPoolServer.cpp
server/thrift_impl/ThreadPoolServer.h
)
else()
set(grpc_service_files
metrics/SystemInfo.cpp
metrics/SystemInfo.h
@ -67,7 +55,6 @@ set(grpc_service_files
grpc/gen-status/status.grpc.pb.cc
grpc/gen-status/status.pb.cc
)
endif()
set(db_files
${CMAKE_CURRENT_SOURCE_DIR}/main.cpp
@ -87,25 +74,16 @@ set(s3_client_files
include_directories(/usr/include)
include_directories("${CUDA_TOOLKIT_ROOT_DIR}/include")
include_directories(thrift/gen-cpp)
include_directories(/usr/include/mysql)
include_directories(grpc/gen-status)
include_directories(grpc/gen-milvus)
if (MILVUS_WITH_THRIFT STREQUAL "ON")
message("Build thrift server")
add_definitions("-DMILVUS_ENABLE_THRIFT")
set(client_thrift_lib
thrift)
else()
message("Build grpc server")
set(client_grpc_lib
grpcpp_channelz
grpc++
grpc
grpc_protobuf
grpc_protoc)
endif()
set(client_grpc_lib
grpcpp_channelz
grpc++
grpc
grpc_protobuf
grpc_protoc)
set(third_party_libs
knowhere
@ -117,7 +95,6 @@ set(third_party_libs
lapack
easyloggingpp
sqlite
${client_thrift_lib}
${client_grpc_lib}
yaml-cpp
prometheus-cpp-push
@ -138,9 +115,6 @@ set(third_party_libs
cudart
)
if (MEGASEARCH_WITH_ARROW STREQUAL "ON")
set(third_party_libs ${third_party_libs} arrow)
endif()
if(${BUILD_FAISS_WITH_MKL} STREQUAL "ON")
set(third_party_libs ${third_party_libs}
${MKL_LIBS})
@ -156,23 +130,14 @@ if (MILVUS_ENABLE_PROFILING STREQUAL "ON")
libunwind)
endif()
if (GPU_VERSION STREQUAL "ON")
link_directories("${CUDA_TOOLKIT_ROOT_DIR}/lib64")
set(engine_libs
pthread
libgomp.a
libgfortran.a
${CUDA_TOOLKIT_ROOT_DIR}/lib64/stubs/libnvidia-ml.so
)
else()
set(engine_libs
pthread
libgomp.a
libgfortran.a
${CUDA_TOOLKIT_ROOT_DIR}/lib64/stubs/libnvidia-ml.so
)
endif ()
link_directories("${CUDA_TOOLKIT_ROOT_DIR}/lib64")
set(engine_libs
pthread
libgomp.a
libgfortran.a
${CUDA_TOOLKIT_ROOT_DIR}/lib64/stubs/libnvidia-ml.so
)
if (NOT ${CMAKE_SYSTEM_PROCESSOR} MATCHES "aarch64")
set(engine_libs
@ -215,34 +180,17 @@ set(knowhere_libs
tbb
)
if (MILVUS_WITH_THRIFT STREQUAL "ON")
add_executable(milvus_server
${config_files}
${server_files}
${thriftserver_files}
${utils_files}
${thrift_service_files}
${metrics_files}
${scheduler_srcs}
)
else()
add_executable(milvus_server
${config_files}
${server_files}
${grpcserver_files}
${utils_files}
${grpc_service_files}
${metrics_files}
${scheduler_srcs}
)
endif()
if(MILVUS_WITH_THRIFT STREQUAL "ON")
target_link_libraries(milvus_server ${server_libs} ${knowhere_libs} ${third_party_libs})
else()
target_link_libraries(milvus_server ${server_libs} ${knowhere_libs} ${third_party_libs})
endif()
add_executable(milvus_server
${config_files}
${server_files}
${grpcserver_files}
${utils_files}
${grpc_service_files}
${metrics_files}
${scheduler_srcs}
)
target_link_libraries(milvus_server ${server_libs} ${knowhere_libs} ${third_party_libs})
install(TARGETS milvus_server DESTINATION bin)

2
cpp/src/grpc/milvus.proto
View File

@ -32,7 +32,7 @@ message Range {
* @brief Record inserted
*/
message RowRecord {
repeated float vector_data = 1; //binary in thrift
repeated float vector_data = 1; //binary vector data
}
/**

48
cpp/src/sdk/CMakeLists.txt
View File

@ -12,43 +12,25 @@ include_directories(/usr/include)
include_directories(include)
include_directories(/usr/local/include)
if (MILVUS_WITH_THRIFT STREQUAL "ON")
aux_source_directory(thrift thrift_client_files)
include_directories(thrift)
include_directories(${CMAKE_SOURCE_DIR}/src/thrift/gen-cpp)
set(thrift_service_files
${CMAKE_SOURCE_DIR}/src/thrift/gen-cpp/MilvusService.cpp
${CMAKE_SOURCE_DIR}/src/thrift/gen-cpp/milvus_constants.cpp
${CMAKE_SOURCE_DIR}/src/thrift/gen-cpp/milvus_types.cpp
)
aux_source_directory(grpc grpc_client_files)
add_library(milvus_sdk STATIC
${interface_files}
${thrift_client_files}
${util_files}
${thrift_service_files}
)
else()
aux_source_directory(grpc grpc_client_files)
include_directories(${CMAKE_SOURCE_DIR}/src/grpc/gen-milvus)
include_directories(${CMAKE_SOURCE_DIR}/src/grpc/gen-status)
include_directories(${CMAKE_SOURCE_DIR}/src/grpc/gen-milvus)
include_directories(${CMAKE_SOURCE_DIR}/src/grpc/gen-status)
set(grpc_service_files
${CMAKE_SOURCE_DIR}/src/grpc/gen-milvus/milvus.grpc.pb.cc
${CMAKE_SOURCE_DIR}/src/grpc/gen-milvus/milvus.pb.cc
${CMAKE_SOURCE_DIR}/src/grpc/gen-status/status.grpc.pb.cc
${CMAKE_SOURCE_DIR}/src/grpc/gen-status/status.pb.cc
)
set(grpc_service_files
${CMAKE_SOURCE_DIR}/src/grpc/gen-milvus/milvus.grpc.pb.cc
${CMAKE_SOURCE_DIR}/src/grpc/gen-milvus/milvus.pb.cc
${CMAKE_SOURCE_DIR}/src/grpc/gen-status/status.grpc.pb.cc
${CMAKE_SOURCE_DIR}/src/grpc/gen-status/status.pb.cc
)
add_library(milvus_sdk STATIC
${interface_files}
${grpc_client_files}
${util_files}
${grpc_service_files}
)
endif()
add_library(milvus_sdk STATIC
${interface_files}
${grpc_client_files}
${util_files}
${grpc_service_files}
)
target_link_libraries(milvus_sdk
${third_party_libs}

6
cpp/src/sdk/examples/CMakeLists.txt
View File

@ -4,8 +4,4 @@
# Proprietary and confidential.
#-------------------------------------------------------------------------------
if (MILVUS_WITH_THRIFT STREQUAL "ON")
add_subdirectory(thriftsimple)
else()
add_subdirectory(grpcsimple)
endif()
add_subdirectory(grpcsimple)

24
cpp/src/sdk/examples/thriftsimple/CMakeLists.txt
View File

@ -1,24 +0,0 @@
#-------------------------------------------------------------------------------
# Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
# Unauthorized copying of this file, via any medium is strictly prohibited.
# Proprietary and confidential.
#-------------------------------------------------------------------------------
aux_source_directory(src src_files)
include_directories(src)
include_directories(../../include)
link_directories(${CMAKE_BINARY_DIR})
add_executable(sdk_simple
./main.cpp
${src_files}
)
target_link_libraries(sdk_simple
milvus_sdk
pthread
)
install(TARGETS sdk_simple DESTINATION bin)

68
cpp/src/sdk/examples/thriftsimple/main.cpp
View File

@ -1,68 +0,0 @@
////////////////////////////////////////////////////////////////////////////////
// Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
// Unauthorized copying of this file, via any medium is strictly prohibited.
// Proprietary and confidential.
////////////////////////////////////////////////////////////////////////////////
#include <getopt.h>
#include <libgen.h>
#include <cstring>
#include <string>
#include "src/ClientTest.h"
void print_help(const std::string &app_name);
int
main(int argc, char *argv[]) {
printf("Client start...\n");
std::string app_name = basename(argv[0]);
static struct option long_options[] = {{"server", optional_argument, 0, 's'},
{"port", optional_argument, 0, 'p'},
{"help", no_argument, 0, 'h'},
{NULL, 0, 0, 0}};
int option_index = 0;
std::string address = "127.0.0.1", port = "19530";
app_name = argv[0];
int value;
while ((value = getopt_long(argc, argv, "s:p:h", long_options, &option_index)) != -1) {
switch (value) {
case 's': {
char *address_ptr = strdup(optarg);
address = address_ptr;
free(address_ptr);
break;
}
case 'p': {
char *port_ptr = strdup(optarg);
port = port_ptr;
free(port_ptr);
break;
}
case 'h':
default:
print_help(app_name);
return EXIT_SUCCESS;
}
}
ClientTest test;
test.Test(address, port);
printf("Client stop...\n");
return 0;
}
void
print_help(const std::string &app_name) {
printf("\n Usage: %s [OPTIONS]\n\n", app_name.c_str());
printf(" Options:\n");
printf(" -s --server Server address, default 127.0.0.1\n");
printf(" -p --port Server port, default 19530\n");
printf(" -h --help Print help information\n");
printf("\n");
}

288
cpp/src/sdk/examples/thriftsimple/src/ClientTest.cpp
View File

@ -1,288 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "ClientTest.h"
#include "MilvusApi.h"
#include <iostream>
#include <time.h>
#include <chrono>
#include <unistd.h>
using namespace ::milvus;
namespace {
std::string GetTableName();
static const std::string TABLE_NAME = GetTableName();
constexpr int64_t TABLE_DIMENSION = 512;
constexpr int64_t BATCH_ROW_COUNT = 100000;
constexpr int64_t NQ = 10;
constexpr int64_t TOP_K = 10;
constexpr int64_t SEARCH_TARGET = 5000; //change this value, result is different
constexpr int64_t ADD_VECTOR_LOOP = 5;
constexpr int64_t SECONDS_EACH_HOUR = 3600;
#define BLOCK_SPLITER std::cout << "===========================================" << std::endl;
void PrintTableSchema(const TableSchema& tb_schema) {
BLOCK_SPLITER
std::cout << "Table name: " << tb_schema.table_name << std::endl;
std::cout << "Table index type: " << (int)tb_schema.index_type << std::endl;
std::cout << "Table dimension: " << tb_schema.dimension << std::endl;
std::cout << "Table store raw data: " << (tb_schema.store_raw_vector ? "true" : "false") << std::endl;
BLOCK_SPLITER
}
void PrintSearchResult(const std::vector<std::pair<int64_t, RowRecord>>& search_record_array,
const std::vector<TopKQueryResult>& topk_query_result_array) {
BLOCK_SPLITER
std::cout << "Returned result count: " << topk_query_result_array.size() << std::endl;
int32_t index = 0;
for(auto& result : topk_query_result_array) {
auto search_id = search_record_array[index].first;
index++;
std::cout << "No." << std::to_string(index) << " vector " << std::to_string(search_id)
<< " top " << std::to_string(result.query_result_arrays.size())
<< " search result:" << std::endl;
for(auto& item : result.query_result_arrays) {
std::cout << "\t" << std::to_string(item.id) << "\tdistance:" << std::to_string(item.distance);
std::cout << std::endl;
}
}
BLOCK_SPLITER
}
std::string CurrentTime() {
time_t tt;
time( &tt );
tt = tt + 8*SECONDS_EACH_HOUR;
tm* t= gmtime( &tt );
std::string str = std::to_string(t->tm_year + 1900) + "_" + std::to_string(t->tm_mon + 1)
+ "_" + std::to_string(t->tm_mday) + "_" + std::to_string(t->tm_hour)
+ "_" + std::to_string(t->tm_min) + "_" + std::to_string(t->tm_sec);
return str;
}
std::string CurrentTmDate(int64_t offset_day = 0) {
time_t tt;
time( &tt );
tt = tt + 8*SECONDS_EACH_HOUR;
tt = tt + 24*SECONDS_EACH_HOUR*offset_day;
tm* t= gmtime( &tt );
std::string str = std::to_string(t->tm_year + 1900) + "-" + std::to_string(t->tm_mon + 1)
+ "-" + std::to_string(t->tm_mday);
return str;
}
std::string GetTableName() {
static std::string s_id(CurrentTime());
return "tbl_" + s_id;
}
TableSchema BuildTableSchema() {
TableSchema tb_schema;
tb_schema.table_name = TABLE_NAME;
tb_schema.index_type = IndexType::gpu_ivfflat;
tb_schema.dimension = TABLE_DIMENSION;
tb_schema.store_raw_vector = true;
return tb_schema;
}
void BuildVectors(int64_t from, int64_t to,
std::vector<RowRecord>& vector_record_array) {
if(to <= from){
return;
}
vector_record_array.clear();
for (int64_t k = from; k < to; k++) {
RowRecord record;
record.data.resize(TABLE_DIMENSION);
for(int64_t i = 0; i < TABLE_DIMENSION; i++) {
record.data[i] = (float)(k%(i+1));
}
vector_record_array.emplace_back(record);
}
}
void Sleep(int seconds) {
std::cout << "Waiting " << seconds << " seconds ..." << std::endl;
sleep(seconds);
}
class TimeRecorder {
public:
TimeRecorder(const std::string& title)
: title_(title) {
start_ = std::chrono::system_clock::now();
}
~TimeRecorder() {
std::chrono::system_clock::time_point end = std::chrono::system_clock::now();
long span = (std::chrono::duration_cast<std::chrono::milliseconds> (end - start_)).count();
std::cout << title_ << " totally cost: " << span << " ms" << std::endl;
}
private:
std::string title_;
std::chrono::system_clock::time_point start_;
};
void CheckResult(const std::vector<std::pair<int64_t, RowRecord>>& search_record_array,
const std::vector<TopKQueryResult>& topk_query_result_array) {
BLOCK_SPLITER
int64_t index = 0;
for(auto& result : topk_query_result_array) {
auto result_id = result.query_result_arrays[0].id;
auto search_id = search_record_array[index++].first;
if(result_id != search_id) {
std::cout << "The top 1 result is wrong: " << result_id
<< " vs. " << search_id << std::endl;
} else {
std::cout << "No." << index-1 << " Check result successfully" << std::endl;
}
}
BLOCK_SPLITER
}
void DoSearch(std::shared_ptr<Connection> conn,
const std::vector<std::pair<int64_t, RowRecord>>& search_record_array,
const std::string& phase_name) {
std::vector<Range> query_range_array;
Range rg;
rg.start_value = CurrentTmDate();
rg.end_value = CurrentTmDate(1);
query_range_array.emplace_back(rg);
std::vector<RowRecord> record_array;
for(auto& pair : search_record_array) {
record_array.push_back(pair.second);
}
std::vector<TopKQueryResult> topk_query_result_array;
{
TimeRecorder rc(phase_name);
Status stat = conn->Search(TABLE_NAME, record_array, query_range_array, TOP_K, topk_query_result_array);
std::cout << "SearchVector function call status: " << stat.ToString() << std::endl;
}
PrintSearchResult(search_record_array, topk_query_result_array);
CheckResult(search_record_array, topk_query_result_array);
}
}
void
ClientTest::Test(const std::string& address, const std::string& port) {
std::shared_ptr<Connection> conn = Connection::Create();
{//connect server
ConnectParam param = {address, port};
Status stat = conn->Connect(param);
std::cout << "Connect function call status: " << stat.ToString() << std::endl;
}
{//server version
std::string version = conn->ServerVersion();
std::cout << "Server version: " << version << std::endl;
}
{//sdk version
std::string version = conn->ClientVersion();
std::cout << "SDK version: " << version << std::endl;
}
{
std::vector<std::string> tables;
Status stat = conn->ShowTables(tables);
std::cout << "ShowTables function call status: " << stat.ToString() << std::endl;
std::cout << "All tables: " << std::endl;
for(auto& table : tables) {
int64_t row_count = 0;
stat = conn->CountTable(table, row_count);
std::cout << "\t" << table << "(" << row_count << " rows)" << std::endl;
}
}
{//create table
TableSchema tb_schema = BuildTableSchema();
Status stat = conn->CreateTable(tb_schema);
std::cout << "CreateTable function call status: " << stat.ToString() << std::endl;
PrintTableSchema(tb_schema);
bool has_table = conn->HasTable(tb_schema.table_name);
if(has_table) {
std::cout << "Table is created" << std::endl;
}
}
{//describe table
TableSchema tb_schema;
Status stat = conn->DescribeTable(TABLE_NAME, tb_schema);
std::cout << "DescribeTable function call status: " << stat.ToString() << std::endl;
PrintTableSchema(tb_schema);
}
//add vectors
std::vector<std::pair<int64_t, RowRecord>> search_record_array;
for (int i = 0; i < ADD_VECTOR_LOOP; i++) {
TimeRecorder recorder("Add vector No." + std::to_string(i));
std::vector<RowRecord> record_array;
int64_t begin_index = i * BATCH_ROW_COUNT;
BuildVectors(begin_index, begin_index + BATCH_ROW_COUNT, record_array);
std::vector<int64_t> record_ids;
Status stat = conn->Insert(TABLE_NAME, record_array, record_ids);
std::cout << "AddVector function call status: " << stat.ToString() << std::endl;
std::cout << "Returned id array count: " << record_ids.size() << std::endl;
if(i == 0) {
for(int64_t k = SEARCH_TARGET; k < SEARCH_TARGET + NQ; k++) {
search_record_array.push_back(
std::make_pair(record_ids[k], record_array[k]));
}
}
}
{//search vectors without index
Sleep(2);
DoSearch(conn, search_record_array, "Search without index");
}
{//wait unit build index finish
TimeRecorder recorder("Build index");
std::cout << "Wait until build all index done" << std::endl;
IndexParam index_param;
index_param.table_name = TABLE_NAME;
Status stat = conn->CreateIndex(index_param);
std::cout << "BuildIndex function call status: " << stat.ToString() << std::endl;
}
{//search vectors after build index finish
DoSearch(conn, search_record_array, "Search after build index finish");
}
{//delete table
Status stat = conn->DropTable(TABLE_NAME);
std::cout << "DeleteTable function call status: " << stat.ToString() << std::endl;
}
{//server status
std::string status = conn->ServerStatus();
std::cout << "Server status before disconnect: " << status << std::endl;
}
Connection::Destroy(conn);
{//server status
std::string status = conn->ServerStatus();
std::cout << "Server status after disconnect: " << status << std::endl;
}
}

13
cpp/src/sdk/examples/thriftsimple/src/ClientTest.h
View File

@ -1,13 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include <string>
class ClientTest {
public:
void Test(const std::string& address, const std::string& port);
};

4
cpp/src/sdk/interface/ConnectionImpl.h
View File

@ -6,11 +6,7 @@
#pragma once
#include "MilvusApi.h"
#ifdef MILVUS_ENABLE_THRIFT
#include "src/sdk/thrift/ClientProxy.h"
#else
#include "src/sdk/grpc/ClientProxy.h"
#endif
namespace milvus {

346
cpp/src/sdk/thrift/ClientProxy.cpp
View File

@ -1,346 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "ClientProxy.h"
namespace milvus {
std::shared_ptr<ThriftClient>&
ClientProxy::ClientPtr() const {
if(client_ptr == nullptr) {
client_ptr = std::make_shared<ThriftClient>();
}
return client_ptr;
}
bool ClientProxy::IsConnected() const {
return (client_ptr != nullptr && connected_);
}
Status
ClientProxy::Connect(const ConnectParam &param) {
Disconnect();
int32_t port = atoi(param.port.c_str());
Status status = ClientPtr()->Connect(param.ip_address, port, THRIFT_PROTOCOL_BINARY);
if(status.ok()) {
connected_ = true;
}
return status;
}
Status
ClientProxy::Connect(const std::string &uri) {
Disconnect();
size_t index = uri.find_first_of(":", 0);
if((index == std::string::npos)) {
return Status::Invalid("Invalid uri");
}
ConnectParam param;
param.ip_address = uri.substr(0, index);
param.port = uri.substr(index + 1);
return Connect(param);
}
Status
ClientProxy::Connected() const {
if(!IsConnected()) {
return Status(StatusCode::NotConnected, "not connected to server");
}
try {
std::string info;
ClientPtr()->interface()->Ping(info, "");
} catch ( std::exception& ex) {
return Status(StatusCode::NotConnected, "connection lost: " + std::string(ex.what()));
}
return Status::OK();
}
Status
ClientProxy::Disconnect() {
if(!IsConnected()) {
return Status(StatusCode::NotConnected, "not connected to server");
}
connected_ = false;
return ClientPtr()->Disconnect();
}
std::string
ClientProxy::ClientVersion() const {
return "";
}
Status
ClientProxy::CreateTable(const TableSchema &param) {
if(!IsConnected()) {
return Status(StatusCode::NotConnected, "not connected to server");
}
try {
thrift::TableSchema schema;
schema.__set_table_name(param.table_name);
schema.__set_index_type((int)param.index_type);
schema.__set_dimension(param.dimension);
schema.__set_store_raw_vector(param.store_raw_vector);
ClientPtr()->interface()->CreateTable(schema);
} catch ( std::exception& ex) {
return Status(StatusCode::UnknownError, "failed to create table: " + std::string(ex.what()));
}
return Status::OK();
}
bool
ClientProxy::HasTable(const std::string &table_name) {
if(!IsConnected()) {
return false;
}
return ClientPtr()->interface()->HasTable(table_name);
}
Status
ClientProxy::DropTable(const std::string &table_name) {
if(!IsConnected()) {
return Status(StatusCode::NotConnected, "not connected to server");
}
try {
ClientPtr()->interface()->DeleteTable(table_name);
} catch ( std::exception& ex) {
return Status(StatusCode::UnknownError, "failed to delete table: " + std::string(ex.what()));
}
return Status::OK();
}
Status
ClientProxy::CreateIndex(const IndexParam &index_param) {
if(!IsConnected()) {
return Status(StatusCode::NotConnected, "not connected to server");
}
try {
ClientPtr()->interface()->BuildIndex(index_param.table_name);
} catch ( std::exception& ex) {
return Status(StatusCode::UnknownError, "failed to build index: " + std::string(ex.what()));
}
return Status::OK();
}
Status
ClientProxy::Insert(const std::string &table_name,
const std::vector<RowRecord> &record_array,
std::vector<int64_t> &id_array) {
if(!IsConnected()) {
return Status(StatusCode::NotConnected, "not connected to server");
}
try {
std::vector<thrift::RowRecord> thrift_records;
for(auto& record : record_array) {
thrift::RowRecord thrift_record;
thrift_record.vector_data.resize(record.data.size() * sizeof(double));
double *dbl = (double *) (const_cast<char *>(thrift_record.vector_data.data()));
for (size_t i = 0; i < record.data.size(); i++) {
dbl[i] = (double) (record.data[i]);
}
thrift_records.emplace_back(thrift_record);
}
ClientPtr()->interface()->AddVector(id_array, table_name, thrift_records);
} catch ( std::exception& ex) {
return Status(StatusCode::UnknownError, "failed to add vector: " + std::string(ex.what()));
}
return Status::OK();
}
Status
ClientProxy::Search(const std::string &table_name,
const std::vector<RowRecord> &query_record_array,
const std::vector<Range> &query_range_array,
int64_t topk,
std::vector<TopKQueryResult> &topk_query_result_array) {
if(!IsConnected()) {
return Status(StatusCode::NotConnected, "not connected to server");
}
try {
//step 1: convert vectors data
std::vector<thrift::RowRecord> thrift_records;
for(auto& record : query_record_array) {
thrift::RowRecord thrift_record;
thrift_record.vector_data.resize(record.data.size() * sizeof(double));
auto dbl = (double *) (const_cast<char *>(thrift_record.vector_data.data()));
for (size_t i = 0; i < record.data.size(); i++) {
dbl[i] = (double) (record.data[i]);
}
thrift_records.emplace_back(thrift_record);
}
//step 2: convert range array
std::vector<thrift::Range> thrift_ranges;
for(auto& range : query_range_array) {
thrift::Range thrift_range;
thrift_range.__set_start_value(range.start_value);
thrift_range.__set_end_value(range.end_value);
thrift_ranges.emplace_back(thrift_range);
}
//step 3: search vectors
std::vector<thrift::TopKQueryBinResult> result_array;
ClientPtr()->interface()->SearchVector2(result_array, table_name, thrift_records, thrift_ranges, topk);
//step 4: convert result array
for(auto& thrift_topk_result : result_array) {
TopKQueryResult result;
size_t id_count = thrift_topk_result.id_array.size()/sizeof(int64_t);
size_t dist_count = thrift_topk_result.distance_array.size()/ sizeof(double);
if(id_count != dist_count) {
return Status(StatusCode::UnknownError, "illegal result");
}
auto id_ptr = (int64_t*)thrift_topk_result.id_array.data();
auto dist_ptr = (double*)thrift_topk_result.distance_array.data();
for(size_t i = 0; i < id_count; i++) {
QueryResult query_result;
query_result.id = id_ptr[i];
query_result.distance = dist_ptr[i];
result.query_result_arrays.emplace_back(query_result);
}
topk_query_result_array.emplace_back(result);
}
} catch ( std::exception& ex) {
return Status(StatusCode::UnknownError, "failed to search vectors: " + std::string(ex.what()));
}
return Status::OK();
}
Status
ClientProxy::DescribeTable(const std::string &table_name, TableSchema &table_schema) {
if(!IsConnected()) {
return Status(StatusCode::NotConnected, "not connected to server");
}
try {
thrift::TableSchema thrift_schema;
ClientPtr()->interface()->DescribeTable(thrift_schema, table_name);
table_schema.table_name = thrift_schema.table_name;
table_schema.index_type = (IndexType)thrift_schema.index_type;
table_schema.dimension = thrift_schema.dimension;
table_schema.store_raw_vector = thrift_schema.store_raw_vector;
} catch ( std::exception& ex) {
return Status(StatusCode::UnknownError, "failed to describe table: " + std::string(ex.what()));
}
return Status::OK();
}
Status
ClientProxy::CountTable(const std::string &table_name, int64_t &row_count) {
if(!IsConnected()) {
return Status(StatusCode::NotConnected, "not connected to server");
}
try {
row_count = ClientPtr()->interface()->GetTableRowCount(table_name);
} catch ( std::exception& ex) {
return Status(StatusCode::UnknownError, "failed to show tables: " + std::string(ex.what()));
}
return Status::OK();
}
Status
ClientProxy::ShowTables(std::vector<std::string> &table_array) {
if(!IsConnected()) {
return Status(StatusCode::NotConnected, "not connected to server");
}
try {
ClientPtr()->interface()->ShowTables(table_array);
} catch ( std::exception& ex) {
return Status(StatusCode::UnknownError, "failed to show tables: " + std::string(ex.what()));
}
return Status::OK();
}
std::string
ClientProxy::ServerVersion() const {
if(!IsConnected()) {
return "";
}
try {
std::string version;
ClientPtr()->interface()->Ping(version, "version");
return version;
} catch ( std::exception& ex) {
return "";
}
}
std::string
ClientProxy::ServerStatus() const {
if(!IsConnected()) {
return "not connected to server";
}
try {
std::string dummy;
ClientPtr()->interface()->Ping(dummy, "");
return "server alive";
} catch ( std::exception& ex) {
return "connection lost";
}
}
Status ClientProxy::DeleteByRange(Range &range, const std::string &table_name) {
return Status::OK();
}
Status ClientProxy::PreloadTable(const std::string &table_name) const {
return Status::OK();
}
Status ClientProxy::DescribeIndex(const std::string &table_name, IndexParam &index_param) const {
index_param.table_name = table_name;
return index_param;
}
Status ClientProxy::DropIndex(const std::string &table_name) const {
return Status::OK();
}
}

72
cpp/src/sdk/thrift/ClientProxy.h
View File

@ -1,72 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include "MilvusApi.h"
#include "ThriftClient.h"
namespace milvus {
class ClientProxy : public Connection {
public:
// Implementations of the Connection interface
virtual Status Connect(const ConnectParam &param) override;
virtual Status Connect(const std::string &uri) override;
virtual Status Connected() const override;
virtual Status Disconnect() override;
virtual Status CreateTable(const TableSchema &param) override;
virtual bool HasTable(const std::string &table_name) override;
virtual Status DropTable(const std::string &table_name) override;
virtual Status CreateIndex(const IndexParam &index_param) override;
virtual Status Insert(const std::string &table_name,
const std::vector<RowRecord> &record_array,
std::vector<int64_t> &id_array) override;
virtual Status Search(const std::string &table_name,
const std::vector<RowRecord> &query_record_array,
const std::vector<Range> &query_range_array,
int64_t topk,
std::vector<TopKQueryResult> &topk_query_result_array) override;
virtual Status DescribeTable(const std::string &table_name, TableSchema &table_schema) override;
virtual Status CountTable(const std::string &table_name, int64_t &row_count) override;
virtual Status ShowTables(std::vector<std::string> &table_array) override;
virtual std::string ClientVersion() const override;
virtual std::string ServerVersion() const override;
virtual std::string ServerStatus() const override;
virtual Status DeleteByRange(Range &range, const std::string &table_name) override;
virtual Status PreloadTable(const std::string &table_name) const override;
virtual Status DescribeIndex(const std::string &table_name, IndexParam &index_param) const override;
virtual Status DropIndex(const std::string &table_name) const override;
private:
std::shared_ptr<ThriftClient>& ClientPtr() const;
bool IsConnected() const;
private:
mutable std::shared_ptr<ThriftClient> client_ptr;
bool connected_ = false;
};
}

102
cpp/src/sdk/thrift/ThriftClient.cpp
View File

@ -1,102 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "ThriftClient.h"
#include "milvus_types.h"
#include "milvus_constants.h"
#include <exception>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/protocol/TJSONProtocol.h>
#include <thrift/protocol/TCompactProtocol.h>
#include <thrift/protocol/TDebugProtocol.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/server/TThreadPoolServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TBufferTransports.h>
#include <thrift/concurrency/PosixThreadFactory.h>
namespace milvus {
using namespace ::apache::thrift;
using namespace ::apache::thrift::protocol;
using namespace ::apache::thrift::transport;
using namespace ::apache::thrift::concurrency;
ThriftClient::ThriftClient() {
}
ThriftClient::~ThriftClient() {
}
ServiceClientPtr
ThriftClient::interface() {
if(client_ == nullptr) {
throw std::exception();
}
return client_;
}
Status
ThriftClient::Connect(const std::string& address, int32_t port, const std::string& protocol) {
try {
stdcxx::shared_ptr<TSocket> socket_ptr(new transport::TSocket(address, port));
stdcxx::shared_ptr<TTransport> transport_ptr(new TBufferedTransport(socket_ptr));
stdcxx::shared_ptr<TProtocol> protocol_ptr;
if(protocol == THRIFT_PROTOCOL_BINARY) {
protocol_ptr.reset(new TBinaryProtocol(transport_ptr));
} else if(protocol == THRIFT_PROTOCOL_JSON) {
protocol_ptr.reset(new TJSONProtocol(transport_ptr));
} else if(protocol == THRIFT_PROTOCOL_COMPACT) {
protocol_ptr.reset(new TCompactProtocol(transport_ptr));
} else {
//CLIENT_LOG_ERROR << "Service protocol: " << protocol << " is not supported currently";
return Status(StatusCode::InvalidAgument, "unsupported protocol");
}
transport_ptr->open();
client_ = std::make_shared<thrift::MilvusServiceClient>(protocol_ptr);
} catch ( std::exception& ex) {
//CLIENT_LOG_ERROR << "connect encounter exception: " << ex.what();
return Status(StatusCode::NotConnected, "failed to connect server" + std::string(ex.what()));
}
return Status::OK();
}
Status
ThriftClient::Disconnect() {
try {
if(client_ != nullptr) {
auto protocol = client_->getInputProtocol();
if(protocol != nullptr) {
auto transport = protocol->getTransport();
if(transport != nullptr) {
transport->close();
}
}
}
} catch ( std::exception& ex) {
//CLIENT_LOG_ERROR << "disconnect encounter exception: " << ex.what();
return Status(StatusCode::UnknownError, "failed to disconnect: " + std::string(ex.what()));
}
return Status::OK();
}
/////////////////////////////////////////////////////////////////////////////////////////////////////////
ThriftClientSession::ThriftClientSession(const std::string& address, int32_t port, const std::string& protocol) {
Connect(address, port, protocol);
}
ThriftClientSession::~ThriftClientSession() {
Disconnect();
}
}

42
cpp/src/sdk/thrift/ThriftClient.h
View File

@ -1,42 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include "MilvusService.h"
#include "Status.h"
#include <memory>
namespace milvus {
using ServiceClientPtr = std::shared_ptr<::milvus::thrift::MilvusServiceClient>;
static const char* THRIFT_PROTOCOL_JSON = "json";
static const char* THRIFT_PROTOCOL_BINARY = "binary";
static const char* THRIFT_PROTOCOL_COMPACT = "compact";
class ThriftClient {
public:
ThriftClient();
virtual ~ThriftClient();
ServiceClientPtr interface();
Status Connect(const std::string& address, int32_t port, const std::string& protocol);
Status Disconnect();
private:
ServiceClientPtr client_;
};
class ThriftClientSession : public ThriftClient {
public:
ThriftClientSession(const std::string& address, int32_t port, const std::string& protocol);
~ThriftClientSession();
};
}

14
cpp/src/server/Server.cpp
View File

@ -5,13 +5,7 @@
////////////////////////////////////////////////////////////////////////////////
#include <thread>
#include "Server.h"
//#include "ServerConfig.h"
#ifdef MILVUS_ENABLE_THRIFT
#include "server/thrift_impl/MilvusServer.h"
#else
#include "server/grpc_impl/GrpcMilvusServer.h"
#endif
#include "utils/Log.h"
#include "utils/SignalUtil.h"
#include "utils/TimeRecorder.h"
@ -227,20 +221,12 @@ Server::LoadConfig() {
void
Server::StartService() {
#ifdef MILVUS_ENABLE_THRIFT
MilvusServer::StartService();
#else
grpc::GrpcMilvusServer::StartService();
#endif
}
void
Server::StopService() {
#ifdef MILVUS_ENABLE_THRIFT
MilvusServer::StopService();
#else
grpc::GrpcMilvusServer::StopService();
#endif
}
}

106
cpp/src/server/thrift_impl/MilvusServer.cpp
View File

@ -1,106 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "MilvusServer.h"
#include "RequestHandler.h"
#include "src/server/ServerConfig.h"
#include "ThreadPoolServer.h"
#include "src/server/DBWrapper.h"
#include "utils/Log.h"
#include "milvus_types.h"
#include "milvus_constants.h"
#include "faiss/utils.h"
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/protocol/TJSONProtocol.h>
#include <thrift/protocol/TDebugProtocol.h>
#include <thrift/protocol/TCompactProtocol.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/server/TThreadPoolServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TBufferTransports.h>
#include <thrift/concurrency/PosixThreadFactory.h>
#include <thread>
#include <iostream>
//extern int distance_compute_blas_threshold;
namespace zilliz {
namespace milvus {
namespace server {
using namespace ::milvus::thrift;
using namespace ::apache::thrift;
using namespace ::apache::thrift::protocol;
using namespace ::apache::thrift::transport;
using namespace ::apache::thrift::server;
using namespace ::apache::thrift::concurrency;
static stdcxx::shared_ptr<TServer> s_server;
void
MilvusServer::StartService() {
if(s_server != nullptr){
StopService();
}
ServerConfig &config = ServerConfig::GetInstance();
ConfigNode server_config = config.GetConfig(CONFIG_SERVER);
ConfigNode engine_config = config.GetConfig(CONFIG_ENGINE);
std::string address = server_config.GetValue(CONFIG_SERVER_ADDRESS, "127.0.0.1");
int32_t port = server_config.GetInt32Value(CONFIG_SERVER_PORT, 19530);
std::string protocol = server_config.GetValue(CONFIG_SERVER_PROTOCOL, "binary");
faiss::distance_compute_blas_threshold = engine_config.GetInt32Value(CONFIG_DCBT,20);
// std::cout<<"distance_compute_blas_threshold = "<< faiss::distance_compute_blas_threshold << std::endl;
try {
DBWrapper::DB();//initialize db
stdcxx::shared_ptr<RequestHandler> handler(new RequestHandler());
stdcxx::shared_ptr<TProcessor> processor(new MilvusServiceProcessor(handler));
stdcxx::shared_ptr<TServerTransport> server_transport(new TServerSocket(address, port));
stdcxx::shared_ptr<TTransportFactory> transport_factory(new TBufferedTransportFactory());
stdcxx::shared_ptr<TProtocolFactory> protocol_factory;
if (protocol == "binary") {
protocol_factory.reset(new TBinaryProtocolFactory());
} else if (protocol == "json") {
protocol_factory.reset(new TJSONProtocolFactory());
} else if (protocol == "compact") {
protocol_factory.reset(new TCompactProtocolFactory());
} else {
// SERVER_LOG_INFO << "Service protocol: " << protocol << " is not supported currently";
return;
}
stdcxx::shared_ptr<ThreadManager> threadManager(ThreadManager::newSimpleThreadManager(16));
stdcxx::shared_ptr<PosixThreadFactory> threadFactory(new PosixThreadFactory());
threadManager->threadFactory(threadFactory);
threadManager->start();
s_server.reset(new ThreadPoolServer(processor,
server_transport,
transport_factory,
protocol_factory,
threadManager));
s_server->serve();
} catch (apache::thrift::TException& ex) {
std::cout << "ERROR! " << ex.what() << std::endl;
kill(0, SIGUSR1);
}
}
void
MilvusServer::StopService() {
if(s_server != nullptr) {
s_server->stop();
}
}
}
}
}

23
cpp/src/server/thrift_impl/MilvusServer.h
View File

@ -1,23 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include <cstdint>
#include <string>
namespace zilliz {
namespace milvus {
namespace server {
class MilvusServer {
public:
static void StartService();
static void StopService();
};
}
}
}

124
cpp/src/server/thrift_impl/RequestHandler.cpp
View File

@ -1,124 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "RequestHandler.h"
#include "RequestTask.h"
#include "utils/TimeRecorder.h"
namespace zilliz {
namespace milvus {
namespace server {
using namespace ::milvus;
RequestHandler::RequestHandler() {
}
void
RequestHandler::CreateTable(const thrift::TableSchema &param) {
BaseTaskPtr task_ptr = CreateTableTask::Create(param);
RequestScheduler::ExecTask(task_ptr);
}
bool
RequestHandler::HasTable(const std::string &table_name) {
bool has_table = false;
BaseTaskPtr task_ptr = HasTableTask::Create(table_name, has_table);
RequestScheduler::ExecTask(task_ptr);
return has_table;
}
void
RequestHandler::DeleteTable(const std::string &table_name) {
BaseTaskPtr task_ptr = DeleteTableTask::Create(table_name);
RequestScheduler::ExecTask(task_ptr);
}
void
RequestHandler::BuildIndex(const std::string &table_name) {
BaseTaskPtr task_ptr = BuildIndexTask::Create(table_name);
RequestScheduler::ExecTask(task_ptr);
}
void
RequestHandler::AddVector(std::vector<int64_t> &_return,
const std::string &table_name,
const std::vector<thrift::RowRecord> &record_array) {
BaseTaskPtr task_ptr = AddVectorTask::Create(table_name, record_array, _return);
RequestScheduler::ExecTask(task_ptr);
}
void
RequestHandler::SearchVector(std::vector<thrift::TopKQueryResult> &_return,
const std::string &table_name,
const std::vector<thrift::RowRecord> &query_record_array,
const std::vector<thrift::Range> &query_range_array,
const int64_t topk) {
// SERVER_LOG_DEBUG << "Entering RequestHandler::SearchVector";
BaseTaskPtr task_ptr = SearchVectorTask1::Create(table_name, std::vector<std::string>(), query_record_array,
query_range_array, topk, _return);
RequestScheduler::ExecTask(task_ptr);
}
void
RequestHandler::SearchVector2(std::vector<thrift::TopKQueryBinResult> & _return,
const std::string& table_name,
const std::vector<thrift::RowRecord> & query_record_array,
const std::vector<thrift::Range> & query_range_array,
const int64_t topk) {
BaseTaskPtr task_ptr = SearchVectorTask2::Create(table_name, std::vector<std::string>(), query_record_array,
query_range_array, topk, _return);
RequestScheduler::ExecTask(task_ptr);
}
void
RequestHandler::SearchVectorInFiles(std::vector<::milvus::thrift::TopKQueryResult> &_return,
const std::string& table_name,
const std::vector<std::string> &file_id_array,
const std::vector<::milvus::thrift::RowRecord> &query_record_array,
const std::vector<::milvus::thrift::Range> &query_range_array,
const int64_t topk) {
// SERVER_LOG_DEBUG << "Entering RequestHandler::SearchVectorInFiles. file_id_array size = " << std::to_string(file_id_array.size());
BaseTaskPtr task_ptr = SearchVectorTask1::Create(table_name, file_id_array, query_record_array,
query_range_array, topk, _return);
RequestScheduler::ExecTask(task_ptr);
}
void
RequestHandler::DescribeTable(thrift::TableSchema &_return, const std::string &table_name) {
BaseTaskPtr task_ptr = DescribeTableTask::Create(table_name, _return);
RequestScheduler::ExecTask(task_ptr);
}
int64_t
RequestHandler::GetTableRowCount(const std::string& table_name) {
int64_t row_count = 0;
{
BaseTaskPtr task_ptr = GetTableRowCountTask::Create(table_name, row_count);
RequestScheduler::ExecTask(task_ptr);
task_ptr->WaitToFinish();
}
return row_count;
}
void
RequestHandler::ShowTables(std::vector<std::string> &_return) {
BaseTaskPtr task_ptr = ShowTablesTask::Create(_return);
RequestScheduler::ExecTask(task_ptr);
}
void
RequestHandler::Ping(std::string& _return, const std::string& cmd) {
BaseTaskPtr task_ptr = PingTask::Create(cmd, _return);
RequestScheduler::ExecTask(task_ptr);
}
}
}
}

206
cpp/src/server/thrift_impl/RequestHandler.h
View File

@ -1,206 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include <cstdint>
#include <string>
#include "MilvusService.h"
namespace zilliz {
namespace milvus {
namespace server {
class RequestHandler : virtual public ::milvus::thrift::MilvusServiceIf {
public:
RequestHandler();
/**
* @brief Create table method
*
* This method is used to create table
*
* @param param, use to provide table information to be created.
*
*
* @param param
*/
void CreateTable(const ::milvus::thrift::TableSchema &param);
/**
* @brief Test table existence method
*
* This method is used to test table existence.
*
* @param table_name, table name is going to be tested.
*
*
* @param table_name
*/
bool HasTable(const std::string &table_name);
/**
* @brief Delete table method
*
* This method is used to delete table.
*
* @param table_name, table name is going to be deleted.
*
*
* @param table_name
*/
void DeleteTable(const std::string& table_name);
/**
* @brief build index by table method
*
* This method is used to build index by table in sync.
*
* @param table_name, table name is going to be built index.
*
*
* @param table_name
*/
void BuildIndex(const std::string &table_name);
/**
* @brief Add vector array to table
*
* This method is used to add vector array to table.
*
* @param table_name, table_name is inserted.
* @param record_array, vector array is inserted.
*
* @return vector id array
*
* @param table_name
* @param record_array
*/
void AddVector(std::vector<int64_t> & _return,
const std::string& table_name,
const std::vector<::milvus::thrift::RowRecord> & record_array);
/**
* @brief Query vector
*
* This method is used to query vector in table.
*
* @param table_name, table_name is queried.
* @param query_record_array, all vector are going to be queried.
* @param query_range_array, optional ranges for conditional search. If not specified, search whole table
* @param topk, how many similarity vectors will be searched.
*
* @return query result array.
*
* @param table_name
* @param query_record_array
* @param query_range_array
* @param topk
*/
void SearchVector(std::vector<::milvus::thrift::TopKQueryResult> & _return,
const std::string& table_name,
const std::vector<::milvus::thrift::RowRecord> & query_record_array,
const std::vector<::milvus::thrift::Range> & query_range_array,
const int64_t topk);
/**
* @brief Query vector
*
* This method is used to query vector in table.
*
* @param table_name, table_name is queried.
* @param query_record_array, all vector are going to be queried.
* @param query_range_array, optional ranges for conditional search. If not specified, search whole table
* @param topk, how many similarity vectors will be searched.
*
* @return query binary result array.
*
* @param table_name
* @param query_record_array
* @param query_range_array
* @param topk
*/
void SearchVector2(std::vector<::milvus::thrift::TopKQueryBinResult> & _return,
const std::string& table_name,
const std::vector<::milvus::thrift::RowRecord> & query_record_array,
const std::vector<::milvus::thrift::Range> & query_range_array,
const int64_t topk);
/**
* @brief Internal use query interface
*
* This method is used to query vector in specified files.
*
* @param file_id_array, specified files id array, queried.
* @param query_record_array, all vector are going to be queried.
* @param query_range_array, optional ranges for conditional search. If not specified, search whole table
* @param topk, how many similarity vectors will be searched.
*
* @return query result array.
*
* @param file_id_array
* @param query_record_array
* @param query_range_array
* @param topk
*/
virtual void SearchVectorInFiles(std::vector<::milvus::thrift::TopKQueryResult> & _return,
const std::string& table_name,
const std::vector<std::string> & file_id_array,
const std::vector<::milvus::thrift::RowRecord> & query_record_array,
const std::vector<::milvus::thrift::Range> & query_range_array,
const int64_t topk);
/**
* @brief Get table schema
*
* This method is used to get table schema.
*
* @param table_name, target table name.
*
* @return table schema
*
* @param table_name
*/
void DescribeTable(::milvus::thrift::TableSchema& _return, const std::string& table_name);
/**
* @brief Get table row count
*
* This method is used to get table row count.
*
* @param table_name, target table name.
*
* @return table row count
*
* @param table_name
*/
int64_t GetTableRowCount(const std::string& table_name);
/**
* @brief List all tables in database
*
* This method is used to list all tables.
*
*
* @return table names.
*/
void ShowTables(std::vector<std::string> & _return);
/**
* @brief Give the server status
*
* This method is used to give the server status.
*
* @return Server status.
*
* @param cmd
*/
void Ping(std::string& _return, const std::string& cmd);
};
}
}
}

218
cpp/src/server/thrift_impl/RequestScheduler.cpp
View File

@ -1,218 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "RequestScheduler.h"
#include "utils/Log.h"
#include "milvus_types.h"
#include "milvus_constants.h"
namespace zilliz {
namespace milvus {
namespace server {
using namespace ::milvus;
namespace {
const std::map<ServerError, thrift::ErrorCode::type> &ErrorMap() {
static const std::map<ServerError, thrift::ErrorCode::type> code_map = {
{SERVER_UNEXPECTED_ERROR, thrift::ErrorCode::UNEXPECTED_ERROR},
{SERVER_UNSUPPORTED_ERROR, thrift::ErrorCode::UNEXPECTED_ERROR},
{SERVER_NULL_POINTER, thrift::ErrorCode::UNEXPECTED_ERROR},
{SERVER_INVALID_ARGUMENT, thrift::ErrorCode::ILLEGAL_ARGUMENT},
{SERVER_FILE_NOT_FOUND, thrift::ErrorCode::FILE_NOT_FOUND},
{SERVER_NOT_IMPLEMENT, thrift::ErrorCode::UNEXPECTED_ERROR},
{SERVER_BLOCKING_QUEUE_EMPTY, thrift::ErrorCode::UNEXPECTED_ERROR},
{SERVER_CANNOT_CREATE_FOLDER, thrift::ErrorCode::CANNOT_CREATE_FOLDER},
{SERVER_CANNOT_CREATE_FILE, thrift::ErrorCode::CANNOT_CREATE_FILE},
{SERVER_CANNOT_DELETE_FOLDER, thrift::ErrorCode::CANNOT_DELETE_FOLDER},
{SERVER_CANNOT_DELETE_FILE, thrift::ErrorCode::CANNOT_DELETE_FILE},
{SERVER_TABLE_NOT_EXIST, thrift::ErrorCode::TABLE_NOT_EXISTS},
{SERVER_INVALID_TABLE_NAME, thrift::ErrorCode::ILLEGAL_TABLE_NAME},
{SERVER_INVALID_TABLE_DIMENSION, thrift::ErrorCode::ILLEGAL_DIMENSION},
{SERVER_INVALID_TIME_RANGE, thrift::ErrorCode::ILLEGAL_RANGE},
{SERVER_INVALID_VECTOR_DIMENSION, thrift::ErrorCode::ILLEGAL_DIMENSION},
{SERVER_INVALID_INDEX_TYPE, thrift::ErrorCode::ILLEGAL_INDEX_TYPE},
{SERVER_INVALID_ROWRECORD, thrift::ErrorCode::ILLEGAL_ROWRECORD},
{SERVER_INVALID_ROWRECORD_ARRAY, thrift::ErrorCode::ILLEGAL_ROWRECORD},
{SERVER_INVALID_TOPK, thrift::ErrorCode::ILLEGAL_TOPK},
{SERVER_ILLEGAL_VECTOR_ID, thrift::ErrorCode::ILLEGAL_VECTOR_ID},
{SERVER_ILLEGAL_SEARCH_RESULT, thrift::ErrorCode::ILLEGAL_SEARCH_RESULT},
{SERVER_CACHE_ERROR, thrift::ErrorCode::CACHE_FAILED},
{DB_META_TRANSACTION_FAILED, thrift::ErrorCode::META_FAILED},
{SERVER_BUILD_INDEX_ERROR, thrift::ErrorCode::BUILD_INDEX_ERROR},
};
return code_map;
}
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
BaseTask::BaseTask(const std::string& task_group, bool async)
: task_group_(task_group),
async_(async),
done_(false),
error_code_(SERVER_SUCCESS) {
}
BaseTask::~BaseTask() {
WaitToFinish();
}
ServerError BaseTask::Execute() {
error_code_ = OnExecute();
done_ = true;
finish_cond_.notify_all();
return error_code_;
}
ServerError BaseTask::SetError(ServerError error_code, const std::string& error_msg) {
error_code_ = error_code;
error_msg_ = error_msg;
SERVER_LOG_ERROR << error_msg_;
return error_code_;
}
ServerError BaseTask::WaitToFinish() {
std::unique_lock <std::mutex> lock(finish_mtx_);
finish_cond_.wait(lock, [this] { return done_; });
return error_code_;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
RequestScheduler::RequestScheduler()
: stopped_(false) {
Start();
}
RequestScheduler::~RequestScheduler() {
Stop();
}
void RequestScheduler::ExecTask(BaseTaskPtr& task_ptr) {
if(task_ptr == nullptr) {
return;
}
RequestScheduler& scheduler = RequestScheduler::GetInstance();
scheduler.ExecuteTask(task_ptr);
if(!task_ptr->IsAsync()) {
task_ptr->WaitToFinish();
ServerError err = task_ptr->ErrorCode();
if (err != SERVER_SUCCESS) {
thrift::Exception ex;
ex.__set_code(ErrorMap().at(err));
std::string msg = task_ptr->ErrorMsg();
if(msg.empty()){
msg = "Error message not set";
}
ex.__set_reason(msg);
throw ex;
}
}
}
void RequestScheduler::Start() {
if(!stopped_) {
return;
}
stopped_ = false;
}
void RequestScheduler::Stop() {
if(stopped_) {
return;
}
SERVER_LOG_INFO << "Scheduler gonna stop...";
{
std::lock_guard<std::mutex> lock(queue_mtx_);
for(auto iter : task_groups_) {
if(iter.second != nullptr) {
iter.second->Put(nullptr);
}
}
}
for(auto iter : execute_threads_) {
if(iter == nullptr)
continue;
iter->join();
}
stopped_ = true;
SERVER_LOG_INFO << "Scheduler stopped";
}
ServerError RequestScheduler::ExecuteTask(const BaseTaskPtr& task_ptr) {
if(task_ptr == nullptr) {
return SERVER_NULL_POINTER;
}
ServerError err = PutTaskToQueue(task_ptr);
if(err != SERVER_SUCCESS) {
return err;
}
if(task_ptr->IsAsync()) {
return SERVER_SUCCESS;//async execution, caller need to call WaitToFinish at somewhere
}
return task_ptr->WaitToFinish();//sync execution
}
namespace {
void TakeTaskToExecute(TaskQueuePtr task_queue) {
if(task_queue == nullptr) {
return;
}
while(true) {
BaseTaskPtr task = task_queue->Take();
if (task == nullptr) {
break;//stop the thread
}
try {
ServerError err = task->Execute();
if(err != SERVER_SUCCESS) {
SERVER_LOG_ERROR << "Task failed with code: " << err;
}
} catch (std::exception& ex) {
SERVER_LOG_ERROR << "Task failed to execute: " << ex.what();
}
}
}
}
ServerError RequestScheduler::PutTaskToQueue(const BaseTaskPtr& task_ptr) {
std::lock_guard<std::mutex> lock(queue_mtx_);
std::string group_name = task_ptr->TaskGroup();
if(task_groups_.count(group_name) > 0) {
task_groups_[group_name]->Put(task_ptr);
} else {
TaskQueuePtr queue = std::make_shared<TaskQueue>();
queue->Put(task_ptr);
task_groups_.insert(std::make_pair(group_name, queue));
//start a thread
ThreadPtr thread = std::make_shared<std::thread>(&TakeTaskToExecute, queue);
execute_threads_.push_back(thread);
SERVER_LOG_INFO << "Create new thread for task group: " << group_name;
}
return SERVER_SUCCESS;
}
}
}
}

88
cpp/src/server/thrift_impl/RequestScheduler.h
View File

@ -1,88 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include "utils/BlockingQueue.h"
#include <map>
#include <vector>
#include <thread>
namespace zilliz {
namespace milvus {
namespace server {
class BaseTask {
protected:
BaseTask(const std::string& task_group, bool async = false);
virtual ~BaseTask();
public:
ServerError Execute();
ServerError WaitToFinish();
std::string TaskGroup() const { return task_group_; }
ServerError ErrorCode() const { return error_code_; }
std::string ErrorMsg() const { return error_msg_; }
bool IsAsync() const { return async_; }
protected:
virtual ServerError OnExecute() = 0;
ServerError SetError(ServerError error_code, const std::string& msg);
protected:
mutable std::mutex finish_mtx_;
std::condition_variable finish_cond_;
std::string task_group_;
bool async_;
bool done_;
ServerError error_code_;
std::string error_msg_;
};
using BaseTaskPtr = std::shared_ptr<BaseTask>;
using TaskQueue = BlockingQueue<BaseTaskPtr>;
using TaskQueuePtr = std::shared_ptr<TaskQueue>;
using ThreadPtr = std::shared_ptr<std::thread>;
class RequestScheduler {
public:
static RequestScheduler& GetInstance() {
static RequestScheduler scheduler;
return scheduler;
}
void Start();
void Stop();
ServerError ExecuteTask(const BaseTaskPtr& task_ptr);
static void ExecTask(BaseTaskPtr& task_ptr);
protected:
RequestScheduler();
virtual ~RequestScheduler();
ServerError PutTaskToQueue(const BaseTaskPtr& task_ptr);
private:
mutable std::mutex queue_mtx_;
std::map<std::string, TaskQueuePtr> task_groups_;
std::vector<ThreadPtr> execute_threads_;
bool stopped_;
};
}
}
}

779
cpp/src/server/thrift_impl/RequestTask.cpp
View File

@ -1,779 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "RequestTask.h"
#include "src/server/ServerConfig.h"
#include "utils/CommonUtil.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include "src/server/DBWrapper.h"
#include "version.h"
#ifdef MILVUS_ENABLE_PROFILING
#include "gperftools/profiler.h"
#endif
namespace zilliz {
namespace milvus {
namespace server {
using namespace ::milvus;
static const char* DQL_TASK_GROUP = "dql";
static const char* DDL_DML_TASK_GROUP = "ddl_dml";
static const char* PING_TASK_GROUP = "ping";
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},
{4, engine::EngineType::NSG_MIX},
};
if(map_type.find(type) == map_type.end()) {
return engine::EngineType::INVALID;
}
return map_type[type];
}
int IndexType(engine::EngineType type) {
static std::map<engine::EngineType, int> map_type = {
{engine::EngineType::INVALID, 0},
{engine::EngineType::FAISS_IDMAP, 1},
{engine::EngineType::FAISS_IVFFLAT, 2},
{engine::EngineType::FAISS_IVFSQ8, 3},
{engine::EngineType::NSG_MIX, 4},
};
if(map_type.find(type) == map_type.end()) {
return 0;
}
return map_type[type];
}
void
ConvertRowRecordToFloatArray(const std::vector<thrift::RowRecord>& record_array,
uint64_t dimension,
std::vector<float>& float_array,
ServerError& error_code,
std::string& error_msg) {
uint64_t vec_count = record_array.size();
float_array.resize(vec_count*dimension);//allocate enough memory
for(uint64_t i = 0; i < vec_count; i++) {
const auto& record = record_array[i];
if(record.vector_data.empty()) {
error_code = SERVER_INVALID_ROWRECORD;
error_msg = "Rowrecord float array is empty";
return;
}
uint64_t vec_dim = record.vector_data.size()/sizeof(double);//how many double value?
if(vec_dim != dimension) {
error_code = SERVER_INVALID_VECTOR_DIMENSION;
error_msg = "Invalid rowrecord dimension: " + std::to_string(vec_dim)
+ " vs. table dimension:" + std::to_string(dimension);
return;
}
//convert double array to float array(thrift has no float type)
const double* d_p = reinterpret_cast<const double*>(record.vector_data.data());
for(uint64_t d = 0; d < vec_dim; d++) {
float_array[i*vec_dim + d] = (float)(d_p[d]);
}
}
}
static constexpr long DAY_SECONDS = 86400;
void
ConvertTimeRangeToDBDates(const std::vector<thrift::Range> &range_array,
std::vector<DB_DATE>& dates,
ServerError& error_code,
std::string& error_msg) {
dates.clear();
for(auto& range : range_array) {
time_t tt_start, tt_end;
tm tm_start, tm_end;
if(!CommonUtil::TimeStrToTime(range.start_value, tt_start, tm_start)){
error_code = SERVER_INVALID_TIME_RANGE;
error_msg = "Invalid time range: " + range.start_value;
return;
}
if(!CommonUtil::TimeStrToTime(range.end_value, tt_end, tm_end)){
error_code = SERVER_INVALID_TIME_RANGE;
error_msg = "Invalid time range: " + range.start_value;
return;
}
long days = (tt_end > tt_start) ? (tt_end - tt_start)/DAY_SECONDS : (tt_start - tt_end)/DAY_SECONDS;
if(days == 0) {
error_code = SERVER_INVALID_TIME_RANGE;
error_msg = "Invalid time range: " + range.start_value + " to " + range.end_value;
return ;
}
for(long i = 0; i < days; i++) {
time_t tt_day = tt_start + DAY_SECONDS*i;
tm tm_day;
CommonUtil::ConvertTime(tt_day, tm_day);
long date = tm_day.tm_year*10000 + tm_day.tm_mon*100 + tm_day.tm_mday;//according to db logic
dates.push_back(date);
}
}
}
std::string
GetCurrTimeStr() {
char tm_buf[20] = {0};
time_t tt;
time(&tt);
tt = tt + 8 * 60 * 60;
tm* t = gmtime(&tt);
sprintf(tm_buf, "%4d%02d%02d_%02d%02d%02d", (t->tm_year+1900), (t->tm_mon+1), (t->tm_mday),
(t->tm_hour), (t->tm_min), (t->tm_sec));
return tm_buf;
}
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
CreateTableTask::CreateTableTask(const thrift::TableSchema& schema)
: BaseTask(DDL_DML_TASK_GROUP),
schema_(schema) {
}
BaseTaskPtr CreateTableTask::Create(const thrift::TableSchema& schema) {
return std::shared_ptr<BaseTask>(new CreateTableTask(schema));
}
ServerError CreateTableTask::OnExecute() {
TimeRecorder rc("CreateTableTask");
try {
//step 1: check arguments
ServerError res = SERVER_SUCCESS;
res = ValidationUtil::ValidateTableName(schema_.table_name);
if(res != SERVER_SUCCESS) {
return SetError(res, "Invalid table name: " + schema_.table_name);
}
res = ValidationUtil::ValidateTableDimension(schema_.dimension);
if(res != SERVER_SUCCESS) {
return SetError(res, "Invalid table dimension: " + std::to_string(schema_.dimension));
}
res = ValidationUtil::ValidateTableIndexType(schema_.index_type);
if(res != SERVER_SUCCESS) {
return SetError(res, "Invalid index type: " + std::to_string(schema_.index_type));
}
//step 2: construct table schema
engine::meta::TableSchema table_info;
table_info.dimension_ = (uint16_t)schema_.dimension;
table_info.table_id_ = schema_.table_name;
table_info.engine_type_ = (int)EngineType(schema_.index_type);
table_info.store_raw_data_ = schema_.store_raw_vector;
//step 3: create table
engine::Status stat = DBWrapper::DB()->CreateTable(table_info);
if(!stat.ok()) {//table could exist
return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
}
} catch (std::exception& ex) {
SERVER_LOG_ERROR << "CreateTableTask encounter exception: " << ex.what();
return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
}
rc.ElapseFromBegin("totally cost");
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
DescribeTableTask::DescribeTableTask(const std::string &table_name, thrift::TableSchema &schema)
: BaseTask(DDL_DML_TASK_GROUP),
table_name_(table_name),
schema_(schema) {
schema_.table_name = table_name_;
}
BaseTaskPtr DescribeTableTask::Create(const std::string& table_name, thrift::TableSchema& schema) {
return std::shared_ptr<BaseTask>(new DescribeTableTask(table_name, schema));
}
ServerError DescribeTableTask::OnExecute() {
TimeRecorder rc("DescribeTableTask");
try {
//step 1: check arguments
ServerError res = SERVER_SUCCESS;
res = ValidationUtil::ValidateTableName(table_name_);
if(res != SERVER_SUCCESS) {
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);
if(!stat.ok()) {
return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
}
schema_.table_name = table_info.table_id_;
schema_.index_type = IndexType((engine::EngineType)table_info.engine_type_);
schema_.dimension = table_info.dimension_;
schema_.store_raw_vector = table_info.store_raw_data_;
} catch (std::exception& ex) {
SERVER_LOG_ERROR << "DescribeTableTask encounter exception: " << ex.what();
return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
}
rc.ElapseFromBegin("totally cost");
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
BuildIndexTask::BuildIndexTask(const std::string& table_name)
: BaseTask(DDL_DML_TASK_GROUP),
table_name_(table_name) {
}
BaseTaskPtr BuildIndexTask::Create(const std::string& table_name) {
return std::shared_ptr<BaseTask>(new BuildIndexTask(table_name));
}
ServerError BuildIndexTask::OnExecute() {
try {
TimeRecorder rc("BuildIndexTask");
//step 1: check arguments
ServerError res = SERVER_SUCCESS;
res = ValidationUtil::ValidateTableName(table_name_);
if(res != SERVER_SUCCESS) {
return SetError(res, "Invalid table name: " + table_name_);
}
bool has_table = false;
engine::Status stat = DBWrapper::DB()->HasTable(table_name_, has_table);
if(!has_table) {
return SetError(SERVER_TABLE_NOT_EXIST, "Table " + table_name_ + " not exists");
}
//step 2: check table existence
stat = DBWrapper::DB()->BuildIndex(table_name_);
if(!stat.ok()) {
return SetError(SERVER_BUILD_INDEX_ERROR, "Engine failed: " + stat.ToString());
}
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
SERVER_LOG_ERROR << "BuildIndexTask encounter exception: " << ex.what();
return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
}
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
HasTableTask::HasTableTask(const std::string& table_name, bool& has_table)
: BaseTask(DDL_DML_TASK_GROUP),
table_name_(table_name),
has_table_(has_table) {
}
BaseTaskPtr HasTableTask::Create(const std::string& table_name, bool& has_table) {
return std::shared_ptr<BaseTask>(new HasTableTask(table_name, has_table));
}
ServerError HasTableTask::OnExecute() {
try {
TimeRecorder rc("HasTableTask");
//step 1: check arguments
ServerError res = SERVER_SUCCESS;
res = ValidationUtil::ValidateTableName(table_name_);
if(res != SERVER_SUCCESS) {
return SetError(res, "Invalid table name: " + table_name_);
}
//step 2: check table existence
engine::Status stat = DBWrapper::DB()->HasTable(table_name_, has_table_);
if(!stat.ok()) {
return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
}
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
SERVER_LOG_ERROR << "HasTableTask encounter exception: " << ex.what();
return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
}
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
DeleteTableTask::DeleteTableTask(const std::string& table_name)
: BaseTask(DDL_DML_TASK_GROUP),
table_name_(table_name) {
}
BaseTaskPtr DeleteTableTask::Create(const std::string& table_name) {
return std::shared_ptr<BaseTask>(new DeleteTableTask(table_name));
}
ServerError DeleteTableTask::OnExecute() {
try {
TimeRecorder rc("DeleteTableTask");
//step 1: check arguments
ServerError res = SERVER_SUCCESS;
res = ValidationUtil::ValidateTableName(table_name_);
if(res != SERVER_SUCCESS) {
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);
if(!stat.ok()) {
if(stat.IsNotFound()) {
return SetError(SERVER_TABLE_NOT_EXIST, "Table " + table_name_ + " not exists");
} else {
return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
}
}
//step 3: delete table
std::vector<DB_DATE> dates;
stat = DBWrapper::DB()->DeleteTable(table_name_, dates);
if(!stat.ok()) {
return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
}
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
SERVER_LOG_ERROR << "DeleteTableTask encounter exception: " << ex.what();
return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
}
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
ShowTablesTask::ShowTablesTask(std::vector<std::string>& tables)
: BaseTask(DDL_DML_TASK_GROUP),
tables_(tables) {
}
BaseTaskPtr ShowTablesTask::Create(std::vector<std::string>& tables) {
return std::shared_ptr<BaseTask>(new ShowTablesTask(tables));
}
ServerError ShowTablesTask::OnExecute() {
std::vector<engine::meta::TableSchema> schema_array;
engine::Status stat = DBWrapper::DB()->AllTables(schema_array);
if(!stat.ok()) {
return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
}
tables_.clear();
for(auto& schema : schema_array) {
tables_.push_back(schema.table_id_);
}
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
AddVectorTask::AddVectorTask(const std::string& table_name,
const std::vector<thrift::RowRecord>& record_array,
std::vector<int64_t>& record_ids)
: BaseTask(DDL_DML_TASK_GROUP),
table_name_(table_name),
record_array_(record_array),
record_ids_(record_ids) {
record_ids_.clear();
}
BaseTaskPtr AddVectorTask::Create(const std::string& table_name,
const std::vector<thrift::RowRecord>& record_array,
std::vector<int64_t>& record_ids) {
return std::shared_ptr<BaseTask>(new AddVectorTask(table_name, record_array, record_ids));
}
ServerError AddVectorTask::OnExecute() {
try {
TimeRecorder rc("AddVectorTask");
//step 1: check arguments
ServerError res = SERVER_SUCCESS;
res = ValidationUtil::ValidateTableName(table_name_);
if(res != SERVER_SUCCESS) {
return SetError(res, "Invalid table name: " + table_name_);
}
if(record_array_.empty()) {
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);
if(!stat.ok()) {
if(stat.IsNotFound()) {
return SetError(SERVER_TABLE_NOT_EXIST, "Table " + table_name_ + " not exists");
} else {
return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
}
}
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
//step 3: prepare float data
std::vector<float> vec_f;
ServerError error_code = SERVER_SUCCESS;
std::string error_msg;
ConvertRowRecordToFloatArray(record_array_, table_info.dimension_, vec_f, error_code, error_msg);
if(error_code != SERVER_SUCCESS) {
return SetError(error_code, error_msg);
}
rc.RecordSection("prepare vectors data");
//step 4: insert vectors
uint64_t vec_count = (uint64_t)record_array_.size();
stat = DBWrapper::DB()->InsertVectors(table_name_, vec_count, vec_f.data(), record_ids_);
if(!stat.ok()) {
return SetError(SERVER_CACHE_ERROR, "Cache error: " + stat.ToString());
}
if(record_ids_.size() != vec_count) {
std::string msg = "Add " + std::to_string(vec_count) + " vectors but only return "
+ std::to_string(record_ids_.size()) + " id";
return SetError(SERVER_ILLEGAL_VECTOR_ID, msg);
}
#ifdef MILVUS_ENABLE_PROFILING
ProfilerStop();
#endif
rc.RecordSection("add vectors to engine");
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
SERVER_LOG_ERROR << "AddVectorTask encounter exception: " << ex.what();
return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
}
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
SearchVectorTaskBase::SearchVectorTaskBase(const std::string &table_name,
const std::vector<std::string>& file_id_array,
const std::vector<thrift::RowRecord> &query_record_array,
const std::vector<thrift::Range> &query_range_array,
const int64_t top_k)
: BaseTask(DQL_TASK_GROUP),
table_name_(table_name),
file_id_array_(file_id_array),
record_array_(query_record_array),
range_array_(query_range_array),
top_k_(top_k) {
}
ServerError SearchVectorTaskBase::OnExecute() {
try {
std::string title = "SearchVectorTask(n=" + std::to_string(record_array_.size())
+ " k=" + std::to_string(top_k_) + ")";
TimeRecorder rc(title);
//step 1: check arguments
ServerError res = SERVER_SUCCESS;
res = ValidationUtil::ValidateTableName(table_name_);
if(res != SERVER_SUCCESS) {
return SetError(res, "Invalid table name: " + table_name_);
}
if(top_k_ <= 0 || top_k_ > 1024) {
return SetError(SERVER_INVALID_TOPK, "Invalid topk: " + std::to_string(top_k_));
}
if(record_array_.empty()) {
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);
if(!stat.ok()) {
if(stat.IsNotFound()) {
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;
ConvertTimeRangeToDBDates(range_array_, dates, error_code, error_msg);
if(error_code != SERVER_SUCCESS) {
return SetError(error_code, error_msg);
}
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
//step 3: prepare float data
std::vector<float> vec_f;
ConvertRowRecordToFloatArray(record_array_, table_info.dimension_, vec_f, error_code, error_msg);
if(error_code != SERVER_SUCCESS) {
return SetError(error_code, error_msg);
}
double span_prepare = rc.RecordSection("prepare vector data");
//step 4: search vectors
engine::QueryResults results;
uint64_t record_count = (uint64_t)record_array_.size();
if(file_id_array_.empty()) {
stat = DBWrapper::DB()->Query(table_name_, (size_t) top_k_, record_count, vec_f.data(), dates, results);
} else {
stat = DBWrapper::DB()->Query(table_name_, file_id_array_, (size_t) top_k_, record_count, vec_f.data(), dates, results);
}
double span_search = rc.RecordSection("search vectors from engine");
if(!stat.ok()) {
return SetError(DB_META_TRANSACTION_FAILED, "Engine failed: " + stat.ToString());
}
if(results.empty()) {
return SERVER_SUCCESS; //empty table
}
if(results.size() != record_count) {
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);
}
//step 5: construct result array
ConstructResult(results);
#ifdef MILVUS_ENABLE_PROFILING
ProfilerStop();
#endif
double span_result = rc.RecordSection("construct result");
rc.ElapseFromBegin("totally cost");
//step 6: print time cost percent
double total_cost = span_check + span_prepare + span_search + span_result;
SERVER_LOG_DEBUG << title << ": check validation(" << (span_check/total_cost)*100.0 << "%)"
<< " prepare data(" << (span_prepare/total_cost)*100.0 << "%)"
<< " search(" << (span_search/total_cost)*100.0 << "%)"
<< " construct result(" << (span_result/total_cost)*100.0 << "%)";
} catch (std::exception& ex) {
SERVER_LOG_ERROR << "SearchVectorTask encounter exception: " << ex.what();
return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
}
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
SearchVectorTask1::SearchVectorTask1(const std::string &table_name,
const std::vector<std::string>& file_id_array,
const std::vector<thrift::RowRecord> &query_record_array,
const std::vector<thrift::Range> &query_range_array,
const int64_t top_k,
std::vector<thrift::TopKQueryResult> &result_array)
: SearchVectorTaskBase(table_name, file_id_array, query_record_array, query_range_array, top_k),
result_array_(result_array) {
}
BaseTaskPtr SearchVectorTask1::Create(const std::string& table_name,
const std::vector<std::string>& file_id_array,
const std::vector<thrift::RowRecord> & query_record_array,
const std::vector<thrift::Range> & query_range_array,
const int64_t top_k,
std::vector<thrift::TopKQueryResult>& result_array) {
return std::shared_ptr<BaseTask>(new SearchVectorTask1(table_name, file_id_array,
query_record_array, query_range_array, top_k, result_array));
}
ServerError SearchVectorTask1::ConstructResult(engine::QueryResults& results) {
for(uint64_t i = 0; i < results.size(); i++) {
auto& result = results[i];
const auto& record = record_array_[i];
thrift::TopKQueryResult thrift_topk_result;
for(auto& pair : result) {
thrift::QueryResult thrift_result;
thrift_result.__set_id(pair.first);
thrift_result.__set_distance(pair.second);
thrift_topk_result.query_result_arrays.emplace_back(thrift_result);
}
result_array_.emplace_back(thrift_topk_result);
}
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
SearchVectorTask2::SearchVectorTask2(const std::string &table_name,
const std::vector<std::string>& file_id_array,
const std::vector<thrift::RowRecord> &query_record_array,
const std::vector<thrift::Range> &query_range_array,
const int64_t top_k,
std::vector<thrift::TopKQueryBinResult> &result_array)
: SearchVectorTaskBase(table_name, file_id_array, query_record_array, query_range_array, top_k),
result_array_(result_array) {
}
BaseTaskPtr SearchVectorTask2::Create(const std::string& table_name,
const std::vector<std::string>& file_id_array,
const std::vector<thrift::RowRecord> & query_record_array,
const std::vector<thrift::Range> & query_range_array,
const int64_t top_k,
std::vector<thrift::TopKQueryBinResult>& result_array) {
return std::shared_ptr<BaseTask>(new SearchVectorTask2(table_name, file_id_array,
query_record_array, query_range_array, top_k, result_array));
}
ServerError SearchVectorTask2::ConstructResult(engine::QueryResults& results) {
for(size_t i = 0; i < results.size(); i++) {
auto& result = results[i];
thrift::TopKQueryBinResult thrift_topk_result;
if(result.empty()) {
result_array_.emplace_back(thrift_topk_result);
continue;
}
std::string str_ids, str_distances;
str_ids.resize(sizeof(engine::IDNumber)*result.size());
str_distances.resize(sizeof(double)*result.size());
engine::IDNumber* ids_ptr = (engine::IDNumber*)str_ids.data();
double* distance_ptr = (double*)str_distances.data();
for(size_t k = 0; k < result.size(); k++) {
auto& pair = result[k];
ids_ptr[k] = pair.first;
distance_ptr[k] = pair.second;
}
thrift_topk_result.__set_id_array(str_ids);
thrift_topk_result.__set_distance_array(str_distances);
result_array_.emplace_back(thrift_topk_result);
}
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
GetTableRowCountTask::GetTableRowCountTask(const std::string& table_name, int64_t& row_count)
: BaseTask(DDL_DML_TASK_GROUP),
table_name_(table_name),
row_count_(row_count) {
}
BaseTaskPtr GetTableRowCountTask::Create(const std::string& table_name, int64_t& row_count) {
return std::shared_ptr<BaseTask>(new GetTableRowCountTask(table_name, row_count));
}
ServerError GetTableRowCountTask::OnExecute() {
try {
TimeRecorder rc("GetTableRowCountTask");
//step 1: check arguments
ServerError res = SERVER_SUCCESS;
res = ValidationUtil::ValidateTableName(table_name_);
if(res != SERVER_SUCCESS) {
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;
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
SERVER_LOG_ERROR << "GetTableRowCountTask encounter exception: " << ex.what();
return SetError(SERVER_UNEXPECTED_ERROR, ex.what());
}
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
PingTask::PingTask(const std::string& cmd, std::string& result)
: BaseTask(PING_TASK_GROUP),
cmd_(cmd),
result_(result) {
}
BaseTaskPtr PingTask::Create(const std::string& cmd, std::string& result) {
return std::shared_ptr<BaseTask>(new PingTask(cmd, result));
}
ServerError PingTask::OnExecute() {
if(cmd_ == "version") {
result_ = MILVUS_VERSION;
}
return SERVER_SUCCESS;
}
}
}
}

232
cpp/src/server/thrift_impl/RequestTask.h
View File

@ -1,232 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include "RequestScheduler.h"
#include "utils/Error.h"
#include "db/Types.h"
#include "milvus_types.h"
#include <condition_variable>
#include <memory>
namespace zilliz {
namespace milvus {
namespace server {
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class CreateTableTask : public BaseTask {
public:
static BaseTaskPtr Create(const ::milvus::thrift::TableSchema& schema);
protected:
CreateTableTask(const ::milvus::thrift::TableSchema& schema);
ServerError OnExecute() override;
private:
const ::milvus::thrift::TableSchema& schema_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class HasTableTask : public BaseTask {
public:
static BaseTaskPtr Create(const std::string& table_name, bool& has_table);
protected:
HasTableTask(const std::string& table_name, bool& has_table);
ServerError OnExecute() override;
private:
std::string table_name_;
bool& has_table_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class DescribeTableTask : public BaseTask {
public:
static BaseTaskPtr Create(const std::string& table_name, ::milvus::thrift::TableSchema& schema);
protected:
DescribeTableTask(const std::string& table_name, ::milvus::thrift::TableSchema& schema);
ServerError OnExecute() override;
private:
std::string table_name_;
::milvus::thrift::TableSchema& schema_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class DeleteTableTask : public BaseTask {
public:
static BaseTaskPtr Create(const std::string& table_name);
protected:
DeleteTableTask(const std::string& table_name);
ServerError OnExecute() override;
private:
std::string table_name_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class BuildIndexTask : public BaseTask {
public:
static BaseTaskPtr Create(const std::string& table_name);
protected:
BuildIndexTask(const std::string& table_name);
ServerError OnExecute() override;
private:
std::string table_name_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class ShowTablesTask : public BaseTask {
public:
static BaseTaskPtr Create(std::vector<std::string>& tables);
protected:
ShowTablesTask(std::vector<std::string>& tables);
ServerError OnExecute() override;
private:
std::vector<std::string>& tables_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class AddVectorTask : public BaseTask {
public:
static BaseTaskPtr Create(const std::string& table_name,
const std::vector<::milvus::thrift::RowRecord>& record_array,
std::vector<int64_t>& record_ids_);
protected:
AddVectorTask(const std::string& table_name,
const std::vector<::milvus::thrift::RowRecord>& record_array,
std::vector<int64_t>& record_ids_);
ServerError OnExecute() override;
private:
std::string table_name_;
const std::vector<::milvus::thrift::RowRecord>& record_array_;
std::vector<int64_t>& record_ids_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class SearchVectorTaskBase : public BaseTask {
protected:
SearchVectorTaskBase(const std::string& table_name,
const std::vector<std::string>& file_id_array,
const std::vector<::milvus::thrift::RowRecord> & query_record_array,
const std::vector<::milvus::thrift::Range> & query_range_array,
const int64_t top_k);
ServerError OnExecute() override;
virtual ServerError ConstructResult(engine::QueryResults& results) = 0;
protected:
std::string table_name_;
std::vector<std::string> file_id_array_;
int64_t top_k_;
const std::vector<::milvus::thrift::RowRecord>& record_array_;
const std::vector<::milvus::thrift::Range>& range_array_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class SearchVectorTask1 : public SearchVectorTaskBase {
public:
static BaseTaskPtr Create(const std::string& table_name,
const std::vector<std::string>& file_id_array,
const std::vector<::milvus::thrift::RowRecord> & query_record_array,
const std::vector<::milvus::thrift::Range> & query_range_array,
const int64_t top_k,
std::vector<::milvus::thrift::TopKQueryResult>& result_array);
protected:
SearchVectorTask1(const std::string& table_name,
const std::vector<std::string>& file_id_array,
const std::vector<::milvus::thrift::RowRecord> & query_record_array,
const std::vector<::milvus::thrift::Range> & query_range_array,
const int64_t top_k,
std::vector<::milvus::thrift::TopKQueryResult>& result_array);
ServerError ConstructResult(engine::QueryResults& results) override;
private:
std::vector<::milvus::thrift::TopKQueryResult>& result_array_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class SearchVectorTask2 : public SearchVectorTaskBase {
public:
static BaseTaskPtr Create(const std::string& table_name,
const std::vector<std::string>& file_id_array,
const std::vector<::milvus::thrift::RowRecord> & query_record_array,
const std::vector<::milvus::thrift::Range> & query_range_array,
const int64_t top_k,
std::vector<::milvus::thrift::TopKQueryBinResult>& result_array);
protected:
SearchVectorTask2(const std::string& table_name,
const std::vector<std::string>& file_id_array,
const std::vector<::milvus::thrift::RowRecord> & query_record_array,
const std::vector<::milvus::thrift::Range> & query_range_array,
const int64_t top_k,
std::vector<::milvus::thrift::TopKQueryBinResult>& result_array);
ServerError ConstructResult(engine::QueryResults& results) override;
private:
std::vector<::milvus::thrift::TopKQueryBinResult>& result_array_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class GetTableRowCountTask : public BaseTask {
public:
static BaseTaskPtr Create(const std::string& table_name, int64_t& row_count);
protected:
GetTableRowCountTask(const std::string& table_name, int64_t& row_count);
ServerError OnExecute() override;
private:
std::string table_name_;
int64_t& row_count_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class PingTask : public BaseTask {
public:
static BaseTaskPtr Create(const std::string& cmd, std::string& result);
protected:
PingTask(const std::string& cmd, std::string& result);
ServerError OnExecute() override;
private:
std::string cmd_;
std::string& result_;
};
}
}
}

36
cpp/src/server/thrift_impl/ThreadPoolServer.cpp
View File

@ -1,36 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "metrics/Metrics.h"
#include "ThreadPoolServer.h"
namespace zilliz {
namespace milvus {
namespace server {
void
ThreadPoolServer::onClientConnected(const std::shared_ptr<apache::thrift::server::TConnectedClient> &pClient) {
server::Metrics::GetInstance().ConnectionGaugeIncrement();
TThreadPoolServer::onClientConnected(pClient);
}
void
ThreadPoolServer::onClientDisconnected(apache::thrift::server::TConnectedClient *pClient) {
server::Metrics::GetInstance().ConnectionGaugeDecrement();
TThreadPoolServer::onClientDisconnected(pClient);
}
zilliz::milvus::server::ThreadPoolServer::ThreadPoolServer(const std::shared_ptr<apache::thrift::TProcessor> &processor,
const std::shared_ptr<apache::thrift::transport::TServerTransport> &serverTransport,
const std::shared_ptr<apache::thrift::transport::TTransportFactory> &transportFactory,
const std::shared_ptr<apache::thrift::protocol::TProtocolFactory> &protocolFactory,
const std::shared_ptr<apache::thrift::concurrency::ThreadManager> &threadManager)
: TThreadPoolServer(processor, serverTransport, transportFactory, protocolFactory, threadManager) {
}
}
}
}

33
cpp/src/server/thrift_impl/ThreadPoolServer.h
View File

@ -1,33 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include <thrift/server/TThreadPoolServer.h>
namespace zilliz {
namespace milvus {
namespace server {
class ThreadPoolServer : public apache::thrift::server::TThreadPoolServer {
public:
ThreadPoolServer(
const std::shared_ptr<apache::thrift::TProcessor>& processor,
const std::shared_ptr<apache::thrift::transport::TServerTransport>& serverTransport,
const std::shared_ptr<apache::thrift::transport::TTransportFactory>& transportFactory,
const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& protocolFactory,
const std::shared_ptr<apache::thrift::concurrency::ThreadManager>& threadManager
= apache::thrift::concurrency::ThreadManager::newSimpleThreadManager());
protected:
void onClientConnected(const std::shared_ptr<apache::thrift::server::TConnectedClient>& pClient) override ;
void onClientDisconnected(apache::thrift::server::TConnectedClient* pClient) override ;
};
}
}
}

4
cpp/src/thrift/cpp_gen.sh
View File

@ -1,4 +0,0 @@
#!/bin/bash
thrift -r --gen cpp ./milvus.thrift

5457
cpp/src/thrift/gen-cpp/MilvusService.cpp
View File

File diff suppressed because it is too large Load Diff

1972
cpp/src/thrift/gen-cpp/MilvusService.h
View File

File diff suppressed because it is too large Load Diff

241
cpp/src/thrift/gen-cpp/MilvusService_server.skeleton.cpp
View File

@ -1,241 +0,0 @@
// This autogenerated skeleton file illustrates how to build a server.
// You should copy it to another filename to avoid overwriting it.
#include "MilvusService.h"
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TSimpleServer.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TBufferTransports.h>
using namespace ::apache::thrift;
using namespace ::apache::thrift::protocol;
using namespace ::apache::thrift::transport;
using namespace ::apache::thrift::server;
using namespace ::milvus::thrift;
class MilvusServiceHandler : virtual public MilvusServiceIf {
public:
MilvusServiceHandler() {
// Your initialization goes here
}
/**
* @brief Create table method
*
* This method is used to create table
*
* @param param, use to provide table information to be created.
*
*
* @param param
*/
void CreateTable(const TableSchema& param) {
// Your implementation goes here
printf("CreateTable\n");
}
/**
* @brief Test table existence method
*
* This method is used to test table existence.
*
* @param table_name, table name is going to be tested.
*
*
* @param table_name
*/
bool HasTable(const std::string& table_name) {
// Your implementation goes here
printf("HasTable\n");
}
/**
* @brief Delete table method
*
* This method is used to delete table.
*
* @param table_name, table name is going to be deleted.
*
*
* @param table_name
*/
void DeleteTable(const std::string& table_name) {
// Your implementation goes here
printf("DeleteTable\n");
}
/**
* @brief Build index by table method
*
* This method is used to build index by table in sync mode.
*
* @param table_name, table is going to be built index.
*
*
* @param table_name
*/
void BuildIndex(const std::string& table_name) {
// Your implementation goes here
printf("BuildIndex\n");
}
/**
* @brief Add vector array to table
*
* This method is used to add vector array to table.
*
* @param table_name, table_name is inserted.
* @param record_array, vector array is inserted.
*
* @return vector id array
*
* @param table_name
* @param record_array
*/
void AddVector(std::vector<int64_t> & _return, const std::string& table_name, const std::vector<RowRecord> & record_array) {
// Your implementation goes here
printf("AddVector\n");
}
/**
* @brief Query vector
*
* This method is used to query vector in table.
*
* @param table_name, table_name is queried.
* @param query_record_array, all vector are going to be queried.
* @param query_range_array, optional ranges for conditional search. If not specified, search whole table
* @param topk, how many similarity vectors will be searched.
*
* @return query result array.
*
* @param table_name
* @param query_record_array
* @param query_range_array
* @param topk
*/
void SearchVector(std::vector<TopKQueryResult> & _return, const std::string& table_name, const std::vector<RowRecord> & query_record_array, const std::vector<Range> & query_range_array, const int64_t topk) {
// Your implementation goes here
printf("SearchVector\n");
}
/**
* @brief Query vector
*
* This method is used to query vector in table.
*
* @param table_name, table_name is queried.
* @param query_record_array, all vector are going to be queried.
* @param query_range_array, optional ranges for conditional search. If not specified, search whole table
* @param topk, how many similarity vectors will be searched.
*
* @return query binary result array.
*
* @param table_name
* @param query_record_array
* @param query_range_array
* @param topk
*/
void SearchVector2(std::vector<TopKQueryBinResult> & _return, const std::string& table_name, const std::vector<RowRecord> & query_record_array, const std::vector<Range> & query_range_array, const int64_t topk) {
// Your implementation goes here
printf("SearchVector2\n");
}
/**
* @brief Internal use query interface
*
* This method is used to query vector in specified files.
*
* @param file_id_array, specified files id array, queried.
* @param query_record_array, all vector are going to be queried.
* @param query_range_array, optional ranges for conditional search. If not specified, search whole table
* @param topk, how many similarity vectors will be searched.
*
* @return query result array.
*
* @param table_name
* @param file_id_array
* @param query_record_array
* @param query_range_array
* @param topk
*/
void SearchVectorInFiles(std::vector<TopKQueryResult> & _return, const std::string& table_name, const std::vector<std::string> & file_id_array, const std::vector<RowRecord> & query_record_array, const std::vector<Range> & query_range_array, const int64_t topk) {
// Your implementation goes here
printf("SearchVectorInFiles\n");
}
/**
* @brief Get table schema
*
* This method is used to get table schema.
*
* @param table_name, target table name.
*
* @return table schema
*
* @param table_name
*/
void DescribeTable(TableSchema& _return, const std::string& table_name) {
// Your implementation goes here
printf("DescribeTable\n");
}
/**
* @brief Get table row count
*
* This method is used to get table row count.
*
* @param table_name, target table name.
*
* @return table row count
*
* @param table_name
*/
int64_t GetTableRowCount(const std::string& table_name) {
// Your implementation goes here
printf("GetTableRowCount\n");
}
/**
* @brief List all tables in database
*
* This method is used to list all tables.
*
*
* @return table names.
*/
void ShowTables(std::vector<std::string> & _return) {
// Your implementation goes here
printf("ShowTables\n");
}
/**
* @brief Give the server status
*
* This method is used to give the server status.
*
* @return Server status.
*
* @param cmd
*/
void Ping(std::string& _return, const std::string& cmd) {
// Your implementation goes here
printf("Ping\n");
}
};
int main(int argc, char **argv) {
int port = 9090;
::apache::thrift::stdcxx::shared_ptr<MilvusServiceHandler> handler(new MilvusServiceHandler());
::apache::thrift::stdcxx::shared_ptr<TProcessor> processor(new MilvusServiceProcessor(handler));
::apache::thrift::stdcxx::shared_ptr<TServerTransport> serverTransport(new TServerSocket(port));
::apache::thrift::stdcxx::shared_ptr<TTransportFactory> transportFactory(new TBufferedTransportFactory());
::apache::thrift::stdcxx::shared_ptr<TProtocolFactory> protocolFactory(new TBinaryProtocolFactory());
TSimpleServer server(processor, serverTransport, transportFactory, protocolFactory);
server.serve();
return 0;
}

17
cpp/src/thrift/gen-cpp/milvus_constants.cpp
View File

@ -1,17 +0,0 @@
/**
* Autogenerated by Thrift Compiler (0.12.0)
*
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
* @generated
*/
#include "milvus_constants.h"
namespace milvus { namespace thrift {
const milvusConstants g_milvus_constants;
milvusConstants::milvusConstants() {
}
}} // namespace

24
cpp/src/thrift/gen-cpp/milvus_constants.h
View File

@ -1,24 +0,0 @@
/**
* Autogenerated by Thrift Compiler (0.12.0)
*
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
* @generated
*/
#ifndef milvus_CONSTANTS_H
#define milvus_CONSTANTS_H
#include "milvus_types.h"
namespace milvus { namespace thrift {
class milvusConstants {
public:
milvusConstants();
};
extern const milvusConstants g_milvus_constants;
}} // namespace
#endif

899
cpp/src/thrift/gen-cpp/milvus_types.cpp
View File

@ -1,899 +0,0 @@
/**
* Autogenerated by Thrift Compiler (0.12.0)
*
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
* @generated
*/
#include "milvus_types.h"
#include <algorithm>
#include <ostream>
#include <thrift/TToString.h>
namespace milvus { namespace thrift {
int _kErrorCodeValues[] = {
ErrorCode::SUCCESS,
ErrorCode::UNEXPECTED_ERROR,
ErrorCode::CONNECT_FAILED,
ErrorCode::PERMISSION_DENIED,
ErrorCode::TABLE_NOT_EXISTS,
ErrorCode::ILLEGAL_ARGUMENT,
ErrorCode::ILLEGAL_RANGE,
ErrorCode::ILLEGAL_DIMENSION,
ErrorCode::ILLEGAL_INDEX_TYPE,
ErrorCode::ILLEGAL_TABLE_NAME,
ErrorCode::ILLEGAL_TOPK,
ErrorCode::ILLEGAL_ROWRECORD,
ErrorCode::ILLEGAL_VECTOR_ID,
ErrorCode::ILLEGAL_SEARCH_RESULT,
ErrorCode::FILE_NOT_FOUND,
ErrorCode::META_FAILED,
ErrorCode::CACHE_FAILED,
ErrorCode::CANNOT_CREATE_FOLDER,
ErrorCode::CANNOT_CREATE_FILE,
ErrorCode::CANNOT_DELETE_FOLDER,
ErrorCode::CANNOT_DELETE_FILE,
ErrorCode::BUILD_INDEX_ERROR
};
const char* _kErrorCodeNames[] = {
"SUCCESS",
"UNEXPECTED_ERROR",
"CONNECT_FAILED",
"PERMISSION_DENIED",
"TABLE_NOT_EXISTS",
"ILLEGAL_ARGUMENT",
"ILLEGAL_RANGE",
"ILLEGAL_DIMENSION",
"ILLEGAL_INDEX_TYPE",
"ILLEGAL_TABLE_NAME",
"ILLEGAL_TOPK",
"ILLEGAL_ROWRECORD",
"ILLEGAL_VECTOR_ID",
"ILLEGAL_SEARCH_RESULT",
"FILE_NOT_FOUND",
"META_FAILED",
"CACHE_FAILED",
"CANNOT_CREATE_FOLDER",
"CANNOT_CREATE_FILE",
"CANNOT_DELETE_FOLDER",
"CANNOT_DELETE_FILE",
"BUILD_INDEX_ERROR"
};
const std::map<int, const char*> _ErrorCode_VALUES_TO_NAMES(::apache::thrift::TEnumIterator(22, _kErrorCodeValues, _kErrorCodeNames), ::apache::thrift::TEnumIterator(-1, NULL, NULL));
std::ostream& operator<<(std::ostream& out, const ErrorCode::type& val) {
std::map<int, const char*>::const_iterator it = _ErrorCode_VALUES_TO_NAMES.find(val);
if (it != _ErrorCode_VALUES_TO_NAMES.end()) {
out << it->second;
} else {
out << static_cast<int>(val);
}
return out;
}
Exception::~Exception() throw() {
}
void Exception::__set_code(const ErrorCode::type val) {
this->code = val;
}
void Exception::__set_reason(const std::string& val) {
this->reason = val;
}
std::ostream& operator<<(std::ostream& out, const Exception& obj)
{
obj.printTo(out);
return out;
}
uint32_t Exception::read(::apache::thrift::protocol::TProtocol* iprot) {
::apache::thrift::protocol::TInputRecursionTracker tracker(*iprot);
uint32_t xfer = 0;
std::string fname;
::apache::thrift::protocol::TType ftype;
int16_t fid;
xfer += iprot->readStructBegin(fname);
using ::apache::thrift::protocol::TProtocolException;
while (true)
{
xfer += iprot->readFieldBegin(fname, ftype, fid);
if (ftype == ::apache::thrift::protocol::T_STOP) {
break;
}
switch (fid)
{
case 1:
if (ftype == ::apache::thrift::protocol::T_I32) {
int32_t ecast0;
xfer += iprot->readI32(ecast0);
this->code = (ErrorCode::type)ecast0;
this->__isset.code = true;
} else {
xfer += iprot->skip(ftype);
}
break;
case 2:
if (ftype == ::apache::thrift::protocol::T_STRING) {
xfer += iprot->readString(this->reason);
this->__isset.reason = true;
} else {
xfer += iprot->skip(ftype);
}
break;
default:
xfer += iprot->skip(ftype);
break;
}
xfer += iprot->readFieldEnd();
}
xfer += iprot->readStructEnd();
return xfer;
}
uint32_t Exception::write(::apache::thrift::protocol::TProtocol* oprot) const {
uint32_t xfer = 0;
::apache::thrift::protocol::TOutputRecursionTracker tracker(*oprot);
xfer += oprot->writeStructBegin("Exception");
xfer += oprot->writeFieldBegin("code", ::apache::thrift::protocol::T_I32, 1);
xfer += oprot->writeI32((int32_t)this->code);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldBegin("reason", ::apache::thrift::protocol::T_STRING, 2);
xfer += oprot->writeString(this->reason);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldStop();
xfer += oprot->writeStructEnd();
return xfer;
}
void swap(Exception &a, Exception &b) {
using ::std::swap;
swap(a.code, b.code);
swap(a.reason, b.reason);
swap(a.__isset, b.__isset);
}
Exception::Exception(const Exception& other1) : TException() {
code = other1.code;
reason = other1.reason;
__isset = other1.__isset;
}
Exception& Exception::operator=(const Exception& other2) {
code = other2.code;
reason = other2.reason;
__isset = other2.__isset;
return *this;
}
void Exception::printTo(std::ostream& out) const {
using ::apache::thrift::to_string;
out << "Exception(";
out << "code=" << to_string(code);
out << ", " << "reason=" << to_string(reason);
out << ")";
}
const char* Exception::what() const throw() {
try {
std::stringstream ss;
ss << "TException - service has thrown: " << *this;
this->thriftTExceptionMessageHolder_ = ss.str();
return this->thriftTExceptionMessageHolder_.c_str();
} catch (const std::exception&) {
return "TException - service has thrown: Exception";
}
}
TableSchema::~TableSchema() throw() {
}
void TableSchema::__set_table_name(const std::string& val) {
this->table_name = val;
}
void TableSchema::__set_index_type(const int32_t val) {
this->index_type = val;
}
void TableSchema::__set_dimension(const int64_t val) {
this->dimension = val;
}
void TableSchema::__set_store_raw_vector(const bool val) {
this->store_raw_vector = val;
}
std::ostream& operator<<(std::ostream& out, const TableSchema& obj)
{
obj.printTo(out);
return out;
}
uint32_t TableSchema::read(::apache::thrift::protocol::TProtocol* iprot) {
::apache::thrift::protocol::TInputRecursionTracker tracker(*iprot);
uint32_t xfer = 0;
std::string fname;
::apache::thrift::protocol::TType ftype;
int16_t fid;
xfer += iprot->readStructBegin(fname);
using ::apache::thrift::protocol::TProtocolException;
bool isset_table_name = false;
while (true)
{
xfer += iprot->readFieldBegin(fname, ftype, fid);
if (ftype == ::apache::thrift::protocol::T_STOP) {
break;
}
switch (fid)
{
case 1:
if (ftype == ::apache::thrift::protocol::T_STRING) {
xfer += iprot->readString(this->table_name);
isset_table_name = true;
} else {
xfer += iprot->skip(ftype);
}
break;
case 2:
if (ftype == ::apache::thrift::protocol::T_I32) {
xfer += iprot->readI32(this->index_type);
this->__isset.index_type = true;
} else {
xfer += iprot->skip(ftype);
}
break;
case 3:
if (ftype == ::apache::thrift::protocol::T_I64) {
xfer += iprot->readI64(this->dimension);
this->__isset.dimension = true;
} else {
xfer += iprot->skip(ftype);
}
break;
case 4:
if (ftype == ::apache::thrift::protocol::T_BOOL) {
xfer += iprot->readBool(this->store_raw_vector);
this->__isset.store_raw_vector = true;
} else {
xfer += iprot->skip(ftype);
}
break;
default:
xfer += iprot->skip(ftype);
break;
}
xfer += iprot->readFieldEnd();
}
xfer += iprot->readStructEnd();
if (!isset_table_name)
throw TProtocolException(TProtocolException::INVALID_DATA);
return xfer;
}
uint32_t TableSchema::write(::apache::thrift::protocol::TProtocol* oprot) const {
uint32_t xfer = 0;
::apache::thrift::protocol::TOutputRecursionTracker tracker(*oprot);
xfer += oprot->writeStructBegin("TableSchema");
xfer += oprot->writeFieldBegin("table_name", ::apache::thrift::protocol::T_STRING, 1);
xfer += oprot->writeString(this->table_name);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldBegin("index_type", ::apache::thrift::protocol::T_I32, 2);
xfer += oprot->writeI32(this->index_type);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldBegin("dimension", ::apache::thrift::protocol::T_I64, 3);
xfer += oprot->writeI64(this->dimension);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldBegin("store_raw_vector", ::apache::thrift::protocol::T_BOOL, 4);
xfer += oprot->writeBool(this->store_raw_vector);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldStop();
xfer += oprot->writeStructEnd();
return xfer;
}
void swap(TableSchema &a, TableSchema &b) {
using ::std::swap;
swap(a.table_name, b.table_name);
swap(a.index_type, b.index_type);
swap(a.dimension, b.dimension);
swap(a.store_raw_vector, b.store_raw_vector);
swap(a.__isset, b.__isset);
}
TableSchema::TableSchema(const TableSchema& other3) {
table_name = other3.table_name;
index_type = other3.index_type;
dimension = other3.dimension;
store_raw_vector = other3.store_raw_vector;
__isset = other3.__isset;
}
TableSchema& TableSchema::operator=(const TableSchema& other4) {
table_name = other4.table_name;
index_type = other4.index_type;
dimension = other4.dimension;
store_raw_vector = other4.store_raw_vector;
__isset = other4.__isset;
return *this;
}
void TableSchema::printTo(std::ostream& out) const {
using ::apache::thrift::to_string;
out << "TableSchema(";
out << "table_name=" << to_string(table_name);
out << ", " << "index_type=" << to_string(index_type);
out << ", " << "dimension=" << to_string(dimension);
out << ", " << "store_raw_vector=" << to_string(store_raw_vector);
out << ")";
}
Range::~Range() throw() {
}
void Range::__set_start_value(const std::string& val) {
this->start_value = val;
}
void Range::__set_end_value(const std::string& val) {
this->end_value = val;
}
std::ostream& operator<<(std::ostream& out, const Range& obj)
{
obj.printTo(out);
return out;
}
uint32_t Range::read(::apache::thrift::protocol::TProtocol* iprot) {
::apache::thrift::protocol::TInputRecursionTracker tracker(*iprot);
uint32_t xfer = 0;
std::string fname;
::apache::thrift::protocol::TType ftype;
int16_t fid;
xfer += iprot->readStructBegin(fname);
using ::apache::thrift::protocol::TProtocolException;
while (true)
{
xfer += iprot->readFieldBegin(fname, ftype, fid);
if (ftype == ::apache::thrift::protocol::T_STOP) {
break;
}
switch (fid)
{
case 1:
if (ftype == ::apache::thrift::protocol::T_STRING) {
xfer += iprot->readString(this->start_value);
this->__isset.start_value = true;
} else {
xfer += iprot->skip(ftype);
}
break;
case 2:
if (ftype == ::apache::thrift::protocol::T_STRING) {
xfer += iprot->readString(this->end_value);
this->__isset.end_value = true;
} else {
xfer += iprot->skip(ftype);
}
break;
default:
xfer += iprot->skip(ftype);
break;
}
xfer += iprot->readFieldEnd();
}
xfer += iprot->readStructEnd();
return xfer;
}
uint32_t Range::write(::apache::thrift::protocol::TProtocol* oprot) const {
uint32_t xfer = 0;
::apache::thrift::protocol::TOutputRecursionTracker tracker(*oprot);
xfer += oprot->writeStructBegin("Range");
xfer += oprot->writeFieldBegin("start_value", ::apache::thrift::protocol::T_STRING, 1);
xfer += oprot->writeString(this->start_value);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldBegin("end_value", ::apache::thrift::protocol::T_STRING, 2);
xfer += oprot->writeString(this->end_value);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldStop();
xfer += oprot->writeStructEnd();
return xfer;
}
void swap(Range &a, Range &b) {
using ::std::swap;
swap(a.start_value, b.start_value);
swap(a.end_value, b.end_value);
swap(a.__isset, b.__isset);
}
Range::Range(const Range& other5) {
start_value = other5.start_value;
end_value = other5.end_value;
__isset = other5.__isset;
}
Range& Range::operator=(const Range& other6) {
start_value = other6.start_value;
end_value = other6.end_value;
__isset = other6.__isset;
return *this;
}
void Range::printTo(std::ostream& out) const {
using ::apache::thrift::to_string;
out << "Range(";
out << "start_value=" << to_string(start_value);
out << ", " << "end_value=" << to_string(end_value);
out << ")";
}
RowRecord::~RowRecord() throw() {
}
void RowRecord::__set_vector_data(const std::string& val) {
this->vector_data = val;
}
std::ostream& operator<<(std::ostream& out, const RowRecord& obj)
{
obj.printTo(out);
return out;
}
uint32_t RowRecord::read(::apache::thrift::protocol::TProtocol* iprot) {
::apache::thrift::protocol::TInputRecursionTracker tracker(*iprot);
uint32_t xfer = 0;
std::string fname;
::apache::thrift::protocol::TType ftype;
int16_t fid;
xfer += iprot->readStructBegin(fname);
using ::apache::thrift::protocol::TProtocolException;
bool isset_vector_data = false;
while (true)
{
xfer += iprot->readFieldBegin(fname, ftype, fid);
if (ftype == ::apache::thrift::protocol::T_STOP) {
break;
}
switch (fid)
{
case 1:
if (ftype == ::apache::thrift::protocol::T_STRING) {
xfer += iprot->readBinary(this->vector_data);
isset_vector_data = true;
} else {
xfer += iprot->skip(ftype);
}
break;
default:
xfer += iprot->skip(ftype);
break;
}
xfer += iprot->readFieldEnd();
}
xfer += iprot->readStructEnd();
if (!isset_vector_data)
throw TProtocolException(TProtocolException::INVALID_DATA);
return xfer;
}
uint32_t RowRecord::write(::apache::thrift::protocol::TProtocol* oprot) const {
uint32_t xfer = 0;
::apache::thrift::protocol::TOutputRecursionTracker tracker(*oprot);
xfer += oprot->writeStructBegin("RowRecord");
xfer += oprot->writeFieldBegin("vector_data", ::apache::thrift::protocol::T_STRING, 1);
xfer += oprot->writeBinary(this->vector_data);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldStop();
xfer += oprot->writeStructEnd();
return xfer;
}
void swap(RowRecord &a, RowRecord &b) {
using ::std::swap;
swap(a.vector_data, b.vector_data);
}
RowRecord::RowRecord(const RowRecord& other7) {
vector_data = other7.vector_data;
}
RowRecord& RowRecord::operator=(const RowRecord& other8) {
vector_data = other8.vector_data;
return *this;
}
void RowRecord::printTo(std::ostream& out) const {
using ::apache::thrift::to_string;
out << "RowRecord(";
out << "vector_data=" << to_string(vector_data);
out << ")";
}
QueryResult::~QueryResult() throw() {
}
void QueryResult::__set_id(const int64_t val) {
this->id = val;
}
void QueryResult::__set_distance(const double val) {
this->distance = val;
}
std::ostream& operator<<(std::ostream& out, const QueryResult& obj)
{
obj.printTo(out);
return out;
}
uint32_t QueryResult::read(::apache::thrift::protocol::TProtocol* iprot) {
::apache::thrift::protocol::TInputRecursionTracker tracker(*iprot);
uint32_t xfer = 0;
std::string fname;
::apache::thrift::protocol::TType ftype;
int16_t fid;
xfer += iprot->readStructBegin(fname);
using ::apache::thrift::protocol::TProtocolException;
while (true)
{
xfer += iprot->readFieldBegin(fname, ftype, fid);
if (ftype == ::apache::thrift::protocol::T_STOP) {
break;
}
switch (fid)
{
case 1:
if (ftype == ::apache::thrift::protocol::T_I64) {
xfer += iprot->readI64(this->id);
this->__isset.id = true;
} else {
xfer += iprot->skip(ftype);
}
break;
case 2:
if (ftype == ::apache::thrift::protocol::T_DOUBLE) {
xfer += iprot->readDouble(this->distance);
this->__isset.distance = true;
} else {
xfer += iprot->skip(ftype);
}
break;
default:
xfer += iprot->skip(ftype);
break;
}
xfer += iprot->readFieldEnd();
}
xfer += iprot->readStructEnd();
return xfer;
}
uint32_t QueryResult::write(::apache::thrift::protocol::TProtocol* oprot) const {
uint32_t xfer = 0;
::apache::thrift::protocol::TOutputRecursionTracker tracker(*oprot);
xfer += oprot->writeStructBegin("QueryResult");
xfer += oprot->writeFieldBegin("id", ::apache::thrift::protocol::T_I64, 1);
xfer += oprot->writeI64(this->id);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldBegin("distance", ::apache::thrift::protocol::T_DOUBLE, 2);
xfer += oprot->writeDouble(this->distance);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldStop();
xfer += oprot->writeStructEnd();
return xfer;
}
void swap(QueryResult &a, QueryResult &b) {
using ::std::swap;
swap(a.id, b.id);
swap(a.distance, b.distance);
swap(a.__isset, b.__isset);
}
QueryResult::QueryResult(const QueryResult& other9) {
id = other9.id;
distance = other9.distance;
__isset = other9.__isset;
}
QueryResult& QueryResult::operator=(const QueryResult& other10) {
id = other10.id;
distance = other10.distance;
__isset = other10.__isset;
return *this;
}
void QueryResult::printTo(std::ostream& out) const {
using ::apache::thrift::to_string;
out << "QueryResult(";
out << "id=" << to_string(id);
out << ", " << "distance=" << to_string(distance);
out << ")";
}
TopKQueryResult::~TopKQueryResult() throw() {
}
void TopKQueryResult::__set_query_result_arrays(const std::vector<QueryResult> & val) {
this->query_result_arrays = val;
}
std::ostream& operator<<(std::ostream& out, const TopKQueryResult& obj)
{
obj.printTo(out);
return out;
}
uint32_t TopKQueryResult::read(::apache::thrift::protocol::TProtocol* iprot) {
::apache::thrift::protocol::TInputRecursionTracker tracker(*iprot);
uint32_t xfer = 0;
std::string fname;
::apache::thrift::protocol::TType ftype;
int16_t fid;
xfer += iprot->readStructBegin(fname);
using ::apache::thrift::protocol::TProtocolException;
while (true)
{
xfer += iprot->readFieldBegin(fname, ftype, fid);
if (ftype == ::apache::thrift::protocol::T_STOP) {
break;
}
switch (fid)
{
case 1:
if (ftype == ::apache::thrift::protocol::T_LIST) {
{
this->query_result_arrays.clear();
uint32_t _size11;
::apache::thrift::protocol::TType _etype14;
xfer += iprot->readListBegin(_etype14, _size11);
this->query_result_arrays.resize(_size11);
uint32_t _i15;
for (_i15 = 0; _i15 < _size11; ++_i15)
{
xfer += this->query_result_arrays[_i15].read(iprot);
}
xfer += iprot->readListEnd();
}
this->__isset.query_result_arrays = true;
} else {
xfer += iprot->skip(ftype);
}
break;
default:
xfer += iprot->skip(ftype);
break;
}
xfer += iprot->readFieldEnd();
}
xfer += iprot->readStructEnd();
return xfer;
}
uint32_t TopKQueryResult::write(::apache::thrift::protocol::TProtocol* oprot) const {
uint32_t xfer = 0;
::apache::thrift::protocol::TOutputRecursionTracker tracker(*oprot);
xfer += oprot->writeStructBegin("TopKQueryResult");
xfer += oprot->writeFieldBegin("query_result_arrays", ::apache::thrift::protocol::T_LIST, 1);
{
xfer += oprot->writeListBegin(::apache::thrift::protocol::T_STRUCT, static_cast<uint32_t>(this->query_result_arrays.size()));
std::vector<QueryResult> ::const_iterator _iter16;
for (_iter16 = this->query_result_arrays.begin(); _iter16 != this->query_result_arrays.end(); ++_iter16)
{
xfer += (*_iter16).write(oprot);
}
xfer += oprot->writeListEnd();
}
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldStop();
xfer += oprot->writeStructEnd();
return xfer;
}
void swap(TopKQueryResult &a, TopKQueryResult &b) {
using ::std::swap;
swap(a.query_result_arrays, b.query_result_arrays);
swap(a.__isset, b.__isset);
}
TopKQueryResult::TopKQueryResult(const TopKQueryResult& other17) {
query_result_arrays = other17.query_result_arrays;
__isset = other17.__isset;
}
TopKQueryResult& TopKQueryResult::operator=(const TopKQueryResult& other18) {
query_result_arrays = other18.query_result_arrays;
__isset = other18.__isset;
return *this;
}
void TopKQueryResult::printTo(std::ostream& out) const {
using ::apache::thrift::to_string;
out << "TopKQueryResult(";
out << "query_result_arrays=" << to_string(query_result_arrays);
out << ")";
}
TopKQueryBinResult::~TopKQueryBinResult() throw() {
}
void TopKQueryBinResult::__set_id_array(const std::string& val) {
this->id_array = val;
}
void TopKQueryBinResult::__set_distance_array(const std::string& val) {
this->distance_array = val;
}
std::ostream& operator<<(std::ostream& out, const TopKQueryBinResult& obj)
{
obj.printTo(out);
return out;
}
uint32_t TopKQueryBinResult::read(::apache::thrift::protocol::TProtocol* iprot) {
::apache::thrift::protocol::TInputRecursionTracker tracker(*iprot);
uint32_t xfer = 0;
std::string fname;
::apache::thrift::protocol::TType ftype;
int16_t fid;
xfer += iprot->readStructBegin(fname);
using ::apache::thrift::protocol::TProtocolException;
bool isset_id_array = false;
bool isset_distance_array = false;
while (true)
{
xfer += iprot->readFieldBegin(fname, ftype, fid);
if (ftype == ::apache::thrift::protocol::T_STOP) {
break;
}
switch (fid)
{
case 1:
if (ftype == ::apache::thrift::protocol::T_STRING) {
xfer += iprot->readBinary(this->id_array);
isset_id_array = true;
} else {
xfer += iprot->skip(ftype);
}
break;
case 2:
if (ftype == ::apache::thrift::protocol::T_STRING) {
xfer += iprot->readBinary(this->distance_array);
isset_distance_array = true;
} else {
xfer += iprot->skip(ftype);
}
break;
default:
xfer += iprot->skip(ftype);
break;
}
xfer += iprot->readFieldEnd();
}
xfer += iprot->readStructEnd();
if (!isset_id_array)
throw TProtocolException(TProtocolException::INVALID_DATA);
if (!isset_distance_array)
throw TProtocolException(TProtocolException::INVALID_DATA);
return xfer;
}
uint32_t TopKQueryBinResult::write(::apache::thrift::protocol::TProtocol* oprot) const {
uint32_t xfer = 0;
::apache::thrift::protocol::TOutputRecursionTracker tracker(*oprot);
xfer += oprot->writeStructBegin("TopKQueryBinResult");
xfer += oprot->writeFieldBegin("id_array", ::apache::thrift::protocol::T_STRING, 1);
xfer += oprot->writeBinary(this->id_array);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldBegin("distance_array", ::apache::thrift::protocol::T_STRING, 2);
xfer += oprot->writeBinary(this->distance_array);
xfer += oprot->writeFieldEnd();
xfer += oprot->writeFieldStop();
xfer += oprot->writeStructEnd();
return xfer;
}
void swap(TopKQueryBinResult &a, TopKQueryBinResult &b) {
using ::std::swap;
swap(a.id_array, b.id_array);
swap(a.distance_array, b.distance_array);
}
TopKQueryBinResult::TopKQueryBinResult(const TopKQueryBinResult& other19) {
id_array = other19.id_array;
distance_array = other19.distance_array;
}
TopKQueryBinResult& TopKQueryBinResult::operator=(const TopKQueryBinResult& other20) {
id_array = other20.id_array;
distance_array = other20.distance_array;
return *this;
}
void TopKQueryBinResult::printTo(std::ostream& out) const {
using ::apache::thrift::to_string;
out << "TopKQueryBinResult(";
out << "id_array=" << to_string(id_array);
out << ", " << "distance_array=" << to_string(distance_array);
out << ")";
}
}} // namespace

394
cpp/src/thrift/gen-cpp/milvus_types.h
View File

@ -1,394 +0,0 @@
/**
* Autogenerated by Thrift Compiler (0.12.0)
*
* DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
* @generated
*/
#ifndef milvus_TYPES_H
#define milvus_TYPES_H
#include <iosfwd>
#include <thrift/Thrift.h>
#include <thrift/TApplicationException.h>
#include <thrift/TBase.h>
#include <thrift/protocol/TProtocol.h>
#include <thrift/transport/TTransport.h>
#include <thrift/stdcxx.h>
namespace milvus { namespace thrift {
struct ErrorCode {
enum type {
SUCCESS = 0,
UNEXPECTED_ERROR = 1,
CONNECT_FAILED = 2,
PERMISSION_DENIED = 3,
TABLE_NOT_EXISTS = 4,
ILLEGAL_ARGUMENT = 5,
ILLEGAL_RANGE = 6,
ILLEGAL_DIMENSION = 7,
ILLEGAL_INDEX_TYPE = 8,
ILLEGAL_TABLE_NAME = 9,
ILLEGAL_TOPK = 10,
ILLEGAL_ROWRECORD = 11,
ILLEGAL_VECTOR_ID = 12,
ILLEGAL_SEARCH_RESULT = 13,
FILE_NOT_FOUND = 14,
META_FAILED = 15,
CACHE_FAILED = 16,
CANNOT_CREATE_FOLDER = 17,
CANNOT_CREATE_FILE = 18,
CANNOT_DELETE_FOLDER = 19,
CANNOT_DELETE_FILE = 20,
BUILD_INDEX_ERROR = 21
};
};
extern const std::map<int, const char*> _ErrorCode_VALUES_TO_NAMES;
std::ostream& operator<<(std::ostream& out, const ErrorCode::type& val);
class Exception;
class TableSchema;
class Range;
class RowRecord;
class QueryResult;
class TopKQueryResult;
class TopKQueryBinResult;
typedef struct _Exception__isset {
_Exception__isset() : code(false), reason(false) {}
bool code :1;
bool reason :1;
} _Exception__isset;
class Exception : public ::apache::thrift::TException {
public:
Exception(const Exception&);
Exception& operator=(const Exception&);
Exception() : code((ErrorCode::type)0), reason() {
}
virtual ~Exception() throw();
ErrorCode::type code;
std::string reason;
_Exception__isset __isset;
void __set_code(const ErrorCode::type val);
void __set_reason(const std::string& val);
bool operator == (const Exception & rhs) const
{
if (!(code == rhs.code))
return false;
if (!(reason == rhs.reason))
return false;
return true;
}
bool operator != (const Exception &rhs) const {
return !(*this == rhs);
}
bool operator < (const Exception & ) const;
uint32_t read(::apache::thrift::protocol::TProtocol* iprot);
uint32_t write(::apache::thrift::protocol::TProtocol* oprot) const;
virtual void printTo(std::ostream& out) const;
mutable std::string thriftTExceptionMessageHolder_;
const char* what() const throw();
};
void swap(Exception &a, Exception &b);
std::ostream& operator<<(std::ostream& out, const Exception& obj);
typedef struct _TableSchema__isset {
_TableSchema__isset() : index_type(true), dimension(true), store_raw_vector(true) {}
bool index_type :1;
bool dimension :1;
bool store_raw_vector :1;
} _TableSchema__isset;
class TableSchema : public virtual ::apache::thrift::TBase {
public:
TableSchema(const TableSchema&);
TableSchema& operator=(const TableSchema&);
TableSchema() : table_name(), index_type(0), dimension(0LL), store_raw_vector(false) {
}
virtual ~TableSchema() throw();
std::string table_name;
int32_t index_type;
int64_t dimension;
bool store_raw_vector;
_TableSchema__isset __isset;
void __set_table_name(const std::string& val);
void __set_index_type(const int32_t val);
void __set_dimension(const int64_t val);
void __set_store_raw_vector(const bool val);
bool operator == (const TableSchema & rhs) const
{
if (!(table_name == rhs.table_name))
return false;
if (!(index_type == rhs.index_type))
return false;
if (!(dimension == rhs.dimension))
return false;
if (!(store_raw_vector == rhs.store_raw_vector))
return false;
return true;
}
bool operator != (const TableSchema &rhs) const {
return !(*this == rhs);
}
bool operator < (const TableSchema & ) const;
uint32_t read(::apache::thrift::protocol::TProtocol* iprot);
uint32_t write(::apache::thrift::protocol::TProtocol* oprot) const;
virtual void printTo(std::ostream& out) const;
};
void swap(TableSchema &a, TableSchema &b);
std::ostream& operator<<(std::ostream& out, const TableSchema& obj);
typedef struct _Range__isset {
_Range__isset() : start_value(false), end_value(false) {}
bool start_value :1;
bool end_value :1;
} _Range__isset;
class Range : public virtual ::apache::thrift::TBase {
public:
Range(const Range&);
Range& operator=(const Range&);
Range() : start_value(), end_value() {
}
virtual ~Range() throw();
std::string start_value;
std::string end_value;
_Range__isset __isset;
void __set_start_value(const std::string& val);
void __set_end_value(const std::string& val);
bool operator == (const Range & rhs) const
{
if (!(start_value == rhs.start_value))
return false;
if (!(end_value == rhs.end_value))
return false;
return true;
}
bool operator != (const Range &rhs) const {
return !(*this == rhs);
}
bool operator < (const Range & ) const;
uint32_t read(::apache::thrift::protocol::TProtocol* iprot);
uint32_t write(::apache::thrift::protocol::TProtocol* oprot) const;
virtual void printTo(std::ostream& out) const;
};
void swap(Range &a, Range &b);
std::ostream& operator<<(std::ostream& out, const Range& obj);
class RowRecord : public virtual ::apache::thrift::TBase {
public:
RowRecord(const RowRecord&);
RowRecord& operator=(const RowRecord&);
RowRecord() : vector_data() {
}
virtual ~RowRecord() throw();
std::string vector_data;
void __set_vector_data(const std::string& val);
bool operator == (const RowRecord & rhs) const
{
if (!(vector_data == rhs.vector_data))
return false;
return true;
}
bool operator != (const RowRecord &rhs) const {
return !(*this == rhs);
}
bool operator < (const RowRecord & ) const;
uint32_t read(::apache::thrift::protocol::TProtocol* iprot);
uint32_t write(::apache::thrift::protocol::TProtocol* oprot) const;
virtual void printTo(std::ostream& out) const;
};
void swap(RowRecord &a, RowRecord &b);
std::ostream& operator<<(std::ostream& out, const RowRecord& obj);
typedef struct _QueryResult__isset {
_QueryResult__isset() : id(false), distance(false) {}
bool id :1;
bool distance :1;
} _QueryResult__isset;
class QueryResult : public virtual ::apache::thrift::TBase {
public:
QueryResult(const QueryResult&);
QueryResult& operator=(const QueryResult&);
QueryResult() : id(0), distance(0) {
}
virtual ~QueryResult() throw();
int64_t id;
double distance;
_QueryResult__isset __isset;
void __set_id(const int64_t val);
void __set_distance(const double val);
bool operator == (const QueryResult & rhs) const
{
if (!(id == rhs.id))
return false;
if (!(distance == rhs.distance))
return false;
return true;
}
bool operator != (const QueryResult &rhs) const {
return !(*this == rhs);
}
bool operator < (const QueryResult & ) const;
uint32_t read(::apache::thrift::protocol::TProtocol* iprot);
uint32_t write(::apache::thrift::protocol::TProtocol* oprot) const;
virtual void printTo(std::ostream& out) const;
};
void swap(QueryResult &a, QueryResult &b);
std::ostream& operator<<(std::ostream& out, const QueryResult& obj);
typedef struct _TopKQueryResult__isset {
_TopKQueryResult__isset() : query_result_arrays(false) {}
bool query_result_arrays :1;
} _TopKQueryResult__isset;
class TopKQueryResult : public virtual ::apache::thrift::TBase {
public:
TopKQueryResult(const TopKQueryResult&);
TopKQueryResult& operator=(const TopKQueryResult&);
TopKQueryResult() {
}
virtual ~TopKQueryResult() throw();
std::vector<QueryResult> query_result_arrays;
_TopKQueryResult__isset __isset;
void __set_query_result_arrays(const std::vector<QueryResult> & val);
bool operator == (const TopKQueryResult & rhs) const
{
if (!(query_result_arrays == rhs.query_result_arrays))
return false;
return true;
}
bool operator != (const TopKQueryResult &rhs) const {
return !(*this == rhs);
}
bool operator < (const TopKQueryResult & ) const;
uint32_t read(::apache::thrift::protocol::TProtocol* iprot);
uint32_t write(::apache::thrift::protocol::TProtocol* oprot) const;
virtual void printTo(std::ostream& out) const;
};
void swap(TopKQueryResult &a, TopKQueryResult &b);
std::ostream& operator<<(std::ostream& out, const TopKQueryResult& obj);
class TopKQueryBinResult : public virtual ::apache::thrift::TBase {
public:
TopKQueryBinResult(const TopKQueryBinResult&);
TopKQueryBinResult& operator=(const TopKQueryBinResult&);
TopKQueryBinResult() : id_array(), distance_array() {
}
virtual ~TopKQueryBinResult() throw();
std::string id_array;
std::string distance_array;
void __set_id_array(const std::string& val);
void __set_distance_array(const std::string& val);
bool operator == (const TopKQueryBinResult & rhs) const
{
if (!(id_array == rhs.id_array))
return false;
if (!(distance_array == rhs.distance_array))
return false;
return true;
}
bool operator != (const TopKQueryBinResult &rhs) const {
return !(*this == rhs);
}
bool operator < (const TopKQueryBinResult & ) const;
uint32_t read(::apache::thrift::protocol::TProtocol* iprot);
uint32_t write(::apache::thrift::protocol::TProtocol* oprot) const;
virtual void printTo(std::ostream& out) const;
};
void swap(TopKQueryBinResult &a, TopKQueryBinResult &b);
std::ostream& operator<<(std::ostream& out, const TopKQueryBinResult& obj);
}} // namespace
#endif

246
cpp/src/thrift/milvus.thrift
View File

@ -1,246 +0,0 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
namespace cpp milvus.thrift
namespace py milvus.thrift
namespace d milvus.thrift
namespace dart milvus.thrift
namespace java milvus.thrift
namespace perl milvus.thrift
namespace php milvus.thrift
namespace haxe milvus.thrift
namespace netcore milvus.thrift
enum ErrorCode {
SUCCESS = 0,
UNEXPECTED_ERROR,
CONNECT_FAILED,
PERMISSION_DENIED,
TABLE_NOT_EXISTS,
ILLEGAL_ARGUMENT,
ILLEGAL_RANGE,
ILLEGAL_DIMENSION,
ILLEGAL_INDEX_TYPE,
ILLEGAL_TABLE_NAME,
ILLEGAL_TOPK,
ILLEGAL_ROWRECORD,
ILLEGAL_VECTOR_ID,
ILLEGAL_SEARCH_RESULT,
FILE_NOT_FOUND,
META_FAILED,
CACHE_FAILED,
CANNOT_CREATE_FOLDER,
CANNOT_CREATE_FILE,
CANNOT_DELETE_FOLDER,
CANNOT_DELETE_FILE,
BUILD_INDEX_ERROR,
}
exception Exception {
1: ErrorCode code;
2: string reason;
}
/**
* @brief Table Schema
*/
struct TableSchema {
1: required string table_name; ///< Table name
2: i32 index_type = 0; ///< Index type, optional: 0-invalid, 1-idmap, 2-ivflat
3: i64 dimension = 0; ///< Vector dimension
4: bool store_raw_vector = false; ///< Store raw data
}
/**
* @brief Range Schema
*/
struct Range {
1: string start_value; ///< Range start
2: string end_value; ///< Range stop
}
/**
* @brief Record inserted
*/
struct RowRecord {
1: required binary vector_data; ///< Vector data, double array
}
/**
* @brief Query result
*/
struct QueryResult {
1: i64 id; ///< Output result
2: double distance; ///< Vector similarity distance
}
/**
* @brief TopK query result
*/
struct TopKQueryResult {
1: list<QueryResult> query_result_arrays; ///< TopK query result
}
/**
* @brief TopK query binary result
*/
struct TopKQueryBinResult {
1: required binary id_array; ///< id array, interger array
2: required binary distance_array; ///< distance array, double array
}
service MilvusService {
/**
* @brief Create table method
*
* This method is used to create table
*
* @param param, use to provide table information to be created.
*
*/
void CreateTable(2: TableSchema param) throws(1: Exception e);
/**
* @brief Test table existence method
*
* This method is used to test table existence.
*
* @param table_name, table name is going to be tested.
*
*/
bool HasTable(2: string table_name) throws(1: Exception e);
/**
* @brief Delete table method
*
* This method is used to delete table.
*
* @param table_name, table name is going to be deleted.
*
*/
void DeleteTable(2: string table_name) throws(1: Exception e);
/**
* @brief Build index by table method
*
* This method is used to build index by table in sync mode.
*
* @param table_name, table is going to be built index.
*
*/
void BuildIndex(2: string table_name) throws(1: Exception e);
/**
* @brief Add vector array to table
*
* This method is used to add vector array to table.
*
* @param table_name, table_name is inserted.
* @param record_array, vector array is inserted.
*
* @return vector id array
*/
list<i64> AddVector(2: string table_name,
3: list<RowRecord> record_array) throws(1: Exception e);
/**
* @brief Query vector
*
* This method is used to query vector in table.
*
* @param table_name, table_name is queried.
* @param query_record_array, all vector are going to be queried.
* @param query_range_array, optional ranges for conditional search. If not specified, search whole table
* @param topk, how many similarity vectors will be searched.
*
* @return query result array.
*/
list<TopKQueryResult> SearchVector(2: string table_name,
3: list<RowRecord> query_record_array,
4: list<Range> query_range_array,
5: i64 topk) throws(1: Exception e);
/**
* @brief Query vector
*
* This method is used to query vector in table.
*
* @param table_name, table_name is queried.
* @param query_record_array, all vector are going to be queried.
* @param query_range_array, optional ranges for conditional search. If not specified, search whole table
* @param topk, how many similarity vectors will be searched.
*
* @return query binary result array.
*/
list<TopKQueryBinResult> SearchVector2(2: string table_name,
3: list<RowRecord> query_record_array,
4: list<Range> query_range_array,
5: i64 topk) throws(1: Exception e);
/**
* @brief Internal use query interface
*
* This method is used to query vector in specified files.
*
* @param file_id_array, specified files id array, queried.
* @param query_record_array, all vector are going to be queried.
* @param query_range_array, optional ranges for conditional search. If not specified, search whole table
* @param topk, how many similarity vectors will be searched.
*
* @return query result array.
*/
list<TopKQueryResult> SearchVectorInFiles(2: string table_name,
3: list<string> file_id_array,
4: list<RowRecord> query_record_array,
5: list<Range> query_range_array,
6: i64 topk) throws(1: Exception e);
/**
* @brief Get table schema
*
* This method is used to get table schema.
*
* @param table_name, target table name.
*
* @return table schema
*/
TableSchema DescribeTable(2: string table_name) throws(1: Exception e);
/**
* @brief Get table row count
*
* This method is used to get table row count.
*
* @param table_name, target table name.
*
* @return table row count
*/
i64 GetTableRowCount(2: string table_name) throws(1: Exception e);
/**
* @brief List all tables in database
*
* This method is used to list all tables.
*
*
* @return table names.
*/
list<string> ShowTables() throws(1: Exception e);
/**
* @brief Give the server status
*
* This method is used to give the server status.
*
* @return Server status.
*/
string Ping(2: string cmd) throws(1: Exception e);
}

1
cpp/thirdparty/versions.txt vendored
View File

@ -15,7 +15,6 @@ ROCKSDB_VERSION=v6.0.2
SNAPPY_VERSION=1.1.7
SQLITE_VERSION=3280000
SQLITE_ORM_VERSION=master
THRIFT_VERSION=v0.12.0
YAMLCPP_VERSION=0.6.2
ZLIB_VERSION=v1.2.11
ZSTD_VERSION=v1.4.0