From fedd50c3bc5213266b5d68e70bd854866b908a57 Mon Sep 17 00:00:00 2001 From: Sophia Ugochukwu <104243303+SophiaUgo@users.noreply.github.com> Date: Tue, 28 Jul 2026 17:41:21 +0000 Subject: [PATCH] Add metadata RPC plumbing for namespace index registration Fix namespace index registration RPC wiring Fix namespace index registration RPC routing Format routing key match arm Signed-off-by: Sophia Ugochukwu <104243303+SophiaUgo@users.noreply.github.com> --- crates/nokv-client/src/service.rs | 82 +++++++++++++++++++++++++++--- crates/nokv-protocol/src/lib.rs | 27 +++++++++- crates/nokv-server/src/rpc/mod.rs | 12 +++-- crates/nokv-server/src/rpc/wire.rs | 71 ++++++++++++++++++++++---- 4 files changed, 169 insertions(+), 23 deletions(-) diff --git a/crates/nokv-client/src/service.rs b/crates/nokv-client/src/service.rs index 24742abc..c870d161 100644 --- a/crates/nokv-client/src/service.rs +++ b/crates/nokv-client/src/service.rs @@ -13,15 +13,17 @@ use nokv_meta::{ NamespaceFacetSummary, NamespaceFacetValue, NamespaceFieldSource, NamespaceFieldSourceKind, NamespaceFieldValue, NamespaceFilterCapability, NamespaceFindField, NamespaceFindRequest, NamespaceFindResult, NamespaceGrepMatch, NamespaceGrepRequest, NamespaceGrepResult, - NamespaceInclude, NamespaceIndexValue, NamespaceListOptions, NamespaceListPage, - NamespacePredicate, NamespacePredicateOp, NamespacePredicateValue, NamespaceQueryCatalog, - NamespaceReadFormat, NamespaceReadItem, NamespaceReadOptions, NamespaceReadPage, - NamespaceRecordCount, NamespaceRecordType, NamespaceSchema, NamespaceSort, - NamespaceSortDirection, NamespaceSortField, PublishArtifactStagedSession, - RecordCountProvenance, RenameReplaceResult, RestoreInitialization, RestoreOutcome, - RestoreState, SnapshotRenewOutcome, SubtreeDelta, UpdateAttr, XattrSetMode, + NamespaceInclude, NamespaceIndexField, NamespaceIndexRegistration, NamespaceIndexRow, + NamespaceIndexValue, NamespaceListOptions, NamespaceListPage, NamespacePredicate, + NamespacePredicateOp, NamespacePredicateValue, NamespaceQueryCatalog, NamespaceReadFormat, + NamespaceReadItem, NamespaceReadOptions, NamespaceReadPage, NamespaceRecordCount, + NamespaceRecordType, NamespaceSchema, NamespaceSort, NamespaceSortDirection, + NamespaceSortField, PublishArtifactStagedSession, RecordCountProvenance, RenameReplaceResult, + RestoreInitialization, RestoreOutcome, RestoreState, SnapshotRenewOutcome, SubtreeDelta, + UpdateAttr, XattrSetMode, }; use nokv_object::ObjectReadPlan; + use nokv_protocol::{ decode_envelope, decode_name_cursor, decode_xattr_name, encode_advisory_lock_kind, encode_file_type, encode_name_cursor, encode_request, encode_xattr_name, request_routing_key, @@ -34,7 +36,8 @@ use nokv_protocol::{ WireNamespaceFieldSourceKind, WireNamespaceFieldValue, WireNamespaceFilterCapability, WireNamespaceFindField, WireNamespaceFindRequest, WireNamespaceFindResult, WireNamespaceGrepMatch, WireNamespaceGrepRequest, WireNamespaceGrepResult, - WireNamespaceInclude, WireNamespaceIndexValue, WireNamespaceListPage, WireNamespacePredicate, + WireNamespaceInclude, WireNamespaceIndexField, WireNamespaceIndexRegistration, + WireNamespaceIndexRow, WireNamespaceIndexValue, WireNamespaceListPage, WireNamespacePredicate, WireNamespacePredicateOp, WireNamespacePredicateValue, WireNamespaceQueryCatalog, WireNamespaceReadFormat, WireNamespaceReadItem, WireNamespaceReadOptions, WireNamespaceReadPage, WireNamespaceRecordCount, WireNamespaceRecordType, WireNamespaceSchema, @@ -1310,6 +1313,18 @@ impl MetadataClient { } } + pub fn register_namespace_index( + &self, + registration: NamespaceIndexRegistration, + ) -> Result<(), ClientError> { + match self.call(MetadataRpcRequest::RegisterNamespaceIndex { + registration: Box::new(wire_namespace_index_registration(®istration)?), + })? { + MetadataRpcResult::Unit => Ok(()), + other => Err(unexpected_result(other)), + } + } + pub fn read_page( &self, path: &str, @@ -3054,6 +3069,57 @@ fn wire_namespace_find_request( }) } +fn wire_namespace_index_registration( + registration: &NamespaceIndexRegistration, +) -> Result { + Ok(WireNamespaceIndexRegistration { + path: registration.path.clone(), + fields: registration + .fields + .iter() + .map(wire_namespace_index_field) + .collect(), + rows: registration + .rows + .iter() + .map(wire_namespace_index_row) + .collect(), + }) +} + +fn wire_namespace_find_field(field: &NamespaceFindField) -> WireNamespaceFindField { + WireNamespaceFindField { + id: field.id.clone(), + } +} + +fn wire_namespace_index_field(field: &NamespaceIndexField) -> WireNamespaceIndexField { + WireNamespaceIndexField { + field: wire_namespace_find_field(&field.field), + operators: field + .operators + .iter() + .map(wire_namespace_predicate_op) + .collect(), + sortable: field.sortable, + facetable: field.facetable, + } +} + +fn wire_namespace_index_row(row: &NamespaceIndexRow) -> WireNamespaceIndexRow { + WireNamespaceIndexRow { + path: row.path.clone(), + values: row.values.iter().map(wire_namespace_index_value).collect(), + } +} + +fn wire_namespace_index_value(value: &NamespaceIndexValue) -> WireNamespaceIndexValue { + WireNamespaceIndexValue { + field: wire_namespace_find_field(&value.field), + value: wire_namespace_predicate_value(&value.value), + } +} + fn wire_namespace_aggregate_request( request: &NamespaceAggregateRequest, ) -> Result { diff --git a/crates/nokv-protocol/src/lib.rs b/crates/nokv-protocol/src/lib.rs index d39af5b8..59aa1152 100644 --- a/crates/nokv-protocol/src/lib.rs +++ b/crates/nokv-protocol/src/lib.rs @@ -121,6 +121,9 @@ pub enum MetadataRpcRequest { path: String, options: Box, }, + RegisterNamespaceIndex { + registration: Box, + }, CreateDir { parent: u64, name: String, @@ -869,6 +872,27 @@ pub struct WireNamespaceIndexValue { pub value: WireNamespacePredicateValue, } +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct WireNamespaceIndexField { + pub field: WireNamespaceFindField, + pub operators: Vec, + pub sortable: bool, + pub facetable: bool, +} + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct WireNamespaceIndexRow { + pub path: String, + pub values: Vec, +} + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +pub struct WireNamespaceIndexRegistration { + pub path: String, + pub fields: Vec, + pub rows: Vec, +} + #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum WireNamespaceSortDirection { @@ -1837,7 +1861,8 @@ pub fn request_routing_key(request: &MetadataRpcRequest) -> RoutingKey<'_> { // No addressable key: target the default/root shard. MetadataRpcRequest::Batch { .. } | MetadataRpcRequest::BootstrapRoot { .. } - | MetadataRpcRequest::UpdateRootAttrs { .. } => RoutingKey::Default, + | MetadataRpcRequest::UpdateRootAttrs { .. } + | MetadataRpcRequest::RegisterNamespaceIndex { .. } => RoutingKey::Default, } } diff --git a/crates/nokv-server/src/rpc/mod.rs b/crates/nokv-server/src/rpc/mod.rs index b7fab4a4..37045386 100644 --- a/crates/nokv-server/src/rpc/mod.rs +++ b/crates/nokv-server/src/rpc/mod.rs @@ -35,9 +35,9 @@ use crate::server::{Server, ServerError}; use batch::{create_path_batch_envelopes, execute_batch, CreatePathKind}; use wire::{ dentry_name, err_envelope, inode_id, namespace_aggregate_request, namespace_find_request, - namespace_grep_request, namespace_read_options, prepared_artifact, protocol_error, - staged_object_set, update_attr, wire_body_read_plan, wire_dentry, - wire_namespace_aggregate_result, wire_namespace_card, wire_namespace_find_result, + namespace_grep_request, namespace_index_registration, namespace_read_options, + prepared_artifact, protocol_error, staged_object_set, update_attr, wire_body_read_plan, + wire_dentry, wire_namespace_aggregate_result, wire_namespace_card, wire_namespace_find_result, wire_namespace_grep_result, wire_namespace_list_page, wire_namespace_read_page, wire_open_path_read_plan, wire_prepared_artifact, wire_subtree_delta, xattr_set_mode, }; @@ -285,6 +285,11 @@ fn execute_unfenced( result: Box::new(wire_namespace_find_result(&result)?), }) } + MetadataRpcRequest::RegisterNamespaceIndex { registration } => { + slot.service() + .register_namespace_index(namespace_index_registration(*registration))?; + Ok(MetadataRpcResult::Unit) + } MetadataRpcRequest::AggregatePaths { request } => { let result = slot .service() @@ -1176,6 +1181,7 @@ fn refreshes_metadata_view(request: &MetadataRpcRequest) -> bool { | MetadataRpcRequest::PrepareArtifactPath { .. } | MetadataRpcRequest::RefreshPreparedArtifactObjectGcEpoch { .. } | MetadataRpcRequest::PublishPreparedArtifact { .. } + | MetadataRpcRequest::RegisterNamespaceIndex { .. } | MetadataRpcRequest::PublishPreparedArtifactStagedSession { .. } => false, } } diff --git a/crates/nokv-server/src/rpc/wire.rs b/crates/nokv-server/src/rpc/wire.rs index 3d0ecc40..8b1952d9 100644 --- a/crates/nokv-server/src/rpc/wire.rs +++ b/crates/nokv-server/src/rpc/wire.rs @@ -7,13 +7,16 @@ use nokv_meta::{ NamespaceFacetValue, NamespaceFieldSource, NamespaceFieldSourceKind, NamespaceFieldValue, NamespaceFilterCapability, NamespaceFindField, NamespaceFindRequest, NamespaceFindResult, NamespaceGrepMatch, NamespaceGrepRequest, NamespaceGrepResult, NamespaceInclude, - NamespaceIndexValue, NamespaceListPage, NamespacePredicate, NamespacePredicateOp, - NamespacePredicateValue, NamespaceQueryCatalog, NamespaceReadFormat, NamespaceReadOptions, - NamespaceReadPage, NamespaceRecordCount, NamespaceRecordType, NamespaceSchema, NamespaceSort, + NamespaceIndexField, NamespaceIndexRegistration, NamespaceIndexRow, NamespaceIndexValue, + NamespaceListPage, NamespacePredicate, NamespacePredicateOp, NamespacePredicateValue, + NamespaceQueryCatalog, NamespaceReadFormat, NamespaceReadOptions, NamespaceReadPage, + NamespaceRecordCount, NamespaceRecordType, NamespaceSchema, NamespaceSort, NamespaceSortDirection, NamespaceSortField, PreparedArtifact, RecordCountProvenance, SubtreeDelta, SubtreeDeltaKind, UpdateAttr, XattrSetMode, }; + use nokv_object::{ObjectKey, ObjectReadBlock, StagedObject, StagedObjectSet}; + use nokv_protocol::{ MetadataProtocolError, MetadataRpcEnvelope, MetadataRpcResult, WireAdvisoryLock, WireBodyReadPlan, WireDentryWithAttr, WireMetadataError, WireNamespaceAggregateGroup, @@ -24,14 +27,15 @@ use nokv_protocol::{ WireNamespaceFieldSource, WireNamespaceFieldSourceKind, WireNamespaceFieldValue, WireNamespaceFilterCapability, WireNamespaceFindField, WireNamespaceFindRequest, WireNamespaceFindResult, WireNamespaceGrepMatch, WireNamespaceGrepRequest, - WireNamespaceGrepResult, WireNamespaceInclude, WireNamespaceIndexValue, WireNamespaceListPage, - WireNamespacePredicate, WireNamespacePredicateOp, WireNamespacePredicateValue, - WireNamespaceQueryCatalog, WireNamespaceReadFormat, WireNamespaceReadItem, - WireNamespaceReadOptions, WireNamespaceReadPage, WireNamespaceRecordCount, - WireNamespaceRecordType, WireNamespaceSchema, WireNamespaceSort, WireNamespaceSortDirection, - WireNamespaceSortField, WireObjectReadBlock, WireOpenPathReadPlan, WirePathMetadata, - WirePreparedArtifact, WireReadLease, WireRecordCountProvenance, WireStagedObjectSet, - WireSubtreeDelta, WireSubtreeDeltaKind, WireUpdateAttr, WireXattrSetMode, + WireNamespaceGrepResult, WireNamespaceInclude, WireNamespaceIndexField, + WireNamespaceIndexRegistration, WireNamespaceIndexRow, WireNamespaceIndexValue, + WireNamespaceListPage, WireNamespacePredicate, WireNamespacePredicateOp, + WireNamespacePredicateValue, WireNamespaceQueryCatalog, WireNamespaceReadFormat, + WireNamespaceReadItem, WireNamespaceReadOptions, WireNamespaceReadPage, + WireNamespaceRecordCount, WireNamespaceRecordType, WireNamespaceSchema, WireNamespaceSort, + WireNamespaceSortDirection, WireNamespaceSortField, WireObjectReadBlock, WireOpenPathReadPlan, + WirePathMetadata, WirePreparedArtifact, WireReadLease, WireRecordCountProvenance, + WireStagedObjectSet, WireSubtreeDelta, WireSubtreeDeltaKind, WireUpdateAttr, WireXattrSetMode, }; use nokv_types::{DentryName, InodeId, MountId}; @@ -792,6 +796,51 @@ pub(super) fn namespace_find_request( }) } +fn namespace_index_value(value: WireNamespaceIndexValue) -> NamespaceIndexValue { + NamespaceIndexValue { + field: namespace_find_field(value.field), + value: namespace_predicate_value(value.value), + } +} + +fn namespace_index_row(row: WireNamespaceIndexRow) -> NamespaceIndexRow { + NamespaceIndexRow { + path: row.path, + values: row.values.into_iter().map(namespace_index_value).collect(), + } +} + +fn namespace_index_field(field: WireNamespaceIndexField) -> NamespaceIndexField { + NamespaceIndexField { + field: namespace_find_field(field.field), + operators: field + .operators + .into_iter() + .map(namespace_predicate_op) + .collect(), + sortable: field.sortable, + facetable: field.facetable, + } +} + +pub(super) fn namespace_index_registration( + registration: WireNamespaceIndexRegistration, +) -> NamespaceIndexRegistration { + NamespaceIndexRegistration { + path: registration.path, + fields: registration + .fields + .into_iter() + .map(namespace_index_field) + .collect(), + rows: registration + .rows + .into_iter() + .map(namespace_index_row) + .collect(), + } +} + pub(super) fn namespace_aggregate_request( request: WireNamespaceAggregateRequest, ) -> Result {