WIP: indexer rt bulk prototype

.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)

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)
 ```
 <!-- end -->
 
-<!-- example bulk_insert -->
 ## 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.
+
+<!-- example bulk_insert -->
 <!-- intro -->
 ### Bulk insert examples
 ##### SQL:

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=$<BOOL:${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 $<TARGET_FILE: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 $<TARGET_FILE:searchd> )
+			SET_TESTS_PROPERTIES ( ${INDEXER_RT_FLUENTBIT_TEST} PROPERTIES LABELS RT_INDEXER RUN_SERIAL TRUE )
+		endif ()
 	endif ()
 
 #	fixup_test_name ( tst "Internal src/tests" )

src/client_session.h

@@ -17,6 +17,7 @@
 #include "queryprofile.h"
 #include "searchdaemon.h"
 #include "searchdsql.h"
+#include <cstdio>
 #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<char[]> m_pIndexerRtBulkBuffer;
+	int m_iIndexerRtBulkPid = -1;
+	int64_t m_iIndexerRtBulkIndexId = -1;
+	CSphString m_sIndexerRtBulkTable;
+	CSphString m_sIndexerRtBulkDir;
+	CSphString m_sIndexerRtBulkConfig;
+	CSphString m_sIndexerRtBulkIndex;
 	CSphVector<int64_t> m_dLastIds;
 	CSphVector<CSphString> m_dLastIdStrings;
 	QueryProfile_c m_tProfile;

src/fileutils.cpp

@@ -15,6 +15,10 @@
 #include "std/crc32.h"
 #include <sys/stat.h>
 
+#if defined(__APPLE__)
+	#include <mach-o/dyld.h>
+#endif
+
 #if _WIN32
 	#define getcwd		_getcwd
 	#include <shlwapi.h>
@@ -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<char> 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;

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";