From ae29fb4aedc00c68eea48af0980b4406be9f6153 Mon Sep 17 00:00:00 2001 From: Maksim Dimitrov Date: Mon, 9 Nov 2026 16:42:16 +0200 Subject: [PATCH 1/2] store: Ignore invalid indexes of a copy source A copy took every index of the source into account, including invalid ones. An index is invalid when a concurrent build or drop of it failed or was interrupted, and while a concurrent build is still running. A copy rebuilt postponed indexes that an operator was dropping, or whose build had failed on the source. A copy now reads only the source's valid indexes. Dumps still read all of them, since they record the schema as it is. --- store/postgres/src/deployment_store.rs | 12 ++++--- store/postgres/src/relational/index.rs | 23 +++++++++---- store/postgres/src/subgraph_store.rs | 5 ++- store/test-store/tests/postgres/graft.rs | 41 +++++++++++++++++++++++- 4 files changed, 68 insertions(+), 13 deletions(-) diff --git a/store/postgres/src/deployment_store.rs b/store/postgres/src/deployment_store.rs index ab589d95577..2b7c14e9c3a 100644 --- a/store/postgres/src/deployment_store.rs +++ b/store/postgres/src/deployment_store.rs @@ -725,7 +725,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 +758,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 +934,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, diff --git a/store/postgres/src/relational/index.rs b/store/postgres/src/relational/index.rs index ae399513788..a962e7ef676 100644 --- a/store/postgres/src/relational/index.rs +++ b/store/postgres/src/relational/index.rs @@ -901,16 +901,25 @@ impl IndexList { IndexList { indexes } } - pub async fn load(conn: &mut AsyncPgConnection, layout: &Layout) -> Result { + /// 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 indexes = 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(); + let indexes_from_table = catalog::indexes_for_table( + conn, + schema_name.as_str(), + table.name.as_str(), + valid_only, + ) + .await? + .into_iter() + .map(CreateIndex::parse) + .collect(); indexes.insert(table.name.to_string(), indexes_from_table); } Ok(Self::new(indexes)) diff --git a/store/postgres/src/subgraph_store.rs b/store/postgres/src/subgraph_store.rs index 77ef932bbbd..5c4d940f220 100644 --- a/store/postgres/src/subgraph_store.rs +++ b/store/postgres/src/subgraph_store.rs @@ -1420,10 +1420,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..84898a1bbf2 100644 --- a/store/test-store/tests/postgres/graft.rs +++ b/store/test-store/tests/postgres/graft.rs @@ -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,44 @@ 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 builds the postponed indexes of entity types its source lacks. #[test] fn graft_new_table_gets_postponed_indexes() { From 2b2c5483da09d2ed5626f7b6881d740692adcbe8 Mon Sep 17 00:00:00 2001 From: Maksim Dimitrov Date: Mon, 9 Nov 2026 17:01:23 +0200 Subject: [PATCH 2/2] graph, store: Mirror source indexes on copy/graft Since #6434, a copy or graft created the indexes it gets with its tables from its own layout instead of from its source. Indexes dropped on the source came back, and manual indexes were lost. Before #6434, a copy followed its source only when postponement was on. A copy or graft now gets the indexes its source has, for each table the source has with the same mutability. graph-node's own indexes come from the layout, even where the source tuned them, and its attribute indexes are left out where the source has their columns but doesn't index them. A source index under a graph-node name is copied only when the table has its columns but doesn't index them, such as a GIN index on a list. The source's other indexes are copied with the statement Postgres prints for them, with only the schema name changed, each in its own savepoint. Those that can't be copied are logged with the reason. Tables that are new or changed mutability get the layout's indexes, also at the end of the copy. GRAPH_STORE_COPY_MIRROR_INDEXES, true by default, turns this off. Fixes #6722 --- docs/environment-variables.md | 7 + graph/src/env/store.rs | 7 + store/postgres/examples/layout.rs | 2 +- store/postgres/src/copy.rs | 19 +- store/postgres/src/deployment_store.rs | 43 ++- store/postgres/src/relational.rs | 3 +- store/postgres/src/relational/ddl.rs | 20 +- store/postgres/src/relational/ddl_tests.rs | 168 +++++++++++- store/postgres/src/relational/index.rs | 251 +++++++++++++++--- store/postgres/src/relational/prune.rs | 2 +- store/postgres/src/relational/restore.rs | 1 + store/postgres/src/subgraph_store.rs | 43 ++- store/test-store/tests/postgres/graft.rs | 141 +++++++++- store/test-store/tests/postgres/relational.rs | 3 +- .../tests/postgres/relational_bytes.rs | 2 +- 15 files changed, 625 insertions(+), 87 deletions(-) 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 2b7c14e9c3a..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() { @@ -2121,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 a962e7ef676..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,30 +878,178 @@ 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, + } } /// Load the indexes of `layout`, only the valid ones with `valid_only`. @@ -907,22 +1058,29 @@ impl IndexList { layout: &Layout, valid_only: bool, ) -> Result { - let mut indexes = HashMap::new(); + 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( + let defns = catalog::indexes_for_table( conn, schema_name.as_str(), table.name.as_str(), valid_only, ) - .await? - .into_iter() - .map(CreateIndex::parse) - .collect(); - indexes.insert(table.name.to_string(), indexes_from_table); + .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)) } } @@ -1387,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); @@ -1406,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 5c4d940f220..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?; diff --git a/store/test-store/tests/postgres/graft.rs b/store/test-store/tests/postgres/graft.rs index 84898a1bbf2..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::*; @@ -1009,6 +1009,145 @@ fn graft_ignores_invalid_source_index() { }) } +// 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") }