graph, store: Use a much larger fetch_size for fdw

Changing the fetch_size to 10,000 from its default of 100 will have a
dramatic effect on the time copy operations take.

The fetch_size can be controlled with the environment variable
GRAPH_STORE_FDW_FETCH_SIZE

Author: lutter

graph/src/env/store.rs

@@ -126,6 +126,9 @@ pub struct EnvVarsStore {
     /// used to work around Postgres errors complaining 'number of
     /// parameters must be between 0 and 65535' when inserting entities
     pub insert_extra_cols: usize,
+    /// The number of rows to fetch from the foreign data wrapper in one go,
+    /// this will be set as the option 'fetch_size' on all foreign servers
+    pub fdw_fetch_size: usize,
 }
 
 // This does not print any values avoid accidentally leaking any sensitive env vars
@@ -175,6 +178,7 @@ impl From<InnerStore> for EnvVarsStore {
             disable_block_cache_for_lookup: x.disable_block_cache_for_lookup,
             last_rollup_from_poi: x.last_rollup_from_poi,
             insert_extra_cols: x.insert_extra_cols,
+            fdw_fetch_size: x.fdw_fetch_size,
         }
     }
 }
@@ -238,6 +242,8 @@ pub struct InnerStore {
     last_rollup_from_poi: bool,
     #[envconfig(from = "GRAPH_STORE_INSERT_EXTRA_COLS", default = "0")]
     insert_extra_cols: usize,
+    #[envconfig(from = "GRAPH_STORE_FDW_FETCH_SIZE", default = "10000")]
+    fdw_fetch_size: usize,
 }
 
 #[derive(Clone, Copy, Debug)]

store/postgres/src/connection_pool.rs

@@ -150,7 +150,11 @@ impl ForeignServer {
             "\
         create server \"{name}\"
                foreign data wrapper postgres_fdw
-               options (host '{remote_host}', port '{remote_port}', dbname '{remote_db}', updatable 'false');
+               options (host '{remote_host}', \
+                        port '{remote_port}', \
+                        dbname '{remote_db}', \
+                        fetch_size '{fetch_size}', \
+                        updatable 'false');
         create user mapping
                for current_user server \"{name}\"
                options (user '{remote_user}', password '{remote_password}');",
@@ -160,6 +164,7 @@ impl ForeignServer {
             remote_db = self.dbname,
             remote_user = self.user,
             remote_password = self.password,
+            fetch_size = ENV_VARS.store.fdw_fetch_size,
         );
         Ok(conn.batch_execute(&query)?)
     }
@@ -178,17 +183,22 @@ impl ForeignServer {
         let query = format!(
             "\
         alter server \"{name}\"
-              options (set host '{remote_host}', {set_port} port '{remote_port}', set dbname '{remote_db}');
+              options (set host '{remote_host}', \
+                       {set_port} port '{remote_port}', \
+                       set dbname '{remote_db}, \
+                       {set_fetch_size} fetch_size '{fetch_size}');
         alter user mapping
               for current_user server \"{name}\"
               options (set user '{remote_user}', set password '{remote_password}');",
             name = self.name,
             remote_host = self.host,
             set_port = set_or_add("port"),
+            set_fetch_size = set_or_add("fetch_size"),
             remote_port = self.port,
             remote_db = self.dbname,
             remote_user = self.user,
             remote_password = self.password,
+            fetch_size = ENV_VARS.store.fdw_fetch_size,
         );
         Ok(conn.batch_execute(&query)?)
     }