store: create a copy's attribute indexes from the source's index list

store/postgres/src/deployment_store.rs

@@ -176,13 +176,17 @@ impl DeploymentStore {
         DeploymentStore(Arc::new(store))
     }
 
+    /// `index_def` is the index list of the deployment this one is copied or
+    /// grafted from. When it is given, the destination's schema is created
+    /// with the source's indexes rather than the schema's default set.
     pub(crate) async fn create_deployment(
         &self,
         schema: &InputSchema,
         deployment: DeploymentCreate,
         site: Arc<Site>,
         replace: bool,
         on_sync: OnSync,
+        index_def: Option<&IndexList>,
     ) -> Result<(), StoreError> {
         let mut conn = self.pool.get_permitted().await?;
         conn.transaction::<_, StoreError, _>(async |conn| {
@@ -210,11 +214,12 @@ impl DeploymentStore {
                 let query = format!("create schema {}", &site.namespace);
                 conn.batch_execute(&query).await?;
 
-                let _ = Layout::create_relational_schema(
+                let _ = Layout::create_relational_schema_with_source_indexes(
                     conn,
                     site.clone(),
                     schema,
                     entities_with_causality_region.into_iter().collect(),
+                    index_def,
                 )
                 .await?;
 

store/postgres/src/relational.rs

@@ -56,6 +56,7 @@ use std::str::FromStr;
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
+use crate::relational::index::IndexList;
 use crate::relational::value::{FromOidRow, OidRow};
 use crate::relational_queries::{
     ConflictingEntitiesData, ConflictingEntitiesQuery, EntityDataExt, FindChangesQuery,
@@ -381,12 +382,33 @@ impl Layout {
         site: Arc<Site>,
         schema: &InputSchema,
         entities_with_causality_region: BTreeSet<EntityType>,
+    ) -> Result<Layout, StoreError> {
+        Self::create_relational_schema_with_source_indexes(
+            conn,
+            site,
+            schema,
+            entities_with_causality_region,
+            None,
+        )
+        .await
+    }
+
+    /// Like `create_relational_schema`, but the indexes that are not postponed
+    /// are taken from `source_indexes`. A copy or graft passes the source
+    /// deployment's index list so the destination ends up with the indexes the
+    /// source actually carries, not the schema's default set.
+    pub async fn create_relational_schema_with_source_indexes(
+        conn: &mut AsyncPgConnection,
+        site: Arc<Site>,
+        schema: &InputSchema,
+        entities_with_causality_region: BTreeSet<EntityType>,
+        source_indexes: Option<&IndexList>,
     ) -> Result<Layout, StoreError> {
         let catalog =
             Catalog::for_creation(conn, site.cheap_clone(), entities_with_causality_region).await?;
         let layout = Self::new(site, schema, catalog)?;
         let sql = layout
-            .as_ddl()
+            .as_ddl_with_source_indexes(source_indexes)
             .map_err(|_| StoreError::Unknown(anyhow!("failed to generate DDL for layout")))?;
         conn.batch_execute(&sql).await?;
         Ok(layout)

store/postgres/src/relational/ddl.rs

@@ -9,7 +9,10 @@ use crate::relational::{
     BLOCK_COLUMN, BLOCK_RANGE_COLUMN, BYTE_ARRAY_PREFIX_SIZE, ColumnType, STRING_PREFIX_SIZE,
     VID_COLUMN,
 };
-use crate::{block_range::CAUSALITY_REGION_COLUMN, relational::index::Cond};
+use crate::{
+    block_range::CAUSALITY_REGION_COLUMN,
+    relational::index::{Cond, IndexList},
+};
 
 use super::{
     Column, Layout, SqlName, Table,
@@ -30,6 +33,17 @@ impl Layout {
     /// See the unit tests at the end of this file for the actual DDL that
     /// gets generated
     pub fn as_ddl(&self) -> Result<String, fmt::Error> {
+        self.as_ddl_with_source_indexes(None)
+    }
+
+    /// Like `as_ddl`, but takes the indexes that are not postponed from
+    /// `source_indexes` instead of from this layout. Callers that create the
+    /// schema for a copy or graft pass the source deployment's index list so
+    /// the destination ends up with the indexes the source actually has.
+    pub fn as_ddl_with_source_indexes(
+        &self,
+        source_indexes: Option<&IndexList>,
+    ) -> Result<String, fmt::Error> {
         let mut out = String::new();
 
         // Output enums first so table definitions can reference them
@@ -42,7 +56,7 @@ impl Layout {
         // Output 'create table' statements for all tables
         let creat = self.index_creator(false, false);
         for table in tables {
-            table.as_ddl(&self.input_schema, &creat, &mut out)?;
+            table.as_ddl(&self.input_schema, &creat, source_indexes, &mut out)?;
         }
 
         Ok(out)
@@ -410,14 +424,32 @@ impl Table {
     ///
     /// See the unit tests at the end of this file for the actual DDL that
     /// gets generated
+    /// `as_ddl` writes the table and the indexes that do not wait for the data
+    /// to be loaded.
+    ///
+    /// `source_indexes` is the index list of the deployment this one is copied
+    /// or grafted from. When it is given, the attribute indexes are taken from
+    /// the source instead of the layout, so that a copy carries the source's
+    /// index set rather than the schema's default one. The time-travel and
+    /// aggregate indexes are always created from the layout: they are part of
+    /// the schema every deployment needs, not something an operator drops.
     pub(crate) fn as_ddl(
         &self,
         schema: &InputSchema,
         creat: &IndexCreator,
+        source_indexes: Option<&IndexList>,
         out: &mut String,
     ) -> fmt::Result {
         self.create_table(out)?;
-        for idx in self.indexes(schema)? {
+
+        let mut indexes = self.time_travel_indexes();
+        match source_indexes {
+            Some(source) => indexes.extend(source.indexes_for_copy_table(self).cloned()),
+            None => self.add_attribute_indexes(&mut indexes),
+        }
+        self.add_aggregate_indexes(schema, &mut indexes)?;
+
+        for idx in indexes {
             if !idx.to_postpone() {
                 writeln!(out, "{};", creat.to_sql(&idx)?)?;
             }

store/postgres/src/relational/ddl_tests.rs

@@ -3,6 +3,7 @@ use pretty_assertions::assert_eq;
 
 use super::*;
 
+use crate::relational::index::CreateIndex;
 use crate::{deployment_store::generate_index_creation_sql, layout_for_tests::make_dummy_site};
 
 const ID_TYPE: ColumnType = ColumnType::String;
@@ -888,3 +889,63 @@ type Data @entity(immutable: true) {
     block: Block!
 }
 "#;
+
+/// The copy and graft paths create the destination from the source's index
+/// list, so a deployment whose indexes were dropped before copying does not
+/// come back with the schema's default set. `as_ddl` on its own still writes
+/// the layout's indexes, which is what a fresh deployment needs.
+#[test]
+fn source_indexes_replace_the_layout_indexes_on_creation() {
+    let layout = test_layout(THING_GQL);
+    let namespace = layout.site.namespace.as_str().to_string();
+
+    // `thing` carries a gist attribute index in this schema. A source that
+    // does not have it must not get it on the copy, even though the layout
+    // describes it.
+    let gist_on_thing =
+        format!("create index attr_0_1_thing_big_thing on \"{namespace}\".\"thing\" using gist");
+    let layout_ddl = layout.as_ddl().expect("failed to generate DDL");
+    assert!(
+        layout_ddl.contains(&gist_on_thing),
+        "the layout is expected to describe a gist index for `thing`:\n{layout_ddl}"
+    );
+
+    let from_empty = layout
+        .as_ddl_with_source_indexes(Some(&IndexList::new(HashMap::new())))
+        .expect("failed to generate DDL");
+    assert!(
+        !from_empty.contains(&gist_on_thing),
+        "a source without that index must not produce it:\n{from_empty}"
+    );
+
+    // The time-travel indexes are not an operator's to drop: they describe
+    // how history is stored, so they are written regardless of the source.
+    for always in ["brin_thing", "thing_block_range_closed"] {
+        assert!(
+            from_empty.contains(always),
+            "`{always}` is part of the schema and must still be created:\n{from_empty}"
+        );
+    }
+
+    // An attribute index the source does carry is kept. This is the other
+    // direction of the same check: reading the source must not silently drop
+    // what it has.
+    let table = layout.table(&SqlName::from("Thing")).unwrap();
+    let kept = table
+        .indexes(&layout.input_schema)
+        .unwrap()
+        .into_iter()
+        .find(|idx| {
+            matches!(idx, CreateIndex::Parsed { name, .. }
+                if *name == "attr_0_1_thing_big_thing")
+        })
+        .expect("the schema is expected to describe the gist index");
+    let source = IndexList::new(HashMap::from([(table.name.to_string(), vec![kept])]));
+    let from_source = layout
+        .as_ddl_with_source_indexes(Some(&source))
+        .expect("failed to generate DDL");
+    assert!(
+        from_source.contains(&gist_on_thing),
+        "an index the source carries must be created:\n{from_source}"
+    );
+}

store/postgres/src/relational/index.rs

@@ -946,6 +946,28 @@ impl IndexList {
                 && !(ci.is_id() && table.immutable)
         })
     }
+
+    /// The indexes to create up front when a deployment is copied or grafted
+    /// from this one.
+    ///
+    /// Unlike `indexes_for_table`, this keeps the attribute indexes that
+    /// `indexes_for_table` drops. Those are dropped there because the layout
+    /// creates them anyway, which does not hold when the attribute indexes are
+    /// taken from the source: a source that carries `attr_N_0_<table>_id`
+    /// must keep it on the copy. The indexes the layout always creates
+    /// (`brin_*` and the closed-range ones) are dropped here for the same
+    /// reason they are dropped there.
+    pub fn indexes_for_copy_table<'a>(
+        &'a self,
+        table: &'a Table,
+    ) -> impl Iterator<Item = &'a CreateIndex> {
+        static EMPTY: Vec<CreateIndex> = vec![];
+        let indexes = self.indexes.get(table.name.as_str()).unwrap_or(&EMPTY);
+
+        indexes
+            .iter()
+            .filter(move |ci| ci.fields_exist_in_dest(table) && !ci.is_default_non_attr_index())
+    }
 }
 
 #[cfg(test)]

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?;
 

store/postgres/src/subgraph_store.rs

@@ -416,7 +416,14 @@ impl SubgraphStore {
 
         // Create the actual databases schema and metadata entries
         deployment_store
-            .create_deployment(schema, deployment, site.clone(), replace, OnSync::None)
+            .create_deployment(
+                schema,
+                deployment,
+                site.clone(),
+                replace,
+                OnSync::None,
+                None,
+            )
             .await?;
 
         // FIXME: This simultaneously holds a `primary_conn` and a shard connection, which can
@@ -900,13 +907,19 @@ impl Inner {
             .get(&shard)
             .ok_or_else(|| StoreError::UnknownShard(shard.to_string()))?;
 
+        // The destination gets the source's index set, not the schema's
+        // default one: an operator who dropped indexes from the source to
+        // save space should not get them all back on every copy.
+        let src_indexes = src_store.load_indexes(src.cheap_clone()).await?;
+
         deployment_store
             .create_deployment(
                 &src_layout.input_schema,
                 deployment,
                 dst.clone(),
                 false,
                 on_sync,
+                Some(&src_indexes),
             )
             .await?;