feat: [FC-86] implement multivalue faceted search using Elasticsearch

Makefile

@@ -59,3 +59,25 @@ test: test_with_es ## run tests and generate coverage report
 install-local: ## installs your local edx-search into the LMS and CMS python virtualenvs
 	docker exec -t edx.devstack.lms bash -c '. /edx/app/edxapp/venvs/edxapp/bin/activate && cd /edx/app/edxapp/edx-platform && pip uninstall -y edx-search && pip install -e /edx/src/edx-search && pip freeze | grep edx-search'
 	docker exec -t edx.devstack.cms bash -c '. /edx/app/edxapp/venvs/edxapp/bin/activate && cd /edx/app/edxapp/edx-platform && pip uninstall -y edx-search && pip install -e /edx/src/edx-search && pip freeze | grep edx-search'
+
+test-meili: meili-up
+	@echo "Running Meilisearch tests..."
+	@MEILISEARCH_MASTER_KEY=test_master_key \
+	MEILISEARCH_URL=http://localhost:7700 \
+	pytest -v \
+	    search/tests/test_meilisearch.py \
+	    search/tests/test_course_discovery.py -k Meilisearch \
+	    search/tests/test_course_discovery_views.py -k Meilisearch || true
+	@$(MAKE) meili-down
+
+meili-up:
+	@echo "Starting Meilisearch..."
+	@docker compose up -d test_meilisearch
+	@echo "Waiting for Meilisearch to be healthy..."
+	@timeout 15 bash -c \
+    	'until curl -sf http://localhost:7700/health > /dev/null; do echo "Waiting..."; sleep 1; done'
+
+
+meili-down:
+	@echo "Shutting down Meilisearch..."
+	@docker compose down -v test_meilisearch

docker-compose.yml

@@ -19,6 +19,22 @@ services:
     ports:
       - "9200:9200"
 
+  test_meilisearch:
+    image: getmeili/meilisearch:v1.7
+    container_name: test_meilisearch
+    ports:
+      - "7700:7700"
+    environment:
+      MEILISEARCH_MASTER_KEY: test_master_key
+      MEILISEARCH_URL: http://localhost:7700
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:7700/health"]
+      interval: 2s
+      timeout: 1s
+      retries: 10
+
+version: '3.8'
+
 volumes:
   data01:
     driver: local

edxsearch/settings.py

@@ -26,7 +26,7 @@
 # This is just a container for running tests
 DEBUG = True
 
-ALLOWED_HOSTS = []
+ALLOWED_HOSTS = ['testserver', 'localhost', '127.0.0.1']
 
 TEMPLATES = [
     {
@@ -128,3 +128,6 @@
         }
     }
 }
+
+MEILISEARCH_API_KEY = os.environ.get("MEILISEARCH_MASTER_KEY", "test_master_key")
+MEILISEARCH_URL = os.environ.get("MEILISEARCH_URL", "http://meilisearch")

search/api.py

@@ -122,7 +122,7 @@ def emit_api_timing_event(search_term, course_id, filter_generation_timer, proce
     })
 
 
-def course_discovery_search(search_term=None, size=20, from_=0, field_dictionary=None):
+def course_discovery_search(search_term=None, size=20, from_=0, field_dictionary=None, is_multivalue=False):
     """
     Course Discovery activities against the search engine index of course details
     """
@@ -155,6 +155,7 @@ def course_discovery_search(search_term=None, size=20, from_=0, field_dictionary
         filter_dictionary={"enrollment_end": DateRange(datetime.utcnow(), None)},
         exclude_dictionary=exclude_dictionary,
         aggregation_terms=course_discovery_aggregations(),
+        is_multivalue=is_multivalue,
     )
 
     return results

search/elastic.py

@@ -22,7 +22,7 @@
 RESERVED_CHARACTERS = "+=><!(){}[]^~*:\\/&|?"
 
 
