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