Repository: mlflow/mlflow

Issue:
[BUG] Error when logging batch of metrics via mlflow.log_metrics
### Issues Policy acknowledgement

- [X] I have read and agree to submit bug reports in accordance with the [issues policy](https://www.github.com/mlflow/mlflow/blob/master/ISSUE_POLICY.md)

### Where did you encounter this bug?

Local machine

### Willingness to contribute

No. I cannot contribute a bug fix at this time.

### MLflow version

- Client: 2.15.0
- Tracking server: 2.15.0


### System information

- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Ubuntu 22.04
- **Python version**: 3.10.12
- **yarn version, if running the dev UI**: N/A


### Describe the problem

When logging a batch of metrics `mlflow.log_metrics`, mlflow raises an exception:
```
mlflow.exceptions.MlflowException: Got invalid value 1.0 for metric 'x' (timestamp=1722625307265). Please specify value as a valid double (64-bit floating point)
```

Logging each metric separately using `mlflow.log_metric` works fine.

### Tracking information

N/A

### Code to reproduce issue

```python
import numpy as np
import mlflow

x = np.array(1.)
y = np.array(2.)

with mlflow.start_run():
     # this won't work
    mlflow.log_metrics(metrics={'x': x, 'y': y}, step=1)

    # but this works
    mlflow.log_metric(key='x', value=x, step=1)
    mlflow.log_metric(key='y', value=y, step=1)
```


### Stack trace

```python
Traceback (most recent call last):
  File "<stdin>", line 2, in <module>
  File "/home/n10/venv/lib/python3.10/site-packages/mlflow/tracking/fluent.py", line 916, in log_metrics
    return MlflowClient().log_batch(
  File "/home/n10/venv/lib/python3.10/site-packages/mlflow/tracking/client.py", line 1846, in log_batch
    return self._tracking_client.log_batch(
  File "/home/n10/venv/lib/python3.10/site-packages/mlflow/tracking/_tracking_service/client.py", line 732, in log_batch
    self.store.log_batch(run_id=run_id, metrics=metrics_batch, params=[], tags=[])
  File "/home/n10/venv/lib/python3.10/site-packages/mlflow/store/tracking/file_store.py", line 1070, in log_batch
    metrics, params, tags = _validate_batch_log_data(metrics, params, tags)
  File "/home/n10/venv/lib/python3.10/site-packages/mlflow/utils/validation.py", line 345, in _validate_batch_log_data
    _validate_metric(metric.key, metric.value, metric.timestamp, metric.step)
  File "/home/n10/venv/lib/python3.10/site-packages/mlflow/utils/validation.py", line 156, in _validate_metric
    raise MlflowException(
mlflow.exceptions.MlflowException: Got invalid value 1.0 for metric 'x' (timestamp=1722625307265). Please specify value as a valid double (64-bit floating point)
```


### Other info / logs

N/A

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

- [ ] `area/artifacts`: Artifact stores and artifact logging
- [ ] `area/build`: Build and test infrastructure for MLflow
- [ ] `area/deployments`: MLflow Deployments client APIs, server, and third-party Deployments integrations
- [ ] `area/docs`: MLflow documentation pages
- [ ] `area/examples`: Example code
- [ ] `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
- [X] `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:
github-actions[bot]: <!-- documentation preview -->


Documentation preview for e614bf4417d723252fa3cffa279246f14b32292f will be available when [this CircleCI job](https://circleci.com/gh/mlflow/mlflow/117834)
completes successfully.

- [Top page](https://output.circle-artifacts.com/output/job/b1a42be1-1379-4f76-8591-dfa03345918c/artifacts/0/docs/build/html/index.html)
- [Changed pages](https://output.circle-artifacts.com/output/job/b1a42be1-1379-4f76-8591-dfa03345918c/artifacts/0/docs/build/html/diff.html)

<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/10225995947.

</details>

serena-ruan: @beomsun0829 Thanks for fixing this! Could you add a unit test?
beomsun0829: @serena-ruan I commited unit tests, check it out!

Patch:
diff --git a/dev/generate_protos.py b/dev/generate_protos.py
index 046b827a3..d75a46ce5 100644
--- a/dev/generate_protos.py
+++ b/dev/generate_protos.py
@@ -57,6 +57,8 @@ uc_proto_files = [
     "databricks_uc_registry_messages.proto",
     "databricks_uc_registry_service.proto",
     "databricks_filesystem_service.proto",
+    "unity_catalog_oss_messages.proto",
+    "unity_catalog_oss_service.proto",
 ]
 facet_proto_files = ["facet_feature_statistics.proto"]
 python_proto_files = basic_proto_files + uc_proto_files + facet_proto_files
@@ -79,6 +81,10 @@ python_gencode_replacements = [
         "from . import databricks_managed_catalog_messages_pb2 as databricks_managed_"
         "catalog_messages_pb2",
     ),
+    (
+        "import unity_catalog_oss_messages_pb2 as unity__catalog__oss__messages__pb2",
+        "from . import unity_catalog_oss_messages_pb2 as unity_catalog_oss_messages_pb2",
+    ),
 ]
 
 
diff --git a/mlflow/data/delta_dataset_source.py b/mlflow/data/delta_dataset_source.py
index 20b4499bc..9c7887927 100644
--- a/mlflow/data/delta_dataset_source.py
+++ b/mlflow/data/delta_dataset_source.py
@@ -7,7 +7,7 @@ from mlflow.protos.databricks_managed_catalog_messages_pb2 import (
     GetTable,
     GetTableResponse,
 )
-from mlflow.protos.databricks_managed_catalog_service_pb2 import UnityCatalogService
+from mlflow.protos.databricks_managed_catalog_service_pb2 import DatabricksUnityCatalogService
 from mlflow.protos.databricks_pb2 import INVALID_PARAMETER_VALUE
 from mlflow.utils._spark_utils import _get_active_spark_session
 from mlflow.utils._unity_catalog_utils import get_full_name_from_sc
@@ -119,7 +119,7 @@ class DeltaDatasetSource(DatasetSource):
         try:
             req_body = message_to_json(GetTable(full_name_arg=table_name))
             _METHOD_TO_INFO = extract_api_info_for_service(
-                UnityCatalogService, _REST_API_PATH_PREFIX
+                DatabricksUnityCatalogService, _REST_API_PATH_PREFIX
             )
             db_creds = get_databricks_host_creds()
             endpoint, method = _METHOD_TO_INFO[GetTable]
diff --git a/mlflow/protos/databricks_managed_catalog_service.proto b/mlflow/protos/databricks_managed_catalog_service.proto
index 64bd28c8b..0e0c587db 100644
--- a/mlflow/protos/databricks_managed_catalog_service.proto
+++ b/mlflow/protos/databricks_managed_catalog_service.proto
@@ -13,7 +13,7 @@ option (scalapb.options) = {
   flat_package: true,
 };
 
-service UnityCatalogService {
+service DatabricksUnityCatalogService {
   rpc getTable (GetTable) returns (GetTableResponse) {
     option (rpc) = {
       endpoints: {
diff --git a/mlflow/protos/databricks_managed_catalog_service_pb2.py b/mlflow/protos/databricks_managed_catalog_service_pb2.py
index 78e4007c9..4abc14780 100644
--- a/mlflow/protos/databricks_managed_catalog_service_pb2.py
+++ b/mlflow/protos/databricks_managed_catalog_service_pb2.py
@@ -21,7 +21,7 @@ if Version(google.protobuf.__version__).major >= 5:
   from .scalapb import scalapb_pb2 as scalapb_dot_scalapb__pb2
 
 
-  DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n(databricks_managed_catalog_service.proto\x12\x15mlflow.managedcatalog\x1a\x10\x64\x61tabricks.proto\x1a)databricks_managed_catalog_messages.proto\x1a\x15scalapb/scalapb.proto2\xa8\x01\n\x13UnityCatalogService\x12\x90\x01\n\x08getTable\x12\x1f.mlflow.managedcatalog.GetTable\x1a\'.mlflow.managedcatalog.GetTableResponse\":\xf2\x86\x19\x36\n2\n\x03GET\x12%/unity-catalog/tables/{full_name_arg}\x1a\x04\x08\x02\x10\x00\x10\x03\x42\x34\n\'com.databricks.api.proto.managedcatalog\x90\x01\x01\xa0\x01\x01\xe2?\x02\x10\x01')
+  DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n(databricks_managed_catalog_service.proto\x12\x15mlflow.managedcatalog\x1a\x10\x64\x61tabricks.proto\x1a)databricks_managed_catalog_messages.proto\x1a\x15scalapb/scalapb.proto2\xb2\x01\n\x1d\x44\x61tabricksUnityCatalogService\x12\x90\x01\n\x08getTable\x12\x1f.mlflow.managedcatalog.GetTable\x1a\'.mlflow.managedcatalog.GetTableResponse\":\xf2\x86\x19\x36\n2\n\x03GET\x12%/unity-catalog/tables/{full_name_arg}\x1a\x04\x08\x02\x10\x00\x10\x03\x42\x34\n\'com.databricks.api.proto.managedcatalog\x90\x01\x01\xa0\x01\x01\xe2?\x02\x10\x01')
 
   _globals = globals()
   _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
@@ -29,10 +29,10 @@ if Version(google.protobuf.__version__).major >= 5:
   if not _descriptor._USE_C_DESCRIPTORS:
     _globals['DESCRIPTOR']._loaded_options = None
     _globals['DESCRIPTOR']._serialized_options = b'\n\'com.databricks.api.proto.managedcatalog\220\001\001\240\001\001\342?\002\020\001'
-    _globals['_UNITYCATALOGSERVICE'].methods_by_name['getTable']._loaded_options = None
-    _globals['_UNITYCATALOGSERVICE'].methods_by_name['getTable']._serialized_options = b'\362\206\0316\n2\n\003GET\022%/unity-catalog/tables/{full_name_arg}\032\004\010\002\020\000\020\003'
-    _globals['_UNITYCATALOGSERVICE']._serialized_start=152
-    _globals['_UNITYCATALOGSERVICE']._serialized_end=320
+    _globals['_DATABRICKSUNITYCATALOGSERVICE'].methods_by_name['getTable']._loaded_options = None
+    _globals['_DATABRICKSUNITYCATALOGSERVICE'].methods_by_name['getTable']._serialized_options = b'\362\206\0316\n2\n\003GET\022%/unity-catalog/tables/{full_name_arg}\032\004\010\002\020\000\020\003'
+    _globals['_DATABRICKSUNITYCATALOGSERVICE']._serialized_start=152
+    _globals['_DATABRICKSUNITYCATALOGSERVICE']._serialized_end=330
   _builder.BuildServices(DESCRIPTOR, 'databricks_managed_catalog_service_pb2', _globals)
   # @@protoc_insertion_point(module_scope)
 
@@ -58,26 +58,26 @@ else:
   from .scalapb import scalapb_pb2 as scalapb_dot_scalapb__pb2
 
 
-  DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n(databricks_managed_catalog_service.proto\x12\x15mlflow.managedcatalog\x1a\x10\x64\x61tabricks.proto\x1a)databricks_managed_catalog_messages.proto\x1a\x15scalapb/scalapb.proto2\xa8\x01\n\x13UnityCatalogService\x12\x90\x01\n\x08getTable\x12\x1f.mlflow.managedcatalog.GetTable\x1a\'.mlflow.managedcatalog.GetTableResponse\":\xf2\x86\x19\x36\n2\n\x03GET\x12%/unity-catalog/tables/{full_name_arg}\x1a\x04\x08\x02\x10\x00\x10\x03\x42\x34\n\'com.databricks.api.proto.managedcatalog\x90\x01\x01\xa0\x01\x01\xe2?\x02\x10\x01')
+  DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n(databricks_managed_catalog_service.proto\x12\x15mlflow.managedcatalog\x1a\x10\x64\x61tabricks.proto\x1a)databricks_managed_catalog_messages.proto\x1a\x15scalapb/scalapb.proto2\xb2\x01\n\x1d\x44\x61tabricksUnityCatalogService\x12\x90\x01\n\x08getTable\x12\x1f.mlflow.managedcatalog.GetTable\x1a\'.mlflow.managedcatalog.GetTableResponse\":\xf2\x86\x19\x36\n2\n\x03GET\x12%/unity-catalog/tables/{full_name_arg}\x1a\x04\x08\x02\x10\x00\x10\x03\x42\x34\n\'com.databricks.api.proto.managedcatalog\x90\x01\x01\xa0\x01\x01\xe2?\x02\x10\x01')
 
 
 
-  _UNITYCATALOGSERVICE = DESCRIPTOR.services_by_name['UnityCatalogService']
+  _DATABRICKSUNITYCATALOGSERVICE = DESCRIPTOR.services_by_name['DatabricksUnityCatalogService']
   if _descriptor._USE_C_DESCRIPTORS == False:
 
     DESCRIPTOR._options = None
     DESCRIPTOR._serialized_options = b'\n\'com.databricks.api.proto.managedcatalog\220\001\001\240\001\001\342?\002\020\001'
-    _UNITYCATALOGSERVICE.methods_by_name['getTable']._options = None
-    _UNITYCATALOGSERVICE.methods_by_name['getTable']._serialized_options = b'\362\206\0316\n2\n\003GET\022%/unity-catalog/tables/{full_name_arg}\032\004\010\002\020\000\020\003'
-    _UNITYCATALOGSERVICE._serialized_start=152
-    _UNITYCATALOGSERVICE._serialized_end=320
-  UnityCatalogService = service_reflection.GeneratedServiceType('UnityCatalogService', (_service.Service,), dict(
-    DESCRIPTOR = _UNITYCATALOGSERVICE,
+    _DATABRICKSUNITYCATALOGSERVICE.methods_by_name['getTable']._options = None
+    _DATABRICKSUNITYCATALOGSERVICE.methods_by_name['getTable']._serialized_options = b'\362\206\0316\n2\n\003GET\022%/unity-catalog/tables/{full_name_arg}\032\004\010\002\020\000\020\003'
+    _DATABRICKSUNITYCATALOGSERVICE._serialized_start=152
+    _DATABRICKSUNITYCATALOGSERVICE._serialized_end=330
+  DatabricksUnityCatalogService = service_reflection.GeneratedServiceType('DatabricksUnityCatalogService', (_service.Service,), dict(
+    DESCRIPTOR = _DATABRICKSUNITYCATALOGSERVICE,
     __module__ = 'databricks_managed_catalog_service_pb2'
     ))
 
-  UnityCatalogService_Stub = service_reflection.GeneratedServiceStubType('UnityCatalogService_Stub', (UnityCatalogService,), dict(
-    DESCRIPTOR = _UNITYCATALOGSERVICE,
+  DatabricksUnityCatalogService_Stub = service_reflection.GeneratedServiceStubType('DatabricksUnityCatalogService_Stub', (DatabricksUnityCatalogService,), dict(
+    DESCRIPTOR = _DATABRICKSUNITYCATALOGSERVICE,
     __module__ = 'databricks_managed_catalog_service_pb2'
     ))
 
diff --git a/mlflow/protos/unity_catalog_oss_messages.proto b/mlflow/protos/unity_catalog_oss_messages.proto
new file mode 100644
index 000000000..273340522
--- /dev/null
+++ b/mlflow/protos/unity_catalog_oss_messages.proto
@@ -0,0 +1,92 @@
+syntax = "proto2";
+
+package mlflow.unitycatalog;
+
+import "scalapb/scalapb.proto";
+import "databricks.proto";
+
+option java_package = "com.databricks.api.proto.managedcatalog";
+option java_generate_equals_and_hash = true;
+option (scalapb.options) = {
+  flat_package: true,
+};
+
+// Registered Model Management API
+
+message CreateRegisteredModel {
+  optional RegisteredModelInfo registered_model_info = 1;
+
+  message Response {
+    optional RegisteredModelInfo registered_model_info = 1;
+  }
+}
+
+message RegisteredModelInfo {
+  // Input fields:
+
+  // [Create:REQ Update:OPT] Name of Model, relative to parent Schema.
+  optional string name = 1;
+  // [Create:REQ Update:IGN] Name of parent Catalog.
+  optional string catalog_name = 2;
+  // [Create:REQ Update:IGN] Name of parent Schema relative to its parent Catalog.
+  optional string schema_name = 3;
+  // [Create:IGN Update:OPT] Username of current owner of Model.
+  optional string owner = 4;
+  // [Create,Update:OPT] User-provided free-form text description.
+  optional string comment = 5;
+  // [Create:OPT] Registered model tags
+  repeated TagKeyValue tags = 20;
+
+  // Output-only fields:
+
+  // [Create,Update:IGN] Storage root URL for model
+  optional string storage_location = 6;
+  // [Create,Update:IGN] Unique identifier of parent Metastore.
+  optional string metastore_id = 7;
+  // [Create,Update:IGN] Full name of Model, in form of <catalog_name>.<schema_name>.<model_name>
+  optional string full_name = 9;
+  // [Create,Update:IGN] Time at which this Model was created, in epoch milliseconds.
+  optional int64 created_at = 11;
+  // [Create,Update:IGN] Username of Model creator.
+  optional string created_by = 12;
+  // [Create,Update:IGN] Time at which this Model was last modified, in epoch milliseconds.
+  optional int64 updated_at = 13;
+  // [Create,Update:IGN] Username of user who last modified the Model.
+  optional string updated_by = 14;
+  // [Create,Update:IGN] common - type of securable
+  optional SecurableType securable_type = 15;
+  // [Create,Update:IGN] common - kind of securable
+  optional SecurableKind securable_kind = 16;
+  // [Create,Update:IGN] ID of the securable that can be used in paging tokens
+  optional string id = 18;
+  // [Create,Update:IGN] Registered model aliases
+  repeated RegisteredModelAliasInfo aliases = 19;
+  // [Create,Update:IGN] Indicate if current model info contains only browsable metadata
+  optional bool browse_only = 21;
+  // [Create,Update:IGN] Encryption options to use when accessing cloud storage.
+  //  optional EncryptionDetails encryption_details = 22;
+}
+
+message TagKeyValue {
+  // name of the tag
+  optional string key = 1;
+  // value of the tag associated with the key, could be optional
+  optional string value = 2;
+}
+
+enum SecurableType {
+  UNKNOWN_SECURABLE_TYPE = 0;  // For detecting field not being set to a supported value
+  FUNCTION = 6;
+}
+
+enum SecurableKind {
+  UNKNOWN_SECURABLE_KIND = 0;  // For detecting field not being set to a supported value
+  FUNCTION_REGISTERED_MODEL = 71;
+}
+
+message RegisteredModelAliasInfo {
+  // Name of the alias.
+  optional string alias_name = 1;
+  // Numeric model version that alias will reference.
+  optional int64 version_num = 2;
+}
diff --git a/mlflow/protos/unity_catalog_oss_messages_pb2.py b/mlflow/protos/unity_catalog_oss_messages_pb2.py
new file mode 100644
index 000000000..2bbe5950b
--- /dev/null
+++ b/mlflow/protos/unity_catalog_oss_messages_pb2.py
@@ -0,0 +1,139 @@
+
+import google.protobuf
+from packaging.version import Version
+if Version(google.protobuf.__version__).major >= 5:
+  # -*- coding: utf-8 -*-
+  # Generated by the protocol buffer compiler.  DO NOT EDIT!
+  # source: unity_catalog_oss_messages.proto
+  # Protobuf Python Version: 5.26.0
+  """Generated protocol buffer code."""
+  from google.protobuf import descriptor as _descriptor
+  from google.protobuf import descriptor_pool as _descriptor_pool
+  from google.protobuf import symbol_database as _symbol_database
+  from google.protobuf.internal import builder as _builder
+  # @@protoc_insertion_point(imports)
+
+  _sym_db = _symbol_database.Default()
+
+
+  from .scalapb import scalapb_pb2 as scalapb_dot_scalapb__pb2
+  from . import databricks_pb2 as databricks__pb2
+
+
+  DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n unity_catalog_oss_messages.proto\x12\x13mlflow.unitycatalog\x1a\x15scalapb/scalapb.proto\x1a\x10\x64\x61tabricks.proto\"\xb5\x01\n\x15\x43reateRegisteredModel\x12G\n\x15registered_model_info\x18\x01 \x01(\x0b\x32(.mlflow.unitycatalog.RegisteredModelInfo\x1aS\n\x08Response\x12G\n\x15registered_model_info\x18\x01 \x01(\x0b\x32(.mlflow.unitycatalog.RegisteredModelInfo\"\x8a\x04\n\x13RegisteredModelInfo\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x14\n\x0c\x63\x61talog_name\x18\x02 \x01(\t\x12\x13\n\x0bschema_name\x18\x03 \x01(\t\x12\r\n\x05owner\x18\x04 \x01(\t\x12\x0f\n\x07\x63omment\x18\x05 \x01(\t\x12.\n\x04tags\x18\x14 \x03(\x0b\x32 .mlflow.unitycatalog.TagKeyValue\x12\x18\n\x10storage_location\x18\x06 \x01(\t\x12\x14\n\x0cmetastore_id\x18\x07 \x01(\t\x12\x11\n\tfull_name\x18\t \x01(\t\x12\x12\n\ncreated_at\x18\x0b \x01(\x03\x12\x12\n\ncreated_by\x18\x0c \x01(\t\x12\x12\n\nupdated_at\x18\r \x01(\x03\x12\x12\n\nupdated_by\x18\x0e \x01(\t\x12:\n\x0esecurable_type\x18\x0f \x01(\x0e\x32\".mlflow.unitycatalog.SecurableType\x12:\n\x0esecurable_kind\x18\x10 \x01(\x0e\x32\".mlflow.unitycatalog.SecurableKind\x12\n\n\x02id\x18\x12 \x01(\t\x12>\n\x07\x61liases\x18\x13 \x03(\x0b\x32-.mlflow.unitycatalog.RegisteredModelAliasInfo\x12\x13\n\x0b\x62rowse_only\x18\x15 \x01(\x08\")\n\x0bTagKeyValue\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t\"C\n\x18RegisteredModelAliasInfo\x12\x12\n\nalias_name\x18\x01 \x01(\t\x12\x13\n\x0bversion_num\x18\x02 \x01(\x03*9\n\rSecurableType\x12\x1a\n\x16UNKNOWN_SECURABLE_TYPE\x10\x00\x12\x0c\n\x08\x46UNCTION\x10\x06*J\n\rSecurableKind\x12\x1a\n\x16UNKNOWN_SECURABLE_KIND\x10\x00\x12\x1d\n\x19\x46UNCTION_REGISTERED_MODEL\x10GB1\n\'com.databricks.api.proto.managedcatalog\xa0\x01\x01\xe2?\x02\x10\x01')
+
+  _globals = globals()
+  _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
+  _builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'unity_catalog_oss_messages_pb2', _globals)
+  if not _descriptor._USE_C_DESCRIPTORS:
+    _globals['DESCRIPTOR']._loaded_options = None
+    _globals['DESCRIPTOR']._serialized_options = b'\n\'com.databricks.api.proto.managedcatalog\240\001\001\342?\002\020\001'
+    _globals['_SECURABLETYPE']._serialized_start=919
+    _globals['_SECURABLETYPE']._serialized_end=976
+    _globals['_SECURABLEKIND']._serialized_start=978
+    _globals['_SECURABLEKIND']._serialized_end=1052
+    _globals['_CREATEREGISTEREDMODEL']._serialized_start=99
+    _globals['_CREATEREGISTEREDMODEL']._serialized_end=280
+    _globals['_CREATEREGISTEREDMODEL_RESPONSE']._serialized_start=197
+    _globals['_CREATEREGISTEREDMODEL_RESPONSE']._serialized_end=280
+    _globals['_REGISTEREDMODELINFO']._serialized_start=283
+    _globals['_REGISTEREDMODELINFO']._serialized_end=805
+    _globals['_TAGKEYVALUE']._serialized_start=807
+    _globals['_TAGKEYVALUE']._serialized_end=848
+    _globals['_REGISTEREDMODELALIASINFO']._serialized_start=850
+    _globals['_REGISTEREDMODELALIASINFO']._serialized_end=917
+  # @@protoc_insertion_point(module_scope)
+
+else:
+  # -*- coding: utf-8 -*-
+  # Generated by the protocol buffer compiler.  DO NOT EDIT!
+  # source: unity_catalog_oss_messages.proto
+  """Generated protocol buffer code."""
+  from google.protobuf.internal import enum_type_wrapper
+  from google.protobuf import descriptor as _descriptor
+  from google.protobuf import descriptor_pool as _descriptor_pool
+  from google.protobuf import message as _message
+  from google.protobuf import reflection as _reflection
+  from google.protobuf import symbol_database as _symbol_database
+  # @@protoc_insertion_point(imports)
+
+  _sym_db = _symbol_database.Default()
+
+
+  from .scalapb import scalapb_pb2 as scalapb_dot_scalapb__pb2
+  from . import databricks_pb2 as databricks__pb2
+
+
+  DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n unity_catalog_oss_messages.proto\x12\x13mlflow.unitycatalog\x1a\x15scalapb/scalapb.proto\x1a\x10\x64\x61tabricks.proto\"\xb5\x01\n\x15\x43reateRegisteredModel\x12G\n\x15registered_model_info\x18\x01 \x01(\x0b\x32(.mlflow.unitycatalog.RegisteredModelInfo\x1aS\n\x08Response\x12G\n\x15registered_model_info\x18\x01 \x01(\x0b\x32(.mlflow.unitycatalog.RegisteredModelInfo\"\x8a\x04\n\x13RegisteredModelInfo\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x14\n\x0c\x63\x61talog_name\x18\x02 \x01(\t\x12\x13\n\x0bschema_name\x18\x03 \x01(\t\x12\r\n\x05owner\x18\x04 \x01(\t\x12\x0f\n\x07\x63omment\x18\x05 \x01(\t\x12.\n\x04tags\x18\x14 \x03(\x0b\x32 .mlflow.unitycatalog.TagKeyValue\x12\x18\n\x10storage_location\x18\x06 \x01(\t\x12\x14\n\x0cmetastore_id\x18\x07 \x01(\t\x12\x11\n\tfull_name\x18\t \x01(\t\x12\x12\n\ncreated_at\x18\x0b \x01(\x03\x12\x12\n\ncreated_by\x18\x0c \x01(\t\x12\x12\n\nupdated_at\x18\r \x01(\x03\x12\x12\n\nupdated_by\x18\x0e \x01(\t\x12:\n\x0esecurable_type\x18\x0f \x01(\x0e\x32\".mlflow.unitycatalog.SecurableType\x12:\n\x0esecurable_kind\x18\x10 \x01(\x0e\x32\".mlflow.unitycatalog.SecurableKind\x12\n\n\x02id\x18\x12 \x01(\t\x12>\n\x07\x61liases\x18\x13 \x03(\x0b\x32-.mlflow.unitycatalog.RegisteredModelAliasInfo\x12\x13\n\x0b\x62rowse_only\x18\x15 \x01(\x08\")\n\x0bTagKeyValue\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t\"C\n\x18RegisteredModelAliasInfo\x12\x12\n\nalias_name\x18\x01 \x01(\t\x12\x13\n\x0bversion_num\x18\x02 \x01(\x03*9\n\rSecurableType\x12\x1a\n\x16UNKNOWN_SECURABLE_TYPE\x10\x00\x12\x0c\n\x08\x46UNCTION\x10\x06*J\n\rSecurableKind\x12\x1a\n\x16UNKNOWN_SECURABLE_KIND\x10\x00\x12\x1d\n\x19\x46UNCTION_REGISTERED_MODEL\x10GB1\n\'com.databricks.api.proto.managedcatalog\xa0\x01\x01\xe2?\x02\x10\x01')
+
+  _SECURABLETYPE = DESCRIPTOR.enum_types_by_name['SecurableType']
+  SecurableType = enum_type_wrapper.EnumTypeWrapper(_SECURABLETYPE)
+  _SECURABLEKIND = DESCRIPTOR.enum_types_by_name['SecurableKind']
+  SecurableKind = enum_type_wrapper.EnumTypeWrapper(_SECURABLEKIND)
+  UNKNOWN_SECURABLE_TYPE = 0
+  FUNCTION = 6
+  UNKNOWN_SECURABLE_KIND = 0
+  FUNCTION_REGISTERED_MODEL = 71
+
+
+  _CREATEREGISTEREDMODEL = DESCRIPTOR.message_types_by_name['CreateRegisteredModel']
+  _CREATEREGISTEREDMODEL_RESPONSE = _CREATEREGISTEREDMODEL.nested_types_by_name['Response']
+  _REGISTEREDMODELINFO = DESCRIPTOR.message_types_by_name['RegisteredModelInfo']
+  _TAGKEYVALUE = DESCRIPTOR.message_types_by_name['TagKeyValue']
+  _REGISTEREDMODELALIASINFO = DESCRIPTOR.message_types_by_name['RegisteredModelAliasInfo']
+  CreateRegisteredModel = _reflection.GeneratedProtocolMessageType('CreateRegisteredModel', (_message.Message,), {
+
+    'Response' : _reflection.GeneratedProtocolMessageType('Response', (_message.Message,), {
+      'DESCRIPTOR' : _CREATEREGISTEREDMODEL_RESPONSE,
+      '__module__' : 'unity_catalog_oss_messages_pb2'
+      # @@protoc_insertion_point(class_scope:mlflow.unitycatalog.CreateRegisteredModel.Response)
+      })
+    ,
+    'DESCRIPTOR' : _CREATEREGISTEREDMODEL,
+    '__module__' : 'unity_catalog_oss_messages_pb2'
+    # @@protoc_insertion_point(class_scope:mlflow.unitycatalog.CreateRegisteredModel)
+    })
+  _sym_db.RegisterMessage(CreateRegisteredModel)
+  _sym_db.RegisterMessage(CreateRegisteredModel.Response)
+
+  RegisteredModelInfo = _reflection.GeneratedProtocolMessageType('RegisteredModelInfo', (_message.Message,), {
+    'DESCRIPTOR' : _REGISTEREDMODELINFO,
+    '__module__' : 'unity_catalog_oss_messages_pb2'
+    # @@protoc_insertion_point(class_scope:mlflow.unitycatalog.RegisteredModelInfo)
+    })
+  _sym_db.RegisterMessage(RegisteredModelInfo)
+
+  TagKeyValue = _reflection.GeneratedProtocolMessageType('TagKeyValue', (_message.Message,), {
+    'DESCRIPTOR' : _TAGKEYVALUE,
+    '__module__' : 'unity_catalog_oss_messages_pb2'
+    # @@protoc_insertion_point(class_scope:mlflow.unitycatalog.TagKeyValue)
+    })
+  _sym_db.RegisterMessage(TagKeyValue)
+
+  RegisteredModelAliasInfo = _reflection.GeneratedProtocolMessageType('RegisteredModelAliasInfo', (_message.Message,), {
+    'DESCRIPTOR' : _REGISTEREDMODELALIASINFO,
+    '__module__' : 'unity_catalog_oss_messages_pb2'
+    # @@protoc_insertion_point(class_scope:mlflow.unitycatalog.RegisteredModelAliasInfo)
+    })
+  _sym_db.RegisterMessage(RegisteredModelAliasInfo)
+
+  if _descriptor._USE_C_DESCRIPTORS == False:
+
+    DESCRIPTOR._options = None
+    DESCRIPTOR._serialized_options = b'\n\'com.databricks.api.proto.managedcatalog\240\001\001\342?\002\020\001'
+    _SECURABLETYPE._serialized_start=919
+    _SECURABLETYPE._serialized_end=976
+    _SECURABLEKIND._serialized_start=978
+    _SECURABLEKIND._serialized_end=1052
+    _CREATEREGISTEREDMODEL._serialized_start=99
+    _CREATEREGISTEREDMODEL._serialized_end=280
+    _CREATEREGISTEREDMODEL_RESPONSE._serialized_start=197
+    _CREATEREGISTEREDMODEL_RESPONSE._serialized_end=280
+    _REGISTEREDMODELINFO._serialized_start=283
+    _REGISTEREDMODELINFO._serialized_end=805
+    _TAGKEYVALUE._serialized_start=807
+    _TAGKEYVALUE._serialized_end=848
+    _REGISTEREDMODELALIASINFO._serialized_start=850
+    _REGISTEREDMODELALIASINFO._serialized_end=917
+  # @@protoc_insertion_point(module_scope)
+
diff --git a/mlflow/protos/unity_catalog_oss_service.proto b/mlflow/protos/unity_catalog_oss_service.proto
new file mode 100644
index 000000000..8921aeb26
--- /dev/null
+++ b/mlflow/protos/unity_catalog_oss_service.proto
@@ -0,0 +1,29 @@
+syntax = "proto2";
+
+package mlflow.unitycatalog;
+
+import "databricks.proto";
+import "scalapb/scalapb.proto";
+import "unity_catalog_oss_messages.proto";
+
+option java_package = "com.databricks.api.proto.managedcatalog";
+option java_generate_equals_and_hash = true;
+option py_generic_services = true;
+option (scalapb.options) = {
+  flat_package: true,
+};
+
+service UnityCatalogService {
+  // Registered Model Management API
+  rpc createRegisteredModel(CreateRegisteredModel) returns (CreateRegisteredModel.Response) {
+    option (rpc) = {
+      endpoints: {
+        method: "POST",
+        path: "/unity-catalog/models",
+        since { major: 2, minor: 1 },
+      },
+      // https://docs.databricks.com/api/workspace/registeredmodels/create
+      visibility: PUBLIC,
+    };
+  }
+}
diff --git a/mlflow/protos/unity_catalog_oss_service_pb2.py b/mlflow/protos/unity_catalog_oss_service_pb2.py
new file mode 100644
index 000000000..a718ed295
--- /dev/null
+++ b/mlflow/protos/unity_catalog_oss_service_pb2.py
@@ -0,0 +1,86 @@
+
+import google.protobuf
+from packaging.version import Version
+if Version(google.protobuf.__version__).major >= 5:
+  # -*- coding: utf-8 -*-
+  # Generated by the protocol buffer compiler.  DO NOT EDIT!
+  # source: unity_catalog_oss_service.proto
+  # Protobuf Python Version: 5.26.0
+  """Generated protocol buffer code."""
+  from google.protobuf import descriptor as _descriptor
+  from google.protobuf import descriptor_pool as _descriptor_pool
+  from google.protobuf import symbol_database as _symbol_database
+  from google.protobuf.internal import builder as _builder
+  # @@protoc_insertion_point(imports)
+
+  _sym_db = _symbol_database.Default()
+
+
+  from . import databricks_pb2 as databricks__pb2
+  from .scalapb import scalapb_pb2 as scalapb_dot_scalapb__pb2
+  from . import unity_catalog_oss_messages_pb2 as unity_catalog_oss_messages_pb2
+
+
+  DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1funity_catalog_oss_service.proto\x12\x13mlflow.unitycatalog\x1a\x10\x64\x61tabricks.proto\x1a\x15scalapb/scalapb.proto\x1a unity_catalog_oss_messages.proto2\xbd\x01\n\x13UnityCatalogService\x12\xa5\x01\n\x15\x63reateRegisteredModel\x12*.mlflow.unitycatalog.CreateRegisteredModel\x1a\x33.mlflow.unitycatalog.CreateRegisteredModel.Response\"+\xf2\x86\x19\'\n#\n\x04POST\x12\x15/unity-catalog/models\x1a\x04\x08\x02\x10\x01\x10\x01\x42\x34\n\'com.databricks.api.proto.managedcatalog\x90\x01\x01\xa0\x01\x01\xe2?\x02\x10\x01')
+
+  _globals = globals()
+  _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
+  _builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'unity_catalog_oss_service_pb2', _globals)
+  if not _descriptor._USE_C_DESCRIPTORS:
+    _globals['DESCRIPTOR']._loaded_options = None
+    _globals['DESCRIPTOR']._serialized_options = b'\n\'com.databricks.api.proto.managedcatalog\220\001\001\240\001\001\342?\002\020\001'
+    _globals['_UNITYCATALOGSERVICE'].methods_by_name['createRegisteredModel']._loaded_options = None
+    _globals['_UNITYCATALOGSERVICE'].methods_by_name['createRegisteredModel']._serialized_options = b'\362\206\031\'\n#\n\004POST\022\025/unity-catalog/models\032\004\010\002\020\001\020\001'
+    _globals['_UNITYCATALOGSERVICE']._serialized_start=132
+    _globals['_UNITYCATALOGSERVICE']._serialized_end=321
+  _builder.BuildServices(DESCRIPTOR, 'unity_catalog_oss_service_pb2', _globals)
+  # @@protoc_insertion_point(module_scope)
+
+else:
+  # -*- coding: utf-8 -*-
+  # Generated by the protocol buffer compiler.  DO NOT EDIT!
+  # source: unity_catalog_oss_service.proto
+  """Generated protocol buffer code."""
+  from google.protobuf import descriptor as _descriptor
+  from google.protobuf import descriptor_pool as _descriptor_pool
+  from google.protobuf import message as _message
+  from google.protobuf import reflection as _reflection
+  from google.protobuf import symbol_database as _symbol_database
+  from google.protobuf import service as _service
+  from google.protobuf import service_reflection
+  # @@protoc_insertion_point(imports)
+
+  _sym_db = _symbol_database.Default()
+
+
+  from . import databricks_pb2 as databricks__pb2
+  from .scalapb import scalapb_pb2 as scalapb_dot_scalapb__pb2
+  from . import unity_catalog_oss_messages_pb2 as unity_catalog_oss_messages_pb2
+
+
+  DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1funity_catalog_oss_service.proto\x12\x13mlflow.unitycatalog\x1a\x10\x64\x61tabricks.proto\x1a\x15scalapb/scalapb.proto\x1a unity_catalog_oss_messages.proto2\xbd\x01\n\x13UnityCatalogService\x12\xa5\x01\n\x15\x63reateRegisteredModel\x12*.mlflow.unitycatalog.CreateRegisteredModel\x1a\x33.mlflow.unitycatalog.CreateRegisteredModel.Response\"+\xf2\x86\x19\'\n#\n\x04POST\x12\x15/unity-catalog/models\x1a\x04\x08\x02\x10\x01\x10\x01\x42\x34\n\'com.databricks.api.proto.managedcatalog\x90\x01\x01\xa0\x01\x01\xe2?\x02\x10\x01')
+
+
+
+  _UNITYCATALOGSERVICE = DESCRIPTOR.services_by_name['UnityCatalogService']
+  if _descriptor._USE_C_DESCRIPTORS == False:
+
+    DESCRIPTOR._options = None
+    DESCRIPTOR._serialized_options = b'\n\'com.databricks.api.proto.managedcatalog\220\001\001\240\001\001\342?\002\020\001'
+    _UNITYCATALOGSERVICE.methods_by_name['createRegisteredModel']._options = None
+    _UNITYCATALOGSERVICE.methods_by_name['createRegisteredModel']._serialized_options = b'\362\206\031\'\n#\n\004POST\022\025/unity-catalog/models\032\004\010\002\020\001\020\001'
+    _UNITYCATALOGSERVICE._serialized_start=132
+    _UNITYCATALOGSERVICE._serialized_end=321
+  UnityCatalogService = service_reflection.GeneratedServiceType('UnityCatalogService', (_service.Service,), dict(
+    DESCRIPTOR = _UNITYCATALOGSERVICE,
+    __module__ = 'unity_catalog_oss_service_pb2'
+    ))
+
+  UnityCatalogService_Stub = service_reflection.GeneratedServiceStubType('UnityCatalogService_Stub', (UnityCatalogService,), dict(
+    DESCRIPTOR = _UNITYCATALOGSERVICE,
+    __module__ = 'unity_catalog_oss_service_pb2'
+    ))
+
+
+  # @@protoc_insertion_point(module_scope)
+
diff --git a/mlflow/store/_unity_catalog/registry_oss/__init__.py b/mlflow/store/_unity_catalog/registry_oss/__init__.py
new file mode 100644
index 000000000..2d5c9ebda
--- /dev/null
+++ b/mlflow/store/_unity_catalog/registry_oss/__init__.py
@@ -0,0 +1,1 @@
+from mlflow.store._unity_catalog.registry_oss import rest_store_oss  # noqa: F401
diff --git a/mlflow/store/_unity_catalog/registry_oss/rest_store_oss.py b/mlflow/store/_unity_catalog/registry_oss/rest_store_oss.py
new file mode 100644
index 000000000..93d89efea
--- /dev/null
+++ b/mlflow/store/_unity_catalog/registry_oss/rest_store_oss.py
@@ -0,0 +1,138 @@
+import functools
+
+from mlflow.protos.unity_catalog_oss_messages_pb2 import (
+    CreateRegisteredModel,
+    RegisteredModelInfo,
+    TagKeyValue,
+)
+from mlflow.protos.unity_catalog_oss_service_pb2 import UnityCatalogService
+from mlflow.store.model_registry.base_rest_store import BaseRestStore
+from mlflow.utils._unity_catalog_oss_utils import registered_model_from_uc_oss_proto
+from mlflow.utils._unity_catalog_utils import get_full_name_from_sc
+from mlflow.utils.annotations import experimental
+from mlflow.utils.databricks_utils import get_databricks_host_creds
+from mlflow.utils.proto_json_utils import message_to_json
+from mlflow.utils.rest_utils import (
+    _UC_OSS_REST_API_PATH_PREFIX,
+    extract_all_api_info_for_service,
+    extract_api_info_for_service,
+)
+
+_METHOD_TO_INFO = extract_api_info_for_service(UnityCatalogService, _UC_OSS_REST_API_PATH_PREFIX)
+_METHOD_TO_ALL_INFO = extract_all_api_info_for_service(
+    UnityCatalogService, _UC_OSS_REST_API_PATH_PREFIX
+)
+
+
+@experimental
+class UnityCatalogOssStore(BaseRestStore):
+    """
+    Client for an Open Source Unity Catalog Server accessed via REST API calls.
+    """
+
+    def __init__(self, store_uri):
+        super().__init__(get_host_creds=functools.partial(get_databricks_host_creds, store_uri))
+
+    def _get_response_from_method(self, method):
+        method_to_response = {
+            CreateRegisteredModel: RegisteredModelInfo,
+        }
+        return method_to_response[method]()
+
+    def _get_endpoint_from_method(self, method):
+        return _METHOD_TO_INFO[method]
+
+    def _get_all_endpoints_from_method(self, method):
+        return _METHOD_TO_ALL_INFO[method]
+
+    def create_registered_model(self, name, tags=None, description=None):
+        full_name = get_full_name_from_sc(name, None)
+        [catalog_name, schema_name, model_name] = full_name.split(".")
+        comment = description if description else ""
+        tags = [TagKeyValue(key=tag.key, value=tag.value) for tag in (tags or [])]
+        # RegisteredModelInfo is inlined in the request and the response.
+        # https://docs.databricks.com/api/workspace/registeredmodels/create
+        req_body = message_to_json(
+            RegisteredModelInfo(
+                name=model_name,
+                catalog_name=catalog_name,
+                schema_name=schema_name,
+                comment=comment,
+                tags=tags,
+            )
+        )
+        registered_model_info = self._call_endpoint(CreateRegisteredModel, req_body)
+        return registered_model_from_uc_oss_proto(registered_model_info)
+
+    def update_registered_model(self, name, description):
+        raise NotImplementedError("Method not implemented")
+
+    def rename_registered_model(self, name, new_name):
+        raise NotImplementedError("Method not implemented")
+
+    def delete_registered_model(self, name):
+        raise NotImplementedError("Method not implemented")
+
+    def search_registered_models(
+        self, filter_string=None, max_results=None, order_by=None, page_token=None
+    ):
+        raise NotImplementedError("Method not implemented")
+
+    def get_registered_model(self, name):
+        raise NotImplementedError("Method not implemented")
+
+    def get_latest_versions(self, name, stages=None):
+        raise NotImplementedError("Method not implemented")
+
+    def set_registered_model_tag(self, name, tag):
+        raise NotImplementedError("Method not implemented")
+
+    def delete_registered_model_tag(self, name, key):
+        raise NotImplementedError("Method not implemented")
+
+    def create_model_version(
+        self,
+        name,
+        source,
+        run_id=None,
+        tags=None,
+        run_link=None,
+        description=None,
+        local_model_path=None,
+    ):
+        raise NotImplementedError("Method not implemented")
+
+    def update_model_version(self, name, version, description):
+        raise NotImplementedError("Method not implemented")
+
+    def transition_model_version_stage(self, name, version, stage, archive_existing_versions):
+        raise NotImplementedError("Method not implemented")
+
+    def delete_model_version(self, name, version):
+        raise NotImplementedError("Method not implemented")
+
+    def get_model_version(self, name, version):
+        raise NotImplementedError("Method not implemented")
+
+    def get_model_version_download_uri(self, name, version):
+        raise NotImplementedError("Method not implemented")
+
+    def search_model_versions(
+        self, filter_string=None, max_results=None, order_by=None, page_token=None
+    ):
+        raise NotImplementedError("Method not implemented")
+
+    def set_model_version_tag(self, name, version, tag):
+        raise NotImplementedError("Method not implemented")
+
+    def delete_model_version_tag(self, name, version, key):
+        raise NotImplementedError("Method not implemented")
+
+    def set_registered_model_alias(self, name, alias, version):
+        raise NotImplementedError("Method not implemented")
+
+    def delete_registered_model_alias(self, name, alias):
+        raise NotImplementedError("Method not implemented")
+
+    def get_model_version_by_alias(self, name, alias):
+        raise NotImplementedError("Method not implemented")
diff --git a/mlflow/tracking/_tracking_service/client.py b/mlflow/tracking/_tracking_service/client.py
index 4e4adba45..70f5dd386 100644
--- a/mlflow/tracking/_tracking_service/client.py
+++ b/mlflow/tracking/_tracking_service/client.py
@@ -696,6 +696,16 @@ class TrackingServiceClient:
         if len(metrics) == 0 and len(params) == 0 and len(tags) == 0:
             return
 
+        metrics = [
+            Metric(
+                metric.key,
+                convert_metric_value_to_float_if_possible(metric.value),
+                metric.timestamp,
+                metric.step,
+            )
+            for metric in metrics
+        ]
+
         param_batches = chunk_list(params, MAX_PARAMS_TAGS_PER_BATCH)
         tag_batches = chunk_list(tags, MAX_PARAMS_TAGS_PER_BATCH)
 
diff --git a/mlflow/utils/_unity_catalog_oss_utils.py b/mlflow/utils/_unity_catalog_oss_utils.py
new file mode 100644
index 000000000..273d2a8b2
--- /dev/null
+++ b/mlflow/utils/_unity_catalog_oss_utils.py
@@ -0,0 +1,28 @@
+from typing import List, Optional
+
+from mlflow.entities.model_registry import RegisteredModel, RegisteredModelAlias, RegisteredModelTag
+from mlflow.protos.unity_catalog_oss_messages_pb2 import RegisteredModelInfo, TagKeyValue
+
+
+def registered_model_from_uc_oss_proto(uc_oss_proto: RegisteredModelInfo) -> RegisteredModel:
+    return RegisteredModel(
+        name=f"{uc_oss_proto.catalog_name}.{uc_oss_proto.schema_name}.{uc_oss_proto.name}",
+        creation_timestamp=uc_oss_proto.created_at,
+        last_updated_timestamp=uc_oss_proto.updated_at,
+        description=uc_oss_proto.comment,
+        aliases=[
+            RegisteredModelAlias(alias=alias.alias_name, version=str(alias.version_num))
+            for alias in (uc_oss_proto.aliases or [])
+        ],
+        tags=[
+            RegisteredModelTag(key=tag.key, value=tag.value) for tag in (uc_oss_proto.tags or [])
+        ],
+    )
+
+
+def uc_oss_registered_model_tag_from_mlflow_tags(
+    tags: Optional[List[RegisteredModelTag]],
+) -> List[TagKeyValue]:
+    if tags is None:
+        return []
+    return [TagKeyValue(key=t.key, value=t.value) for t in tags]
diff --git a/mlflow/utils/rest_utils.py b/mlflow/utils/rest_utils.py
index 0b2320c8e..aae758e11 100644
--- a/mlflow/utils/rest_utils.py
+++ b/mlflow/utils/rest_utils.py
@@ -36,6 +36,7 @@ from mlflow.utils.string_utils import strip_suffix
 
 RESOURCE_NON_EXISTENT = "RESOURCE_DOES_NOT_EXIST"
 _REST_API_PATH_PREFIX = "/api/2.0"
+_UC_OSS_REST_API_PATH_PREFIX = "/api/2.1"
 _TRACE_REST_API_PATH_PREFIX = f"{_REST_API_PATH_PREFIX}/mlflow/traces"
 
 


Test patch:
diff --git a/tests/store/_unity_catalog/conftest.py b/tests/store/_unity_catalog/conftest.py
index 7efe6ae50..5f336ecc1 100644
--- a/tests/store/_unity_catalog/conftest.py
+++ b/tests/store/_unity_catalog/conftest.py
@@ -28,6 +28,15 @@ def mock_databricks_uc_host_creds():
         yield
 
 
+@pytest.fixture
+def mock_databricks_uc_oss_host_creds():
+    with mock.patch(
+        "mlflow.store._unity_catalog.registry_oss.rest_store_oss.get_databricks_host_creds",
+        side_effect=mock_host_creds,
+    ):
+        yield
+
+
 @pytest.fixture
 def configure_client_for_uc(mock_databricks_uc_host_creds):
     """
diff --git a/tests/store/_unity_catalog/oss_registry/test_unity_catalog_oss_rest_store.py b/tests/store/_unity_catalog/oss_registry/test_unity_catalog_oss_rest_store.py
new file mode 100644
index 000000000..7ddc6ed8f
--- /dev/null
+++ b/tests/store/_unity_catalog/oss_registry/test_unity_catalog_oss_rest_store.py
@@ -0,0 +1,74 @@
+import json
+from unittest import mock
+
+import pytest
+
+from mlflow.entities.model_registry import RegisteredModelTag
+from mlflow.protos.unity_catalog_oss_messages_pb2 import (
+    RegisteredModelInfo,
+)
+from mlflow.store._unity_catalog.registry_oss.rest_store_oss import UnityCatalogOssStore
+from mlflow.utils._unity_catalog_oss_utils import uc_oss_registered_model_tag_from_mlflow_tags
+from mlflow.utils.proto_json_utils import message_to_json
+
+from tests.helper_functions import mock_http_200
+from tests.store._unity_catalog.conftest import _REGISTRY_HOST_CREDS
+
+
+@pytest.fixture
+def store(mock_databricks_uc_oss_host_creds):
+    with mock.patch("mlflow.utils.databricks_utils.get_databricks_host_creds"):
+        return UnityCatalogOssStore(store_uri="databricks-uc")
+
+
+def _args(endpoint, method, json_body, host_creds, extra_headers):
+    res = {
+        "host_creds": host_creds,
+        "endpoint": f"/api/2.1/unity-catalog/{endpoint}",
+        "method": method,
+        "extra_headers": extra_headers,
+    }
+    if extra_headers is None:
+        del res["extra_headers"]
+    if method == "GET":
+        res["params"] = json.loads(json_body)
+    else:
+        res["json"] = json.loads(json_body)
+    return res
+
+
+def _verify_requests(
+    http_request,
+    endpoint,
+    method,
+    proto_message,
+    host_creds=_REGISTRY_HOST_CREDS,
+    extra_headers=None,
+):
+    json_body = message_to_json(proto_message)
+    call_args = _args(endpoint, method, json_body, host_creds, extra_headers)
+    http_request.assert_any_call(**call_args)
+
+
+@mock_http_200
+def test_create_registered_model(mock_http, store):
+    description = "best model ever"
+    tags = [
+        RegisteredModelTag(key="key", value="value"),
+        RegisteredModelTag(key="anotherKey", value="some other value"),
+    ]
+    store.create_registered_model(
+        name="catalog_1.schema_1.model_1", description=description, tags=tags
+    )
+    _verify_requests(
+        mock_http,
+        "models",
+        "POST",
+        RegisteredModelInfo(
+            name="model_1",
+            catalog_name="catalog_1",
+            schema_name="schema_1",
+            comment=description,
+            tags=uc_oss_registered_model_tag_from_mlflow_tags(tags),
+        ),
+    )
diff --git a/tests/tracking/_tracking_service/test_tracking_service_client.py b/tests/tracking/_tracking_service/test_tracking_service_client.py
index 9cd440e46..f818913ec 100644
--- a/tests/tracking/_tracking_service/test_tracking_service_client.py
+++ b/tests/tracking/_tracking_service/test_tracking_service_client.py
@@ -6,7 +6,7 @@ import pydantic
 import pytest
 
 from mlflow import MlflowClient
-from mlflow.entities import Run, RunInfo
+from mlflow.entities import Metric, Param, Run, RunInfo, RunTag
 from mlflow.entities.trace_data import TraceData
 from mlflow.entities.trace_info import TraceInfo
 from mlflow.entities.trace_status import TraceStatus
@@ -430,3 +430,65 @@ def test_search_traces_with_filestore(tmp_path):
         assert res2[0].info == trace_infos[0]
         assert res2.token is None
         assert mock_download_trace_data.call_count == 3
+
+
+@pytest.fixture
+def tracking_client_log_batch(tmp_path):
+    client = TrackingServiceClient(tmp_path.as_uri())
+    exp_id = client.create_experiment("test_log_batch")
+    run = client.create_run(exp_id)
+    return client, run.info.run_id
+
+
+def test_log_batch(tracking_client_log_batch):
+    client, run_id = tracking_client_log_batch
+
+    metrics = [
+        Metric(key="metric1", value=1.0, timestamp=12345, step=0),
+        Metric(key="metric2", value=2.0, timestamp=23456, step=1),
+    ]
+
+    params = [Param(key="param1", value="value1"), Param(key="param2", value="value2")]
+
+    tags = [RunTag(key="tag1", value="value1"), RunTag(key="tag2", value="value2")]
+
+    client.log_batch(run_id=run_id, metrics=metrics, params=params, tags=tags)
+    run_data = client.get_run(run_id).data
+
+    expected_tags = {tag.key: tag.value for tag in tags}
+    expected_tags["mlflow.runName"] = run_data.tags["mlflow.runName"]
+
+    assert run_data.metrics == {metric.key: metric.value for metric in metrics}
+    assert run_data.params == {param.key: param.value for param in params}
+    assert run_data.tags == expected_tags
+
+
+def test_log_batch_with_empty_data(tracking_client_log_batch):
+    client, run_id = tracking_client_log_batch
+
+    client.log_batch(run_id=run_id, metrics=[], params=[], tags=[])
+    run_data = client.get_run(run_id).data
+
+    assert run_data.metrics == {}
+    assert run_data.params == {}
+    assert run_data.tags == {"mlflow.runName": run_data.tags["mlflow.runName"]}
+
+
+def test_log_batch_with_numpy_array(tracking_client_log_batch):
+    import numpy as np
+
+    client, run_id = tracking_client_log_batch
+
+    metrics = [Metric(key="metric1", value=np.array(1.0), timestamp=12345, step=0)]
+    params = [Param(key="param1", value="value1")]
+    tags = [RunTag(key="tag1", value="value1")]
+
+    client.log_batch(run_id=run_id, metrics=metrics, params=params, tags=tags)
+    run_data = client.get_run(run_id).data
+
+    expected_tags = {tag.key: tag.value for tag in tags}
+    expected_tags["mlflow.runName"] = run_data.tags["mlflow.runName"]
+
+    assert run_data.metrics == {metric.key: metric.value for metric in metrics}
+    assert run_data.params == {param.key: param.value for param in params}
+    assert run_data.tags == expected_tags
diff --git a/tests/utils/test_unity_catalog_oss_utils.py b/tests/utils/test_unity_catalog_oss_utils.py
new file mode 100644
index 000000000..0d2922218
--- /dev/null
+++ b/tests/utils/test_unity_catalog_oss_utils.py
@@ -0,0 +1,44 @@
+from mlflow.entities.model_registry import RegisteredModel, RegisteredModelAlias, RegisteredModelTag
+from mlflow.protos.unity_catalog_oss_messages_pb2 import (
+    RegisteredModelAliasInfo,
+    RegisteredModelInfo,
+    TagKeyValue,
+)
+from mlflow.utils._unity_catalog_oss_utils import registered_model_from_uc_oss_proto
+
+
+def test_registered_model_from_uc_oss_proto():
+    expected_registered_model = RegisteredModel(
+        name="catalog.schema.name",
+        creation_timestamp=1,
+        last_updated_timestamp=2,
+        description="description",
+        aliases=[
+            RegisteredModelAlias(alias="alias1", version="1"),
+            RegisteredModelAlias(alias="alias2", version="2"),
+        ],
+        tags=[
+            RegisteredModelTag(key="key1", value="value"),
+            RegisteredModelTag(key="key2", value=""),
+        ],
+    )
+
+    uc_oss_proto = RegisteredModelInfo(
+        name="name",
+        catalog_name="catalog",
+        schema_name="schema",
+        created_at=1,
+        updated_at=2,
+        comment="description",
+        aliases=[
+            RegisteredModelAliasInfo(alias_name="alias1", version_num=1),
+            RegisteredModelAliasInfo(alias_name="alias2", version_num=2),
+        ],
+        tags=[
+            TagKeyValue(key="key1", value="value"),
+            TagKeyValue(key="key2", value=""),
+        ],
+    )
+
+    actual_registered_model = registered_model_from_uc_oss_proto(uc_oss_proto)
+    assert actual_registered_model == expected_registered_model