-def _translate_hits(es_response):
+def _translate_hits(es_response, aggregation_terms, is_multivalue=False):
     """
     Provide result set in our desired format from elasticsearch results.
 
@@ -98,7 +98,7 @@ def translate_result(result):
 
     def translate_agg_bucket(bucket, agg_result):
         """
-        Any conversion from ES aggregations result into our search engine syntax
+        Convert ES aggregation result following our search engine syntax.
 
         agg_result argument needs for getting total number of
         documents per bucket.
@@ -108,19 +108,29 @@ def translate_agg_bucket(bucket, agg_result):
         :return: dict
         """
         agg_item = agg_result[bucket]
-        terms = {
-            bucket["key"]: bucket["doc_count"]
-            for bucket in agg_item["buckets"]
-        }
-        total_docs = (
-            agg_result[_get_total_doc_key(bucket)]["value"]
-            + agg_item["sum_other_doc_count"]
-            + agg_item["doc_count_error_upper_bound"]
-        )
+
+        if is_multivalue:
+            values_agg = agg_item["values"]
+            terms = {
+                bucket["key"]: bucket["doc_count"]
+                for bucket in values_agg["buckets"]
+            }
+            total_docs = sum(terms.values())
+        else:
+            terms = {
+                bucket["key"]: bucket["doc_count"]
+                for bucket in agg_item["buckets"]
+            }
+            total_docs = (
+                    agg_result[_get_total_doc_key(bucket)]["value"]
+                    + agg_item["sum_other_doc_count"]
+                    + agg_item["doc_count_error_upper_bound"]
+            )
+
         return {
             "terms": terms,
             "total": total_docs,
-            "other": agg_item["sum_other_doc_count"],
+            "other": agg_item.get("sum_other_doc_count", 0),
         }
 
     results = list(map(translate_result, es_response["hits"]["hits"]))
@@ -131,11 +141,19 @@ def translate_agg_bucket(bucket, agg_result):
         "results": results,
     }
     if "aggregations" in es_response:
-        response["aggs"] = {
-            bucket: translate_agg_bucket(bucket, es_response["aggregations"])
-            for bucket in es_response["aggregations"]
-            if "total_" not in bucket
-        }
+        if is_multivalue:
+            global_aggs = es_response["aggregations"].get("global_aggs", {})
+            response["aggs"] = {
+                facet: translate_agg_bucket(facet, global_aggs)
+                for facet in global_aggs
+                if facet in aggregation_terms
+            }
+        else:
+            response["aggs"] = {
+                bucket: translate_agg_bucket(bucket, es_response["aggregations"])
+                for bucket in es_response["aggregations"]
+                if "total_" not in bucket
+            }
 
     return response
 
@@ -240,6 +258,47 @@ def _process_aggregation_terms(aggregation_terms):
     return elastic_aggs
 
 
+def _process_multivalue_aggregations(aggregation_terms, field_dictionary):
+    """
+    Calculate facet distributions for multi-value faceted search.
+    """
+    aggs = {}
+    for facet_field, options in aggregation_terms.items():
+        filters_excluding_facet = {
+            field: value for field, value in field_dictionary.items()
+            if field != facet_field
+        }
+        filter_clauses = [
+            _get_filter_field(field, value)
+            for field, value in filters_excluding_facet.items()
+            if value
+        ]
+        facet_filter = {
+            "bool": {
+                "must": filter_clauses
+            }
+        } if filter_clauses else {"match_all": {}}
+
+        aggs[facet_field] = {
+            "filter": facet_filter,
+            "aggs": {
+                "values": {
+                    "terms": {
+                        "field": facet_field,
+                        **options
+                    }
+                }
+            }
+        }
+
+    return {
+        "global_aggs": {
+            "global": {},
+            "aggs": aggs
+        }
+    }
+
+
 class ElasticSearchEngine(SearchEngine):
     """
     ElasticSearch implementation of SearchEngine abstraction
@@ -652,8 +711,13 @@ def search(self,
                 }
 
         body = {"query": query}
