IndexStorage implementation for postgres (#2954)

* feat: initial implementation

* feat: use db migrations

* feat: drop_prefix improvements

* fix: issues introduced by merge with main

* Address PR comments

---------

Co-authored-by: kmatasfp <33095685+kmatas@users.noreply.github.com>

Author: kmatasfp

Cargo.lock

@@ -64,6 +64,7 @@ http-body-util = { workspace = true }
 humansize = { workspace = true }
 humantime-serde = { workspace = true }
 hyper = { workspace = true }
+include_dir = { workspace = true }
 itertools = { workspace = true }
 lazy_static = { workspace = true }
 log = { workspace = true }

golem-worker-executor/Cargo.toml

@@ -0,0 +1,15 @@
+CREATE TABLE index_storage (
+    namespace TEXT NOT NULL,
+    key TEXT NOT NULL,
+    id BIGINT NOT NULL,
+    value BYTEA NOT NULL,
+    PRIMARY KEY (namespace, key, id)
+);
+
+ALTER TABLE index_storage SET (
+    autovacuum_vacuum_scale_factor = 0.01,
+    autovacuum_vacuum_threshold = 1024,
+    autovacuum_analyze_scale_factor = 0.02,
+    autovacuum_analyze_threshold = 1024,
+    autovacuum_vacuum_cost_limit = 2000
+);

golem-worker-executor/db/migration/postgres/001_init.sql

@@ -65,6 +65,7 @@ use crate::services::{
     HasWorkerActivator, HasWorkerService,
 };
 use crate::storage::indexed::multi_sqlite::MultiSqliteIndexedStorage;
+use crate::storage::indexed::postgres::PostgresIndexedStorage;
 use crate::storage::indexed::redis::RedisIndexedStorage;
 use crate::storage::indexed::sqlite::SqliteIndexedStorage;
 use crate::storage::indexed::IndexedStorage;
@@ -391,6 +392,13 @@ pub async fn create_worker_executor_impl<Ctx: WorkerCtx, A: Bootstrap<Ctx> + ?Si
             let pool = RedisPool::configured(redis).await?;
             Arc::new(RedisIndexedStorage::new(pool.clone()))
         }
+        IndexedStorageConfig::Postgres(postgres) => {
+            Arc::new(
+                PostgresIndexedStorage::configured(postgres)
+                    .await
+                    .map_err(|err| anyhow!(err))?,
+            )
+        }
         IndexedStorageConfig::KVStoreSqlite(_) => {
             let sqlite = sqlite
                 .clone()

golem-worker-executor/src/lib.rs

@@ -16,7 +16,7 @@ use anyhow::Context;
 use figment::providers::{Format, Toml};
 use figment::Figment;
 use golem_common::config::{
-    ConfigExample, ConfigLoader, DbSqliteConfig, HasConfigExamples, RedisConfig,
+    ConfigExample, ConfigLoader, DbPostgresConfig, DbSqliteConfig, HasConfigExamples, RedisConfig,
 };
 use golem_common::model::base64::Base64;
 use golem_common::model::RetryConfig;
@@ -607,6 +607,7 @@ impl SafeDisplay for KeyValueStorageInMemoryConfig {
 pub enum IndexedStorageConfig {
     KVStoreRedis(IndexedStorageKVStoreRedisConfig),
     Redis(RedisConfig),
+    Postgres(IndexedStoragePostgresConfig),
     KVStoreSqlite(IndexedStorageKVStoreSqliteConfig),
     KVStoreMultiSqlite(IndexedStorageKVStoreMultiSqliteConfig),
     Sqlite(DbSqliteConfig),
@@ -626,6 +627,10 @@ impl SafeDisplay for IndexedStorageConfig {
                 let _ = writeln!(&mut result, "redis:");
                 let _ = writeln!(&mut result, "{}", inner.to_safe_string_indented());
             }
+            IndexedStorageConfig::Postgres(inner) => {
+                let _ = writeln!(&mut result, "postgres:");
+                let _ = writeln!(&mut result, "{}", inner.to_safe_string_indented());
+            }
             IndexedStorageConfig::KVStoreSqlite(inner) => {
                 let _ = writeln!(&mut result, "sqlite kv-store:");
                 let _ = writeln!(&mut result, "{}", inner.to_safe_string_indented());
@@ -704,6 +709,31 @@ impl SafeDisplay for IndexedStorageInMemoryConfig {
     }
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct IndexedStoragePostgresConfig {
+    #[serde(flatten)]
+    pub postgres: DbPostgresConfig,
+    #[serde(default = "default_indexed_storage_postgres_drop_prefix_delete_batch_size")]
+    pub drop_prefix_delete_batch_size: u64,
+}
+
+impl SafeDisplay for IndexedStoragePostgresConfig {
+    fn to_safe_string(&self) -> String {
+        let mut result = String::new();
+        let _ = writeln!(&mut result, "{}", self.postgres.to_safe_string_indented());
+        let _ = writeln!(
+            &mut result,
+            "drop prefix delete batch size: {}",
+            self.drop_prefix_delete_batch_size
+        );
+        result
+    }
+}
+
+fn default_indexed_storage_postgres_drop_prefix_delete_batch_size() -> u64 {
+    1024
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct MemoryConfig {
     pub system_memory_override: Option<u64>,

golem-worker-executor/src/services/golem_config.rs

@@ -22,6 +22,7 @@ use golem_common::serialization::{deserialize, serialize};
 
 pub mod memory;
 pub mod multi_sqlite;
+pub mod postgres;
 pub mod redis;
 pub mod sqlite;