perf(ingester): remove namespace lookup query
Now DML operations contain the namespace ID, the ingester has all
necessary data to initialise the NamespaceData buffer node without
having to query the catalog.
//! Shard level data buffer structures.
-use std::{collections::HashMap, sync::Arc};
+use std::sync::Arc;
use data_types::{NamespaceId, ShardId, ShardIndex};
use dml::DmlOperation;
use iox_catalog::interface::Catalog;
use metric::U64Counter;
use parking_lot::RwLock;
-use snafu::{OptionExt, ResultExt};
use write_summary::ShardProgress;
use super::{
@@ -15,15 +14,15 @@ use super::{
partition::resolver::PartitionProvider,
DmlApplyAction,
};
-use crate::lifecycle::LifecycleHandle;
+use crate::{arcmap::ArcMap, lifecycle::LifecycleHandle};
/// A double-referenced map where [`NamespaceData`] can be looked up by name, or
/// ID.
#[derive(Debug, Default)]
struct DoubleRef {
// TODO(4880): this can be removed when IDs are sent over the wire.
- by_name: HashMap<NamespaceName, Arc<NamespaceData>>,
- by_id: HashMap<NamespaceId, Arc<NamespaceData>>,
+ by_name: ArcMap<NamespaceName, NamespaceData>,
+ by_id: ArcMap<NamespaceId, NamespaceData>,
}
impl DoubleRef {
@@ -31,17 +30,17 @@ impl DoubleRef {
let id = ns.namespace_id();
let ns = Arc::new(ns);
- self.by_name.insert(name, Arc::clone(&ns));
- self.by_id.insert(id, Arc::clone(&ns));
+ self.by_name.insert(&name, Arc::clone(&ns));
+ self.by_id.insert(&id, Arc::clone(&ns));
ns
}
fn by_name(&self, name: &NamespaceName) -> Option<Arc<NamespaceData>> {
- self.by_name.get(name).map(Arc::clone)
+ self.by_name.get(name)
}
fn by_id(&self, id: NamespaceId) -> Option<Arc<NamespaceData>> {
- self.by_id.get(&id).map(Arc::clone)
+ self.by_id.get(&id)
}
}
@@ -104,7 +103,7 @@ impl ShardData {
let namespace_data = match self.namespace(&NamespaceName::from(dml_operation.namespace())) {
Some(d) => d,
None => {
- self.insert_namespace(dml_operation.namespace(), &**catalog)
+ self.insert_namespace(dml_operation.namespace(), dml_operation.namespace_id())
.await?
}
};
@@ -135,17 +134,9 @@ impl ShardData {
async fn insert_namespace(
&self,
namespace: &str,
- catalog: &dyn Catalog,
+ namespace_id: NamespaceId,
) -> Result<Arc<NamespaceData>, super::Error> {
- let mut repos = catalog.repositories().await;
-
let ns_name = NamespaceName::from(namespace);
- let namespace = repos
- .namespaces()
- .get_by_name(namespace)
- .await
- .context(super::CatalogSnafu)?
- .context(super::NamespaceNotFoundSnafu { namespace })?;
let mut n = self.namespaces.write();
@@ -158,7 +149,7 @@ impl ShardData {
n.insert(
ns_name.clone(),
NamespaceData::new(
- namespace.id,
+ namespace_id,
ns_name,
self.shard_id,
Arc::clone(&self.partition_provider),
@@ -171,13 +162,7 @@ impl ShardData {
/// Return the progress of this shard
pub(super) async fn progress(&self) -> ShardProgress {
- let namespaces: Vec<_> = self
- .namespaces
- .read()
- .by_id
- .values()
- .map(Arc::clone)
- .collect();
+ let namespaces: Vec<_> = self.namespaces.read().by_id.values();
let mut progress = ShardProgress::new();