+
+        is_multivalue = kwargs.pop("is_multivalue", False)
         if aggregation_terms:
-            body["aggs"] = _process_aggregation_terms(aggregation_terms)
+            if is_multivalue:
+                body["aggs"] = _process_multivalue_aggregations(aggregation_terms, field_dictionary)
+            else:
+                body["aggs"] = _process_aggregation_terms(aggregation_terms)
 
         if log_search_params:
             log.info(f"full elastic search body {body}")
@@ -664,4 +728,4 @@ def search(self,
             log.exception("error while searching index - %r", ex)
             raise
 
-        return _translate_hits(es_response)
+        return _translate_hits(es_response, aggregation_terms, is_multivalue)

search/meilisearch.py

@@ -70,6 +70,7 @@
 from django.conf import settings
 from django.utils import timezone
 
+from search.api import course_discovery_filter_fields
 from search.search_engine_base import SearchEngine
 from search.utils import ValueRange
 
@@ -168,6 +169,7 @@ def search(
         """
         See meilisearch docs: https://www.meilisearch.com/docs/reference/api/search
         """
+        is_multivalue = kwargs.pop("is_multivalue", False)
         opt_params = get_search_params(
             field_dictionary=field_dictionary,
             filter_dictionary=filter_dictionary,
@@ -178,8 +180,36 @@ def search(
         if log_search_params:
             logger.info("Search query: opt_params=%s", opt_params)
         meilisearch_results = self.meilisearch_index.search(query_string, opt_params)
-        processed_results = process_results(meilisearch_results, self.index_name)
-        return processed_results
+
+        if is_multivalue:
+            self._expand_facet_distibutions(field_dictionary, query_string, opt_params, meilisearch_results)
+
+        return process_results(meilisearch_results, self.index_name)
+
+    def _expand_facet_distibutions(self, field_dictionary, query_string, opt_params, meilisearch_results):
+        """
+        For each selected facet, get all its available options within the selected filters.
+        """
+        for facet in field_dictionary.keys():
+            expanded_facet_distribution = self._get_expanded_distribution(
+                query_string,
+                facet,
+                opt_params.get("filter", []),
+            )
+            meilisearch_results.setdefault("facetDistribution", {})[facet] = expanded_facet_distribution
+
+    def _get_expanded_distribution(self, query, facet_to_exclude, filter_rules):
+        """
+        Run a secondary query excluding one facet to get its full distribution.
+        Only return distribution data, without any actual results.
+        """
+        secondary_opt_params = {
+            'facets': [facet_to_exclude],
+            'filter': [rule for rule in filter_rules if not rule.startswith(f"{facet_to_exclude} = ")],
+            'limit': 0,
+        }
+        result = self.meilisearch_index.search(query, secondary_opt_params)
+        return result.get("facetDistribution", {}).get(facet_to_exclude, {})
 
     def remove(self, doc_ids, **kwargs):
         """
@@ -407,16 +437,20 @@ def get_filter_rules(
     Convert inclusion/exclusion rules.
     """
     rules = []
-    for key, value in rule_dict.items():
-        if isinstance(value, list):
-            for v in value:
-                rules.append(
-                    get_filter_rule(key, v, exclude=exclude, optional=optional)
-                )
+    filter_fields = course_discovery_filter_fields()
+    for rule_name, rule_value in rule_dict.items():
+        if isinstance(rule_value, list):
+            if rule_name in filter_fields:
+                rules.append(" OR ".join(f'{rule_name} = "{nested_value}"' for nested_value in rule_value))
+            else:
+                rules += [
+                    get_filter_rule(
+                        rule_name, nested_value, exclude=exclude, optional=optional
+                    ) for nested_value in rule_value
+                ]
         else:
-            rules.append(
-                get_filter_rule(key, value, exclude=exclude, optional=optional)
-            )
+            rules.append(get_filter_rule(rule_name, rule_value, exclude=exclude, optional=optional))
+
     return rules