refactor: store reverse column ID map for cached tables (#6360)

querier/src/cache/namespace.rs

@@ -208,6 +208,7 @@ pub struct CachedTable {
     pub id: TableId,
     pub schema: Arc<Schema>,
     pub column_id_map: HashMap<ColumnId, Arc<str>>,
+    pub column_id_map_rev: HashMap<Arc<str>, ColumnId>,
     pub primary_key_column_ids: Vec<ColumnId>,
 }
 
@@ -221,6 +222,12 @@ impl CachedTable {
                 .iter()
                 .map(|(_id, name)| name.len())
                 .sum::<usize>()
+            + (self.column_id_map_rev.capacity() * size_of::<(Arc<str>, ColumnId)>())
+            + self
+                .column_id_map_rev
+                .iter()
+                .map(|(name, _id)| name.len())
+                .sum::<usize>()
             + (self.primary_key_column_ids.capacity() * size_of::<ColumnId>())
     }
 }
@@ -237,10 +244,12 @@ impl From<TableSchema> for CachedTable {
         let id = table.id;
         let schema: Arc<Schema> = Arc::new(table.try_into().expect("Catalog table schema broken"));
 
-        let column_id_map_rev: HashMap<Arc<str>, ColumnId> = column_id_map
+        let mut column_id_map_rev: HashMap<Arc<str>, ColumnId> = column_id_map
             .iter()
             .map(|(v, k)| (Arc::clone(k), *v))
             .collect();
+        column_id_map_rev.shrink_to_fit();
+
         let mut primary_key_column_ids: Vec<ColumnId> = schema
             .primary_key()
             .into_iter()
@@ -256,6 +265,7 @@ impl From<TableSchema> for CachedTable {
             id,
             schema,
             column_id_map,
+            column_id_map_rev,
             primary_key_column_ids,
         }
     }
@@ -372,6 +382,11 @@ mod tests {
                             (col112.column.id, Arc::from(col112.column.name.clone())),
                             (col113.column.id, Arc::from(col113.column.name.clone())),
                         ]),
+                        column_id_map_rev: HashMap::from([
+                            (Arc::from(col111.column.name.clone()), col111.column.id),
+                            (Arc::from(col112.column.name.clone()), col112.column.id),
+                            (Arc::from(col113.column.name.clone()), col113.column.id),
+                        ]),
                         primary_key_column_ids: vec![col112.column.id, col113.column.id],
                     }),
                 ),
@@ -391,6 +406,10 @@ mod tests {
                             (col121.column.id, Arc::from(col121.column.name.clone())),
                             (col122.column.id, Arc::from(col122.column.name.clone())),
                         ]),
+                        column_id_map_rev: HashMap::from([
+                            (Arc::from(col121.column.name.clone()), col121.column.id),
+                            (Arc::from(col122.column.name.clone()), col122.column.id),
+                        ]),
                         primary_key_column_ids: vec![col122.column.id],
                     }),
                 ),
@@ -419,6 +438,10 @@ mod tests {
                         col211.column.id,
                         Arc::from(col211.column.name.clone()),
                     )]),
+                    column_id_map_rev: HashMap::from([(
+                        Arc::from(col211.column.name.clone()),
+                        col211.column.id,
+                    )]),
                     primary_key_column_ids: vec![col211.column.id],
                 }),
             )]),

querier/src/cache/projected_schema.rs

@@ -159,7 +159,7 @@ impl ProjectedSchemaCache {
 #[cfg(test)]
 mod tests {
     use iox_time::SystemProvider;
-    use schema::builder::SchemaBuilder;
+    use schema::{builder::SchemaBuilder, TIME_COLUMN_NAME};
 
     use crate::cache::ram::test_util::test_ram_pool;
 
@@ -199,19 +199,20 @@ mod tests {
             (ColumnId::new(1), Arc::from("t1")),
             (ColumnId::new(2), Arc::from("t2")),
             (ColumnId::new(3), Arc::from("t3")),
-            (ColumnId::new(4), Arc::from("time")),
+            (ColumnId::new(4), Arc::from(TIME_COLUMN_NAME)),
         ]);
         let column_id_map_b = HashMap::from([
             (ColumnId::new(1), Arc::from("t1")),
             (ColumnId::new(2), Arc::from("t2")),
             (ColumnId::new(3), Arc::from("t3")),
             (ColumnId::new(5), Arc::from("t4")),
-            (ColumnId::new(4), Arc::from("time")),
+            (ColumnId::new(4), Arc::from(TIME_COLUMN_NAME)),
         ]);
         let table_1a = Arc::new(CachedTable {
             id: table_id_1,
             schema: Arc::clone(&table_schema_a),
             column_id_map: column_id_map_a.clone(),
+            column_id_map_rev: reverse_map(&column_id_map_a),
             primary_key_column_ids: vec![
                 ColumnId::new(1),
                 ColumnId::new(2),
@@ -223,6 +224,7 @@ mod tests {
             id: table_id_1,
             schema: Arc::clone(&table_schema_b),
             column_id_map: column_id_map_b.clone(),
+            column_id_map_rev: reverse_map(&column_id_map_b),
             primary_key_column_ids: vec![
                 ColumnId::new(1),
                 ColumnId::new(2),
@@ -234,6 +236,7 @@ mod tests {
             id: table_id_2,
             schema: Arc::clone(&table_schema_a),
             column_id_map: column_id_map_a.clone(),
+            column_id_map_rev: reverse_map(&column_id_map_a),
             primary_key_column_ids: vec![
                 ColumnId::new(1),
                 ColumnId::new(2),
@@ -316,4 +319,12 @@ mod tests {
             .await;
         assert!(Arc::ptr_eq(&projection_1, &projection_7));
     }
+
+    fn reverse_map<K, V>(map: &HashMap<K, V>) -> HashMap<V, K>
+    where
+        K: Clone,
+        V: Clone + std::hash::Hash + Eq,
+    {
+        map.iter().map(|(k, v)| (v.clone(), k.clone())).collect()
+    }
 }