From 47e1ef91b00ab4642abfda09a93f24c655d8a5a8 Mon Sep 17 00:00:00 2001 From: Maksim Dimitrov Date: Mon, 9 Nov 2026 12:36:29 +0200 Subject: [PATCH 1/2] store: Skip postponed indexes that already exist under another name The background build of postponed indexes skipped an index only if one with the same name existed. Index names contain the positions of the table and the column. A copy's positions can differ from its source's, and older versions copied indexes under the source's names. If such a copy's flag was reset, the build added a second index on every shifted column. The build now compares the shape of the indexes, not their names. It only counts valid indexes, so that the leftover of an interrupted build is still dropped and rebuilt. --- store/postgres/src/catalog.rs | 22 ++- store/postgres/src/copy.rs | 2 +- store/postgres/src/deployment.rs | 9 +- store/postgres/src/deployment_store.rs | 61 +++++-- store/postgres/src/relational/index.rs | 202 ++++++++++++++++++++++- store/postgres/src/subgraph_store.rs | 34 ++++ store/test-store/src/store.rs | 88 +++++++--- store/test-store/tests/postgres/graft.rs | 65 ++++++++ 8 files changed, 427 insertions(+), 56 deletions(-) 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..709954d4bca 100644 --- a/store/postgres/src/copy.rs +++ b/store/postgres/src/copy.rs @@ -337,7 +337,7 @@ impl CopyState { } conn.transaction(async |conn| { - deployment::set_postponed_indexes_created(conn, &self.dst.site).await + deployment::set_postponed_indexes_created(conn, &self.dst.site, true).await }) .await?; 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..15978fdf260 100644 --- a/store/postgres/src/deployment_store.rs +++ b/store/postgres/src/deployment_store.rs @@ -711,7 +711,7 @@ 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()) } @@ -723,6 +723,40 @@ impl DeploymentStore { IndexList::load(&mut conn, &layout).await } + 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 + } + + // 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. pub(crate) async fn drop_index( &self, @@ -1706,14 +1740,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 +1813,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 +1837,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..96b0487bb98 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,6 +674,20 @@ impl CreateIndex { } } + /// The user-defined columns this index covers. + pub fn attribute_columns(&self) -> Vec<&str> { + match self { + CreateIndex::Unknown { .. } => vec![], + CreateIndex::Parsed { columns, .. } => columns + .iter() + .filter_map(|expr| match expr { + Expr::Column(name) | Expr::Prefix(name, _) => Some(name.as_str()), + _ => None, + }) + .collect(), + } + } + /// 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`. @@ -898,6 +913,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 +977,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) @@ -1348,4 +1412,138 @@ mod tests { "index columns exist in table regardless of order" ); } + + fn thing_table(layout: &Layout) -> &Table { + layout + .table_for_entity(&layout.input_schema.entity_type("Thing").unwrap()) + .unwrap() + } + + #[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}"); + } + } + + 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..e3460c09698 100644 --- a/store/postgres/src/subgraph_store.rs +++ b/store/postgres/src/subgraph_store.rs @@ -1394,6 +1394,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, 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..71b7dfdd4e3 100644 --- a/store/test-store/tests/postgres/graft.rs +++ b/store/test-store/tests/postgres/graft.rs @@ -670,3 +670,68 @@ 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(()) + }) +} From 0d0f6bceba987d4b5218886bf39cd81006b4a764 Mon Sep 17 00:00:00 2001 From: Maksim Dimitrov Date: Mon, 9 Nov 2026 15:18:36 +0200 Subject: [PATCH 2/2] store: Copy postponed indexes of a graft/copy source under own names At the end of a copy or graft, `create_indexes` recreated the postponed indexes that existed on the source. It then always marked the destination as having built them. If the source hadn't built them yet, the destination never got them. Entity types that a graft added never got theirs either. The copy now reads the source's flag when it starts, before the source's indexes and on the source's shard. At the end, the destination gets its own postponed indexes for the columns the source has indexed, and for columns and entity types the source doesn't have. They are named for the destination, so its own build later doesn't add a second index on a column whose position changed. The destination then gets the source's flag. This also clears a flag that an older version set before the copy was interrupted. Fixes #6749 --- store/postgres/src/copy.rs | 71 ++++--- store/postgres/src/deployment_store.rs | 21 +- store/postgres/src/relational/index.rs | 134 ------------ store/postgres/src/subgraph_store.rs | 10 +- store/postgres/src/writable.rs | 8 +- store/test-store/tests/postgres/graft.rs | 259 +++++++++++++++++++++++ 6 files changed, 318 insertions(+), 185 deletions(-) diff --git a/store/postgres/src/copy.rs b/store/postgres/src/copy.rs index 709954d4bca..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, true).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_store.rs b/store/postgres/src/deployment_store.rs index 15978fdf260..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; @@ -716,11 +717,19 @@ impl DeploymentStore { 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?; - - IndexList::load(&mut conn, &layout).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 { @@ -1583,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 {}", @@ -1619,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); } diff --git a/store/postgres/src/relational/index.rs b/store/postgres/src/relational/index.rs index 96b0487bb98..ae399513788 100644 --- a/store/postgres/src/relational/index.rs +++ b/store/postgres/src/relational/index.rs @@ -688,77 +688,6 @@ 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 { - 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 - .iter() - .any(|c| c.name.as_str() == column_name) - } - - 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 /// `true`, make it a concurrent index creation. If `if_not_exists` is /// `true` add a `if not exists` clause to the index creation. @@ -986,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)] @@ -1374,45 +1279,6 @@ 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 - .table_for_entity(&layout.input_schema.entity_type("Thing").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, - }; - - assert!( - index.fields_exist_in_dest(table), - "index columns exist in table regardless of order" - ); - } - fn thing_table(layout: &Layout) -> &Table { layout .table_for_entity(&layout.input_schema.entity_type("Thing").unwrap()) diff --git a/store/postgres/src/subgraph_store.rs b/store/postgres/src/subgraph_store.rs index e3460c09698..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::{ @@ -1571,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/tests/postgres/graft.rs b/store/test-store/tests/postgres/graft.rs index 71b7dfdd4e3..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! { @@ -735,3 +754,243 @@ fn late_build_skips_columns_indexed_under_other_names() { 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(()) + }) +}