diff --git a/docs/environment-variables.md b/docs/environment-variables.md index 4bfdd29b540..58381990fe0 100644 --- a/docs/environment-variables.md +++ b/docs/environment-variables.md @@ -298,6 +298,13 @@ those. - `GRAPH_POSTPONE_ATTRIBUTE_INDEX_CREATION`: During the coping of a subgraph postponing creation of certain indexes (btree, attribute based ones), would speed up syncing +- `GRAPH_STORE_COPY_MIRROR_INDEXES`: Whether copies and grafts take the + indexes created with their tables from their source: attribute indexes + dropped on the source stay dropped, and the source's other indexes, + including manual ones, are carried over. When `false`, these indexes + come from the destination's own layout, as before. Set it on the node + that runs `graphman copy create` and on the nodes that handle deployments + (for grafts). Defaults to `true`. - `GRAPH_STORE_INSERT_EXTRA_COLS`: Makes it possible to work around bugs in the subgraph writing code that manifest as Postgres errors saying 'number of parameters must be between 0 and 65535' Such errors are always diff --git a/graph/src/env/store.rs b/graph/src/env/store.rs index 92c5564d17f..81aa2466f94 100644 --- a/graph/src/env/store.rs +++ b/graph/src/env/store.rs @@ -141,6 +141,10 @@ pub struct EnvVarsStore { /// Whether to create GIN indexes for array attributes. Set by /// `GRAPH_STORE_CREATE_GIN_INDEXES`. The default is `false` pub create_gin_indexes: bool, + /// Whether copies and grafts take the indexes created with their tables + /// from their source instead of their own layout. Set by + /// `GRAPH_STORE_COPY_MIRROR_INDEXES`. The default is `true`. + pub copy_mirror_indexes: bool, /// Temporary env var in case we need to quickly rollback PR #5010 pub use_brin_for_all_query_types: bool, /// Temporary env var to disable certain lookups in the chain store @@ -239,6 +243,7 @@ impl TryFrom for EnvVarsStore { write_batch_duration: Duration::from_secs(x.write_batch_duration_in_secs), write_batch_size: x.write_batch_size * 1_000, create_gin_indexes: x.create_gin_indexes, + copy_mirror_indexes: x.copy_mirror_indexes, use_brin_for_all_query_types: x.use_brin_for_all_query_types, disable_block_cache_for_lookup: x.disable_block_cache_for_lookup, ignore_block_cache: x.ignore_block_cache, @@ -346,6 +351,8 @@ pub struct InnerStore { write_batch_memoize: bool, #[envconfig(from = "GRAPH_STORE_CREATE_GIN_INDEXES", default = "false")] create_gin_indexes: bool, + #[envconfig(from = "GRAPH_STORE_COPY_MIRROR_INDEXES", default = "true")] + copy_mirror_indexes: bool, #[envconfig(from = "GRAPH_STORE_USE_BRIN_FOR_ALL_QUERY_TYPES", default = "false")] use_brin_for_all_query_types: bool, #[envconfig(from = "GRAPH_STORE_DISABLE_BLOCK_CACHE_FOR_LOOKUP", default = "false")] diff --git a/store/postgres/examples/layout.rs b/store/postgres/examples/layout.rs index e7fde150fe9..05b1ac74e7f 100644 --- a/store/postgres/examples/layout.rs +++ b/store/postgres/examples/layout.rs @@ -64,7 +64,7 @@ fn print_delete_all(layout: &Layout) { } fn print_ddl(layout: &Layout) { - let ddl = ensure(layout.as_ddl(), "Failed to generate DDL"); + let ddl = ensure(layout.as_ddl(None), "Failed to generate DDL"); println!("{}", ddl); } diff --git a/store/postgres/src/copy.rs b/store/postgres/src/copy.rs index fd076ab0e1b..7090dbb2a8a 100644 --- a/store/postgres/src/copy.rs +++ b/store/postgres/src/copy.rs @@ -32,7 +32,10 @@ use crate::{ dynds::DataSourcesTable, pool::TxnCallback, primary::{DeploymentId, Primary, Site}, - relational::{Layout, Table, index::IndexList}, + relational::{ + Layout, Table, + index::{IndexList, mirrored_table}, + }, relational_queries as rq, vid_batcher::{VidBatcher, VidRange}, }; @@ -318,19 +321,7 @@ impl CopyState { 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 src = mirrored_table(dst_table, (self.src.as_ref(), &source.list)); let idxs = dst_table.postponed_indexes_to_create(&self.dst.input_schema, &existing, src)?; creat.execute_many(conn, &idxs).await?; diff --git a/store/postgres/src/deployment_store.rs b/store/postgres/src/deployment_store.rs index ab589d95577..75c8dac143c 100644 --- a/store/postgres/src/deployment_store.rs +++ b/store/postgres/src/deployment_store.rs @@ -53,7 +53,7 @@ use crate::deployment::{self, OnSync}; use crate::detail::ErrorDetail; use crate::dynds::DataSourcesTable; use crate::primary::{DeploymentId, Primary}; -use crate::relational::index::{CreateIndex, IndexCreator, IndexList, Method}; +use crate::relational::index::{CreateIndex, IndexCreator, IndexList, Method, SourceIndex}; use crate::relational::{self, Layout, LayoutCache, STATEMENT_TIMEOUT, SqlName, Table}; use crate::relational_queries::{FromEntityData, JSONData}; use crate::{AsyncPgConnection, advisory_lock, catalog, retry}; @@ -177,6 +177,8 @@ impl DeploymentStore { DeploymentStore(Arc::new(store)) } + /// For a copy or graft that mirrors its source, `src` is the source's + /// layout and indexes. pub(crate) async fn create_deployment( &self, schema: &InputSchema, @@ -184,7 +186,9 @@ impl DeploymentStore { site: Arc, replace: bool, on_sync: OnSync, + src: Option<(&Layout, &IndexList)>, ) -> Result<(), StoreError> { + let logger = self.logger.cheap_clone(); let mut conn = self.pool.get_permitted().await?; conn.transaction::<_, StoreError, _>(async |conn| { let exists = deployment::exists(conn, &site).await?; @@ -211,13 +215,17 @@ impl DeploymentStore { let query = format!("create schema {}", &site.namespace); conn.batch_execute(&query).await?; - let _ = Layout::create_relational_schema( + let layout = Layout::create_relational_schema( conn, site.clone(), schema, entities_with_causality_region.into_iter().collect(), + src, ) .await?; + if let Some(src) = src { + copy_source_indexes(&logger, conn, &layout, src).await?; + } // Create data sources table if site.schema_version.private_data_sources() { @@ -725,7 +733,9 @@ impl DeploymentStore { 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?; + // Invalid indexes are leftovers of interrupted concurrent drops or + // builds. + let list = IndexList::load(&mut conn, &layout, true).await?; Ok(CopySource { list, postponed_indexes_created, @@ -756,11 +766,13 @@ impl DeploymentStore { name: &str, table: &str, expr: &str, + unique: bool, ) -> Result<(), StoreError> { let mut conn = self.pool.get_permitted().await?; + let unique = if unique { "unique " } else { "" }; let query = format!( - "create index \"{}\" on \"{}\".\"{}\" using btree ({})", - name, site.namespace, table, expr + "create {}index concurrently \"{}\" on \"{}\".\"{}\" using btree ({})", + unique, name, site.namespace, table, expr ); sql_query(query).execute(&mut conn).await?; Ok(()) @@ -930,7 +942,7 @@ impl DeploymentStore { // Layout and index list are schema metadata — safe to load outside // the snapshot transaction let layout = self.layout(&mut conn, site.cheap_clone()).await?; - let index_list = IndexList::load(&mut conn, &layout).await?; + let index_list = IndexList::load(&mut conn, &layout, false).await?; // Use REPEATABLE READ to get a consistent MVCC snapshot for the // entire dump. All queries inside see the same database state, @@ -2117,6 +2129,37 @@ impl DeploymentStore { } } +/// Create the indexes of the copy source `src` that `layout` doesn't create +/// itself. Each runs in its own savepoint, so one that fails is skipped. +async fn copy_source_indexes( + logger: &Logger, + conn: &mut AsyncPgConnection, + layout: &Layout, + src: (&Layout, &IndexList), +) -> Result<(), StoreError> { + for (defn, action) in layout.source_indexes(src)? { + match action { + SourceIndex::Copy(sql) => { + let res = conn + .transaction::<_, StoreError, _>(async |conn| { + sql_query(&sql).execute(conn).await?; + Ok(()) + }) + .await; + if let Err(e) = res { + warn!(logger, "Could not copy an index of the source"; + "index" => &defn, "error" => e.to_string()); + } + } + SourceIndex::Skip(reason) => { + warn!(logger, "Not copying an index of the source"; + "index" => &defn, "reason" => reason); + } + } + } + Ok(()) +} + /// Tries to fetch a [`Table`] either by its Entity name or its SQL name. /// /// Since we allow our input to be either camel-case or snake-case, we must retry the diff --git a/store/postgres/src/relational.rs b/store/postgres/src/relational.rs index 4059e736635..7b6b7c71284 100644 --- a/store/postgres/src/relational.rs +++ b/store/postgres/src/relational.rs @@ -381,12 +381,13 @@ impl Layout { site: Arc, schema: &InputSchema, entities_with_causality_region: BTreeSet, + src: Option<(&Layout, &index::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(src) .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..daf4f152638 100644 --- a/store/postgres/src/relational/ddl.rs +++ b/store/postgres/src/relational/ddl.rs @@ -13,7 +13,9 @@ use crate::{block_range::CAUSALITY_REGION_COLUMN, relational::index::Cond}; use super::{ Column, Layout, SqlName, Table, - index::{CreateIndex, Expr, IndexCreator, Method, PrefixKind}, + index::{ + CreateIndex, Expr, IndexCreator, IndexList, Method, PrefixKind, SourceTable, mirrored_table, + }, }; // In debug builds (for testing etc.) unconditionally create exclusion constraints, in release @@ -28,8 +30,9 @@ impl Layout { /// and `create index` etc. statements needed in the database schema /// /// See the unit tests at the end of this file for the actual DDL that - /// gets generated - pub fn as_ddl(&self) -> Result { + /// gets generated. For a copy or graft, `src` is the source's layout and + /// indexes. + pub fn as_ddl(&self, src: Option<(&Layout, &IndexList)>) -> Result { let mut out = String::new(); // Output enums first so table definitions can reference them @@ -42,7 +45,8 @@ 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)?; + let src = src.and_then(|src| mirrored_table(table, src)); + table.as_ddl(&self.input_schema, &creat, src, &mut out)?; } Ok(out) @@ -406,7 +410,9 @@ impl Table { } /// Generate the DDL for one table, i.e. one `create table` statement - /// and all `create index` statements for the table's columns + /// and all `create index` statements for the table's columns. With + /// `src`, the table of a copy source, attribute indexes the source + /// dropped are left out. /// /// See the unit tests at the end of this file for the actual DDL that /// gets generated @@ -414,11 +420,13 @@ impl Table { &self, schema: &InputSchema, creat: &IndexCreator, + src: Option, out: &mut String, ) -> fmt::Result { self.create_table(out)?; for idx in self.indexes(schema)? { - if !idx.to_postpone() { + let dropped = src.is_some_and(|src| idx.is_attr_index() && src.dropped(&idx)); + if !idx.to_postpone() && !dropped { 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..0fc91089d43 100644 --- a/store/postgres/src/relational/ddl_tests.rs +++ b/store/postgres/src/relational/ddl_tests.rs @@ -8,9 +8,13 @@ use crate::{deployment_store::generate_index_creation_sql, layout_for_tests::mak const ID_TYPE: ColumnType = ColumnType::String; fn test_layout(gql: &str) -> Arc { + test_layout_in(gql, "sgd0815") +} + +fn test_layout_in(gql: &str, nsp: &str) -> Arc { let subgraph = DeploymentHash::new("subgraph").unwrap(); let schema = InputSchema::parse_latest(gql, subgraph.clone()).expect("Test schema invalid"); - let namespace = Namespace::new("sgd0815".to_owned()).unwrap(); + let namespace = Namespace::new(nsp.to_owned()).unwrap(); let site = Arc::new(make_dummy_site(subgraph, namespace, "anet".to_string())); let ents = { match schema.entity_type("FileThing") { @@ -190,31 +194,31 @@ create index concurrently if not exists attr_1_7_scalar_color #[test] fn generate_ddl() { let layout = test_layout(THING_GQL); - let sql = layout.as_ddl().expect("Failed to generate DDL"); + let sql = layout.as_ddl(None).expect("Failed to generate DDL"); assert_eq!(THING_DDL, &sql); // Use `assert_eq!` to also test the formatting. let layout = test_layout(MUSIC_GQL); - let sql = layout.as_ddl().expect("Failed to generate DDL"); + let sql = layout.as_ddl(None).expect("Failed to generate DDL"); check_eqv(MUSIC_DDL, &sql); let layout = test_layout(FOREST_GQL); - let sql = layout.as_ddl().expect("Failed to generate DDL"); + let sql = layout.as_ddl(None).expect("Failed to generate DDL"); check_eqv(FOREST_DDL, &sql); let layout = test_layout(FULLTEXT_GQL); - let sql = layout.as_ddl().expect("Failed to generate DDL"); + let sql = layout.as_ddl(None).expect("Failed to generate DDL"); check_eqv(FULLTEXT_DDL, &sql); let layout = test_layout(FORWARD_ENUM_GQL); - let sql = layout.as_ddl().expect("Failed to generate DDL"); + let sql = layout.as_ddl(None).expect("Failed to generate DDL"); check_eqv(FORWARD_ENUM_SQL, &sql); let layout = test_layout(TS_GQL); - let sql = layout.as_ddl().expect("Failed to generate DDL"); + let sql = layout.as_ddl(None).expect("Failed to generate DDL"); check_eqv(TS_SQL, &sql); let layout = test_layout(LIFETIME_GQL); - let sql = layout.as_ddl().expect("Failed to generate DDL"); + let sql = layout.as_ddl(None).expect("Failed to generate DDL"); check_eqv(LIFETIME_SQL, &sql); } @@ -341,7 +345,7 @@ fn postponed_indexes_with_block_column() { // Create everything; when postpone is enabled (debug builds), the // attribute btree index is omitted from the DDL - let sql = layout.as_ddl().unwrap(); + let sql = layout.as_ddl(None).unwrap(); assert!(sql.contains(&cr(BLOCK_IDX))); // The block$ time-travel index appears exactly once let count = sql.matches(BLOCK_IDX).count(); @@ -362,6 +366,152 @@ fn postponed_indexes_with_block_column() { assert!(sql.contains(&cr(ATTR_IDX))); } +const MIRROR_GQL: &str = " +type Thing @entity { + id: ID! + owner: Owner! + name: String! +} +type Owner @entity { + id: ID! +}"; + +fn index_list(tables: Vec<(&str, Vec<&str>)>) -> index::IndexList { + let definitions = tables + .into_iter() + .map(|(table, defns)| { + let defns = defns.into_iter().map(str::to_string).collect(); + (table.to_string(), defns) + }) + .collect(); + index::IndexList::new(definitions) +} + +#[test] +fn ddl_mirrors_source_indexes() { + // The source doesn't index `thing.owner`. + let src = test_layout_in(MIRROR_GQL, "sgd0816"); + let indexes = index_list(vec![ + ( + "thing", + vec!["CREATE INDEX attr_0_0_thing_id ON sgd0816.thing USING btree (id)"], + ), + ( + "owner", + vec!["CREATE INDEX attr_1_0_owner_id ON sgd0816.owner USING btree (id)"], + ), + ]); + let layout = test_layout(&format!("{MIRROR_GQL}\ntype Pet @entity {{ id: ID! }}")); + let sql = layout.as_ddl(Some((&src, &indexes))).unwrap(); + + assert!(sql.contains("attr_0_0_thing_id"), "{sql}"); + assert!(!sql.contains("attr_0_1_thing_owner"), "{sql}"); + assert!(sql.contains("attr_1_0_owner_id"), "{sql}"); + // `pet` doesn't exist in the source. + assert!(sql.contains("attr_2_0_pet_id"), "{sql}"); +} + +#[test] +fn source_indexes_copied_verbatim() { + const GQL: &str = " + enum Status { Active, Inactive } + type Thing @entity { + id: ID! + from: String! + tags: [String!]! + status: Status! + } + type Other @entity { + id: ID! + }"; + let src = test_layout_in( + &GQL.replace("id: ID!\n }", "id: ID!\n color: String!\n }"), + "sgd0816", + ); + let indexes = index_list(vec![ + ( + "thing", + vec![ + "CREATE INDEX attr_0_0_thing_id ON sgd0816.thing USING btree (id) WITH (fillfactor='70')", + "CREATE INDEX attr_0_2_thing_tags ON sgd0816.thing USING gin (tags)", + r#"CREATE INDEX manual_thing_from ON sgd0816.thing USING btree ("from" DESC)"#, + "CREATE INDEX manual_thing_active ON sgd0816.thing USING btree (id) WHERE (status = 'Active'::sgd0816.status)", + "CREATE INDEX brin_thing ON sgd0816.thing USING brin (lower(block_range), COALESCE(upper(block_range), 2147483647), vid) WITH (pages_per_range='32')", + r#"CREATE INDEX "my-idx" ON sgd0816.thing USING btree (id)"#, + ], + ), + ( + "other", + vec![ + r#"CREATE INDEX attr_1_1_other_color ON sgd0816.other USING btree ("left"(color, 256))"#, + "CREATE INDEX manual_other_color ON sgd0816.other USING btree (color)", + ], + ), + ]); + let layout = test_layout(GQL); + let actions: HashMap<_, _> = layout + .source_indexes((&src, &indexes)) + .unwrap() + .into_iter() + .collect(); + + use index::SourceIndex::*; + let copy = |sql: &str| Copy(sql.to_string()); + // Indexes under graph-node's names are left out, tuned or on a removed + // column; `tags` is copied since the layout doesn't index it. + let expected: HashMap = [ + ( + "CREATE INDEX attr_0_2_thing_tags ON sgd0816.thing USING gin (tags)", + copy("CREATE INDEX attr_0_2_thing_tags ON sgd0815.thing USING gin (tags)"), + ), + ( + r#"CREATE INDEX manual_thing_from ON sgd0816.thing USING btree ("from" DESC)"#, + copy(r#"CREATE INDEX manual_thing_from ON sgd0815.thing USING btree ("from" DESC)"#), + ), + ( + "CREATE INDEX manual_thing_active ON sgd0816.thing USING btree (id) WHERE (status = 'Active'::sgd0816.status)", + copy("CREATE INDEX manual_thing_active ON sgd0815.thing USING btree (id) WHERE (status = 'Active'::sgd0815.status)"), + ), + ( + r#"CREATE INDEX "my-idx" ON sgd0816.thing USING btree (id)"#, + Skip("it can't be parsed"), + ), + ( + "CREATE INDEX manual_other_color ON sgd0816.other USING btree (color)", + Skip("it uses a column this table doesn't have"), + ), + ] + .into_iter() + .map(|(defn, action)| (defn.to_string(), action)) + .collect(); + assert_eq!(expected, actions); + + // The layout's `id` index replaces the source's tuned one. + let sql = layout.as_ddl(Some((&src, &indexes))).unwrap(); + assert!(sql.contains("attr_0_0_thing_id"), "{sql}"); +} + +#[test] +fn ddl_ignores_source_indexes_after_mutability_change() { + let src = test_layout_in( + &MIRROR_GQL.replace("type Thing @entity", "type Thing @entity(immutable: true)"), + "sgd0816", + ); + let manual = "CREATE INDEX manual_thing_name ON sgd0816.thing USING btree (name)"; + let indexes = index_list(vec![("thing", vec![manual])]); + let layout = test_layout(MIRROR_GQL); + let sql = layout.as_ddl(Some((&src, &indexes))).unwrap(); + + assert!(sql.contains("attr_0_0_thing_id"), "{sql}"); + assert_eq!( + vec![( + manual.to_string(), + index::SourceIndex::Skip("its table changed mutability") + )], + layout.source_indexes((&src, &indexes)).unwrap() + ); +} + const THING_GQL: &str = r#" type Thing @entity { id: ID! diff --git a/store/postgres/src/relational/index.rs b/store/postgres/src/relational/index.rs index ae399513788..609a4d6265d 100644 --- a/store/postgres/src/relational/index.rs +++ b/store/postgres/src/relational/index.rs @@ -10,7 +10,7 @@ use graph::components::store::StoreError; use graph::itertools::Itertools; use graph::prelude::{ BlockNumber, lazy_static, - regex::{Captures, Regex}, + regex::{self, Captures, Regex}, }; use graph::schema::InputSchema; @@ -644,17 +644,10 @@ impl CreateIndex { return false; } - fn has_prefix(s: &str, prefix: &str) -> bool { - s.starts_with(prefix) - || s.ends_with("\"") && s.starts_with(format!("\"{}", prefix).as_str()) - } match self { CreateIndex::Unknown { .. } => false, CreateIndex::Parsed { - name, - columns, - method, - .. + columns, method, .. } => { if *method != Method::BTree { return false; @@ -662,11 +655,23 @@ impl CreateIndex { if columns.len() == 1 && columns[0].is_id() { return false; } - has_prefix(name, "attr_") && self.is_attribute_index() + self.is_attr_index() } } } + /// Whether this index has the name of an attribute index graph-node creates. + pub fn has_attr_name(&self) -> bool { + self.name().is_some_and(|name| { + name.starts_with("attr_") || name.ends_with('"') && name.starts_with("\"attr_") + }) + } + + /// Whether this is an attribute index that graph-node created. + pub fn is_attr_index(&self) -> bool { + self.has_attr_name() && self.is_attribute_index() + } + pub fn name(&self) -> Option<&str> { match self { CreateIndex::Unknown { .. } => None, @@ -855,17 +860,15 @@ impl Table { /// 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. + /// its columns but no `attr_` index on them. pub(crate) fn postponed_indexes_to_create( &self, schema: &InputSchema, existing: &[CreateIndex], - src: Option<(&Table, &[CreateIndex])>, + src: Option, ) -> Result, StoreError> { - let indexes = self.indexes(schema).map_err(|e| { - StoreError::ConstraintViolation(format!("failed to generate indexes: {}", e)) - })?; - Ok(indexes + Ok(self + .layout_indexes(schema)? .into_iter() .filter(|idx| idx.to_postpone()) .filter(|idx| { @@ -875,45 +878,209 @@ impl Table { { 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) + src.is_none_or(|src| !src.dropped(idx)) }) .collect()) } + + /// The indexes of the copy source `src` that this table doesn't create + /// itself, each with its definition on the source and what to do with + /// it. + pub(crate) fn source_indexes( + &self, + schema: &InputSchema, + src: SourceTable, + ) -> Result, StoreError> { + let own = self.layout_indexes(schema)?; + let has_column = |name: &str| self.columns.iter().any(|c| c.name.as_str() == name); + let src_nsp = regex::escape(src.table.nsp.as_str()); + let src_nsp = Regex::new(&format!(r"\b{}\.", src_nsp)).unwrap(); + + let mut indexes = Vec::new(); + for (idx, defn) in src.indexes.iter().zip(src.definitions) { + let CreateIndex::Parsed { + name, + columns, + cond, + .. + } = idx + else { + indexes.push((defn.clone(), SourceIndex::Skip("it can't be parsed"))); + continue; + }; + let columns_indexed = |o: &CreateIndex| { + o.is_attr_index() && o.attribute_columns() == idx.attribute_columns() + }; + // Postgres quotes names like `brin_poi2$`; the layout's are unquoted. + let builtin_name = |o: &CreateIndex| o.name() == Some(name.trim_matches('"')); + if idx.is_default_non_attr_index() + || (idx.is_id() && self.immutable) + || own + .iter() + .any(|o| !o.is_attr_index() && (idx.is_same_index(o) || builtin_name(o))) + || (idx.has_attr_name() + && (own.iter().any(columns_indexed) + || !idx.attribute_columns().into_iter().all(has_column))) + { + // graph-node's own: this table creates the ones it needs + continue; + } + // An expression or condition that isn't parsed can use any column. + let opaque = columns.iter().any(|expr| matches!(expr, Expr::Unknown(_))) + || matches!(cond, Some(Cond::Unknown(_))); + let action = if (opaque + && !src + .table + .columns + .iter() + .all(|c| has_column(c.name.as_str()))) + || !idx.attribute_columns().into_iter().all(has_column) + { + SourceIndex::Skip("it uses a column this table doesn't have") + } else { + let sql = src_nsp.replace_all(defn, format!("{}.", self.nsp)); + SourceIndex::Copy(sql.into_owned()) + }; + indexes.push((defn.clone(), action)); + } + Ok(indexes) + } + + fn layout_indexes(&self, schema: &InputSchema) -> Result, StoreError> { + self.indexes(schema).map_err(|e| { + StoreError::ConstraintViolation(format!("failed to generate indexes: {}", e)) + }) + } +} + +impl Layout { + /// The indexes of the copy source `src` that this layout doesn't create + /// itself, each with its definition on the source and what to do with + /// it. + pub(crate) fn source_indexes( + &self, + src: (&Layout, &IndexList), + ) -> Result, StoreError> { + let (src_layout, src_indexes) = src; + let mut indexes = Vec::new(); + for table in self.tables.values() { + let Some(src_table) = src_layout.table(table.name.as_str()) else { + continue; + }; + let mut actions = + table.source_indexes(&self.input_schema, src_indexes.source_table(src_table))?; + if src_table.immutable != table.immutable { + for (_, action) in &mut actions { + if let SourceIndex::Copy(_) = action { + *action = SourceIndex::Skip("its table changed mutability"); + } + } + } + indexes.extend(actions); + } + Ok(indexes) + } +} + +/// The source table that `table` mirrors in a copy from `src`: the one with +/// the same name and mutability. Index kinds differ between mutable and +/// immutable tables. +pub(crate) fn mirrored_table<'a>( + table: &Table, + src: (&'a Layout, &'a IndexList), +) -> Option> { + let (src_layout, src_indexes) = src; + let src_table = src_layout + .table(table.name.as_str()) + .filter(|src_table| src_table.immutable == table.immutable)?; + Some(src_indexes.source_table(src_table)) +} + +/// A table of a copy source and its indexes. +#[derive(Clone, Copy)] +pub(crate) struct SourceTable<'a> { + table: &'a Table, + indexes: &'a [CreateIndex], + /// The definitions of `indexes` as Postgres printed them. + definitions: &'a [String], +} + +impl SourceTable<'_> { + /// Whether the source has the columns of `idx` but no `attr_` index on them. + pub(crate) fn dropped(&self, idx: &CreateIndex) -> bool { + let columns = idx.attribute_columns(); + columns + .iter() + .all(|col| self.table.columns.iter().any(|c| c.name.as_str() == *col)) + && !self + .indexes + .iter() + .any(|s| s.has_attr_name() && s.attribute_columns() == columns) + } +} + +/// What happens to an index of a copy source that the destination doesn't +/// create itself. +#[derive(Debug, PartialEq)] +pub(crate) enum SourceIndex { + /// Create it with this statement. + Copy(String), + /// Leave it out, for this reason. + Skip(&'static str), } #[derive(Debug)] pub struct IndexList { pub(crate) indexes: HashMap>, + /// The definitions of `indexes` as Postgres printed them. + definitions: HashMap>, } impl IndexList { - pub fn new(indexes: HashMap>) -> Self { - IndexList { indexes } + /// Parse `definitions`, the index definitions of each table. + pub fn new(definitions: HashMap>) -> Self { + let indexes = definitions + .iter() + .map(|(table, defns)| { + let indexes = defns.iter().cloned().map(CreateIndex::parse).collect(); + (table.clone(), indexes) + }) + .collect(); + IndexList { + indexes, + definitions, + } } - pub async fn load(conn: &mut AsyncPgConnection, layout: &Layout) -> Result { - let mut indexes = HashMap::new(); + /// Load the indexes of `layout`, only the valid ones with `valid_only`. + pub async fn load( + conn: &mut AsyncPgConnection, + layout: &Layout, + valid_only: bool, + ) -> Result { + let mut definitions = HashMap::new(); 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(), false) - .await? - .into_iter() - .map(CreateIndex::parse) - .collect(); - indexes.insert(table.name.to_string(), indexes_from_table); + let defns = catalog::indexes_for_table( + conn, + schema_name.as_str(), + table.name.as_str(), + valid_only, + ) + .await?; + definitions.insert(table.name.to_string(), defns); + } + Ok(Self::new(definitions)) + } + + /// `table` of this list's deployment, with its indexes. + pub(crate) fn source_table<'a>(&'a self, table: &'a Table) -> SourceTable<'a> { + let name = table.name.as_str(); + SourceTable { + table, + indexes: self.indexes.get(name).map(Vec::as_slice).unwrap_or(&[]), + definitions: self.definitions.get(name).map(Vec::as_slice).unwrap_or(&[]), } - Ok(Self::new(indexes)) } } @@ -1378,18 +1545,25 @@ mod tests { #[test] fn postponed_indexes_to_create_from_source() { - // The source indexes `email` with a different shape and doesn't - // index `name`. + // The source indexes `email` with a different shape, has a tuned + // `age` index and doesn't index `name`. let src_layout = test_layout( "type Thing @entity { id: ID! email: String! name: String! + age: Int! }", ); - let src_indexes = [CreateIndex::parse( - r#"CREATE INDEX attr_0_1_thing_email ON sgd0815.thing USING btree (email)"#.to_string(), - )]; + let src_indexes = IndexList::new(HashMap::from([( + "thing".to_string(), + vec![ + r#"CREATE INDEX attr_0_1_thing_email ON sgd0815.thing USING btree (email)"# + .to_string(), + "CREATE INDEX attr_0_3_thing_age ON sgd0815.thing USING btree (age) WITH (fillfactor='70')" + .to_string(), + ], + )])); let layout = test_layout(POSTPONED_GQL); let table = thing_table(&layout); @@ -1397,7 +1571,7 @@ mod tests { .postponed_indexes_to_create( &layout.input_schema, &[], - Some((thing_table(&src_layout), &src_indexes)), + Some(src_indexes.source_table(thing_table(&src_layout))), ) .unwrap(); assert_eq!(vec!["age", "data", "email"], sorted_columns(&to_create)); 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/relational/restore.rs b/store/postgres/src/relational/restore.rs index 081e6eebc3b..9860da0fd3b 100644 --- a/store/postgres/src/relational/restore.rs +++ b/store/postgres/src/relational/restore.rs @@ -309,6 +309,7 @@ pub async fn create_schema( site.clone(), &input_schema, entities_with_causality_region, + None, ) .await?; diff --git a/store/postgres/src/subgraph_store.rs b/store/postgres/src/subgraph_store.rs index 77ef932bbbd..6ce444f1094 100644 --- a/store/postgres/src/subgraph_store.rs +++ b/store/postgres/src/subgraph_store.rs @@ -29,10 +29,10 @@ use graph::{ }, internal_error, prelude::{ - ApiVersion, BlockNumber, BlockPtr, ChainStore, DeploymentHash, EntityOperation, Logger, - MetricsRegistry, NodeId, PartialBlockPtr, StoreError, StoreEvent, SubgraphDeploymentEntity, - SubgraphName, SubgraphStore as SubgraphStoreTrait, SubgraphVersionSwitchingMode, anyhow, - lazy_static, o, + ApiVersion, BlockNumber, BlockPtr, ChainStore, DeploymentHash, ENV_VARS, EntityOperation, + Logger, MetricsRegistry, NodeId, PartialBlockPtr, StoreError, StoreEvent, + SubgraphDeploymentEntity, SubgraphName, SubgraphStore as SubgraphStoreTrait, + SubgraphVersionSwitchingMode, anyhow, lazy_static, o, }, schema::{ApiSchema, InputSchema}, url::Url, @@ -341,7 +341,7 @@ impl SubgraphStore { self.evict(schema.id())?; let graft_base = deployment.graft_base.as_ref(); - let (site, deployment_store, node_id) = { + let (site, deployment_store, node_id, base_layout) = { // We need to deal with two situations: // (1) We are really creating a new subgraph; it therefore needs // to go in the shard and onto the node that the placement @@ -366,6 +366,7 @@ impl SubgraphStore { .get(&site.shard) .ok_or_else(|| StoreError::UnknownShard(site.shard.to_string()))?; + let mut validated_base = None; if let Some(graft_base) = graft_base { // Perform schema compatibility validation when: // - the site was just created (first-time deployment), or @@ -406,15 +407,35 @@ impl SubgraphStore { pconn .record_active_copy(base_layout.site.as_ref(), site.as_ref()) .await?; + validated_base = Some(base_layout); } } - (site, deployment_store, node_id) + (site, deployment_store, node_id, validated_base) + }; + + let src = match base_layout { + Some(base_layout) if ENV_VARS.store.copy_mirror_indexes => { + let indexes = self + .load_copy_source(base_layout.site.cheap_clone()) + .await? + .list; + Some((base_layout, indexes)) + } + _ => None, }; // 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, + src.as_ref() + .map(|(layout, indexes)| (layout.as_ref(), indexes)), + ) .await?; // FIXME: This simultaneously holds a `primary_conn` and a shard connection, which can @@ -898,6 +919,11 @@ impl Inner { .get(&shard) .ok_or_else(|| StoreError::UnknownShard(shard.to_string()))?; + let src_indexes = if ENV_VARS.store.copy_mirror_indexes { + Some(src_store.load_copy_source(src.cheap_clone()).await?.list) + } else { + None + }; deployment_store .create_deployment( &src_layout.input_schema, @@ -905,6 +931,9 @@ impl Inner { dst.clone(), false, on_sync, + src_indexes + .as_ref() + .map(|indexes| (src_layout.as_ref(), indexes)), ) .await?; @@ -1420,10 +1449,13 @@ impl Inner { name: &str, table: &str, expr: &str, + unique: bool, ) -> 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 + store + .create_btree_index(&site, name, table, expr, unique) + .await } pub async fn drop_index_for_deployment( diff --git a/store/test-store/tests/postgres/graft.rs b/store/test-store/tests/postgres/graft.rs index 1e917c4b14b..4e6814ba1cb 100644 --- a/store/test-store/tests/postgres/graft.rs +++ b/store/test-store/tests/postgres/graft.rs @@ -1,6 +1,6 @@ use graph::blockchain::block_stream::FirehoseCursor; use graph::schema::InputSchema; -use graph_store_postgres::command_support::OnSync; +use graph_store_postgres::command_support::{OnSync, index::Method}; use lazy_static::lazy_static; use std::{marker::PhantomData, str::FromStr}; use test_store::*; @@ -741,7 +741,7 @@ fn late_build_skips_columns_indexed_under_other_names() { )); 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") + .create_btree_index_for_deployment(&dst, "attr_9_9_user_age", "user", "age", false) .await?; test_store::start_subgraph(&dst).await?; store.set_postponed_indexes_created(&dst, false).await?; @@ -793,6 +793,7 @@ fn graft_resumes_after_indexes_copied_under_source_names() { "attr_0_3_user_email", "user", r#"left("email", 256)"#, + false, ) .await?; @@ -970,6 +971,183 @@ fn graft_keeps_dropped_postponed_index() { }) } +// A graft ignores an invalid index on its source. +#[test] +fn graft_ignores_invalid_source_index() { + run_test(|store, src| async move { + build_postponed_indexes(&store, &src).await?; + store + .drop_index_for_deployment(&src, "attr_0_4_user_age") + .await?; + // Fails on the two versions of user 3, both aged 28, and leaves an + // invalid index behind. + let res = store + .create_btree_index_for_deployment(&src, "attr_0_4_user_age", "user", "age", true) + .await; + assert!(res.is_err()); + + let dst_id = DeploymentHash::new("grafted_invalid").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_2_user_email", + "attr_0_4_user_favorite_color", + "attr_0_5_user_added", + ], + postponed_attr_indexes(&store, &dst, USER).await? + ); + Ok(()) + }) +} + +// A graft has the source's manual indexes and not the ones it dropped. +#[test] +fn graft_mirrors_source_indexes() { + run_test(|store, src| async move { + store + .drop_index_for_deployment(&src, "attr_0_0_user_id") + .await?; + store + .create_manual_index(&src, USER, vec!["email".to_string()], Method::BTree, None) + .await?; + + let dst_id = DeploymentHash::new("grafted_mirror").unwrap(); + let dst = create_grafted_subgraph( + &dst_id, + GRAFT_GQL, + TEST_SUBGRAPH_ID.as_str(), + BLOCKS[1].clone(), + ) + .await?; + + let names: Vec = store + .indexes_for_entity(&dst, USER) + .await? + .into_iter() + .filter_map(|idx| idx.name().map(str::to_string)) + .collect(); + assert!(!names.iter().any(|n| n == "attr_0_0_user_id"), "{names:?}"); + assert!(names.iter().any(|n| n == "manual_user_email"), "{names:?}"); + Ok(()) + }) +} + +// A source index that can't be created on the graft doesn't stop the graft +// or the copies after it. +#[test] +fn graft_skips_source_index_that_cannot_be_created() { + run_test(|store, src| async move { + // The graft has a table named `pet`. + store + .create_btree_index_for_deployment(&src, "pet", "user", "name", false) + .await?; + store + .create_btree_index_for_deployment(&src, "zz_user_email", "user", "email", false) + .await?; + + let dst_id = DeploymentHash::new("grafted_failing").unwrap(); + let dst = create_grafted_subgraph( + &dst_id, + GRAFT_NEW_TABLE_GQL, + TEST_SUBGRAPH_ID.as_str(), + BLOCKS[1].clone(), + ) + .await?; + + let names: Vec = store + .indexes_for_entity(&dst, USER) + .await? + .into_iter() + .filter_map(|idx| idx.name().map(str::to_string)) + .collect(); + assert!(!names.iter().any(|n| n == "pet"), "{names:?}"); + assert!(names.iter().any(|n| n == "zz_user_email"), "{names:?}"); + Ok(()) + }) +} + +// A graft that changes a table's mutability gets the layout's postponed +// indexes for it, even ones the source dropped. +#[test] +fn graft_with_changed_mutability_gets_layout_indexes() { + 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_immutable").unwrap(); + let dst = create_grafted_subgraph( + &dst_id, + GRAFT_IMMUTABLE_GQL, + TEST_SUBGRAPH_ID.as_str(), + BLOCKS[0].clone(), + ) + .await?; + + assert_eq!( + vec![ + "attr_0_0_user_name", + "attr_0_1_user_email", + "attr_0_2_user_age", + "attr_0_3_user_favorite_color", + "attr_0_4_user_added", + ], + postponed_attr_indexes(&store, &dst, USER).await? + ); + Ok(()) + }) +} + +// Same as `graft_mirrors_source_indexes`, for a copy to another shard. +// Needs a test configuration with at least two shards. +#[test] +fn copy_mirrors_source_indexes() { + run_test(|store, src| async move { + let Some(dst_shard) = other_shard(&store, &src).await? else { + return Ok(()); + }; + store + .drop_index_for_deployment(&src, "attr_0_0_user_id") + .await?; + store + .create_manual_index(&src, USER, vec!["email".to_string()], Method::BTree, None) + .await?; + + let dst = store + .copy_deployment( + &src, + dst_shard, + NODE_ID.clone(), + BLOCKS[1].clone(), + OnSync::None, + ) + .await?; + test_store::start_subgraph(&dst).await?; + // `indexes_for_entity` looks up the active deployment for a hash. + store.activate(&dst).await?; + + let names: Vec = store + .indexes_for_entity(&dst, USER) + .await? + .into_iter() + .filter_map(|idx| idx.name().map(str::to_string)) + .collect(); + assert!(!names.iter().any(|n| n == "attr_0_0_user_id"), "{names:?}"); + assert!(names.iter().any(|n| n == "manual_user_email"), "{names:?}"); + Ok(()) + }) +} + // A graft builds the postponed indexes of entity types its source lacks. #[test] fn graft_new_table_gets_postponed_indexes() { diff --git a/store/test-store/tests/postgres/relational.rs b/store/test-store/tests/postgres/relational.rs index c4946b408c5..4d21aa5532a 100644 --- a/store/test-store/tests/postgres/relational.rs +++ b/store/test-store/tests/postgres/relational.rs @@ -514,7 +514,7 @@ async fn create_schema(conn: &mut AsyncPgConnection) -> Layout { let query = format!("create schema {}", NAMESPACE.as_str()); conn.batch_execute(&query).await.unwrap(); - Layout::create_relational_schema(conn, Arc::new(site), &schema, BTreeSet::new()) + Layout::create_relational_schema(conn, Arc::new(site), &schema, BTreeSet::new(), None) .await .expect("Failed to create relational schema") } @@ -667,6 +667,7 @@ async fn copy_populates_fulltext() { Arc::new(dst_site), &THINGS_SCHEMA, BTreeSet::new(), + None, ) .await .expect("Failed to create destination relational schema"); diff --git a/store/test-store/tests/postgres/relational_bytes.rs b/store/test-store/tests/postgres/relational_bytes.rs index 1d608114eac..5ab5bb042c9 100644 --- a/store/test-store/tests/postgres/relational_bytes.rs +++ b/store/test-store/tests/postgres/relational_bytes.rs @@ -170,7 +170,7 @@ async fn create_schema(conn: &mut AsyncPgConnection) -> Layout { NAMESPACE.clone(), NETWORK_NAME.to_string(), ); - Layout::create_relational_schema(conn, Arc::new(site), &schema, BTreeSet::new()) + Layout::create_relational_schema(conn, Arc::new(site), &schema, BTreeSet::new(), None) .await .expect("Failed to create relational schema") }