Repository: mlflow/mlflow

Issue:
[FR] Introduce page_token argument to MlflowClient.search_model_versions
### Willingness to contribute

No. I cannot contribute this feature at this time.

### Proposal Summary

Introduce a `page_token` argument to https://mlflow.org/docs/latest/python_api/mlflow.client.html#mlflow.client.MlflowClient.search_model_versions in the MLflow Python client.

### Motivation

> #### What is the use case for this feature? Currently, some MLflow backends (e.g. RestStore) limit the number of results returned by `search_model_versions` requests. Clients must fetch subsequent results by making new requests using a provided page token. Unfortunately, the MLflow Python client's https://mlflow.org/docs/latest/python_api/mlflow.client.html#mlflow.client.MlflowClient.search_model_versions method doesn't accept page tokens, so users can't easily load the next page of results.

> #### Why is this use case valuable to support for MLflow users in general? ^ 

> #### Why is this use case valuable to support for your project(s) or organization? ^ 

> #### Why is it currently difficult to achieve this use case? https://mlflow.org/docs/latest/python_api/mlflow.client.html#mlflow.client.MlflowClient.search_model_versions doesn't accept a page token


### Details

We'll need to introduce a `page_token=None` argument to https://mlflow.org/docs/latest/python_api/mlflow.client.html#mlflow.client.MlflowClient.search_model_versions. We'll also need to introduce `page_token=None` to https://github.com/mlflow/mlflow/blob/7f2152370b8e93ed7bf8cc3e5e1e7ff39cd69ca5/mlflow/store/model_registry/abstract_store.py#L217 and all stores that extend from the Model Registry `AbstractStore`.

### What component(s) does this bug affect?

- [ ] `area/artifacts`: Artifact stores and artifact logging
- [ ] `area/build`: Build and test infrastructure for MLflow
- [ ] `area/docs`: MLflow documentation pages
- [ ] `area/examples`: Example code
- [X] `area/model-registry`: Model Registry service, APIs, and the fluent client calls for Model Registry
- [ ] `area/models`: MLmodel format, model serialization/deserialization, flavors
- [ ] `area/recipes`: Recipes, Recipe APIs, Recipe configs, Recipe Templates
- [ ] `area/projects`: MLproject format, project running backends
- [ ] `area/scoring`: MLflow Model server, model deployment tools, Spark UDFs
- [ ] `area/server-infra`: MLflow Tracking server backend
- [ ] `area/tracking`: Tracking Service, tracking client APIs, autologging

### What interface(s) does this bug affect?

- [ ] `area/uiux`: Front-end, user experience, plotting, JavaScript, JavaScript dev server
- [ ] `area/docker`: Docker use across MLflow's components, such as MLflow Projects and MLflow Models
- [ ] `area/sqlalchemy`: Use of SQLAlchemy in the Tracking Service or Model Registry
- [ ] `area/windows`: Windows support

### What language(s) does this bug affect?

- [ ] `language/r`: R APIs and clients
- [ ] `language/java`: Java APIs and clients
- [ ] `language/new`: Proposals for new client languages

### What integration(s) does this bug affect?

- [ ] `integrations/azure`: Azure and Azure ML integrations
- [ ] `integrations/sagemaker`: SageMaker integrations
- [ ] `integrations/databricks`: Databricks integrations

Discussion:
mlflow-automation: 
<!-- documentation preview -->

Documentation preview for 18efcc55c989eba7939d70f6623f9b643b8f74a5 will be available [here](https://output.circle-artifacts.com/output/job/7683952d-ce66-4985-813d-26b58fd3c607/artifacts/0/docs/build/html/index.html) when [this CircleCI job](https://circleci.com/gh/mlflow/mlflow/38997) completes successfully.

<details>
<summary>More info</summary>

- Ignore this comment if this PR does not change the documentation.
- It takes a few minutes for the preview to be available.
- The preview is updated when a new commit is pushed to this PR.
- This comment was created by https://github.com/mlflow/mlflow/actions/runs/3800248182.

</details>

serena-ruan: Hi @dbczumar, now for searching model versions I added order by name ASC and version ASC by default, should I order by version DESC instead? Because the larger the number the newer the version is, want to hear your thoughts on this :)
WeichenXu123: > Hi @dbczumar, now for searching model versions I added order by name ASC and version ASC by default, should I order by version DESC instead? Because the larger the number the newer the version is, want to hear your thoughts on this :)

Shall we order by created date desc ?
serena-ruan: @dbczumar I'm ready for review :D
serena-ruan: @BenWilson2 @harupy Could you help me rerun the cancelled tasks? And if it passes I'm waiting for approval :D
BenWilson2: @serena-ruan the tasks were cancelled due to a timeout from pip resolution tree logic. Could you merge master branch into your PR (that should fix it :) )?
serena-ruan: Hi @dbczumar, @WeichenXu123 do you have any more comments? (Or approval :D)
harupy: @serena-ruan Let me test this on Databricks.
harupy: I tested on Databricks and it worked :)
serena-ruan: @harupy Should I update this newly added 'mlflow/store/_unity_catalog/registry/rest_store.py' to follow the same function pattern of search_model_versions? The proto definition doesn't include order_by in it as for now.
mlflow-automation: 
Failed to find a documentation preview for 8262f8201e6335f46ae13d7a3bc8d7997275cbac.

<details>
<summary>More info</summary>

- If the `ci/circleci: build_doc` job status is successful, you can see the preview with the following steps:
  1. Click `Details`.
  2. Click `Artifacts`.
  3. Click `docs/build/html/index.html`.
- This comment was created by https://github.com/mlflow/mlflow/actions/runs/4229153155.

</details>

serena-ruan: > @harupy Should I update this newly added 'mlflow/store/_unity_catalog/registry/rest_store.py' to follow the same function pattern of search_model_versions? The proto definition doesn't include order_by in it as for now.

The failing lint test is for this newly added `mlflow/store/_unity_catalog/registry/rest_store.py`, should we fix it or ignore pylint here?
harupy: @serena-ruan I pushed a commit to fix the failed checks.

