Rename converted operations from agg to facet, Solr's term

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.
 

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 `<name>_iterations` and related parameters, and the standard value source `workload.py` registers for it. Listed in `CONVERTED.md` |
 
 ## Requires manual review
 

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"(?<![A-Za-z0-9])agg(?![A-Za-z0-9])")
+
 
 def detect_workload_format_from_file(workload_dir: str) -> 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 ``<name>_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'(?<![A-Za-z0-9_-]){quoted}(?=[_-][A-Za-z0-9])', new),
+                ]
+                if in_operations:
+                    patterns.append((rf'("name"\s*:\s*)"{quoted}"', rf'\1"{new}"'))
+                for pattern, replacement in patterns:
+                    text, hits = re.subn(pattern, replacement, text)
+                    count += hits
+            if count:
+                with open(path, "w", encoding="utf-8") as f:
+                    f.write(text)
+                touched[os.path.relpath(path, output_dir)] = count
+    return touched
+
+
+def _write_converted_marker(output_dir: str, source_dir: str, skipped: list, issues: list,
+                            renamed: dict = None):
     """Write a CONVERTED.md marker file documenting the conversion."""
     timestamp = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
     skipped_section = ""
@@ -821,6 +930,16 @@ def _write_converted_marker(output_dir: str, source_dir: str, skipped: list, iss
     if issues:
         issues_section = "\n## Conversion Issues\n\n" + "\n".join(f"- {i}" for i in issues) + "\n"
 
+    renamed_section = ""
+    if renamed:
+        renamed_section = (
+            "\n## Renamed Operations\n\nSolr calls an aggregation a facet, so these "
+            "operations, their `<name>_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.

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