Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 1 addition & 93 deletions crates/graphforge-api/src/composite_publish.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1011,99 +1011,7 @@ fn require_capabilities(
Ok(())
}

/// Cumulative probe I/O and serial decoder peaks. The identity containers are
/// reported separately: decoder counters are not a bound on process memory.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct EdgeOwnerProbeWork {
candidate_routes: usize,
target_memberships: usize,
resolved_targets: usize,
route_name_bytes: usize,
physical_bytes: u64,
authentication_bytes: u64,
authenticated_snapshot_bytes: u64,
authenticated_snapshot_peak_bytes: u64,
physical_rows: u64,
physical_blocks: u64,
fragments_considered: u64,
row_groups_considered: u64,
row_groups_selected: u64,
decoder_peak_bytes: u64,
decoder_page_reservation_bytes: u64,
}

/// Construction and ordinary mutation have different exploratory property
/// owners. Probe only the candidate routes for requested identities, once per
/// route, using authenticated row presence (including a newest tombstone).
fn resolve_existing_edge_property_owners(
inventory: &graphforge_storage::AuthenticatedPropertyInventory,
owners: &mut BTreeMap<Uuid, String>,
) -> Result<EdgeOwnerProbeWork, GfError> {
use graphforge_storage::PropertyRouteKind;
let mut candidates = BTreeMap::<String, BTreeSet<[u8; 16]>>::new();
for (uuid, route) in owners.iter() {
for candidate in [route.as_str(), "_exploratory"] {
if inventory
.route_schema(PropertyRouteKind::Edge, candidate)
.is_some()
{
if let Some(targets) = candidates.get_mut(candidate) {
targets.insert(uuid.into_bytes());
} else {
candidates.insert(candidate.to_owned(), BTreeSet::from([uuid.into_bytes()]));
}
}
}
}
let mut resolved = BTreeMap::new();
let mut total = EdgeOwnerProbeWork {
candidate_routes: candidates.len(),
target_memberships: candidates.values().map(BTreeSet::len).sum(),
route_name_bytes: candidates.keys().map(String::len).sum(),
..EdgeOwnerProbeWork::default()
};
for (route, targets) in &candidates {
let (present, work) =
graphforge_storage::read_authenticated_property_presence_for_inventory(
inventory,
PropertyRouteKind::Edge,
route,
targets,
)?;
total.physical_bytes += work.physical_bytes;
total.authentication_bytes += work.authentication_bytes;
total.authenticated_snapshot_bytes += work.authenticated_snapshot_bytes;
total.authenticated_snapshot_peak_bytes = total
.authenticated_snapshot_peak_bytes
.max(work.authenticated_snapshot_peak_bytes);
total.physical_rows += work.physical_rows;
total.physical_blocks += work.physical_blocks;
total.fragments_considered += work.fragments_considered;
total.row_groups_considered += work.row_groups_considered;
total.row_groups_selected += work.row_groups_selected;
total.decoder_peak_bytes = total.decoder_peak_bytes.max(work.decoder_peak_bytes);
total.decoder_page_reservation_bytes = total
.decoder_page_reservation_bytes
.max(work.decoder_page_reservation_bytes);
for uuid in present {
if resolved
.insert(Uuid::from_bytes(uuid), route.as_str())
.is_some()
{
return Err(GfError::Storage(
"composite edge property owner is ambiguous".into(),
));
}
}
}
// No property row means ordinary writer ownership, already derived from
// topology. Same-request creations are added by the caller afterwards.
total.resolved_targets = resolved.len();
for (uuid, route) in resolved {
owners.insert(uuid, route.to_owned());
}
Ok(total)
}
use graphforge_storage::resolve_existing_edge_property_owners;

#[allow(clippy::too_many_lines)]
fn build_validation_snapshot(
Expand Down
Loading