Patch:
diff --git a/mlflow/java/client/src/main/java/org/mlflow/api/proto/ModelRegistry.java b/mlflow/java/client/src/main/java/org/mlflow/api/proto/ModelRegistry.java
index d7d4f392b..2762ffdf0 100644
--- a/mlflow/java/client/src/main/java/org/mlflow/api/proto/ModelRegistry.java
+++ b/mlflow/java/client/src/main/java/org/mlflow/api/proto/ModelRegistry.java
@@ -27093,7 +27093,8 @@ public final class ModelRegistry {
 
     /**
      * <pre>
-     * Maximum number of models desired. Max threshold is 200K.
+     * Maximum number of models desired. Max threshold is 200K. Backends may choose a lower default
+     * value and maximum threshold.
      * </pre>
      *
      * <code>optional int64 max_results = 2 [default = 200000];</code>
@@ -27102,7 +27103,8 @@ public final class ModelRegistry {
     boolean hasMaxResults();
     /**
      * <pre>
-     * Maximum number of models desired. Max threshold is 200K.
+     * Maximum number of models desired. Max threshold is 200K. Backends may choose a lower default
+     * value and maximum threshold.
      * </pre>
      *
      * <code>optional int64 max_results = 2 [default = 200000];</code>
@@ -28509,7 +28511,8 @@ public final class ModelRegistry {
     private long maxResults_;
     /**
      * <pre>
-     * Maximum number of models desired. Max threshold is 200K.
+     * Maximum number of models desired. Max threshold is 200K. Backends may choose a lower default
+     * value and maximum threshold.
      * </pre>
      *
      * <code>optional int64 max_results = 2 [default = 200000];</code>
@@ -28521,7 +28524,8 @@ public final class ModelRegistry {
     }
     /**
      * <pre>
-     * Maximum number of models desired. Max threshold is 200K.
+     * Maximum number of models desired. Max threshold is 200K. Backends may choose a lower default
+     * value and maximum threshold.
      * </pre>
      *
      * <code>optional int64 max_results = 2 [default = 200000];</code>
@@ -29173,7 +29177,8 @@ public final class ModelRegistry {
       private long maxResults_ = 200000L;
       /**
        * <pre>
-       * Maximum number of models desired. Max threshold is 200K.
+       * Maximum number of models desired. Max threshold is 200K. Backends may choose a lower default
+       * value and maximum threshold.
        * </pre>
        *
        * <code>optional int64 max_results = 2 [default = 200000];</code>
@@ -29185,7 +29190,8 @@ public final class ModelRegistry {
       }
       /**
        * <pre>
-       * Maximum number of models desired. Max threshold is 200K.
+       * Maximum number of models desired. Max threshold is 200K. Backends may choose a lower default
+       * value and maximum threshold.
        * </pre>
        *
        * <code>optional int64 max_results = 2 [default = 200000];</code>
@@ -29197,7 +29203,8 @@ public final class ModelRegistry {
       }
       /**
        * <pre>
-       * Maximum number of models desired. Max threshold is 200K.
+       * Maximum number of models desired. Max threshold is 200K. Backends may choose a lower default
+       * value and maximum threshold.
        * </pre>
        *
        * <code>optional int64 max_results = 2 [default = 200000];</code>
@@ -29212,7 +29219,8 @@ public final class ModelRegistry {
       }
       /**
        * <pre>
-       * Maximum number of models desired. Max threshold is 200K.
+       * Maximum number of models desired. Max threshold is 200K. Backends may choose a lower default
+       * value and maximum threshold.
        * </pre>
        *
        * <code>optional int64 max_results = 2 [default = 200000];</code>
diff --git a/mlflow/protos/model_registry.proto b/mlflow/protos/model_registry.proto
index 84627e7fd..45ddfade2 100644
--- a/mlflow/protos/model_registry.proto
+++ b/mlflow/protos/model_registry.proto
@@ -529,7 +529,8 @@ message SearchModelVersions {
   // with string values wrapped in single quotes.
   optional string filter = 1;
 
-  // Maximum number of models desired. Max threshold is 200K.
+  // Maximum number of models desired. Max threshold is 200K. Backends may choose a lower default
+  // value and maximum threshold.
   optional int64 max_results = 2 [default = 200000];
 
   // List of columns to be ordered by including model name, version, stage with an
diff --git a/mlflow/server/handlers.py b/mlflow/server/handlers.py
index 43300fd3a..2c75acd72 100644
--- a/mlflow/server/handlers.py
+++ b/mlflow/server/handlers.py
@@ -1508,10 +1508,26 @@ def _get_model_version_download_uri():
 @catch_mlflow_exception
 @_disable_if_artifacts_only
 def _search_model_versions():
-    request_message = _get_request_message(SearchModelVersions())
-    model_versions = _get_model_registry_store().search_model_versions(request_message.filter)
+    request_message = _get_request_message(
+        SearchModelVersions(),
+        schema={
+            "filter": [_assert_string],
+            "max_results": [_assert_intlike, lambda x: _assert_less_than_or_equal(x, 200_000)],
+            "order_by": [_assert_array, _assert_item_type_string],
+            "page_token": [_assert_string],
+        },
+    )
+    store = _get_model_registry_store()
+    model_versions = store.search_model_versions(
+        filter_string=request_message.filter,
+        max_results=request_message.max_results,
+        order_by=request_message.order_by,
+        page_token=request_message.page_token,
+    )
     response_message = SearchModelVersions.Response()
     response_message.model_versions.extend([e.to_proto() for e in model_versions])
+    if model_versions.token:
+        response_message.next_page_token = model_versions.token
     return _wrap_response(response_message)
 
 
diff --git a/mlflow/store/_unity_catalog/registry/rest_store.py b/mlflow/store/_unity_catalog/registry/rest_store.py
index d5ab405dc..bc9fd6911 100644
--- a/mlflow/store/_unity_catalog/registry/rest_store.py
+++ b/mlflow/store/_unity_catalog/registry/rest_store.py
@@ -339,15 +339,23 @@ class UcModelRegistryStore(BaseRestStore):
         response_proto = self._call_endpoint(GetModelVersionDownloadUriRequest, req_body)
         return response_proto.artifact_uri
 
-    def search_model_versions(self, filter_string):
+    def search_model_versions(
+        self, filter_string=None, max_results=None, order_by=None, page_token=None
+    ):
         """
         Search for model versions in backend that satisfy the filter criteria.
 
         :param filter_string: A filter string expression. Currently supports a single filter
                               condition either name of model like ``name = 'model_name'`` or
                               ``run_id = '...'``.
-        :return: PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
-                 objects.
+        :param max_results: Maximum number of model versions desired.
+        :param order_by: List of column names with ASC|DESC annotation, to be used for ordering
+                         matching search results.
+        :param page_token: Token specifying the next page of results. It should be obtained from
+                            a ``search_model_versions`` call.
+        :return: A PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
+                 objects that satisfy the search expressions. The pagination token for the next
+                 page can be obtained via the ``token`` attribute of the object.
         """
         req_body = message_to_json(SearchModelVersionsRequest(filter=filter_string))
         response_proto = self._call_endpoint(SearchModelVersionsRequest, req_body)
diff --git a/mlflow/store/model_registry/__init__.py b/mlflow/store/model_registry/__init__.py
index 091c7481b..e172912ca 100644
--- a/mlflow/store/model_registry/__init__.py
+++ b/mlflow/store/model_registry/__init__.py
@@ -3,3 +3,5 @@ DEFAULT_LOCAL_FILE_AND_ARTIFACT_PATH = "./mlruns"
 
 SEARCH_REGISTERED_MODEL_MAX_RESULTS_DEFAULT = 100
 SEARCH_REGISTERED_MODEL_MAX_RESULTS_THRESHOLD = 1000
+SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT = 200_000
+SEARCH_MODEL_VERSION_MAX_RESULTS_THRESHOLD = 200_000
diff --git a/mlflow/store/model_registry/abstract_store.py b/mlflow/store/model_registry/abstract_store.py
index 20d0fb529..f427bba11 100644
--- a/mlflow/store/model_registry/abstract_store.py
+++ b/mlflow/store/model_registry/abstract_store.py
@@ -215,15 +215,23 @@ class AbstractStore:
         pass
 
     @abstractmethod
-    def search_model_versions(self, filter_string):
+    def search_model_versions(
+        self, filter_string=None, max_results=None, order_by=None, page_token=None
+    ):
         """
         Search for model versions in backend that satisfy the filter criteria.
 
         :param filter_string: A filter string expression. Currently supports a single filter
                               condition either name of model like ``name = 'model_name'`` or
                               ``run_id = '...'``.
-        :return: PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
-                 objects.
+        :param max_results: Maximum number of model versions desired.
+        :param order_by: List of column names with ASC|DESC annotation, to be used for ordering
+                         matching search results.
+        :param page_token: Token specifying the next page of results. It should be obtained from
+                            a ``search_model_versions`` call.
+        :return: A PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
+                 objects that satisfy the search expressions. The pagination token for the next
+                 page can be obtained via the ``token`` attribute of the object.
         """
         pass
 
diff --git a/mlflow/store/model_registry/file_store.py b/mlflow/store/model_registry/file_store.py
index 75a545c70..32244f8ab 100644
--- a/mlflow/store/model_registry/file_store.py
+++ b/mlflow/store/model_registry/file_store.py
@@ -30,6 +30,7 @@ from mlflow.store.model_registry.abstract_store import AbstractStore
 from mlflow.store.model_registry import (
     DEFAULT_LOCAL_FILE_AND_ARTIFACT_PATH,
     SEARCH_REGISTERED_MODEL_MAX_RESULTS_THRESHOLD,
+    SEARCH_MODEL_VERSION_MAX_RESULTS_THRESHOLD,
 )
 from mlflow.utils.search_utils import SearchUtils, SearchModelUtils, SearchModelVersionUtils
 from mlflow.utils.string_utils import is_string_type
@@ -525,6 +526,7 @@ class FileStore(AbstractStore):
                 creation_time = get_current_time_millis()
                 registered_model = self.get_registered_model(name)
                 registered_model.last_updated_timestamp = creation_time
+                self._save_registered_model_as_meta_file(registered_model)
                 version = next_version(name)
                 model_version = ModelVersion(
                     name=name,
@@ -688,29 +690,56 @@ class FileStore(AbstractStore):
             model_versions.append(self._get_model_version_from_dir(directory))
         return model_versions
 
-    def search_model_versions(self, filter_string):
+    def search_model_versions(
+        self, filter_string=None, max_results=None, order_by=None, page_token=None
+    ):
         """
         Search for model versions in backend that satisfy the filter criteria.
 
         :param filter_string: A filter string expression. Currently supports a single filter
                               condition either name of model like ``name = 'model_name'`` or
                               ``run_id = '...'``.
-        :return: PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
-                 objects.
+        :param max_results: Maximum number of model versions desired.
+        :param order_by: List of column names with ASC|DESC annotation, to be used for ordering
+                         matching search results.
+        :param page_token: Token specifying the next page of results. It should be obtained from
+                            a ``search_model_versions`` call.
+        :return: A PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
+                 objects that satisfy the search expressions. The pagination token for the next
+                 page can be obtained via the ``token`` attribute of the object.
         """
-        # should we filter name directly if the condition is about name?
-        # do all model versions under the same registered_model have the same name?
+        if not isinstance(max_results, int) or max_results < 1:
+            raise MlflowException(
+                "Invalid value for max_results. It must be a positive integer,"
+                f" but got {max_results}",
+                INVALID_PARAMETER_VALUE,
+            )
+
+        if max_results > SEARCH_MODEL_VERSION_MAX_RESULTS_THRESHOLD:
+            raise MlflowException(
+                "Invalid value for request parameter max_results. It must be at most "
+                f"{SEARCH_MODEL_VERSION_MAX_RESULTS_THRESHOLD}, but got value {max_results}",
+                INVALID_PARAMETER_VALUE,
+            )
+
         registered_model_paths = self._get_all_registered_model_paths()
         model_versions = []
         for path in registered_model_paths:
             model_versions.extend(self._list_model_versions_under_path(path))
         filtered_mvs = SearchModelVersionUtils.filter(model_versions, filter_string)
-        filtered_mvs = sorted(
-            (mv for mv in filtered_mvs if mv.current_stage != STAGE_DELETED_INTERNAL),
-            key=lambda mv: mv.last_updated_timestamp,
-            reverse=True,
+
+        sorted_mvs = SearchModelVersionUtils.sort(
+            filtered_mvs,
+            order_by or ["last_updated_timestamp DESC", "name ASC", "version_number DESC"],
         )
-        return PagedList(filtered_mvs, None)
+        start_offset = SearchUtils.parse_start_offset_from_page_token(page_token)
+        final_offset = start_offset + max_results
+
+        paginated_mvs = sorted_mvs[start_offset:final_offset]
+        next_page_token = None
+        if final_offset < len(sorted_mvs):
+            next_page_token = SearchUtils.create_page_token(final_offset)
+        return PagedList(paginated_mvs, next_page_token)
 
     def _get_registered_model_version_tag_path(self, name, version, tag_name):
         _validate_model_name(name)
diff --git a/mlflow/store/model_registry/rest_store.py b/mlflow/store/model_registry/rest_store.py
index 6f46b57dd..b4c21a25d 100644
--- a/mlflow/store/model_registry/rest_store.py
+++ b/mlflow/store/model_registry/rest_store.py
@@ -299,17 +299,32 @@ class RestStore(BaseRestStore):
         response_proto = self._call_endpoint(GetModelVersionDownloadUri, req_body)
         return response_proto.artifact_uri
 
-    def search_model_versions(self, filter_string):
+    def search_model_versions(
+        self, filter_string=None, max_results=None, order_by=None, page_token=None
+    ):
         """
         Search for model versions in backend that satisfy the filter criteria.
 
         :param filter_string: A filter string expression. Currently supports a single filter
                               condition either name of model like ``name = 'model_name'`` or
                               ``run_id = '...'``.
-        :return: PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
-                 objects.
+        :param max_results: Maximum number of model versions desired.
+        :param order_by: List of column names with ASC|DESC annotation, to be used for ordering
+                         matching search results.
+        :param page_token: Token specifying the next page of results. It should be obtained from
+                            a ``search_model_versions`` call.
+        :return: A PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
+                 objects that satisfy the search expressions. The pagination token for the next
+                 page can be obtained via the ``token`` attribute of the object.
         """
-        req_body = message_to_json(SearchModelVersions(filter=filter_string))
+        req_body = message_to_json(
+            SearchModelVersions(
+                filter=filter_string,
+                max_results=max_results,
+                order_by=order_by,
+                page_token=page_token,
+            )
+        )
         response_proto = self._call_endpoint(SearchModelVersions, req_body)
         model_versions = [ModelVersion.from_proto(mvd) for mvd in response_proto.model_versions]
         return PagedList(model_versions, response_proto.next_page_token)
diff --git a/mlflow/store/model_registry/sqlalchemy_store.py b/mlflow/store/model_registry/sqlalchemy_store.py
index 8ec88ae50..8c13c6206 100644
--- a/mlflow/store/model_registry/sqlalchemy_store.py
+++ b/mlflow/store/model_registry/sqlalchemy_store.py
@@ -21,6 +21,8 @@ import mlflow.store.db.utils
 from mlflow.store.model_registry import (
     SEARCH_REGISTERED_MODEL_MAX_RESULTS_DEFAULT,
     SEARCH_REGISTERED_MODEL_MAX_RESULTS_THRESHOLD,
+    SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT,
+    SEARCH_MODEL_VERSION_MAX_RESULTS_THRESHOLD,
 )
 from mlflow.store.db.base_sql_model import Base
 from mlflow.store.entities.paged_list import PagedList
@@ -282,6 +284,13 @@ class SqlAlchemyStore(AbstractStore):
             sql_registered_model = self._get_registered_model(session, name)
             session.delete(sql_registered_model)
 
+    def _compute_next_token(self, max_results_for_query, current_size, offset, max_results):
+        next_token = None
+        if max_results_for_query == current_size:
+            final_offset = offset + max_results
+            next_token = SearchUtils.create_page_token(final_offset)
+        return next_token
+
     def search_registered_models(
         self,
         filter_string=None,
@@ -321,13 +330,6 @@ class SqlAlchemyStore(AbstractStore):
         # this remediates having to make another query which returns no items.
         max_results_for_query = max_results + 1
 
-        def compute_next_token(current_size):
-            next_token = None
-            if max_results_for_query == current_size:
-                final_offset = offset + max_results
-                next_token = SearchUtils.create_page_token(final_offset)
-            return next_token
-
         with self.ManagedSessionMaker() as session:
             query = (
                 filter_query.options(*self._get_eager_registered_model_query_options())
@@ -337,7 +339,9 @@ class SqlAlchemyStore(AbstractStore):
             if page_token:
                 query = query.offset(offset)
             sql_registered_models = session.execute(query).scalars(SqlRegisteredModel).all()
-            next_page_token = compute_next_token(len(sql_registered_models))
+            next_page_token = self._compute_next_token(
+                max_results_for_query, len(sql_registered_models), offset, max_results
+            )
             rm_entities = [rm.to_mlflow_entity() for rm in sql_registered_models][:max_results]
             return PagedList(rm_entities, next_page_token)
 
@@ -411,18 +415,34 @@ class SqlAlchemyStore(AbstractStore):
             comparator = f["comparator"]
             value = f["value"]
             if type_ == "attribute":
-                if key not in ("name", "source_path", "run_id"):
+                if key not in SearchModelVersionUtils.VALID_SEARCH_ATTRIBUTE_KEYS:
                     raise MlflowException(
                         f"Invalid attribute name: {key}", error_code=INVALID_PARAMETER_VALUE
                     )
-                if comparator not in ("=", "!=", "LIKE", "ILIKE", "IN") or (
-                    comparator == "IN" and key != "run_id"
+                if key in SearchModelVersionUtils.NUMERIC_ATTRIBUTES:
+                    if (
+                        comparator
+                        not in SearchModelVersionUtils.VALID_NUMERIC_ATTRIBUTE_COMPARATORS
+                    ):
+                        raise MlflowException(
+                            f"Invalid comparator for attribute {key}: {comparator}",
+                            error_code=INVALID_PARAMETER_VALUE,
+                        )
+                elif (
+                    comparator not in SearchModelVersionUtils.VALID_STRING_ATTRIBUTE_COMPARATORS
+                    or (comparator == "IN" and key != "run_id")
                 ):
                     raise MlflowException(
                         f"Invalid comparator for attribute: {comparator}",
                         error_code=INVALID_PARAMETER_VALUE,
                     )
-                attr = getattr(SqlModelVersion, "source" if key == "source_path" else key)
+                if key == "source_path":
+                    key_name = "source"
+                elif key == "version_number":
+                    key_name = "version"
+                else:
+                    key_name = key
+                attr = getattr(SqlModelVersion, key_name)
                 if comparator == "IN":
                     # Note: Here the run_id values in databases contain only lower case letters,
                     # so we already filter out comparison values containing upper case letters
@@ -823,31 +843,118 @@ class SqlAlchemyStore(AbstractStore):
             sql_model_version = self._get_sql_model_version(session, name, version)
             return sql_model_version.source
 
-    def search_model_versions(self, filter_string):
+    def search_model_versions(
+        self,
+        filter_string=None,
+        max_results=SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT,
+        order_by=None,
+        page_token=None,
+    ):
         """
         Search for model versions in backend that satisfy the filter criteria.
 
         :param filter_string: A filter string expression. Currently supports a single filter
                               condition either name of model like ``name = 'model_name'`` or
                               ``run_id = '...'``.
-        :return: PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
-                 objects sorted by last updated time in descending order.
+        :param max_results: Maximum number of model versions desired.
+        :param order_by: List of column names with ASC|DESC annotation, to be used for ordering
+                         matching search results.
+        :param page_token: Token specifying the next page of results. It should be obtained from
+                            a ``search_model_versions`` call.
+        :return: A PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
+                 objects that satisfy the search expressions. The pagination token for the next
+                 page can be obtained via the ``token`` attribute of the object.
         """
+        if not isinstance(max_results, int) or max_results < 1:
+            raise MlflowException(
+                "Invalid value for max_results. It must be a positive integer,"
+                f" but got {max_results}",
+                INVALID_PARAMETER_VALUE,
+            )
+
+        if max_results > SEARCH_MODEL_VERSION_MAX_RESULTS_THRESHOLD:
+            raise MlflowException(
+                "Invalid value for request parameter max_results. It must be at most "
+                f"{SEARCH_MODEL_VERSION_MAX_RESULTS_THRESHOLD}, but got value {max_results}",
+                INVALID_PARAMETER_VALUE,
+            )
+
         parsed_filters = SearchModelVersionUtils.parse_search_filter(filter_string)
 
         filter_query = self._get_search_model_versions_filter_clauses(
             parsed_filters, self.engine.dialect.name
         )
 
+        parsed_orderby = self._parse_search_model_versions_order_by(
+            order_by or ["last_updated_timestamp DESC", "name ASC", "version_number DESC"]
+        )
+        offset = SearchUtils.parse_start_offset_from_page_token(page_token)
+        # we query for max_results + 1 items to check whether there is another page to return.
+        # this remediates having to make another query which returns no items.
+        max_results_for_query = max_results + 1
+
         with self.ManagedSessionMaker() as session:
-            stmt = (
+            query = (
                 filter_query.options(*self._get_eager_model_version_query_options())
                 .filter(SqlModelVersion.current_stage != STAGE_DELETED_INTERNAL)
-                .order_by(SqlModelVersion.last_updated_time.desc())
+                .order_by(*parsed_orderby)
+                .limit(max_results_for_query)
             )
-            sql_model_versions = session.execute(stmt).scalars(SqlModelVersion).all()
-            model_versions = [mv.to_mlflow_entity() for mv in sql_model_versions]
-            return PagedList(model_versions, None)
+            if page_token:
+                query = query.offset(offset)
+            sql_model_versions = session.execute(query).scalars(SqlModelVersion).all()
+            next_page_token = self._compute_next_token(
+                max_results_for_query, len(sql_model_versions), offset, max_results
+            )
+            model_versions = [mv.to_mlflow_entity() for mv in sql_model_versions][:max_results]
+            return PagedList(model_versions, next_page_token)
+
+    @classmethod
+    def _parse_search_model_versions_order_by(cls, order_by_list):
+        """Sorts a set of model versions based on their natural ordering and an overriding set
+        of order_bys. Model versions are naturally ordered first by name ascending, then by
+        version ascending.
+        """
+        clauses = []
+        observed_order_by_clauses = set()
+        if order_by_list:
+            for order_by_clause in order_by_list:
+                (
+                    _,
+                    key,
+                    ascending,
+                ) = SearchModelVersionUtils.parse_order_by_for_search_model_versions(
+                    order_by_clause
+                )
+                if key not in SearchModelVersionUtils.VALID_ORDER_BY_ATTRIBUTE_KEYS:
+                    raise MlflowException(
+                        f"Invalid order by key '{key}' specified. "
+                        "Valid keys are "
+                        f"{SearchModelVersionUtils.VALID_ORDER_BY_ATTRIBUTE_KEYS}",
+                        error_code=INVALID_PARAMETER_VALUE,
+                    )
+                else:
+                    if key == "version_number":
+                        field = SqlModelVersion.version
+                    elif key == "creation_timestamp":
+                        field = SqlModelVersion.creation_time
+                    elif key == "last_updated_timestamp":
+                        field = SqlModelVersion.last_updated_time
+                    else:
+                        field = getattr(SqlModelVersion, key)
+                if field.key in observed_order_by_clauses:
+                    raise MlflowException(f"`order_by` contains duplicate fields: {order_by_list}")
+                observed_order_by_clauses.add(field.key)
+                if ascending:
+                    clauses.append(field.asc())
+                else:
+                    clauses.append(field.desc())
+
+        if SqlModelVersion.name.key not in observed_order_by_clauses:
+            clauses.append(SqlModelVersion.name.asc())
+        if SqlModelVersion.version.key not in observed_order_by_clauses:
+            clauses.append(SqlModelVersion.version.desc())
+        return clauses
 
     @classmethod
     def _get_model_version_tag(cls, session, name, version, key):
diff --git a/mlflow/tracking/_model_registry/client.py b/mlflow/tracking/_model_registry/client.py
index c9f0f701f..838335ce5 100644
--- a/mlflow/tracking/_model_registry/client.py
+++ b/mlflow/tracking/_model_registry/client.py
@@ -9,7 +9,10 @@ from time import sleep
 import logging
 
 from mlflow.exceptions import MlflowException
-from mlflow.store.model_registry import SEARCH_REGISTERED_MODEL_MAX_RESULTS_DEFAULT
+from mlflow.store.model_registry import (
+    SEARCH_REGISTERED_MODEL_MAX_RESULTS_DEFAULT,
+    SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT,
+)
 from mlflow.entities.model_registry import RegisteredModelTag, ModelVersionTag
 from mlflow.entities.model_registry.model_version_status import ModelVersionStatus
 from mlflow.tracking._model_registry import utils, DEFAULT_AWAIT_MAX_SLEEP_SECONDS
@@ -275,16 +278,29 @@ class ModelRegistryClient:
         """
         return self.store.get_model_version_download_uri(name, version)
 
-    def search_model_versions(self, filter_string):
+    def search_model_versions(
+        self,
+        filter_string=None,
+        max_results=SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT,
+        order_by=None,
+        page_token=None,
+    ):
         """
         Search for model versions in backend that satisfy the filter criteria.
 
         :param filter_string: A filter string expression. Currently supports a single filter
                               condition either name of model like ``name = 'model_name'`` or
                               ``run_id = '...'``.
-        :return: PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion` objects.
+        :param max_results: Maximum number of model versions desired.
+        :param order_by: List of column names with ASC|DESC annotation, to be used for ordering
+                         matching search results.
+        :param page_token: Token specifying the next page of results. It should be obtained from
+                            a ``search_model_versions`` call.
+        :return: A PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
+                 objects that satisfy the search expressions. The pagination token for the next
+                 page can be obtained via the ``token`` attribute of the object.
         """
-        return self.store.search_model_versions(filter_string)
+        return self.store.search_model_versions(filter_string, max_results, order_by, page_token)
 
     def get_model_version_stages(self, name, version):
         """
diff --git a/mlflow/tracking/client.py b/mlflow/tracking/client.py
index cc170f916..8f81892f9 100644
--- a/mlflow/tracking/client.py
+++ b/mlflow/tracking/client.py
@@ -19,7 +19,10 @@ from mlflow.entities.model_registry import RegisteredModel, ModelVersion
 from mlflow.entities.model_registry.model_version_stages import ALL_STAGES
 from mlflow.exceptions import MlflowException
 from mlflow.protos.databricks_pb2 import FEATURE_DISABLED
-from mlflow.store.model_registry import SEARCH_REGISTERED_MODEL_MAX_RESULTS_DEFAULT
+from mlflow.store.model_registry import (
+    SEARCH_REGISTERED_MODEL_MAX_RESULTS_DEFAULT,
+    SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT,
+)
 from mlflow.store.tracking import SEARCH_MAX_RESULTS_DEFAULT
 from mlflow.tracking._model_registry.client import ModelRegistryClient
 from mlflow.tracking._model_registry import utils as registry_utils
@@ -2623,7 +2626,13 @@ class MlflowClient:
         """
         return self._get_registry_client().get_model_version_download_uri(name, version)
 
-    def search_model_versions(self, filter_string: str) -> PagedList[ModelVersion]:
+    def search_model_versions(
+        self,
+        filter_string: Optional[str] = None,
+        max_results: int = SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT,
+        order_by: Optional[List[str]] = None,
+        page_token: Optional[str] = None,
+    ) -> PagedList[ModelVersion]:
         """
         Search for model versions in backend that satisfy the filter criteria.
 
@@ -2649,7 +2658,14 @@ class MlflowClient:
             Logical operators
               - ``AND``: Combines two sub-queries and returns True if both of them are True.
 
-        :return: PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion` objects.
+        :param max_results: Maximum number of model versions desired.
+        :param order_by: List of column names with ASC|DESC annotation, to be used for ordering
+                         matching search results.
+        :param page_token: Token specifying the next page of results. It should be obtained from
+                            a ``search_model_versions`` call.
+        :return: A PagedList of :py:class:`mlflow.entities.model_registry.ModelVersion`
+                 objects that satisfy the search expressions. The pagination token for the next
+                 page can be obtained via the ``token`` attribute of the object.
 
         .. code-block:: python
             :caption: Example
@@ -2684,7 +2700,9 @@ class MlflowClient:
             ------------------------------------------------------------------------------------
             name=CordobaWeatherForecastModel; run_id=e14afa2f47a040728060c1699968fd43; version=2
         """
-        return self._get_registry_client().search_model_versions(filter_string)
+        return self._get_registry_client().search_model_versions(
+            filter_string, max_results, order_by, page_token
+        )
 
     def get_model_version_stages(
         self, name: str, version: str  # pylint: disable=unused-argument
diff --git a/mlflow/utils/search_utils.py b/mlflow/utils/search_utils.py
index 779da5334..e1eb68665 100644
--- a/mlflow/utils/search_utils.py
+++ b/mlflow/utils/search_utils.py
@@ -18,6 +18,7 @@ from sqlparse.tokens import Token as TokenType
 from packaging.version import Version
 
 from mlflow.entities import RunInfo
+from mlflow.entities.model_registry.model_version_stages import STAGE_DELETED_INTERNAL
 from mlflow.exceptions import MlflowException
 from mlflow.protos.databricks_pb2 import INVALID_PARAMETER_VALUE
 from mlflow.store.db.db_types import MYSQL, MSSQL, SQLITE, POSTGRES
@@ -1087,11 +1088,18 @@ class SearchModelUtils(SearchUtils):
 
 
 class SearchModelVersionUtils(SearchUtils):
-    NUMERIC_ATTRIBUTES = {"creation_timestamp", "last_updated_timestamp"}
+    NUMERIC_ATTRIBUTES = {"version_number", "creation_timestamp", "last_updated_timestamp"}
     VALID_SEARCH_ATTRIBUTE_KEYS = {
         "name",
+        "version_number",
         "run_id",
-        "source_path",  # this is for sql store only
+        "source_path",
+    }
+    VALID_ORDER_BY_ATTRIBUTE_KEYS = {
+        "name",
+        "version_number",
+        "creation_timestamp",
+        "last_updated_timestamp",
     }
     VALID_STRING_ATTRIBUTE_COMPARATORS = {"!=", "=", "LIKE", "ILIKE", "IN"}
 
@@ -1102,17 +1110,14 @@ class SearchModelVersionUtils(SearchUtils):
         value = sed.get("value")
         comparator = sed.get("comparator").upper()
 
-        # what comparators do we support here?
         if cls.is_string_attribute(key_type, key, comparator):
-            if key == "source_path":
-                lhs = getattr(mv, key, getattr(mv, "source"))
-            else:
-                lhs = getattr(mv, key)
+            lhs = getattr(mv, "source" if key == "source_path" else key)
         elif cls.is_numeric_attribute(key_type, key, comparator):
+            if key == "version_number":
+                key = "version"
             lhs = getattr(mv, key)
             value = int(value)
         elif cls.is_tag(key_type, comparator):
-            # if the filter doesn't apply, do we return False or?
             lhs = mv.tags.get(key, None)
         else:
             raise MlflowException(
@@ -1129,6 +1134,7 @@ class SearchModelVersionUtils(SearchUtils):
     @classmethod
     def filter(cls, model_versions, filter_string):  # pylint: disable=arguments-renamed
         """Filters a set of model versions based on a search filter string."""
+        model_versions = [mv for mv in model_versions if mv.current_stage != STAGE_DELETED_INTERNAL]
         if not filter_string:
             return model_versions
         parsed = cls.parse_search_filter(filter_string)
@@ -1138,6 +1144,41 @@ class SearchModelVersionUtils(SearchUtils):
 
         return [mv for mv in model_versions if model_version_matches(mv)]
 
+    @classmethod
+    def parse_order_by_for_search_model_versions(cls, order_by):
+        token_value, is_ascending = cls._parse_order_by_string(order_by)
+        identifier = SearchExperimentsUtils._get_identifier(
+            token_value.strip(), cls.VALID_ORDER_BY_ATTRIBUTE_KEYS
+        )
+        return identifier["type"], identifier["key"], is_ascending
+
+    @classmethod
+    def _get_sort_key(cls, order_by_list):
+        order_by = []
+        parsed_order_by = map(cls.parse_order_by_for_search_model_versions, order_by_list or [])
+        for type_, key, ascending in parsed_order_by:
+            if type_ == "attribute":
+                # Need to add this mapping because version is a keyword in sql
+                if key == "version_number":
+                    key = "version"
+                order_by.append((key, ascending))
+            else:
+                raise MlflowException.invalid_parameter_value(f"Invalid order_by entity: {type_}")
+
+        # Add a tie-breaker
+        if not any(key == "name" for key, _ in order_by):
+            order_by.append(("name", True))
+        if not any(key == "version_number" for key, _ in order_by):
+            order_by.append(("version", False))
+
+        return lambda model_version: tuple(
+            _apply_reversor(model_version, k, asc) for (k, asc) in order_by
+        )
+
+    @classmethod
+    def sort(cls, model_versions, order_by_list):  # pylint: disable=arguments-renamed
+        return sorted(model_versions, key=cls._get_sort_key(order_by_list))
+
     @classmethod
     def _get_model_version_search_identifier(cls, identifier, valid_attributes):
         tokens = identifier.split(".", maxsplit=1)
@@ -1171,9 +1212,56 @@ class SearchModelVersionUtils(SearchUtils):
         left, comparator, right = stripped_comparison
         comp = cls._get_model_version_search_identifier(left.value, cls.VALID_SEARCH_ATTRIBUTE_KEYS)
         comp["comparator"] = comparator.value.upper()
-        comp["value"] = SearchModelUtils._get_value(comp.get("type"), comp.get("key"), right)
+        comp["value"] = cls._get_value(comp.get("type"), comp.get("key"), right)
         return comp
 
+    @classmethod
+    def _get_value(cls, identifier_type, key, token):
+        if identifier_type == cls._TAG_IDENTIFIER:
+            if token.ttype in cls.STRING_VALUE_TYPES or isinstance(token, Identifier):
+                return cls._strip_quotes(token.value, expect_quoted_value=True)
+            raise MlflowException(
+                "Expected a quoted string value for "
+                "{identifier_type} (e.g. 'my-value'). Got value "
+                "{value}".format(identifier_type=identifier_type, value=token.value),
+                error_code=INVALID_PARAMETER_VALUE,
+            )
+        elif identifier_type == cls._ATTRIBUTE_IDENTIFIER:
+            if token.ttype in cls.STRING_VALUE_TYPES or isinstance(token, Identifier):
+                return cls._strip_quotes(token.value, expect_quoted_value=True)
+            elif isinstance(token, Parenthesis):
+                if key != "run_id":
+                    raise MlflowException(
+                        "Only the 'run_id' attribute supports comparison with a list of quoted "
+                        "string values.",
+                        error_code=INVALID_PARAMETER_VALUE,
+                    )
+                return cls._parse_run_ids(token)
+            elif token.ttype in cls.NUMERIC_VALUE_TYPES:
+                if key not in cls.NUMERIC_ATTRIBUTES:
+                    raise MlflowException(
+                        f"Only the '{cls.NUMERIC_ATTRIBUTES}' attributes support comparison with "
+                        "numeric values.",
+                        error_code=INVALID_PARAMETER_VALUE,
+                    )
+                if token.ttype == TokenType.Literal.Number.Integer:
+                    return int(token.value)
+                elif token.ttype == TokenType.Literal.Number.Float:
+                    return float(token.value)
+            else:
+                raise MlflowException(
+                    "Expected a quoted string value or a list of quoted string values for "
+                    "attributes. Got value {value}".format(value=token.value),
+                    error_code=INVALID_PARAMETER_VALUE,
+                )
+        else:
+            # Expected to be either "param" or "metric".
+            raise MlflowException(
+                "Invalid identifier type. Expected one of "
+                "{}.".format([cls._ATTRIBUTE_IDENTIFIER, cls._TAG_IDENTIFIER]),
+                error_code=INVALID_PARAMETER_VALUE,
+            )
+
     @classmethod
     def _process_statement(cls, statement):
         invalids = list(filter(cls._invalid_statement_token_search_model_version, statement.tokens))
@@ -1184,7 +1272,6 @@ class SearchModelVersionUtils(SearchUtils):
             )
         return [cls._get_comparison(t) for t in statement.tokens if isinstance(t, Comparison)]
 
-    # What should we support here?
     @classmethod
     def _invalid_statement_token_search_model_version(cls, token):
         if isinstance(token, (Comparison, Identifier, Parenthesis)):


Test patch:
diff --git a/tests/server/test_handlers.py b/tests/server/test_handlers.py
index bff6121b7..d901e0dc1 100644
--- a/tests/server/test_handlers.py
+++ b/tests/server/test_handlers.py
@@ -45,6 +45,10 @@ from mlflow.server.handlers import (
 from mlflow.server import BACKEND_STORE_URI_ENV_VAR, app
 from mlflow.store.entities.paged_list import PagedList
 from mlflow.protos.service_pb2 import CreateExperiment, SearchRuns
+from mlflow.store.model_registry import (
+    SEARCH_REGISTERED_MODEL_MAX_RESULTS_DEFAULT,
+    SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT,
+)
 from mlflow.protos.model_registry_pb2 import (
     CreateRegisteredModel,
     UpdateRegisteredModel,
@@ -336,14 +340,24 @@ def test_search_registered_models(mock_get_request_message, mock_model_registry_
     mock_model_registry_store.search_registered_models.return_value = PagedList(rmds, None)
     resp = _search_registered_models()
     _, args = mock_model_registry_store.search_registered_models.call_args
-    assert args == {"filter_string": "", "max_results": 100, "order_by": [], "page_token": ""}
+    assert args == {
+        "filter_string": "",
+        "max_results": SEARCH_REGISTERED_MODEL_MAX_RESULTS_DEFAULT,
+        "order_by": [],
+        "page_token": "",
+    }
     assert json.loads(resp.get_data()) == {"registered_models": jsonify(rmds)}
 
     mock_get_request_message.return_value = SearchRegisteredModels(filter="hello")
     mock_model_registry_store.search_registered_models.return_value = PagedList(rmds[:1], "tok")
     resp = _search_registered_models()
     _, args = mock_model_registry_store.search_registered_models.call_args
-    assert args == {"filter_string": "hello", "max_results": 100, "order_by": [], "page_token": ""}
+    assert args == {
+        "filter_string": "hello",
+        "max_results": SEARCH_REGISTERED_MODEL_MAX_RESULTS_DEFAULT,
+        "order_by": [],
+        "page_token": "",
+    }
     assert json.loads(resp.get_data()) == {
         "registered_models": jsonify(rmds[:1]),
         "next_page_token": "tok",
@@ -559,13 +573,12 @@ def test_get_model_version_download_uri(mock_get_request_message, mock_model_reg
 
 
 def test_search_model_versions(mock_get_request_message, mock_model_registry_store):
-    mock_get_request_message.return_value = SearchModelVersions(filter="source_path = 'A/B/CD'")
     mvds = [
         ModelVersion(
             name="model_1",
             version="5",
             creation_timestamp=100,
-            last_updated_timestamp=1200,
+            last_updated_timestamp=3200,
             description="v 5",
             user_id="u1",
             current_stage="Production",
@@ -591,7 +604,7 @@ def test_search_model_versions(mock_get_request_message, mock_model_registry_sto
             name="ads_model",
             version="8",
             creation_timestamp=200,
-            last_updated_timestamp=2000,
+            last_updated_timestamp=1000,
             description="v 8",
             user_id="u1",
             current_stage="Staging",
@@ -604,7 +617,7 @@ def test_search_model_versions(mock_get_request_message, mock_model_registry_sto
             name="fraud_detection_model",
             version="345",
             creation_timestamp=1000,
-            last_updated_timestamp=1001,
+            last_updated_timestamp=999,
             description="newest version",
             user_id="u12",
             current_stage="None",
@@ -614,12 +627,57 @@ def test_search_model_versions(mock_get_request_message, mock_model_registry_sto
             status_message=None,
         ),
     ]
-    mock_model_registry_store.search_model_versions.return_value = mvds
+    mock_get_request_message.return_value = SearchModelVersions(filter="source_path = 'A/B/CD'")
+    mock_model_registry_store.search_model_versions.return_value = PagedList(mvds, None)
     resp = _search_model_versions()
-    args, _ = mock_model_registry_store.search_model_versions.call_args
-    assert args == ("source_path = 'A/B/CD'",)
+    mock_model_registry_store.search_model_versions.assert_called_with(
+        filter_string="source_path = 'A/B/CD'",
+        max_results=SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT,
+        order_by=[],
+        page_token="",
+    )
     assert json.loads(resp.get_data()) == {"model_versions": jsonify(mvds)}
 
+    mock_get_request_message.return_value = SearchModelVersions(filter="name='model_1'")
+    mock_model_registry_store.search_model_versions.return_value = PagedList(mvds[:1], "tok")
+    resp = _search_model_versions()
+    mock_model_registry_store.search_model_versions.assert_called_with(
+        filter_string="name='model_1'",
+        max_results=SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT,
+        order_by=[],
+        page_token="",
+    )
+    assert json.loads(resp.get_data()) == {
+        "model_versions": jsonify(mvds[:1]),
+        "next_page_token": "tok",
+    }
+
+    mock_get_request_message.return_value = SearchModelVersions(filter="version<=12", max_results=2)
+    mock_model_registry_store.search_model_versions.return_value = PagedList(
+        [mvds[0], mvds[2]], "next"
+    )
+    resp = _search_model_versions()
+    mock_model_registry_store.search_model_versions.assert_called_with(
+        filter_string="version<=12", max_results=2, order_by=[], page_token=""
+    )
+    assert json.loads(resp.get_data()) == {
+        "model_versions": jsonify([mvds[0], mvds[2]]),
+        "next_page_token": "next",
+    }
+
+    mock_get_request_message.return_value = SearchModelVersions(
+        filter="version<=12", max_results=2, order_by=["version DESC"], page_token="prev"
+    )
+    mock_model_registry_store.search_model_versions.return_value = PagedList(mvds[1:3], "next")
+    resp = _search_model_versions()
+    mock_model_registry_store.search_model_versions.assert_called_with(
+        filter_string="version<=12", max_results=2, order_by=["version DESC"], page_token="prev"
+    )
+    assert json.loads(resp.get_data()) == {
+        "model_versions": jsonify(mvds[1:3]),
+        "next_page_token": "next",
+    }
+
 
 def test_set_model_version_tag(mock_get_request_message, mock_model_registry_store):
     name = "model1"
diff --git a/tests/store/model_registry/test_file_store.py b/tests/store/model_registry/test_file_store.py
index 995dce44b..1d984e7e9 100644
--- a/tests/store/model_registry/test_file_store.py
+++ b/tests/store/model_registry/test_file_store.py
@@ -197,6 +197,7 @@ def _create_model_version(
     run_link=None,
     description=None,
 ):
+    time.sleep(0.001)
     return fs.create_model_version(
         name, source, run_id, tags, run_link=run_link, description=description
     )
@@ -388,7 +389,7 @@ def test_create_model_version(store):
     store.create_registered_model(name)
     run_id = uuid.uuid4().hex
     with mock.patch("time.time", return_value=456778):
-        mv1 = store.create_model_version(name, "a/b/CD", run_id)
+        mv1 = _create_model_version(store, name, "a/b/CD", run_id)
         assert mv1.name == name
         assert mv1.version == 1
 
@@ -627,6 +628,15 @@ def test_delete_model_version(store):
     assert exception_context.value.error_code == ErrorCode.Name(RESOURCE_DOES_NOT_EXIST)
 
 
+def _search_model_versions(fs, filter_string=None, max_results=10, order_by=None, page_token=None):
+    return fs.search_model_versions(
+        filter_string=filter_string,
+        max_results=max_results,
+        order_by=order_by,
+        page_token=page_token,
+    )
+
+
 def test_search_model_versions(store):
     # create some model versions
     name = "test_for_search_MV"
@@ -644,10 +654,14 @@ def test_search_model_versions(store):
     assert mv4.version == 4
 
     def search_versions(filter_string):
-        return [mvd.version for mvd in store.search_model_versions(filter_string)]
+        return [mvd.version for mvd in _search_model_versions(store, filter_string)]
 
     # search using name should return all 4 versions
-    assert set(search_versions(f"name='{name}'")) == {1, 2, 3, 4}
+    assert set(search_versions("name='%s'" % name)) == {1, 2, 3, 4}
+
+    # search using version
+    assert set(search_versions("version_number=2")) == {2}
+    assert set(search_versions("version_number<=3")) == {1, 2, 3}
 
     # search using run_id_1 should return version 1
     assert set(search_versions(f"run_id='{run_id_1}'")) == {1}
@@ -733,6 +747,14 @@ def test_search_model_versions(store):
         search_versions("run_id IN ('runid1',,'runid2')")
     assert exception_context.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)
 
+    # search using source_path "A/D" should return version 3 and 4
+    assert set(search_versions("source_path = 'A/D'")) == {3, 4}
+
+    # search using source_path "A" should not return anything
+    assert len(search_versions("source_path = 'A'")) == 0
+    assert len(search_versions("source_path = 'A/'")) == 0
+    assert len(search_versions("source_path = ''")) == 0
+
     # delete mv4. search should not return version 4
     store.delete_model_version(name=mv4.name, version=mv4.version)
     assert set(search_versions("")) == {1, 2, 3}
@@ -749,7 +771,7 @@ def test_search_model_versions(store):
         name=mv1.name, version=mv1.version, description="Online prediction model!"
     )
 
-    mvds = store.search_model_versions(f"run_id = '{run_id_1}'")
+    mvds = store.search_model_versions("run_id = '%s'" % run_id_1, max_results=10)
     assert len(mvds) == 1
     assert isinstance(mvds[0], ModelVersion)
     assert mvds[0].current_stage == "Production"
@@ -758,6 +780,99 @@ def test_search_model_versions(store):
     assert mvds[0].description == "Online prediction model!"
 
 
+def test_search_model_versions_order_by_simple(store):
+    # create some model versions
+    names = ["RM1", "RM2", "RM3", "RM4", "RM1", "RM4"]
+    sources = ["A"] * 3 + ["B"] * 3
+    run_ids = [uuid.uuid4().hex for _ in range(6)]
+    for name in set(names):
+        store.create_registered_model(name)
+    for i in range(6):
+        _create_model_version(store, name=names[i], source=sources[i], run_id=run_ids[i])
+        time.sleep(0.001)  # sleep for windows fs timestamp precision issues
+
+    # by default order by last_updated_timestamp DESC
+    mvs = _search_model_versions(store).to_list()
+    assert [mv.name for mv in mvs] == names[::-1]
+    assert [mv.version for mv in mvs] == [2, 2, 1, 1, 1, 1]
+
+    # order by name DESC
+    mvs = _search_model_versions(store, order_by=["name DESC"])
+    assert [mv.name for mv in mvs] == sorted(names)[::-1]
+    assert [mv.version for mv in mvs] == [2, 1, 1, 1, 2, 1]
+
+    # order by version DESC
+    mvs = _search_model_versions(store, order_by=["version_number DESC"])
+    assert [mv.name for mv in mvs] == ["RM1", "RM4", "RM1", "RM2", "RM3", "RM4"]
+    assert [mv.version for mv in mvs] == [2, 2, 1, 1, 1, 1]
+
+    # order by creation_timestamp DESC
+    mvs = _search_model_versions(store, order_by=["creation_timestamp DESC"])
+    assert [mv.name for mv in mvs] == names[::-1]
+    assert [mv.version for mv in mvs] == [2, 2, 1, 1, 1, 1]
+
+    # order by last_updated_timestamp ASC
+    store.update_model_version(names[0], 1, "latest updated")
+    mvs = _search_model_versions(store, order_by=["last_updated_timestamp ASC"])
+    assert mvs[-1].name == names[0]
+    assert mvs[-1].version == 1
+
+
+def test_search_model_versions_pagination(store):
+    def search_versions(filter_string, page_token=None, max_results=10):
+        result = _search_model_versions(
+            store, filter_string=filter_string, page_token=page_token, max_results=max_results
+        )
+        return result.to_list(), result.token
+
+    name = "test_for_search_MV_pagination"
+    store.create_registered_model(name)
+    mvs = [_create_model_version(store, name) for _ in range(50)][::-1]
+
+    # test flow with fixed max_results
+    returned_mvs = []
+    query = "name LIKE 'test_for_search_MV_pagination%'"
+    result, token = search_versions(query, page_token=None, max_results=5)
+    returned_mvs.extend(result)
+    while token:
+        result, token = search_versions(query, page_token=token, max_results=5)
+        returned_mvs.extend(result)
+    assert mvs == returned_mvs
+
+    # test that pagination will return all valid results in sorted order
+    # by name ascending
+    result, token1 = search_versions(query, max_results=5)
+    assert token1 is not None
+    assert result == mvs[0:5]
+
+    result, token2 = search_versions(query, page_token=token1, max_results=10)
+    assert token2 is not None
+    assert result == mvs[5:15]
+
+    result, token3 = search_versions(query, page_token=token2, max_results=20)
+    assert token3 is not None
+    assert result == mvs[15:35]
+
+    result, token4 = search_versions(query, page_token=token3, max_results=100)
+    # assert that page token is None
+    assert token4 is None
+    assert result == mvs[35:]
+
+    # test that providing a completely invalid page token throws
+    with pytest.raises(
+        MlflowException, match=r"Invalid page token, could not base64-decode"
+    ) as exception_context:
+        search_versions(query, page_token="evilhax", max_results=20)
+    assert exception_context.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)
+
+    # test that providing too large of a max_results throws
+    with pytest.raises(
+        MlflowException, match=r"Invalid value for max_results."
+    ) as exception_context:
+        search_versions(query, page_token="evilhax", max_results=1e15)
+    assert exception_context.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)
+
+
 def test_search_model_versions_by_tag(store):
     # create some model versions
     name = "test_for_search_MV_by_tag"
@@ -783,7 +898,7 @@ def test_search_model_versions_by_tag(store):
     assert mv2.version == 2
 
     def search_versions(filter_string):
-        return [mvd.version for mvd in store.search_model_versions(filter_string)]
+        return [mvd.version for mvd in _search_model_versions(store, filter_string)]
 
     assert search_versions(f"name = '{name}' and tag.t2 = 'xyz'") == [1]
     assert search_versions("name = 'wrong_name' and tag.t2 = 'xyz'") == []
@@ -808,9 +923,9 @@ class SearchRegisteredModelsResult(NamedTuple):
 
 
 def _search_registered_models(
-    fs, filter_string=None, max_results=10, order_by=None, page_token=None
+    store, filter_string=None, max_results=10, order_by=None, page_token=None
 ):
-    result = fs.search_registered_models(
+    result = store.search_registered_models(
         filter_string=filter_string,
         max_results=max_results,
         order_by=order_by,
@@ -990,8 +1105,7 @@ def test_search_registered_models_by_tag(store):
 
 def test_search_registered_models_order_by_simple(store):
     # create some registered models
-    prefix = "test_for_search_"
-    names = [prefix + name for name in ["RM1", "RM2", "RM3", "RM4", "RM4A", "RM4ab"]]
+    names = ["RM1", "RM2", "RM3", "RM4", "RM4A", "RM4ab"]
     for name in names:
         store.create_registered_model(name)
         time.sleep(0.001)  # sleep for windows store timestamp precision issues
@@ -1307,13 +1421,12 @@ def test_pyfunc_model_registry_with_file_store(store):
     with mlflow.start_run():
         mlflow.log_param("A", "B")
 
-    models = store.search_registered_models(max_results=10)
-    assert len(models) == 2
-    assert models[0].name == "model1"
-    assert models[1].name == "model2"
-    mv1 = store.search_model_versions("name = 'model1'")
-    assert len(mv1) == 2
-    assert mv1[0].name == "model1"
-    mv2 = store.search_model_versions("name = 'model2'")
-    assert len(mv2) == 1
-    assert mv2[0].name == "model2"
+        models = store.search_registered_models(max_results=10)
+        assert len(models) == 2
+        assert models[0].name == "model1"
+        assert models[1].name == "model2"
+        mv1 = store.search_model_versions("name = 'model1'", max_results=10)
+        assert len(mv1) == 2
+        assert mv1[0].name == "model1"
+        mv2 = store.search_model_versions("name = 'model2'", max_results=10)
+        assert len(mv2) == 1 and mv2[0].name == "model2"
diff --git a/tests/store/model_registry/test_rest_store.py b/tests/store/model_registry/test_rest_store.py
index be2d42f30..d0dd59bb0 100644
--- a/tests/store/model_registry/test_rest_store.py
+++ b/tests/store/model_registry/test_rest_store.py
@@ -322,13 +322,34 @@ def test_get_model_version_download_uri(store, creds):
 
 def test_search_model_versions(store, creds):
     with mock_http_request_200() as mock_http:
-        store.search_model_versions(filter_string="name='model_12'")
+        store.search_model_versions()
+    _verify_requests(mock_http, creds, "model-versions/search", "GET", SearchModelVersions())
+
+
+@pytest.mark.parametrize("filter_string", [None, "name = 'model_12'"])
+@pytest.mark.parametrize("max_results", [None, 400])
+@pytest.mark.parametrize("page_token", [None, "blah"])
+@pytest.mark.parametrize("order_by", ["version DESC", "creation_time DESC"])
+def test_search_model_versions_params(
+    store, creds, filter_string, max_results, page_token, order_by
+):
+    params = {
+        "filter_string": filter_string,
+        "max_results": max_results,
+        "page_token": page_token,
+        "order_by": order_by,
+    }
+    params = {k: v for k, v in params.items() if v is not None}
+    with mock_http_request_200() as mock_http:
+        store.search_model_versions(**params)
+    if "filter_string" in params:
+        params["filter"] = params.pop("filter_string")
     _verify_requests(
         mock_http,
         creds,
         "model-versions/search",
         "GET",
-        SearchModelVersions(filter="name='model_12'"),
+        SearchModelVersions(**params),
     )
 
 
diff --git a/tests/store/model_registry/test_sqlalchemy_store.py b/tests/store/model_registry/test_sqlalchemy_store.py
index 9551937ad..e02a98bd0 100644
--- a/tests/store/model_registry/test_sqlalchemy_store.py
+++ b/tests/store/model_registry/test_sqlalchemy_store.py
@@ -755,12 +755,21 @@ class TestSqlAlchemyStoreSqlite(unittest.TestCase):
         mv4 = self._mv_maker(name=name, source="A/D", run_id=run_id_3)
         assert mv4.version == 4
 
-        def search_versions(filter_string):
-            return [mvd.version for mvd in self.store.search_model_versions(filter_string)]
+        def search_versions(filter_string, max_results=10, order_by=None, page_token=None):
+            return [
+                mvd.version
+                for mvd in self.store.search_model_versions(
+                    filter_string, max_results, order_by, page_token
+                )
+            ]
 
         # search using name should return all 4 versions
         assert set(search_versions("name='%s'" % name)) == {1, 2, 3, 4}
 
+        # search using version
+        assert set(search_versions("version_number=2")) == {2}
+        assert set(search_versions("version_number<=3")) == {1, 2, 3}
+
         # search using run_id_1 should return version 1
         assert set(search_versions("run_id='%s'" % run_id_1)) == {1}
 
@@ -871,7 +880,7 @@ class TestSqlAlchemyStoreSqlite(unittest.TestCase):
             name=mv1.name, version=mv1.version, description="Online prediction model!"
         )
 
-        mvds = self.store.search_model_versions("run_id = '%s'" % run_id_1)
+        mvds = self.store.search_model_versions("run_id = '%s'" % run_id_1, max_results=10)
         assert len(mvds) == 1
         assert isinstance(mvds[0], ModelVersion)
         assert mvds[0].current_stage == "Production"
@@ -879,6 +888,130 @@ class TestSqlAlchemyStoreSqlite(unittest.TestCase):
         assert mvds[0].source == "A/B"
         assert mvds[0].description == "Online prediction model!"
 
+    def test_search_model_versions_order_by_simple(self):
+        # create some model versions
+        names = ["RM1", "RM2", "RM3", "RM4", "RM1", "RM4"]
+        sources = ["A"] * 3 + ["B"] * 3
+        run_ids = [uuid.uuid4().hex for _ in range(6)]
+        for name in set(names):
+            self._rm_maker(name)
+        for i in range(6):
+            self._mv_maker(name=names[i], source=sources[i], run_id=run_ids[i])
+
+        # by default order by last_updated_timestamp DESC
+        mvs = self.store.search_model_versions(filter_string=None)
+        assert [mv.name for mv in mvs] == names[::-1]
+        assert [mv.version for mv in mvs] == [2, 2, 1, 1, 1, 1]
+
+        # order by name DESC
+        mvs = self.store.search_model_versions(filter_string=None, order_by=["name DESC"])
+        assert [mv.name for mv in mvs] == sorted(names)[::-1]
+        assert [mv.version for mv in mvs] == [2, 1, 1, 1, 2, 1]
+
+        # order by version DESC
+        mvs = self.store.search_model_versions(filter_string=None, order_by=["version_number DESC"])
+        assert [mv.name for mv in mvs] == ["RM1", "RM4", "RM1", "RM2", "RM3", "RM4"]
+        assert [mv.version for mv in mvs] == [2, 2, 1, 1, 1, 1]
+
+        # order by creation_timestamp DESC
+        mvs = self.store.search_model_versions(
+            filter_string=None, order_by=["creation_timestamp DESC"]
+        )
+        assert [mv.name for mv in mvs] == names[::-1]
+        assert [mv.version for mv in mvs] == [2, 2, 1, 1, 1, 1]
+
+        # order by last_updated_timestamp ASC
+        self.store.update_model_version(names[0], 1, "latest updated")
+        mvs = self.store.search_model_versions(
+            filter_string=None, order_by=["last_updated_timestamp ASC"]
+        )
+        assert mvs[-1].name == names[0]
+        assert mvs[-1].version == 1
+
+    def test_search_model_versions_order_by_errors(self):
+        # create some model versions
+        name = "RM1"
+        self._rm_maker(name)
+        for _ in range(6):
+            self._mv_maker(name=name)
+        query = "name LIKE 'RM%'"
+        # test that invalid columns throw even if they come after valid columns
+        with pytest.raises(
+            MlflowException, match=r"Invalid attribute key '.+' specified"
+        ) as exception_context:
+            self.store.search_model_versions(
+                query,
+                page_token=None,
+                order_by=["name ASC", "run_id DESC"],
+                max_results=5,
+            )
+        assert exception_context.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)
+        # test that invalid columns with random text throw even if they come after valid columns
+        with pytest.raises(
+            MlflowException, match=r"Invalid order_by clause '.+'"
+        ) as exception_context:
+            self.store.search_model_versions(
+                query,
+                page_token=None,
+                order_by=["name ASC", "last_updated_timestamp DESC blah"],
+                max_results=5,
+            )
+        assert exception_context.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)
+
+    def test_search_model_versions_pagination(self):
+        def search_versions(filter_string, page_token=None, max_results=10):
+            result = self.store.search_model_versions(
+                filter_string=filter_string, page_token=page_token, max_results=max_results
+            )
+            return result.to_list(), result.token
+
+        name = "test_for_search_MV_pagination"
+        self._rm_maker(name)
+        mvs = [self._mv_maker(name) for _ in range(50)][::-1]
+
+        # test flow with fixed max_results
+        returned_mvs = []
+        query = "name LIKE 'test_for_search_MV_pagination%'"
+        result, token = search_versions(query, page_token=None, max_results=5)
+        returned_mvs.extend(result)
+        while token:
+            result, token = search_versions(query, page_token=token, max_results=5)
+            returned_mvs.extend(result)
+        assert mvs == returned_mvs
+
+        # test that pagination will return all valid results in sorted order
+        # by name ascending
+        result, token1 = search_versions(query, max_results=5)
+        assert token1 is not None
+        assert result == mvs[0:5]
+
+        result, token2 = search_versions(query, page_token=token1, max_results=10)
+        assert token2 is not None
+        assert result == mvs[5:15]
+
+        result, token3 = search_versions(query, page_token=token2, max_results=20)
+        assert token3 is not None
+        assert result == mvs[15:35]
+
+        result, token4 = search_versions(query, page_token=token3, max_results=100)
+        # assert that page token is None
+        assert token4 is None
+        assert result == mvs[35:]
+
+        # test that providing a completely invalid page token throws
+        with pytest.raises(
+            MlflowException, match=r"Invalid page token, could not base64-decode"
+        ) as exception_context:
+            search_versions(query, page_token="evilhax", max_results=20)
+        assert exception_context.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)
+
+        # test that providing too large of a max_results throws
+        with pytest.raises(
+            MlflowException, match=r"Invalid value for max_results\."
+        ) as exception_context:
+            search_versions(query, page_token="evilhax", max_results=1e15)
+        assert exception_context.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)
+
     def test_search_model_versions_by_tag(self):
         # create some model versions
         name = "test_for_search_MV_by_tag"
diff --git a/tests/tracking/_model_registry/test_model_registry_client.py b/tests/tracking/_model_registry/test_model_registry_client.py
index c30674418..b7467a908 100644
--- a/tests/tracking/_model_registry/test_model_registry_client.py
+++ b/tests/tracking/_model_registry/test_model_registry_client.py
@@ -15,6 +15,10 @@ from mlflow.entities.model_registry import (
 )
 from mlflow.exceptions import MlflowException
 from mlflow.store.entities.paged_list import PagedList
+from mlflow.store.model_registry import (
+    SEARCH_REGISTERED_MODEL_MAX_RESULTS_DEFAULT,
+    SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT,
+)
 from mlflow.tracking._model_registry.client import ModelRegistryClient
 
 
@@ -113,7 +117,9 @@ def test_search_registered_models(mock_store):
         [RegisteredModel("Model 1"), RegisteredModel("Model 2")], ""
     )
     result = newModelRegistryClient().search_registered_models(filter_string="test filter")
-    mock_store.search_registered_models.assert_called_with("test filter", 100, None, None)
+    mock_store.search_registered_models.assert_called_with(
+        "test filter", SEARCH_REGISTERED_MODEL_MAX_RESULTS_DEFAULT, None, None
+    )
     assert len(result) == 2
     assert result.token == ""
 
@@ -337,13 +343,32 @@ def test_get_model_version_download_uri(mock_store):
 
 
 def test_search_model_versions(mock_store):
-    mock_store.search_model_versions.return_value = [
-        ModelVersion(name="Model 1", version="1", creation_timestamp=123),
-        ModelVersion(name="Model 1", version="2", creation_timestamp=124),
+    mvs = [
+        ModelVersion(
+            name="Model 1", version="1", creation_timestamp=123, last_updated_timestamp=123
+        ),
+        ModelVersion(
+            name="Model 1", version="2", creation_timestamp=124, last_updated_timestamp=124
+        ),
+        ModelVersion(
+            name="Model 2", version="1", creation_timestamp=125, last_updated_timestamp=125
+        ),
     ]
+    mock_store.search_model_versions.return_value = PagedList(mvs[:2][::-1], "")
     result = newModelRegistryClient().search_model_versions("name=Model 1")
-    mock_store.search_model_versions.assert_called_once_with("name=Model 1")
-    assert len(result) == 2
+    mock_store.search_model_versions.assert_called_with(
+        "name=Model 1", SEARCH_MODEL_VERSION_MAX_RESULTS_DEFAULT, None, None
+    )
+    assert result == mvs[:2][::-1]
+    assert result.token == ""
+
+    mock_store.search_model_versions.return_value = PagedList([mvs[1], mvs[2], mvs[0]], "")
+    result = newModelRegistryClient().search_model_versions(
+        "version <= 2", max_results=2, order_by="version DESC", page_token="next"
+    )
+    mock_store.search_model_versions.assert_called_with("version <= 2", 2, "version DESC", "next")
+    assert result == [mvs[1], mvs[2], mvs[0]]
+    assert result.token == ""
 
 
 def test_get_model_version_stages(mock_store):
diff --git a/tests/tracking/test_model_registry.py b/tests/tracking/test_model_registry.py
index bc5c661c2..9ab4a4c44 100644
--- a/tests/tracking/test_model_registry.py
+++ b/tests/tracking/test_model_registry.py
@@ -4,7 +4,7 @@ and ensures we can use the tracking API to communicate with it.
 """
 import pytest
 
-from mlflow.entities.model_registry import RegisteredModel
+from mlflow.entities.model_registry import RegisteredModel, ModelVersion
 from mlflow.exceptions import MlflowException
 from mlflow import MlflowClient
 from mlflow.utils.time_utils import get_current_time_millis
@@ -252,6 +252,140 @@ def test_set_registered_model_tag_with_empty_string_as_value(client):
     assert {"tag_key": ""}.items() <= client.get_registered_model(name).tags.items()
 
 
+@pytest.mark.parametrize(
+    ("filter_string", "filter_func"),
+    [
+        (None, lambda mv: True),
+        ("", lambda mv: True),
+        ("name LIKE '%2'", lambda mv: mv.name.endswith("2")),
+        ("name ILIKE '%rm%00%'", lambda mv: "00" in mv.name),
+        ("name LIKE '%rm%00%'", lambda mv: False),
+        ("name = 'badname'", lambda mv: False),
+        ("name = 'CreateRMsearchForMV03'", lambda mv: mv.name == "CreateRMsearchForMV03"),
+    ],
+)
+def test_search_model_versions_filter_string(
+    client,
+    filter_string,
+    filter_func,
+):
+    names = [f"CreateRMsearchForMV{i:03}" for i in range(29)]
+    for name in names:
+        client.create_registered_model(name)
+    mvs = [
+        client.create_model_version(name, "path/to/model", "run_id") for name in names + names[:10]
+    ]
+    for mv in mvs:
+        assert isinstance(mv, ModelVersion)
+    mvs = mvs[::-1]
+
+    def verify_pagination(mv_getter_with_token, expected_mvs):
+        result_mvs = []
+        result = mv_getter_with_token(None)
+        result_mvs.extend(result)
+        while result.token:
+            result = mv_getter_with_token(result.token)
+            result_mvs.extend(result)
+        assert [mv.name for mv in expected_mvs] == [mv.name for mv in result_mvs]
+        assert [mv.version for mv in expected_mvs] == [mv.version for mv in result_mvs]
+
+    expected_mvs = sorted(
+        filter(filter_func, mvs), key=lambda x: x.last_updated_timestamp, reverse=True
+    )
+    verify_pagination(
+        lambda tok: client.search_model_versions(
+            filter_string=filter_string,
+            page_token=tok,
+        ),
+        expected_mvs,
+    )
+
+
+@pytest.mark.parametrize("max_results", [1, 8, 100])
+def test_search_model_versions_max_results(client, max_results):
+    names = [f"CreateRMsearchForMV{i:03}" for i in range(29)]
+    for name in names:
+        client.create_registered_model(name)
+    mvs = [
+        client.create_model_version(name, "path/to/model", "run_id") for name in names + names[:10]
+    ]
+    for mv in mvs:
+        assert isinstance(mv, ModelVersion)
+    mvs = mvs[::-1]
+
+    def verify_pagination(mv_getter_with_token, expected_mvs):
+        result_mvs = []
+        result = mv_getter_with_token(None)
+        result_mvs.extend(result)
+        while result.token:
+            result = mv_getter_with_token(result.token)
+            result_mvs.extend(result)
+        assert [mv.name for mv in expected_mvs] == [mv.name for mv in result_mvs]
+        assert [mv.version for mv in expected_mvs] == [mv.version for mv in result_mvs]
+
+    expected_mvs = sorted(mvs, key=lambda x: x.last_updated_timestamp, reverse=True)
+    verify_pagination(
+        lambda tok: client.search_model_versions(
+            max_results=max_results,
+            page_token=tok,
+        ),
+        expected_mvs,
+    )
+
+
+@pytest.mark.parametrize(
+    ("order_by", "order_by_key", "order_by_desc"),
+    [
+        (None, None, False),
+        (["name DESC"], lambda mv: (mv.name, mv.version), True),
+        (
+            ["version_number DESC"],
+            lambda mv: (-int(mv.version), mv.name),
+            False,
+        ),
+    ],
+)
+def test_search_model_versions_order_by(
+    client,
+    order_by,
+    order_by_key,
+    order_by_desc,
+):
+    names = [f"CreateRMsearchForMV{i:03}" for i in range(29)]
+    for name in names:
+        client.create_registered_model(name)
+    mvs = [
+        client.create_model_version(name, "path/to/model", "run_id") for name in names + names[:10]
+    ]
+    for mv in mvs:
+        assert isinstance(mv, ModelVersion)
+    mvs = mvs[::-1]
+
+    def verify_pagination(mv_getter_with_token, expected_mvs):
+        result_mvs = []
+        result = mv_getter_with_token(None)
+        result_mvs.extend(result)
+        while result.token:
+            result = mv_getter_with_token(result.token)
+            result_mvs.extend(result)
+        assert [mv.name for mv in expected_mvs] == [mv.name for mv in result_mvs]
+        assert [mv.version for mv in expected_mvs] == [mv.version for mv in result_mvs]
+
+    if order_by_key:
+        expected_mvs = sorted(mvs, key=order_by_key, reverse=order_by_desc)
+    else:
+        expected_mvs = sorted(
+            mvs, key=lambda x: (-int(x.last_updated_timestamp), x.name, -int(x.version))
+        )
+    verify_pagination(
+        lambda tok: client.search_model_versions(
+            order_by=order_by,
+            page_token=tok,
+        ),
+        expected_mvs,
+    )
+
+
 def test_create_and_query_model_version_flow(client):
     name = "CreateMVTest"
     tags = {"key": "value", "another key": "some other value", "numeric value": 12345}