Introduce a document summary for collapsing and a way to safely roll it out (#1333)

Author: papa99do

src/marqo/core/constants.py

@@ -31,6 +31,7 @@
 MARQO_COLLAPSE_FIELDS_MINIMUM_VERSION = semver.VersionInfo.parse('2.23.0')
 MARQO_TYPEAHEAD_SCHEMA_MINIMUM_VERSION = semver.VersionInfo.parse('2.23.0')
 MARQO_UPDATE_SCHEMA_MINIMUM_VERSION = semver.VersionInfo.parse('2.23.0')
+MARQO_COLLAPSE_MINIMAL_SUMMARY_MINIMUM_VERSION = semver.VersionInfo.parse('2.24.6')
 
 # For score modifiers
 QUERY_INPUT_SCORE_MODIFIERS_MULT_WEIGHTS_2_9 = 'marqo__mult_weights'

src/marqo/core/index_management/index_management.py

@@ -313,6 +313,24 @@ def apply_latest_schema_template(self, index_name: str, force: bool = False, dry
                     f"Please recreate the index with a newer version of Marqo to use this feature."
                 )
 
+            # Validate that index's marqo_version is not greater than current version
+            current_version_parsed = semver.VersionInfo.parse(version.get_version())
+            if existing_index.parsed_marqo_version() > current_version_parsed:
+                raise InternalError(
+                    f"Cannot update schema for index '{index_name}' created with Marqo version "
+                    f"{existing_index.marqo_version} using current Marqo version {version.get_version()}. "
+                    f"The index was created with a newer version of Marqo than is currently running."
+                )
+
+            # Early return if schema is already at current version
+            if existing_index.schema_template_version == version.get_version():
+                logger.info(f'Index {index_name} schema is already at version {version.get_version()}')
+                return {
+                    "updated": False,
+                    "schemaChanged": False,
+                    "reason": f"Schema is already at current Marqo version {version.get_version()}"
+                }
+
             # Generate new schema from current settings using latest template
             new_schema = SemiStructuredVespaSchema.generate_vespa_schema(existing_index)
 
@@ -343,7 +361,6 @@ def apply_latest_schema_template(self, index_name: str, force: bool = False, dry
                 "newSchema": new_schema,
                 "schemaDiff": schema_diff,
                 "configChangeActions": {},
-                "reason": ""
             }
 
             # Scenario 1: Schemas are identical - no changes needed

src/marqo/core/index_management/vespa_application_package.py

@@ -20,7 +20,7 @@
 from marqo.core.models import MarqoIndex
 from marqo.core.typeahead.typeahead_vespa_schema import TypeaheadVespaSchema
 import marqo.logging
-from marqo.vespa.exceptions import VespaError
+from marqo import version as marqo_version
 from marqo.vespa.vespa_client import VespaClient
 
 
@@ -772,7 +772,10 @@ def update_index_setting_and_schema(self, index: MarqoIndex, schema: str,
 
         version = index.version + 1 if index.version is not None else 1
         self._store.save_file(schema, 'schemas', f'{index.schema_name}.sd')
-        self._index_setting_store.save_index_setting(index.copy(update={'version': version}))
+        self._index_setting_store.save_index_setting(index.copy(update={
+            'version': version,
+            'schema_template_version': marqo_version.get_version()
+        }))
         self._persist_index_settings()
 
         if prepare_only:

src/marqo/core/models/marqo_index.py

@@ -540,6 +540,7 @@ class SemiStructuredMarqoIndex(UnstructuredMarqoIndex):
     string_array_fields: Optional[List[
         StringArrayField]]  # This is required so that when saving a document containing string array fields, we can make changes to the schema on the fly. Ref: https://github.com/marqo-ai/marqo/blob/cfea70adea7039d1586c94e36adae8e66cabe306/src/marqo/core/semi_structured_vespa_index/semi_structured_vespa_schema_template_2_16.sd.jinja2#L83
     collapse_fields: Optional[List[CollapseField]] = None
+    schema_template_version: Optional[str] = None
 
     def __init__(self, **data):
         super().__init__(**data)
@@ -560,6 +561,16 @@ def is_collapse_field(self, field_name: str) -> bool:
             return False
         return field_name in [field.name for field in self.collapse_fields]
 
+    def parsed_schema_template_version(self) -> semver.VersionInfo:
+        """
+        Get the schema template version as a semver object.
+        Falls back to marqo_version if schema_template_version is not set (backward compatibility).
+        """
+        if self.schema_template_version is not None:
+            return semver.VersionInfo.parse(self.schema_template_version)
+        # Backward compatibility: old indexes don't have schema_template_version
+        return self.parsed_marqo_version()
+
     @property
     def field_map(self) -> Dict[str, Field]:
         """
@@ -651,6 +662,11 @@ def generate():
 
         return self._cache_or_get('tensor_subfield_map', generate)
 
+    # TODO: Update index_supports_* properties to use parsed_schema_template_version()
+    # instead of parsed_marqo_version() for more accurate feature detection
+    # after schema updates. This will allow features to be detected based on
+    # the deployed schema template version rather than index creation version.
+
     @property
     def index_supports_partial_updates(self) -> bool:
         """
@@ -687,6 +703,17 @@ def index_supports_sorty_by(self) -> bool:
             'index_supports_sort_by',
             lambda: self.parsed_marqo_version() >= constants.MARQO_SORT_BY_MINIMUM_VERSION)
 
+    @property
+    def index_supports_collapse_minimal_summary(self) -> bool:
+        """
+        Check if the index schema supports collapse-minimal-summary.
+        This summary class was added in version 2.24.6 to optimize
+        collapse query performance by returning minimal fields.
+        """
+        return self._cache_or_get(
+            'index_supports_collapse_minimal_summary',
+            lambda: self.parsed_schema_template_version() >= constants.MARQO_COLLAPSE_MINIMAL_SUMMARY_MINIMUM_VERSION)
+
 
 _PROTECTED_FIELD_NAMES = ['_id', '_tensor_facets', '_highlights', '_score', '_found']
 _VESPA_NAME_PATTERN = r'[a-zA-Z_][a-zA-Z0-9_]*'

src/marqo/core/models/marqo_index_request.py

@@ -42,6 +42,7 @@ class UnstructuredMarqoIndexRequest(MarqoIndexRequest):
     treat_urls_and_pointers_as_media: bool
     filter_string_max_length: int
     collapse_fields: Optional[List[marqo_index.CollapseField]] = None
+    schema_template_version: Optional[str] = None  # For testing: allows simulating older schema template versions
 
     @root_validator
     def validate_collapse_fields(cls, values):

src/marqo/core/semi_structured_vespa_index/semi_structured_vespa_index.py

@@ -122,7 +122,7 @@ def _to_vespa_hybrid_query(self, marqo_query):
         return query
 
     def _generate_collapse_query_params(self, collapse_field_name: str):
-        return {
+        params = {
             'collapsefield': collapse_field_name,
             'collapsesize': 1,  # currently fixed to 1, will support multiple if needed in the future
 
@@ -133,6 +133,12 @@ def _generate_collapse_query_params(self, collapse_field_name: str):
             'marqo__ranking.tensor.lexical': common.RANK_PROFILE_HYBRID_EMBEDDING_SIMILARITY_THEN_BM25 + '_diversity',
         }
 
+        # Only use minimal summary if the schema supports it (version check)
+        if self.get_marqo_index().index_supports_collapse_minimal_summary:
+            params['collapse.summary'] = 'collapse-minimal-summary'
+
+        return params
+
     def _add_relevance_cutoff_and_sort_by_params(self, marqo_query, query):
         if marqo_query.relevance_cutoff:
             query["marqo__hybrid.relevanceCutoff.method"] = marqo_query.relevance_cutoff.method

src/marqo/core/semi_structured_vespa_index/semi_structured_vespa_schema.py

@@ -55,6 +55,7 @@ def _generate_marqo_index(self, schema_name: str) -> SemiStructuredMarqoIndex:
             vector_numeric_type=self._index_request.vector_numeric_type,
             hnsw_config=self._index_request.hnsw_config,
             marqo_version=self._index_request.marqo_version,
+            schema_template_version=self._index_request.schema_template_version or self._index_request.marqo_version,
             created_at=self._index_request.created_at,
             updated_at=self._index_request.updated_at,
             lexical_fields=[],

src/marqo/core/semi_structured_vespa_index/semi_structured_vespa_schema_template_2_16.sd.jinja2

@@ -403,4 +403,10 @@ schema {{ index.schema_name }} {
         summary {{ field.embeddings_field_name }} type tensor<float>(p{}, x[{{ dimension }}]) {}
         {%- endfor %}
     }
+
+    {% if collapse_field -%}
+    document-summary collapse-minimal-summary {
+        summary {{ collapse_field.name }} type string {}
+    }
+    {%- endif %}
 }

src/marqo/version.py

@@ -1,4 +1,4 @@
-__version__ = "2.24.5"
+__version__ = "2.24.6"
 
 def get_version() -> str:
     return f"{__version__}"

tests/integ_tests/core/index_management/test_index_management_schema_update.py

@@ -3,6 +3,7 @@
 from datetime import datetime, timedelta
 from unittest.mock import patch
 
+from marqo import version
 from marqo.core.exceptions import IndexNotFoundError, InternalError, UnsupportedFeatureError
 from marqo.core.models.add_docs_params import AddDocsParams
 from marqo.core.models.marqo_index import FieldType
@@ -41,12 +42,16 @@ def setUpClass(cls) -> None:
         # Bootstrap Vespa
         cls.index_management.bootstrap_vespa()
 
+        cls.original_marqo_version = '2.23.0'
+        cls.original_schema_template_version = '2.24.0'
+
         # pre-create all indexes for tests
         index_requests = [
-            cls.unstructured_marqo_index_request(name='test_schema_update_index'),
-            cls.unstructured_marqo_index_request(name='test_config_change_restart'),
-            cls.unstructured_marqo_index_request(name='test_config_change_reindex'),
-            cls.unstructured_marqo_index_request(name='test_config_change_refeed'),
+            cls.unstructured_marqo_index_request(name='test_schema_update_index', marqo_version=cls.original_marqo_version, schema_template_version=cls.original_schema_template_version),
+            cls.unstructured_marqo_index_request(name='test_config_change_restart', marqo_version=cls.original_marqo_version, schema_template_version=cls.original_schema_template_version),
+            cls.unstructured_marqo_index_request(name='test_config_change_reindex', marqo_version=cls.original_marqo_version, schema_template_version=cls.original_schema_template_version),
+            cls.unstructured_marqo_index_request(name='test_config_change_refeed', marqo_version=cls.original_marqo_version, schema_template_version=cls.original_schema_template_version),
+            cls.unstructured_marqo_index_request(name='test_schema_template_version_current'),
             cls.unstructured_marqo_index_request(name='old_version_index', marqo_version='2.22.0'),
             cls.unstructured_marqo_index_request(name='legacy_unstructured_index', marqo_version='2.12.0'),
             cls.structured_marqo_index_request(
@@ -57,12 +62,13 @@ def setUpClass(cls) -> None:
         ]
         cls.create_indexes(index_requests)
 
-        # feed in a doc to add a 'title' lexical field to 'test_config_change_restart'
-        cls.add_documents(cls.config, AddDocsParams(
-            index_name='test_config_change_restart',
-            docs=[{'_id': '1', 'title': 'hello'}],
-            tensor_fields=[]
-        ))
+        # feed in a doc to add a 'title' lexical field to 'test_config_change_restart', keep the schema version unchanged
+        with patch('marqo.version.get_version', return_value=cls.original_schema_template_version):
+            cls.add_documents(cls.config, AddDocsParams(
+                index_name='test_config_change_restart',
+                docs=[{'_id': '1', 'title': 'hello'}],
+                tensor_fields=[]
+            ))
 
     def _get_validation_overrides(self) -> str:
         app = self.vespa_client.download_application()
@@ -96,7 +102,10 @@ def test_update_schema_no_changes(self):
         self.assertEqual(original_schema, result['oldSchema'])
         self.assertEqual(original_schema, result['newSchema'])
 
-        self.assertEqual(original_version, self.index_management.get_index(test_index_name).version)
+        # Verify version and schema_template_version not updated (remains None when no changes)
+        updated_index = self.index_management.get_index(test_index_name)
+        self.assertEqual(self.original_schema_template_version, updated_index.schema_template_version)
+        self.assertEqual(original_version, updated_index.version)
 
     @patch('marqo.core.semi_structured_vespa_index.semi_structured_vespa_schema.SemiStructuredVespaSchema.generate_vespa_schema')
     def test_update_schema_successful(self, mock_generate_schema):
@@ -126,8 +135,11 @@ def test_update_schema_successful(self, mock_generate_schema):
 
         # Verify schema was actually deployed to Vespa
         self.assertEqual(modified_schema, self._get_schema_from_vespa(saved_index.schema_name))
-        # Verify the index version is also updated
-        self.assertEqual(original_version + 1, self.index_management.get_index(test_index_name).version)
+
+        # Verify schema_template_version updated to current version; and index version becomes +1
+        updated_index = self.index_management.get_index(test_index_name)
+        self.assertEqual(version.get_version(), updated_index.schema_template_version)
+        self.assertEqual(original_version + 1, updated_index.version)
 
     @patch('marqo.core.semi_structured_vespa_index.semi_structured_vespa_schema.SemiStructuredVespaSchema.generate_vespa_schema')
     def test_update_schema_dry_run_prevents_deployment(self, mock_generate_schema):
@@ -157,6 +169,10 @@ def test_update_schema_dry_run_prevents_deployment(self, mock_generate_schema):
         # Verify schema was NOT deployed to Vespa (original schema still present)
         self.assertEqual(original_schema, self._get_schema_from_vespa(saved_index.schema_name))
 
+        # Verify schema_template_version not updated in dry run
+        updated_index = self.index_management.get_index(test_index_name)
+        self.assertEqual(self.original_schema_template_version, updated_index.schema_template_version)
+
     # ============================================================================
     # Error Case Tests
     # ============================================================================
@@ -182,7 +198,7 @@ def test_update_schema_wrong_index_type_legacy_unstructured(self):
 
         self.assertIn('only semi-structured indexes support schema updates', str(ctx.exception))
 
-    def test_update_schema_version_too_old(self):
+    def test_update_schema_template_version_too_old(self):
         """Should raise UnsupportedFeatureError for indexes created with Marqo < 2.23.0."""
         with self.assertRaisesStrict(UnsupportedFeatureError) as ctx:
             self.index_management.apply_latest_schema_template('old_version_index')
@@ -191,6 +207,22 @@ def test_update_schema_version_too_old(self):
                       str(ctx.exception))
         self.assertIn('created with Marqo 2.22.0', str(ctx.exception))
 
+    def test_shortcut_when_schema_template_version_current(self):
+        """When schema_template_version matches current version, shortcut is triggered."""
+        current_version = version.get_version()
+        index_name = 'test_schema_template_version_current'
+
+        result = self.index_management.apply_latest_schema_template(index_name)
+
+        # Verify shortcut response
+        self.assertFalse(result['updated'])
+        self.assertFalse(result['schemaChanged'])
+        self.assertIn(f'already at current Marqo version {current_version}', result['reason'])
+
+        # Verify schema_template_version unchanged
+        index = self.index_management.get_index(index_name)
+        self.assertEqual(current_version, index.schema_template_version)
+
     # ============================================================================
     # configChangeActions Tests
     # ============================================================================
@@ -244,6 +276,10 @@ def test_update_schema_with_restart_actions(self, mock_generate_schema):
         # Verify schema was NOT deployed
         self.assertEqual(original_schema, self._get_schema_from_vespa(saved_index.schema_name))
 
+        # Verify schema_template_version not updated when blocked
+        blocked_index = self.index_management.get_index(test_index_name)
+        self.assertEqual(self.original_schema_template_version, blocked_index.schema_template_version)
+
         # Verify it updates the schema when forced set to true
         result_forced = self.index_management.apply_latest_schema_template(
             test_index_name,
@@ -254,6 +290,10 @@ def test_update_schema_with_restart_actions(self, mock_generate_schema):
         self.assertIn('Update forced despite required actions', result_forced['reason'])
         self.assertEqual(modified_schema, self._get_schema_from_vespa(saved_index.schema_name))
 
+        # Verify schema_template_version updated when forced
+        forced_index = self.index_management.get_index(test_index_name)
+        self.assertEqual(version.get_version(), forced_index.schema_template_version)
+
     @patch('marqo.core.semi_structured_vespa_index.semi_structured_vespa_schema.SemiStructuredVespaSchema.generate_vespa_schema')
     def test_update_schema_with_refeed_actions(self, mock_generate_schema):
         """When re-feed actions are required and force=False, should block deployment.
@@ -296,6 +336,10 @@ def test_update_schema_with_refeed_actions(self, mock_generate_schema):
 
         self.assertEqual(original_schema, self._get_schema_from_vespa(saved_index.schema_name))
 
+        # Verify schema_template_version not updated when blocked
+        blocked_index = self.index_management.get_index(test_index_name)
+        self.assertEqual(self.original_schema_template_version, blocked_index.schema_template_version)
+
         # Force update
         result_forced = self.index_management.apply_latest_schema_template(
             test_index_name,
@@ -309,6 +353,10 @@ def test_update_schema_with_refeed_actions(self, mock_generate_schema):
 
         self.assertEqual(modified_schema, self._get_schema_from_vespa(saved_index.schema_name))
 
+        # Verify schema_template_version updated when forced
+        forced_index = self.index_management.get_index(test_index_name)
+        self.assertEqual(version.get_version(), forced_index.schema_template_version)
+
     @patch('marqo.core.semi_structured_vespa_index.semi_structured_vespa_schema.SemiStructuredVespaSchema.generate_vespa_schema')
     def test_update_schema_with_reindex_actions(self, mock_generate_schema):
         """When reindex actions are required and force=False, should block deployment.
@@ -357,6 +405,10 @@ def test_update_schema_with_reindex_actions(self, mock_generate_schema):
 
         self.assertEqual(original_schema, self._get_schema_from_vespa(saved_index.schema_name))
 
+        # Verify schema_template_version not updated when blocked
+        blocked_index = self.index_management.get_index(test_index_name)
+        self.assertEqual(self.original_schema_template_version, blocked_index.schema_template_version)
+
         # Force update
         result_forced = self.index_management.apply_latest_schema_template(
             test_index_name,
@@ -369,3 +421,7 @@ def test_update_schema_with_reindex_actions(self, mock_generate_schema):
         self.assertIn('Update forced despite required actions', result_forced['reason'])
         self.assertIn('reindex', result_forced['configChangeActions'])
         self.assertEqual(modified_schema, self._get_schema_from_vespa(saved_index.schema_name))
+
+        # Verify schema_template_version updated when forced
+        forced_index = self.index_management.get_index(test_index_name)
+        self.assertEqual(version.get_version(), forced_index.schema_template_version)