This is an automated email from the ASF dual-hosted git repository.
janhoy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/solr-orbit.git
The following commit(s) were added to refs/heads/main by this push:
new e132cbbd Convert OpenSearch's index-stats merge wait to Solr's
wait-for-merges (#80)
e132cbbd is described below
commit e132cbbdc8216a4a6c04892940bc32b6c078757a
Author: Serhiy Bzhezytskyy <[email protected]>
AuthorDate: Thu Oct 15 10:16:17 2026 +0300
Convert OpenSearch's index-stats merge wait to Solr's wait-for-merges (#80)
---
docs/converter/what-converts.md | 1 +
solrorbit/conversion/workload_converter.py | 92 +++++++++++++++-
tests/unit/solr/test_workload_converter.py | 164 +++++++++++++++++++++++++++++
3 files changed, 255 insertions(+), 2 deletions(-)
diff --git a/docs/converter/what-converts.md b/docs/converter/what-converts.md
index f7b064ef..fdaeeed2 100644
--- a/docs/converter/what-converts.md
+++ b/docs/converter/what-converts.md
@@ -23,6 +23,7 @@ The table below summarizes which OpenSearch Benchmark
constructs are automatical
| `terms` aggregations | Converted to Solr `terms` facet |
| `date_histogram` aggregations | Converted to Solr `range` facet with
calendar gap |
| `avg` / `sum` / `min` / `max` aggregations | Converted to Solr function
query stats |
+| `index-stats` retried until `merges.current` reaches zero | →
`wait-for-merges`, with the polling expressed as `retry-wait-period` and
`max-wait-seconds` |
## Requires manual review
diff --git a/solrorbit/conversion/workload_converter.py
b/solrorbit/conversion/workload_converter.py
index b99dcb6d..025fbd8b 100644
--- a/solrorbit/conversion/workload_converter.py
+++ b/solrorbit/conversion/workload_converter.py
@@ -209,6 +209,18 @@ _UNSUPPORTED_OPS = {
"delete-pipeline",
}
+# An index-stats operation polling this condition path is a wait-for-merges in
Solr
+_MERGE_WAIT_CONDITION_PATH = "merges.current"
+_MERGE_WAIT_RETRY_WAIT_PERIOD = 2.0
+_MERGE_WAIT_MAX_WAIT_SECONDS = 600
+
+_MERGE_WAIT_OBJECT = re.compile(
+ r'\{[^{}]*?"operation-type"\s*:\s*"index-stats"\s*,'
+ r'[^{}]*?"condition"\s*:\s*\{[^{}]*?"path"\s*:\s*"[^"]*'
+ + re.escape(_MERGE_WAIT_CONDITION_PATH)
+ + r'"[^{}]*?\}[^{}]*?\}'
+)
+
def detect_workload_format_from_file(workload_dir: str) -> bool:
"""
@@ -455,7 +467,12 @@ def _process_collected_files(source_dir: str, output_dir:
str, issues: list, ski
ops_list, tokens = _parse_jinja_fragment(raw, wrap_array=True)
except ValueError as exc:
issues.append(f"{subdir}/{filename}: cannot parse as JSON
fragment ({exc}); copied verbatim")
- shutil.copy2(src_path, dst_path)
+ merges_converted = _convert_merge_wait_text(raw)
+ if merges_converted == raw:
+ shutil.copy2(src_path, dst_path)
+ else:
+ with open(dst_path, "w", encoding="utf-8") as f:
+ f.write(merges_converted)
continue
# Convert each operation in the fragment; filter out skipped ones
@@ -552,6 +569,73 @@ def _has_auto_date_histogram(aggs: dict) -> bool:
return False
+def _is_merge_wait(op: dict) -> bool:
+ """
+ Return True if *op* is OpenSearch Benchmark's "wait until merges finish"
idiom:
+ an index-stats operation retried until the active merge count reaches zero.
+ """
+ condition = op.get("condition")
+ if not isinstance(condition, dict):
+ return False
+ path = condition.get("path")
+ if not isinstance(path, str) or not
path.endswith(_MERGE_WAIT_CONDITION_PATH):
+ return False
+ return condition.get("expected-value") == 0
+
+
+def _convert_merge_wait(op: dict) -> None:
+ """
+ Rewrite *op* in-place as a Solr ``wait-for-merges`` operation.
+
+ The polling OpenSearch expresses with ``condition`` and
``retry-until-success`` is
+ the runner's own loop in Solr, so those keys are replaced by its wait
parameters.
+ """
+ converted = {}
+ if "name" in op:
+ converted["name"] = op["name"]
+ converted["operation-type"] = "wait-for-merges"
+ converted["retry-wait-period"] = _MERGE_WAIT_RETRY_WAIT_PERIOD
+ converted["max-wait-seconds"] = _MERGE_WAIT_MAX_WAIT_SECONDS
+ if "include-in-reporting" in op:
+ converted["include-in-reporting"] = op["include-in-reporting"]
+
+ replaced = {"name", "operation-type", "type", "index", "indices",
"condition",
+ "retry-until-success", "include-in-reporting"}
+ converted.update({key: value for key, value in op.items() if key not in
replaced})
+
+ op.clear()
+ op.update(converted)
+
+
+def _convert_merge_wait_text(text: str) -> str:
+ """
+ Rewrite every merge-wait operation object in *text* as ``wait-for-merges``.
+
+ Used for fragments whose Jinja2 directives keep them from being parsed as
JSON,
+ where the operation object itself is still plain JSON.
+ """
+ def replace(match):
+ matched = match.group(0)
+ try:
+ op = json.loads(matched)
+ except json.JSONDecodeError:
+ return matched
+ if not isinstance(op, dict) or not _is_merge_wait(op):
+ return matched
+
+ _convert_merge_wait(op)
+ line_start = text.rfind("\n", 0, match.start()) + 1
+ indent = text[line_start:match.start()]
+ indent = indent[:len(indent) - len(indent.lstrip())]
+
+ first_key = re.match(r"\{\s*\n([ \t]*)", matched)
+ step = len(first_key.group(1)) - len(indent) if first_key else 4
+ rendered = json.dumps(op, indent=max(step, 1))
+ return rendered.replace("\n", "\n" + indent)
+
+ return _MERGE_WAIT_OBJECT.sub(replace, text)
+
+
def _convert_operation(op, issues, skipped, source_dir, output_dir):
"""Convert an operation definition dict in-place.
@@ -560,6 +644,10 @@ def _convert_operation(op, issues, skipped, source_dir,
output_dir):
op_type = op.get("operation-type") or op.get("type", "")
op_name = op.get("name", op_type)
+ if op_type == "index-stats" and _is_merge_wait(op):
+ _convert_merge_wait(op)
+ return True
+
if op_type in _UNSUPPORTED_OPS:
logger.warning("Skipping unsupported operation '%s' (type: %s)",
op_name, op_type)
skipped.append(op_name)
@@ -752,7 +840,7 @@ def _process_external_collected_files(source_dir: str,
output_dir: str, issues:
return _serialise_jinja_fragment(ops_list, tokens, wrap_array=True)
except ValueError:
# Complex Jinja2 — fall back to text substitution for known
op-type strings
- result = raw
+ result = _convert_merge_wait_text(raw)
for old_op, new_op in _OP_MAP.items():
if old_op != new_op:
result = re.sub(
diff --git a/tests/unit/solr/test_workload_converter.py
b/tests/unit/solr/test_workload_converter.py
index 480117bd..29640f7e 100644
--- a/tests/unit/solr/test_workload_converter.py
+++ b/tests/unit/solr/test_workload_converter.py
@@ -24,6 +24,9 @@ import unittest
from solrorbit.conversion.workload_converter import (
CONVERTED_MARKER,
+ _convert_merge_wait,
+ _convert_merge_wait_text,
+ _convert_operation,
convert_opensearch_workload,
detect_workload_format_from_file,
is_already_converted,
@@ -466,5 +469,166 @@ class TestDateHistogramFacetGap(unittest.TestCase):
self.assertEqual("+1MONTH", gap)
+class TestMergeWaitConversion(unittest.TestCase):
+ """Tests for OpenSearch's index-stats merge wait becoming Solr's
wait-for-merges."""
+
+ def _osb_merge_wait(self):
+ return {
+ "name": "wait-until-merges-finish",
+ "operation-type": "index-stats",
+ "index": "_all",
+ "condition": {"path": "_all.total.merges.current",
"expected-value": 0},
+ "retry-until-success": True,
+ "include-in-reporting": False,
+ }
+
+ def test_converted_operation_is_the_solr_operation(self):
+ op = self._osb_merge_wait()
+ _convert_merge_wait(op)
+ self.assertEqual({
+ "name": "wait-until-merges-finish",
+ "operation-type": "wait-for-merges",
+ "retry-wait-period": 2.0,
+ "max-wait-seconds": 600,
+ "include-in-reporting": False,
+ }, op)
+
+ def test_convert_operation_leaves_other_index_stats_alone(self):
+ op = {"name": "index-stats", "operation-type": "index-stats"}
+ _convert_operation(op, [], [], "", "")
+ self.assertEqual("index-stats", op["operation-type"])
+
+ def test_convert_operation_leaves_another_condition_path_alone(self):
+ op = self._osb_merge_wait()
+ op["condition"]["path"] = "_all.total.docs.count"
+ _convert_operation(op, [], [], "", "")
+ self.assertEqual("index-stats", op["operation-type"])
+
+ def test_text_conversion_of_a_jinja_fragment(self):
+ raw = """{
+ "operation": {
+ "operation-type": "force-merge",
+ "request-timeout": {{ request_timeout | default(60) | tojson }}
+ }
+},
+{
+ "name": "wait-until-merges-finish",
+ "operation": {
+ "operation-type": "index-stats",
+ "index": "_all",
+ "condition": {
+ "path": "_all.total.merges.current",
+ "expected-value": 0
+ },
+ "retry-until-success": true,
+ "include-in-reporting": false
+ }
+}"""
+ converted = _convert_merge_wait_text(raw)
+ self.assertNotIn("index-stats", converted)
+ self.assertIn('"operation-type": "wait-for-merges"', converted)
+ self.assertIn('"retry-wait-period": 2.0', converted)
+ self.assertIn('"max-wait-seconds": 600', converted)
+ self.assertIn("{{ request_timeout | default(60) | tojson }}",
converted)
+
+ def test_text_conversion_follows_the_fragment_indentation(self):
+ template = """{{
+{i}"name": "wait-until-merges-finish",
+{i}"operation": {{
+{i}{i}"operation-type": "index-stats",
+{i}{i}"index": "_all",
+{i}{i}"condition": {{
+{i}{i}{i}"path": "_all.total.merges.current",
+{i}{i}{i}"expected-value": 0
+{i}{i}}},
+{i}{i}"include-in-reporting": false
+{i}}}
+}}"""
+ for indent in (" ", " "):
+ with self.subTest(indent=len(indent)):
+ converted = _convert_merge_wait_text(template.format(i=indent))
+ self.assertIn(f'\n{indent * 2}"operation-type":
"wait-for-merges",', converted)
+ self.assertIn(f"\n{indent}}}\n}}", converted)
+
+ def test_text_conversion_leaves_another_condition_alone(self):
+ raw = """{
+ "operation": {
+ "operation-type": "index-stats",
+ "index": "_all",
+ "condition": {
+ "path": "_all.total.docs.count",
+ "expected-value": 0
+ }
+ }
+}"""
+ self.assertEqual(raw, _convert_merge_wait_text(raw))
+
+ def test_conversion_of_an_external_fragment_end_to_end(self):
+ """The merge wait reaches the output even from a fragment Jinja2 keeps
unparseable."""
+ fragment = """{
+ "operation": {
+ "operation-type": "force-merge",
+ "request-timeout": {{ request_timeout | default(60) | tojson }}{%- if
max_num_segments is defined %},
+ "max-num-segments": {{ max_num_segments | tojson }}
+ {%- endif %}
+ }
+},
+{
+ "name": "wait-until-merges-finish",
+ "operation": {
+ "operation-type": "index-stats",
+ "index": "_all",
+ "condition": {
+ "path": "_all.total.merges.current",
+ "expected-value": 0
+ },
+ "retry-until-success": true,
+ "include-in-reporting": false
+ }
+}"""
+ with tempfile.TemporaryDirectory() as tmpdir,
tempfile.TemporaryDirectory() as dst:
+ shared = os.path.join(tmpdir, "common_operations")
+ os.makedirs(shared)
+ with open(os.path.join(shared, "force_merge.json"), "w") as f:
+ f.write(fragment)
+
+ src = os.path.join(tmpdir, "my_workload")
+ os.makedirs(os.path.join(src, "test_procedures"))
+ with open(os.path.join(src, "workload.json"), "w") as f:
+ json.dump({"indices": [], "challenges": []}, f)
+ with open(os.path.join(src, "test_procedures", "default.json"),
"w") as f:
+ f.write('{\n "name": "default",\n "schedule": [\n'
+ ' {{
benchmark.collect(parts="../../common_operations/force_merge.json") }}\n'
+ " ]\n}")
+
+ convert_opensearch_workload(src, dst)
+
+ with open(os.path.join(dst, "common_operations",
"force_merge.json")) as f:
+ out = f.read()
+ self.assertNotIn("index-stats", out)
+ self.assertIn('"operation-type": "wait-for-merges"', out)
+ self.assertIn('"max-wait-seconds": 600', out)
+
+ def test_conversion_end_to_end(self):
+ with tempfile.TemporaryDirectory() as src,
tempfile.TemporaryDirectory() as dst:
+ with open(os.path.join(src, "workload.json"), "w") as f:
+ json.dump({
+ "indices": [],
+ "challenges": [
+ {
+ "name": "default",
+ "schedule": [{"operation":
self._osb_merge_wait()}],
+ }
+ ],
+ }, f)
+ convert_opensearch_workload(src, dst)
+ with open(os.path.join(dst, "workload.json")) as f:
+ out = json.load(f)
+ op = out["challenges"][0]["schedule"][0]["operation"]
+ self.assertEqual("wait-for-merges", op["operation-type"])
+ self.assertEqual(600, op["max-wait-seconds"])
+ self.assertNotIn("condition", op)
+
+
if __name__ == "__main__":
unittest.main()
