From a54387c2a5389c0c7d3f6f64408db305ed306fa8 Mon Sep 17 00:00:00 2001 From: Serhiy Bzhezytskyy Date: Fri, 4 Sep 2026 22:33:24 +0300 Subject: [PATCH] Rename converted operations from agg to facet, Solr's term MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An OpenSearch workload names its aggregation operations after the aggregation: nyc_taxis has date_histogram_agg and distance_amount_agg, geonames has country_agg_uncached, country_agg_cached and numeric-term-cardinality-agg-high. Solr calls that a facet, and the workloads already shipped in solr-orbit-workloads use the facet names — so every regeneration produced five operations the checked-in workloads do not have, and re-running the converter over a workload directory would silently rename them back. An operation name is not only the "name" key. It is what a test procedure schedules, the prefix of its _iterations, _target_throughput, _search_clients and _time_period parameters, and the key workload.py registers a standard value source under. Renaming only the definition leaves a workload that loads and then cannot find its value source, so the rewrite covers all of them, longest name first so country_agg is not rewritten inside country_agg_cached. The token is bounded on both sides: aggs, aggregation and the aggs-query-* operations keep their names. The map is built from operations/ and from the operations of workload.json only, never from a test procedure, so a procedure named after an aggregation is left alone. Regenerating nyc_taxis and geonames from the upstream workloads now produces every operation name, test procedure reference, parameter prefix and value source the shipped workloads use, with nothing left only in the checked-in copy, and the only lines that change in the generated output are the names themselves. The converter's docstring promised a result dict of output_dir, issues and skipped operations, which stops being true here, so it and the two converter doc pages are updated with it. --- docs/converter/usage.md | 1 + docs/converter/what-converts.md | 1 + solrorbit/conversion/workload_converter.py | 127 ++++++++++++++++- tests/unit/solr/test_workload_converter.py | 152 +++++++++++++++++++++ 4 files changed, 277 insertions(+), 4 deletions(-) diff --git a/docs/converter/usage.md b/docs/converter/usage.md index 0625c922..7f56dcf0 100644 --- a/docs/converter/usage.md +++ b/docs/converter/usage.md @@ -31,6 +31,7 @@ The converter produces a copy of the workload in the output directory with the f - OpenSearch JSON DSL search bodies translated to Solr JSON query format - Date range filters converted from custom formats (e.g., `dd/MM/yyyy`) to ISO 8601 - Aggregations translated to Solr facet syntax +- Operations named after an aggregation renamed from `agg` to `facet`, along with every reference to them A `CONVERTED.md` file is written to the output directory summarizing what was converted and flagging any items that require manual review. diff --git a/docs/converter/what-converts.md b/docs/converter/what-converts.md index f7b064ef..b9a4cfc0 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 | +| Operation names containing `agg` | Renamed to `facet`, along with every reference to the operation: the test procedures that schedule it, its `_iterations` and related parameters, and the standard value source `workload.py` registers for it. Listed in `CONVERTED.md` | ## Requires manual review diff --git a/solrorbit/conversion/workload_converter.py b/solrorbit/conversion/workload_converter.py index b99dcb6d..ad8da5da 100644 --- a/solrorbit/conversion/workload_converter.py +++ b/solrorbit/conversion/workload_converter.py @@ -27,10 +27,12 @@ The converter: - Renames ``indices`` → ``collections`` and generates schema.xml files from mappings - Renames operation types using the same map as migrate_workload.py + - Renames operations named after an aggregation from ``agg`` to ``facet``, Solr's term, + and rewrites every reference to them - Translates OpenSearch search bodies to Solr JSON Query DSL - Preserves ``corpora`` as-is (dataset files are compatible with both formats) - Writes a ``CONVERTED.md`` marker file to prevent double-conversion - - Returns a summary dict with output_dir, issues, and skipped operations + - Returns a summary dict with output_dir, issues, skipped and renamed operations """ import json @@ -209,6 +211,8 @@ def _load_workload_json(workload_path: str) -> dict: "delete-pipeline", } +_AGG_TOKEN = re.compile(r"(? bool: """ @@ -314,8 +318,12 @@ def convert_opensearch_workload(source_dir: str, output_dir: str) -> dict: # --- Follow external benchmark.collect() refs and make the workload self-contained --- _process_external_collected_files(source_dir, output_dir, issues, skipped) + # --- Rename operations to Solr terminology and update every reference --- + renamed = _collect_operation_renames(output_dir) + _apply_operation_renames(output_dir, renamed) + # --- Write CONVERTED.md marker --- - _write_converted_marker(output_dir, source_dir, skipped, issues) + _write_converted_marker(output_dir, source_dir, skipped, issues, renamed) logger.info( "Workload conversion complete: %s → %s (%d ops, %d skipped, %d issues)", @@ -326,6 +334,7 @@ def convert_opensearch_workload(source_dir: str, output_dir: str) -> dict: "output_dir": os.path.abspath(output_dir), "issues": issues, "skipped": skipped, + "renamed": renamed, } @@ -810,7 +819,107 @@ def replacer(m): _process_one_file(out_file, src_file) -def _write_converted_marker(output_dir: str, source_dir: str, skipped: list, issues: list): +def _solr_operation_name(name: str) -> str: + """Rename an operation from OpenSearch's ``agg`` to Solr's ``facet``.""" + return _AGG_TOKEN.sub("facet", name) + + +def _collect_operation_renames(output_dir: str) -> dict: + """ + Map every converted operation name that carries the ``agg`` token to its Solr name. + + Only ``operations/`` files and the operations of ``workload.json`` — the top-level + array and the ones defined inline in a schedule — are read, so a test procedure that + happens to be named after an aggregation is not mistaken for an operation. + """ + renames = {} + candidates = [] + operations_dir = os.path.join(output_dir, "operations") + if os.path.isdir(operations_dir): + for entry in sorted(os.listdir(operations_dir)): + if entry.endswith(".json"): + candidates.append(os.path.join(operations_dir, entry)) + + for path in candidates: + with open(path, encoding="utf-8") as f: + text = f.read() + for name in re.findall(r'"name"\s*:\s*"([^"]+)"', text): + solr_name = _solr_operation_name(name) + if solr_name != name: + renames[name] = solr_name + + workload_json = os.path.join(output_dir, "workload.json") + if os.path.isfile(workload_json): + with open(workload_json, encoding="utf-8") as f: + try: + data = _parse_jinja_fragment(f.read())[0] + except ValueError: + data = None + if isinstance(data, dict): + inline = list(data.get("operations", [])) + for procedures in ("challenges", "test_procedures"): + for procedure in data.get(procedures, []): + if not isinstance(procedure, dict): + continue + for task in procedure.get("schedule", []): + if isinstance(task, dict) and isinstance(task.get("operation"), dict): + inline.append(task["operation"]) + for op in inline: + name = op.get("name") if isinstance(op, dict) else None + if name and _solr_operation_name(name) != name: + renames[name] = _solr_operation_name(name) + return renames + + +def _apply_operation_renames(output_dir: str, renames: dict) -> dict: + """ + Rewrite every reference to a renamed operation across the converted workload. + + An operation's name is also the prefix its ``_iterations``-style parameters + are built from, the string ``workload.py`` registers a value source under, and the + string a test procedure schedules. The rewrite is textual because a test procedure + whose Jinja2 cannot be parsed as JSON is copied verbatim, so there is no object to + walk; longest name first, so ``x-agg`` cannot be rewritten inside ``x-agg-cached``. + + Returns the number of references rewritten per file, relative to ``output_dir``. + """ + if not renames: + return {} + ordered = sorted(renames, key=len, reverse=True) + touched = {} + for root, dirs, files in os.walk(output_dir): + dirs[:] = [d for d in dirs if d not in {"__pycache__", ".git", "configsets"}] + for entry in sorted(files): + if not entry.endswith((".json", ".py")): + continue + path = os.path.join(root, entry) + with open(path, encoding="utf-8") as f: + original = f.read() + text = original + count = 0 + in_operations = os.path.basename(root) == "operations" or entry == "workload.json" + for old in ordered: + new = renames[old] + quoted = re.escape(old) + patterns = [ + (rf'("operation"\s*:\s*)"{quoted}"', rf'\1"{new}"'), + (rf'(register_[a-z_]+\(\s*["\']){quoted}(["\'])', rf'\1{new}\2'), + (rf'(?_iterations`-style parameters and every test " + "procedure that schedules them were renamed:\n\n" + + "\n".join(f"- `{old}` → `{new}`" for old, new in sorted(renamed.items())) + + "\n" + ) + content = f"""# Workload Conversion Record This workload was automatically converted from OpenSearch Benchmark format to @@ -831,7 +950,7 @@ def _write_converted_marker(output_dir: str, source_dir: str, skipped: list, iss - **Source workload**: `{os.path.abspath(source_dir)}` - **Converted at**: `{timestamp}` - **Converter version**: solr.conversion.workload_converter v1.0 -{skipped_section}{issues_section} +{skipped_section}{issues_section}{renamed_section} ## Notes - Search operation bodies have been translated to Solr JSON Query DSL format. diff --git a/tests/unit/solr/test_workload_converter.py b/tests/unit/solr/test_workload_converter.py index 423dc9c7..923c9ffe 100644 --- a/tests/unit/solr/test_workload_converter.py +++ b/tests/unit/solr/test_workload_converter.py @@ -24,6 +24,9 @@ from solrorbit.conversion.workload_converter import ( CONVERTED_MARKER, + _apply_operation_renames, + _collect_operation_renames, + _solr_operation_name, convert_opensearch_workload, detect_workload_format_from_file, is_already_converted, @@ -339,6 +342,155 @@ def test_value_count_metric(self): self.assertEqual("countvals(vendor_id)", result["doc_count"]) +class TestSolrOperationName(unittest.TestCase): + def test_agg_token_becomes_facet(self): + self.assertEqual("date_histogram_facet", _solr_operation_name("date_histogram_agg")) + self.assertEqual("country_facet_uncached", _solr_operation_name("country_agg_uncached")) + self.assertEqual("distance_amount_facet", _solr_operation_name("distance_amount_agg")) + + def test_hyphenated_names(self): + self.assertEqual( + "numeric-term-cardinality-facet-high", + _solr_operation_name("numeric-term-cardinality-agg-high"), + ) + + def test_agg_as_a_substring_is_left_alone(self): + for name in ("aggs-query-large", "aggregation-heavy", "baggage_claim", "agglomerate"): + self.assertEqual(name, _solr_operation_name(name)) + + +class TestCollectOperationRenames(unittest.TestCase): + def _write(self, root, rel, text): + path = os.path.join(root, rel) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as f: + f.write(text) + + def test_reads_operations_directory(self): + with tempfile.TemporaryDirectory() as out: + self._write(out, "operations/default.json", json.dumps([ + {"name": "date_histogram_agg", "operation-type": "search"}, + {"name": "default", "operation-type": "search"}, + ])) + self.assertEqual( + {"date_histogram_agg": "date_histogram_facet"}, + _collect_operation_renames(out), + ) + + def test_reads_inline_schedule_operations(self): + with tempfile.TemporaryDirectory() as out: + self._write(out, "workload.json", json.dumps({ + "collections": [], + "test_procedures": [{ + "name": "default", + "schedule": [{"operation": {"name": "country_agg", "operation-type": "search"}}], + }], + })) + self.assertEqual({"country_agg": "country_facet"}, _collect_operation_renames(out)) + + def test_a_test_procedure_named_after_an_aggregation_is_not_renamed(self): + """`streaming-agg-clickbench` is a test procedure, not an operation.""" + with tempfile.TemporaryDirectory() as out: + self._write(out, "operations/default.json", json.dumps( + [{"name": "default", "operation-type": "search"}])) + self._write(out, "test_procedures/streaming.json", json.dumps( + {"name": "streaming-agg-clickbench", "schedule": [{"operation": "default"}]})) + self.assertEqual({}, _collect_operation_renames(out)) + + +class TestApplyOperationRenames(unittest.TestCase): + def _make_workload(self, out): + os.makedirs(os.path.join(out, "operations")) + os.makedirs(os.path.join(out, "test_procedures")) + with open(os.path.join(out, "operations", "default.json"), "w") as f: + f.write(json.dumps([ + {"name": "country_agg", "operation-type": "search"}, + {"name": "country_agg_cached", "operation-type": "search"}, + ], indent=2)) + with open(os.path.join(out, "test_procedures", "default.json"), "w") as f: + f.write( + '[\n' + ' {\n' + ' "operation": "country_agg",\n' + ' "iterations": {{country_agg_iterations | default(100)}},\n' + ' "clients": {{country_agg_search_clients | default(1)}}\n' + ' },\n' + ' {\n' + ' "operation": "country_agg_cached",\n' + ' "iterations": {{country_agg_cached_iterations | default(100)}}\n' + ' }\n' + ']\n' + ) + with open(os.path.join(out, "workload.py"), "w") as f: + f.write( + 'def register(registry):\n' + ' registry.register_standard_value_source("country_agg", src)\n' + ) + + def test_every_reference_is_rewritten(self): + with tempfile.TemporaryDirectory() as out: + self._make_workload(out) + renames = _collect_operation_renames(out) + self.assertEqual({ + "country_agg": "country_facet", + "country_agg_cached": "country_facet_cached", + }, renames) + _apply_operation_renames(out, renames) + + with open(os.path.join(out, "operations", "default.json")) as f: + ops = json.load(f) + self.assertEqual(["country_facet", "country_facet_cached"], + [op["name"] for op in ops]) + + with open(os.path.join(out, "test_procedures", "default.json")) as f: + procedure = f.read() + self.assertIn('"operation": "country_facet"', procedure) + self.assertIn('"operation": "country_facet_cached"', procedure) + self.assertIn("country_facet_iterations", procedure) + self.assertIn("country_facet_search_clients", procedure) + self.assertIn("country_facet_cached_iterations", procedure) + self.assertNotIn("agg", procedure) + + with open(os.path.join(out, "workload.py")) as f: + self.assertIn('register_standard_value_source("country_facet"', f.read()) + + def test_no_renames_leaves_the_tree_untouched(self): + with tempfile.TemporaryDirectory() as out: + self._make_workload(out) + before = os.path.getmtime(os.path.join(out, "operations", "default.json")) + self.assertEqual({}, _apply_operation_renames(out, {})) + self.assertEqual(before, os.path.getmtime(os.path.join(out, "operations", "default.json"))) + + +class TestConvertRenamesOperations(unittest.TestCase): + def test_inline_operation_is_renamed_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": { + "name": "date_histogram_agg", + "operation-type": "search", + "body": {"query": {"match_all": {}}}, + } + }], + }], + }, f) + result = convert_opensearch_workload(src, dst) + self.assertEqual({"date_histogram_agg": "date_histogram_facet"}, result["renamed"]) + + with open(os.path.join(dst, "workload.json")) as f: + out = json.load(f) + operation = out["challenges"][0]["schedule"][0]["operation"] + self.assertEqual("date_histogram_facet", operation["name"]) + + with open(os.path.join(dst, CONVERTED_MARKER)) as f: + self.assertIn("`date_histogram_agg` → `date_histogram_facet`", f.read()) + + class TestCalendarIntervalToSolrGap(unittest.TestCase): def test_known_intervals(self): self.assertEqual("+1DAY", _calendar_interval_to_solr_gap("day"))