diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 91d290811a..412b82217f 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -427,6 +427,24 @@ jobs: timeout: 20 MCL_RUNTIME_ARTIFACT_NAME: mcl_runtime_linux_debug_x86_64 + test_linux_debug_rt_indexer: + if: always() && (github.event_name == 'pull_request' && github.event.pull_request.base.ref == 'update-buddy-version') != true && needs.build_linux_debug.result == 'success' && (needs.mcl_runtime_linux_debug_cached.outputs.cache_hit == 'true' || needs.mcl_runtime_linux_debug.result == 'success') + name: Linux debug mode tests (rt-indexer) + needs: [build_linux_debug, mcl_runtime_linux_debug_cached, mcl_runtime_linux_debug, meta] + permissions: + checks: write + contents: read + uses: ./.github/workflows/test_template.yml + secrets: inherit + with: + build_artifact_name: debug_build + artifact_name: debug_test_results_rt_indexer + results_name: "Linux debug RT indexer test results" + suite_name: rt-indexer + CTEST_REGEX: "^(rtidx_|Indexer RT bulk)" + timeout: 35 + MCL_RUNTIME_ARTIFACT_NAME: mcl_runtime_linux_debug_x86_64 + test_linux_debug_mcl: if: always() && (github.event_name == 'pull_request' && github.event.pull_request.base.ref == 'update-buddy-version') != true && needs.build_linux_debug.result == 'success' && (needs.mcl_runtime_linux_debug_cached.outputs.cache_hit == 'true' || needs.mcl_runtime_linux_debug.result == 'success') name: Linux debug mode tests (MCL) @@ -833,6 +851,24 @@ jobs: timeout: 15 MCL_RUNTIME_ARTIFACT_NAME: mcl_runtime_linux_release_x86_64 + test_linux_release_rt_indexer: + if: always() && (github.event_name == 'pull_request' && github.event.pull_request.base.ref == 'update-buddy-version') != true && needs.build_linux_release_tests.result == 'success' && (needs.mcl_runtime_linux_release_cached.outputs.cache_hit == 'true' || needs.mcl_runtime_linux_release.result == 'success') + name: Linux release mode tests (rt-indexer) + needs: [build_linux_release_tests, mcl_runtime_linux_release_cached, mcl_runtime_linux_release, meta] + permissions: + checks: write + contents: read + uses: ./.github/workflows/test_template.yml + secrets: inherit + with: + build_artifact_name: release_test_build + artifact_name: release_test_results_rt_indexer + results_name: "Linux release RT indexer test results" + suite_name: rt-indexer + CTEST_REGEX: "^(rtidx_|Indexer RT bulk)" + timeout: 15 + MCL_RUNTIME_ARTIFACT_NAME: mcl_runtime_linux_release_x86_64 + test_linux_release_mcl: if: always() && (github.event_name == 'pull_request' && github.event.pull_request.base.ref == 'update-buddy-version') != true && needs.build_linux_release_tests.result == 'success' && (needs.mcl_runtime_linux_release_cached.outputs.cache_hit == 'true' || needs.mcl_runtime_linux_release.result == 'success') name: Linux release mode tests (MCL) diff --git a/manual/english/Data_creation_and_modification/Adding_documents_to_a_table/Adding_documents_to_a_real-time_table.md b/manual/english/Data_creation_and_modification/Adding_documents_to_a_table/Adding_documents_to_a_real-time_table.md index 5fe99e6899..00260eacee 100755 --- a/manual/english/Data_creation_and_modification/Adding_documents_to_a_table/Adding_documents_to_a_real-time_table.md +++ b/manual/english/Data_creation_and_modification/Adding_documents_to_a_table/Adding_documents_to_a_real-time_table.md @@ -846,7 +846,6 @@ CALL UUID_SHORT(3) ``` - ## Bulk adding documents You can insert not just a single document into a real-time table, but as many as you'd like. It's perfectly fine to insert batches of tens of thousands of documents into a real-time table. However, it's important to keep the following points in mind: * The larger the batch, the higher the latency of each insert operation @@ -877,6 +876,98 @@ The `/bulk` (Manticore mode) endpoint supports [Chunked transfer encoding](https * decreases response time * allows you to bypass [max_packet_size](../../Server_settings/Searchd.md#max_packet_size) and transfer batches much larger than the maximum allowed value of `max_packet_size` (128MB), for example, 1GB at a time. +### Indexer-assisted bulk insertion + +> NOTE: This insertion mode is experimental. + +For large append-only loads into a local real-time table, Manticore Search can stream `INSERT` rows through `indexer`, build one disk chunk, and attach that chunk to the table when the transaction commits. This avoids adding every row through the regular real-time insertion path. Rows remain invisible until the chunk is attached, and a failed operation or `ROLLBACK` leaves the table unchanged. + +Use this mode when loading a large batch for which producing a disk chunk directly is preferable to building a RAM chunk first. It supports both row-wise and columnar tables, including full-text fields, numeric attributes, strings, JSON, MVA/MVA64, and float vectors with KNN indexes. + +#### SQL + +Enable the mode for the current SQL session, start an explicit transaction, run one or more `INSERT` statements against the same table, and commit: + +```sql +SET indexer_rt_bulk=1; +BEGIN; +INSERT INTO products(id,title,price) VALUES + (101,'Crossbody Bag with Tassel',19.85), + (102,'Microfiber Sheet Set',19.99); +INSERT INTO products(id,title,price) VALUES + (103,'Pet Hair Remover Glove',7.99); +COMMIT; +SET indexer_rt_bulk=0; +``` + +`SET indexer_rt_bulk=1` applies to the current connection. While it is enabled, every `INSERT` requires an active transaction started with `BEGIN` or `START TRANSACTION`. `COMMIT` builds and atomically attaches the disk chunk; `ROLLBACK`, a failed insert, or closing the connection discards the staged batch. + +#### HTTP `/bulk` + +Add `indexer_rt_bulk=1` to the `/bulk` query string. The request body remains standard newline-delimited JSON (NDJSON), and each operation must be `insert`: + +```bash +curl -X POST 'http://localhost:9308/bulk?indexer_rt_bulk=1' \ + -H 'Content-Type: application/x-ndjson' \ + --data-binary $'{"insert":{"table":"products","id":101,"doc":{"title":"Crossbody Bag with Tassel","price":19.85}}}\n{"insert":{"table":"products","id":102,"doc":{"title":"Microfiber Sheet Set","price":19.99}}}\n' +``` + +The endpoint also supports chunked transfer encoding in this mode, so the server can stream a request larger than `max_packet_size` into `indexer` without buffering the complete request body. As with ordinary `/bulk` transactions, changing the target table or adding an empty line commits the preceding group; the entire multi-table request is therefore not one atomic transaction. + +#### Fluent Bit + +Fluent Bit's Elasticsearch output sends data to `/_bulk` rather than `/bulk`. To enable indexer-assisted insertion for that output, set its `pipeline` option to the reserved value `indexer_rt_bulk`, use the `index` write operation, and provide the name of a record field that contains the document ID: + +```ini +[OUTPUT] + name es + match site_access_logs + host manticore + port 9308 + index site_access_logs + pipeline indexer_rt_bulk + write_operation index + id_key id +``` + +Every record's `id` value must be a non-zero decimal string, for example `"123"`. Fluent Bit does not add `_id` when `id_key` refers to a numeric value, and `generate_id On` produces UUID strings, which this experimental mode does not support. If the source does not already contain a suitable stable numeric string, add one with a Fluent Bit filter before the output runs. IDs must remain stable when Fluent Bit retries a chunk. For a single tailed file, one option is to expose its byte offset and convert it to a string with a Lua filter: + +```ini +[INPUT] + name tail + path /var/log/example.log + offset_key fluent_bit_offset + +[FILTER] + name lua + match site_access_logs + script add_numeric_id.lua + call add_numeric_id +``` + +```lua +function add_numeric_id(tag, timestamp, record) + record["id"] = tostring(record["fluent_bit_offset"] + 1) + record["fluent_bit_offset"] = nil + return 2, timestamp, record +end +``` + +Offsets are unique only within one file. Pipelines that tail multiple files should derive a collision-free numeric ID from a stable source identifier and offset instead. + +With `pipeline indexer_rt_bulk`, each Fluent Bit request is committed as one disk chunk. The `index` action replaces a document when its ID already exists in the table. IDs must be unique within each request; duplicate IDs in one request and other bulk actions are rejected. If any record in the request is invalid, Manticore Search rejects the request and attaches none of its records. Successful assisted responses for this reserved Fluent Bit pipeline contain `"errors": false` and an empty `items` array to avoid serializing, transferring, and parsing one redundant success object per record. Error responses retain per-record details. + +Without `pipeline indexer_rt_bulk`, the same output continues to use regular real-time insertion. The assisted mode is most useful when bulk success responses or indexing work are a bottleneck. Fluent Bit flushes independent chunks, and assisted mode starts `indexer` for each request, so small batches on a low-latency connection may not improve wall-clock time. Benchmark with representative records, schema, chunking, workers, and network conditions before enabling it in production. + +#### Current limitations + +* The target must be one local real-time table per transaction. Distributed, sharded, replicated, percolate, and plain tables are not supported. +* Only `INSERT` is supported. `REPLACE`, `UPDATE`, and `DELETE` are rejected. +* Every row must provide an explicit numeric, non-zero document ID. Auto-generated and UUID document IDs are not supported. +* The feature is unavailable on Windows and in static builds. +* The `indexer` executable from the same Manticore Search build or installation must be next to the running `searchd` executable. Manticore Search resolves only that sibling executable; it does not search `PATH`. If it is missing or cannot be executed, only the assisted insertion fails—normal startup and regular insertion remain available. + + ### Bulk insert examples ##### SQL: diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 504b7756ff..3f864cd571 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -193,7 +193,7 @@ diag ( LMANTICORE_BISON SEARCHD_BISON LMANTICORE_FLEX SEARCHD_FLEX ) target_sources ( lmanticore PUBLIC ${LMANTICORE_BISON} ${LMANTICORE_FLEX} ${HEADERS} ${CHARSET_FILES} ${CHARSET_TEMPLATE} "../misc/manticore.natvis" ) target_link_libraries ( lmanticore PUBLIC uni-algo::uni-algo ) -add_library ( lsearchd OBJECT searchdha.cpp api_reply_stream.cpp http/http_parser.c searchdhttp.cpp +add_library ( lsearchd OBJECT searchdha.cpp api_reply_stream.cpp http/http_parser.c searchdhttp.cpp indexer_rt_bulk.cpp searchdtask.cpp taskping.cpp taskmalloctrim.cpp taskglobalidf.cpp tasksavestate.cpp taskflushbinlog.cpp taskflushattrs.cpp taskflushmutable.cpp taskpreread.cpp searchdaemon.cpp searchdfields.cpp searchdconfig.cpp @@ -207,6 +207,7 @@ add_library ( lsearchd OBJECT searchdha.cpp api_reply_stream.cpp http/http_parse facetutils.cpp auth/auth.cpp auth/auth_common.cpp auth/auth_proto_http.cpp auth/auth_proto_mysql.cpp auth/auth_proto_api.cpp auth/auth_perms.cpp auth/gcm_nonce.cpp auth/auth_log.cpp auth/auth_bootstrap.cpp) +target_compile_definitions ( lsearchd PRIVATE STATIC_BINARY=$ ) target_sources ( lsearchd PUBLIC ${SEARCHD_SRCS_TESTABLE} ${SEARCHD_H} ${SEARCHD_BISON} ${SEARCHD_FLEX} ) add_library ( digest_sha1 digest_sha1.cpp ) add_library ( digest_sha256 digest_sha256.cpp ) @@ -436,6 +437,20 @@ if (NOT TEST_SPECIAL_EXTERNAL) COMMAND ${PYTHON_EXECUTABLE} reserved.py ) SET_TESTS_PROPERTIES ( ${RESERVED_TEST} PROPERTIES LABELS LINTER ) endif () + + if (PYTHONINTERP_FOUND AND NOT STATIC_BINARY) + fixup_test_name ( INDEXER_RT_MISSING_SIBLING_TEST "Indexer RT bulk missing sibling" ) + add_test ( NAME ${INDEXER_RT_MISSING_SIBLING_TEST} + COMMAND ${PYTHON_EXECUTABLE} ${MANTICORE_SOURCE_DIR}/test/integration/indexer_rt_bulk_missing_sibling_test.py + --searchd $ ) + SET_TESTS_PROPERTIES ( ${INDEXER_RT_MISSING_SIBLING_TEST} PROPERTIES LABELS RT_INDEXER RUN_SERIAL TRUE ) + + fixup_test_name ( INDEXER_RT_FLUENTBIT_TEST "Indexer RT bulk Fluent Bit" ) + add_test ( NAME ${INDEXER_RT_FLUENTBIT_TEST} + COMMAND ${PYTHON_EXECUTABLE} ${MANTICORE_SOURCE_DIR}/test/integration/indexer_rt_bulk_fluentbit_test.py + --searchd $ ) + SET_TESTS_PROPERTIES ( ${INDEXER_RT_FLUENTBIT_TEST} PROPERTIES LABELS RT_INDEXER RUN_SERIAL TRUE ) + endif () endif () # fixup_test_name ( tst "Internal src/tests" ) diff --git a/src/client_session.h b/src/client_session.h index 751610bb9b..21f21da959 100644 --- a/src/client_session.h +++ b/src/client_session.h @@ -17,6 +17,7 @@ #include "queryprofile.h" #include "searchdaemon.h" #include "searchdsql.h" +#include #include "searchd_shard.h" #include "sphinxpq.h" @@ -54,6 +55,15 @@ class ClientSession_c final public: bool m_bAutoCommit = true; bool m_bInTransaction = false; + bool m_bIndexerRtBulk = false; + FILE * m_pIndexerRtBulkStream = nullptr; + std::unique_ptr m_pIndexerRtBulkBuffer; + int m_iIndexerRtBulkPid = -1; + int64_t m_iIndexerRtBulkIndexId = -1; + CSphString m_sIndexerRtBulkTable; + CSphString m_sIndexerRtBulkDir; + CSphString m_sIndexerRtBulkConfig; + CSphString m_sIndexerRtBulkIndex; CSphVector m_dLastIds; CSphVector m_dLastIdStrings; QueryProfile_c m_tProfile; diff --git a/src/fileutils.cpp b/src/fileutils.cpp index 35d8ed0439..4cd2a41e82 100644 --- a/src/fileutils.cpp +++ b/src/fileutils.cpp @@ -15,6 +15,10 @@ #include "std/crc32.h" #include +#if defined(__APPLE__) + #include +#endif + #if _WIN32 #define getcwd _getcwd #include @@ -976,6 +980,23 @@ CSphString GetExecutablePath() CHAR szPath[MAX_PATH]; GetModuleFileName ( hModule, szPath, MAX_PATH ); return szPath; +#elif defined(__APPLE__) + uint32_t uPathSize = 0; + if ( _NSGetExecutablePath ( nullptr, &uPathSize )!=-1 || !uPathSize ) + return ""; + + CSphVector dPath; + dPath.Resize ( uPathSize ); + if ( _NSGetExecutablePath ( dPath.Begin(), &uPathSize ) ) + return ""; + + char * szResolved = realpath ( dPath.Begin(), nullptr ); + if ( !szResolved ) + return dPath.Begin(); + + CSphString sExecutable = szResolved; + free ( szResolved ); + return sExecutable; #else char szPath[PATH_MAX]; ssize_t tLen; diff --git a/src/gtests/gtests_functions.cpp b/src/gtests/gtests_functions.cpp index d0e081907c..6cca85b394 100644 --- a/src/gtests/gtests_functions.cpp +++ b/src/gtests/gtests_functions.cpp @@ -20,6 +20,7 @@ #include "histogram.h" #include "conversion.h" #include "digest_sha1.h" +#include "fileutils.h" #include "std/openhash.h" // Miscelaneous short functional tests: TDigest, SpanSearch, @@ -1601,6 +1602,19 @@ TEST ( functions, path ) ASSERT_STREQ ( sFile14.cstr(), "pq2" ); } + +TEST ( functions, executable_path ) +{ + CSphString sExecutable = GetExecutablePath(); + ASSERT_FALSE ( sExecutable.IsEmpty() ); + ASSERT_TRUE ( IsPathAbsolute ( sExecutable ) ) << sExecutable.cstr(); +#if _WIN32 + ASSERT_STREQ ( GetBaseName ( sExecutable ), "gmanticoretest.exe" ); +#else + ASSERT_STREQ ( GetBaseName ( sExecutable ), "gmanticoretest" ); +#endif +} + TEST ( functions, IsTriviallyCopyable ) { EXPECT_TRUE ( IS_TRIVIALLY_COPYABLE ( DWORD ) ) << "DWORD"; diff --git a/src/indexer_rt_bulk.cpp b/src/indexer_rt_bulk.cpp new file mode 100644 index 0000000000..3a984596ac --- /dev/null +++ b/src/indexer_rt_bulk.cpp @@ -0,0 +1,689 @@ +#if defined(__linux__) && !defined(_GNU_SOURCE) +#define _GNU_SOURCE +#endif + +// +// Copyright (c) 2017-2026, Manticore Software LTD +// +// Experimental indexer-assisted RT bulk loader (dev#2761). +// + +#include "indexer_rt_bulk.h" + +#include "client_session.h" +#include "fileutils.h" +#include "indexfiles.h" +#include "indexsettings.h" +#include "knnmisc.h" +#include "searchdaemon.h" +#include "searchdsql.h" +#include "sphinxrt.h" +#include "threadutils.h" + +#include +#include + +#if !_WIN32 +#include +#include +#include +#include +#include +#include + +extern char ** environ; +#endif + +bool AttachIndexerRtBulkChunk ( const CSphString & sTable, int64_t iIndexId, const CSphString & sPath, CSphString & sError ); + + +static bool ValidateFloatVectorValue ( const CSphColumnInfo & tAttr, const SqlInsert_t & tValue, int iRow, CSphString & sError ) +{ + if ( tValue.m_iType==SqlInsert_t::TOK_NULL ) + return true; + + if ( tValue.m_iType!=SqlInsert_t::CONST_MVA || !tValue.m_pVals ) + { + sError.SetSprintf ( "row %d, attribute '%s': float_vector requires a tuple value", iRow+1, tAttr.m_sName.cstr() ); + return false; + } + + const auto & dValues = *tValue.m_pVals; + if ( tAttr.IsIndexedKNN() && dValues.GetLength()!=tAttr.m_tKNN.m_iDims ) + { + sError.SetSprintf ( "row %d, attribute '%s': KNN index requires %d vector entries; %d specified", iRow+1, tAttr.m_sName.cstr(), tAttr.m_tKNN.m_iDims, dValues.GetLength() ); + return false; + } + + for ( const auto & tItem : dValues ) + if ( !std::isfinite ( tItem.m_fValue ) ) + { + sError.SetSprintf ( "row %d, attribute '%s': float_vector entries must be finite", iRow+1, tAttr.m_sName.cstr() ); + return false; + } + + return true; +} + +static void AppendCsvEscaped ( StringBuilder_c & sOut, const char * szValue ) +{ + const char * pStart = szValue ? szValue : ""; + for ( const char * p = pStart; ; ++p ) + { + if ( *p!='"' && *p!='\0' ) + continue; + sOut.AppendRawChunk ( Str_t { pStart, int ( p-pStart ) } ); + if ( !*p ) + break; + sOut.AppendRawChunk ( Str_t { "\"\"", 2 } ); + pStart = p+1; + } +} + + +static void AppendCsvQuoted ( StringBuilder_c & sOut, const char * szValue ) +{ + sOut << '"'; + AppendCsvEscaped ( sOut, szValue ); + sOut << '"'; +} + + +static bool GetIndexerPath ( CSphString & sIndexer, CSphString & sError ) +{ +#if STATIC_BINARY + sError = "indexer RT bulk is unavailable in static builds"; + return false; +#else + CSphString sExecutable = GetExecutablePath(); + if ( sExecutable.IsEmpty() ) + { + sError = "failed to locate the running searchd executable for indexer RT bulk"; + return false; + } + + sIndexer.SetSprintf ( "%sindexer", GetPathOnly ( sExecutable ).cstr() ); + return true; +#endif +} + + +#if !_WIN32 +static CSphString GetIndexerRtBulkOutput ( const ClientSession_c & tSession ) +{ + CSphString sPath; + sPath.SetSprintf ( "%s/indexer.log", tSession.m_sIndexerRtBulkDir.cstr() ); + FILE * fp = fopen ( sPath.cstr(), "rb" ); + if ( !fp ) + return {}; + + char sBuffer[4097]; + size_t iLength = fread ( sBuffer, 1, sizeof(sBuffer)-1, fp ); + fclose ( fp ); + while ( iLength && sphIsSpace ( sBuffer[iLength-1] ) ) + --iLength; + sBuffer[iLength] = '\0'; + return CSphString ( sBuffer ); +} + + +static bool WaitIndexerRtBulk ( ClientSession_c & tSession, CSphString & sError ) +{ + if ( tSession.m_iIndexerRtBulkPid<0 ) + return true; + + int iStatus = 0; + pid_t iResult; + do + { + iResult = waitpid ( tSession.m_iIndexerRtBulkPid, &iStatus, 0 ); + } while ( iResult<0 && errno==EINTR ); + tSession.m_iIndexerRtBulkPid = -1; + + if ( iResult<0 ) + { + sError.SetSprintf ( "failed waiting for indexer RT bulk process: %s", strerrorm ( errno ) ); + return false; + } + if ( WIFEXITED ( iStatus ) && WEXITSTATUS ( iStatus )==0 ) + return true; + if ( WIFEXITED ( iStatus ) ) + sError.SetSprintf ( "indexer RT bulk build failed with status %d", WEXITSTATUS ( iStatus ) ); + else if ( WIFSIGNALED ( iStatus ) ) + sError.SetSprintf ( "indexer RT bulk build was killed by signal %d", WTERMSIG ( iStatus ) ); + else + sError = "indexer RT bulk build failed"; + CSphString sOutput = GetIndexerRtBulkOutput ( tSession ); + if ( !sOutput.IsEmpty() ) + { + CSphString sStatus = sError; + sError.SetSprintf ( "%s: %s", sStatus.cstr(), sOutput.cstr() ); + } + return false; +} + + +static void StopIndexerRtBulk ( ClientSession_c & tSession ) +{ + if ( tSession.m_iIndexerRtBulkPid<0 ) + return; + + // The indexer's csvpipe command is its child, so terminate the entire process group. + const pid_t iPid = tSession.m_iIndexerRtBulkPid; + kill ( -iPid, SIGTERM ); + kill ( iPid, SIGTERM ); + int iStatus = 0; + for ( int i=0; i<100; ++i ) + { + pid_t iResult = waitpid ( iPid, &iStatus, WNOHANG ); + if ( iResult==iPid || ( iResult<0 && errno==ECHILD ) ) + { + tSession.m_iIndexerRtBulkPid = -1; + return; + } + if ( iResult<0 && errno!=EINTR ) + break; + usleep ( 10000 ); + } + + kill ( -iPid, SIGKILL ); + kill ( iPid, SIGKILL ); + while ( waitpid ( iPid, &iStatus, 0 )<0 && errno==EINTR ) {} + tSession.m_iIndexerRtBulkPid = -1; +} + + +static bool StartIndexerRtBulk ( ClientSession_c & tSession, CSphString & sError ) +{ + CSphString sIndexer; + if ( !GetIndexerPath ( sIndexer, sError ) ) + return false; + + int dSockets[2] = { -1, -1 }; + int iSocketType = SOCK_STREAM; + #ifdef SOCK_CLOEXEC + iSocketType |= SOCK_CLOEXEC; + #endif + if ( socketpair ( AF_UNIX, iSocketType, 0, dSockets )<0 ) + { + sError.SetSprintf ( "failed to create indexer RT bulk stream: %s", strerrorm ( errno ) ); + return false; + } + #ifndef SOCK_CLOEXEC + if ( fcntl ( dSockets[0], F_SETFD, FD_CLOEXEC )<0 || fcntl ( dSockets[1], F_SETFD, FD_CLOEXEC )<0 ) + { + close ( dSockets[0] ); + close ( dSockets[1] ); + sError.SetSprintf ( "failed to protect indexer RT bulk stream descriptors: %s", strerrorm ( errno ) ); + return false; + } + #endif + + const char * szIndexer = sIndexer.cstr(); + const char * szConfig = tSession.m_sIndexerRtBulkConfig.cstr(); + CSphString sOutput; + sOutput.SetSprintf ( "%s/indexer.log", tSession.m_sIndexerRtBulkDir.cstr() ); + int iOutput = open ( sOutput.cstr(), O_WRONLY | O_CREAT | O_TRUNC, 0600 ); + if ( iOutput<0 ) + { + close ( dSockets[0] ); + close ( dSockets[1] ); + sError.SetSprintf ( "failed to create indexer RT bulk log '%s': %s", sOutput.cstr(), strerrorm ( errno ) ); + return false; + } + posix_spawn_file_actions_t tActions; + posix_spawnattr_t tAttrs; + bool bActionsInit = false; + bool bAttrsInit = false; + int iSpawnError = posix_spawn_file_actions_init ( &tActions ); + if ( !iSpawnError ) + { + bActionsInit = true; + iSpawnError = posix_spawn_file_actions_adddup2 ( &tActions, dSockets[1], STDIN_FILENO ); + } + if ( !iSpawnError && dSockets[0]!=STDIN_FILENO ) + iSpawnError = posix_spawn_file_actions_addclose ( &tActions, dSockets[0] ); + if ( !iSpawnError && dSockets[1]!=STDIN_FILENO ) + iSpawnError = posix_spawn_file_actions_addclose ( &tActions, dSockets[1] ); + if ( !iSpawnError ) + iSpawnError = posix_spawn_file_actions_adddup2 ( &tActions, iOutput, STDOUT_FILENO ); + if ( !iSpawnError ) + iSpawnError = posix_spawn_file_actions_adddup2 ( &tActions, iOutput, STDERR_FILENO ); + if ( !iSpawnError && iOutput!=STDOUT_FILENO && iOutput!=STDERR_FILENO ) + iSpawnError = posix_spawn_file_actions_addclose ( &tActions, iOutput ); + #if defined(__linux__) + if ( !iSpawnError ) + iSpawnError = posix_spawn_file_actions_addclosefrom_np ( &tActions, STDERR_FILENO+1 ); + #endif + if ( !iSpawnError ) + { + iSpawnError = posix_spawnattr_init ( &tAttrs ); + bAttrsInit = !iSpawnError; + } + if ( !iSpawnError ) + { + sigset_t tSignals; + sigemptyset ( &tSignals ); + iSpawnError = posix_spawnattr_setsigmask ( &tAttrs, &tSignals ); + if ( !iSpawnError ) + iSpawnError = posix_spawnattr_setpgroup ( &tAttrs, 0 ); + if ( !iSpawnError ) + { + short iFlags = POSIX_SPAWN_SETPGROUP | POSIX_SPAWN_SETSIGMASK; + #ifdef POSIX_SPAWN_CLOEXEC_DEFAULT + iFlags |= POSIX_SPAWN_CLOEXEC_DEFAULT; + #endif + iSpawnError = posix_spawnattr_setflags ( &tAttrs, iFlags ); + } + } + + pid_t iPid = -1; + if ( !iSpawnError ) + { + char sConfigArg[] = "--config"; + char sIndexArg[] = "indexer_rt_bulk_chunk"; + char * dArgv[] = { const_cast( szIndexer ), sConfigArg, const_cast( szConfig ), sIndexArg, nullptr }; + iSpawnError = posix_spawn ( &iPid, szIndexer, &tActions, &tAttrs, dArgv, environ ); + } + if ( bAttrsInit ) + posix_spawnattr_destroy ( &tAttrs ); + if ( bActionsInit ) + posix_spawn_file_actions_destroy ( &tActions ); + close ( iOutput ); + close ( dSockets[1] ); + if ( iSpawnError ) + { + close ( dSockets[0] ); + sError.SetSprintf ( "failed to start indexer RT bulk process '%s': %s", szIndexer, strerrorm ( iSpawnError ) ); + return false; + } + + tSession.m_iIndexerRtBulkPid = iPid; + tSession.m_pIndexerRtBulkStream = fdopen ( dSockets[0], "wb" ); + if ( !tSession.m_pIndexerRtBulkStream ) + { + int iFdopenErrno = errno; + close ( dSockets[0] ); + StopIndexerRtBulk ( tSession ); + sError.SetSprintf ( "failed to open indexer RT bulk stream: %s", strerrorm ( iFdopenErrno ) ); + return false; + } + constexpr size_t iBufferSize = 256*1024; + tSession.m_pIndexerRtBulkBuffer = std::make_unique ( iBufferSize ); + if ( setvbuf ( tSession.m_pIndexerRtBulkStream, tSession.m_pIndexerRtBulkBuffer.get(), _IOFBF, iBufferSize ) ) + { + fclose ( tSession.m_pIndexerRtBulkStream ); + tSession.m_pIndexerRtBulkStream = nullptr; + tSession.m_pIndexerRtBulkBuffer.reset(); + StopIndexerRtBulk ( tSession ); + sError = "failed to configure indexer RT bulk stream buffering"; + return false; + } + return true; +} +#endif + + +static void AppendInsertValueToCsv ( StringBuilder_c & sOut, const SqlInsert_t & tValue, ESphAttr eType, bool bDocid ) +{ + sOut << '"'; + switch ( tValue.m_iType ) + { + case SqlInsert_t::QUOTED_STRING: + AppendCsvEscaped ( sOut, tValue.m_sVal.cstr() ); + break; + + case SqlInsert_t::CONST_FLOAT: + sOut.Appendf ( "%.9g", tValue.m_fVal ); + break; + + case SqlInsert_t::CONST_INT: + if ( bDocid ) + sOut << tValue.GetValueUint(); + else + sOut << tValue.GetValueInt(); + break; + + case SqlInsert_t::CONST_MVA: + if ( tValue.m_pVals ) + { + bool bFirst = true; + for ( const auto & tItem : *tValue.m_pVals ) + { + if ( !bFirst ) + sOut.RawC ( ' ' ); + bFirst = false; + if ( eType==SPH_ATTR_FLOAT_VECTOR ) + sOut.Appendf ( "%.9g", tItem.m_fValue ); + else + sOut << tItem.m_iValue; + } + } + break; + + case SqlInsert_t::TOK_NULL: + break; + + default: + AppendCsvEscaped ( sOut, tValue.m_sVal.cstr() ); + break; + } + sOut << '"'; +} + + +static int FindInsertColumn ( const SqlStmt_t & tStmt, const CSphString & sName ) +{ + return tStmt.m_dInsertSchema.GetFirst ( [&sName] ( const CSphString & sColumn ) { return sColumn==sName; } ); +} + + +static const char * CsvAttrDirective ( ESphAttr eType ) +{ + switch ( eType ) + { + case SPH_ATTR_INTEGER: return "csvpipe_attr_uint"; + case SPH_ATTR_TIMESTAMP: return "csvpipe_attr_timestamp"; + case SPH_ATTR_BOOL: return "csvpipe_attr_bool"; + case SPH_ATTR_FLOAT: return "csvpipe_attr_float"; + case SPH_ATTR_BIGINT: return "csvpipe_attr_bigint"; + case SPH_ATTR_STRING: return "csvpipe_attr_string"; + case SPH_ATTR_JSON: return "csvpipe_attr_json"; + case SPH_ATTR_UINT32SET: return "csvpipe_attr_multi"; + case SPH_ATTR_INT64SET: return "csvpipe_attr_multi_64"; + case SPH_ATTR_FLOAT_VECTOR: return "csvpipe_attr_float_vector"; + default: return nullptr; + } +} + + +static constexpr ESphAttr g_dCsvAttrOrder[] = +{ + SPH_ATTR_INTEGER, SPH_ATTR_TIMESTAMP, SPH_ATTR_BOOL, SPH_ATTR_FLOAT, SPH_ATTR_BIGINT, + SPH_ATTR_UINT32SET, SPH_ATTR_INT64SET, SPH_ATTR_FLOAT_VECTOR, SPH_ATTR_STRING, SPH_ATTR_JSON +}; + + +void CleanupIndexerRtBulk ( ClientSession_c & tSession ) +{ + if ( tSession.m_pIndexerRtBulkStream ) + { + fclose ( tSession.m_pIndexerRtBulkStream ); + tSession.m_pIndexerRtBulkStream = nullptr; + } + tSession.m_pIndexerRtBulkBuffer.reset(); + + #if !_WIN32 + StopIndexerRtBulk ( tSession ); + #endif + + if ( !tSession.m_sIndexerRtBulkIndex.IsEmpty() ) + IndexFiles_c ( tSession.m_sIndexerRtBulkIndex, "indexer RT bulk" ).UnlinkExisted(); + + if ( !tSession.m_sIndexerRtBulkConfig.IsEmpty() ) + ::unlink ( tSession.m_sIndexerRtBulkConfig.cstr() ); + if ( !tSession.m_sIndexerRtBulkDir.IsEmpty() ) + { + CSphString sOutput; + sOutput.SetSprintf ( "%s/indexer.log", tSession.m_sIndexerRtBulkDir.cstr() ); + ::unlink ( sOutput.cstr() ); + } + + if ( !tSession.m_sIndexerRtBulkDir.IsEmpty() ) + ::rmdir ( tSession.m_sIndexerRtBulkDir.cstr() ); + + tSession.m_sIndexerRtBulkTable = ""; + tSession.m_iIndexerRtBulkIndexId = -1; + tSession.m_sIndexerRtBulkDir = ""; + tSession.m_sIndexerRtBulkConfig = ""; + tSession.m_sIndexerRtBulkIndex = ""; +} + + +static bool CheckSchemaSupported ( const CSphSchema & tSchema, CSphString & sError ) +{ + const CSphColumnInfo * pDocid = tSchema.GetAttr ( sphGetDocidName() ); + if ( !pDocid || pDocid->IsUuidLinkedDocid() ) + { + sError = "indexer RT bulk prototype supports numeric document ids only"; + return false; + } + + for ( int i=0; i pRt { pServed }; + const CSphSchema & tSchema = pRt->GetMatchSchema(); + if ( !CheckSchemaSupported ( tSchema, sError ) ) + return false; + + static std::atomic uBulkId { 0 }; + CSphString sParent = GetPathOnly ( pRt->GetFilebase() ); + tSession.m_sIndexerRtBulkDir.SetSprintf ( "%s/indexer-rt-bulk-%d-%u", sParent.cstr(), GetOsProcessId(), ++uBulkId ); + if ( !MkDir ( tSession.m_sIndexerRtBulkDir.cstr() ) ) + { + sError.SetSprintf ( "failed to create bulk staging directory '%s': %s", tSession.m_sIndexerRtBulkDir.cstr(), strerrorm ( errno ) ); + return false; + } + + tSession.m_sIndexerRtBulkConfig.SetSprintf ( "%s/indexer.conf", tSession.m_sIndexerRtBulkDir.cstr() ); + tSession.m_sIndexerRtBulkIndex.SetSprintf ( "%s/chunk", tSession.m_sIndexerRtBulkDir.cstr() ); + FILE * fpConfig = fopen ( tSession.m_sIndexerRtBulkConfig.cstr(), "wb" ); + if ( !fpConfig ) + { + sError.SetSprintf ( "failed to create bulk staging config in '%s': %s", tSession.m_sIndexerRtBulkDir.cstr(), strerrorm ( errno ) ); + CleanupIndexerRtBulk ( tSession ); + return false; + } + + fprintf ( fpConfig, "source indexer_rt_bulk_source {\n type = csvpipe\n csvpipe_command = /bin/cat\n" ); + for ( int i=0; im_eAttrType==SPH_ATTR_STRING ? "csvpipe_field_string" : "csvpipe_field", tField.m_sName.cstr() ); + } + for ( ESphAttr eType : g_dCsvAttrOrder ) + for ( int i=0; iGetSettings().m_dKNN.GetLength() ) + fprintf ( fpConfig, "\n knn = %s", FormatKNNConfigStr ( pRt->GetSettings().m_dKNN ).cstr() ); + fprintf ( fpConfig, "\n}\n" ); + fclose ( fpConfig ); + + tSession.m_sIndexerRtBulkTable = tStmt.m_sIndex; + tSession.m_iIndexerRtBulkIndexId = pRt->GetIndexId(); + #if _WIN32 + sError = "streaming indexer RT bulk is not implemented on Windows"; + CleanupIndexerRtBulk ( tSession ); + return false; + #else + if ( !StartIndexerRtBulk ( tSession, sError ) ) + { + CleanupIndexerRtBulk ( tSession ); + return false; + } + #endif + return true; +} + + +bool StageIndexerRtBulk ( ClientSession_c & tSession, const SqlStmt_t & tStmt, CSphString & sError ) +{ + if ( !tSession.m_bInTransaction ) + { + sError = "indexer_rt_bulk requires an active transaction; use BEGIN before INSERT"; + return false; + } + if ( tStmt.m_eStmt!=STMT_INSERT ) + { + sError = "indexer_rt_bulk prototype supports INSERT only"; + return false; + } + + auto pServed = GetServed ( tStmt.m_sIndex ); + if ( !ServedDesc_t::IsMutable ( pServed ) ) + { + sError.SetSprintf ( "table '%s' is absent, or not real-time", tStmt.m_sIndex.cstr() ); + return false; + } + if ( FindInsertColumn ( tStmt, sphGetDocidName() )<0 ) + { + sError = "indexer RT bulk prototype requires an explicit id"; + return false; + } + if ( !InitIndexerRtBulk ( tSession, tStmt, pServed, sError ) ) + return false; + + RIdx_T pRt { pServed }; + const CSphSchema & tSchema = pRt->GetMatchSchema(); + const int iColumns = tStmt.m_iSchemaSz; + for ( int i=0; i dColumns; + auto AddColumn = [&] ( const CSphString & sName, ESphAttr eType, bool bDocid ) { + dColumns.Add ( { FindInsertColumn ( tStmt, sName ), eType, bDocid } ); + }; + AddColumn ( sphGetDocidName(), SPH_ATTR_BIGINT, true ); + for ( int i=0; i + void CSphSource::SetDict ( const DictRefPtr_c& pDict ) { assert ( pDict ); @@ -987,3 +989,31 @@ void CSphSource::ParseFieldMVA ( int iAttr, const char * szValue ) if ( pDigit ) m_dMvas[iAttr].Add ( sphToInt64 ( pDigit ) ); } + + +bool CSphSource::ParseFieldFloatVector ( int iAttr, const char * szValue ) +{ + if ( !szValue ) + return true; + + const char * pValue = szValue; + while ( *pValue ) + { + while ( *pValue && ( sphIsSpace ( *pValue ) || *pValue==',' || *pValue=='(' || *pValue==')' ) ) + ++pValue; + if ( !*pValue ) + break; + + char * pEnd = nullptr; + float fValue = strtof ( pValue, &pEnd ); + if ( pEnd==pValue ) + return false; + if ( !std::isfinite ( fValue ) ) + return false; + + m_dMvas[iAttr].Add ( sphF2DW ( fValue ) ); + pValue = pEnd; + } + + return true; +} diff --git a/src/indexing_sources/source_document.h b/src/indexing_sources/source_document.h index 0c43e8c4f7..b58753d30b 100644 --- a/src/indexing_sources/source_document.h +++ b/src/indexing_sources/source_document.h @@ -185,6 +185,7 @@ class CSphSource : public CSphSourceSettings, public AttrSource_i protected: void ParseFieldMVA ( int iAttr, const char * szValue ); + bool ParseFieldFloatVector ( int iAttr, const char * szValue ); bool CheckFileField ( const BYTE * sField ); int LoadFileField ( BYTE ** ppField, CSphString & sError ); diff --git a/src/indexing_sources/source_svpipe.cpp b/src/indexing_sources/source_svpipe.cpp index 2f57e0781f..f4f6a41c14 100644 --- a/src/indexing_sources/source_svpipe.cpp +++ b/src/indexing_sources/source_svpipe.cpp @@ -188,6 +188,9 @@ bool CSphSource_BaseSV::SetupPipe ( const CSphConfigSection & hSource, FILE * pP CSphString sColumn; for ( const auto& tVal : hSource ) { + if ( tVal.first=="csvpipe_attr_order" ) + continue; + const CSphVariant * pVal = &tVal.second; while ( pVal ) { @@ -390,6 +393,14 @@ bool CSphSource_BaseSV::StoreAttribute ( int iAttr, int iOff ) ParseFieldMVA ( tRemap.m_iAttr, sVal ); break; + case SPH_ATTR_FLOAT_VECTOR: + if ( !ParseFieldFloatVector ( tRemap.m_iAttr, sVal ) ) + { + DecorateMessage ( "invalid float vector value '%s'", sVal ); + return false; + } + break; + case SPH_ATTR_TOKENCOUNT: m_tDocInfo.SetAttr ( tAttr.m_tLocator, 0 ); break; @@ -572,6 +583,7 @@ bool CSphSource_TSV::SetupSchema ( const CSphConfigSection & hSource, bool bWord bOk &= ConfigureAttrs ( hSource("tsvpipe_attr_bigint"), SPH_ATTR_BIGINT, m_tSchema, sError ); bOk &= ConfigureAttrs ( hSource("tsvpipe_attr_multi"), SPH_ATTR_UINT32SET, m_tSchema, sError ); bOk &= ConfigureAttrs ( hSource("tsvpipe_attr_multi_64"), SPH_ATTR_INT64SET, m_tSchema, sError ); + bOk &= ConfigureAttrs ( hSource("tsvpipe_attr_float_vector"), SPH_ATTR_FLOAT_VECTOR, m_tSchema, sError ); bOk &= ConfigureAttrs ( hSource("tsvpipe_attr_string"), SPH_ATTR_STRING, m_tSchema, sError ); bOk &= ConfigureAttrs ( hSource("tsvpipe_attr_json"), SPH_ATTR_JSON, m_tSchema, sError ); bOk &= ConfigureAttrs ( hSource("tsvpipe_field_string"), SPH_ATTR_STRING, m_tSchema, sError ); @@ -800,6 +812,7 @@ bool CSphSource_CSV::SetupSchema ( const CSphConfigSection & hSource, bool bWord bOk &= ConfigureAttrs ( hSource("csvpipe_attr_bigint"), SPH_ATTR_BIGINT, m_tSchema, sError ); bOk &= ConfigureAttrs ( hSource("csvpipe_attr_multi"), SPH_ATTR_UINT32SET, m_tSchema, sError ); bOk &= ConfigureAttrs ( hSource("csvpipe_attr_multi_64"), SPH_ATTR_INT64SET, m_tSchema, sError ); + bOk &= ConfigureAttrs ( hSource("csvpipe_attr_float_vector"), SPH_ATTR_FLOAT_VECTOR, m_tSchema, sError ); bOk &= ConfigureAttrs ( hSource("csvpipe_attr_string"), SPH_ATTR_STRING, m_tSchema, sError ); bOk &= ConfigureAttrs ( hSource("csvpipe_attr_json"), SPH_ATTR_JSON, m_tSchema, sError ); bOk &= ConfigureAttrs ( hSource("csvpipe_field_string"), SPH_ATTR_STRING, m_tSchema, sError ); @@ -807,6 +820,31 @@ bool CSphSource_CSV::SetupSchema ( const CSphConfigSection & hSource, bool bWord if ( !bOk ) return false; + CSphString sAttrOrder = hSource.GetStr ( "csvpipe_attr_order" ); + if ( !sAttrOrder.IsEmpty() ) + { + StrVec_t dNames; + sphSplit ( dNames, sAttrOrder.cstr(), "," ); + CSphVector dUsed ( m_tSchema.GetAttrsCount() ); + dUsed.ZeroVec(); + CSphSchema tOrdered ( m_tSchema.GetName() ); + for ( const CSphString & sName : dNames ) + { + int iAttr = m_tSchema.GetAttrIndex ( sName.cstr() ); + if ( iAttr<0 || dUsed[iAttr] ) + { + sError.SetSprintf ( "csvpipe_attr_order contains %s attribute '%s'", iAttr<0 ? "unknown" : "duplicate", sName.cstr() ); + return false; + } + tOrdered.AddAttr ( m_tSchema.GetAttr(iAttr), true ); + dUsed[iAttr] = true; + } + for ( int i=0; im_bIndexerRtBulk ) + CleanupIndexerRtBulk ( *pSession ); pSession->m_bInTransaction = true; tOut.Ok ( 0 ); } @@ -4989,6 +4992,17 @@ void sphHandleMysqlCommitRollback ( StmtErrorReporter_i& tOut, Str_t sQuery, boo pSession->m_bInTransaction = false; int iDeleted = 0; + if ( pSession->m_bIndexerRtBulk && !pSession->m_sIndexerRtBulkTable.IsEmpty() ) + { + if ( !bCommit ) + CleanupIndexerRtBulk ( *pSession ); + else if ( !FinalizeIndexerRtBulk ( *pSession, sError ) ) + { + tOut.Error ( "%s", sError.cstr() ); + return; + } + } + if ( pSession->m_tShardTxn.HasPendingData() ) { if ( bCommit ) @@ -5076,6 +5090,17 @@ void sphHandleMysqlInsert ( StmtErrorReporter_i & tOut, const SqlStmt_t & tStmt } assert ( pServed ); + if ( pSession->m_bIndexerRtBulk ) + { + if ( !StageIndexerRtBulk ( *pSession, tStmt, pSession->m_sError ) ) + { + CleanupIndexerRtBulk ( *pSession ); + tOut.Error ( "%s", pSession->m_sError.cstr() ); + } + else + tOut.Ok ( tStmt.m_iRowsAffected ); + return; + } Threads::Coro::ScopedWriteTable_c tWriting { pServed->Locker() }; if ( !tWriting.CanWrite() ) { @@ -9379,6 +9404,8 @@ static bool HandleSetLocal ( CSphString& sError, const CSphString& sName, int64_ // per-session AUTOCOMMIT bool bAutoCommit = ( iSetValue != 0 ); auto pSession = session::Info().GetClientSession(); + if ( !pSession->m_sIndexerRtBulkTable.IsEmpty() ) + CleanupIndexerRtBulk ( *pSession ); pSession->m_bAutoCommit = bAutoCommit; pSession->m_bInTransaction = false; @@ -9401,6 +9428,15 @@ static bool HandleSetLocal ( CSphString& sError, const CSphString& sName, int64_ return true; } + if ( sName == "indexer_rt_bulk" ) + { + auto pSession = session::Info().GetClientSession(); + if ( !iSetValue ) + CleanupIndexerRtBulk ( *pSession ); + pSession->m_bIndexerRtBulk = !!iSetValue; + return true; + } + if ( sName == "collation_connection" ) { // per-session COLLATION_CONNECTION @@ -9981,6 +10017,63 @@ void HandleMysqlAttach ( RowBuffer_i & tOut, const SqlStmt_t & tStmt, CSphString } +bool AttachIndexerRtBulkChunk ( const CSphString & sTable, int64_t iIndexId, const CSphString & sPath, CSphString & sError ) +{ + auto pServed = GetServed ( sTable ); + if ( !ServedDesc_t::IsMutable ( pServed ) ) + { + sError.SetSprintf ( "table '%s' disappeared while finalizing indexer RT bulk", sTable.cstr() ); + return false; + } + { + RIdx_T pRt { pServed }; + if ( pRt->GetIndexId()!=iIndexId ) + { + sError.SetSprintf ( "table '%s' was replaced while finalizing indexer RT bulk", sTable.cstr() ); + return false; + } + } + + auto pPlain = sphCreateIndexPhrase ( "indexer_rt_bulk_chunk", sPath ); + StrVec_t dWarnings; + if ( !pPlain->Prealloc ( false, nullptr, dWarnings ) ) + { + sError.SetSprintf ( "failed loading indexer RT bulk chunk: %s", pPlain->GetLastError().cstr() ); + return false; + } + + if ( pPlain->GetMatchSchema().HasKNNAttrs() && !pPlain->AlterKNN ( sError ) ) + { + sError.SetSprintf ( "failed building KNN index for indexer RT bulk chunk: %s", sError.cstr() ); + return false; + } + + bool bFatal = false; + bool bAttached = false; + { + auto pCurrent = GetServed ( sTable ); + if ( !ServedDesc_t::IsMutable ( pCurrent ) ) + { + sError.SetSprintf ( "table '%s' disappeared while finalizing indexer RT bulk", sTable.cstr() ); + return false; + } + + WIdx_T pRt { pCurrent }; + auto pRegistered = GetServed ( sTable ); + if ( pRegistered.Ptr()!=pCurrent.Ptr() || pRt->GetIndexId()!=iIndexId ) + { + sError.SetSprintf ( "table '%s' was replaced while finalizing indexer RT bulk", sTable.cstr() ); + return false; + } + + bAttached = pRt->AttachDiskIndex ( pPlain.get(), false, bFatal, sError ); + } + if ( bAttached ) + pPlain.release(); + return bAttached; +} + + void HandleMysqlFlushRtindex ( RowBuffer_i & tOut, const SqlStmt_t & tStmt ) { CSphString sError; @@ -12507,6 +12600,7 @@ void ClientSession_c::FreezeLastMeta() ClientSession_c::~ClientSession_c () { + CleanupIndexerRtBulk ( *this ); m_tShardTxn.Cleanup(); UnlockTables(this); } diff --git a/src/searchdbuddy.cpp b/src/searchdbuddy.cpp index 39985f669e..6c35b4155c 100644 --- a/src/searchdbuddy.cpp +++ b/src/searchdbuddy.cpp @@ -21,6 +21,7 @@ #define BOOST_PROCESS_VERSION 1 #include #include +#include #include #include #include diff --git a/src/searchdhttp.cpp b/src/searchdhttp.cpp index 347b986c85..7d84d009b1 100644 --- a/src/searchdhttp.cpp +++ b/src/searchdhttp.cpp @@ -22,6 +22,7 @@ #include "accumulator.h" #include "networking_daemon.h" #include "client_session.h" +#include "indexer_rt_bulk.h" #include "tracer.h" #include "searchdbuddy.h" #include "compressed_http.h" @@ -32,6 +33,8 @@ #include "auth/auth_proto_http.h" #include "netfetch.h" +#include + static bool g_bLogBadHttpReq = env_exists ( "MANTICORE_LOG_HTTP_BAD_REQ" ); // log content of bad http requests, ruled by this env variable static int g_iLogHttpData = env_long ( "MANTICORE_LOG_HTTP_DATA" ).value_or(0); // verbose logging of http data, ruled by this env variable @@ -175,7 +178,7 @@ static Endpoint_t g_dEndpoints[(size_t)EHTTP_ENDPOINT::TOTAL] = { "pq", "json/pq" }, { "cli", nullptr }, { "cli_json", nullptr }, - { "_bulk", nullptr }, + { "_bulk", "_bulk/" }, { "token", nullptr } }; @@ -2279,6 +2282,18 @@ class HttpHandler_JsonBulk_c : public HttpHandler_c, public HttpJsonUpdateTraits if ( !CheckNDJson() ) return false; + auto pSession = session::Info().GetClientSession(); + const bool bIndexerRtBulk = m_tOptions.Exists ( "indexer_rt_bulk" ) && m_tOptions["indexer_rt_bulk"]=="1"; + if ( bIndexerRtBulk ) + pSession->m_bIndexerRtBulk = true; + AT_SCOPE_EXIT ( [pSession, bIndexerRtBulk] { + if ( bIndexerRtBulk ) + { + CleanupIndexerRtBulk ( *pSession ); + pSession->m_bIndexerRtBulk = false; + } + }); + JsonObj_c tResults ( true ); bool bResult = false; int iCurLine = 0; @@ -2364,6 +2379,12 @@ class HttpHandler_JsonBulk_c : public HttpHandler_c, public HttpJsonUpdateTraits } SetQueryOptions ( m_tOptions, tStmt ); + if ( bIndexerRtBulk && tStmt.m_eStmt!=STMT_INSERT ) + { + m_sError = "indexer_rt_bulk prototype supports INSERT only"; + RollbackBulkTxn ( tTxnState ); + return FinishBulk ( tResults, false, iCurLine, iLastTxStartLine, EHTTP_STATUS::_400 ); + } switch ( tStmt.m_eStmt ) { @@ -2578,6 +2599,9 @@ class HttpHandlerEsBulk_c : public HttpCompatBaseHandler_c, public HttpJsonUpdat bool ProcessTnx ( const VecTraits_T & dTnx, VecTraits_T & dDocs, JsonObj_c & tItems ); bool Validate(); void ReportLogError ( const char * sError, HttpErrorType_e eType , EHTTP_STATUS eStatus, bool bLogOnly ); + bool IsIndexerRtBulk() const; + + bool m_bIndexerRtBulk { false }; }; class HttpTokenHandler_c final: public HttpHandler_c, public HttpOptionTrait_t @@ -3617,11 +3641,31 @@ bool HttpHandlerEsBulk_c::Validate() return true; } + +bool HttpHandlerEsBulk_c::IsIndexerRtBulk() const +{ + return ( m_hOpts.Exists ( "indexer_rt_bulk" ) && m_hOpts["indexer_rt_bulk"]=="1" ) + || ( m_hOpts.Exists ( "pipeline" ) && m_hOpts["pipeline"]=="indexer_rt_bulk" ); +} + + bool HttpHandlerEsBulk_c::Process() { if ( !Validate() ) return false; + m_bIndexerRtBulk = IsIndexerRtBulk(); + auto pSession = session::Info().GetClientSession(); + if ( m_bIndexerRtBulk ) + pSession->m_bIndexerRtBulk = true; + AT_SCOPE_EXIT ( [pSession, bIndexerRtBulk=m_bIndexerRtBulk] { + if ( bIndexerRtBulk ) + { + CleanupIndexerRtBulk ( *pSession ); + pSession->m_bIndexerRtBulk = false; + } + }); + auto & tCrashQuery = GlobalCrashQueryGetRef(); tCrashQuery.m_dQuery = S2B ( GetBody() ); @@ -3664,6 +3708,42 @@ bool HttpHandlerEsBulk_c::Process() return false; } } + + if ( m_bIndexerRtBulk ) + { + if ( dDocs.IsEmpty() ) + { + ReportLogError ( "indexer_rt_bulk requires at least one document", HttpErrorType_e::ActionRequestValidation, EHTTP_STATUS::_400, false ); + return false; + } + + const CSphString & sIndex = dDocs[0].m_sIndex; + std::unordered_set hDocids; + for ( const BulkDoc_t & tDoc : dDocs ) + { + if ( tDoc.m_sAction!="index" ) + { + ReportLogError ( "indexer_rt_bulk supports Elasticsearch bulk index actions only", HttpErrorType_e::ActionRequestValidation, EHTTP_STATUS::_400, false ); + return false; + } + if ( tDoc.m_sIndex!=sIndex ) + { + ReportLogError ( "indexer_rt_bulk requires one target table per request", HttpErrorType_e::ActionRequestValidation, EHTTP_STATUS::_400, false ); + return false; + } + if ( tDoc.m_eDocid!=BulkDocid_e::NUMERIC || !tDoc.m_tDocid ) + { + ReportLogError ( "indexer_rt_bulk requires an explicit non-zero numeric _id", HttpErrorType_e::ActionRequestValidation, EHTTP_STATUS::_400, false ); + return false; + } + if ( !hDocids.insert ( tDoc.m_tDocid ).second ) + { + ReportLogError ( "indexer_rt_bulk requires unique document ids within one request", HttpErrorType_e::ActionRequestValidation, EHTTP_STATUS::_400, false ); + return false; + } + } + } + CSphVector dTnx; const BulkDoc_t * pLastDoc = dDocs.Begin(); for ( const BulkDoc_t * pCurDoc = pLastDoc + 1; pCurDoc & dTnx, VecT dErrors.Add ( { iDoc, m_sError } ); continue; } + // Atomic plain-chunk attachment replaces matching ids; feed the reserved ES index action through the INSERT-only CSV serializer. + if ( m_bIndexerRtBulk && tStmt.m_eStmt==STMT_REPLACE ) + tStmt.m_eStmt = STMT_INSERT; bool bAction = false; JsonObj_c tResult = JsonNull; @@ -3843,6 +3926,16 @@ bool HttpHandlerEsBulk_c::ProcessTnx ( const VecTraits_T & dTnx, VecT } } + if ( m_bIndexerRtBulk && !dErrors.IsEmpty() ) + { + ProcessRollback ( FromStr ( sIdx ) ); + session::SetInTrans ( false ); + bOk = false; + for ( const auto & tErr : dErrors ) + AddEsError ( tErr.first, tErr.second, "mapper_parsing_exception", dDocs[tErr.first], tItems ); + continue; + } + // FIXME!!! check commit of empty accum JsonObj_c tResult; DocID_t tDocId = 0; @@ -3856,7 +3949,7 @@ bool HttpHandlerEsBulk_c::ProcessTnx ( const VecTraits_T & dTnx, VecT CSphString sUpdError; sUpdError.SetSprintf ( "[_doc][%s]: document missing", tUpdDoc.m_sDocid.scstr() ); AddEsError ( -1, sUpdError, "document_missing_exception", tUpdDoc, tItems ); - } else + } else if ( !m_bIndexerRtBulk ) { for ( int i=0; i * FetchMVA ( DocID_t tDocId, int iAttr, const C pMva = tMvaContainer.m_tContainer[iAttr]->Find(tDocId); } - if ( pMva ) + if ( pMva && tAttr.m_eAttrType==SPH_ATTR_FLOAT_VECTOR && tAttr.IsIndexedKNN() && tAttr.m_tKNN.m_eHNSWSimilarity==knn::HNSWSimilarity_e::COSINE ) + { + CSphVector dVector; + dVector.Resize ( pMva->GetLength() ); + ARRAY_FOREACH ( i, dVector ) + dVector[i] = sphDW2F ( (DWORD)(*pMva)[i] ); + NormalizeVec ( dVector ); + ARRAY_FOREACH ( i, dVector ) + (*pMva)[i] = sphF2DW ( dVector[i] ); + } + + if ( pMva && tAttr.m_eAttrType!=SPH_ATTR_FLOAT_VECTOR ) SortMva ( *pMva, ( tAttr.m_eAttrType==SPH_ATTR_UINT32SET ) ); return pMva; @@ -5014,6 +5025,7 @@ bool CSphIndex_VLN::Build_StoreBlobAttrs ( DocID_t tDocId, std::pair * pMva = FetchMVA ( tDocId, i, tAttr, tMvaContainer, tSource, bForceSource ); bOk = tBlobRowBuilder.SetAttr ( iBlobAttr++, pMva ? (const BYTE*)(pMva->Begin()) : nullptr, pMva ? pMva->GetLength()*sizeof(int64_t) : 0, BlobAttrInput_e::MVA_INT64, sError ); @@ -5072,6 +5084,7 @@ static void Builder_StoreAttrs ( const CSphSchema & tSchema, DocID_t tDocId, CSp case SPH_ATTR_UINT32SET: case SPH_ATTR_INT64SET: + case SPH_ATTR_FLOAT_VECTOR: { const CSphVector * pMva = FetchMVA ( tDocId, i, tAttr, tMvaContainer, tSource, false ); pBuilder->SetAttr ( iAttrId, pMva ? pMva->Begin() : nullptr, pMva ? pMva->GetLength() : 0 ); diff --git a/src/sphinxutils.cpp b/src/sphinxutils.cpp index d8a10d8c20..55e35226f4 100644 --- a/src/sphinxutils.cpp +++ b/src/sphinxutils.cpp @@ -916,7 +916,8 @@ static KeyDesc_t g_dKeysSource[] = { "tsvpipe_attr_bigint", KEY_LIST, NULL }, { "tsvpipe_attr_multi", KEY_LIST, NULL }, { "tsvpipe_attr_multi_64", KEY_LIST, NULL }, - { "tsvpipe_attr_string", KEY_LIST, NULL }, + { "tsvpipe_attr_float_vector", KEY_LIST, NULL }, + { "tsvpipe_attr_string", KEY_LIST, NULL }, { "tsvpipe_attr_json", KEY_LIST, NULL }, { "tsvpipe_field_string", KEY_LIST, NULL }, { "csvpipe_command", 0, NULL }, @@ -928,9 +929,11 @@ static KeyDesc_t g_dKeysSource[] = { "csvpipe_attr_bigint", KEY_LIST, NULL }, { "csvpipe_attr_multi", KEY_LIST, NULL }, { "csvpipe_attr_multi_64", KEY_LIST, NULL }, - { "csvpipe_attr_string", KEY_LIST, NULL }, + { "csvpipe_attr_float_vector", KEY_LIST, NULL }, + { "csvpipe_attr_string", KEY_LIST, NULL }, { "csvpipe_attr_json", KEY_LIST, NULL }, { "csvpipe_field_string", KEY_LIST, NULL }, + { "csvpipe_attr_order", 0, NULL }, { "csvpipe_delimiter", 0, NULL }, { NULL, 0, NULL } }; @@ -1038,6 +1041,7 @@ static KeyDesc_t g_dKeysIndex[] = { "docstore_compression", 0, nullptr }, { "docstore_compression_level", 0, nullptr }, { "columnar_attrs", 0, nullptr }, + { "engine", 0, nullptr }, { "columnar_no_fast_fetch", 0, nullptr }, { "rowwise_attrs", 0, nullptr }, { "columnar_strings_no_hash", 0, nullptr }, diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 1ad91d1b59..42f1dcaa85 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -98,6 +98,10 @@ function (add_ubertest testN tst_name LABELS prefix label flags) SET_TESTS_PROPERTIES ( "${_tstn}" PROPERTIES RESOURCE_GROUPS "datadir:1,mysql:1" ) endif () _test_prop ("${_tstn}") + if ("${flags}" MATCHES "--rt-indexer") + set_property (TEST "${_tstn}" APPEND PROPERTY ENVIRONMENT "MANTICORE_NO_BUDDY=1") + SET_TESTS_PROPERTIES ("${_tstn}" PROPERTIES RUN_SERIAL TRUE) + endif () endfunction () function (add_ubertests testN tst_name LABELS) @@ -107,6 +111,12 @@ function (add_ubertests testN tst_name LABELS) if (NOT "NON-RT" IN_LIST LABELS) add_ubertest ("${testN}" "${tst_name}" "${LABELS}" "rt" "RT" "--rt --ignore-weights") endif () + # Indexer-assisted loading is meaningful only for tests that have both a + # source and an index for the RT conversion path. Exclude semantic conflicts + # explicitly with . + if ("RT-INDEXER" IN_LIST LABELS AND NOT "NON-RT-INDEXER" IN_LIST LABELS) + add_ubertest ("${testN}" "${tst_name}" "${LABELS}" "rtidx" "RT_INDEXER" "--rt-indexer --ignore-weights") + endif () endfunction () fixup_test_name(tst "Cleaning guess cache") @@ -131,9 +141,11 @@ foreach ( test ${tests} ) if ( EXISTS "${test}/test.xml" ) # open the file with the test and read _chunk of 512 bytes file ( READ ${test}/test.xml _test_head LIMIT 512 ) + file ( READ ${test}/test.xml _test_xml ) # convert it to lower (CMAKE regexes are case-sensitive) STRING ( TOLOWER "${_test_head}" _lower_test_head ) + STRING ( TOLOWER "${_test_xml}" _lower_test_xml ) # extract test name. It is not simple 'match (.*)' since we want non-lower-case, # so find the chunk in lowered-case, then take the clause by substring from original chunk @@ -169,6 +181,12 @@ foreach ( test ${tests} ) LIST ( APPEND LABELS "SKIP_DB" ) endif () + if (NOT "NON-RT" IN_LIST LABELS AND NOT "NON-RT-INDEXER" IN_LIST LABELS AND NOT "SKIP_INDEXER" IN_LIST LABELS + AND "${_lower_test_xml}" MATCHES "(^|[\n\r])[ ]*source[ ]+[^ \n\r{]+" + AND "${_lower_test_xml}" MATCHES "(^|[\n\r])[ ]*index[ ]+[^ \n\r:{]+") + LIST ( APPEND LABELS "RT-INDEXER" ) + endif () + else () if (testN STREQUAL "028" ) # the only non-xml test we have set ( test_name "spelldump...") diff --git a/test/helpers.inc b/test/helpers.inc index 984c2f03ff..ecc9c29720 100644 --- a/test/helpers.inc +++ b/test/helpers.inc @@ -1651,6 +1651,12 @@ function IsRt() return $g_locals['rt_mode']; } +function IsRtIndexer() +{ + global $g_locals; + return !empty ( $g_locals['rt_indexer_mode'] ); +} + function IsColumnar() { global $g_locals; @@ -4421,7 +4427,8 @@ class SphinxConfig function InsertIntoIndexer ( &$error ) { - global $sd_address, $sd_sphinxql_port, $action_retries, $action_wait_timeout; + global $sd_address; + $assisted_commits = 0; $address = $sd_address; if ($address == "localhost") $address = "127.0.0.1"; @@ -4437,9 +4444,19 @@ class SphinxConfig if ( $cn !== false ) $cn->close(); $cn = new mysqli( $address, "", "", "Manticore", $port ); + if ( $cn->connect_errno ) + { + $error = "failed to connect to $connect_string: ".$cn->connect_error; + return false; + } + + if ( IsRtIndexer() && mysqli_wr ( "SET indexer_rt_bulk=1", $cn ) === false ) + { + $error = $cn->error; + $cn->close(); + return false; + } } - if ( $cn === false ) - return false; $corrected_cols = array(); foreach ( array_keys($data["orders"]) as $key ) @@ -4447,8 +4464,19 @@ class SphinxConfig $cols = join ( ", ", $corrected_cols ); $prefix = "INSERT INTO $name ($cols) VALUES "; - $accum = ""; + $in_transaction = false; + + if ( IsRtIndexer() && count ( $data['values'] ) ) + { + if ( mysqli_wr ( "BEGIN", $cn ) === false ) + { + $error = $cn->error; + $cn->close(); + return false; + } + $in_transaction = true; + } // mva shouldn't be quoted, e.g. "insert into rt (id,gid,mva) values ('1','2',(1,2))" $is_mva = array(); @@ -4468,12 +4496,14 @@ class SphinxConfig $i++; } - if ( ( strlen ($accum) + strlen ($query) ) > 8192000 ) /// 8192000 ) ///error; + if ( $in_transaction ) + mysqli_wr ( "ROLLBACK", $cn ); + $cn->close(); return false; } $accum=""; @@ -4484,18 +4514,32 @@ class SphinxConfig $accum .= "($query)"; } // final chunk; - if ( $accum !="" ) + if ( $accum !="" && mysqli_wr ( $prefix.$accum, $cn ) === false ) { - $result = mysqli_wr ( $prefix.$accum, $cn ); - if ( $result === false ) - { - $error = $cn->error; - return false; - } + $error = $cn->error; + if ( $in_transaction ) + mysqli_wr ( "ROLLBACK", $cn ); + $cn->close(); + return false; } + + if ( $in_transaction && mysqli_wr ( "COMMIT", $cn ) === false ) + { + $error = $cn->error; + mysqli_wr ( "ROLLBACK", $cn ); + $cn->close(); + return false; + } + if ( $in_transaction ) + $assisted_commits++; } if ( $cn ) $cn->close(); + if ( IsRtIndexer() && !$assisted_commits ) + { + $error = "RT indexer mode did not load any fixture data"; + return false; + } return true; } @@ -4794,6 +4838,12 @@ function CheckConfig ( $config, $path ) return false; } + if ( $config->Requires("non-rt-indexer") && IsRtIndexer() ) + { + printf ( "SKIPPING %s, %s - explicitly incompatible test skipped in RT indexer mode\n", $path, $config->Name () ); + return false; + } + if ( $config->Requires("non-columnar") && IsColumnar() ) { printf ( "SKIPPING %s, %s - explicitly non-columnar test skipped in columnar mode\n", $path, $config->Name () ); @@ -5146,8 +5196,12 @@ function RunTest ( $test_dir, $skipdemo, $usemarks ) continue; // in case of RT index - run "insert into" instead of indexer - if ( IsRt () ) - $config->InsertIntoIndexer ( $error ); + if ( IsRt () && !$config->InsertIntoIndexer ( $error ) && IsRtIndexer() ) + { + if ( !HandleFailure ( $config, $report, "$error\n", $nfailed ) ) + $log .= " subtest $subtest: error loading RT fixture data; see $report_file\n"; + continue; + } $config->ResetResults(); diff --git a/test/integration/indexer_rt_bulk_benchmark.py b/test/integration/indexer_rt_bulk_benchmark.py new file mode 100644 index 0000000000..b6a9b33afa --- /dev/null +++ b/test/integration/indexer_rt_bulk_benchmark.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +"""Benchmark dev#2761's indexer-assisted HTTP bulk path against direct RT bulk. + +Generate an NDJSON corpus with rows shaped as Manticore /bulk insert operations, +then start searchd and run this script under a one-CPU limiter, for example: + + ROWS=1000000 REPS=3 DATA=/path/to/taxi.ndjson \ + cpulimit --include-children -l 100 python3 indexer_rt_bulk_benchmark.py + +The daemon must listen on MYSQL_PORT/HTTP_PORT and have the matching indexer +binary beside the running searchd executable. +""" +import http.client +import json +import os +from pathlib import Path +import statistics +import threading +import time + +import pymysql + +ROWS = int(os.environ.get("ROWS", "1000000")) +REPS = int(os.environ.get("REPS", "3")) +BATCH = int(os.environ.get("BATCH", "1000")) +HOST = os.environ.get("HOST", "127.0.0.1") +MYSQL_PORT = int(os.environ.get("MYSQL_PORT", "19406")) +HTTP_PORT = int(os.environ.get("HTTP_PORT", "19408")) +DATA = Path(os.environ["DATA"]) +SCHEMA = "(pickup_datetime timestamp, dropoff_datetime timestamp, passenger_count int, trip_distance float, fare_amount float, pickup text, dropoff text)" + + +def sql(statement): + conn = pymysql.connect(host=HOST, port=MYSQL_PORT, user="root", autocommit=True) + try: + with conn.cursor() as cur: + cur.execute(statement) + return cur.fetchall() + finally: + conn.close() + + +def prepare(table): + sql(f"DROP TABLE IF EXISTS {table}") + sql(f"CREATE TABLE {table} {SCHEMA}") + + +def request(path, body, encode_chunked=False): + conn = http.client.HTTPConnection(HOST, HTTP_PORT, timeout=600) + conn.request("POST", path, body=body, headers={"Content-Type": "application/x-ndjson"}, encode_chunked=encode_chunked) + reply = conn.getresponse() + payload = reply.read() + conn.close() + if reply.status != 200: + raise RuntimeError(f"HTTP {reply.status}: {payload[:1000]!r}") + parsed = json.loads(payload) + if parsed.get("errors"): + raise RuntimeError(str(parsed)[:2000]) + + +def direct_rt(table): + with DATA.open("rb") as source: + batch = [] + for line in source: + batch.append(line.replace(b'"table":"taxi"', f'"table":"{table}"'.encode(), 1)) + if len(batch) == BATCH: + request("/bulk", b"".join(batch)) + batch.clear() + if batch: + request("/bulk", b"".join(batch)) + + +def chunks(table): + old = b'"table":"taxi"' + new = f'"table":"{table}"'.encode() + with DATA.open("rb") as source: + block = bytearray() + for line in source: + block.extend(line.replace(old, new, 1)) + if len(block) >= 1024 * 1024: + yield bytes(block) + block.clear() + if block: + yield bytes(block) + + +def indexer_rt(table): + request("/bulk?indexer_rt_bulk=1", chunks(table), encode_chunked=True) + + +def descendants_rss(root_pid): + try: + rows = os.popen("ps -axo pid=,ppid=,rss=").read().splitlines() + procs = [tuple(map(int, row.split())) for row in rows] + except Exception: + return 0 + active = {root_pid} + changed = True + while changed: + changed = False + for pid, ppid, _ in procs: + if ppid in active and pid not in active: + active.add(pid) + changed = True + return sum(rss for pid, _, rss in procs if pid in active) + + +def run(mode, rep): + table = f"taxi_{mode}_{rep}" + prepare(table) + root_pid = os.getppid() + peak = 0 + stop = threading.Event() + + def sample(): + nonlocal peak + while not stop.is_set(): + peak = max(peak, descendants_rss(root_pid)) + stop.wait(0.05) + + sampler = threading.Thread(target=sample, daemon=True) + sampler.start() + started = time.perf_counter() + (direct_rt if mode == "direct" else indexer_rt)(table) + elapsed = time.perf_counter() - started + stop.set() + sampler.join() + count = sql(f"SELECT COUNT(*) FROM {table}")[0][0] + status = dict(sql(f"SHOW TABLE {table} STATUS")) + if count != ROWS: + raise RuntimeError(f"{mode}: expected {ROWS}, got {count}") + result = { + "mode": mode, + "rep": rep, + "seconds": elapsed, + "rows_per_sec": ROWS / elapsed, + "rows": count, + "disk_chunks": int(status.get("disk_chunks", 0)), + "ram_bytes": int(status.get("ram_bytes", 0)), + "disk_bytes": int(status.get("disk_bytes", 0)), + "peak_tree_rss_kb": peak, + } + print(json.dumps(result), flush=True) + sql(f"DROP TABLE {table}") + return result + + +def main(): + results = [] + for rep in range(1, REPS + 1): + order = ("direct", "indexer") if rep % 2 else ("indexer", "direct") + for mode in order: + results.append(run(mode, rep)) + summary = {} + for mode in ("direct", "indexer"): + subset = [row for row in results if row["mode"] == mode] + summary[mode] = { + "median_seconds": statistics.median(row["seconds"] for row in subset), + "median_rows_per_sec": statistics.median(row["rows_per_sec"] for row in subset), + "max_peak_tree_rss_kb": max(row["peak_tree_rss_kb"] for row in subset), + } + summary["speedup"] = summary["direct"]["median_seconds"] / summary["indexer"]["median_seconds"] + print(json.dumps({"summary": summary}, indent=2), flush=True) + + +if __name__ == "__main__": + main() diff --git a/test/integration/indexer_rt_bulk_fluentbit_test.py b/test/integration/indexer_rt_bulk_fluentbit_test.py new file mode 100644 index 0000000000..5bed7325c2 --- /dev/null +++ b/test/integration/indexer_rt_bulk_fluentbit_test.py @@ -0,0 +1,229 @@ +#!/usr/bin/env python +"""Verify Fluent Bit-compatible Elasticsearch bulk uses indexer-assisted RT insertion.""" + +from __future__ import print_function + +import argparse +try: + import http.client as http_client +except ImportError: + import httplib as http_client +import json +import os +import shutil +import signal +import socket +import subprocess +import tempfile +import time + + +def free_port(): + sock = socket.socket() + try: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + finally: + sock.close() + + +def request(port, path, body): + connection = http_client.HTTPConnection("127.0.0.1", port, timeout=30) + connection.request("POST", path, body=body, headers={"Content-Type": "application/x-ndjson"}) + response = connection.getresponse() + payload = response.read().decode("utf-8", errors="replace") + connection.close() + return response.status, payload + + +def sql(port, statement): + status, payload = request(port, "/sql?mode=raw", statement) + assert status == 200, (statement, status, payload) + result = json.loads(payload)[0] + assert not result.get("error"), (statement, result) + return result.get("data", []) + + +def count(port, table): + return sql(port, "SELECT COUNT(*) FROM {}".format(table))[0]["count(*)"] + + +def row(port, table, docid): + rows = sql(port, "SELECT id,title,gid FROM {} WHERE id={}".format(table, docid)) + return rows[0] if rows else None + + +def disk_chunks(port, table): + status = sql(port, "SHOW TABLE {} STATUS".format(table)) + return int(dict((item["Variable_name"], item["Value"]) for item in status)["disk_chunks"]) + + +def bulk(port, lines, pipeline=True): + path = "/_bulk/?pipeline=indexer_rt_bulk" if pipeline else "/_bulk/" + body = "".join(json.dumps(line, separators=(",", ":")) + "\n" for line in lines) + status, payload = request(port, path, body) + return status, json.loads(payload) + + +def wait_ready(port, process): + deadline = time.monotonic() + 10 if hasattr(time, "monotonic") else time.time() + 10 + now = time.monotonic if hasattr(time, "monotonic") else time.time + last_error = "" + while now() < deadline: + if process.poll() is not None: + raise AssertionError("searchd exited during startup with status {}".format(process.returncode)) + try: + status, payload = request(port, "/sql?mode=raw", "SELECT 1") + if status == 200 and '"1":1' in payload: + return + last_error = "HTTP {}: {}".format(status, payload) + except (OSError, socket.error) as exc: + last_error = str(exc) + time.sleep(0.05) + raise AssertionError("searchd did not become ready: {}".format(last_error)) + + +def wait_process(process, timeout): + deadline = time.time() + timeout + while process.poll() is None and time.time() < deadline: + time.sleep(0.05) + return process.poll() is not None + + +def staging_dirs(data_dir): + found = [] + for root, names, _ in os.walk(data_dir): + for name in names: + if name.startswith("indexer-rt-bulk-"): + found.append(os.path.join(root, name)) + return found + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--searchd", required=True) + args = parser.parse_args() + + root = tempfile.mkdtemp(prefix="manticore-indexer-fluentbit-") + process = None + output = None + try: + data_dir = os.path.join(root, "data") + os.mkdir(data_dir) + port = free_port() + config = os.path.join(root, "manticore.conf") + with open(config, "w") as config_file: + config_file.write( + "searchd {{\n" + " listen = 127.0.0.1:{port}:http\n" + " log = {log}\n" + " query_log = {query_log}\n" + " pid_file = {pid_file}\n" + " data_dir = {data_dir}\n" + " workers = threads\n" + " watchdog = 0\n" + "}}\n".format( + port=port, + log=os.path.join(root, "searchd.log"), + query_log=os.path.join(root, "query.log"), + pid_file=os.path.join(root, "searchd.pid"), + data_dir=data_dir, + ) + ) + + environment = os.environ.copy() + environment["MANTICORE_NO_BUDDY"] = "1" + output = open(os.path.join(root, "searchd-console.log"), "wb") + process = subprocess.Popen( + [args.searchd, "--config", config, "--nodetach"], + cwd=root, + env=environment, + stdout=output, + stderr=output, + ) + wait_ready(port, process) + + table = "indexer_rt_fluent_bit" + other = table + "_other" + sql(port, "CREATE TABLE {}(title TEXT, gid INTEGER)".format(table)) + sql(port, "CREATE TABLE {}(title TEXT, gid INTEGER)".format(other)) + + status, payload = bulk(port, [ + {"index": {"_index": table, "_id": "401"}}, + {"title": "first fluent bit row", "gid": 1}, + {"index": {"_index": table, "_id": "402"}}, + {"title": "second fluent bit row", "gid": 2}, + ]) + assert status == 200 and payload["errors"] is False, payload + assert payload["items"] == [], payload + assert count(port, table) == 2 + assert disk_chunks(port, table) == 1 + + status, payload = bulk(port, [ + {"index": {"_index": table, "_id": "401"}}, + {"title": "replaced fluent bit row", "gid": 10}, + ]) + assert status == 200 and payload["errors"] is False, payload + assert count(port, table) == 2 + assert row(port, table, 401) == {"id": 401, "title": "replaced fluent bit row", "gid": 10} + + status, payload = bulk(port, [ + {"index": {"_index": table, "_id": "405"}}, + {"title": "must not attach to first", "gid": 5}, + {"index": {"_index": other, "_id": "406"}}, + {"title": "must not attach to second", "gid": 6}, + ]) + assert status == 400 and "one target table" in payload["error"]["reason"], payload + assert count(port, table) == 2 and count(port, other) == 0 + + status, payload = bulk(port, [ + {"index": {"_index": table, "_id": "408"}}, + {"title": "first duplicate", "gid": 8}, + {"index": {"_index": table, "_id": "408"}}, + {"title": "second duplicate", "gid": 9}, + ]) + assert status == 400 and "unique document ids" in payload["error"]["reason"], payload + assert count(port, table) == 2 + + status, payload = bulk(port, [ + {"create": {"_index": table, "_id": "407"}}, + {"title": "unsupported create", "gid": 7}, + ]) + assert status == 400 and "bulk index actions only" in payload["error"]["reason"], payload + assert count(port, table) == 2 + + status, payload = request(port, "/_bulk/?pipeline=indexer_rt_bulk", "\n") + payload = json.loads(payload) + assert status == 400 and "at least one document" in payload["error"]["reason"], payload + + before_chunks = disk_chunks(port, table) + status, payload = bulk(port, [ + {"index": {"_index": table, "_id": "408"}}, + {"title": "must not attach", "gid": 8}, + {"index": {"_index": table}}, + {"title": "missing id", "gid": 9}, + ]) + assert status == 400 and "explicit non-zero numeric _id" in payload["error"]["reason"], payload + assert count(port, table) == 2 and disk_chunks(port, table) == before_chunks + + status, payload = bulk(port, [ + {"create": {"_index": table, "_id": "403"}}, + {"title": "ordinary trailing slash", "gid": 3}, + ], pipeline=False) + assert status == 200 and payload["errors"] is False, payload + assert len(payload["items"]) == 1 and count(port, table) == 3 + assert not staging_dirs(data_dir), staging_dirs(data_dir) + assert process.poll() is None, "Fluent Bit bulk test stopped searchd" + finally: + if process is not None and process.poll() is None: + process.send_signal(signal.SIGTERM) + if not wait_process(process, 10): + process.kill() + wait_process(process, 5) + if output is not None: + output.close() + shutil.rmtree(root, ignore_errors=True) + + +if __name__ == "__main__": + main() diff --git a/test/integration/indexer_rt_bulk_missing_sibling_test.py b/test/integration/indexer_rt_bulk_missing_sibling_test.py new file mode 100644 index 0000000000..f19c4afce7 --- /dev/null +++ b/test/integration/indexer_rt_bulk_missing_sibling_test.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python +"""Verify that indexer-assisted ingestion lazily requires an absolute sibling.""" + +from __future__ import print_function + +import argparse +try: + import http.client as http_client +except ImportError: + import httplib as http_client +import json +import os +import shutil +import signal +import socket +import subprocess +import tempfile +import time + + +def free_port(): + sock = socket.socket() + try: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + finally: + sock.close() + + +def request(port, path, body): + connection = http_client.HTTPConnection("127.0.0.1", port, timeout=5) + connection.request("POST", path, body=body, headers={"Content-Type": "application/x-ndjson"}) + response = connection.getresponse() + payload = response.read().decode("utf-8", errors="replace") + connection.close() + return response.status, payload + + +def sql(port, statement): + return request(port, "/sql?mode=raw", statement) + + +def wait_ready(port, process): + deadline = time.monotonic() + 10 if hasattr(time, "monotonic") else time.time() + 10 + now = time.monotonic if hasattr(time, "monotonic") else time.time + last_error = "" + while now() < deadline: + if process.poll() is not None: + raise AssertionError("searchd exited during startup with status {}".format(process.returncode)) + try: + status, payload = sql(port, "SELECT 1") + if status == 200 and '"1":1' in payload: + return + last_error = "HTTP {}: {}".format(status, payload) + except (OSError, socket.error) as exc: + last_error = str(exc) + time.sleep(0.05) + raise AssertionError("searchd did not become ready: {}".format(last_error)) + + +def wait_process(process, timeout): + deadline = time.time() + timeout + while process.poll() is None and time.time() < deadline: + time.sleep(0.05) + return process.poll() is not None + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--searchd", required=True) + args = parser.parse_args() + + root = tempfile.mkdtemp(prefix="manticore-indexer-sibling-") + process = None + devnull = None + try: + actual_bindir = os.path.join(root, "actual") + launch_bindir = os.path.join(root, "launch") + data_dir = os.path.join(root, "data") + os.mkdir(actual_bindir) + os.mkdir(launch_bindir) + os.mkdir(data_dir) + searchd = os.path.join(actual_bindir, "searchd") + launch_searchd = os.path.join(launch_bindir, "searchd") + shutil.copy2(args.searchd, searchd) + os.symlink(searchd, launch_searchd) + missing_indexer = os.path.realpath(os.path.join(actual_bindir, "indexer")) + path_marker = os.path.join(root, "path-indexer-ran") + path_indexer = os.path.join(launch_bindir, "indexer") + with open(path_indexer, "w") as script: + script.write("#!/bin/sh\n: > '{}'\nexit 0\n".format(path_marker.replace("'", "'\\''"))) + os.chmod(path_indexer, 0o755) + port = free_port() + config = os.path.join(root, "manticore.conf") + with open(config, "w") as config_file: + config_file.write( + "searchd {{\n" + " listen = 127.0.0.1:{port}:http\n" + " log = {log}\n" + " query_log = {query_log}\n" + " pid_file = {pid_file}\n" + " data_dir = {data_dir}\n" + " workers = threads\n" + " watchdog = 0\n" + "}}\n".format( + port=port, + log=os.path.join(root, "searchd.log"), + query_log=os.path.join(root, "query.log"), + pid_file=os.path.join(root, "searchd.pid"), + data_dir=data_dir, + ) + ) + + environment = os.environ.copy() + environment["MANTICORE_NO_BUDDY"] = "1" + environment["MANTICORE_INDEXER_RT_INDEXER"] = path_indexer + environment["PATH"] = launch_bindir + devnull = open(os.devnull, "wb") + process = subprocess.Popen( + [launch_searchd, "--config", config, "--nodetach"], + cwd=root, + env=environment, + stdout=devnull, + stderr=devnull, + ) + + wait_ready(port, process) + status, payload = sql(port, "CREATE TABLE missing_sibling(title TEXT)") + assert status == 200, payload + + body = json.dumps({"insert": {"index": "missing_sibling", "id": 1, "doc": {"title": "test"}}}) + "\n" + status, payload = request(port, "/bulk?indexer_rt_bulk=1", body) + assert status != 200, payload + assert missing_indexer in payload, payload + assert "indexer" in payload.lower(), payload + assert not os.path.exists(path_marker), "searchd launched indexer from PATH" + + diagnostic = "diagnostic-marker-from-indexer" + with open(missing_indexer, "w") as script: + script.write("#!/bin/sh\n/bin/cat >/dev/null\necho '{}' >&2\nexit 1\n".format(diagnostic)) + os.chmod(missing_indexer, 0o755) + status, payload = request(port, "/bulk?indexer_rt_bulk=1", body) + assert status != 200, payload + assert diagnostic in payload, payload + + status, payload = sql(port, "SELECT COUNT(*) FROM missing_sibling") + assert status == 200 and '"count(*)":0' in payload.lower(), payload + assert process.poll() is None, "missing sibling stopped searchd" + finally: + if process is not None and process.poll() is None: + process.send_signal(signal.SIGTERM) + if not wait_process(process, 10): + process.kill() + wait_process(process, 5) + if devnull is not None: + devnull.close() + shutil.rmtree(root, ignore_errors=True) + + +if __name__ == "__main__": + main() diff --git a/test/integration/indexer_rt_bulk_test.py b/test/integration/indexer_rt_bulk_test.py new file mode 100644 index 0000000000..26132a1984 --- /dev/null +++ b/test/integration/indexer_rt_bulk_test.py @@ -0,0 +1,680 @@ +#!/usr/bin/env python3 +"""Focused integration test for the indexer-assisted RT bulk prototype.""" + +import argparse +import http.client +import json +import os +from pathlib import Path +import threading +import time + +import pymysql + + +def sql_conn(port): + return pymysql.connect(host="127.0.0.1", port=port, user="root", autocommit=True, ssl_disabled=True) + + +def query(conn, statement, args=None): + with conn.cursor() as cur: + cur.execute(statement, args) + return cur.fetchall() + + +def count(conn, table): + return query(conn, f"SELECT COUNT(*) FROM {table}")[0][0] + + +def table_status(conn, table): + return dict(query(conn, f"SHOW TABLE {table} STATUS")) + + +def disk_chunks(conn, table): + return int(table_status(conn, table)["disk_chunks"]) + + +def staging_dirs(data_dir): + directories = [] + for root, names, _ in os.walk(data_dir): + for name in names: + if name.startswith("indexer-rt-bulk-"): + path = Path(root, name) + if path.exists(): + directories.append(path) + return directories + + +def indexer_pids(): + pids = [] + for entry in Path("/proc").glob("[0-9]*"): + try: + command = (entry / "cmdline").read_bytes().replace(b"\0", b" ") + except (FileNotFoundError, PermissionError, ProcessLookupError): + continue + if b"indexer_rt_bulk_chunk" in command: + pids.append(int(entry.name)) + if pids or Path("/proc").exists(): + return pids + output = os.popen("ps -axo pid=,command=").read().splitlines() + return [int(line.split(None, 1)[0]) for line in output if "indexer_rt_bulk_chunk" in line] + + +def wait_for_streaming_indexer(data_dir, timeout=5): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + directories = staging_dirs(data_dir) + if len(directories) == 1 and indexer_pids(): + assert not (directories[0] / "input.csv").exists(), "streaming path created an input.csv" + return directories[0] + time.sleep(0.02) + raise AssertionError("indexer did not start before request/transaction completion") + + +def sql_transaction_test(port, data_dir): + writer = sql_conn(port) + reader = sql_conn(port) + query(writer, "DROP TABLE IF EXISTS indexer_rt_sql") + query(writer, "CREATE TABLE indexer_rt_sql(title TEXT, gid INTEGER, price FLOAT)") + query(writer, "INSERT INTO indexer_rt_sql(id,title,gid,price) VALUES (10,'existing RT row',100,10.0)") + query(writer, "SET indexer_rt_bulk=1") + query(writer, "BEGIN") + query(writer, "INSERT INTO indexer_rt_sql(id,title,gid,price) VALUES (1,'comma, quote \\\' and line\\nfeed',10,1.25)") + wait_for_streaming_indexer(data_dir) + # This exceeds normal socket and pipe buffers. Returning from INSERT before + # COMMIT proves the indexer/csvpipe consumer is actively draining the stream. + large_title = "second searchable row " + "x" * (2 * 1024 * 1024) + query(writer, "INSERT INTO indexer_rt_sql(id,title,gid,price) VALUES (2,%s,20,2.5)", (large_title,)) + assert count(reader, "indexer_rt_sql") == 1, "staged SQL rows became visible before COMMIT" + query(writer, "COMMIT") + assert count(reader, "indexer_rt_sql") == 3 + rows = query(reader, "SELECT id,gid,price FROM indexer_rt_sql WHERE MATCH('searchable')") + assert rows == ((2, 20, 2.5),), rows + + query(writer, "BEGIN") + query(writer, "INSERT INTO indexer_rt_sql(id,title,gid,price) VALUES (3,'rollback row',30,3.5)") + query(writer, "ROLLBACK") + assert count(reader, "indexer_rt_sql") == 3, "ROLLBACK leaked a staged row" + writer.close() + reader.close() + + +def sql_disconnect_cleanup_test(port, data_dir): + writer = sql_conn(port) + reader = sql_conn(port) + query(writer, "DROP TABLE IF EXISTS indexer_rt_disconnect") + query(writer, "CREATE TABLE indexer_rt_disconnect(title TEXT, gid INTEGER)") + query(writer, "SET indexer_rt_bulk=1") + query(writer, "BEGIN") + query(writer, "INSERT INTO indexer_rt_disconnect(id,title,gid) VALUES (11,'abandoned row',1)") + assert count(reader, "indexer_rt_disconnect") == 0 + writer.close() + deadline = time.monotonic() + 5 + while staging_dirs(data_dir) and time.monotonic() < deadline: + time.sleep(0.02) + assert not staging_dirs(data_dir), "disconnect left a staging directory behind" + assert count(reader, "indexer_rt_disconnect") == 0 + + writer = sql_conn(port) + query(writer, "SET indexer_rt_bulk=1") + try: + query(writer, "INSERT INTO indexer_rt_disconnect(id,title,gid) VALUES (12,'missing begin',2)") + raise AssertionError("bulk mode unexpectedly accepted INSERT outside BEGIN") + except pymysql.MySQLError as exc: + assert "requires an active transaction" in str(exc), exc + assert not staging_dirs(data_dir), "INSERT outside BEGIN left staging state behind" + + query(writer, "BEGIN") + query(writer, "INSERT INTO indexer_rt_disconnect(id,title,gid) VALUES (13,'staged before replace',3)") + try: + query(writer, "REPLACE INTO indexer_rt_disconnect(id,title,gid) VALUES (14,'unsupported replace',4)") + raise AssertionError("bulk mode unexpectedly accepted REPLACE") + except pymysql.MySQLError as exc: + assert "supports INSERT only" in str(exc), exc + deadline = time.monotonic() + 5 + while staging_dirs(data_dir) and time.monotonic() < deadline: + time.sleep(0.02) + assert not staging_dirs(data_dir), "rejected REPLACE did not cancel the streaming pipeline" + query(writer, "COMMIT") + assert count(reader, "indexer_rt_disconnect") == 0, "COMMIT attached rows from a failed ingestion" + writer.close() + reader.close() + + +def concurrent_streams_test(port, data_dir): + first = sql_conn(port) + second = sql_conn(port) + reader = sql_conn(port) + query(reader, "DROP TABLE IF EXISTS indexer_rt_concurrent_a") + query(reader, "DROP TABLE IF EXISTS indexer_rt_concurrent_b") + query(reader, "CREATE TABLE indexer_rt_concurrent_a(title TEXT)") + query(reader, "CREATE TABLE indexer_rt_concurrent_b(title TEXT)") + for connection, table, docid in ( + (first, "indexer_rt_concurrent_a", 1), + (second, "indexer_rt_concurrent_b", 2), + ): + query(connection, "SET indexer_rt_bulk=1") + query(connection, "BEGIN") + query(connection, f"INSERT INTO {table}(id,title) VALUES ({docid},'concurrent stream')") + + deadline = time.monotonic() + 5 + while len(indexer_pids()) < 2 and time.monotonic() < deadline: + time.sleep(0.02) + assert len(indexer_pids()) >= 2, "two streaming transactions did not start two indexers" + + outcome = {} + + def commit_first(): + outcome["rows"] = query(first, "COMMIT") + + committer = threading.Thread(target=commit_first, daemon=True) + committer.start() + committer.join(timeout=10) + assert not committer.is_alive(), "first COMMIT waited for the second stream to close" + assert count(reader, "indexer_rt_concurrent_a") == 1 + assert count(reader, "indexer_rt_concurrent_b") == 0 + query(second, "ROLLBACK") + first.close() + second.close() + reader.close() + assert not staging_dirs(data_dir), "concurrent streams left staging state behind" + + +def transaction_invariant_test(port, data_dir): + writer = sql_conn(port) + admin = sql_conn(port) + query(admin, "DROP TABLE IF EXISTS indexer_rt_invariant") + query(admin, "CREATE TABLE indexer_rt_invariant(title TEXT)") + query(writer, "SET indexer_rt_bulk=1") + query(writer, "BEGIN") + query(writer, "INSERT INTO indexer_rt_invariant(id,title) VALUES (1,'cancel on mode change')") + wait_for_streaming_indexer(data_dir) + query(writer, "SET autocommit=1") + assert not staging_dirs(data_dir), "SET autocommit did not cancel the streaming transaction" + assert count(admin, "indexer_rt_invariant") == 0 + + query(writer, "BEGIN") + query(writer, "INSERT INTO indexer_rt_invariant(id,title) VALUES (2,'old table object')") + wait_for_streaming_indexer(data_dir) + query(admin, "DROP TABLE indexer_rt_invariant") + query(admin, "CREATE TABLE indexer_rt_invariant(title TEXT)") + try: + query(writer, "COMMIT") + raise AssertionError("COMMIT attached the chunk to a replacement table") + except pymysql.MySQLError as exc: + assert "was replaced" in str(exc), exc + assert count(admin, "indexer_rt_invariant") == 0 + assert not staging_dirs(data_dir), "replacement-table rejection left staging state behind" + writer.close() + admin.close() + + +def all_types_and_knn_test(port, data_dir): + writer = sql_conn(port) + reader = sql_conn(port) + table = "indexer_rt_all_types" + query(writer, f"DROP TABLE IF EXISTS {table}") + query(writer, f"CREATE TABLE {table} (id BIGINT, title TEXT, i INTEGER, bits BIT(5), bi BIGINT, b BOOL, ts TIMESTAMP, f FLOAT, s STRING, j JSON, m MULTI, m64 MULTI64, v FLOAT_VECTOR KNN_TYPE='hnsw' KNN_DIMS='4' HNSW_SIMILARITY='l2')") + query(writer, "SET indexer_rt_bulk=1") + query(writer, "BEGIN") + query(writer, f"INSERT INTO {table} (id,title,i,bits,bi,b,ts,f,s,j,m,m64,v) VALUES " + "(1,'alpha, \\\"quoted\\\"',42,31,-9223372036854775807,1,1700000000,3.1415927,'hello, \\\"csv\\\"','{\\\"name\\\":\\\"alpha\\\",\\\"n\\\":7}',(3,1,2),(9223372036854775806,-5),(-0.25,0.5,0,0))," + "(2,'beta',0,17,9223372036854775807,0,2147483647,-0.125,'','{\\\"name\\\":\\\"beta\\\",\\\"ok\\\":true}',(),(),(1,0,0,0))," + "(3,'gamma',4294967295,1,-1,1,1,1.5,'unicode café','[1,2,3]',(9,9,8),(7,7,-9),(0,1,0,0))") + query(writer, f"INSERT INTO {table} (id,title,v) VALUES (4,'defaults',(0,0,1,0))") + assert count(reader, table) == 0, "all-type rows became visible before COMMIT" + query(writer, "COMMIT") + + rows = sorted(query(reader, f"SELECT id,title,i,bits,bi,b,ts,f,s,j,m,m64,v FROM {table}")) + assert len(rows) == 4, rows + assert rows[0][:7] == (1, 'alpha, "quoted"', 42, 31, -9223372036854775807, 1, 1700000000), rows[0] + assert abs(rows[0][7] - 3.1415927) < 1e-5 + assert rows[0][8] == 'hello, "csv"' + assert json.loads(rows[0][9]) == {"name": "alpha", "n": 7} + assert rows[0][10:13] == ("1,2,3", "-5,9223372036854775806", "-0.250000,0.500000,0.000000,0.000000") + assert rows[1][2:7] == (0, 17, 9223372036854775807, 0, 2147483647) + assert rows[1][10:12] == ("", "") + assert rows[2][2] == 4294967295 + assert rows[2][3] == 1 + assert rows[2][8] == "unicode café" + assert json.loads(rows[2][9]) == [1, 2, 3] + assert rows[2][10:12] == ("8,9", "-9,7") + assert rows[3][2:12] == (0, 0, 0, 0, 0, 0.0, "", None, "", "") + assert disk_chunks(reader, table) == 1 + + knn = query(reader, f"SELECT id,knn_dist() FROM {table} WHERE knn(v,4,(0.9,0.1,0,0)) ORDER BY knn_dist() ASC") + assert [row[0] for row in knn] == [2, 1, 3, 4], knn + expected = (0.02, 1.4825, 1.62, 1.82) + assert all(abs(row[1] - expected[i]) < 1e-5 for i, row in enumerate(knn)), knn + assert query(reader, f"SELECT id FROM {table} WHERE knn(v,4,(0.9,0.1,0,0)) ORDER BY knn_dist() ASC") == ((2,), (1,), (3,), (4,)) + + writer.close() + reader.close() + + +def unindexed_float_vector_test(port): + writer = sql_conn(port) + reader = sql_conn(port) + table = "indexer_rt_unindexed_vector" + query(writer, f"DROP TABLE IF EXISTS {table}") + query(writer, f"CREATE TABLE {table} (title TEXT, v FLOAT_VECTOR)") + query(writer, "SET indexer_rt_bulk=1") + query(writer, "BEGIN") + query(writer, f"INSERT INTO {table}(id,title,v) VALUES (1,'three decimals',(0.25,-1.5,2.75)),(2,'one decimal',(9.5))") + query(writer, "COMMIT") + rows = sorted(query(reader, f"SELECT id,v FROM {table}")) + assert rows == [(1, "0.250000,-1.500000,2.750000"), (2, "9.500000")], rows + assert disk_chunks(reader, table) == 1 + writer.close() + reader.close() + + +def field_forms_and_id_boundary_test(port): + writer = sql_conn(port) + reader = sql_conn(port) + for table, definition, column, selectable in ( + ("indexer_rt_field_attr", "both STRING INDEXED ATTRIBUTE", "both", True), + ("indexer_rt_field_indexed", "indexed_value TEXT INDEXED", "indexed_value", False), + ("indexer_rt_field_stored", "saved_value TEXT STORED", "saved_value", True), + ): + query(writer, f"DROP TABLE IF EXISTS {table}") + query(writer, f"CREATE TABLE {table} ({definition})") + query(writer, "SET indexer_rt_bulk=1") + query(writer, "BEGIN") + query(writer, f"INSERT INTO {table}(id,{column}) VALUES (1,'alpha beta'),(2,'gamma')") + query(writer, "COMMIT") + assert query(reader, f"SELECT id FROM {table} WHERE MATCH('alpha')") == ((1,),) + if selectable: + assert query(reader, f"SELECT id,{column} FROM {table}") == ((1, "alpha beta"), (2, "gamma")) + assert disk_chunks(reader, table) == 1 + + direct = "indexer_rt_id_boundary_direct" + assisted = "indexer_rt_id_boundary_assisted" + for table, bulk in ((direct, 0), (assisted, 1)): + query(writer, f"DROP TABLE IF EXISTS {table}") + query(writer, f"CREATE TABLE {table} (title TEXT)") + query(writer, f"SET indexer_rt_bulk={bulk}") + if bulk: + query(writer, "BEGIN") + query(writer, f"INSERT INTO {table}(id,title) VALUES (9223372036854775808,'high bit'),(18446744073709551615,'maximum')") + if bulk: + query(writer, "COMMIT") + direct_rows = query(reader, f"SELECT id,title FROM {direct}") + assisted_rows = query(reader, f"SELECT id,title FROM {assisted}") + assert assisted_rows == direct_rows == ((-9223372036854775808, "high bit"), (-1, "maximum")), (direct_rows, assisted_rows) + assert disk_chunks(reader, assisted) == 1 + writer.close() + reader.close() + + +def columnar_all_types_test(port): + writer = sql_conn(port) + reader = sql_conn(port) + table = "indexer_rt_columnar_types" + query(writer, f"DROP TABLE IF EXISTS {table}") + query(writer, f"CREATE TABLE {table} (title TEXT, i INTEGER, bi BIGINT, f FLOAT, s STRING, j JSON, m MULTI, m64 MULTI64, v FLOAT_VECTOR) ENGINE='columnar'") + query(writer, "SET indexer_rt_bulk=1") + query(writer, "BEGIN") + query(writer, f"INSERT INTO {table}(id,title,i,bi,f,s,j,m,m64,v) VALUES (1,'columnar searchable',4294967295,-7,-0.25,'café','{{\\\"x\\\":1}}',(3,1,3),(9,-2),(0.25,-1.5))") + query(writer, "COMMIT") + rows = query(reader, f"SELECT id,i,bi,f,s,j,m,m64,v FROM {table}") + assert len(rows) == 1 and rows[0][:3] == (1, 4294967295, -7), rows + assert abs(rows[0][3] + 0.25) < 1e-6 + assert rows[0][4] == "café" and json.loads(rows[0][5]) == {"x": 1} + assert rows[0][6:] == ("1,3", "-2,9", "0.250000,-1.500000"), rows + assert query(reader, f"SELECT id FROM {table} WHERE MATCH('searchable')") == ((1,),) + assert disk_chunks(reader, table) == 1 + writer.close() + reader.close() + + +def columnar_and_cosine_knn_test(port): + writer = sql_conn(port) + reader = sql_conn(port) + cases = ( + ("indexer_rt_columnar_vector", "l2", " ENGINE='columnar'", (1.0, 1.0)), + ("indexer_rt_cosine_vector", "cosine", "", (0.70710677, 0.70710677)), + ("indexer_rt_columnar_cosine_vector", "cosine", " ENGINE='columnar'", (0.70710677, 0.70710677)), + ) + + for table, similarity, engine, expected_third in cases: + query(writer, f"DROP TABLE IF EXISTS {table}") + query(writer, f"CREATE TABLE {table} (title TEXT, v FLOAT_VECTOR KNN_TYPE='hnsw' KNN_DIMS='4' HNSW_SIMILARITY='{similarity}'{engine})") + query(writer, "SET indexer_rt_bulk=1") + query(writer, "BEGIN") + query(writer, f"INSERT INTO {table}(id,title,v) VALUES (1,'x',(1,0,0,0)),(2,'y',(0,1,0,0)),(3,'z',(1,1,0,0))") + query(writer, "COMMIT") + + rows = sorted(query(reader, f"SELECT id,v FROM {table}")) + vector = tuple(float(value) for value in rows[2][1].split(',')) + assert abs(vector[0] - expected_third[0]) < 1e-5 and abs(vector[1] - expected_third[1]) < 1e-5, rows + knn = query(reader, f"SELECT id,knn_dist() FROM {table} WHERE knn(v,3,(0.9,0.1,0,0)) ORDER BY knn_dist() ASC") + assert [row[0] for row in knn] == [1, 3, 2], knn + assert disk_chunks(reader, table) == 1 + + direct = "indexer_rt_cosine_vector_direct" + query(writer, f"DROP TABLE IF EXISTS {direct}") + query(writer, f"CREATE TABLE {direct} (title TEXT, v FLOAT_VECTOR KNN_TYPE='hnsw' KNN_DIMS='4' HNSW_SIMILARITY='cosine')") + query(writer, "SET indexer_rt_bulk=0") + query(writer, f"INSERT INTO {direct}(id,title,v) VALUES (1,'x',(1,0,0,0)),(2,'y',(0,1,0,0)),(3,'z',(1,1,0,0))") + direct_rows = sorted(query(reader, f"SELECT id,v FROM {direct}")) + assisted_rows = sorted(query(reader, "SELECT id,v FROM indexer_rt_cosine_vector")) + columnar_assisted_rows = sorted(query(reader, "SELECT id,v FROM indexer_rt_columnar_cosine_vector")) + assert direct_rows == assisted_rows == columnar_assisted_rows, (direct_rows, assisted_rows, columnar_assisted_rows) + direct_knn = query(reader, f"SELECT id,knn_dist() FROM {direct} WHERE knn(v,3,(0.9,0.1,0,0)) ORDER BY knn_dist() ASC") + assisted_knn = query(reader, "SELECT id,knn_dist() FROM indexer_rt_cosine_vector WHERE knn(v,3,(0.9,0.1,0,0)) ORDER BY knn_dist() ASC") + columnar_assisted_knn = query(reader, "SELECT id,knn_dist() FROM indexer_rt_columnar_cosine_vector WHERE knn(v,3,(0.9,0.1,0,0)) ORDER BY knn_dist() ASC") + assert [row[0] for row in direct_knn] == [row[0] for row in assisted_knn] + assert all(abs(direct_knn[i][1] - assisted_knn[i][1]) < 1e-5 for i in range(len(direct_knn))), (direct_knn, assisted_knn) + assert [row[0] for row in direct_knn] == [row[0] for row in columnar_assisted_knn] + assert all(abs(direct_knn[i][1] - columnar_assisted_knn[i][1]) < 1e-5 for i in range(len(direct_knn))), (direct_knn, columnar_assisted_knn) + + writer.close() + reader.close() + + +def vector_rejection_atomicity_test(port, data_dir): + writer = sql_conn(port) + reader = sql_conn(port) + table = "indexer_rt_bad_vectors" + query(writer, f"DROP TABLE IF EXISTS {table}") + query(writer, f"CREATE TABLE {table} (title TEXT, v FLOAT_VECTOR KNN_TYPE='hnsw' KNN_DIMS='4' HNSW_SIMILARITY='l2')") + query(writer, "SET indexer_rt_bulk=1") + + for bad_vector, expected in ( + ("(1,2,3)", "requires 4 vector entries"), + ("(1,2,3,4,5)", "requires 4 vector entries"), + ("()", "requires a tuple value"), + ("(1e400,0,0,0)", "entries must be finite"), + ("(nan,0,0,0)", "entries must be finite"), + ("(inf,0,0,0)", "entries must be finite"), + ): + query(writer, "BEGIN") + query(writer, f"INSERT INTO {table}(id,title,v) VALUES (1,'valid before rejection',(1,0,0,0))") + try: + query(writer, f"INSERT INTO {table}(id,title,v) VALUES (2,'invalid',{bad_vector})") + raise AssertionError(f"accepted invalid vector {bad_vector}") + except pymysql.MySQLError as exc: + assert expected in str(exc), exc + query(writer, "COMMIT") + assert count(reader, table) == 0, f"{bad_vector} attached a partial chunk" + assert disk_chunks(reader, table) == 0, f"{bad_vector} attached a partial disk chunk" + assert not staging_dirs(data_dir), f"{bad_vector} left staging state behind" + assert not indexer_pids(), f"{bad_vector} left an indexer worker behind" + + query(writer, "BEGIN") + try: + query(writer, f"INSERT INTO {table}(id,title,v) VALUES (3,'malformed','not-a-vector')") + raise AssertionError("accepted malformed float_vector text") + except pymysql.MySQLError as exc: + assert "float_vector" in str(exc).lower(), exc + query(writer, "COMMIT") + assert count(reader, table) == 0 + assert disk_chunks(reader, table) == 0 + assert not staging_dirs(data_dir) + assert not indexer_pids() + writer.close() + reader.close() + + +def send_chunked_bulk(port, first_sent, finish, response): + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=30) + conn.putrequest("POST", "/bulk?indexer_rt_bulk=1") + conn.putheader("Content-Type", "application/x-ndjson") + conn.putheader("Transfer-Encoding", "chunked") + conn.endheaders() + + def chunk(obj): + payload = (json.dumps(obj, separators=(",", ":")) + "\n").encode() + conn.send(f"{len(payload):X}\r\n".encode() + payload + b"\r\n") + + chunk({"insert": {"index": "indexer_rt_http", "id": 101, "doc": {"title": "first streamed row " + "x" * 16384, "gid": 1}}}) + first_sent.set() + finish.wait(timeout=20) + chunk({"insert": {"index": "indexer_rt_http", "id": 102, "doc": {"title": "second streamed row", "gid": 2}}}) + conn.send(b"0\r\n\r\n") + reply = conn.getresponse() + response["status"] = reply.status + response["body"] = json.loads(reply.read()) + conn.close() + + +def http_streaming_test(sql_port, http_port, data_dir): + reader = sql_conn(sql_port) + query(reader, "DROP TABLE IF EXISTS indexer_rt_http") + query(reader, "CREATE TABLE indexer_rt_http(title TEXT, gid INTEGER)") + + first_sent = threading.Event() + finish = threading.Event() + response = {} + sender = threading.Thread(target=send_chunked_bulk, args=(http_port, first_sent, finish, response), daemon=True) + sender.start() + assert first_sent.wait(timeout=5), "sender did not transmit the first HTTP chunk" + wait_for_streaming_indexer(data_dir) + assert count(reader, "indexer_rt_http") == 0, "HTTP row became visible before request EOF" + finish.set() + sender.join(timeout=30) + assert not sender.is_alive(), "chunked bulk request did not finish" + assert response["status"] == 200, response + assert response["body"]["errors"] is False, response + assert count(reader, "indexer_rt_http") == 2 + assert int(table_status(reader, "indexer_rt_http")["disk_chunks"]) == 1 + assert not staging_dirs(data_dir), "successful HTTP bulk left staging state behind" + ids = {row[0] for row in query(reader, "SELECT id FROM indexer_rt_http WHERE MATCH('streamed')")} + assert ids == {101, 102}, ids + reader.close() + + +def malformed_http_cleanup_test(sql_port, http_port, data_dir): + reader = sql_conn(sql_port) + query(reader, "DROP TABLE IF EXISTS indexer_rt_http_bad") + query(reader, "CREATE TABLE indexer_rt_http_bad(title TEXT, gid INTEGER)") + + conn = http.client.HTTPConnection("127.0.0.1", http_port, timeout=30) + body = ( + json.dumps({"insert": {"index": "indexer_rt_http_bad", "id": 201, "doc": {"title": "must rollback", "gid": 1}}}) + + "\n{not-json}\n" + ) + conn.request("POST", "/bulk?indexer_rt_bulk=1", body=body, headers={"Content-Type": "application/x-ndjson"}) + reply = conn.getresponse() + reply.read() + conn.close() + assert reply.status == 400, reply.status + assert count(reader, "indexer_rt_http_bad") == 0 + assert not staging_dirs(data_dir), "failed HTTP bulk left staging state behind" + reader.close() + + +def unsupported_http_operation_test(sql_port, http_port, data_dir): + reader = sql_conn(sql_port) + query(reader, "DROP TABLE IF EXISTS indexer_rt_http_ops") + query(reader, "CREATE TABLE indexer_rt_http_ops(title TEXT, gid INTEGER)") + query(reader, "INSERT INTO indexer_rt_http_ops(id,title,gid) VALUES (301,'keep me',1)") + + conn = http.client.HTTPConnection("127.0.0.1", http_port, timeout=30) + body = json.dumps({"delete": {"index": "indexer_rt_http_ops", "id": 301}}) + "\n" + conn.request("POST", "/bulk?indexer_rt_bulk=1", body=body, headers={"Content-Type": "application/x-ndjson"}) + reply = conn.getresponse() + payload = json.loads(reply.read()) + conn.close() + assert reply.status == 400, (reply.status, payload) + assert "supports INSERT only" in payload["error"], payload + assert count(reader, "indexer_rt_http_ops") == 1 + assert not staging_dirs(data_dir), "rejected HTTP operation left staging state behind" + reader.close() + + + +def es_bulk_fluent_bit_test(sql_port, http_port, data_dir): + reader = sql_conn(sql_port) + table = "indexer_rt_fluent_bit" + query(reader, f"DROP TABLE IF EXISTS {table}") + query(reader, f"CREATE TABLE {table}(title TEXT, gid INTEGER)") + + docs = [ + {"index": {"_index": table, "_id": "401"}}, + {"title": "first fluent bit row", "gid": 1}, + {"index": {"_index": table, "_id": "402"}}, + {"title": "second fluent bit row", "gid": 2}, + ] + body = "".join(json.dumps(doc, separators=(",", ":")) + "\n" for doc in docs) + conn = http.client.HTTPConnection("127.0.0.1", http_port, timeout=30) + conn.request("POST", "/_bulk/?pipeline=indexer_rt_bulk", body=body, headers={"Content-Type": "application/x-ndjson"}) + reply = conn.getresponse() + payload = json.loads(reply.read()) + conn.close() + assert reply.status == 200, (reply.status, payload) + assert payload["errors"] is False, payload + assert payload["items"] == [], payload + assert count(reader, table) == 2 + assert disk_chunks(reader, table) == 1 + assert {row[0] for row in query(reader, f"SELECT id FROM {table}")} == {401, 402} + assert not staging_dirs(data_dir), "successful Fluent Bit bulk left staging state behind" + + replacement = ( + json.dumps({"index": {"_index": table, "_id": "401"}}) + + "\n" + + json.dumps({"title": "replaced fluent bit row", "gid": 10}) + + "\n" + ) + conn = http.client.HTTPConnection("127.0.0.1", http_port, timeout=30) + conn.request("POST", "/_bulk/?pipeline=indexer_rt_bulk", body=replacement, headers={"Content-Type": "application/x-ndjson"}) + reply = conn.getresponse() + payload = json.loads(reply.read()) + conn.close() + assert reply.status == 200, (reply.status, payload) + assert payload["items"] == [], payload + assert count(reader, table) == 2 + assert query(reader, f"SELECT title, gid FROM {table} WHERE id=401")[0] == ("replaced fluent bit row", 10) + + other_table = table + "_other" + query(reader, f"DROP TABLE IF EXISTS {other_table}") + query(reader, f"CREATE TABLE {other_table}(title TEXT, gid INTEGER)") + multiple_tables = ( + json.dumps({"index": {"_index": table, "_id": "405"}}) + + "\n" + + json.dumps({"title": "must not attach to first", "gid": 5}) + + "\n" + + json.dumps({"index": {"_index": other_table, "_id": "406"}}) + + "\n" + + json.dumps({"title": "must not attach to second", "gid": 6}) + + "\n" + ) + conn = http.client.HTTPConnection("127.0.0.1", http_port, timeout=30) + conn.request("POST", "/_bulk/?pipeline=indexer_rt_bulk", body=multiple_tables, headers={"Content-Type": "application/x-ndjson"}) + reply = conn.getresponse() + payload = json.loads(reply.read()) + conn.close() + assert reply.status == 400, (reply.status, payload) + assert "one target table" in payload["error"]["reason"], payload + assert count(reader, table) == 2 + assert count(reader, other_table) == 0 + assert not staging_dirs(data_dir), "rejected multi-table bulk left staging state behind" + + duplicate_ids = ( + json.dumps({"index": {"_index": table, "_id": "408"}}) + + "\n" + + json.dumps({"title": "first duplicate", "gid": 8}) + + "\n" + + json.dumps({"index": {"_index": table, "_id": "408"}}) + + "\n" + + json.dumps({"title": "second duplicate", "gid": 9}) + + "\n" + ) + conn = http.client.HTTPConnection("127.0.0.1", http_port, timeout=30) + conn.request("POST", "/_bulk/?pipeline=indexer_rt_bulk", body=duplicate_ids, headers={"Content-Type": "application/x-ndjson"}) + reply = conn.getresponse() + payload = json.loads(reply.read()) + conn.close() + assert reply.status == 400, (reply.status, payload) + assert "unique document ids" in payload["error"]["reason"], payload + assert count(reader, table) == 2 + + create_action = ( + json.dumps({"create": {"_index": table, "_id": "407"}}) + + "\n" + + json.dumps({"title": "unsupported create", "gid": 7}) + + "\n" + ) + conn = http.client.HTTPConnection("127.0.0.1", http_port, timeout=30) + conn.request("POST", "/_bulk/?pipeline=indexer_rt_bulk", body=create_action, headers={"Content-Type": "application/x-ndjson"}) + reply = conn.getresponse() + payload = json.loads(reply.read()) + conn.close() + assert reply.status == 400, (reply.status, payload) + assert "bulk index actions only" in payload["error"]["reason"], payload + assert count(reader, table) == 2 + + conn = http.client.HTTPConnection("127.0.0.1", http_port, timeout=30) + conn.request("POST", "/_bulk/?pipeline=indexer_rt_bulk", body="\n", headers={"Content-Type": "application/x-ndjson"}) + reply = conn.getresponse() + payload = json.loads(reply.read()) + conn.close() + assert reply.status == 400, (reply.status, payload) + assert "at least one document" in payload["error"]["reason"], payload + assert count(reader, table) == 2 + + ordinary = ( + json.dumps({"create": {"_index": table, "_id": "403"}}) + + "\n" + + json.dumps({"title": "ordinary trailing slash", "gid": 3}) + + "\n" + ) + conn = http.client.HTTPConnection("127.0.0.1", http_port, timeout=30) + conn.request("POST", "/_bulk/", body=ordinary, headers={"Content-Type": "application/x-ndjson"}) + reply = conn.getresponse() + payload = json.loads(reply.read()) + conn.close() + assert reply.status == 200, (reply.status, payload) + assert payload["errors"] is False, payload + assert len(payload["items"]) == 1, payload + assert count(reader, table) == 3 + + before_chunks = disk_chunks(reader, table) + missing_id = ( + json.dumps({"index": {"_index": table, "_id": "404"}}) + + "\n" + + json.dumps({"title": "must not attach", "gid": 4}) + + "\n" + + json.dumps({"index": {"_index": table}}) + + "\n" + + json.dumps({"title": "no id", "gid": 5}) + + "\n" + ) + conn = http.client.HTTPConnection("127.0.0.1", http_port, timeout=30) + conn.request("POST", "/_bulk/?pipeline=indexer_rt_bulk", body=missing_id, headers={"Content-Type": "application/x-ndjson"}) + reply = conn.getresponse() + payload = json.loads(reply.read()) + conn.close() + assert reply.status == 400, (reply.status, payload) + assert "explicit non-zero numeric _id" in payload["error"]["reason"], payload + assert count(reader, table) == 3, "rejected Elasticsearch bulk attached a partial chunk" + assert disk_chunks(reader, table) == before_chunks + assert not staging_dirs(data_dir), "rejected Fluent Bit bulk left staging state behind" + reader.close() + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--sql-port", type=int, default=19306) + parser.add_argument("--http-port", type=int, default=19308) + parser.add_argument("--data-dir", default="/tmp/manticore-indexer-rt/data") + args = parser.parse_args() + sql_transaction_test(args.sql_port, args.data_dir) + sql_disconnect_cleanup_test(args.sql_port, args.data_dir) + concurrent_streams_test(args.sql_port, args.data_dir) + transaction_invariant_test(args.sql_port, args.data_dir) + all_types_and_knn_test(args.sql_port, args.data_dir) + unindexed_float_vector_test(args.sql_port) + field_forms_and_id_boundary_test(args.sql_port) + columnar_all_types_test(args.sql_port) + columnar_and_cosine_knn_test(args.sql_port) + vector_rejection_atomicity_test(args.sql_port, args.data_dir) + http_streaming_test(args.sql_port, args.http_port, args.data_dir) + malformed_http_cleanup_test(args.sql_port, args.http_port, args.data_dir) + unsupported_http_operation_test(args.sql_port, args.http_port, args.data_dir) + es_bulk_fluent_bit_test(args.sql_port, args.http_port, args.data_dir) + print("indexer-assisted SQL, HTTP bulk, and Fluent Bit tests: PASS") + + +if __name__ == "__main__": + main() diff --git a/test/test_009/test.xml b/test/test_009/test.xml index 4d3ae7dfc4..7c94f10f8a 100644 --- a/test/test_009/test.xml +++ b/test/test_009/test.xml @@ -1,6 +1,7 @@ attributes over network + indexer { diff --git a/test/test_015/test.xml b/test/test_015/test.xml index fc609fb7af..9762d602aa 100644 --- a/test/test_015/test.xml +++ b/test/test_015/test.xml @@ -3,6 +3,8 @@ phrase matching vs duplicate keywords + + indexer { diff --git a/test/test_019/test.xml b/test/test_019/test.xml index b924c08c97..aaff6023bf 100644 --- a/test/test_019/test.xml +++ b/test/test_019/test.xml @@ -3,6 +3,8 @@ extended queries + + indexer { diff --git a/test/test_022/test.xml b/test/test_022/test.xml index aca4856576..1d0c351fe6 100644 --- a/test/test_022/test.xml +++ b/test/test_022/test.xml @@ -3,6 +3,8 @@ wordforms vs case folding + + indexer { diff --git a/test/test_126/test.xml b/test/test_126/test.xml index be0df66c0f..5a14526f15 100644 --- a/test/test_126/test.xml +++ b/test/test_126/test.xml @@ -3,6 +3,8 @@ expand keywords + + indexer { diff --git a/test/test_150/test.xml b/test/test_150/test.xml index 15665e2de3..cc358c29fb 100644 --- a/test/test_150/test.xml +++ b/test/test_150/test.xml @@ -3,6 +3,8 @@ keywords dictionary vs expansion limit + + indexer { diff --git a/test/test_193/test.xml b/test/test_193/test.xml index c4262d843f..190025c4d2 100644 --- a/test/test_193/test.xml +++ b/test/test_193/test.xml @@ -3,6 +3,8 @@ stopwords vs wordforms vs multiforms + + indexer { diff --git a/test/test_194/test.xml b/test/test_194/test.xml index d8cc0b6230..f256dc5a20 100644 --- a/test/test_194/test.xml +++ b/test/test_194/test.xml @@ -5,6 +5,7 @@ + diff --git a/test/test_196/test.xml b/test/test_196/test.xml index 053e9f1d20..40587a0857 100644 --- a/test/test_196/test.xml +++ b/test/test_196/test.xml @@ -3,6 +3,8 @@ wordforms: comments, after morph modifier, multiple wordform files + + indexer { diff --git a/test/test_206/test.xml b/test/test_206/test.xml index 604192a6ec..7f11ddbb20 100644 --- a/test/test_206/test.xml +++ b/test/test_206/test.xml @@ -5,6 +5,7 @@ + diff --git a/test/test_207/test.xml b/test/test_207/test.xml index 8ffd947031..ce67811b25 100644 --- a/test/test_207/test.xml +++ b/test/test_207/test.xml @@ -3,6 +3,7 @@ + aot morphology in utf-8 diff --git a/test/test_219/test.xml b/test/test_219/test.xml index fcdddf6d55..77241a2668 100644 --- a/test/test_219/test.xml +++ b/test/test_219/test.xml @@ -3,6 +3,7 @@ + aot morphology for english and german diff --git a/test/test_222/test.xml b/test/test_222/test.xml index cad39d7233..8a9cfdb3c9 100644 --- a/test/test_222/test.xml +++ b/test/test_222/test.xml @@ -3,6 +3,7 @@ + cooperation of index_exact_word with expanding by lematizer, expand diff --git a/test/test_324/test.xml b/test/test_324/test.xml index 84a084643d..a7854e9f19 100644 --- a/test/test_324/test.xml +++ b/test/test_324/test.xml @@ -3,6 +3,8 @@ distributed KEYWORDS + + indexer { diff --git a/test/test_337/test.xml b/test/test_337/test.xml index ff53b13b49..aa03d3e9ba 100644 --- a/test/test_337/test.xml +++ b/test/test_337/test.xml @@ -5,6 +5,7 @@ + diff --git a/test/test_342/test.xml b/test/test_342/test.xml index 8e29ac581f..f4a138d2e6 100644 --- a/test/test_342/test.xml +++ b/test/test_342/test.xml @@ -3,6 +3,8 @@ expand keywords - query option + + indexer { diff --git a/test/test_348/test.xml b/test/test_348/test.xml index a7a759f393..4c683f9c26 100644 --- a/test/test_348/test.xml +++ b/test/test_348/test.xml @@ -3,6 +3,8 @@ expand keywords - multiple options + + indexer { diff --git a/test/test_365/test.xml b/test/test_365/test.xml index 46f5c2555a..f7defba3e6 100644 --- a/test/test_365/test.xml +++ b/test/test_365/test.xml @@ -3,6 +3,8 @@ KEYWORDS for bigram + + indexer { diff --git a/test/test_390/test.xml b/test/test_390/test.xml index f707dc502d..e6f423bddc 100644 --- a/test/test_390/test.xml +++ b/test/test_390/test.xml @@ -3,6 +3,8 @@ document storage + + indexer { diff --git a/test/test_392/test.xml b/test/test_392/test.xml index 1ae9f2da04..966377c00d 100644 --- a/test/test_392/test.xml +++ b/test/test_392/test.xml @@ -3,6 +3,8 @@ document storage vs highlighting + + indexer { diff --git a/test/test_397/test.xml b/test/test_397/test.xml index fd666fa2d8..bf01144ea2 100644 --- a/test/test_397/test.xml +++ b/test/test_397/test.xml @@ -6,6 +6,7 @@ + diff --git a/test/test_431/test.xml b/test/test_431/test.xml index a82f960804..4863adea75 100644 --- a/test/test_431/test.xml +++ b/test/test_431/test.xml @@ -5,6 +5,7 @@ + diff --git a/test/test_433/test.xml b/test/test_433/test.xml index 3c21e34525..36accf3d42 100644 --- a/test/test_433/test.xml +++ b/test/test_433/test.xml @@ -5,6 +5,7 @@ + diff --git a/test/test_435/test.xml b/test/test_435/test.xml index 8cbf226ecf..7cbcee91da 100644 --- a/test/test_435/test.xml +++ b/test/test_435/test.xml @@ -5,6 +5,7 @@ + diff --git a/test/test_440/test.xml b/test/test_440/test.xml index 2ccbbccb3f..bddd13bc54 100644 --- a/test/test_440/test.xml +++ b/test/test_440/test.xml @@ -5,6 +5,7 @@ + diff --git a/test/test_447/test.xml b/test/test_447/test.xml index b754187d5c..abe4b3165b 100644 --- a/test/test_447/test.xml +++ b/test/test_447/test.xml @@ -5,6 +5,7 @@ + diff --git a/test/test_451/test.xml b/test/test_451/test.xml index 3909764706..09407964d4 100644 --- a/test/test_451/test.xml +++ b/test/test_451/test.xml @@ -3,6 +3,8 @@ stored fields vs field_string attribute + + indexer { diff --git a/test/test_463/test.xml b/test/test_463/test.xml index 11ec2f40da..fd4998aa43 100644 --- a/test/test_463/test.xml +++ b/test/test_463/test.xml @@ -3,6 +3,7 @@ + aot morphology vs original forms diff --git a/test/test_471/test.xml b/test/test_471/test.xml index 61117b39a1..d1effb9643 100644 --- a/test/test_471/test.xml +++ b/test/test_471/test.xml @@ -1,6 +1,8 @@ global idf in non RT mode + + indexer { diff --git a/test/test_494/test.xml b/test/test_494/test.xml index e7f8dba50a..a76b25edfb 100644 --- a/test/test_494/test.xml +++ b/test/test_494/test.xml @@ -3,6 +3,8 @@ plain table direct query defaults from config + + indexer { diff --git a/test/test_514/test.xml b/test/test_514/test.xml index d53d7e5d4b..92e7b1e0de 100644 --- a/test/test_514/test.xml +++ b/test/test_514/test.xml @@ -5,6 +5,7 @@ + diff --git a/test/ubertest.php b/test/ubertest.php index b2daeb7459..58643d44ca 100644 --- a/test/ubertest.php +++ b/test/ubertest.php @@ -40,9 +40,10 @@ print ( "--strict\t\tterminate on the first failure (for automatic runs)\n" ); print ( "--strict-verbose\tterminate on the first failure and copy the last report to report.txt (for automatic runs)\n" ); print ( "--managed\t\tdon't run searchd during test (for debugging)\n" ); - print ( "--skip-indexer\t\tskip DB creation and indexer stages and go directly to queries/custom tests\n"); - print ( "--rt\t\t\ttest RT backend (auto-convert all local indexes)\n" ); - print ( "--columnar\t\t\ttest attrs as columnar\n" ); + print ( "--skip-indexer skip DB creation and indexer stages and go directly to queries/custom tests\n"); + print ( "--rt test RT backend (auto-convert all local indexes)\n" ); + print ( "--rt-indexer test RT backend and load converted indexes through indexer\n" ); + print ( "--columnar test attrs as columnar\n" ); print ( "--no-drop-db\t\tKeep test db tables after the test (for debugging)\n"); print ( "--keep-all\t\tKeep test db and all test data (like generated configs, etc.) after the test (for debugging)\n"); print ( "--no-demo\t\tJust skip all tests without models. Else - run them, but never fail (for debugging)\n"); @@ -67,6 +68,7 @@ $locals = array(); $locals['rt_mode'] = false; +$locals['rt_indexer_mode'] = false; $locals['columnar_mode'] = false; $locals['testdir'] = ''; $locals['scriptdir'] = ''; @@ -107,6 +109,7 @@ else if ( $arg=="-tt" ) $locals['scriptdir'] = $args[++$i]; else if ( $arg=="--ctest" ) { $locals['ctest'] = true; $ctest = true; $force_guess = false; } else if ( $arg=="--rt" ) $locals['rt_mode'] = true; + else if ( $arg=="--rt-indexer" ) { $locals['rt_mode'] = true; $locals['rt_indexer_mode'] = true; } else if ( $arg=="--columnar" ) $locals['columnar_mode'] = true; else if ( $arg=="--strict" ) $g_strict = true; else if ( $arg=="--strict-verbose" ) { $g_strict = true; $g_strictverbose = true; } @@ -187,6 +190,14 @@ PublishLocals ( $locals, false ); +if ( !empty ( $locals['rt_indexer_mode'] ) ) +{ + if ( $windows ) + die ( "--rt-indexer is not supported on Windows\n" ); + + putenv ( "MANTICORE_NO_BUDDY=1" ); +} + if ( !getenv("MANTICORE_MODULES") ) { $module_dirs = array (