diff --git a/store/postgres/src/deployment_store.rs b/store/postgres/src/deployment_store.rs index d764d451ad6..06f5c3155b1 100644 --- a/store/postgres/src/deployment_store.rs +++ b/store/postgres/src/deployment_store.rs @@ -176,6 +176,9 @@ impl DeploymentStore { DeploymentStore(Arc::new(store)) } + /// `index_def` is the index list of the deployment this one is copied or + /// grafted from. When it is given, the destination's schema is created + /// with the source's indexes rather than the schema's default set. pub(crate) async fn create_deployment( &self, schema: &InputSchema, @@ -183,6 +186,7 @@ impl DeploymentStore { site: Arc, replace: bool, on_sync: OnSync, + index_def: Option<&IndexList>, ) -> Result<(), StoreError> { let mut conn = self.pool.get_permitted().await?; conn.transaction::<_, StoreError, _>(async |conn| { @@ -210,11 +214,12 @@ impl DeploymentStore { let query = format!("create schema {}", &site.namespace); conn.batch_execute(&query).await?; - let _ = Layout::create_relational_schema( + let _ = Layout::create_relational_schema_with_source_indexes( conn, site.clone(), schema, entities_with_causality_region.into_iter().collect(), + index_def, ) .await?; diff --git a/store/postgres/src/relational.rs b/store/postgres/src/relational.rs index 4059e736635..836a403f4d3 100644 --- a/store/postgres/src/relational.rs +++ b/store/postgres/src/relational.rs @@ -56,6 +56,7 @@ use std::str::FromStr; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; +use crate::relational::index::IndexList; use crate::relational::value::{FromOidRow, OidRow}; use crate::relational_queries::{ ConflictingEntitiesData, ConflictingEntitiesQuery, EntityDataExt, FindChangesQuery, @@ -381,12 +382,33 @@ impl Layout { site: Arc, schema: &InputSchema, entities_with_causality_region: BTreeSet, + ) -> Result { + Self::create_relational_schema_with_source_indexes( + conn, + site, + schema, + entities_with_causality_region, + None, + ) + .await + } + + /// Like `create_relational_schema`, but the indexes that are not postponed + /// are taken from `source_indexes`. A copy or graft passes the source + /// deployment's index list so the destination ends up with the indexes the + /// source actually carries, not the schema's default set. + pub async fn create_relational_schema_with_source_indexes( + conn: &mut AsyncPgConnection, + site: Arc, + schema: &InputSchema, + entities_with_causality_region: BTreeSet, + source_indexes: Option<&IndexList>, ) -> Result { let catalog = Catalog::for_creation(conn, site.cheap_clone(), entities_with_causality_region).await?; let layout = Self::new(site, schema, catalog)?; let sql = layout - .as_ddl() + .as_ddl_with_source_indexes(source_indexes) .map_err(|_| StoreError::Unknown(anyhow!("failed to generate DDL for layout")))?; conn.batch_execute(&sql).await?; Ok(layout) diff --git a/store/postgres/src/relational/ddl.rs b/store/postgres/src/relational/ddl.rs index c1179b5b1a1..35b8c38550c 100644 --- a/store/postgres/src/relational/ddl.rs +++ b/store/postgres/src/relational/ddl.rs @@ -9,7 +9,10 @@ use crate::relational::{ BLOCK_COLUMN, BLOCK_RANGE_COLUMN, BYTE_ARRAY_PREFIX_SIZE, ColumnType, STRING_PREFIX_SIZE, VID_COLUMN, }; -use crate::{block_range::CAUSALITY_REGION_COLUMN, relational::index::Cond}; +use crate::{ + block_range::CAUSALITY_REGION_COLUMN, + relational::index::{Cond, IndexList}, +}; use super::{ Column, Layout, SqlName, Table, @@ -30,6 +33,17 @@ impl Layout { /// See the unit tests at the end of this file for the actual DDL that /// gets generated pub fn as_ddl(&self) -> Result { + self.as_ddl_with_source_indexes(None) + } + + /// Like `as_ddl`, but takes the indexes that are not postponed from + /// `source_indexes` instead of from this layout. Callers that create the + /// schema for a copy or graft pass the source deployment's index list so + /// the destination ends up with the indexes the source actually has. + pub fn as_ddl_with_source_indexes( + &self, + source_indexes: Option<&IndexList>, + ) -> Result { let mut out = String::new(); // Output enums first so table definitions can reference them @@ -42,7 +56,7 @@ impl Layout { // Output 'create table' statements for all tables let creat = self.index_creator(false, false); for table in tables { - table.as_ddl(&self.input_schema, &creat, &mut out)?; + table.as_ddl(&self.input_schema, &creat, source_indexes, &mut out)?; } Ok(out) @@ -410,14 +424,32 @@ impl Table { /// /// See the unit tests at the end of this file for the actual DDL that /// gets generated + /// `as_ddl` writes the table and the indexes that do not wait for the data + /// to be loaded. + /// + /// `source_indexes` is the index list of the deployment this one is copied + /// or grafted from. When it is given, the attribute indexes are taken from + /// the source instead of the layout, so that a copy carries the source's + /// index set rather than the schema's default one. The time-travel and + /// aggregate indexes are always created from the layout: they are part of + /// the schema every deployment needs, not something an operator drops. pub(crate) fn as_ddl( &self, schema: &InputSchema, creat: &IndexCreator, + source_indexes: Option<&IndexList>, out: &mut String, ) -> fmt::Result { self.create_table(out)?; - for idx in self.indexes(schema)? { + + let mut indexes = self.time_travel_indexes(); + match source_indexes { + Some(source) => indexes.extend(source.indexes_for_copy_table(self).cloned()), + None => self.add_attribute_indexes(&mut indexes), + } + self.add_aggregate_indexes(schema, &mut indexes)?; + + for idx in indexes { if !idx.to_postpone() { writeln!(out, "{};", creat.to_sql(&idx)?)?; } diff --git a/store/postgres/src/relational/ddl_tests.rs b/store/postgres/src/relational/ddl_tests.rs index f32aa46d97e..9165910c529 100644 --- a/store/postgres/src/relational/ddl_tests.rs +++ b/store/postgres/src/relational/ddl_tests.rs @@ -3,6 +3,7 @@ use pretty_assertions::assert_eq; use super::*; +use crate::relational::index::CreateIndex; use crate::{deployment_store::generate_index_creation_sql, layout_for_tests::make_dummy_site}; const ID_TYPE: ColumnType = ColumnType::String; @@ -888,3 +889,63 @@ type Data @entity(immutable: true) { block: Block! } "#; + +/// The copy and graft paths create the destination from the source's index +/// list, so a deployment whose indexes were dropped before copying does not +/// come back with the schema's default set. `as_ddl` on its own still writes +/// the layout's indexes, which is what a fresh deployment needs. +#[test] +fn source_indexes_replace_the_layout_indexes_on_creation() { + let layout = test_layout(THING_GQL); + let namespace = layout.site.namespace.as_str().to_string(); + + // `thing` carries a gist attribute index in this schema. A source that + // does not have it must not get it on the copy, even though the layout + // describes it. + let gist_on_thing = + format!("create index attr_0_1_thing_big_thing on \"{namespace}\".\"thing\" using gist"); + let layout_ddl = layout.as_ddl().expect("failed to generate DDL"); + assert!( + layout_ddl.contains(&gist_on_thing), + "the layout is expected to describe a gist index for `thing`:\n{layout_ddl}" + ); + + let from_empty = layout + .as_ddl_with_source_indexes(Some(&IndexList::new(HashMap::new()))) + .expect("failed to generate DDL"); + assert!( + !from_empty.contains(&gist_on_thing), + "a source without that index must not produce it:\n{from_empty}" + ); + + // The time-travel indexes are not an operator's to drop: they describe + // how history is stored, so they are written regardless of the source. + for always in ["brin_thing", "thing_block_range_closed"] { + assert!( + from_empty.contains(always), + "`{always}` is part of the schema and must still be created:\n{from_empty}" + ); + } + + // An attribute index the source does carry is kept. This is the other + // direction of the same check: reading the source must not silently drop + // what it has. + let table = layout.table(&SqlName::from("Thing")).unwrap(); + let kept = table + .indexes(&layout.input_schema) + .unwrap() + .into_iter() + .find(|idx| { + matches!(idx, CreateIndex::Parsed { name, .. } + if *name == "attr_0_1_thing_big_thing") + }) + .expect("the schema is expected to describe the gist index"); + let source = IndexList::new(HashMap::from([(table.name.to_string(), vec![kept])])); + let from_source = layout + .as_ddl_with_source_indexes(Some(&source)) + .expect("failed to generate DDL"); + assert!( + from_source.contains(&gist_on_thing), + "an index the source carries must be created:\n{from_source}" + ); +} diff --git a/store/postgres/src/relational/index.rs b/store/postgres/src/relational/index.rs index dc66a04f9a7..d98fc1607c4 100644 --- a/store/postgres/src/relational/index.rs +++ b/store/postgres/src/relational/index.rs @@ -946,6 +946,28 @@ impl IndexList { && !(ci.is_id() && table.immutable) }) } + + /// The indexes to create up front when a deployment is copied or grafted + /// from this one. + /// + /// Unlike `indexes_for_table`, this keeps the attribute indexes that + /// `indexes_for_table` drops. Those are dropped there because the layout + /// creates them anyway, which does not hold when the attribute indexes are + /// taken from the source: a source that carries `attr_N_0__id` + /// must keep it on the copy. The indexes the layout always creates + /// (`brin_*` and the closed-range ones) are dropped here for the same + /// reason they are dropped there. + pub fn indexes_for_copy_table<'a>( + &'a self, + table: &'a Table, + ) -> impl Iterator { + static EMPTY: Vec = vec![]; + let indexes = self.indexes.get(table.name.as_str()).unwrap_or(&EMPTY); + + indexes + .iter() + .filter(move |ci| ci.fields_exist_in_dest(table) && !ci.is_default_non_attr_index()) + } } #[cfg(test)] diff --git a/store/postgres/src/relational/prune.rs b/store/postgres/src/relational/prune.rs index 2bfb7e6c9c9..29446bf2aad 100644 --- a/store/postgres/src/relational/prune.rs +++ b/store/postgres/src/relational/prune.rs @@ -57,7 +57,7 @@ impl TablePair { // In case of pruning we don't do delayed creation of indexes, // as the asumption is that there is not that much data inserted. let creat = src_layout.index_creator(false, false); - dst.as_ddl(&src_layout.input_schema, &creat, &mut query)?; + dst.as_ddl(&src_layout.input_schema, &creat, None, &mut query)?; } conn.batch_execute(&query).await?; diff --git a/store/postgres/src/subgraph_store.rs b/store/postgres/src/subgraph_store.rs index b4435cae0cb..d47a9be390e 100644 --- a/store/postgres/src/subgraph_store.rs +++ b/store/postgres/src/subgraph_store.rs @@ -416,7 +416,14 @@ impl SubgraphStore { // Create the actual databases schema and metadata entries deployment_store - .create_deployment(schema, deployment, site.clone(), replace, OnSync::None) + .create_deployment( + schema, + deployment, + site.clone(), + replace, + OnSync::None, + None, + ) .await?; // FIXME: This simultaneously holds a `primary_conn` and a shard connection, which can @@ -900,6 +907,11 @@ impl Inner { .get(&shard) .ok_or_else(|| StoreError::UnknownShard(shard.to_string()))?; + // The destination gets the source's index set, not the schema's + // default one: an operator who dropped indexes from the source to + // save space should not get them all back on every copy. + let src_indexes = src_store.load_indexes(src.cheap_clone()).await?; + deployment_store .create_deployment( &src_layout.input_schema, @@ -907,6 +919,7 @@ impl Inner { dst.clone(), false, on_sync, + Some(&src_indexes), ) .await?;