feast-dev · ntkathole · Jul 22, 2026 · Jul 22, 2026
@@ -48,6 +48,19 @@ message SavedDatasetSpec {
 
   // User defined metadata
   map<string, string> tags = 7;
+
+  // Optional logical namespace for hierarchical grouping.
+  // Maps to the top-level prefix in Iceberg REST Catalog API.
+  // Empty string means not set (no namespace scoping).
+  string namespace = 9;
+
+  // Optional sub-grouping within a namespace.
+  // Maps to the namespace level in Iceberg REST Catalog API.
+  // Empty string means not set (dataset sits directly under namespace).
+  string collection = 10;
+
+  // Description of the saved dataset.
+  string description = 11;
 }
 
 message SavedDatasetStorage {

@@ -417,6 +417,10 @@ message ListSavedDatasetsRequest {
     map<string,string> tags = 3;
     PaginationParams pagination = 4;
     SortingParams sorting = 5;
+    // Optional logical namespace filter. Empty string means no filter.
+    string namespace = 6;
+    // Optional collection filter. Empty string means no filter.
+    string collection = 7;
 }
 
 message ListSavedDatasetsResponse {

@@ -43,6 +43,9 @@ class RegisterDatasetRequest(BaseModel):
     full_feature_names: bool = False
     feature_service_name: Optional[str] = None
     allow_override: bool = False
+    namespace: str = ""
+    collection: str = ""
+    description: str = ""
 
 
 class CreateDatasetRequest(BaseModel):
@@ -75,10 +78,22 @@ def list_saved_datasets_all(
         limit: int = Query(50, ge=1, le=100),
         sort_by: str = Query(None),
         sort_order: str = Query("asc"),
+        namespace: Optional[str] = Query(
+            None, description="Filter by namespace (logical grouping)"
+        ),
+        collection: Optional[str] = Query(
+            None, description="Filter by collection (sub-grouping within namespace)"
+        ),
         include_relationships: bool = Query(
             False, description="Include relationships for each saved dataset"
         ),
     ):
+        extra_params = {}
+        if namespace is not None:
+            extra_params["namespace"] = namespace
+        if collection is not None:
+            extra_params["collection"] = collection
+
         return aggregate_across_projects(
             grpc_handler=grpc_handler,
             list_method=grpc_handler.ListSavedDatasets,
@@ -91,6 +106,7 @@ def list_saved_datasets_all(
             sort_by=sort_by,
             sort_order=sort_order,
             include_relationships=include_relationships,
+            extra_request_params=extra_params or None,
         )
 
     @router.get("/saved_datasets/data/{name}")
@@ -230,6 +246,12 @@ def list_saved_datasets(
         project: str = Query(...),
         allow_cache: bool = Query(default=True),
         tags: Dict[str, str] = Depends(parse_tags),
+        namespace: Optional[str] = Query(
+            None, description="Filter by namespace (logical grouping)"
+        ),
+        collection: Optional[str] = Query(
+            None, description="Filter by collection (sub-grouping within namespace)"
+        ),
         include_relationships: bool = Query(
             False, description="Include relationships for each saved dataset"
         ),
@@ -240,6 +262,8 @@ def list_saved_datasets(
             project=project,
             allow_cache=allow_cache,
             tags=tags,
+            namespace=namespace or "",
+            collection=collection or "",
             pagination=create_grpc_pagination_params(pagination_params),
             sorting=create_grpc_sorting_params(sorting_params),
         )
@@ -347,6 +371,9 @@ def register_saved_dataset(payload: RegisterDatasetRequest = Body(...)):
             full_feature_names=payload.full_feature_names,
             storage=storage_proto,
             tags=payload.tags,
+            namespace=payload.namespace,
+            collection=payload.collection,
+            description=payload.description,
         )
         if payload.feature_service_name:
             spec.feature_service_name = payload.feature_service_name

@@ -4554,20 +4554,30 @@ def delete_project(self, name: str, commit: bool = True) -> None:
         return self.registry.delete_project(name, commit=commit)
 
     def list_saved_datasets(
-        self, allow_cache: bool = False, tags: Optional[dict[str, str]] = None
+        self,
+        allow_cache: bool = False,
+        tags: Optional[dict[str, str]] = None,
+        namespace: Optional[str] = None,
+        collection: Optional[str] = None,
     ) -> List[SavedDataset]:
         """
         Retrieves the list of saved datasets from the registry.
 
         Args:
             allow_cache: Whether to allow returning saved datasets from a cached registry.
             tags: Filter by tags.
+            namespace: Filter by logical namespace grouping.
+            collection: Filter by collection sub-grouping within namespace.
 
         Returns:
             A list of saved datasets.
         """
         return self.registry.list_saved_datasets(
-            self.project, allow_cache=allow_cache, tags=tags
+            self.project,
+            allow_cache=allow_cache,
+            tags=tags,
+            namespace=namespace,
+            collection=collection,
         )
 
     async def initialize(self) -> None:

@@ -698,6 +698,8 @@ def list_saved_datasets(
         project: str,
         allow_cache: bool = False,
         tags: Optional[dict[str, str]] = None,
+        namespace: Optional[str] = None,
+        collection: Optional[str] = None,
     ) -> List[SavedDataset]:
         """
         Retrieves a list of all saved datasets in specified project
@@ -706,6 +708,8 @@ def list_saved_datasets(
             project: Feast project
             allow_cache: Whether to allow returning this dataset from a cached registry
             tags: Filter by tags
+            namespace: Filter by logical namespace grouping
+            collection: Filter by collection sub-grouping within namespace
 
         Returns:
             Returns the list of SavedDatasets

@@ -344,7 +344,11 @@ def get_saved_dataset(
 
     @abstractmethod
     def _list_saved_datasets(
-        self, project: str, tags: Optional[dict[str, str]] = None
+        self,
+        project: str,
+        tags: Optional[dict[str, str]] = None,
+        namespace: Optional[str] = None,
+        collection: Optional[str] = None,
     ) -> List[SavedDataset]:
         pass
 
@@ -353,13 +357,21 @@ def list_saved_datasets(
         project: str,
         allow_cache: bool = False,
         tags: Optional[dict[str, str]] = None,
+        namespace: Optional[str] = None,
+        collection: Optional[str] = None,
     ) -> List[SavedDataset]:
         if allow_cache:
             self._refresh_cached_registry_if_necessary()
             return proto_registry_utils.list_saved_datasets(
-                self.cached_registry_proto, project, tags
+                self.cached_registry_proto,
+                project,
+                tags,
+                namespace=namespace,
+                collection=collection,
             )
-        return self._list_saved_datasets(project, tags)
+        return self._list_saved_datasets(
+            project, tags, namespace=namespace, collection=collection
+        )
 
     @abstractmethod
     def _get_validation_reference(self, name: str, project: str) -> ValidationReference:

@@ -418,14 +418,23 @@ def list_data_sources(
 
 @registry_proto_cache_with_tags
 def list_saved_datasets(
-    registry_proto: RegistryProto, project: str, tags: Optional[dict[str, str]]
+    registry_proto: RegistryProto,
+    project: str,
+    tags: Optional[dict[str, str]],
+    namespace: Optional[str] = None,
+    collection: Optional[str] = None,
 ) -> List[SavedDataset]:
     saved_datasets = []
     for saved_dataset in registry_proto.saved_datasets:
-        if saved_dataset.spec.project == project and utils.has_all_tags(
-            saved_dataset.spec.tags, tags
-        ):
-            saved_datasets.append(SavedDataset.from_proto(saved_dataset))
+        if saved_dataset.spec.project != project:
+            continue
+        if not utils.has_all_tags(saved_dataset.spec.tags, tags):
+            continue
+        if namespace is not None and saved_dataset.spec.namespace != namespace:
+            continue
+        if collection is not None and saved_dataset.spec.collection != collection:
+            continue
+        saved_datasets.append(SavedDataset.from_proto(saved_dataset))
     return saved_datasets
 
 

@@ -1305,11 +1305,19 @@ def list_saved_datasets(
         project: str,
         allow_cache: bool = False,
         tags: Optional[dict[str, str]] = None,
+        namespace: Optional[str] = None,
+        collection: Optional[str] = None,
     ) -> List[SavedDataset]:
         registry_proto = self._get_registry_proto(
             project=project, allow_cache=allow_cache
         )
-        return proto_registry_utils.list_saved_datasets(registry_proto, project, tags)
+        return proto_registry_utils.list_saved_datasets(
+            registry_proto,
+            project,
+            tags,
+            namespace=namespace,
+            collection=collection,
+        )
 
     def delete_saved_dataset(self, name: str, project: str, commit: bool = True):
         self._prepare_registry_for_changes(project)

@@ -511,9 +511,15 @@ def list_saved_datasets(
         project: str,
         allow_cache: bool = False,
         tags: Optional[dict[str, str]] = None,
+        namespace: Optional[str] = None,
+        collection: Optional[str] = None,
     ) -> List[SavedDataset]:
         request = RegistryServer_pb2.ListSavedDatasetsRequest(
-            project=project, allow_cache=allow_cache, tags=tags
+            project=project,
+            allow_cache=allow_cache,
+            tags=tags,
+            namespace=namespace or "",
+            collection=collection or "",
         )
         response = self.stub.ListSavedDatasets(request)
         return [

@@ -945,20 +945,31 @@ def list_saved_datasets(
         project: str,
         allow_cache: bool = False,
         tags: Optional[dict[str, str]] = None,
+        namespace: Optional[str] = None,
+        collection: Optional[str] = None,
     ) -> List[SavedDataset]:
         if allow_cache:
             registry_proto = self._refresh_cached_registry_if_necessary()
             return proto_registry_utils.list_saved_datasets(
-                registry_proto, project, tags
+                registry_proto,
+                project,
+                tags,
+                namespace=namespace,
+                collection=collection,
             )
-        return self._list_objects(
+        results = self._list_objects(
             "SAVED_DATASETS",
             project,
             SavedDatasetProto,
             SavedDataset,
             "SAVED_DATASET_PROTO",
             tags=tags,
         )
+        if namespace is not None:
+            results = [sd for sd in results if sd.namespace == namespace]
+        if collection is not None:
+            results = [sd for sd in results if sd.collection == collection]
+        return results
 
     def list_stream_feature_views(
         self,

@@ -1076,9 +1076,14 @@ def _list_feature_views(
         )
 
     def _list_saved_datasets(
-        self, project: str, tags: Optional[dict[str, str]] = None, **kwargs
+        self,
+        project: str,
+        tags: Optional[dict[str, str]] = None,
+        namespace: Optional[str] = None,
+        collection: Optional[str] = None,
+        **kwargs,
     ) -> List[SavedDataset]:
-        return self._list_objects(
+        results = self._list_objects(
             saved_datasets,
             project,
             SavedDatasetProto,
@@ -1087,6 +1092,11 @@ def _list_saved_datasets(
             tags=tags,
             **kwargs,
         )
+        if namespace is not None:
+            results = [sd for sd in results if sd.namespace == namespace]
+        if collection is not None:
+            results = [sd for sd in results if sd.collection == collection]
+        return results
 
     def _list_on_demand_feature_views(
         self, project: str, tags: Optional[dict[str, str]], **kwargs