Fix external merge in python index module

keyvi/include/keyvi/index/internal/merge_job.h

@@ -22,8 +22,10 @@
 #include <chrono>  //NOLINT
 #include <functional>
 #include <memory>
+#include <sstream>
 #include <string>
 #include <thread>  // NOLINT
+#include <utility>
 #include <vector>
 
 #include <boost/process/v2/process.hpp>
@@ -157,7 +159,16 @@ class MergeJob final {
   void DoExternalProcessMerge(boost::asio::io_context* external_process_ctx) {
     payload_.start_time_ = std::chrono::system_clock::now();
 
+    std::istringstream iss(payload_.settings_.GetKeyviMergerBin());
+    std::string executable;
+    iss >> executable;
+
     std::vector<std::string> args;
+    std::string token;
+    while (iss >> token) {
+      args.push_back(std::move(token));
+    }
+
     args.push_back("-m");
     args.push_back("5242880");
 
@@ -169,8 +180,7 @@ class MergeJob final {
     args.push_back("-o");
     args.push_back(payload_.output_filename_.string());
 
-    external_process_.reset(
-        new boost::process::v2::process(*external_process_ctx, payload_.settings_.GetKeyviMergerBin(), args));
+    external_process_ = std::make_shared<boost::process::v2::process>(*external_process_ctx, executable, args);
   }
 
   bool TryFinalizeMerge() {

python/src/py/keyvi/_pycore/keyvimerger.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-'''
+"""
 keyvi - A key value store.
 
 Copyright 2018 Hendrik Muhs<hendrik.muhs@gmail.com>
@@ -15,21 +15,31 @@
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
-'''
+"""
 
 import keyvi.compiler
 import sys
 from argparse import ArgumentParser
 
-if __name__ == '__main__':
-    argument_parser = ArgumentParser(description='keyvi')
-
-    argument_parser.add_argument('-i', '--input-file', action='append', default=[], dest='input_file')
-    argument_parser.add_argument('-o', '--output-file', type=str)
-    argument_parser.add_argument('-m', '--memory-limit', type=str, help='amount of main memory to use')
-    argument_parser.add_argument('-p', '--parameter', action='append', default=[], dest='params',
-                              type=lambda kv: kv.split("="),
-                              help='An option; format is -p xxx=yyy')
+if __name__ == "__main__":
+    argument_parser = ArgumentParser(description="keyvi")
+
+    argument_parser.add_argument(
+        "-i", "--input-file", action="append", default=[], dest="input_file"
+    )
+    argument_parser.add_argument("-o", "--output-file", type=str)
+    argument_parser.add_argument(
+        "-m", "--memory-limit", type=str, help="amount of main memory to use"
+    )
+    argument_parser.add_argument(
+        "-p",
+        "--parameter",
+        action="append",
+        default=[],
+        dest="params",
+        type=lambda kv: kv.split("="),
+        help="An option; format is -p xxx=yyy",
+    )
 
     args = argument_parser.parse_args()
 
@@ -41,11 +51,11 @@
 
         merger = keyvi.compiler.JsonDictionaryMerger(params)
         for f in args.input_file:
-            merger.Add(f)
+            merger.add(f)
 
-        merger.Merge(args.output_file)
+        merger.merge(args.output_file)
     else:
-        print ("ERROR: arguments wrong or missing.")
+        print("ERROR: arguments wrong or missing.")
         sys.exit(1)
 
     sys.exit(0)

python/tests/index/index_external_merge_test.py

@@ -0,0 +1,126 @@
+# -*- coding: utf-8 -*-
+# Usage: py.test tests
+
+from keyvi.index import Index, ReadOnlyIndex
+import os
+import tempfile
+import time
+
+
+def test_external_merge():
+    with tempfile.TemporaryDirectory() as test_dir:
+        index_dir = os.path.join(test_dir, "index")
+        index = Index(index_dir, {"segment_external_merge_key_threshold": "100"})
+
+        for batch in range(10):
+            key_values = []
+            for i in range(50):
+                key = "key-{:05d}".format(batch * 50 + i)
+                value = "value-{}".format(batch * 50 + i)
+                key_values.append((key, value))
+            index.bulk_set(key_values)
+            index.flush()
+
+        # wait for external merge to complete
+        time.sleep(5)
+
+        # verify all keys are still accessible
+        for i in range(500):
+            key = "key-{:05d}".format(i)
+            assert key in index, "missing key: {}".format(key)
+            match = index[key]
+            assert match.value == "value-{}".format(i)
+
+
+def test_external_merge_with_deletes():
+    with tempfile.TemporaryDirectory() as test_dir:
+        index_dir = os.path.join(test_dir, "index")
+        index = Index(index_dir, {"segment_external_merge_key_threshold": "100"})
+
+        for batch in range(10):
+            key_values = []
+            for i in range(50):
+                key = "key-{:05d}".format(batch * 50 + i)
+                value = "value-{}".format(batch * 50 + i)
+                key_values.append((key, value))
+            index.bulk_set(key_values)
+            index.flush()
+
+        # delete every other key
+        for i in range(0, 500, 2):
+            index.delete("key-{:05d}".format(i))
+        index.flush()
+
+        # wait for external merge to complete
+        time.sleep(5)
+
+        for i in range(500):
+            key = "key-{:05d}".format(i)
+            if i % 2 == 0:
+                assert key not in index, "key should be deleted: {}".format(key)
+            else:
+                assert key in index, "missing key: {}".format(key)
+                match = index[key]
+                assert match.value == "value-{}".format(i)
+
+
+def test_external_merge_with_overwrite():
+    with tempfile.TemporaryDirectory() as test_dir:
+        index_dir = os.path.join(test_dir, "index")
+        index = Index(index_dir, {"segment_external_merge_key_threshold": "100"})
+
+        # write initial data across multiple segments
+        for batch in range(5):
+            key_values = []
+            for i in range(50):
+                key = "key-{:05d}".format(batch * 50 + i)
+                value = "value-v1-{}".format(batch * 50 + i)
+                key_values.append((key, value))
+            index.bulk_set(key_values)
+            index.flush()
+
+        # overwrite some keys with new values in new segments
+        for batch in range(5):
+            key_values = []
+            for i in range(50):
+                key = "key-{:05d}".format(batch * 50 + i)
+                value = "value-v2-{}".format(batch * 50 + i)
+                key_values.append((key, value))
+            index.bulk_set(key_values)
+            index.flush()
+
+        # wait for external merge to complete
+        time.sleep(5)
+
+        # verify that the latest values are returned
+        for i in range(250):
+            key = "key-{:05d}".format(i)
+            assert key in index, "missing key: {}".format(key)
+            match = index[key]
+            assert match.value == "value-v2-{}".format(i), (
+                "expected v2 value for {}, got {}".format(key, match.value)
+            )
+
+
+def test_external_merge_read_only_index():
+    with tempfile.TemporaryDirectory() as test_dir:
+        index_dir = os.path.join(test_dir, "index")
+        index = Index(index_dir, {"segment_external_merge_key_threshold": "100"})
+
+        for batch in range(10):
+            key_values = []
+            for i in range(50):
+                key = "key-{:05d}".format(batch * 50 + i)
+                value = "value-{}".format(batch * 50 + i)
+                key_values.append((key, value))
+            index.bulk_set(key_values)
+            index.flush()
+
+        # wait for external merge to complete
+        time.sleep(5)
+
+        # open a read-only index and verify all data
+        reader = ReadOnlyIndex(index_dir)
+        for i in range(500):
+            key = "key-{:05d}".format(i)
+            assert key in reader, "missing key in reader: {}".format(key)