diff --git a/store/postgres/src/catalog.rs b/store/postgres/src/catalog.rs index 0fbee8ff416..a4f733a9880 100644 --- a/store/postgres/src/catalog.rs +++ b/store/postgres/src/catalog.rs @@ -912,31 +912,37 @@ pub(crate) async fn index_creation_is_running( .map_err::(Into::into) } +/// The index definitions of a table, only the valid ones with `valid_only`. pub(crate) async fn indexes_for_table( conn: &mut AsyncPgConnection, schema_name: &str, table_name: &str, + valid_only: bool, ) -> Result, StoreError> { #[derive(Queryable, QueryableByName)] - struct IndexName { + struct IndexDef { #[diesel(sql_type = Text)] - #[diesel(column_name = indexdef)] def: String, } let query = " select - indexdef + pg_get_indexdef(i.indexrelid) as def from - pg_indexes + pg_index i + join pg_class t on t.oid = i.indrelid + join pg_class c on c.oid = i.indexrelid + join pg_namespace n on n.oid = t.relnamespace where - schemaname = $1 - and tablename = $2 - order by indexname"; + n.nspname = $1 + and t.relname = $2 + and (i.indisvalid or not $3) + order by c.relname"; let results = sql_query(query) .bind::(schema_name) .bind::(table_name) - .load::(conn) + .bind::(valid_only) + .load::(conn) .await .map_err::(Into::into)?; diff --git a/store/postgres/src/copy.rs b/store/postgres/src/copy.rs index d35e585a653..fd076ab0e1b 100644 --- a/store/postgres/src/copy.rs +++ b/store/postgres/src/copy.rs @@ -12,7 +12,6 @@ //! operation can resume after an interruption, for example, because //! `graph-node` was restarted while the copy was running. use std::{ - collections::HashSet, convert::TryFrom, future::Future, pin::Pin, @@ -103,6 +102,14 @@ pub enum Status { Cancelled, } +/// The source of a copy, as read when the copy starts. +pub(crate) struct CopySource { + /// The source's indexes. + pub(crate) list: IndexList, + /// Whether the source has built its postponed indexes. + pub(crate) postponed_indexes_created: bool, +} + struct CopyState { src: Arc, dst: Arc, @@ -299,45 +306,39 @@ impl CopyState { self.finished.iter().chain(self.unfinished.iter()) } - /// Create the indexes that were postponed at the start of the - /// copy/graft operation: first the ones that existed in the source - /// subgraph, then any indexes for new fields that don't exist in the - /// source. + /// Create the postponed indexes `dst` needs and give it the source's + /// `postponed_indexes_created` flag. async fn create_indexes( &self, conn: &mut AsyncPgConnection, - index_list: &IndexList, + source: &CopySource, ) -> Result<(), StoreError> { let creat = self.dst.index_creator(false, true); - let dst_nsp = self.dst.site.namespace.to_string(); - - // Create the indexes that existed in the source subgraph - for table in self.all_tables() { - let idxs = index_list - .indexes_for_table(&table.dst) - .filter(|idx| idx.to_postpone()) - .map(|idx| idx.with_nsp(dst_nsp.clone())) - .collect::, _>>()?; + for dst_table in self.dst.tables.values() { + // May already hold postponed indexes from an interrupted run or an older version's copy. + let existing = dst_table.valid_indexes(conn).await?; + let src = self + .src + .table_for_entity(&dst_table.object) + .ok() + .map(|src_table| { + let indexes = source + .list + .indexes + .get(src_table.name.as_str()) + .map(Vec::as_slice) + .unwrap_or(&[]); + (src_table.as_ref(), indexes) + }); + let idxs = + dst_table.postponed_indexes_to_create(&self.dst.input_schema, &existing, src)?; creat.execute_many(conn, &idxs).await?; } - // Create indexes for new fields that don't exist in the source - for table in self.all_tables() { - let src_columns: HashSet<&str> = - table.src.columns.iter().map(|c| c.name.as_str()).collect(); - let new_idxs: Vec<_> = table - .dst - .indexes(&self.dst.input_schema) - .map_err(|_| internal_error!("failed to generate indexes for copy"))? - .into_iter() - .filter(|idx| idx.to_postpone() && idx.references_column_not_in(&src_columns)) - .collect(); - creat.execute_many(conn, &new_idxs).await?; - } - + let created = source.postponed_indexes_created; conn.transaction(async |conn| { - deployment::set_postponed_indexes_created(conn, &self.dst.site).await + deployment::set_postponed_indexes_created(conn, &self.dst.site, created).await }) .await?; @@ -1191,7 +1192,7 @@ impl Connection { } } - async fn copy_data_internal(&mut self, index_list: IndexList) -> Result { + async fn copy_data_internal(&mut self, source: CopySource) -> Result { let src = self.src.clone(); let dst = self.dst.clone(); let target_block = self.target_block.clone(); @@ -1282,7 +1283,7 @@ impl Connection { } debug_assert!(self.conn.is_some()); - state.create_indexes(self.get_conn()?, &index_list).await?; + state.create_indexes(self.get_conn()?, &source).await?; self.copy_private_data_sources(&state).await?; @@ -1299,7 +1300,7 @@ impl Connection { /// block is guaranteed to not be subject to chain reorgs. All data up /// to and including `target_block` will be copied. /// - /// The parameter index_list is a list of indexes that exist on the `src`. + /// The parameter `source` describes the indexes of `src`. /// /// The copy logic makes heavy use of the fact that the `vid` and /// `block_range` of entity versions are related since for two entity @@ -1309,7 +1310,7 @@ impl Connection { /// lower(v1.block_range) => v2.vid > v1.vid` and we can therefore stop /// the copying of each table as soon as we hit `max_vid = max { v.vid | /// lower(v.block_range) <= target_block.number }`. - pub async fn copy_data(mut self, index_list: IndexList) -> Result { + pub async fn copy_data(mut self, source: CopySource) -> Result { // We require sole access to the destination site, and that we get a // consistent view of what has been copied so far. In general, that // is always true. It can happen though that this function runs when @@ -1329,7 +1330,7 @@ impl Connection { }; conn.lock(&self.logger, &dst_site).await?; - let res = self.copy_data_internal(index_list).await; + let res = self.copy_data_internal(source).await; match self.conn.as_mut() { None => { diff --git a/store/postgres/src/deployment.rs b/store/postgres/src/deployment.rs index 76cde305bcb..5ad5eb83ba0 100644 --- a/store/postgres/src/deployment.rs +++ b/store/postgres/src/deployment.rs @@ -854,17 +854,18 @@ pub async fn postponed_indexes_created( Ok(created) } -/// Mark the postponed (deferred) attribute indexes as created for this -/// deployment. Once set, we never recreate these indexes — even if an -/// external process removes them. +/// Record whether the postponed (deferred) attribute indexes have been +/// created for this deployment. Once set, we never recreate these indexes — +/// even if an external process removes them. pub async fn set_postponed_indexes_created( conn: &mut AsyncPgConnection, site: &Site, + created: bool, ) -> Result<(), StoreError> { use deployment as d; update(d::table.filter(d::id.eq(site.id))) - .set(d::postponed_indexes_created.eq(true)) + .set(d::postponed_indexes_created.eq(created)) .execute(conn) .await?; Ok(()) diff --git a/store/postgres/src/deployment_store.rs b/store/postgres/src/deployment_store.rs index d764d451ad6..ab589d95577 100644 --- a/store/postgres/src/deployment_store.rs +++ b/store/postgres/src/deployment_store.rs @@ -48,6 +48,7 @@ use graph::prelude::{ use graph::schema::{ApiSchema, EntityKey, EntityType, InputSchema}; use crate::block_range::{BLOCK_COLUMN, BLOCK_RANGE_COLUMN}; +use crate::copy::CopySource; use crate::deployment::{self, OnSync}; use crate::detail::ErrorDetail; use crate::dynds::DataSourcesTable; @@ -711,16 +712,58 @@ impl DeploymentStore { let table = resolve_table_name(&layout, &entity_name)?; let table_name = &table.name; let indexes = - catalog::indexes_for_table(&mut conn, schema_name.as_str(), table_name.as_str()) + catalog::indexes_for_table(&mut conn, schema_name.as_str(), table_name.as_str(), false) .await?; Ok(indexes.into_iter().map(CreateIndex::parse).collect()) } - pub(crate) async fn load_indexes(&self, site: Arc) -> Result { + /// Load `site` as the source of a copy. + pub(crate) async fn load_copy_source(&self, site: Arc) -> Result { let mut conn = self.pool.get_permitted().await?; + // Read the flag before the indexes, so that a set flag never comes + // with a list from the middle of the background build. + let postponed_indexes_created = + deployment::postponed_indexes_created(&mut conn, &site).await?; let layout = self.layout(&mut conn, site).await?; + let list = IndexList::load(&mut conn, &layout).await?; + Ok(CopySource { + list, + postponed_indexes_created, + }) + } + + pub(crate) async fn postponed_indexes_created(&self, site: &Site) -> Result { + let mut conn = self.pool.get_permitted().await?; + deployment::postponed_indexes_created(&mut conn, site).await + } - IndexList::load(&mut conn, &layout).await + // Only used for tests + #[cfg(debug_assertions)] + pub(crate) async fn set_postponed_indexes_created( + &self, + site: &Site, + created: bool, + ) -> Result<(), StoreError> { + let mut conn = self.pool.get_permitted().await?; + deployment::set_postponed_indexes_created(&mut conn, site, created).await + } + + // Only used for tests + #[cfg(debug_assertions)] + pub(crate) async fn create_btree_index( + &self, + site: &Site, + name: &str, + table: &str, + expr: &str, + ) -> Result<(), StoreError> { + let mut conn = self.pool.get_permitted().await?; + let query = format!( + "create index \"{}\" on \"{}\".\"{}\" using btree ({})", + name, site.namespace, table, expr + ); + sql_query(query).execute(&mut conn).await?; + Ok(()) } /// Drops an index for a given deployment, concurrently. @@ -1549,12 +1592,12 @@ impl DeploymentStore { &self, logger: &Logger, site: Arc, - graft_src: Option<(Arc, BlockPtr, SubgraphDeploymentEntity, IndexList)>, + graft_src: Option<(Arc, BlockPtr, SubgraphDeploymentEntity, CopySource)>, ) -> Result<(), StoreError> { let dst = self.find_layout(site.cheap_clone()).await?; // If `graft_src` is `Some`, then there is a pending graft. - if let Some((src, block, src_deployment, index_list)) = graft_src { + if let Some((src, block, src_deployment, source)) = graft_src { info!( logger, "Initializing graft by copying data from {} to {}", @@ -1585,7 +1628,7 @@ impl DeploymentStore { dst_manifest_idx_and_name, ) .await?; - let status = copy_conn.copy_data(index_list).await?; + let status = copy_conn.copy_data(source).await?; if status == crate::copy::Status::Cancelled { return Err(StoreError::Canceled); } @@ -1706,14 +1749,6 @@ impl DeploymentStore { catalog::index_creation_is_running(&mut conn, site.namespace.as_str()).await } - async fn postponed_indexes_created( - store: &DeploymentStore, - site: &Site, - ) -> Result { - let mut conn = store.pool.get_permitted().await?; - deployment::postponed_indexes_created(&mut conn, site).await - } - async fn wait_for_concurrent_index_creation( logger: &Logger, store: &DeploymentStore, @@ -1787,19 +1822,18 @@ impl DeploymentStore { // connection for the whole time. Instead, we get a new // connection for each index that we create. - if postponed_indexes_created(&store, &site).await? { + if store.postponed_indexes_created(&site).await? { return Ok(()); } for table in layout.tables.values() { - let indexes = table.indexes(&layout.input_schema).map_err(|e| { - StoreError::ConstraintViolation(format!("failed to generate indexes: {}", e)) - })?; + let existing = { + let mut conn = store.pool.get_permitted().await?; + table.valid_indexes(&mut conn).await? + }; + let indexes = + table.postponed_indexes_to_create(&layout.input_schema, &existing, None)?; for idx in indexes { - if !idx.to_postpone() { - continue; - } - wait_for_concurrent_index_creation(&logger, &store, &site).await?; create_index(&logger, &store, &layout, &site, &idx).await?; @@ -1812,7 +1846,7 @@ impl DeploymentStore { } let mut conn = store.pool.get_permitted().await?; - deployment::set_postponed_indexes_created(&mut conn, &site).await?; + deployment::set_postponed_indexes_created(&mut conn, &site, true).await?; Ok(()) } diff --git a/store/postgres/src/relational/index.rs b/store/postgres/src/relational/index.rs index dc66a04f9a7..ae399513788 100644 --- a/store/postgres/src/relational/index.rs +++ b/store/postgres/src/relational/index.rs @@ -12,6 +12,7 @@ use graph::prelude::{ BlockNumber, lazy_static, regex::{Captures, Regex}, }; +use graph::schema::InputSchema; use crate::block_range::{BLOCK_COLUMN, BLOCK_RANGE_COLUMN}; use crate::relational::{BYTE_ARRAY_PREFIX_SIZE, STRING_PREFIX_SIZE}; @@ -591,7 +592,7 @@ impl CreateIndex { self.is_attribute_index() || self.is_default_non_attr_index() } - fn is_same_index(&self, other: &CreateIndex) -> bool { + pub(crate) fn is_same_index(&self, other: &CreateIndex) -> bool { match (self, other) { (CreateIndex::Unknown { .. }, _) | (_, CreateIndex::Unknown { .. }) => false, ( @@ -673,75 +674,18 @@ impl CreateIndex { } } - /// Return `true` if any of the attribute columns (user-defined columns, - /// not infrastructure columns like `vid`, `block$`, etc.) referenced by - /// this index are NOT in `columns`. - pub fn references_column_not_in(&self, columns: &std::collections::HashSet<&str>) -> bool { + /// The user-defined columns this index covers. + pub fn attribute_columns(&self) -> Vec<&str> { match self { - CreateIndex::Unknown { .. } => false, - CreateIndex::Parsed { columns: exprs, .. } => exprs.iter().any(|expr| match expr { - Expr::Column(name) | Expr::Prefix(name, _) => !columns.contains(name.as_str()), - _ => false, - }), - } - } - - pub fn fields_exist_in_dest(&self, dest_table: &Table) -> bool { - fn column_exists(dest_table: &Table, column_name: &str) -> bool { - dest_table - .columns + CreateIndex::Unknown { .. } => vec![], + CreateIndex::Parsed { columns, .. } => columns .iter() - .any(|c| c.name.as_str() == column_name) + .filter_map(|expr| match expr { + Expr::Column(name) | Expr::Prefix(name, _) => Some(name.as_str()), + _ => None, + }) + .collect(), } - - match self { - CreateIndex::Unknown { defn: _ } => return true, - CreateIndex::Parsed { - columns: parsed_cols, - .. - } => { - for c in parsed_cols { - match c { - Expr::Column(column_name) => { - if !column_exists(dest_table, column_name) { - return false; - } - } - Expr::Prefix(column_name, _) => { - if !column_exists(dest_table, column_name) { - return false; - } - } - Expr::BlockRange | Expr::BlockRangeLower | Expr::BlockRangeUpper => { - if dest_table.immutable { - return false; - } - } - Expr::Vid => (), - Expr::Block => { - if !dest_table.immutable { - return false; - } - } - Expr::Unknown(expression) => { - if expression.contains("block_range") && dest_table.immutable { - return false; - } - if !dest_table - .columns - .iter() - .any(|c| expression.contains(c.name.as_str())) - && !expression.contains("block_range") - && !expression.contains("vid") - { - return false; - } - } - } - } - } - } - true } /// Generate a SQL statement that creates this index. If `concurrent` is @@ -898,6 +842,55 @@ impl Layout { } } +impl Table { + /// The valid indexes on this table. + pub(crate) async fn valid_indexes( + &self, + conn: &mut AsyncPgConnection, + ) -> Result, StoreError> { + let defs = + catalog::indexes_for_table(conn, self.nsp.as_str(), self.name.as_str(), true).await?; + Ok(defs.into_iter().map(CreateIndex::parse).collect()) + } + + /// The postponed indexes this table still needs. Leaves out an index if + /// `existing` has one of the same shape, or if the copy source `src` has + /// its columns but no postponed index on them. + pub(crate) fn postponed_indexes_to_create( + &self, + schema: &InputSchema, + existing: &[CreateIndex], + src: Option<(&Table, &[CreateIndex])>, + ) -> Result, StoreError> { + let indexes = self.indexes(schema).map_err(|e| { + StoreError::ConstraintViolation(format!("failed to generate indexes: {}", e)) + })?; + Ok(indexes + .into_iter() + .filter(|idx| idx.to_postpone()) + .filter(|idx| { + if existing + .iter() + .any(|e| e.to_postpone() && e.is_same_index(idx)) + { + return false; + } + let Some((src_table, src_indexes)) = src else { + return true; + }; + let columns = idx.attribute_columns(); + let src_has_columns = columns + .iter() + .all(|col| src_table.columns.iter().any(|c| c.name.as_str() == *col)); + !src_has_columns + || src_indexes + .iter() + .any(|s| s.to_postpone() && s.attribute_columns() == columns) + }) + .collect()) + } +} + #[derive(Debug)] pub struct IndexList { pub(crate) indexes: HashMap>, @@ -913,7 +906,7 @@ impl IndexList { let schema_name = layout.site.namespace.clone(); for table in layout.tables.values() { let indexes_from_table = - catalog::indexes_for_table(conn, schema_name.as_str(), table.name.as_str()) + catalog::indexes_for_table(conn, schema_name.as_str(), table.name.as_str(), false) .await? .into_iter() .map(CreateIndex::parse) @@ -922,30 +915,6 @@ impl IndexList { } Ok(Self::new(indexes)) } - - /// Return all indexes for `table` from this list. Only indexes that are - /// for columns that actually exist on `table` are returned. In - /// addition, indexes that are always created when a deployment is - /// created independent of the configuration are also not returned. - pub fn indexes_for_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| { - // First we check if the fields do exist in the destination subgraph. - // In case of grafting that is not given. - ci.fields_exist_in_dest(table) - // Then we check if the index is one of the default indexes not based on - // the attributes. Those will be created anyway and we should skip them. - && !ci.is_default_non_attr_index() - // Then ID based indexes in the immutable tables are also created initially - // and should be skipped. - && !(ci.is_id() && table.immutable) - }) - } } #[cfg(test)] @@ -1310,42 +1279,137 @@ mod tests { parse_many(sqls, exp); } - #[test] - fn fields_exist_in_dest_out_of_order() { - let gql = "type Thing @entity { - id: Bytes! - early: String! - mid: String! - late: String! - }"; - let layout = test_layout(gql); - let table = layout + fn thing_table(layout: &Layout) -> &Table { + layout .table_for_entity(&layout.input_schema.entity_type("Thing").unwrap()) - .unwrap(); + .unwrap() + } - // Index references columns in reverse order vs the table definition. - // Table columns: id, early, mid, late - // Index columns: late, mid, early - // The consuming iterator finds 'late', advances past 'mid' and 'early', - // then can't find 'mid' because the iterator is already past it. - let index = CreateIndex::Parsed { - unique: false, - name: "test_reverse".to_string(), - nsp: "sgd0815".to_string(), - table: "thing".to_string(), - method: Method::BTree, - columns: vec![ - Expr::Column("late".to_string()), - Expr::Column("mid".to_string()), - Expr::Column("early".to_string()), - ], - cond: None, - with: None, - }; + #[test] + fn postponed_shapes_round_trip() { + let layout = test_layout( + "type Thing @entity(immutable: true) { + id: ID! + name: String! + data: Bytes! + age: Int! + owner: Owner! + } + type Owner @entity { + id: ID! + }", + ); + let generated: Vec<_> = thing_table(&layout) + .indexes(&layout.input_schema) + .unwrap() + .into_iter() + .filter(|idx| idx.to_postpone()) + .collect(); + + // As Postgres prints them. + let printed = [ + ( + r#"CREATE INDEX attr_9_1_thing_name ON sgd0815.thing USING btree ("left"(name, 256))"#, + "name", + ), + ( + r#"CREATE INDEX attr_9_2_thing_data ON sgd0815.thing USING btree ("substring"(data, 1, 64))"#, + "data", + ), + ( + r#"CREATE INDEX attr_9_3_thing_age ON sgd0815.thing USING btree (age)"#, + "age", + ), + ( + r#"CREATE INDEX attr_9_4_thing_owner ON sgd0815.thing USING btree (owner, "block$")"#, + "owner", + ), + ]; + assert_eq!(printed.len(), generated.len(), "{generated:?}"); + for (defn, column) in printed { + let idx = CreateIndex::parse(defn.to_string()); + assert!(idx.to_postpone(), "{defn}"); + assert_eq!(vec![column], idx.attribute_columns(), "{defn}"); + assert!(generated.iter().any(|g| idx.is_same_index(g)), "{defn}"); + } + } - assert!( - index.fields_exist_in_dest(table), - "index columns exist in table regardless of order" + const POSTPONED_GQL: &str = "type Thing @entity { + id: ID! + name: String! + email: String! + age: Int! + data: Bytes! + }"; + + fn sorted_columns(indexes: &[CreateIndex]) -> Vec { + let mut columns: Vec<_> = indexes + .iter() + .map(|idx| idx.attribute_columns().join(",")) + .collect(); + columns.sort(); + columns + } + + #[test] + fn postponed_indexes_to_create_skips_same_shape() { + let layout = test_layout(POSTPONED_GQL); + let table = thing_table(&layout); + let schema = &layout.input_schema; + + let all = table + .postponed_indexes_to_create(schema, &[], None) + .unwrap(); + assert_eq!(vec!["age", "data", "email", "name"], sorted_columns(&all)); + + // Only `email` is covered: the `data` index has another shape, and + // manual indexes don't count. + let existing = [ + r#"CREATE INDEX attr_9_9_thing_email ON sgd0815.thing USING btree ("left"(email, 256))"#, + r#"CREATE INDEX attr_9_9_thing_data ON sgd0815.thing USING btree (data)"#, + r#"CREATE INDEX manual_thing_age ON sgd0815.thing USING btree (age)"#, + ] + .map(|defn| CreateIndex::parse(defn.to_string())); + let missing = table + .postponed_indexes_to_create(schema, &existing, None) + .unwrap(); + assert_eq!(vec!["age", "data", "name"], sorted_columns(&missing)); + } + + #[test] + fn postponed_indexes_to_create_from_source() { + // The source indexes `email` with a different shape and doesn't + // index `name`. + let src_layout = test_layout( + "type Thing @entity { + id: ID! + email: String! + name: String! + }", + ); + let src_indexes = [CreateIndex::parse( + r#"CREATE INDEX attr_0_1_thing_email ON sgd0815.thing USING btree (email)"#.to_string(), + )]; + let layout = test_layout(POSTPONED_GQL); + let table = thing_table(&layout); + + let to_create = table + .postponed_indexes_to_create( + &layout.input_schema, + &[], + Some((thing_table(&src_layout), &src_indexes)), + ) + .unwrap(); + assert_eq!(vec!["age", "data", "email"], sorted_columns(&to_create)); + let email = to_create + .iter() + .find(|idx| idx.attribute_columns() == ["email"]) + .unwrap(); + let expected = CreateIndex::parse( + r#"CREATE INDEX attr_0_2_thing_email ON sgd0815.thing USING btree ("left"(email, 256))"# + .to_string(), ); + assert_eq!(Some("attr_0_2_thing_email"), email.name()); + assert!(expected.is_same_index(email)); } } diff --git a/store/postgres/src/subgraph_store.rs b/store/postgres/src/subgraph_store.rs index b4435cae0cb..77ef932bbbd 100644 --- a/store/postgres/src/subgraph_store.rs +++ b/store/postgres/src/subgraph_store.rs @@ -43,14 +43,12 @@ use graph::{derive::CheapClone, futures03::future::join_all, prelude::alloy::pri use crate::{ ConnectionPool, NotificationSender, catalog::Catalog, + copy::CopySource, deployment::{OnSync, SubgraphHealth}, primary::{ self, DeploymentId, Mirror as PrimaryMirror, Primary, RestoreAction, RestoreMode, Site, }, - relational::{ - self, Layout, - index::{IndexList, Method}, - }, + relational::{self, Layout, index::Method}, writable::{SourceableStore, WritableStore}, }; use crate::{ @@ -1394,6 +1392,40 @@ impl Inner { store.indexes_for_entity(site, entity_name).await } + #[cfg(debug_assertions)] + pub async fn postponed_indexes_created( + &self, + deployment: &DeploymentLocator, + ) -> Result { + let site = self.find_site(deployment.id.into()).await?; + let store = self.for_site(&site)?; + store.postponed_indexes_created(&site).await + } + + #[cfg(debug_assertions)] + pub async fn set_postponed_indexes_created( + &self, + deployment: &DeploymentLocator, + created: bool, + ) -> Result<(), StoreError> { + let site = self.find_site(deployment.id.into()).await?; + let store = self.for_site(&site)?; + store.set_postponed_indexes_created(&site, created).await + } + + #[cfg(debug_assertions)] + pub async fn create_btree_index_for_deployment( + &self, + deployment: &DeploymentLocator, + name: &str, + table: &str, + expr: &str, + ) -> Result<(), StoreError> { + let site = self.find_site(deployment.id.into()).await?; + let store = self.for_site(&site)?; + store.create_btree_index(&site, name, table, expr).await + } + pub async fn drop_index_for_deployment( &self, deployment: &DeploymentLocator, @@ -1537,9 +1569,9 @@ impl Inner { src_store.load_deployment(site).await } - pub async fn load_indexes(&self, site: Arc) -> Result { + pub(crate) async fn load_copy_source(&self, site: Arc) -> Result { let src_store = self.for_site(&site)?; - src_store.load_indexes(site).await + src_store.load_copy_source(site).await } pub async fn dump( diff --git a/store/postgres/src/writable.rs b/store/postgres/src/writable.rs index 633688aeb1a..84f420d8be7 100644 --- a/store/postgres/src/writable.rs +++ b/store/postgres/src/writable.rs @@ -37,9 +37,9 @@ use tokio::select; use tokio::sync::Notify; use tokio::task::JoinHandle; +use crate::copy::CopySource; use crate::deployment_store::DeploymentStore; use crate::primary::DeploymentId; -use crate::relational::index::IndexList; use crate::{NotificationSender, retry}; use crate::{SubgraphStore, primary, primary::Site, relational::Layout}; @@ -74,8 +74,8 @@ impl WritableSubgraphStore { self.0.find_site(id).await } - async fn load_indexes(&self, site: Arc) -> Result { - self.0.load_indexes(site).await + async fn load_copy_source(&self, site: Arc) -> Result { + self.0.load_copy_source(site).await } } @@ -219,7 +219,7 @@ impl SyncStore { Some((base_id, base_ptr)) => { let src = self.store.layout(&base_id).await?; let deployment_entity = self.store.load_deployment(src.site.clone()).await?; - let indexes = self.store.load_indexes(src.site.clone()).await?; + let indexes = self.store.load_copy_source(src.site.clone()).await?; Some((src, base_ptr, deployment_entity, indexes)) } None => None, diff --git a/store/test-store/src/store.rs b/store/test-store/src/store.rs index ba57dff430d..4959e3438ca 100644 --- a/store/test-store/src/store.rs +++ b/store/test-store/src/store.rs @@ -167,36 +167,37 @@ pub async fn create_subgraph( schema: &str, base: Option<(DeploymentHash, BlockPtr)>, ) -> Result { - let schema = InputSchema::parse_latest(schema, subgraph_id.clone()).unwrap(); - - let manifest = SubgraphManifest:: { - id: subgraph_id.clone(), - spec_version: Version::new(1, 3, 0), - features: BTreeSet::new(), - description: Some(format!("manifest for {}", subgraph_id)), - repository: Some(format!("repo for {}", subgraph_id)), - schema: schema.clone(), - data_sources: vec![], - graft: None, - templates: vec![], - chain: PhantomData, - indexer_hints: None, - }; - + let (schema, manifest) = test_manifest(subgraph_id, schema); create_subgraph_with_manifest(subgraph_id, schema, manifest, base).await } +/// Like `create_subgraph`, but doesn't start the deployment. +pub async fn create_subgraph_unstarted( + subgraph_id: &DeploymentHash, + schema: &str, + base: Option<(DeploymentHash, BlockPtr)>, +) -> Result { + let (schema, manifest) = test_manifest(subgraph_id, schema); + let (name, deployment) = deployment_create(subgraph_id, &manifest, base); + SUBGRAPH_STORE + .create_deployment_replace( + name, + &schema, + deployment, + NODE_ID.clone(), + NETWORK_NAME.to_string(), + SubgraphVersionSwitchingMode::Instant, + ) + .await +} + pub async fn create_subgraph_with_manifest( subgraph_id: &DeploymentHash, schema: InputSchema, manifest: SubgraphManifest, base: Option<(DeploymentHash, BlockPtr)>, ) -> Result { - let mut yaml = serde_yaml::Mapping::new(); - yaml.insert("dataSources".into(), Vec::::new().into()); - let yaml = serde_yaml::to_string(&yaml).unwrap(); - let deployment = DeploymentCreate::new(yaml, &manifest, None).graft(base); - let name = SubgraphName::new_unchecked(subgraph_id.to_string()); + let (name, deployment) = deployment_create(subgraph_id, &manifest, base); let deployment = SUBGRAPH_STORE .create_deployment_replace( name, @@ -207,14 +208,55 @@ pub async fn create_subgraph_with_manifest( SubgraphVersionSwitchingMode::Instant, ) .await?; + start_subgraph(&deployment).await?; + Ok(deployment) +} +/// Start `deployment`; this runs a pending graft's copy. +pub async fn start_subgraph(deployment: &DeploymentLocator) -> Result<(), StoreError> { SUBGRAPH_STORE .cheap_clone() .writable(LOGGER.clone(), deployment.id, Arc::new(Vec::new())) .await? .start_subgraph_deployment(&LOGGER) - .await?; - Ok(deployment) + .await +} + +fn test_manifest( + subgraph_id: &DeploymentHash, + schema: &str, +) -> ( + InputSchema, + SubgraphManifest, +) { + let schema = InputSchema::parse_latest(schema, subgraph_id.clone()).unwrap(); + let manifest = SubgraphManifest:: { + id: subgraph_id.clone(), + spec_version: Version::new(1, 3, 0), + features: BTreeSet::new(), + description: Some(format!("manifest for {}", subgraph_id)), + repository: Some(format!("repo for {}", subgraph_id)), + schema: schema.clone(), + data_sources: vec![], + graft: None, + templates: vec![], + chain: PhantomData, + indexer_hints: None, + }; + (schema, manifest) +} + +fn deployment_create( + subgraph_id: &DeploymentHash, + manifest: &SubgraphManifest, + base: Option<(DeploymentHash, BlockPtr)>, +) -> (SubgraphName, DeploymentCreate) { + let mut yaml = serde_yaml::Mapping::new(); + yaml.insert("dataSources".into(), Vec::::new().into()); + let yaml = serde_yaml::to_string(&yaml).unwrap(); + let deployment = DeploymentCreate::new(yaml, manifest, None).graft(base); + let name = SubgraphName::new_unchecked(subgraph_id.to_string()); + (name, deployment) } pub async fn create_test_subgraph(subgraph_id: &DeploymentHash, schema: &str) -> DeploymentLocator { diff --git a/store/test-store/tests/postgres/graft.rs b/store/test-store/tests/postgres/graft.rs index 6536fc3e06c..1e917c4b14b 100644 --- a/store/test-store/tests/postgres/graft.rs +++ b/store/test-store/tests/postgres/graft.rs @@ -73,6 +73,25 @@ type User @entity(immutable: true) { } "; +const GRAFT_NEW_TABLE_GQL: &str = " +enum Color { yellow, red, blue, green } + +type User @entity { + id: ID!, + name: String, + email: String, + age: Int, + favorite_color: Color, +} + +# An entity type in the dst schema that does not exist in the src schema +type Pet @entity { + id: ID!, + name: String, + age: Int, +} +"; + const USER: &str = "User"; lazy_static! { @@ -670,3 +689,308 @@ fn prune() { }) } } + +/// Sorted names of the postponed indexes on `entity` in `loc`. +async fn postponed_attr_indexes( + store: &DieselSubgraphStore, + loc: &DeploymentLocator, + entity: &str, +) -> Result, StoreError> { + let mut names: Vec = store + .indexes_for_entity(loc, entity) + .await? + .into_iter() + .filter(|idx| idx.to_postpone()) + .filter_map(|idx| idx.name().map(str::to_string)) + .collect(); + names.sort(); + Ok(names) +} + +/// Run the postponed index build of `loc` and wait for it to finish. +async fn build_postponed_indexes( + store: &Arc, + loc: &DeploymentLocator, +) -> Result<(), StoreError> { + store + .cheap_clone() + .writable(LOGGER.clone(), loc.id, Arc::new(Vec::new())) + .await? + .create_postponed_indexes() + .await?; + for _ in 0..60 { + if store.postponed_indexes_created(loc).await? { + return Ok(()); + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + panic!( + "postponed indexes for {} were not created within 30s", + loc.hash + ); +} + +// The build skips a column that is indexed under another name. +#[test] +fn late_build_skips_columns_indexed_under_other_names() { + run_test(|store, _src| async move { + let dst_id = DeploymentHash::new("grafted_renamed").unwrap(); + let base = Some(( + DeploymentHash::new(TEST_SUBGRAPH_ID.as_str()).unwrap(), + BLOCKS[1].clone(), + )); + let dst = test_store::create_subgraph_unstarted(&dst_id, GRAFT_GQL, base).await?; + store + .create_btree_index_for_deployment(&dst, "attr_9_9_user_age", "user", "age") + .await?; + test_store::start_subgraph(&dst).await?; + store.set_postponed_indexes_created(&dst, false).await?; + + build_postponed_indexes(&store, &dst).await?; + let names = postponed_attr_indexes(&store, &dst, USER).await?; + // One index each on name, email, age, favorite_color and added. + assert_eq!(5, names.len(), "{names:?}"); + assert!(names.iter().any(|name| name == "attr_9_9_user_age")); + Ok(()) + }) +} + +// A copy replaces a set flag on the graft with the source's unset one. +#[test] +fn graft_clears_flag_set_before_interrupted_copy() { + run_test(|store, _src| async move { + let dst_id = DeploymentHash::new("grafted_interrupted").unwrap(); + let base = Some(( + DeploymentHash::new(TEST_SUBGRAPH_ID.as_str()).unwrap(), + BLOCKS[1].clone(), + )); + let dst = test_store::create_subgraph_unstarted(&dst_id, GRAFT_GQL, base).await?; + store.set_postponed_indexes_created(&dst, true).await?; + + test_store::start_subgraph(&dst).await?; + assert!(!store.postponed_indexes_created(&dst).await?); + Ok(()) + }) +} + +// A copy skips a column that the graft already indexes under the source's +// name. +#[test] +fn graft_resumes_after_indexes_copied_under_source_names() { + run_test(|store, src| async move { + build_postponed_indexes(&store, &src).await?; + + let dst_id = DeploymentHash::new("grafted_resumed").unwrap(); + let base = Some(( + DeploymentHash::new(TEST_SUBGRAPH_ID.as_str()).unwrap(), + BLOCKS[1].clone(), + )); + let dst = test_store::create_subgraph_unstarted(&dst_id, GRAFT_GQL, base).await?; + // `email` is column 3 in the source but column 2 in the graft. + store + .create_btree_index_for_deployment( + &dst, + "attr_0_3_user_email", + "user", + r#"left("email", 256)"#, + ) + .await?; + + test_store::start_subgraph(&dst).await?; + assert_eq!( + vec![ + "attr_0_1_user_name", + "attr_0_3_user_age", + "attr_0_3_user_email", + "attr_0_4_user_favorite_color", + "attr_0_5_user_added", + ], + postponed_attr_indexes(&store, &dst, USER).await? + ); + Ok(()) + }) +} + +// A graft of an unbuilt source isn't marked as built and builds its own +// indexes (#6749). +#[test] +fn graft_before_postponed_indexes() { + run_test(|store, src| async move { + assert!(!store.postponed_indexes_created(&src).await?); + assert_eq!( + Vec::::new(), + postponed_attr_indexes(&store, &src, USER).await? + ); + + let dst_id = DeploymentHash::new("grafted_postponed").unwrap(); + let dst = create_grafted_subgraph( + &dst_id, + GRAFT_GQL, + TEST_SUBGRAPH_ID.as_str(), + BLOCKS[1].clone(), + ) + .await?; + + assert!(!store.postponed_indexes_created(&dst).await?); + assert_eq!( + vec!["attr_0_5_user_added"], + postponed_attr_indexes(&store, &dst, USER).await? + ); + + build_postponed_indexes(&store, &src).await?; + build_postponed_indexes(&store, &dst).await?; + assert_eq!( + vec![ + "attr_0_1_user_name", + "attr_0_2_user_email", + "attr_0_3_user_age", + "attr_0_4_user_favorite_color", + "attr_0_5_user_added", + ], + postponed_attr_indexes(&store, &dst, USER).await? + ); + Ok(()) + }) +} + +// A graft gets the indexes an unbuilt source has, under the graft's names, +// and its own build doesn't duplicate them. +#[test] +fn graft_copies_postponed_indexes_of_unbuilt_source() { + run_test(|store, src| async move { + build_postponed_indexes(&store, &src).await?; + store.set_postponed_indexes_created(&src, false).await?; + + let dst_id = DeploymentHash::new("grafted_unbuilt").unwrap(); + let dst = create_grafted_subgraph( + &dst_id, + GRAFT_GQL, + TEST_SUBGRAPH_ID.as_str(), + BLOCKS[1].clone(), + ) + .await?; + + let all = vec![ + "attr_0_1_user_name", + "attr_0_2_user_email", + "attr_0_3_user_age", + "attr_0_4_user_favorite_color", + "attr_0_5_user_added", + ]; + assert!(!store.postponed_indexes_created(&dst).await?); + assert_eq!(all, postponed_attr_indexes(&store, &dst, USER).await?); + + build_postponed_indexes(&store, &dst).await?; + assert_eq!(all, postponed_attr_indexes(&store, &dst, USER).await?); + Ok(()) + }) +} + +// Same as `graft_before_postponed_indexes`, for a copy to another shard. +// Needs a test configuration with at least two shards. +#[test] +fn copy_before_postponed_indexes() { + run_test(|store, src| async move { + let Some(dst_shard) = other_shard(&store, &src).await? else { + return Ok(()); + }; + assert!(!store.postponed_indexes_created(&src).await?); + + let dst = store + .copy_deployment( + &src, + dst_shard, + NODE_ID.clone(), + BLOCKS[1].clone(), + OnSync::None, + ) + .await?; + store + .cheap_clone() + .writable(LOGGER.clone(), dst.id, Arc::new(Vec::new())) + .await? + .start_subgraph_deployment(&LOGGER) + .await?; + // `indexes_for_entity` looks up the active deployment for a hash. + store.activate(&dst).await?; + + assert!(!store.postponed_indexes_created(&dst).await?); + assert_eq!( + Vec::::new(), + postponed_attr_indexes(&store, &dst, USER).await? + ); + + build_postponed_indexes(&store, &dst).await?; + assert_eq!( + vec![ + "attr_0_1_user_name", + "attr_0_2_user_bin_name", + "attr_0_3_user_email", + "attr_0_4_user_age", + "attr_0_5_user_seconds_age", + "attr_0_6_user_weight", + "attr_0_7_user_coffee", + "attr_0_8_user_favorite_color", + ], + postponed_attr_indexes(&store, &dst, USER).await? + ); + Ok(()) + }) +} + +// An index dropped on a built source stays dropped on a graft (#6583). +#[test] +fn graft_keeps_dropped_postponed_index() { + run_test(|store, src| async move { + build_postponed_indexes(&store, &src).await?; + store + .drop_index_for_deployment(&src, "attr_0_3_user_email") + .await?; + + let dst_id = DeploymentHash::new("grafted_dropped").unwrap(); + let dst = create_grafted_subgraph( + &dst_id, + GRAFT_GQL, + TEST_SUBGRAPH_ID.as_str(), + BLOCKS[1].clone(), + ) + .await?; + + assert!(store.postponed_indexes_created(&dst).await?); + assert_eq!( + vec![ + "attr_0_1_user_name", + "attr_0_3_user_age", + "attr_0_4_user_favorite_color", + "attr_0_5_user_added", + ], + postponed_attr_indexes(&store, &dst, USER).await? + ); + Ok(()) + }) +} + +// A graft builds the postponed indexes of entity types its source lacks. +#[test] +fn graft_new_table_gets_postponed_indexes() { + run_test(|store, src| async move { + build_postponed_indexes(&store, &src).await?; + + let dst_id = DeploymentHash::new("grafted_new_table").unwrap(); + let dst = create_grafted_subgraph( + &dst_id, + GRAFT_NEW_TABLE_GQL, + TEST_SUBGRAPH_ID.as_str(), + BLOCKS[1].clone(), + ) + .await?; + + assert!(store.postponed_indexes_created(&dst).await?); + // Table positions in index names depend on the schema's type order. + let pet = postponed_attr_indexes(&store, &dst, "Pet").await?; + assert_eq!(2, pet.len(), "{pet:?}"); + assert!(pet.iter().any(|n| n.ends_with("_pet_name")), "{pet:?}"); + assert!(pet.iter().any(|n| n.ends_with("_pet_age")), "{pet:?}"); + Ok(()) + }) +}