feat: implement postgres kvbackend (#4421)

.github/actions/setup-postgres-cluster/action.yml

@@ -0,0 +1,30 @@
+name: Setup PostgreSQL
+description: Deploy PostgreSQL on Kubernetes
+inputs:
+ postgres-replicas:
+ default: 1
+ description: "Number of PostgreSQL replicas"
+ namespace:
+ default: "postgres-namespace"
+ postgres-version:
+ default: "14.2"
+ description: "PostgreSQL version"
+ storage-size:
+ default: "1Gi"
+ description: "Storage size for PostgreSQL"
+
+runs:
+ using: composite
+ steps:
+ - name: Install PostgreSQL
+ shell: bash
+ run: |
+ helm upgrade \
+ --install postgresql oci://registry-1.docker.io/bitnamicharts/postgresql \
+ --set replicaCount=${{ inputs.postgres-replicas }} \
+ --set image.tag=${{ inputs.postgres-version }} \
+ --set persistence.size=${{ inputs.storage-size }} \
+ --set postgresql.username=greptimedb \
+ --set postgresql.password=admin \
+ --create-namespace \
+ -n ${{ inputs.namespace }}

.github/workflows/develop.yml

@@ -338,6 +338,8 @@ jobs:
  uses: ./.github/actions/setup-kafka-cluster
  - name: Setup Etcd cluser
  uses: ./.github/actions/setup-etcd-cluster
+ - name: Setup Postgres cluser
+ uses: ./.github/actions/setup-postgres-cluster
  # Prepares for fuzz tests
  - uses: arduino/setup-protoc@v3
  with:
@@ -476,6 +478,8 @@ jobs:
  uses: ./.github/actions/setup-kafka-cluster
  - name: Setup Etcd cluser
  uses: ./.github/actions/setup-etcd-cluster
+ - name: Setup Postgres cluser
+ uses: ./.github/actions/setup-postgres-cluster
  # Prepares for fuzz tests
  - uses: arduino/setup-protoc@v3
  with:
@@ -702,6 +706,9 @@ jobs:
  - name: Setup minio
  working-directory: tests-integration/fixtures/minio
  run: docker compose -f docker-compose-standalone.yml up -d --wait
+ - name: Setup postgres server
+ working-directory: tests-integration/fixtures/postgres
+ run: docker compose -f docker-compose-standalone.yml up -d --wait
  - name: Run nextest cases
  run: cargo llvm-cov nextest --workspace --lcov --output-path lcov.info -F pyo3_backend -F dashboard
  env:
@@ -718,6 +725,7 @@ jobs:
  GT_MINIO_REGION: us-west-2
  GT_MINIO_ENDPOINT_URL: http://127.0.0.1:9000
  GT_ETCD_ENDPOINTS: http://127.0.0.1:2379
+ GT_POSTGRES_ENDPOINTS: postgres://greptimedb:[email protected]:5432/postgres
  GT_KAFKA_ENDPOINTS: 127.0.0.1:9092
  GT_KAFKA_SASL_ENDPOINTS: 127.0.0.1:9093
  UNITTEST_LOG_DIR: "__unittest_logs"

Cargo.toml

@@ -174,6 +174,7 @@ sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "5
 strum = { version = "0.25", features = ["derive"] }
 tempfile = "3"
 tokio = { version = "1.36", features = ["full"] }
+tokio-postgres = "0.7"
 tokio-stream = { version = "0.1" }
 tokio-util = { version = "0.7", features = ["io-util", "compat"] }
 toml = "0.8.8"

src/cmd/src/metasrv.rs

@@ -22,6 +22,7 @@ use common_telemetry::info;
 use common_telemetry::logging::TracingOptions;
 use common_version::{short_version, version};
 use meta_srv::bootstrap::MetasrvInstance;
+use meta_srv::metasrv::BackendImpl;
 use snafu::ResultExt;
 use tracing_appender::non_blocking::WorkerGuard;
 
@@ -137,6 +138,9 @@ struct StartCommand {
  /// The max operations per txn
  #[clap(long)]
  max_txn_ops: Option<usize>,
+ /// The database backend.
+ #[clap(long, value_enum)]
+ backend: Option<BackendImpl>,
 }
 
 impl StartCommand {
@@ -219,6 +223,12 @@ impl StartCommand {
  opts.max_txn_ops = max_txn_ops;
  }
 
+ if let Some(backend) = &self.backend {
+ opts.backend.clone_from(backend);
+ } else {
+ opts.backend = BackendImpl::default()
+ }
+
  // Disable dashboard in metasrv.
  opts.http.disable_dashboard = true;
 

src/common/meta/Cargo.toml

@@ -6,6 +6,7 @@ license.workspace = true
 
 [features]
 testing = []
+pg_kvbackend = ["dep:tokio-postgres"]
 
 [lints]
 workspace = true
@@ -56,6 +57,7 @@ store-api.workspace = true
 strum.workspace = true
 table.workspace = true
 tokio.workspace = true
+tokio-postgres = { workspace = true, optional = true }
 tonic.workspace = true
 typetag = "0.2"
 

src/common/meta/src/error.rs

@@ -643,6 +643,15 @@ pub enum Error {
  location: Location,
  },
 
+ #[snafu(display("Failed to parse {} from str to utf8", name))]
+ StrFromUtf8 {
+ name: String,
+ #[snafu(source)]
+ error: std::str::Utf8Error,
+ #[snafu(implicit)]
+ location: Location,
+ },
+
  #[snafu(display("Value not exists"))]
  ValueNotExist {
  #[snafu(implicit)]
@@ -651,6 +660,24 @@ pub enum Error {
 
  #[snafu(display("Failed to get cache"))]
  GetCache { source: Arc<Error> },
+
+ #[cfg(feature = "pg_kvbackend")]
+ #[snafu(display("Failed to execute via Postgres"))]
+ PostgresExecution {
+ #[snafu(source)]
+ error: tokio_postgres::Error,
+ #[snafu(implicit)]
+ location: Location,
+ },
+
+ #[cfg(feature = "pg_kvbackend")]
+ #[snafu(display("Failed to connect to Postgres"))]
+ ConnectPostgres {
+ #[snafu(source)]
+ error: tokio_postgres::Error,
+ #[snafu(implicit)]
+ location: Location,
+ },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -704,7 +731,8 @@ impl ErrorExt for Error {
  | UnexpectedLogicalRouteTable { .. }
  | ProcedureOutput { .. }
  | FromUtf8 { .. }
- | MetadataCorruption { .. } => StatusCode::Unexpected,
+ | MetadataCorruption { .. }
+ | StrFromUtf8 { .. } => StatusCode::Unexpected,
 
  SendMessage { .. } | GetKvCache { .. } | CacheNotGet { .. } | RenameTable { .. } => {
  StatusCode::Internal
@@ -749,6 +777,11 @@ impl ErrorExt for Error {
  | ParseNum { .. }
  | InvalidRole { .. }
  | EmptyDdlTasks { .. } => StatusCode::InvalidArguments,
+
+ #[cfg(feature = "pg_kvbackend")]
+ PostgresExecution { .. } => StatusCode::Internal,
+ #[cfg(feature = "pg_kvbackend")]
+ ConnectPostgres { .. } => StatusCode::Internal,
  }
  }
 

src/common/meta/src/kv_backend.rs

@@ -31,6 +31,8 @@ use crate::rpc::KeyValue;
 pub mod chroot;
 pub mod etcd;
 pub mod memory;
+#[cfg(feature = "pg_kvbackend")]
+pub mod postgres;
 pub mod test;
 pub mod txn;