refactor: remove Arc wrap from ReadThroughCache

This Arc was unnecessary in most uses.

router/benches/namespace_schema_cache.rs

@@ -33,7 +33,7 @@ fn init_ns_cache(
     rt.spawn(actor.run());
 
     let cache = MerkleTree::new(cache, handle);
-    let cache = Arc::new(ReadThroughCache::new(cache, Arc::clone(&catalog)));
+    let cache = ReadThroughCache::new(cache, Arc::clone(&catalog));
 
     for (name, schema) in initial_schema {
         cache.put_schema(name, schema);

router/src/dml_handlers/schema_validation.rs

@@ -541,13 +541,13 @@ mod tests {
         // Table exists and is over the column limit because of the race condition,
         {
             // Make two schema validator instances each with their own cache
-            let cache1 = setup_test_cache(&catalog);
+            let cache1 = Arc::new(setup_test_cache(&catalog));
             let handler1 = SchemaValidator::new(
                 catalog.catalog(),
                 Arc::clone(&cache1),
                 &catalog.metric_registry,
             );
-            let cache2 = setup_test_cache(&catalog);
+            let cache2 = Arc::new(setup_test_cache(&catalog));
             let handler2 = SchemaValidator::new(
                 catalog.catalog(),
                 Arc::clone(&cache2),
@@ -753,11 +753,8 @@ mod tests {
         (catalog, namespace)
     }
 
-    fn setup_test_cache(catalog: &TestCatalog) -> Arc<ReadThroughCache<MemoryNamespaceCache>> {
-        Arc::new(ReadThroughCache::new(
-            MemoryNamespaceCache::default(),
-            catalog.catalog(),
-        ))
+    fn setup_test_cache(catalog: &TestCatalog) -> ReadThroughCache<MemoryNamespaceCache> {
+        ReadThroughCache::new(MemoryNamespaceCache::default(), catalog.catalog())
     }
 
     async fn assert_cache<C>(handler: &SchemaValidator<C>, table: &str, col: &str, want: ColumnType)
@@ -789,7 +786,7 @@ mod tests {
         let want_id = namespace.create_table("bananas").await.table.id;
 
         let metrics = Arc::new(metric::Registry::default());
-        let cache = setup_test_cache(&catalog);
+        let cache = Arc::new(setup_test_cache(&catalog));
         let handler = SchemaValidator::new(catalog.catalog(), Arc::clone(&cache), &metrics);
 
         let writes = lp_to_writes("bananas,tag1=A,tag2=B val=42i 123456");

router/src/namespace_cache/read_through_cache.rs

@@ -39,7 +39,7 @@ impl<T> ReadThroughCache<T> {
 }
 
 #[async_trait]
-impl<T> NamespaceCache for Arc<ReadThroughCache<T>>
+impl<T> NamespaceCache for ReadThroughCache<T>
 where
     T: NamespaceCache<ReadError = CacheMissErr>,
 {
@@ -112,7 +112,7 @@ mod tests {
         let metrics = Arc::new(metric::Registry::new());
         let catalog = Arc::new(MemCatalog::new(metrics));
 
-        let cache = Arc::new(ReadThroughCache::new(inner, catalog));
+        let cache = ReadThroughCache::new(inner, catalog);
 
         // Pre-condition: Namespace not in cache or catalog.
         assert_matches!(cache.get_schema(&ns).await, Err(_));
@@ -148,7 +148,7 @@ mod tests {
         let metrics = Arc::new(metric::Registry::new());
         let catalog: Arc<dyn Catalog> = Arc::new(MemCatalog::new(metrics));
 
-        let cache = Arc::new(ReadThroughCache::new(inner, Arc::clone(&catalog)));
+        let cache = ReadThroughCache::new(inner, Arc::clone(&catalog));
 
         // Pre-condition: Namespace not in cache or catalog.
         assert_matches!(cache.get_schema(&ns).await, Err(_));

router/src/namespace_resolver/ns_autocreation.rs

@@ -159,10 +159,7 @@ mod tests {
         let catalog: Arc<dyn Catalog> = Arc::new(MemCatalog::new(metrics));
 
         // Prep the cache before the test to cause a hit
-        let cache = Arc::new(ReadThroughCache::new(
-            MemoryNamespaceCache::default(),
-            Arc::clone(&catalog),
-        ));
+        let cache = ReadThroughCache::new(MemoryNamespaceCache::default(), Arc::clone(&catalog));
         cache.put_schema(
             ns.clone(),
             NamespaceSchema {
@@ -210,10 +207,7 @@ mod tests {
         let metrics = Arc::new(metric::Registry::new());
         let catalog: Arc<dyn Catalog> = Arc::new(MemCatalog::new(metrics));
 
-        let cache = Arc::new(ReadThroughCache::new(
-            MemoryNamespaceCache::default(),
-            Arc::clone(&catalog),
-        ));
+        let cache = ReadThroughCache::new(MemoryNamespaceCache::default(), Arc::clone(&catalog));
 
         let creator = NamespaceAutocreation::new(
             MockNamespaceResolver::default().with_mapping(ns.clone(), NamespaceId::new(1)),
@@ -258,10 +252,7 @@ mod tests {
 
         let metrics = Arc::new(metric::Registry::new());
         let catalog: Arc<dyn Catalog> = Arc::new(MemCatalog::new(metrics));
-        let cache = Arc::new(ReadThroughCache::new(
-            MemoryNamespaceCache::default(),
-            Arc::clone(&catalog),
-        ));
+        let cache = ReadThroughCache::new(MemoryNamespaceCache::default(), Arc::clone(&catalog));
 
         let creator = NamespaceAutocreation::new(
             MockNamespaceResolver::default(),