Repository: xataio/pgstream
Pull request #66 (2024-08-29 15:19:00)

Improve support for elasticsearch replication
This PR improves the support for elasticsearch replication, as well as improving the feature support for opensearch replication. Before this PR we were using a go-elasticsearch library version that supported opensearch (before it introduced version/client/server checks in 2021). This prevented any updates, and it had limitations when using vector fields with elasticsearch.

By separating the opensearch and elasticsearch clients we can address the slight differences for each in the index, mapping and overall library use. The separation is done at the library/pkg level, the rest of changes are meant to align with a decoupled approach, avoiding leaking implementation details. 

Main changes:
- Addition of elasticsearch client and creation of an abstracted search store library with the shared functionality, used as an access point for the rest of the code, abstracting the implementation details.
- Addition of a search store mapper, along with the opensearch and elasticsearch implementations, that takes care of the specificities of each regarding the mapping (ex., [`knn_vector` vs `dense_vector`](https://opster.com/guides/elasticsearch/machine-learning/vector-search-in-opensearch-vs-elasticsearch/)) .
- Update search batch processor to use the newly abstracted library
- Update configuration to be able to provide an elasticsearch or opensearch store url 
- Addition of integration tests to cover the postgres to elasticsearch integration

The commits have been split for ease of reviewing.


Patch:
diff --git a/README.md b/README.md
index 719d329..ad1b9b0 100644
--- a/README.md
+++ b/README.md
@@ -178,7 +178,8 @@ One of exponential/constant backoff policies can be provided for the Kafka commi
 
 | Environment Variable                                         | Default | Required | Description                                                                                                    |
 | ------------------------------------------------------------ | ------- | -------- | -------------------------------------------------------------------------------------------------------------- |
-| PGSTREAM_SEARCH_STORE_URL                                    | N/A     | Yes      | URL for the search store to connect to.                                                                        |
+| PGSTREAM_OPENSEARCH_STORE_URL                                | N/A     | Yes      | URL for the opensearch store to connect to (at least one of the URLs must be provided).                        |
+| PGSTREAM_ELASTICSEARCH_STORE_URL                             | N/A     | Yes      | URL for the elasticsearch store to connect to (at least one of the URLs must be provided).                     |
 | PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT                        | 1s      | No       | Max time interval at which the batch sending to the search store is triggered.                                 |
 | PGSTREAM_SEARCH_INDEXER_BATCH_SIZE                           | 100     | No       | Max number of messages to be sent per batch. When this size is reached, the batch is sent to the search store. |
 | PGSTREAM_SEARCH_INDEXER_MAX_QUEUE_BYTES                      | 100MiB  | No       | Max memory used by the search batch indexer for inflight batches.                                              |
diff --git a/cmd/config.go b/cmd/config.go
index 66a3ab1..2fab128 100644
--- a/cmd/config.go
+++ b/cmd/config.go
@@ -14,7 +14,7 @@ import (
 	kafkacheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/kafka"
 	kafkaprocessor "github.com/xataio/pgstream/pkg/wal/processor/kafka"
 	"github.com/xataio/pgstream/pkg/wal/processor/search"
-	"github.com/xataio/pgstream/pkg/wal/processor/search/opensearch"
+	"github.com/xataio/pgstream/pkg/wal/processor/search/store"
 	"github.com/xataio/pgstream/pkg/wal/processor/translator"
 	"github.com/xataio/pgstream/pkg/wal/processor/webhook/notifier"
 	"github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription/server"
@@ -148,8 +148,9 @@ func parseKafkaWriterConfig(kafkaServers []string, kafkaTopic string) *kafkaproc
 }
 
 func parseSearchProcessorConfig() *stream.SearchProcessorConfig {
-	searchStore := viper.GetString("PGSTREAM_SEARCH_STORE_URL")
-	if searchStore == "" {
+	opensearchStore := viper.GetString("PGSTREAM_OPENSEARCH_STORE_URL")
+	elasticsearchStore := viper.GetString("PGSTREAM_ELASTICSEARCH_STORE_URL")
+	if opensearchStore == "" && elasticsearchStore == "" {
 		return nil
 	}
 
@@ -160,8 +161,9 @@ func parseSearchProcessorConfig() *stream.SearchProcessorConfig {
 			MaxQueueBytes:  viper.GetInt64("PGSTREAM_SEARCH_INDEXER_MAX_QUEUE_BYTES"),
 			CleanupBackoff: parseBackoffConfig("PGSTREAM_SEARCH_INDEXER_CLEANUP"),
 		},
-		Store: opensearch.Config{
-			URL: searchStore,
+		Store: store.Config{
+			OpenSearchURL:    opensearchStore,
+			ElasticsearchURL: elasticsearchStore,
 		},
 		Retrier: search.StoreRetryConfig{
 			Backoff: parseBackoffConfig("PGSTREAM_SEARCH_STORE"),
diff --git a/go.mod b/go.mod
index 27dc787..a9313eb 100644
--- a/go.mod
+++ b/go.mod
@@ -4,7 +4,7 @@ go 1.22.2
 
 require (
 	github.com/cenkalti/backoff/v4 v4.2.1
-	github.com/elastic/go-elasticsearch/v8 v8.0.0-20210311100734-5d6b0c808457
+	github.com/elastic/go-elasticsearch/v8 v8.14.0
 	github.com/go-logr/zerologr v1.2.3
 	github.com/golang-migrate/migrate/v4 v4.17.1
 	github.com/google/go-cmp v0.6.0
@@ -13,6 +13,7 @@ require (
 	github.com/jackc/pgx/v5 v5.6.0
 	github.com/labstack/echo/v4 v4.12.0
 	github.com/mitchellh/mapstructure v1.5.0
+	github.com/opensearch-project/opensearch-go v1.1.0
 	github.com/pterm/pterm v0.12.79
 	github.com/rs/xid v1.5.0
 	github.com/rs/zerolog v1.32.0
@@ -20,7 +21,8 @@ require (
 	github.com/spf13/cobra v1.8.0
 	github.com/spf13/viper v1.18.2
 	github.com/stretchr/testify v1.9.0
-	github.com/testcontainers/testcontainers-go v0.31.0
+	github.com/testcontainers/testcontainers-go v0.33.0
+	github.com/testcontainers/testcontainers-go/modules/elasticsearch v0.33.0
 	github.com/testcontainers/testcontainers-go/modules/kafka v0.31.0
 	github.com/testcontainers/testcontainers-go/modules/opensearch v0.31.0
 	github.com/testcontainers/testcontainers-go/modules/postgres v0.31.0
@@ -34,17 +36,18 @@ require (
 	atomicgo.dev/schedule v0.1.0 // indirect
 	dario.cat/mergo v1.0.0 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
-	github.com/Microsoft/go-winio v0.6.1 // indirect
-	github.com/Microsoft/hcsshim v0.11.4 // indirect
+	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/containerd/console v1.0.3 // indirect
-	github.com/containerd/containerd v1.7.15 // indirect
+	github.com/containerd/containerd v1.7.18 // indirect
 	github.com/containerd/log v0.1.0 // indirect
+	github.com/containerd/platforms v0.2.1 // indirect
 	github.com/cpuguy83/dockercfg v0.3.1 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/distribution/reference v0.5.0 // indirect
-	github.com/docker/docker v25.0.6+incompatible // indirect
+	github.com/distribution/reference v0.6.0 // indirect
+	github.com/docker/docker v27.1.1+incompatible // indirect
 	github.com/docker/go-connections v0.5.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
+	github.com/elastic/elastic-transport-go/v8 v8.6.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
@@ -52,7 +55,6 @@ require (
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt v3.2.2+incompatible // indirect
-	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/gookit/color v1.5.4 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
@@ -63,7 +65,7 @@ require (
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
 	github.com/jackc/puddle/v2 v2.2.1 // indirect
-	github.com/klauspost/compress v1.17.0 // indirect
+	github.com/klauspost/compress v1.17.4 // indirect
 	github.com/labstack/gommon v0.4.2 // indirect
 	github.com/lib/pq v1.10.9 // indirect
 	github.com/lithammer/fuzzysearch v1.1.8 // indirect
@@ -72,6 +74,7 @@ require (
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mattn/go-runewidth v0.0.15 // indirect
+	github.com/moby/docker-image-spec v1.3.1 // indirect
 	github.com/moby/patternmatcher v0.6.0 // indirect
 	github.com/moby/sys/sequential v0.5.0 // indirect
 	github.com/moby/sys/user v0.1.0 // indirect
@@ -106,18 +109,14 @@ require (
 	go.opentelemetry.io/otel/trace v1.27.0 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.9.0 // indirect
-	golang.org/x/crypto v0.22.0 // indirect
+	golang.org/x/crypto v0.24.0 // indirect
 	golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect
-	golang.org/x/mod v0.16.0 // indirect
-	golang.org/x/net v0.24.0 // indirect
-	golang.org/x/sys v0.20.0 // indirect
-	golang.org/x/term v0.19.0 // indirect
-	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/mod v0.17.0 // indirect
+	golang.org/x/net v0.26.0 // indirect
+	golang.org/x/sys v0.21.0 // indirect
+	golang.org/x/term v0.21.0 // indirect
+	golang.org/x/text v0.16.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
-	golang.org/x/tools v0.13.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20231120223509-83a465c0220f // indirect
-	google.golang.org/grpc v1.59.0 // indirect
-	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
diff --git a/go.sum b/go.sum
index fa35309..745e797 100644
--- a/go.sum
+++ b/go.sum
@@ -23,19 +23,20 @@ github.com/MarvinJWendt/testza v0.3.0/go.mod h1:eFcL4I0idjtIx8P9C6KkAuLgATNKpX4/
 github.com/MarvinJWendt/testza v0.4.2/go.mod h1:mSdhXiKH8sg/gQehJ63bINcCKp7RtYewEjXsvsVUPbE=
 github.com/MarvinJWendt/testza v0.5.2 h1:53KDo64C1z/h/d/stCYCPY69bt/OSwjq5KpFNwi+zB4=
 github.com/MarvinJWendt/testza v0.5.2/go.mod h1:xu53QFE5sCdjtMCKk8YMQ2MnymimEctc4n3EjyIYvEY=
-github.com/Microsoft/go-winio v0.6.1 h1:9/kr64B9VUZrLm5YYwbGtUJnMgqWVOdUAXu6Migciow=
-github.com/Microsoft/go-winio v0.6.1/go.mod h1:LRdKpFKfdobln8UmuiYcKPot9D2v6svN5+sAH+4kjUM=
-github.com/Microsoft/hcsshim v0.11.4 h1:68vKo2VN8DE9AdN4tnkWnmdhqdbpUFM8OF3Airm7fz8=
-github.com/Microsoft/hcsshim v0.11.4/go.mod h1:smjE4dvqPX9Zldna+t5FG3rnoHhaB7QYxPRqGcpAD9w=
+github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
+github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
 github.com/atomicgo/cursor v0.0.1/go.mod h1:cBON2QmmrysudxNBFthvMtN32r3jxVRIvzkUiF/RuIk=
+github.com/aws/aws-sdk-go v1.42.27/go.mod h1:OGr6lGMAKGlG9CVrYnWYDKIyb829c6EVBRjxqjmPepc=
 github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
 github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/containerd/console v1.0.3 h1:lIr7SlA5PxZyMV30bDW0MGbiOPXwc63yRuCP0ARubLw=
 github.com/containerd/console v1.0.3/go.mod h1:7LqA/THxQ86k76b8c/EMSiaJ3h1eZkMkXar0TQ1gf3U=
-github.com/containerd/containerd v1.7.15 h1:afEHXdil9iAm03BmhjzKyXnnEBtjaLJefdU7DV0IFes=
-github.com/containerd/containerd v1.7.15/go.mod h1:ISzRRTMF8EXNpJlTzyr2XMhN+j9K302C21/+cr3kUnY=
+github.com/containerd/containerd v1.7.18 h1:jqjZTQNfXGoEaZdW1WwPU0RqSn1Bm2Ay/KJPUuO8nao=
+github.com/containerd/containerd v1.7.18/go.mod h1:IYEk9/IO6wAPUz2bCMVUbsfXjzw5UNP5fLz4PsUygQ4=
 github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I=
 github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo=
+github.com/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpSBQv6A=
+github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/cpuguy83/dockercfg v0.3.1 h1:/FpZ+JaygUR/lZP2NlFI2DVfrOEMAIKP5wWEJdoYe9E=
 github.com/cpuguy83/dockercfg v0.3.1/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc=
@@ -48,10 +49,10 @@ github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dhui/dktest v0.4.1 h1:/w+IWuDXVymg3IrRJCHHOkMK10m9aNVMOyD0X12YVTg=
 github.com/dhui/dktest v0.4.1/go.mod h1:DdOqcUpL7vgyP4GlF3X3w7HbSlz8cEQzwewPveYEQbA=
-github.com/distribution/reference v0.5.0 h1:/FUIFXtfc/x2gpa5/VGfiGLuOIdYa1t65IKK2OFGvA0=
-github.com/distribution/reference v0.5.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
-github.com/docker/docker v25.0.6+incompatible h1:5cPwbwriIcsua2REJe8HqQV+6WlWc1byg2QSXzBxBGg=
-github.com/docker/docker v25.0.6+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
+github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
+github.com/docker/docker v27.1.1+incompatible h1:hO/M4MtV36kzKldqnA37IWhebRA+LnqqcqDja6kVaKY=
+github.com/docker/docker v27.1.1+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj1Br63c=
 github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
@@ -62,8 +63,10 @@ github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 h1:Oy0F4A
 github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0=
 github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
 github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
-github.com/elastic/go-elasticsearch/v8 v8.0.0-20210311100734-5d6b0c808457 h1:5daTns4cQjTWInqBApWdigDJdSPlmVUBo2yqX4wMjys=
-github.com/elastic/go-elasticsearch/v8 v8.0.0-20210311100734-5d6b0c808457/go.mod h1:xe9a/L2aeOgFKKgrO3ibQTnMdpAeL0GC+5/HpGScSa4=
+github.com/elastic/elastic-transport-go/v8 v8.6.0 h1:Y2S/FBjx1LlCv5m6pWAF2kDJAHoSjSRSJCApolgfthA=
+github.com/elastic/elastic-transport-go/v8 v8.6.0/go.mod h1:YLHer5cj0csTzNFXoNQ8qhtGY1GTvSqPnKWKaqQE3Hk=
+github.com/elastic/go-elasticsearch/v8 v8.14.0 h1:1ywU8WFReLLcxE1WJqii3hTtbPUE2hc38ZK/j4mMFow=
+github.com/elastic/go-elasticsearch/v8 v8.14.0/go.mod h1:WRvnlGkSuZyp83M2U8El/LGXpCjYLrvlkSgkAH4O5I4=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
@@ -86,8 +89,6 @@ github.com/golang-jwt/jwt v3.2.2+incompatible h1:IfV12K8xAKAnZqdXVzCZ+TOjboZ2keL
 github.com/golang-jwt/jwt v3.2.2+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I=
 github.com/golang-migrate/migrate/v4 v4.17.1 h1:4zQ6iqL6t6AiItphxJctQb3cFqWiSpMnX7wLTPnnYO4=
 github.com/golang-migrate/migrate/v4 v4.17.1/go.mod h1:m8hinFyWBn0SA4QKHuKh175Pm9wjmxj3S2Mia7dbXzM=
-github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
-github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
 github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
@@ -137,11 +138,13 @@ github.com/jcmturner/gokrb5/v8 v8.4.4 h1:x1Sv4HaTpepFkXbt2IkL29DXRf8sOfZXo8eRKh6
 github.com/jcmturner/gokrb5/v8 v8.4.4/go.mod h1:1btQEpgT6k+unzCwX1KdWMEwPPkkgBtP+F6aCACiMrs=
 github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZY=
 github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc=
+github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
+github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
-github.com/klauspost/compress v1.17.0 h1:Rnbp4K9EjcDuVuHtd0dgA4qNuv9yKDYKK1ulpJwgrqM=
-github.com/klauspost/compress v1.17.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/klauspost/compress v1.17.4 h1:Ej5ixsIri7BrIjBkRZLTo6ghwrEtHFk7ijlczPW4fZ4=
+github.com/klauspost/compress v1.17.4/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.10/go.mod h1:g2LTdtYhdyuGPqyWyv7qRAmj1WBqxuObKfj5c0PQa7c=
 github.com/klauspost/cpuid/v2 v2.0.12/go.mod h1:g2LTdtYhdyuGPqyWyv7qRAmj1WBqxuObKfj5c0PQa7c=
@@ -177,6 +180,8 @@ github.com/mattn/go-runewidth v0.0.15 h1:UNAjwbU9l54TA3KzvqLGxwWjHmMgBUVhBiTjelZ
 github.com/mattn/go-runewidth v0.0.15/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
 github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
 github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
+github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
+github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo=
 github.com/moby/patternmatcher v0.6.0 h1:GmP9lR19aU5GqSSFko+5pRqHi+Ohk1O69aFiKkVGiPk=
 github.com/moby/patternmatcher v0.6.0/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc=
 github.com/moby/sys/sequential v0.5.0 h1:OPvI35Lzn9K04PBbCLW0g4LcFAJgHsvXsRyewg5lXtc=
@@ -191,6 +196,8 @@ github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug=
 github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM=
+github.com/opensearch-project/opensearch-go v1.1.0 h1:eG5sh3843bbU1itPRjA9QXbxcg8LaZ+DjEzQH9aLN3M=
+github.com/opensearch-project/opensearch-go v1.1.0/go.mod h1:+6/XHCuTH+fwsMJikZEWsucZ4eZMma3zNSeLrTtVGbo=
 github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4=
 github.com/pelletier/go-toml/v2 v2.1.0/go.mod h1:tJU2Z3ZkXwnxa4DPO899bsyIoywizdUvyaeZurnPPDc=
 github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
@@ -266,8 +273,10 @@ github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsT
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8=
 github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU=
-github.com/testcontainers/testcontainers-go v0.31.0 h1:W0VwIhcEVhRflwL9as3dhY6jXjVCA27AkmbnZ+UTh3U=
-github.com/testcontainers/testcontainers-go v0.31.0/go.mod h1:D2lAoA0zUFiSY+eAflqK5mcUx/A5hrrORaEQrd0SefI=
+github.com/testcontainers/testcontainers-go v0.33.0 h1:zJS9PfXYT5O0ZFXM2xxXfk4J5UMw/kRiISng037Gxdw=
+github.com/testcontainers/testcontainers-go v0.33.0/go.mod h1:W80YpTa8D5C3Yy16icheD01UTDu+LmXIA2Keo+jWtT8=
+github.com/testcontainers/testcontainers-go/modules/elasticsearch v0.33.0 h1:tVsooNzk7SgYDO1OnqeIgihDYiD/vSBNBqwqCfauIJY=
+github.com/testcontainers/testcontainers-go/modules/elasticsearch v0.33.0/go.mod h1:qmspvRf+Hx0iyqKQUmTg1jTNiO7HHGNrx98t/HksVfg=
 github.com/testcontainers/testcontainers-go/modules/kafka v0.31.0 h1:8B1u+sDwYhTUoMI271wPjnCg9mz3dHGLMWpP7YyF7kE=
 github.com/testcontainers/testcontainers-go/modules/kafka v0.31.0/go.mod h1:W1+yLUfUl8VLTzvmApP2FBHgCk8I5SKKjDWjxWEc33U=
 github.com/testcontainers/testcontainers-go/modules/opensearch v0.31.0 h1:sgo2PJb8oCK7ogJjRxAkidXmt+gPzwtyhZpaxSI5wDo=
@@ -306,8 +315,8 @@ go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0 h1:IeMey
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0/go.mod h1:oVdCUtjq9MK9BlS7TtucsQwUcXcymNiEDjgDD2jMtZU=
 go.opentelemetry.io/otel/metric v1.27.0 h1:hvj3vdEKyeCi4YaYfNjv2NUje8FqKqUY8IlF0FxV/ik=
 go.opentelemetry.io/otel/metric v1.27.0/go.mod h1:mVFgmRlhljgBiuk/MP/oKylr4hs85GZAylncepAX/ak=
-go.opentelemetry.io/otel/sdk v1.19.0 h1:6USY6zH+L8uMH8L3t1enZPR3WFEmSTADlqldyHtJi3o=
-go.opentelemetry.io/otel/sdk v1.19.0/go.mod h1:NedEbbS4w3C6zElbLdPJKOpJQOrGUJ+GfzpjUvI0v1A=
+go.opentelemetry.io/otel/sdk v1.21.0 h1:FTt8qirL1EysG6sTQRZ5TokkU8d0ugCj8htOgThZXQ8=
+go.opentelemetry.io/otel/sdk v1.21.0/go.mod h1:Nna6Yv7PWTdgJHVRD9hIYywQBRx7pbox6nwBnZIxl/E=
 go.opentelemetry.io/otel/trace v1.27.0 h1:IqYb813p7cmbHk0a5y6pD5JPakbVfftRXABGt5/Rscw=
 go.opentelemetry.io/otel/trace v1.27.0/go.mod h1:6RiD1hkAprV4/q+yd2ln1HG9GoPx39SuvvstaLBl+l4=
 go.opentelemetry.io/proto/otlp v1.0.0 h1:T0TX0tmXU8a3CbNXzEKGeU5mIVOdf0oykP+u2lIVU/I=
@@ -321,27 +330,28 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
-golang.org/x/crypto v0.22.0 h1:g1v0xeRhjcugydODzvb3mEM9SQ0HGp9s/nh3COQ/C30=
-golang.org/x/crypto v0.22.0/go.mod h1:vr6Su+7cTlO45qkww3VDJlzDn0ctJvRgYbC2NvXHt+M=
+golang.org/x/crypto v0.24.0 h1:mnl8DM0o513X8fdIkmyFE/5hTYxbwYOjDS/+rK6qpRI=
+golang.org/x/crypto v0.24.0/go.mod h1:Z1PMYSOR5nyMcyAVAIQSKCDwalqy85Aqn1x3Ws4L5DM=
 golang.org/x/exp v0.0.0-20230905200255-921286631fa9 h1:GoHiUyI/Tp2nVkLI2mCxVkOjsbSXD66ic0XW0js0R9g=
 golang.org/x/exp v0.0.0-20230905200255-921286631fa9/go.mod h1:S2oDrQGGwySpoQPVqRShND87VCbxmc6bL1Yd2oYrm6k=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
 golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
-golang.org/x/mod v0.16.0 h1:QX4fJ0Rr5cPQCF7O9lh9Se4pmwfwskqZfq5moyldzic=
-golang.org/x/mod v0.16.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
+golang.org/x/mod v0.17.0 h1:zY54UmvipHiNd+pm+m0x9KhZ9hl1/7QNMyxXbc6ICqA=
+golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20211216030914-fe4d6282115f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
-golang.org/x/net v0.24.0 h1:1PcaxkF854Fu3+lvBIx5SYn9wRlBzzcnHZSiaFFAb0w=
-golang.org/x/net v0.24.0/go.mod h1:2Q7sJY5mzlzWjKtYUEXSlBWCdyaioyXzRB2RtU8KVE8=
+golang.org/x/net v0.26.0 h1:soB7SVo0PWrY4vPW/+ay0jKDNScG2X9wFeYlXIvJsOQ=
+golang.org/x/net v0.26.0/go.mod h1:5YKkiSynbBIh3p6iOc/vibscux0x38BZDkn8sCUPxHE=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -357,6 +367,7 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211013075003-97ac67df715c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -372,8 +383,8 @@ golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y=
-golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws=
+golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -381,17 +392,18 @@ golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuX
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
 golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
-golang.org/x/term v0.19.0 h1:+ThwsDv+tYfnJFhF4L8jITxu1tdTWRTZpdsWgEgjL6Q=
-golang.org/x/term v0.19.0/go.mod h1:2CuTdWZ7KHSQwUzKva0cbMg6q2DMI3Mmxp+gKJbskEk=
+golang.org/x/term v0.21.0 h1:WVXCp+/EBEHOj53Rvu+7KiT/iElMrO8ACK16SMZ3jaA=
+golang.org/x/term v0.21.0/go.mod h1:ooXLefLobQVslOqselCNF4SxFAaoS6KujMbsGzSDmX0=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
-golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
-golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4=
+golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
 golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
 golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
@@ -400,19 +412,17 @@ golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
-golang.org/x/tools v0.13.0 h1:Iey4qkscZuv0VvIt8E0neZjtPVQFSc870HQ448QgEmQ=
-golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/genproto v0.0.0-20231106174013-bbf56f31fb17 h1:wpZ8pe2x1Q3f2KyT5f8oP/fa9rHAKgFPr/HZdNuS+PQ=
-google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17 h1:JpwMPBpFN3uKhdaekDpiNlImDdkUAyiJ6ez/uxGaUSo=
-google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17/go.mod h1:0xJLfVdJqpAPl8tDg1ujOCGzx6LFLttXT5NhllGOXY4=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20231120223509-83a465c0220f h1:ultW7fxlIvee4HYrtnaRPon9HpEgFk5zYpmfMgtKB5I=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20231120223509-83a465c0220f/go.mod h1:L9KNLi232K1/xB6f7AlSX692koaRnKaWSR0stBki0Yc=
-google.golang.org/grpc v1.59.0 h1:Z5Iec2pjwb+LEOqzpB2MR12/eKFhDPhuqW91O+4bwUk=
-google.golang.org/grpc v1.59.0/go.mod h1:aUPDwccQo6OTjy7Hct4AfBPD1GptF4fyUjIkQ9YtF98=
+google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 h1:RFiFrvy37/mpSpdySBDrUdipW/dHwsRwh3J3+A9VgT4=
+google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237/go.mod h1:Z5Iiy3jtmioajWHDGFk7CeugTyHtPvMHA4UTmUkyalE=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 h1:NnYq6UN9ReLM9/Y01KWNOWyI5xQ9kbIms5GGJVwS/Yc=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237/go.mod h1:WtryC6hu0hhx87FDGxWCDptyssuo68sk10vYjF+T9fY=
+google.golang.org/grpc v1.64.1 h1:LKtvyfbX3UGVPFcGqJ9ItpVWW6oN/2XqTxfAnwRRXiA=
+google.golang.org/grpc v1.64.1/go.mod h1:hiQF4LFZelK2WKaP6W0L92zGHtiQdZxk8CrSdvyjeP0=
 google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
 google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
@@ -423,9 +433,10 @@ gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA=
 gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gotest.tools/v3 v3.5.0 h1:Ljk6PdHdOhAb5aDMWXjDLMMhph+BpztA4v1QdqEW2eY=
-gotest.tools/v3 v3.5.0/go.mod h1:isy3WKz7GK6uNw/sbHzfKBLvlvXwUyV06n6brMxxopU=
+gotest.tools/v3 v3.5.1 h1:EENdUnS3pdur5nybKYIh2Vfgc8IUNBjxDPSjtiJcOzU=
+gotest.tools/v3 v3.5.1/go.mod h1:isy3WKz7GK6uNw/sbHzfKBLvlvXwUyV06n6brMxxopU=
diff --git a/internal/es/es_client.go b/internal/searchstore/elasticsearch/elasticsearch_client.go
similarity index 76%
rename from internal/es/es_client.go
rename to internal/searchstore/elasticsearch/elasticsearch_client.go
index 3041755..db8c697 100644
--- a/internal/es/es_client.go
+++ b/internal/searchstore/elasticsearch/elasticsearch_client.go
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package es
+package elasticsearch
 
 import (
 	"bufio"
@@ -14,38 +14,14 @@ import (
 
 	"github.com/elastic/go-elasticsearch/v8"
 	"github.com/elastic/go-elasticsearch/v8/esapi"
+	"github.com/xataio/pgstream/internal/searchstore"
 )
 
-type SearchClient interface {
-	CloseIndex(ctx context.Context, index string) error
-	Count(ctx context.Context, index string) (int, error)
-	CreateIndex(ctx context.Context, index string, body map[string]any) error
-	DeleteByQuery(ctx context.Context, req *DeleteByQueryRequest) error
-	DeleteIndex(ctx context.Context, index []string) error
-	GetIndexAlias(ctx context.Context, name string) (map[string]any, error)
-	GetIndexMappings(ctx context.Context, index string) (*Mappings, error)
-	GetIndicesStats(ctx context.Context, indexPattern string) ([]IndexStats, error)
-	Index(ctx context.Context, req *IndexRequest) error
-	IndexWithID(ctx context.Context, req *IndexWithIDRequest) error
-	IndexExists(ctx context.Context, index string) (bool, error)
-	ListIndices(ctx context.Context, indices []string) ([]string, error)
-	Perform(req *http.Request) (*http.Response, error)
-	PutIndexAlias(ctx context.Context, index []string, name string) error
-	PutIndexMappings(ctx context.Context, index string, body map[string]any) error
-	PutIndexSettings(ctx context.Context, index string, body map[string]any) error
-	RefreshIndex(ctx context.Context, index string) error
-	Search(ctx context.Context, req *SearchRequest) (*SearchResponse, error)
-	SendBulkRequest(ctx context.Context, items []BulkItem) ([]BulkItem, error)
-}
-
 type Client struct {
 	client *elasticsearch.Client
 }
 
-var (
-	ErrResourceNotFound      = errors.New("elasticsearch resource not found")
-	errInvalidSearchEnvelope = errors.New("invalid search response")
-)
+var errInvalidSearchEnvelope = errors.New("invalid search response")
 
 func NewClient(url string) (*Client, error) {
 	es, err := newClient(url)
@@ -55,6 +31,10 @@ func NewClient(url string) (*Client, error) {
 	return &Client{client: es}, nil
 }
 
+func (ec *Client) GetMapper() searchstore.Mapper {
+	return NewMapper()
+}
+
 func (ec *Client) CloseIndex(ctx context.Context, index string) error {
 	res, err := ec.client.Indices.Close(
 		[]string{index},
@@ -85,7 +65,7 @@ func (ec *Client) Count(ctx context.Context, index string) (int, error) {
 		return 0, fmt.Errorf("[Count] error response from Elasticsearch: %w", err)
 	}
 
-	count := &countResponse{}
+	count := &searchstore.CountResponse{}
 	if err := json.NewDecoder(res.Body).Decode(count); err != nil {
 		return 0, fmt.Errorf("[Count] error decoding Elasticsearch response: %w", err)
 	}
@@ -94,7 +74,7 @@ func (ec *Client) Count(ctx context.Context, index string) (int, error) {
 }
 
 func (ec *Client) CreateIndex(ctx context.Context, index string, body map[string]any) error {
-	reader, err := createReader(body)
+	reader, err := searchstore.CreateReader(body)
 	if err != nil {
 		return err
 	}
@@ -114,8 +94,8 @@ func (ec *Client) CreateIndex(ctx context.Context, index string, body map[string
 	return nil
 }
 
-func (ec *Client) DeleteByQuery(ctx context.Context, req *DeleteByQueryRequest) error {
-	reader, err := createReader(req.Query)
+func (ec *Client) DeleteByQuery(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
+	reader, err := searchstore.CreateReader(req.Query)
 	if err != nil {
 		return err
 	}
@@ -156,7 +136,7 @@ func (ec *Client) DeleteIndex(ctx context.Context, index []string) error {
 	return nil
 }
 
-func (ec *Client) Index(ctx context.Context, req *IndexRequest) error {
+func (ec *Client) Index(ctx context.Context, req *searchstore.IndexRequest) error {
 	res, err := ec.client.Index(req.Index,
 		bytes.NewReader(req.Body),
 		ec.client.Index.WithContext(ctx),
@@ -174,7 +154,7 @@ func (ec *Client) Index(ctx context.Context, req *IndexRequest) error {
 	return nil
 }
 
-func (ec *Client) IndexWithID(ctx context.Context, req *IndexWithIDRequest) error {
+func (ec *Client) IndexWithID(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
 	res, err := ec.client.Index(req.Index,
 		bytes.NewReader(req.Body),
 		ec.client.Index.WithContext(ctx),
@@ -235,7 +215,7 @@ func (ec *Client) GetIndexAlias(ctx context.Context, name string) (map[string]an
 	return resMap, nil
 }
 
-func (ec *Client) GetIndexMappings(ctx context.Context, index string) (*Mappings, error) {
+func (ec *Client) GetIndexMappings(ctx context.Context, index string) (*searchstore.Mappings, error) {
 	res, err := ec.client.Indices.GetMapping(
 		ec.client.Indices.GetMapping.WithIndex(index),
 		ec.client.Indices.GetMapping.WithContext(ctx))
@@ -248,7 +228,7 @@ func (ec *Client) GetIndexMappings(ctx context.Context, index string) (*Mappings
 		return nil, fmt.Errorf("[GetIndexMapping] error response from Elasticsearch: %w", err)
 	}
 
-	var indexMappings mappingResponse
+	var indexMappings searchstore.MappingResponse
 	if err = json.NewDecoder(res.Body).Decode(&indexMappings); err != nil {
 		return nil, err
 	}
@@ -260,7 +240,7 @@ func (ec *Client) GetIndexMappings(ctx context.Context, index string) (*Mappings
 
 // GetIndicesStats uses the index stats API to fetch statistics about indices. indexPattern is a
 // wildcard pattern used to select the indices we care about.
-func (ec *Client) GetIndicesStats(ctx context.Context, indexPattern string) ([]IndexStats, error) {
+func (ec *Client) GetIndicesStats(ctx context.Context, indexPattern string) ([]searchstore.IndexStats, error) {
 	res, err := ec.client.Indices.Stats(
 		ec.client.Indices.Stats.WithContext(ctx),
 		ec.client.Indices.Stats.WithIndex(indexPattern),
@@ -270,14 +250,14 @@ func (ec *Client) GetIndicesStats(ctx context.Context, indexPattern string) ([]I
 	}
 	defer res.Body.Close()
 
-	var response indexStatsResponse
+	var response searchstore.IndexStatsResponse
 	if err = json.NewDecoder(res.Body).Decode(&response); err != nil {
 		return nil, fmt.Errorf("[GetIndicesStats] decoding response body: %w", err)
 	}
 
-	usage := make([]IndexStats, 0, len(response.Indices))
+	usage := make([]searchstore.IndexStats, 0, len(response.Indices))
 	for index, r := range response.Indices {
-		usage = append(usage, IndexStats{
+		usage = append(usage, searchstore.IndexStats{
 			Index:            index,
 			TotalSizeBytes:   uint64(r.Total.Store.SizeInBytes),
 			PrimarySizeBytes: uint64(r.Primaries.Store.SizeInBytes),
@@ -341,7 +321,7 @@ func (ec *Client) PutIndexAlias(ctx context.Context, index []string, name string
 // PutIndexMappings add field type mapping data to a previously created ES index
 // Dynamic mapping is disabled upon index creation, so it is a requirement to explicitly define mappings for each column
 func (ec *Client) PutIndexMappings(ctx context.Context, index string, mapping map[string]any) error {
-	reader, err := createReader(mapping)
+	reader, err := searchstore.CreateReader(mapping)
 	if err != nil {
 		return err
 	}
@@ -362,7 +342,7 @@ func (ec *Client) PutIndexMappings(ctx context.Context, index string, mapping ma
 }
 
 func (ec *Client) PutIndexSettings(ctx context.Context, index string, settings map[string]any) error {
-	reader, err := createReader(settings)
+	reader, err := searchstore.CreateReader(settings)
 	if err != nil {
 		return err
 	}
@@ -403,7 +383,7 @@ func (ec *Client) Perform(req *http.Request) (*http.Response, error) {
 	return ec.client.Transport.Perform(req)
 }
 
-func (ec *Client) Search(ctx context.Context, req *SearchRequest) (*SearchResponse, error) {
+func (ec *Client) Search(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
 	res, err := ec.client.Search(ec.parseSearchRequest(ctx, req)...)
 	if err != nil {
 		return nil, fmt.Errorf("[Search] error from Elasticsearch: %w", err)
@@ -413,7 +393,7 @@ func (ec *Client) Search(ctx context.Context, req *SearchRequest) (*SearchRespon
 		return nil, fmt.Errorf("[Search] error response from Elasticsearch: %w", err)
 	}
 
-	var response SearchResponse
+	var response searchstore.SearchResponse
 	err = json.NewDecoder(res.Body).Decode(&response)
 	if err != nil {
 		return nil, fmt.Errorf("[Search] decoding response body: %w: %w", errInvalidSearchEnvelope, err)
@@ -423,10 +403,10 @@ func (ec *Client) Search(ctx context.Context, req *SearchRequest) (*SearchRespon
 }
 
 // SendBulkRequest can perform multiple indexing or delete operations in a single call
-func (ec *Client) SendBulkRequest(ctx context.Context, items []BulkItem) ([]BulkItem, error) {
+func (ec *Client) SendBulkRequest(ctx context.Context, items []searchstore.BulkItem) ([]searchstore.BulkItem, error) {
 	buffer := new(bytes.Buffer)
 
-	if err := encodeBulkItems(buffer, items); err != nil {
+	if err := searchstore.EncodeBulkItems(buffer, items); err != nil {
 		return nil, err
 	}
 
@@ -450,10 +430,10 @@ func (ec *Client) SendBulkRequest(ctx context.Context, items []BulkItem) ([]Bulk
 		return nil, fmt.Errorf("error from Elasticsearch: %d: %s", resp.StatusCode, bodyBytes)
 	}
 
-	return verifyResponse(bodyBytes, items)
+	return searchstore.VerifyResponse(bodyBytes, items)
 }
 
-func (ec *Client) parseSearchRequest(ctx context.Context, req *SearchRequest) []func(*esapi.SearchRequest) {
+func (ec *Client) parseSearchRequest(ctx context.Context, req *searchstore.SearchRequest) []func(*esapi.SearchRequest) {
 	opts := []func(*esapi.SearchRequest){
 		ec.client.Search.WithContext(ctx),
 	}
@@ -483,14 +463,7 @@ func (ec *Client) parseSearchRequest(ctx context.Context, req *SearchRequest) []
 }
 
 func (ec *Client) isErrResponse(res *esapi.Response) error {
-	if res.IsError() {
-		if res.StatusCode == http.StatusNotFound {
-			return fmt.Errorf("%w: %w", ErrResourceNotFound, extractResponseError(res))
-		}
-		return extractResponseError(res)
-	}
-
-	return nil
+	return searchstore.IsErrResponse(newAPIResponse(res))
 }
 
 func newClient(address string) (*elasticsearch.Client, error) {
@@ -508,44 +481,18 @@ func newClient(address string) (*elasticsearch.Client, error) {
 	return elasticsearch.NewClient(cfg)
 }
 
-// createReader returns a reader on the JSON representation of the given value.
-func createReader(value any) (*bytes.Reader, error) {
-	bytesValue, err := json.Marshal(value)
-	if err != nil {
-		return nil, fmt.Errorf("unexpected marshaling error: %w", err)
-	}
-	return bytes.NewReader(bytesValue), nil
+type apiResponse struct {
+	*esapi.Response
 }
 
-func verifyResponse(bodyBytes []byte, items []BulkItem) (failed []BulkItem, err error) {
-	var esResponse BulkResponse
-
-	if err := json.Unmarshal(bodyBytes, &esResponse); err != nil {
-		return nil, fmt.Errorf("error unmarshaling response from es: %w (%s)", err, bodyBytes)
-	}
-
-	if !esResponse.Errors {
-		return []BulkItem{}, nil
-	}
-
-	failed = []BulkItem{}
-	for i, respItem := range esResponse.Items {
-		if items[i].Index != nil {
-			if respItem.Index.Status > 299 {
-				items[i].Status = respItem.Index.Status
-				items[i].Error = respItem.Index.Error
-				failed = append(failed, items[i])
-			}
-		} else if items[i].Delete != nil {
-			if respItem.Delete.Status > 299 {
-				items[i].Status = respItem.Delete.Status
-				items[i].Error = respItem.Delete.Error
-				failed = append(failed, items[i])
-			}
-		}
-	}
+func newAPIResponse(res *esapi.Response) *apiResponse {
+	return &apiResponse{Response: res}
+}
 
-	return failed, nil
+func (r *apiResponse) GetBody() io.ReadCloser {
+	return r.Body
 }
 
-func Ptr[T any](i T) *T { return &i }
+func (r *apiResponse) GetStatusCode() int {
+	return r.StatusCode
+}
diff --git a/internal/searchstore/elasticsearch/elasticsearch_mapper.go b/internal/searchstore/elasticsearch/elasticsearch_mapper.go
new file mode 100644
index 0000000..a987560
--- /dev/null
+++ b/internal/searchstore/elasticsearch/elasticsearch_mapper.go
@@ -0,0 +1,75 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package elasticsearch
+
+import (
+	"github.com/xataio/pgstream/internal/searchstore"
+)
+
+type Mapper struct{}
+
+const (
+	// Lucene’s term byte-length limit is 32766. To cover for the use of UTF-8
+	// text with many non-ASCII characters, the maximum value should be 32766 /
+	// 4 = 8191 since UTF-8 characters may occupy at most 4 bytes.
+	termByteLengthLimit = 32766
+)
+
+func NewMapper() *Mapper {
+	return &Mapper{}
+}
+
+func (m *Mapper) GetDefaultIndexSettings() map[string]any {
+	return map[string]any{
+		"number_of_shards":                 1,
+		"number_of_replicas":               1,
+		"index.mapping.total_fields.limit": 2000,
+	}
+}
+
+func (m *Mapper) FieldMapping(field *searchstore.Field) (map[string]any, error) {
+	switch field.SearchType {
+	case searchstore.IntegerType:
+		return map[string]any{"type": "long"}, nil
+	case searchstore.FloatType:
+		return map[string]any{"type": "double"}, nil
+	case searchstore.BoolType:
+		return map[string]any{"type": "boolean"}, nil
+	case searchstore.TextType, searchstore.JSONType:
+		return map[string]any{"type": "text"}, nil
+	case searchstore.StringType:
+		return map[string]any{
+			"type":         "keyword",
+			"ignore_above": termByteLengthLimit,
+			"fields": map[string]any{
+				"text": map[string]any{
+					"type": "text",
+				},
+			},
+		}, nil
+	case searchstore.TimeType:
+		return map[string]any{
+			"type":   "date",
+			"format": "HH:mm:ss[.SS][x][Z]||HH:mm:ss[.SSS][x][Z]||HH:mm:ss[.SSSSSS][x][Z]",
+		}, nil
+	case searchstore.DateType:
+		return map[string]any{
+			"type":   "date",
+			"format": "date",
+		}, nil
+	case searchstore.DateTimeType, searchstore.DateTimeTZType:
+		return map[string]any{
+			"type":   "date",
+			"format": "yyyy-MM-dd HH:mm:ss[.SSS][x]||yyyy-MM-dd HH:mm:ss[.SS][x]||yyyy-MM-dd HH:mm:ss[.S][x]||yyyy-MM-dd'T'HH:mm:ss[.SSS][X]",
+		}, nil
+	case searchstore.PGVectorType:
+		vectorSettings := map[string]any{
+			"type":  "dense_vector",
+			"index": true,
+			"dims":  field.Metadata.VectorDimension,
+		}
+		return vectorSettings, nil
+	default:
+		return nil, searchstore.ErrUnsupportedSearchFieldType
+	}
+}
diff --git a/internal/es/mocks/mock_client.go b/internal/searchstore/mocks/mock_client.go
similarity index 68%
rename from internal/es/mocks/mock_client.go
rename to internal/searchstore/mocks/mock_client.go
index 32167c0..c13b1cc 100644
--- a/internal/es/mocks/mock_client.go
+++ b/internal/searchstore/mocks/mock_client.go
@@ -6,20 +6,20 @@ import (
 	"context"
 	"net/http"
 
-	"github.com/xataio/pgstream/internal/es"
+	"github.com/xataio/pgstream/internal/searchstore"
 )
 
 type Client struct {
 	CloseIndexFn       func(ctx context.Context, index string) error
 	CountFn            func(ctx context.Context, index string) (int, error)
 	CreateIndexFn      func(ctx context.Context, index string, body map[string]any) error
-	DeleteByQueryFn    func(ctx context.Context, req *es.DeleteByQueryRequest) error
+	DeleteByQueryFn    func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error
 	DeleteIndexFn      func(ctx context.Context, index []string) error
 	GetIndexAliasFn    func(ctx context.Context, name string) (map[string]any, error)
-	GetIndexMappingsFn func(ctx context.Context, index string) (*es.Mappings, error)
-	GetIndicesStatsFn  func(ctx context.Context, pattern string) ([]es.IndexStats, error)
-	IndexFn            func(ctx context.Context, req *es.IndexRequest) error
-	IndexWithIDFn      func(ctx context.Context, req *es.IndexWithIDRequest) error
+	GetIndexMappingsFn func(ctx context.Context, index string) (*searchstore.Mappings, error)
+	GetIndicesStatsFn  func(ctx context.Context, pattern string) ([]searchstore.IndexStats, error)
+	IndexFn            func(ctx context.Context, req *searchstore.IndexRequest) error
+	IndexWithIDFn      func(ctx context.Context, req *searchstore.IndexWithIDRequest) error
 	IndexExistsFn      func(ctx context.Context, index string) (bool, error)
 	ListIndicesFn      func(ctx context.Context, indices []string) ([]string, error)
 	PerformFn          func(req *http.Request) (*http.Response, error)
@@ -27,8 +27,9 @@ type Client struct {
 	PutIndexMappingsFn func(ctx context.Context, index string, body map[string]any) error
 	PutIndexSettingsFn func(ctx context.Context, index string, body map[string]any) error
 	RefreshIndexFn     func(ctx context.Context, index string) error
-	SearchFn           func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error)
-	SendBulkRequestFn  func(ctx context.Context, items []es.BulkItem) ([]es.BulkItem, error)
+	SearchFn           func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error)
+	SendBulkRequestFn  func(ctx context.Context, items []searchstore.BulkItem) ([]searchstore.BulkItem, error)
+	GetMapperFn        func() searchstore.Mapper
 }
 
 func (m *Client) CloseIndex(ctx context.Context, index string) error {
@@ -43,7 +44,7 @@ func (m *Client) CreateIndex(ctx context.Context, index string, body map[string]
 	return m.CreateIndexFn(ctx, index, body)
 }
 
-func (m *Client) DeleteByQuery(ctx context.Context, req *es.DeleteByQueryRequest) error {
+func (m *Client) DeleteByQuery(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 	return m.DeleteByQueryFn(ctx, req)
 }
 
@@ -55,19 +56,19 @@ func (m *Client) GetIndexAlias(ctx context.Context, name string) (map[string]any
 	return m.GetIndexAliasFn(ctx, name)
 }
 
-func (m *Client) GetIndexMappings(ctx context.Context, index string) (*es.Mappings, error) {
+func (m *Client) GetIndexMappings(ctx context.Context, index string) (*searchstore.Mappings, error) {
 	return m.GetIndexMappingsFn(ctx, index)
 }
 
-func (m *Client) GetIndicesStats(ctx context.Context, pattern string) ([]es.IndexStats, error) {
+func (m *Client) GetIndicesStats(ctx context.Context, pattern string) ([]searchstore.IndexStats, error) {
 	return m.GetIndicesStatsFn(ctx, pattern)
 }
 
-func (m *Client) Index(ctx context.Context, req *es.IndexRequest) error {
+func (m *Client) Index(ctx context.Context, req *searchstore.IndexRequest) error {
 	return m.IndexFn(ctx, req)
 }
 
-func (m *Client) IndexWithID(ctx context.Context, req *es.IndexWithIDRequest) error {
+func (m *Client) IndexWithID(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
 	return m.IndexWithIDFn(ctx, req)
 }
 
@@ -99,10 +100,14 @@ func (m *Client) RefreshIndex(ctx context.Context, index string) error {
 	return m.RefreshIndexFn(ctx, index)
 }
 
-func (m *Client) Search(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
+func (m *Client) Search(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
 	return m.SearchFn(ctx, req)
 }
 
-func (m *Client) SendBulkRequest(ctx context.Context, items []es.BulkItem) ([]es.BulkItem, error) {
+func (m *Client) SendBulkRequest(ctx context.Context, items []searchstore.BulkItem) ([]searchstore.BulkItem, error) {
 	return m.SendBulkRequestFn(ctx, items)
 }
+
+func (m *Client) GetMapper() searchstore.Mapper {
+	return m.GetMapperFn()
+}
diff --git a/internal/searchstore/mocks/mock_mapper.go b/internal/searchstore/mocks/mock_mapper.go
new file mode 100644
index 0000000..e6cab65
--- /dev/null
+++ b/internal/searchstore/mocks/mock_mapper.go
@@ -0,0 +1,21 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package mocks
+
+import "github.com/xataio/pgstream/internal/searchstore"
+
+type Mapper struct {
+	GetDefaultIndexSettingsFn func() map[string]any
+	FieldMappingFn            func(*searchstore.Field) (map[string]any, error)
+}
+
+func (m *Mapper) GetDefaultIndexSettings() map[string]any {
+	if m.GetDefaultIndexSettingsFn == nil {
+		return map[string]any{}
+	}
+	return m.GetDefaultIndexSettingsFn()
+}
+
+func (m *Mapper) FieldMapping(f *searchstore.Field) (map[string]any, error) {
+	return m.FieldMappingFn(f)
+}
diff --git a/internal/searchstore/opensearch/opensearch_client.go b/internal/searchstore/opensearch/opensearch_client.go
new file mode 100644
index 0000000..ab64c90
--- /dev/null
+++ b/internal/searchstore/opensearch/opensearch_client.go
@@ -0,0 +1,498 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package opensearch
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+
+	"github.com/opensearch-project/opensearch-go"
+	"github.com/opensearch-project/opensearch-go/opensearchapi"
+	"github.com/xataio/pgstream/internal/searchstore"
+)
+
+type Client struct {
+	client *opensearch.Client
+}
+
+var errInvalidSearchEnvelope = errors.New("invalid search response")
+
+func NewClient(url string) (*Client, error) {
+	os, err := newClient(url)
+	if err != nil {
+		return nil, fmt.Errorf("create opensearch client: %w", err)
+	}
+	return &Client{client: os}, nil
+}
+
+func (c *Client) GetMapper() searchstore.Mapper {
+	return NewMapper()
+}
+
+func (c *Client) CloseIndex(ctx context.Context, index string) error {
+	res, err := c.client.Indices.Close(
+		[]string{index},
+		c.client.Indices.Close.WithContext(ctx),
+	)
+	if err != nil {
+		return fmt.Errorf("[CloseIndex] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return fmt.Errorf("[CloseIndex] error response from OpenSearch: %w", err)
+	}
+
+	return nil
+}
+
+func (c *Client) Count(ctx context.Context, index string) (int, error) {
+	res, err := c.client.Count(
+		c.client.Count.WithIndex(index),
+		c.client.Count.WithContext(ctx))
+	if err != nil {
+		return 0, fmt.Errorf("[Count] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return 0, fmt.Errorf("[Count] error response from OpenSearch: %w", err)
+	}
+
+	count := &searchstore.CountResponse{}
+	if err := json.NewDecoder(res.Body).Decode(count); err != nil {
+		return 0, fmt.Errorf("[Count] error decoding OpenSearch response: %w", err)
+	}
+
+	return count.Count, nil
+}
+
+func (c *Client) CreateIndex(ctx context.Context, index string, body map[string]any) error {
+	reader, err := searchstore.CreateReader(body)
+	if err != nil {
+		return err
+	}
+	res, err := c.client.Indices.Create(index,
+		c.client.Indices.Create.WithContext(ctx),
+		c.client.Indices.Create.WithBody(reader),
+	)
+	if err != nil {
+		return fmt.Errorf("[CreateIndex] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return fmt.Errorf("[CreateIndex] error response from OpenSearch: %w", err)
+	}
+
+	return nil
+}
+
+func (c *Client) DeleteByQuery(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
+	reader, err := searchstore.CreateReader(req.Query)
+	if err != nil {
+		return err
+	}
+
+	res, err := c.client.DeleteByQuery(req.Index,
+		reader,
+		c.client.DeleteByQuery.WithContext(ctx),
+		c.client.DeleteByQuery.WithSlices("auto"),
+		c.client.DeleteByQuery.WithWaitForCompletion(false),
+		c.client.DeleteByQuery.WithRefresh(req.Refresh),
+	)
+	if err != nil {
+		return fmt.Errorf("[DeleteByQuery] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return fmt.Errorf("[DeleteByQuery] error response from OpenSearch: %w", err)
+	}
+
+	return nil
+}
+
+func (c *Client) DeleteIndex(ctx context.Context, index []string) error {
+	res, err := c.client.Indices.Delete(
+		index,
+		c.client.Indices.Delete.WithContext(ctx),
+	)
+	if err != nil {
+		return fmt.Errorf("[DeleteIndex] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return fmt.Errorf("[DeleteIndex] error response from OpenSearch: %w", err)
+	}
+
+	return nil
+}
+
+func (c *Client) Index(ctx context.Context, req *searchstore.IndexRequest) error {
+	res, err := c.client.Index(req.Index,
+		bytes.NewReader(req.Body),
+		c.client.Index.WithContext(ctx),
+		c.client.Index.WithRefresh(req.Refresh),
+	)
+	if err != nil {
+		return fmt.Errorf("[Index] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return fmt.Errorf("[Index] error response from OpenSearch: %w", err)
+	}
+
+	return nil
+}
+
+func (c *Client) IndexWithID(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
+	res, err := c.client.Index(req.Index,
+		bytes.NewReader(req.Body),
+		c.client.Index.WithContext(ctx),
+		c.client.Index.WithRefresh(req.Refresh),
+		c.client.Index.WithDocumentID(req.ID),
+	)
+	if err != nil {
+		return fmt.Errorf("[IndexWithID] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return fmt.Errorf("[IndexWithID] error response from OpenSearch: %w", err)
+	}
+
+	return nil
+}
+
+func (c *Client) IndexExists(ctx context.Context, index string) (bool, error) {
+	res, err := c.client.Indices.Exists([]string{index},
+		c.client.Indices.Exists.WithContext(ctx),
+	)
+	if err != nil {
+		return false, fmt.Errorf("[IndexExists] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.IsError() && res.StatusCode != http.StatusNotFound {
+		return false, fmt.Errorf("[IndexExists] error response from OpenSearch: %w", err)
+	}
+
+	return res.StatusCode == http.StatusOK, nil
+}
+
+func (c *Client) GetIndexAlias(ctx context.Context, name string) (map[string]any, error) {
+	res, err := c.client.Indices.GetAlias(
+		c.client.Indices.GetAlias.WithContext(ctx),
+		c.client.Indices.GetAlias.WithName(name),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("[GetIndexAlias] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return nil, fmt.Errorf("[GetIndexAlias] error response from OpenSearch: %w", err)
+	}
+
+	resMap := map[string]any{}
+	resData, err := io.ReadAll(res.Body)
+	if err != nil {
+		return nil, fmt.Errorf("[GetIndexAlias] error reading OpenSearch response body: %w", err)
+	}
+
+	if err := json.Unmarshal(resData, &resMap); err != nil {
+		return nil, fmt.Errorf("[GetIndexAlias] error unmarshalling OpenSearch response: %w", err)
+	}
+	return resMap, nil
+}
+
+func (c *Client) GetIndexMappings(ctx context.Context, index string) (*searchstore.Mappings, error) {
+	res, err := c.client.Indices.GetMapping(
+		c.client.Indices.GetMapping.WithIndex(index),
+		c.client.Indices.GetMapping.WithContext(ctx))
+	if err != nil {
+		return nil, fmt.Errorf("[GetIndexMapping] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return nil, fmt.Errorf("[GetIndexMapping] error response from OpenSearch: %w", err)
+	}
+
+	var indexMappings searchstore.MappingResponse
+	if err = json.NewDecoder(res.Body).Decode(&indexMappings); err != nil {
+		return nil, err
+	}
+
+	mappings := indexMappings[index]
+
+	return &mappings.Mappings, nil
+}
+
+// GetIndicesStats uses the index stats API to fetch statistics about indices. indexPattern is a
+// wildcard pattern used to select the indices we care about.
+func (c *Client) GetIndicesStats(ctx context.Context, indexPattern string) ([]searchstore.IndexStats, error) {
+	res, err := c.client.Indices.Stats(
+		c.client.Indices.Stats.WithContext(ctx),
+		c.client.Indices.Stats.WithIndex(indexPattern),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("[GetIndicesStats] querying OpenSearch Cat API: %w", err)
+	}
+	defer res.Body.Close()
+
+	var response searchstore.IndexStatsResponse
+	if err = json.NewDecoder(res.Body).Decode(&response); err != nil {
+		return nil, fmt.Errorf("[GetIndicesStats] decoding response body: %w", err)
+	}
+
+	usage := make([]searchstore.IndexStats, 0, len(response.Indices))
+	for index, r := range response.Indices {
+		usage = append(usage, searchstore.IndexStats{
+			Index:            index,
+			TotalSizeBytes:   uint64(r.Total.Store.SizeInBytes),
+			PrimarySizeBytes: uint64(r.Primaries.Store.SizeInBytes),
+		})
+	}
+
+	return usage, nil
+}
+
+// ListIndices returns the list of indices that match the index name pattern on
+// input from the OS cluster
+func (c *Client) ListIndices(ctx context.Context, indices []string) ([]string, error) {
+	res, err := c.client.Cat.Indices(
+		c.client.Cat.Indices.WithContext(ctx),
+		c.client.Cat.Indices.WithIndex(indices...),
+		c.client.Cat.Indices.WithH("index"),
+	)
+	if err != nil {
+		return []string{}, fmt.Errorf("[ListIndices] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return []string{}, fmt.Errorf("[ListIndices] error response from OpenSearch: %w", err)
+	}
+
+	scanner := bufio.NewScanner(res.Body)
+	scanner.Split(bufio.ScanLines)
+
+	resp := []string{}
+	for scanner.Scan() {
+		line := scanner.Text()
+		resp = append(resp, line)
+	}
+
+	if err := scanner.Err(); err != nil {
+		return nil, fmt.Errorf("[ListIndices] error scanning response from OpenSearch: %w", err)
+	}
+
+	return resp, nil
+}
+
+func (c *Client) PutIndexAlias(ctx context.Context, index []string, name string) error {
+	res, err := c.client.Indices.PutAlias(
+		index,
+		name,
+		c.client.Indices.PutAlias.WithContext(ctx),
+	)
+	if err != nil {
+		return fmt.Errorf("[PutIndexAlias] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return fmt.Errorf("[PutIndexAlias] error response from OpenSearch: %w", err)
+	}
+
+	return nil
+}
+
+// PutIndexMappings add field type mapping data to a previously created OpenSearch index
+// Dynamic mapping is disabled upon index creation, so it is a requirement to explicitly define mappings for each column
+func (c *Client) PutIndexMappings(ctx context.Context, index string, mapping map[string]any) error {
+	reader, err := searchstore.CreateReader(mapping)
+	if err != nil {
+		return err
+	}
+	res, err := c.client.Indices.PutMapping(
+		reader,
+		c.client.Indices.PutMapping.WithIndex(index),
+		c.client.Indices.PutMapping.WithContext(ctx))
+	if err != nil {
+		return fmt.Errorf("[PutIndexMappings] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return fmt.Errorf("[PutIndexMappings] error response from OpenSearch: %w", err)
+	}
+
+	return nil
+}
+
+func (c *Client) PutIndexSettings(ctx context.Context, index string, settings map[string]any) error {
+	reader, err := searchstore.CreateReader(settings)
+	if err != nil {
+		return err
+	}
+	res, err := c.client.Indices.PutSettings(
+		reader,
+		c.client.Indices.PutSettings.WithContext(ctx),
+		c.client.Indices.PutSettings.WithIndex(index))
+	if err != nil {
+		return fmt.Errorf("[PutIndexSettings] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return fmt.Errorf("[PutIndexSettings] error response from OpenSearch: %w", err)
+	}
+
+	return nil
+}
+
+func (c *Client) RefreshIndex(ctx context.Context, index string) error {
+	res, err := c.client.Indices.Refresh(
+		c.client.Indices.Refresh.WithIndex(index),
+		c.client.Indices.Refresh.WithContext(ctx),
+	)
+	if err != nil {
+		return fmt.Errorf("[RefreshIndex] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+
+	if err := c.isErrResponse(res); err != nil {
+		return fmt.Errorf("[RefreshIndex] error response from OpenSearch: %w", err)
+	}
+
+	return nil
+}
+
+func (c *Client) Perform(req *http.Request) (*http.Response, error) {
+	return c.client.Transport.Perform(req)
+}
+
+func (c *Client) Search(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
+	res, err := c.client.Search(c.parseSearchRequest(ctx, req)...)
+	if err != nil {
+		return nil, fmt.Errorf("[Search] error from OpenSearch: %w", err)
+	}
+	defer res.Body.Close()
+	if err := c.isErrResponse(res); err != nil {
+		return nil, fmt.Errorf("[Search] error response from OpenSearch: %w", err)
+	}
+
+	var response searchstore.SearchResponse
+	err = json.NewDecoder(res.Body).Decode(&response)
+	if err != nil {
+		return nil, fmt.Errorf("[Search] decoding response body: %w: %w", errInvalidSearchEnvelope, err)
+	}
+
+	return &response, nil
+}
+
+// SendBulkRequest can perform multiple indexing or delete operations in a single call
+func (c *Client) SendBulkRequest(ctx context.Context, items []searchstore.BulkItem) ([]searchstore.BulkItem, error) {
+	buffer := new(bytes.Buffer)
+
+	if err := searchstore.EncodeBulkItems(buffer, items); err != nil {
+		return nil, err
+	}
+
+	req, err := http.NewRequest("POST", "/_bulk", buffer)
+	if err != nil {
+		return nil, fmt.Errorf("new http request: %w", err)
+	}
+	req.Header.Add("Content-Type", "application/x-ndjson")
+	req = req.WithContext(ctx)
+
+	resp, err := c.Perform(req)
+	if err != nil {
+		return nil, fmt.Errorf("perform: %w", err)
+	}
+	defer resp.Body.Close()
+	bodyBytes, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("read body: %w", err)
+	}
+	if resp.StatusCode > 299 {
+		return nil, fmt.Errorf("error from OpenSearch: %d: %s", resp.StatusCode, bodyBytes)
+	}
+
+	return searchstore.VerifyResponse(bodyBytes, items)
+}
+
+func (c *Client) parseSearchRequest(ctx context.Context, req *searchstore.SearchRequest) []func(*opensearchapi.SearchRequest) {
+	opts := []func(*opensearchapi.SearchRequest){
+		c.client.Search.WithContext(ctx),
+	}
+	if req.Index != nil {
+		opts = append(opts, c.client.Search.WithIndex(*req.Index))
+	}
+	if req.ReturnVersion != nil {
+		opts = append(opts, c.client.Search.WithVersion(*req.ReturnVersion))
+	}
+	if req.Size != nil {
+		opts = append(opts, c.client.Search.WithSize(*req.Size))
+	}
+	if req.From != nil {
+		opts = append(opts, c.client.Search.WithFrom(*req.From))
+	}
+	if req.Sort != nil {
+		opts = append(opts, c.client.Search.WithSort(*req.Sort))
+	}
+	if req.Query != nil {
+		opts = append(opts, c.client.Search.WithBody(req.Query))
+	}
+	if req.SourceIncludes != nil {
+		opts = append(opts, c.client.Search.WithSourceIncludes(*req.SourceIncludes))
+	}
+
+	return opts
+}
+
+func (c *Client) isErrResponse(res *opensearchapi.Response) error {
+	return searchstore.IsErrResponse(newAPIResponse(res))
+}
+
+func newClient(address string) (*opensearch.Client, error) {
+	if address == "" {
+		return nil, errors.New("no address provided")
+	}
+
+	cfg := opensearch.Config{
+		Addresses: []string{
+			address,
+		},
+		Transport: http.DefaultTransport,
+	}
+
+	return opensearch.NewClient(cfg)
+}
+
+type apiResponse struct {
+	*opensearchapi.Response
+}
+
+func newAPIResponse(res *opensearchapi.Response) *apiResponse {
+	return &apiResponse{Response: res}
+}
+
+func (r *apiResponse) GetBody() io.ReadCloser {
+	return r.Body
+}
+
+func (r *apiResponse) GetStatusCode() int {
+	return r.StatusCode
+}
diff --git a/internal/searchstore/opensearch/opensearch_mapper.go b/internal/searchstore/opensearch/opensearch_mapper.go
new file mode 100644
index 0000000..9ae5614
--- /dev/null
+++ b/internal/searchstore/opensearch/opensearch_mapper.go
@@ -0,0 +1,78 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package opensearch
+
+import (
+	"github.com/xataio/pgstream/internal/searchstore"
+)
+
+type Mapper struct{}
+
+const (
+	openSearchDefaultEFSearch = 100
+
+	// Lucene’s term byte-length limit is 32766. To cover for the use of UTF-8
+	// text with many non-ASCII characters, the maximum value should be 32766 /
+	// 4 = 8191 since UTF-8 characters may occupy at most 4 bytes.
+	termByteLengthLimit = 32766
+)
+
+func NewMapper() *Mapper {
+	return &Mapper{}
+}
+
+func (m *Mapper) GetDefaultIndexSettings() map[string]any {
+	return map[string]any{
+		"number_of_shards":                 1,
+		"number_of_replicas":               1,
+		"index.mapping.total_fields.limit": 2000,
+		"index.knn":                        true,
+		"knn.algo_param.ef_search":         openSearchDefaultEFSearch,
+	}
+}
+
+func (m *Mapper) FieldMapping(field *searchstore.Field) (map[string]any, error) {
+	switch field.SearchType {
+	case searchstore.IntegerType:
+		return map[string]any{"type": "long"}, nil
+	case searchstore.FloatType:
+		return map[string]any{"type": "double"}, nil
+	case searchstore.BoolType:
+		return map[string]any{"type": "boolean"}, nil
+	case searchstore.TextType, searchstore.JSONType:
+		return map[string]any{"type": "text"}, nil
+	case searchstore.StringType:
+		return map[string]any{
+			"type":         "keyword",
+			"ignore_above": termByteLengthLimit,
+			"fields": map[string]any{
+				"text": map[string]any{
+					"type": "text",
+				},
+			},
+		}, nil
+	case searchstore.TimeType:
+		return map[string]any{
+			"type":   "date",
+			"format": "HH:mm:ss[.SS][x][Z]||HH:mm:ss[.SSS][x][Z]||HH:mm:ss[.SSSSSS][x][Z]",
+		}, nil
+	case searchstore.DateType:
+		return map[string]any{
+			"type":   "date",
+			"format": "date",
+		}, nil
+	case searchstore.DateTimeType, searchstore.DateTimeTZType:
+		return map[string]any{
+			"type":   "date",
+			"format": "yyyy-MM-dd HH:mm:ss[.SSS][x]||yyyy-MM-dd HH:mm:ss[.SS][x]||yyyy-MM-dd HH:mm:ss[.S][x]||yyyy-MM-dd'T'HH:mm:ss[.SSS][X]",
+		}, nil
+	case searchstore.PGVectorType:
+		vectorSettings := map[string]any{
+			"type":      "knn_vector",
+			"dimension": field.Metadata.VectorDimension,
+		}
+		return vectorSettings, nil
+	default:
+		return nil, searchstore.ErrUnsupportedSearchFieldType
+	}
+}
diff --git a/internal/es/es_api.go b/internal/searchstore/search_api.go
similarity index 97%
rename from internal/es/es_api.go
rename to internal/searchstore/search_api.go
index fc490c9..553adc0 100644
--- a/internal/es/es_api.go
+++ b/internal/searchstore/search_api.go
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package es
+package searchstore
 
 import (
 	"bytes"
@@ -250,11 +250,11 @@ type Mappings struct {
 	Dynamic    string
 }
 
-type mappingResponse map[string]struct {
+type MappingResponse map[string]struct {
 	Mappings Mappings
 }
 
-type indexStatsResponse struct {
+type IndexStatsResponse struct {
 	Indices map[string]struct {
 		Primaries struct {
 			Store struct {
@@ -269,11 +269,11 @@ type indexStatsResponse struct {
 	} `json:"indices"`
 }
 
-type countResponse struct {
+type CountResponse struct {
 	Count int `json:"count"`
 }
 
-func encodeBulkItems(buffer *bytes.Buffer, items []BulkItem) error {
+func EncodeBulkItems(buffer *bytes.Buffer, items []BulkItem) error {
 	encoder := json.NewEncoder(buffer)
 
 	for _, item := range items {
diff --git a/internal/searchstore/search_client.go b/internal/searchstore/search_client.go
new file mode 100644
index 0000000..19b7d69
--- /dev/null
+++ b/internal/searchstore/search_client.go
@@ -0,0 +1,76 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package searchstore
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+)
+
+type Client interface {
+	CloseIndex(ctx context.Context, index string) error
+	Count(ctx context.Context, index string) (int, error)
+	CreateIndex(ctx context.Context, index string, body map[string]any) error
+	DeleteByQuery(ctx context.Context, req *DeleteByQueryRequest) error
+	DeleteIndex(ctx context.Context, index []string) error
+	GetIndexAlias(ctx context.Context, name string) (map[string]any, error)
+	GetIndexMappings(ctx context.Context, index string) (*Mappings, error)
+	GetIndicesStats(ctx context.Context, indexPattern string) ([]IndexStats, error)
+	Index(ctx context.Context, req *IndexRequest) error
+	IndexWithID(ctx context.Context, req *IndexWithIDRequest) error
+	IndexExists(ctx context.Context, index string) (bool, error)
+	ListIndices(ctx context.Context, indices []string) ([]string, error)
+	Perform(req *http.Request) (*http.Response, error)
+	PutIndexAlias(ctx context.Context, index []string, name string) error
+	PutIndexMappings(ctx context.Context, index string, body map[string]any) error
+	PutIndexSettings(ctx context.Context, index string, body map[string]any) error
+	RefreshIndex(ctx context.Context, index string) error
+	Search(ctx context.Context, req *SearchRequest) (*SearchResponse, error)
+	SendBulkRequest(ctx context.Context, items []BulkItem) ([]BulkItem, error)
+	GetMapper() Mapper
+}
+
+func Ptr[T any](i T) *T { return &i }
+
+// createReader returns a reader on the JSON representation of the given value.
+func CreateReader(value any) (*bytes.Reader, error) {
+	bytesValue, err := json.Marshal(value)
+	if err != nil {
+		return nil, fmt.Errorf("unexpected marshaling error: %w", err)
+	}
+	return bytes.NewReader(bytesValue), nil
+}
+
+func VerifyResponse(bodyBytes []byte, items []BulkItem) (failed []BulkItem, err error) {
+	var response BulkResponse
+
+	if err := json.Unmarshal(bodyBytes, &response); err != nil {
+		return nil, fmt.Errorf("error unmarshaling response from search store: %w (%s)", err, bodyBytes)
+	}
+
+	if !response.Errors {
+		return []BulkItem{}, nil
+	}
+
+	failed = []BulkItem{}
+	for i, respItem := range response.Items {
+		if items[i].Index != nil {
+			if respItem.Index.Status > 299 {
+				items[i].Status = respItem.Index.Status
+				items[i].Error = respItem.Index.Error
+				failed = append(failed, items[i])
+			}
+		} else if items[i].Delete != nil {
+			if respItem.Delete.Status > 299 {
+				items[i].Status = respItem.Delete.Status
+				items[i].Error = respItem.Delete.Error
+				failed = append(failed, items[i])
+			}
+		}
+	}
+
+	return failed, nil
+}
diff --git a/internal/es/es_errors.go b/internal/searchstore/search_errors.go
similarity index 75%
rename from internal/es/es_errors.go
rename to internal/searchstore/search_errors.go
index 42e3b48..ecbbe8e 100644
--- a/internal/es/es_errors.go
+++ b/internal/searchstore/search_errors.go
@@ -1,15 +1,15 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package es
+package searchstore
 
 import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"strings"
 
-	"github.com/elastic/go-elasticsearch/v8/esapi"
 	"github.com/mitchellh/mapstructure"
 )
 
@@ -79,15 +79,34 @@ const (
 )
 
 var (
-	ErrTooManyRequests      = errors.New("too many requests")
-	ErrTooManyBuckets       = errors.New("too many buckets")
-	ErrTooManyNestedClauses = errors.New("too many nested clauses")
-	ErrTooManyClauses       = errors.New("too many clauses")
+	ErrTooManyRequests            = errors.New("too many requests")
+	ErrTooManyBuckets             = errors.New("too many buckets")
+	ErrTooManyNestedClauses       = errors.New("too many nested clauses")
+	ErrTooManyClauses             = errors.New("too many clauses")
+	ErrUnsupportedSearchFieldType = errors.New("unsupported search field type")
+	ErrResourceNotFound           = errors.New("search resource not found")
 )
 
-func extractResponseError(res *esapi.Response) error {
+type apiResponse interface {
+	GetBody() io.ReadCloser
+	GetStatusCode() int
+	IsError() bool
+}
+
+func IsErrResponse(res apiResponse) error {
+	if res.IsError() {
+		if res.GetStatusCode() == http.StatusNotFound {
+			return fmt.Errorf("%w: %w", ErrResourceNotFound, extractResponseError(res))
+		}
+		return extractResponseError(res)
+	}
+
+	return nil
+}
+
+func extractResponseError(res apiResponse) error {
 	var e map[string]any
-	if err := json.NewDecoder(res.Body).Decode(&e); err != nil {
+	if err := json.NewDecoder(res.GetBody()).Decode(&e); err != nil {
 		return fmt.Errorf("decoding error response: %w", err)
 	}
 
@@ -131,21 +150,22 @@ func extractResponseError(res *esapi.Response) error {
 		}
 	}
 
-	if err, ok := getRetryableError(res.StatusCode); ok {
+	statusCode := res.GetStatusCode()
+	if err, ok := getRetryableError(statusCode); ok {
 		return RetryableError{Cause: err}
 	}
 
-	if res.StatusCode == http.StatusNotFound {
-		return fmt.Errorf("%w: [%d]: %s: %s", ErrResourceNotFound, res.StatusCode, errType, errReason)
+	if statusCode == http.StatusNotFound {
+		return fmt.Errorf("%w: [%d]: %s: %s", ErrResourceNotFound, statusCode, errType, errReason)
 	}
 
-	if res.StatusCode == http.StatusBadRequest {
+	if statusCode == http.StatusBadRequest {
 		switch errType {
 		case ResourceAlreadyExistsException:
 			reason, _ := errReason.(string)
 			return ErrResourceAlreadyExists{Reason: reason}
 		case SnapshotInProgressException:
-			return RetryableError{Cause: fmt.Errorf("[%d] %s: %s", res.StatusCode, errType, errReason)}
+			return RetryableError{Cause: fmt.Errorf("[%d] %s: %s", statusCode, errType, errReason)}
 		default:
 			// Generic bad request
 			return ErrQueryInvalid{
@@ -154,7 +174,7 @@ func extractResponseError(res *esapi.Response) error {
 		}
 	}
 
-	return fmt.Errorf("[%d] %s: %s", res.StatusCode, errType, errReason)
+	return fmt.Errorf("[%d] %s: %s", statusCode, errType, errReason)
 }
 
 func getRetryableError(statusCode int) (error, bool) {
diff --git a/internal/searchstore/search_mapper.go b/internal/searchstore/search_mapper.go
new file mode 100644
index 0000000..7d4a1e9
--- /dev/null
+++ b/internal/searchstore/search_mapper.go
@@ -0,0 +1,34 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package searchstore
+
+type Mapper interface {
+	GetDefaultIndexSettings() map[string]any
+	FieldMapping(*Field) (map[string]any, error)
+}
+
+type Field struct {
+	SearchType Type
+	IsArray    bool
+	Metadata   Metadata
+}
+
+type Metadata struct {
+	VectorDimension int
+}
+
+type Type uint
+
+const (
+	IntegerType Type = iota
+	FloatType
+	BoolType
+	StringType
+	DateTimeTZType
+	DateTimeType
+	DateType
+	TimeType
+	JSONType
+	TextType
+	PGVectorType
+)
diff --git a/kafka2os.env b/kafka2os.env
index f9fa72e..cfb226a 100644
--- a/kafka2os.env
+++ b/kafka2os.env
@@ -13,7 +13,7 @@ PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT=5s
 PGSTREAM_SEARCH_INDEXER_CLEANUP_BACKOFF_INITIAL_INTERVAL=1s
 PGSTREAM_SEARCH_INDEXER_CLEANUP_BACKOFF_MAX_INTERVAL=1m
 PGSTREAM_SEARCH_INDEXER_CLEANUP_BACKOFF_MAX_RETRIES=60
-PGSTREAM_SEARCH_STORE_URL="http://admin:admin@localhost:9200"
+PGSTREAM_OPENSEARCH_STORE_URL="http://admin:admin@localhost:9200"
 PGSTREAM_SEARCH_STORE_BACKOFF_INITIAL_INTERVAL=1s
 PGSTREAM_SEARCH_STORE_BACKOFF_MAX_INTERVAL=1m
 PGSTREAM_SEARCH_STORE_BACKOFF_MAX_RETRIES=0
diff --git a/pg2os.env b/pg2os.env
index 98966ed..8707293 100644
--- a/pg2os.env
+++ b/pg2os.env
@@ -8,7 +8,7 @@ PGSTREAM_SEARCH_INDEXER_BATCH_TIMEOUT=5s
 PGSTREAM_SEARCH_INDEXER_CLEANUP_EXP_BACKOFF_INITIAL_INTERVAL=1s
 PGSTREAM_SEARCH_INDEXER_CLEANUP_EXP_BACKOFF_MAX_INTERVAL=1m
 PGSTREAM_SEARCH_INDEXER_CLEANUP_EXP_BACKOFF_MAX_RETRIES=60
-PGSTREAM_SEARCH_STORE_URL="http://admin:admin@localhost:9200"
+PGSTREAM_OPENSEARCH_STORE_URL="http://admin:admin@localhost:9200"
 PGSTREAM_SEARCH_STORE_EXP_BACKOFF_INITIAL_INTERVAL=1s
 PGSTREAM_SEARCH_STORE_EXP_BACKOFF_MAX_INTERVAL=1m
 PGSTREAM_SEARCH_STORE_EXP_BACKOFF_MAX_RETRIES=0
diff --git a/pkg/stream/config.go b/pkg/stream/config.go
index 12a9a1c..593a25e 100644
--- a/pkg/stream/config.go
+++ b/pkg/stream/config.go
@@ -10,7 +10,7 @@ import (
 	kafkacheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/kafka"
 	kafkaprocessor "github.com/xataio/pgstream/pkg/wal/processor/kafka"
 	"github.com/xataio/pgstream/pkg/wal/processor/search"
-	"github.com/xataio/pgstream/pkg/wal/processor/search/opensearch"
+	"github.com/xataio/pgstream/pkg/wal/processor/search/store"
 	"github.com/xataio/pgstream/pkg/wal/processor/translator"
 	"github.com/xataio/pgstream/pkg/wal/processor/webhook/notifier"
 	"github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription/server"
@@ -49,7 +49,7 @@ type KafkaProcessorConfig struct {
 
 type SearchProcessorConfig struct {
 	Indexer search.IndexerConfig
-	Store   opensearch.Config
+	Store   store.Config
 	Retrier search.StoreRetryConfig
 }
 
diff --git a/pkg/stream/stream_run.go b/pkg/stream/stream_run.go
index ce95401..6a2d4e1 100644
--- a/pkg/stream/stream_run.go
+++ b/pkg/stream/stream_run.go
@@ -18,7 +18,7 @@ import (
 	processinstrumentation "github.com/xataio/pgstream/pkg/wal/processor/instrumentation"
 	kafkaprocessor "github.com/xataio/pgstream/pkg/wal/processor/kafka"
 	"github.com/xataio/pgstream/pkg/wal/processor/search"
-	"github.com/xataio/pgstream/pkg/wal/processor/search/opensearch"
+	"github.com/xataio/pgstream/pkg/wal/processor/search/store"
 	"github.com/xataio/pgstream/pkg/wal/processor/translator"
 	webhooknotifier "github.com/xataio/pgstream/pkg/wal/processor/webhook/notifier"
 	subscriptionserver "github.com/xataio/pgstream/pkg/wal/processor/webhook/subscription/server"
@@ -120,7 +120,7 @@ func Run(ctx context.Context, logger loglib.Logger, config *Config, meter metric
 	case config.Processor.Search != nil:
 		var searchStore search.Store
 		var err error
-		searchStore, err = opensearch.NewStore(config.Processor.Search.Store, opensearch.WithLogger(logger))
+		searchStore, err = store.NewStore(config.Processor.Search.Store, store.WithLogger(logger))
 		if err != nil {
 			return err
 		}
diff --git a/pkg/wal/processor/search/opensearch/opensearch_adapter.go b/pkg/wal/processor/search/store/search_adapter.go
similarity index 84%
rename from pkg/wal/processor/search/opensearch/opensearch_adapter.go
rename to pkg/wal/processor/search/store/search_adapter.go
index 30c8777..81cdf95 100644
--- a/pkg/wal/processor/search/opensearch/opensearch_adapter.go
+++ b/pkg/wal/processor/search/store/search_adapter.go
@@ -1,20 +1,20 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package opensearch
+package store
 
 import (
 	"encoding/json"
 	"fmt"
 
-	"github.com/xataio/pgstream/internal/es"
+	"github.com/xataio/pgstream/internal/searchstore"
 	"github.com/xataio/pgstream/pkg/schemalog"
 	"github.com/xataio/pgstream/pkg/wal/processor/search"
 )
 
 // Adapter converts from/to search types and opensearch types
 type SearchAdapter interface {
-	SearchDocToBulkItem(docs search.Document) es.BulkItem
-	BulkItemsToSearchDocErrs(items []es.BulkItem) []search.DocumentError
+	SearchDocToBulkItem(docs search.Document) searchstore.BulkItem
+	BulkItemsToSearchDocErrs(items []searchstore.BulkItem) []search.DocumentError
 	RecordToLogEntry(rec map[string]any) (*schemalog.LogEntry, error)
 }
 
@@ -32,12 +32,12 @@ func newDefaultAdapter(indexNameAdapter IndexNameAdapter) *adapter {
 	}
 }
 
-func (a *adapter) SearchDocToBulkItem(doc search.Document) es.BulkItem {
+func (a *adapter) SearchDocToBulkItem(doc search.Document) searchstore.BulkItem {
 	indexName := a.indexNameAdapter.SchemaNameToIndex(doc.Schema)
-	item := es.BulkItem{
+	item := searchstore.BulkItem{
 		Doc: doc.Data,
 	}
-	bulkIndex := &es.BulkIndex{
+	bulkIndex := &searchstore.BulkIndex{
 		Index:       indexName.Name(),
 		ID:          doc.ID,
 		Version:     &doc.Version,
@@ -51,7 +51,7 @@ func (a *adapter) SearchDocToBulkItem(doc search.Document) es.BulkItem {
 	return item
 }
 
-func (a *adapter) BulkItemsToSearchDocErrs(items []es.BulkItem) []search.DocumentError {
+func (a *adapter) BulkItemsToSearchDocErrs(items []searchstore.BulkItem) []search.DocumentError {
 	if items == nil {
 		return nil
 	}
@@ -76,7 +76,7 @@ func (a *adapter) RecordToLogEntry(rec map[string]any) (*schemalog.LogEntry, err
 	return &log, nil
 }
 
-func (a *adapter) bulkItemToSearchDocErr(item es.BulkItem) search.DocumentError {
+func (a *adapter) bulkItemToSearchDocErr(item searchstore.BulkItem) search.DocumentError {
 	doc := search.DocumentError{
 		Document: search.Document{
 			Data: item.Doc,
diff --git a/pkg/wal/processor/search/opensearch/opensearch_index_name.go b/pkg/wal/processor/search/store/search_index_name.go
similarity index 98%
rename from pkg/wal/processor/search/opensearch/opensearch_index_name.go
rename to pkg/wal/processor/search/store/search_index_name.go
index 85a9d80..2597e80 100644
--- a/pkg/wal/processor/search/opensearch/opensearch_index_name.go
+++ b/pkg/wal/processor/search/store/search_index_name.go
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package opensearch
+package store
 
 import (
 	"fmt"
diff --git a/pkg/wal/processor/search/opensearch/opensearch_pg_mapper.go b/pkg/wal/processor/search/store/search_pg_mapper.go
similarity index 54%
rename from pkg/wal/processor/search/opensearch/opensearch_pg_mapper.go
rename to pkg/wal/processor/search/store/search_pg_mapper.go
index 70123a6..15d906e 100644
--- a/pkg/wal/processor/search/opensearch/opensearch_pg_mapper.go
+++ b/pkg/wal/processor/search/store/search_pg_mapper.go
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package opensearch
+package store
 
 import (
 	"encoding/json"
@@ -11,140 +11,71 @@ import (
 
 	"github.com/jackc/pgx/v5/pgtype"
 
+	"github.com/xataio/pgstream/internal/searchstore"
 	"github.com/xataio/pgstream/pkg/schemalog"
 	"github.com/xataio/pgstream/pkg/wal/processor/search"
 )
 
-type Mapper struct {
-	pgTypeMap *pgtype.Map
-}
-
-type searchType uint
-
-const (
-	searchTypeInteger searchType = iota
-	searchTypeFloat
-	searchTypeBool
-	searchTypeString
-	searchTypeDateTimeTZ
-	searchTypeDateTime
-	searchTypeDate
-	searchTypeTime
-	searchTypeJSON
-	searchTypeText
-	searchTypePGVector
-)
-
-type searchField struct {
-	searchType searchType
-	isArray    bool
-	metadata   metadata
-}
-
-type metadata struct {
-	vectorDimension int
+type PgMapper struct {
+	searchMapper searchstore.Mapper
+	pgTypeMap    *pgtype.Map
 }
 
 const (
-	// Lucene’s term byte-length limit is 32766. To cover for the use of UTF-8
-	// text with many non-ASCII characters, the maximum value should be 32766 /
-	// 4 = 8191 since UTF-8 characters may occupy at most 4 bytes.
-	termByteLengthLimit = 32766
-
-	// Default ES date_time pattern
+	// Default date_time pattern
 	timestampTZFormat = "2006-01-02T15:04:05.000Z"
 	timestampFormat   = "2006-01-02T15:04:05.000"
 	dateFormat        = "2006-01-02"
 )
 
-// NewPostgresMapper returns a mapper that maps between postgres and opensearch
-// types
-func NewPostgresMapper() *Mapper {
-	return &Mapper{
-		pgTypeMap: pgtype.NewMap(),
+// NewPostgresMapper returns a mapper that maps between postgres and search
+// store types
+func NewPostgresMapper(mapper searchstore.Mapper) *PgMapper {
+	return &PgMapper{
+		searchMapper: mapper,
+		pgTypeMap:    pgtype.NewMap(),
 	}
 }
 
 // ColumnToSearchMapping maps the column on input into the equivalent search mapping
-func (m *Mapper) ColumnToSearchMapping(column schemalog.Column) (map[string]any, error) {
+func (m *PgMapper) ColumnToSearchMapping(column schemalog.Column) (map[string]any, error) {
 	searchField, err := m.columnToSearchField(column)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse pg type (%s): %w", column.DataType, err)
 	}
 
-	switch searchField.searchType {
-	case searchTypeInteger:
-		return map[string]any{"type": "long"}, nil
-	case searchTypeFloat:
-		return map[string]any{"type": "double"}, nil
-	case searchTypeBool:
-		return map[string]any{"type": "boolean"}, nil
-	case searchTypeText, searchTypeJSON:
-		return map[string]any{"type": "text"}, nil
-	case searchTypeString:
-		return map[string]any{
-			"type":         "keyword",
-			"ignore_above": termByteLengthLimit,
-			"fields": map[string]any{
-				"text": map[string]any{
-					"type": "text",
-				},
-			},
-		}, nil
-	case searchTypeTime:
-		return map[string]any{
-			"type":   "date",
-			"format": "HH:mm:ss[.SS][x][Z]||HH:mm:ss[.SSS][x][Z]||HH:mm:ss[.SSSSSS][x][Z]",
-		}, nil
-	case searchTypeDate:
-		return map[string]any{
-			"type":   "date",
-			"format": "date",
-		}, nil
-	case searchTypeDateTime, searchTypeDateTimeTZ:
-		return map[string]any{
-			"type":   "date",
-			"format": "yyyy-MM-dd HH:mm:ss[.SSS][x]||yyyy-MM-dd HH:mm:ss[.SS][x]||yyyy-MM-dd HH:mm:ss[.S][x]||yyyy-MM-dd'T'HH:mm:ss[.SSS][X]",
-		}, nil
-	case searchTypePGVector:
-		vectorSettings := map[string]any{
-			"type":      "knn_vector",
-			"dimension": searchField.metadata.vectorDimension,
-		}
-		return vectorSettings, nil
-	default:
-		return nil, err
-	}
+	return m.searchMapper.FieldMapping(searchField)
 }
 
-// MapColumnValue maps a value emitted from PG into a value that OS can handle.
-// If the column is a timestamp: we need to parse it.
-// If the column is an array of any type except json, we need to map it to a Go slice.
-// If column type is unknown we return nil. This avoids dropping the whole record if one field type is unknown.
-func (m *Mapper) MapColumnValue(column schemalog.Column, value any) (any, error) {
+// MapColumnValue maps a value emitted from PG into a value that the search
+// store can handle. If the column is a timestamp: we need to parse it. If the
+// column is an array of any type except json, we need to map it to a Go slice.
+// If column type is unknown we return nil. This avoids dropping the whole
+// record if one field type is unknown.
+func (m *PgMapper) MapColumnValue(column schemalog.Column, value any) (any, error) {
 	searchField, err := m.columnToSearchField(column)
 	if err != nil {
-		return nil, fmt.Errorf("mapping column from pg to os: %w", err)
+		return nil, fmt.Errorf("mapping column from pg to search store: %w", err)
 	}
 
 	if value == nil {
 		return nil, nil
 	}
 
-	switch searchField.searchType {
-	case searchTypeDateTimeTZ, searchTypeDateTime:
-		if searchField.isArray {
+	switch searchField.SearchType {
+	case searchstore.DateTimeTZType, searchstore.DateTimeType:
+		if searchField.IsArray {
 			return m.mapDateTimeArray(searchField, value)
 		} else {
 			return m.mapDateTime(searchField, value)
 		}
-	case searchTypeDate:
+	case searchstore.DateType:
 		var d pgtype.Date
 		if err := d.Scan(value); err != nil {
-			return nil, fmt.Errorf("mapping date from pg to ES failed: %w (value: %s)", err, value)
+			return nil, fmt.Errorf("mapping date from pg to search store failed: %w (value: %s)", err, value)
 		}
 		return d.Time.Format(dateFormat), nil
-	case searchTypePGVector:
+	case searchstore.PGVectorType:
 		// pgvector vectors come as strings. We need to parse them into arrays of floats.
 		stringContent, ok := value.(string)
 		if !ok {
@@ -157,30 +88,30 @@ func (m *Mapper) MapColumnValue(column schemalog.Column, value any) (any, error)
 		}
 		return array, nil
 	default:
-		if searchField.isArray { // catches all other array types
+		if searchField.IsArray { // catches all other array types
 			// handle arrays
-			switch searchField.searchType {
-			case searchTypeInteger:
+			switch searchField.SearchType {
+			case searchstore.IntegerType:
 				var a pgtype.FlatArray[int64]
 				err := m.pgTypeMap.SQLScanner(&a).Scan(value)
 				return []int64(a), err
-			case searchTypeFloat:
+			case searchstore.FloatType:
 				var a pgtype.FlatArray[float64]
 				err := m.pgTypeMap.SQLScanner(&a).Scan(value)
 				return []float64(a), err
-			case searchTypeBool:
+			case searchstore.BoolType:
 				var a pgtype.FlatArray[bool]
 				err := m.pgTypeMap.SQLScanner(&a).Scan(value)
 				return []bool(a), err
-			case searchTypeString:
+			case searchstore.StringType:
 				var a pgtype.FlatArray[string]
 				err := m.pgTypeMap.SQLScanner(&a).Scan(value)
 				return []string(a), err
-			case searchTypeJSON:
+			case searchstore.JSONType:
 				// nothing to do for json array types
 			default:
 				// should never get here
-				panic(fmt.Sprintf("indexer: unexpected array type: %v", searchField.searchType))
+				panic(fmt.Sprintf("indexer: unexpected array type: %v", searchField.SearchType))
 			}
 		}
 	}
@@ -189,40 +120,40 @@ func (m *Mapper) MapColumnValue(column schemalog.Column, value any) (any, error)
 	return value, nil
 }
 
-func (m *Mapper) columnToSearchField(column schemalog.Column) (*searchField, error) {
+func (m *PgMapper) columnToSearchField(column schemalog.Column) (*searchstore.Field, error) {
 	pgTypeName := column.DataType
 	typeName, isArray, err := m.parsePGType(pgTypeName)
 	if err != nil {
 		return nil, fmt.Errorf("pg to search type: failed to parse pg type: %w", err)
 	}
 
-	metadata := metadata{}
+	metadata := searchstore.Metadata{}
 
-	var searchType searchType
+	var searchType searchstore.Type
 	switch typeName {
 	case "int8", "int2", "int4", "integer", "smallint", "bigint":
-		searchType = searchTypeInteger
+		searchType = searchstore.IntegerType
 	case "float4", "float8", "real", "double precision", "float", "numeric":
-		searchType = searchTypeFloat
+		searchType = searchstore.FloatType
 	case "boolean":
-		searchType = searchTypeBool
+		searchType = searchstore.BoolType
 	case "bytea", "char", "name", "text", "varchar", "bpchar", "xml", "uuid", "character varying", "character", "cidr", "inet", "macaddr", "macaddr8", "interval":
-		searchType = searchTypeString
+		searchType = searchstore.StringType
 	case "jsonb", "json":
-		searchType = searchTypeJSON
+		searchType = searchstore.JSONType
 	case "date":
-		searchType = searchTypeDate
+		searchType = searchstore.DateType
 	case "time", "time with time zone", "time without time zone":
-		searchType = searchTypeTime
+		searchType = searchstore.TimeType
 	case "timestamp", "timestamp without time zone":
-		searchType = searchTypeDateTime
+		searchType = searchstore.DateTimeType
 	case "timestamptz", "timetz", "timestamp with time zone":
-		searchType = searchTypeDateTimeTZ
+		searchType = searchstore.DateTimeTZType
 	default:
 		// pgvector includes the schema (sometimes? seems only a problem when testing locally)
 		if isPGVector(typeName) {
-			searchType = searchTypePGVector
-			metadata.vectorDimension, err = getPGVectorDimension(typeName)
+			searchType = searchstore.PGVectorType
+			metadata.VectorDimension, err = getPGVectorDimension(typeName)
 			if err != nil {
 				return nil, search.ErrTypeInvalid{Input: pgTypeName}
 			}
@@ -231,20 +162,20 @@ func (m *Mapper) columnToSearchField(column schemalog.Column) (*searchField, err
 		}
 	}
 
-	return &searchField{
-		searchType: searchType,
-		isArray:    isArray,
-		metadata:   metadata,
+	return &searchstore.Field{
+		SearchType: searchType,
+		IsArray:    isArray,
+		Metadata:   metadata,
 	}, nil
 }
 
-func (m *Mapper) mapDateTimeArray(searchField *searchField, value any) (any, error) {
-	switch searchField.searchType {
-	case searchTypeDateTimeTZ:
+func (m *PgMapper) mapDateTimeArray(searchField *searchstore.Field, value any) (any, error) {
+	switch searchField.SearchType {
+	case searchstore.DateTimeTZType:
 		var a pgtype.FlatArray[pgtype.Timestamptz]
 		err := m.pgTypeMap.SQLScanner(&a).Scan(value)
 		if err != nil {
-			return nil, fmt.Errorf("mapping timestamptz array from pg to ES failed: %w (value: %s)", err, value)
+			return nil, fmt.Errorf("mapping timestamptz array from pg to search store failed: %w (value: %s)", err, value)
 		}
 
 		dts := make([]string, len(a))
@@ -254,11 +185,11 @@ func (m *Mapper) mapDateTimeArray(searchField *searchField, value any) (any, err
 		}
 
 		return dts, nil
-	case searchTypeDateTime:
+	case searchstore.DateTimeType:
 		var a pgtype.FlatArray[pgtype.Timestamp]
 		err := m.pgTypeMap.SQLScanner(&a).Scan(value)
 		if err != nil {
-			return nil, fmt.Errorf("mapping timestampt array from pg to ES failed: %w (value: %s)", err, value)
+			return nil, fmt.Errorf("mapping timestampt array from pg to search store failed: %w (value: %s)", err, value)
 		}
 
 		dts := make([]string, len(a))
@@ -272,25 +203,25 @@ func (m *Mapper) mapDateTimeArray(searchField *searchField, value any) (any, err
 	return value, nil
 }
 
-func (m *Mapper) mapDateTime(searchField *searchField, value any) (any, error) {
-	switch searchField.searchType {
-	case searchTypeDateTimeTZ:
+func (m *PgMapper) mapDateTime(searchField *searchstore.Field, value any) (any, error) {
+	switch searchField.SearchType {
+	case searchstore.DateTimeTZType:
 		var ts pgtype.Timestamptz
 		if err := ts.Scan(value); err != nil {
-			return nil, fmt.Errorf("mapping timestamptz from pg to ES failed: %w (value: %s)", err, value)
+			return nil, fmt.Errorf("mapping timestamptz from pg to search store failed: %w (value: %s)", err, value)
 		}
 		return ts.Time.Truncate(time.Millisecond).Format(timestampTZFormat), nil
-	case searchTypeDateTime:
+	case searchstore.DateTimeType:
 		var ts pgtype.Timestamp
 		if err := ts.Scan(value); err != nil {
-			return nil, fmt.Errorf("mapping timestamp from pg to ES failed: %w (value: %s)", err, value)
+			return nil, fmt.Errorf("mapping timestamp from pg to search store failed: %w (value: %s)", err, value)
 		}
 		return ts.Time.Truncate(time.Millisecond).Format(timestampFormat), nil
 	}
 	return value, nil
 }
 
-func (m *Mapper) parsePGType(name string) (typeName string, isArray bool, err error) {
+func (m *PgMapper) parsePGType(name string) (typeName string, isArray bool, err error) {
 	inputName := name
 
 	if strings.HasSuffix(name, "[]") { // detect and strip array suffix. this is always last.
diff --git a/pkg/wal/processor/search/opensearch/opensearch_store.go b/pkg/wal/processor/search/store/search_store.go
similarity index 79%
rename from pkg/wal/processor/search/opensearch/opensearch_store.go
rename to pkg/wal/processor/search/store/search_store.go
index 2628f53..6c55dc4 100644
--- a/pkg/wal/processor/search/opensearch/opensearch_store.go
+++ b/pkg/wal/processor/search/store/search_store.go
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package opensearch
+package store
 
 import (
 	"bytes"
@@ -9,47 +9,59 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/xataio/pgstream/internal/es"
+	"github.com/xataio/pgstream/internal/searchstore"
+	elasticsearchstore "github.com/xataio/pgstream/internal/searchstore/elasticsearch"
+	opensearchstore "github.com/xataio/pgstream/internal/searchstore/opensearch"
+
 	loglib "github.com/xataio/pgstream/pkg/log"
 	"github.com/xataio/pgstream/pkg/schemalog"
 	"github.com/xataio/pgstream/pkg/wal/processor/search"
 )
 
 type Store struct {
-	logger           loglib.Logger
-	client           es.SearchClient
-	mapper           search.Mapper
-	adapter          SearchAdapter
-	indexNameAdapter IndexNameAdapter
-	marshaler        func(any) ([]byte, error)
+	logger               loglib.Logger
+	client               searchstore.Client
+	mapper               search.Mapper
+	adapter              SearchAdapter
+	indexNameAdapter     IndexNameAdapter
+	marshaler            func(any) ([]byte, error)
+	defaultIndexSettings map[string]any
 }
 
 type Config struct {
-	URL string
+	OpenSearchURL    string
+	ElasticsearchURL string
 }
 
 type Option func(*Store)
 
 const (
-	openSearchDefaultANNEngine      = "nmslib"
-	openSearchDefaultM              = 48
-	openSearchDefaultEFConstruction = 256
-	openSearchDefaultEFSearch       = 100
-
-	// OpenSearch has a limit of 512 bytes for the ID field. see here:
+	// OpenSearch/Elasticsearch have a limit of 512 bytes for the ID field. see here:
 	// https://www.elastic.co/guide/en/elasticsearch/reference/7.10/mapping-id-field.html
-	osIDFieldLengthLimit = 512
+	idFieldLengthLimit = 512
 
 	schemalogIndexName = "pgstream"
 )
 
 func NewStore(cfg Config, opts ...Option) (*Store, error) {
-	os, err := es.NewClient(cfg.URL)
+	var searchStore searchstore.Client
+	var err error
+	switch {
+	case cfg.OpenSearchURL != "" && cfg.ElasticsearchURL != "":
+		return nil, errors.New("only one store URL must be provided")
+	case cfg.OpenSearchURL == "" && cfg.ElasticsearchURL == "":
+		return nil, errors.New("a store URL must be provided")
+	case cfg.OpenSearchURL != "":
+		searchStore, err = opensearchstore.NewClient(cfg.OpenSearchURL)
+	case cfg.ElasticsearchURL != "":
+		searchStore, err = elasticsearchstore.NewClient(cfg.ElasticsearchURL)
+	}
 	if err != nil {
-		return nil, fmt.Errorf("create elasticsearch client: %w", err)
+		return nil, fmt.Errorf("create search store client: %w", err)
 	}
 
-	s := NewStoreWithClient(os)
+	s := NewStoreWithClient(searchStore)
+
 	for _, opt := range opts {
 		opt(s)
 	}
@@ -57,15 +69,17 @@ func NewStore(cfg Config, opts ...Option) (*Store, error) {
 	return s, nil
 }
 
-func NewStoreWithClient(client es.SearchClient) *Store {
+func NewStoreWithClient(client searchstore.Client) *Store {
+	mapper := client.GetMapper()
 	indexNameAdapter := newDefaultIndexNameAdapter()
 	return &Store{
-		logger:           loglib.NewNoopLogger(),
-		client:           client,
-		indexNameAdapter: indexNameAdapter,
-		adapter:          newDefaultAdapter(indexNameAdapter),
-		mapper:           NewPostgresMapper(),
-		marshaler:        json.Marshal,
+		logger:               loglib.NewNoopLogger(),
+		client:               client,
+		indexNameAdapter:     indexNameAdapter,
+		adapter:              newDefaultAdapter(indexNameAdapter),
+		mapper:               NewPostgresMapper(mapper),
+		marshaler:            json.Marshal,
+		defaultIndexSettings: mapper.GetDefaultIndexSettings(),
 	}
 }
 
@@ -111,7 +125,7 @@ func (s *Store) ApplySchemaChange(ctx context.Context, newEntry *schemalog.LogEn
 
 	// make sure the index and the mapping for the schema exist, and if it
 	// doesn't, align it with latest schema log mapping. This check will allow
-	// us to self recover in case of schema OS index deletion
+	// us to self recover in case of schema search index deletion
 	if err := s.ensureSchemaMapping(ctx, newEntry.SchemaName, existingLogEntry); err != nil {
 		return fmt.Errorf("ensuring schema mapping: %w", err)
 	}
@@ -150,10 +164,10 @@ func (s *Store) ApplySchemaChange(ctx context.Context, newEntry *schemalog.LogEn
 }
 
 func (s *Store) SendDocuments(ctx context.Context, docs []search.Document) ([]search.DocumentError, error) {
-	items := make([]es.BulkItem, 0, len(docs))
+	items := make([]searchstore.BulkItem, 0, len(docs))
 	for _, doc := range docs {
-		if len(doc.ID) > osIDFieldLengthLimit {
-			s.logger.Error(errors.New("ID is longer than 512 bytes"), "opensearch store adapter: error processing document, skipping", loglib.Fields{
+		if len(doc.ID) > idFieldLengthLimit {
+			s.logger.Error(errors.New("ID is longer than 512 bytes"), "error processing document, skipping", loglib.Fields{
 				"severity": "DATALOSS",
 				"id":       doc.ID,
 			})
@@ -183,7 +197,7 @@ func (s *Store) DeleteSchema(ctx context.Context, schemaName string) error {
 	}
 
 	// delete the schema from the schema log index
-	if err := s.client.DeleteByQuery(ctx, &es.DeleteByQueryRequest{
+	if err := s.client.DeleteByQuery(ctx, &searchstore.DeleteByQueryRequest{
 		Index: []string{schemalogIndexName},
 		Query: map[string]any{
 			"query": map[string]any{
@@ -211,10 +225,10 @@ func (s *Store) DeleteTableDocuments(ctx context.Context, schemaName string, tab
 // schema on input. A nil LogEntry will be returned when there's no existing
 // associated logs
 func (s *Store) getLastSchemaLogEntry(ctx context.Context, schemaName string) (*schemalog.LogEntry, error) {
-	query := es.QueryBody{
-		Query: &es.Query{
-			Bool: &es.BoolFilter{
-				Filter: []es.Condition{
+	query := searchstore.QueryBody{
+		Query: &searchstore.Query{
+			Bool: &searchstore.BoolFilter{
+				Filter: []searchstore.Condition{
 					{
 						Term: map[string]any{
 							"schema_name": schemaName,
@@ -230,14 +244,14 @@ func (s *Store) getLastSchemaLogEntry(ctx context.Context, schemaName string) (*
 		return nil, fmt.Errorf("failed to marshal to JSON: %+v, %w", query, err)
 	}
 
-	res, err := s.client.Search(ctx, &es.SearchRequest{
-		Index: es.Ptr(schemalogIndexName),
-		Size:  es.Ptr(1),
-		Sort:  es.Ptr("version:desc"),
+	res, err := s.client.Search(ctx, &searchstore.SearchRequest{
+		Index: searchstore.Ptr(schemalogIndexName),
+		Size:  searchstore.Ptr(1),
+		Sort:  searchstore.Ptr("version:desc"),
 		Query: bytes.NewBuffer(bodyJSON),
 	})
 	if err != nil {
-		if errors.Is(err, es.ErrResourceNotFound) {
+		if errors.Is(err, searchstore.ErrResourceNotFound) {
 			s.logger.Warn(err, "index not found, trying to create it", loglib.Fields{"index": schemalogIndexName})
 			// Create the pgstream index if it was not found.
 			err = s.createSchemaLogIndex(ctx, schemalogIndexName)
@@ -246,7 +260,7 @@ func (s *Store) getLastSchemaLogEntry(ctx context.Context, schemaName string) (*
 			}
 			return nil, search.ErrSchemaNotFound{SchemaName: schemaName}
 		}
-		return nil, fmt.Errorf("get latest schema, failed to search os: %w", mapError(err))
+		return nil, fmt.Errorf("get latest schema, failed to search: %w", mapError(err))
 	}
 
 	if len(res.Hits.Hits) == 0 {
@@ -276,16 +290,10 @@ func (s *Store) createSchema(ctx context.Context, schemaName string) error {
 				},
 			},
 		},
-		"settings": map[string]any{
-			"number_of_shards":                 1,
-			"number_of_replicas":               1,
-			"index.mapping.total_fields.limit": 2000,
-			"index.knn":                        true,
-			"knn.algo_param.ef_search":         openSearchDefaultEFSearch,
-		},
+		"settings": s.defaultIndexSettings,
 	})
 	if err != nil {
-		if errors.As(err, &es.ErrResourceAlreadyExists{}) {
+		if errors.As(err, &searchstore.ErrResourceAlreadyExists{}) {
 			return &search.ErrSchemaAlreadyExists{
 				SchemaName: schemaName,
 			}
@@ -329,7 +337,7 @@ func (s *Store) deleteTableDocuments(ctx context.Context, index IndexName, table
 		return nil
 	}
 
-	req := &es.DeleteByQueryRequest{
+	req := &searchstore.DeleteByQueryRequest{
 		Index: []string{index.Name()},
 		Query: map[string]any{
 			"query": map[string]any{
@@ -417,10 +425,10 @@ func (s *Store) updateMappingAddNewColumns(ctx context.Context, indexName IndexN
 func (s *Store) insertNewSchemaLog(ctx context.Context, m *schemalog.LogEntry) error {
 	logBytes, err := json.Marshal(m)
 	if err != nil {
-		return fmt.Errorf("insert schema log, failed to marshal es doc: %w", err)
+		return fmt.Errorf("insert schema log, failed to marshal search doc: %w", err)
 	}
 
-	err = s.client.IndexWithID(ctx, &es.IndexWithIDRequest{
+	err = s.client.IndexWithID(ctx, &searchstore.IndexWithIDRequest{
 		Index:   schemalogIndexName,
 		ID:      m.ID.String(),
 		Body:    logBytes,
@@ -461,7 +469,7 @@ func (s *Store) ensureSchemaMapping(ctx context.Context, schemaName string, meta
 }
 
 func mapError(err error) error {
-	if errors.As(err, &es.RetryableError{}) {
+	if errors.As(err, &searchstore.RetryableError{}) {
 		return fmt.Errorf("%w: %v", search.ErrRetriable, err.Error())
 	}
 	return err


Test patch:
diff --git a/pkg/stream/integration/helper_test.go b/pkg/stream/integration/helper_test.go
index 22fa624..6249c0d 100644
--- a/pkg/stream/integration/helper_test.go
+++ b/pkg/stream/integration/helper_test.go
@@ -20,7 +20,7 @@ import (
 	"github.com/xataio/pgstream/pkg/wal"
 	kafkacheckpoint "github.com/xataio/pgstream/pkg/wal/checkpointer/kafka"
 	kafkaprocessor "github.com/xataio/pgstream/pkg/wal/processor/kafka"
-	"github.com/xataio/pgstream/pkg/wal/processor/search/opensearch"
+	"github.com/xataio/pgstream/pkg/wal/processor/search/store"
 	"github.com/xataio/pgstream/pkg/wal/processor/translator"
 	"github.com/xataio/pgstream/pkg/wal/processor/webhook"
 	"github.com/xataio/pgstream/pkg/wal/processor/webhook/notifier"
@@ -28,9 +28,10 @@ import (
 )
 
 var (
-	pgurl        string
-	kafkaBrokers []string
-	searchURL    string
+	pgurl            string
+	kafkaBrokers     []string
+	opensearchURL    string
+	elasticsearchURL string
 )
 
 type mockProcessor struct {
@@ -130,12 +131,10 @@ func testKafkaProcessorCfg() stream.ProcessorConfig {
 	}
 }
 
-func testSearchProcessorCfg() stream.ProcessorConfig {
+func testSearchProcessorCfg(storeCfg store.Config) stream.ProcessorConfig {
 	return stream.ProcessorConfig{
 		Search: &stream.SearchProcessorConfig{
-			Store: opensearch.Config{
-				URL: searchURL,
-			},
+			Store: storeCfg,
 		},
 		Translator: &translator.Config{
 			Store: schemalogpg.Config{
diff --git a/pkg/stream/integration/pg_opensearch_integration_test.go b/pkg/stream/integration/pg_opensearch_integration_test.go
deleted file mode 100644
index 35bb2f3..0000000
--- a/pkg/stream/integration/pg_opensearch_integration_test.go
+++ /dev/null
@@ -1,204 +0,0 @@
-// SPDX-License-Identifier: Apache-2.0
-
-package integration
-
-import (
-	"bytes"
-	"context"
-	"encoding/json"
-	"fmt"
-	"os"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/require"
-	"github.com/xataio/pgstream/internal/es"
-	"github.com/xataio/pgstream/pkg/schemalog"
-	"github.com/xataio/pgstream/pkg/stream"
-)
-
-func Test_PostgresToOpensearch(t *testing.T) {
-	if os.Getenv("PGSTREAM_INTEGRATION_TESTS") == "" {
-		t.Skip("skipping integration test...")
-	}
-
-	cfg := &stream.Config{
-		Listener:  testPostgresListenerCfg(),
-		Processor: testSearchProcessorCfg(),
-	}
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	// create a dedicated schema for the opensearch tests to ensure there's no
-	// interference between the other integration tests by having a separate index.
-	testSchema := "pg2os_integration_test"
-	execQuery(t, ctx, fmt.Sprintf("create schema %s", testSchema))
-
-	runStream(t, ctx, cfg)
-
-	client, err := es.NewClient(searchURL)
-	require.NoError(t, err)
-
-	var testTablePgstreamID string
-	testTable := "test"
-	testIndex := fmt.Sprintf("%s-1", testSchema)
-
-	tests := []struct {
-		name  string
-		query string
-
-		validation func() bool
-	}{
-		{
-			name:  "schema event",
-			query: fmt.Sprintf("create table %s.%s(id serial primary key, name text)", testSchema, testTable),
-
-			validation: func() bool {
-				resp := searchSchemaLog(t, ctx, client, testSchema)
-				if resp.Hits.Total.Value <= 0 {
-					return false
-				}
-				hit := resp.Hits.Hits[0].Source
-				testTablePgstreamID = getTablePgstreamID(t, hit, testTable)
-
-				require.Equal(t, false, hit["acked"])
-				require.Equal(t, testSchema, hit["schema_name"])
-				require.Equal(t, float64(2), hit["version"])
-
-				mapping := getIndexMapping(t, ctx, client, testIndex)
-				require.Equal(t, map[string]any{"type": "keyword"}, mapping["_table"])
-				require.Equal(t, map[string]any{"type": "long"}, mapping[fmt.Sprintf("%s-1", testTablePgstreamID)])
-				require.Equal(t, map[string]any{
-					"fields": map[string]any{
-						"text": map[string]any{"type": "text"},
-					},
-					"ignore_above": float64(32766),
-					"type":         "keyword",
-				}, mapping[fmt.Sprintf("%s-2", testTablePgstreamID)])
-				return true
-			},
-		},
-		{
-			name:  "data event",
-			query: fmt.Sprintf("insert into %s.%s(name) values('a')", testSchema, testTable),
-
-			validation: func() bool {
-				resp := searchTable(t, ctx, client, testIndex, testTablePgstreamID)
-				if resp.Hits.Total.Value != 1 {
-					return false
-				}
-				hit := resp.Hits.Hits[0]
-				t.Log(hit)
-				require.Equal(t, fmt.Sprintf("%s_1", testTablePgstreamID), hit.ID)
-				require.Equal(t, "a", hit.Source[fmt.Sprintf("%s-2", testTablePgstreamID)])
-				return true
-			},
-		},
-	}
-
-	for _, tc := range tests {
-		tc := tc
-		t.Run(tc.name, func(t *testing.T) {
-			execQuery(t, ctx, tc.query)
-
-			timer := time.NewTimer(20 * time.Second)
-			defer timer.Stop()
-			ticker := time.NewTicker(time.Second)
-			defer ticker.Stop()
-
-			for {
-				select {
-				case <-timer.C:
-					cancel()
-					t.Error("timeout waiting for opensearch data")
-					return
-				case <-ticker.C:
-					exists, err := client.IndexExists(ctx, testIndex)
-					require.NoError(t, err)
-					if exists && tc.validation() {
-						return
-					}
-				}
-			}
-		})
-	}
-}
-
-func searchSchemaLog(t *testing.T, ctx context.Context, client *es.Client, schemaName string) *es.SearchResponse {
-	query := es.QueryBody{
-		Query: &es.Query{
-			Bool: &es.BoolFilter{
-				Filter: []es.Condition{
-					{
-						Term: map[string]any{
-							"schema_name": schemaName,
-						},
-					},
-				},
-			},
-		},
-	}
-
-	return searchQuery(t, ctx, client, "pgstream", query, es.Ptr("version:desc"))
-}
-
-func searchTable(t *testing.T, ctx context.Context, client *es.Client, index, tableID string) *es.SearchResponse {
-	query := es.QueryBody{
-		Query: &es.Query{
-			Bool: &es.BoolFilter{
-				Filter: []es.Condition{
-					{
-						Term: map[string]any{
-							"_table": tableID,
-						},
-					},
-				},
-			},
-		},
-	}
-
-	return searchQuery(t, ctx, client, index, query, nil)
-}
-
-func searchQuery(t *testing.T, ctx context.Context, client *es.Client, index string, query es.QueryBody, sort *string) *es.SearchResponse {
-	queryBytes, err := json.Marshal(&query)
-	require.NoError(t, err)
-
-	resp, err := client.Search(ctx, &es.SearchRequest{
-		Index: es.Ptr(index),
-		Query: bytes.NewBuffer(queryBytes),
-		Sort:  sort,
-	})
-	require.NoError(t, err)
-
-	return resp
-}
-
-func getIndexMapping(t *testing.T, ctx context.Context, client *es.Client, index string) map[string]any {
-	mapping, err := client.GetIndexMappings(ctx, index)
-	require.NoError(t, err)
-
-	return mapping.Properties
-}
-
-func getTablePgstreamID(t *testing.T, source map[string]any, tableName string) string {
-	sourceBytes, err := json.Marshal(source)
-	require.NoError(t, err)
-
-	schemaLog := &schemalog.LogEntry{}
-	err = json.Unmarshal(sourceBytes, schemaLog)
-	require.NoError(t, err)
-
-	if len(schemaLog.Schema.Tables) != 1 {
-		return ""
-	}
-
-	for _, table := range schemaLog.Schema.Tables {
-		if table.Name == tableName {
-			return table.PgstreamID
-		}
-	}
-
-	return ""
-}
diff --git a/pkg/stream/integration/pg_search_integration_test.go b/pkg/stream/integration/pg_search_integration_test.go
new file mode 100644
index 0000000..3900fb5
--- /dev/null
+++ b/pkg/stream/integration/pg_search_integration_test.go
@@ -0,0 +1,228 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package integration
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/xataio/pgstream/internal/searchstore"
+	"github.com/xataio/pgstream/internal/searchstore/elasticsearch"
+	"github.com/xataio/pgstream/internal/searchstore/opensearch"
+	"github.com/xataio/pgstream/pkg/schemalog"
+	"github.com/xataio/pgstream/pkg/stream"
+	"github.com/xataio/pgstream/pkg/wal/processor/search/store"
+)
+
+func Test_PostgresToSearch(t *testing.T) {
+	if os.Getenv("PGSTREAM_INTEGRATION_TESTS") == "" {
+		t.Skip("skipping integration test...")
+	}
+
+	run := func(t *testing.T, cfg *stream.Config, client searchstore.Client, testSchema string) {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		// create a dedicated schema for the opensearch tests to ensure there's no
+		// interference between the other integration tests by having a separate index.
+		execQuery(t, ctx, fmt.Sprintf("create schema %s", testSchema))
+
+		runStream(t, ctx, cfg)
+
+		var testTablePgstreamID string
+		testTable := "test"
+		testIndex := fmt.Sprintf("%s-1", testSchema)
+
+		tests := []struct {
+			name  string
+			query string
+
+			validation func() bool
+		}{
+			{
+				name:  "schema event",
+				query: fmt.Sprintf("create table %s.%s(id serial primary key, name text)", testSchema, testTable),
+
+				validation: func() bool {
+					resp := searchSchemaLog(t, ctx, client, testSchema)
+					if resp.Hits.Total.Value <= 0 {
+						return false
+					}
+					hit := resp.Hits.Hits[0].Source
+					testTablePgstreamID = getTablePgstreamID(t, hit, testTable)
+
+					require.Equal(t, false, hit["acked"])
+					require.Equal(t, testSchema, hit["schema_name"])
+					require.Equal(t, float64(2), hit["version"])
+
+					mapping := getIndexMapping(t, ctx, client, testIndex)
+					require.Equal(t, map[string]any{"type": "keyword"}, mapping["_table"])
+					require.Equal(t, map[string]any{"type": "long"}, mapping[fmt.Sprintf("%s-1", testTablePgstreamID)])
+					require.Equal(t, map[string]any{
+						"fields": map[string]any{
+							"text": map[string]any{"type": "text"},
+						},
+						"ignore_above": float64(32766),
+						"type":         "keyword",
+					}, mapping[fmt.Sprintf("%s-2", testTablePgstreamID)])
+					return true
+				},
+			},
+			{
+				name:  "data event",
+				query: fmt.Sprintf("insert into %s.%s(name) values('a')", testSchema, testTable),
+
+				validation: func() bool {
+					resp := searchTable(t, ctx, client, testIndex, testTablePgstreamID)
+					if resp.Hits.Total.Value != 1 {
+						return false
+					}
+					hit := resp.Hits.Hits[0]
+					t.Log(hit)
+					require.Equal(t, fmt.Sprintf("%s_1", testTablePgstreamID), hit.ID)
+					require.Equal(t, "a", hit.Source[fmt.Sprintf("%s-2", testTablePgstreamID)])
+					return true
+				},
+			},
+		}
+
+		for _, tc := range tests {
+			tc := tc
+			t.Run(tc.name, func(t *testing.T) {
+				execQuery(t, ctx, tc.query)
+
+				timer := time.NewTimer(20 * time.Second)
+				defer timer.Stop()
+				ticker := time.NewTicker(time.Second)
+				defer ticker.Stop()
+
+				for {
+					select {
+					case <-timer.C:
+						cancel()
+						t.Error("timeout waiting for opensearch data")
+						return
+					case <-ticker.C:
+						exists, err := client.IndexExists(ctx, testIndex)
+						require.NoError(t, err)
+						if exists && tc.validation() {
+							return
+						}
+					}
+				}
+			})
+		}
+	}
+
+	t.Run("postgres to opensearch", func(t *testing.T) {
+		cfg := &stream.Config{
+			Listener: testPostgresListenerCfg(),
+			Processor: testSearchProcessorCfg(store.Config{
+				OpenSearchURL: opensearchURL,
+			}),
+		}
+
+		client, err := opensearch.NewClient(opensearchURL)
+		require.NoError(t, err)
+
+		run(t, cfg, client, "pg2os_integration_test")
+	})
+
+	t.Run("postgres to elasticsearch", func(t *testing.T) {
+		cfg := &stream.Config{
+			Listener: testPostgresListenerCfg(),
+			Processor: testSearchProcessorCfg(store.Config{
+				ElasticsearchURL: elasticsearchURL,
+			}),
+		}
+
+		client, err := elasticsearch.NewClient(elasticsearchURL)
+		require.NoError(t, err)
+
+		run(t, cfg, client, "pg2es_integration_test")
+	})
+}
+
+func searchSchemaLog(t *testing.T, ctx context.Context, client searchstore.Client, schemaName string) *searchstore.SearchResponse {
+	query := searchstore.QueryBody{
+		Query: &searchstore.Query{
+			Bool: &searchstore.BoolFilter{
+				Filter: []searchstore.Condition{
+					{
+						Term: map[string]any{
+							"schema_name": schemaName,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	return searchQuery(t, ctx, client, "pgstream", query, searchstore.Ptr("version:desc"))
+}
+
+func searchTable(t *testing.T, ctx context.Context, client searchstore.Client, index, tableID string) *searchstore.SearchResponse {
+	query := searchstore.QueryBody{
+		Query: &searchstore.Query{
+			Bool: &searchstore.BoolFilter{
+				Filter: []searchstore.Condition{
+					{
+						Term: map[string]any{
+							"_table": tableID,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	return searchQuery(t, ctx, client, index, query, nil)
+}
+
+func searchQuery(t *testing.T, ctx context.Context, client searchstore.Client, index string, query searchstore.QueryBody, sort *string) *searchstore.SearchResponse {
+	queryBytes, err := json.Marshal(&query)
+	require.NoError(t, err)
+
+	resp, err := client.Search(ctx, &searchstore.SearchRequest{
+		Index: searchstore.Ptr(index),
+		Query: bytes.NewBuffer(queryBytes),
+		Sort:  sort,
+	})
+	require.NoError(t, err)
+
+	return resp
+}
+
+func getIndexMapping(t *testing.T, ctx context.Context, client searchstore.Client, index string) map[string]any {
+	mapping, err := client.GetIndexMappings(ctx, index)
+	require.NoError(t, err)
+
+	return mapping.Properties
+}
+
+func getTablePgstreamID(t *testing.T, source map[string]any, tableName string) string {
+	sourceBytes, err := json.Marshal(source)
+	require.NoError(t, err)
+
+	schemaLog := &schemalog.LogEntry{}
+	err = json.Unmarshal(sourceBytes, schemaLog)
+	require.NoError(t, err)
+
+	if len(schemaLog.Schema.Tables) != 1 {
+		return ""
+	}
+
+	for _, table := range schemaLog.Schema.Tables {
+		if table.Name == tableName {
+			return table.PgstreamID
+		}
+	}
+
+	return ""
+}
diff --git a/pkg/stream/integration/setup_test.go b/pkg/stream/integration/setup_test.go
index 2193b33..9b3133b 100644
--- a/pkg/stream/integration/setup_test.go
+++ b/pkg/stream/integration/setup_test.go
@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/modules/elasticsearch"
 	"github.com/testcontainers/testcontainers-go/modules/kafka"
 	"github.com/testcontainers/testcontainers-go/modules/opensearch"
 	"github.com/testcontainers/testcontainers-go/modules/postgres"
@@ -44,6 +45,12 @@ func TestMain(m *testing.M) {
 			log.Fatal(err)
 		}
 		defer oscleanup()
+
+		escleanup, err := setupElasticsearchContainer(ctx)
+		if err != nil {
+			log.Fatal(err)
+		}
+		defer escleanup()
 	}
 
 	os.Exit(m.Run())
@@ -108,7 +115,7 @@ func setupOpenSearchContainer(ctx context.Context) (cleanup, error) {
 		return nil, fmt.Errorf("failed to start opensearch container: %w", err)
 	}
 
-	searchURL, err = ctr.Address(ctx)
+	opensearchURL, err = ctr.Address(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("retrieving url for opensearch container: %w", err)
 	}
@@ -117,3 +124,17 @@ func setupOpenSearchContainer(ctx context.Context) (cleanup, error) {
 		return ctr.Terminate(ctx)
 	}, nil
 }
+
+func setupElasticsearchContainer(ctx context.Context) (cleanup, error) {
+	ctr, err := elasticsearch.Run(ctx, "docker.elastic.co/elasticsearch/elasticsearch:8.9.0",
+		testcontainers.WithEnv(map[string]string{"xpack.security.enabled": "false"})) // disable TLS
+	if err != nil {
+		return nil, fmt.Errorf("failed to start elasticsearch container: %w", err)
+	}
+
+	elasticsearchURL = ctr.Settings.Address
+
+	return func() error {
+		return ctr.Terminate(ctx)
+	}, nil
+}
diff --git a/pkg/wal/processor/search/opensearch/helper_test.go b/pkg/wal/processor/search/store/helper_test.go
similarity index 71%
rename from pkg/wal/processor/search/opensearch/helper_test.go
rename to pkg/wal/processor/search/store/helper_test.go
index 9118367..7f52a32 100644
--- a/pkg/wal/processor/search/opensearch/helper_test.go
+++ b/pkg/wal/processor/search/store/helper_test.go
@@ -1,9 +1,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package opensearch
+package store
 
 import (
-	"github.com/xataio/pgstream/internal/es"
+	"github.com/xataio/pgstream/internal/searchstore"
 	"github.com/xataio/pgstream/pkg/schemalog"
 	"github.com/xataio/pgstream/pkg/wal/processor/search"
 )
@@ -12,8 +12,8 @@ type mockAdapter struct {
 	recordToLogEntryFn         func(map[string]any) (*schemalog.LogEntry, error)
 	schemaNameToIndexFn        func(schemaName string) IndexName
 	indexToSchemaNameFn        func(index string) string
-	searchDocToBulkItemFn      func(docs search.Document) es.BulkItem
-	bulkItemsToSearchDocErrsFn func(items []es.BulkItem) []search.DocumentError
+	searchDocToBulkItemFn      func(docs search.Document) searchstore.BulkItem
+	bulkItemsToSearchDocErrsFn func(items []searchstore.BulkItem) []search.DocumentError
 }
 
 func (m *mockAdapter) RecordToLogEntry(rec map[string]any) (*schemalog.LogEntry, error) {
@@ -28,10 +28,10 @@ func (m *mockAdapter) IndexToSchemaName(index string) string {
 	return m.indexToSchemaNameFn(index)
 }
 
-func (m *mockAdapter) SearchDocToBulkItem(docs search.Document) es.BulkItem {
+func (m *mockAdapter) SearchDocToBulkItem(docs search.Document) searchstore.BulkItem {
 	return m.searchDocToBulkItemFn(docs)
 }
 
-func (m *mockAdapter) BulkItemsToSearchDocErrs(items []es.BulkItem) []search.DocumentError {
+func (m *mockAdapter) BulkItemsToSearchDocErrs(items []searchstore.BulkItem) []search.DocumentError {
 	return m.bulkItemsToSearchDocErrsFn(items)
 }
diff --git a/pkg/wal/processor/search/opensearch/opensearch_pg_mapper_test.go b/pkg/wal/processor/search/store/search_pg_mapper_test.go
similarity index 95%
rename from pkg/wal/processor/search/opensearch/opensearch_pg_mapper_test.go
rename to pkg/wal/processor/search/store/search_pg_mapper_test.go
index 1a2e00a..1c13294 100644
--- a/pkg/wal/processor/search/opensearch/opensearch_pg_mapper_test.go
+++ b/pkg/wal/processor/search/store/search_pg_mapper_test.go
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package opensearch
+package store
 
 import (
 	"errors"
@@ -9,10 +9,13 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+	"github.com/xataio/pgstream/internal/searchstore/opensearch"
 	"github.com/xataio/pgstream/pkg/schemalog"
 	"github.com/xataio/pgstream/pkg/wal/processor/search"
 )
 
+const termByteLengthLimit = 32766
+
 func TestMapper_ColumnToSearchMapping(t *testing.T) {
 	tests := map[string]struct {
 		pg             string
@@ -174,7 +177,7 @@ func TestMapper_ColumnToSearchMapping(t *testing.T) {
 
 	for name, test := range tests {
 		t.Run(name, func(t *testing.T) {
-			m := NewPostgresMapper()
+			m := NewPostgresMapper(opensearch.NewMapper())
 			mapping, err := m.ColumnToSearchMapping(schemalog.Column{
 				DataType: test.pg,
 				Metadata: test.columnMetadata,
@@ -197,7 +200,7 @@ func TestMapper_ColumnToSearchMapping(t *testing.T) {
 
 	for name, test := range errorTests {
 		t.Run(name, func(t *testing.T) {
-			m := NewPostgresMapper()
+			m := NewPostgresMapper(opensearch.NewMapper())
 			_, err := m.ColumnToSearchMapping(schemalog.Column{DataType: test.pg})
 			require.Error(t, err)
 
@@ -279,7 +282,7 @@ func TestMapper_MapColumnValue(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
-			mapper := NewPostgresMapper()
+			mapper := NewPostgresMapper(opensearch.NewMapper())
 			value, err := mapper.MapColumnValue(tc.column, tc.value)
 			if !errors.Is(err, tc.wantErr) {
 				require.Error(t, err, tc.wantErr.Error())
diff --git a/pkg/wal/processor/search/opensearch/opensearch_store_test.go b/pkg/wal/processor/search/store/search_store_test.go
similarity index 65%
rename from pkg/wal/processor/search/opensearch/opensearch_store_test.go
rename to pkg/wal/processor/search/store/search_store_test.go
index 3a1d7b4..ad755cf 100644
--- a/pkg/wal/processor/search/opensearch/opensearch_store_test.go
+++ b/pkg/wal/processor/search/store/search_store_test.go
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
-package opensearch
+package store
 
 import (
 	"bytes"
@@ -12,8 +12,8 @@ import (
 
 	"github.com/rs/xid"
 	"github.com/stretchr/testify/require"
-	"github.com/xataio/pgstream/internal/es"
-	esmocks "github.com/xataio/pgstream/internal/es/mocks"
+	"github.com/xataio/pgstream/internal/searchstore"
+	searchstoremocks "github.com/xataio/pgstream/internal/searchstore/mocks"
 	"github.com/xataio/pgstream/pkg/schemalog"
 	"github.com/xataio/pgstream/pkg/wal/processor/search"
 	searchmocks "github.com/xataio/pgstream/pkg/wal/processor/search/mocks"
@@ -35,9 +35,9 @@ func TestStore_ApplySchemaChange(t *testing.T) {
 		Version:    1,
 	}
 
-	testSearchResponse := &es.SearchResponse{
-		Hits: es.Hits{
-			Hits: []es.Hit{
+	testSearchResponse := &searchstore.SearchResponse{
+		Hits: searchstore.Hits{
+			Hits: []searchstore.Hit{
 				{ID: "doc-1"},
 			},
 		},
@@ -62,26 +62,33 @@ func TestStore_ApplySchemaChange(t *testing.T) {
 
 	tests := []struct {
 		name     string
-		client   es.SearchClient
+		client   searchstore.Client
 		logEntry *schemalog.LogEntry
 
 		wantErr error
 	}{
 		{
-			name:     "ok - nil entry",
-			client:   &esmocks.Client{},
+			name: "ok - nil entry",
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+			},
 			logEntry: nil,
 
 			wantErr: nil,
 		},
 		{
 			name: "ok",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
 					return testSearchResponse, nil
 				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) { return true, nil },
-				IndexWithIDFn: func(ctx context.Context, req *es.IndexWithIDRequest) error {
+				IndexWithIDFn: func(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
 					return nil
 				},
 			},
@@ -91,14 +98,17 @@ func TestStore_ApplySchemaChange(t *testing.T) {
 		},
 		{
 			name: "ok - index doesn't exist",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
-					return nil, es.ErrResourceNotFound
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
+					return nil, searchstore.ErrResourceNotFound
 				},
 				IndexExistsFn:   func(ctx context.Context, index string) (bool, error) { return false, nil },
 				CreateIndexFn:   func(ctx context.Context, index string, body map[string]any) error { return nil },
 				PutIndexAliasFn: func(ctx context.Context, index []string, name string) error { return nil },
-				IndexWithIDFn: func(ctx context.Context, req *es.IndexWithIDRequest) error {
+				IndexWithIDFn: func(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
 					return nil
 				},
 			},
@@ -108,9 +118,12 @@ func TestStore_ApplySchemaChange(t *testing.T) {
 		},
 		{
 			name: "error - ensuring schema exists",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
-					return nil, es.ErrResourceNotFound
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
+					return nil, searchstore.ErrResourceNotFound
 				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) {
 					if index == schemalogIndexName {
@@ -126,8 +139,11 @@ func TestStore_ApplySchemaChange(t *testing.T) {
 		},
 		{
 			name: "error - getting last schema",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
 					return nil, errTest
 				},
 			},
@@ -137,12 +153,15 @@ func TestStore_ApplySchemaChange(t *testing.T) {
 		},
 		{
 			name: "error - schema out of order",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
 					return testSearchResponse, nil
 				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) { return true, nil },
-				IndexWithIDFn: func(ctx context.Context, req *es.IndexWithIDRequest) error {
+				IndexWithIDFn: func(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
 					return nil
 				},
 			},
@@ -157,12 +176,15 @@ func TestStore_ApplySchemaChange(t *testing.T) {
 		},
 		{
 			name: "error - updating mapping",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
 					return testSearchResponse, nil
 				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) { return true, nil },
-				IndexWithIDFn: func(ctx context.Context, req *es.IndexWithIDRequest) error { return errTest },
+				IndexWithIDFn: func(ctx context.Context, req *searchstore.IndexWithIDRequest) error { return errTest },
 			},
 			logEntry: newLogEntry,
 
@@ -170,8 +192,11 @@ func TestStore_ApplySchemaChange(t *testing.T) {
 		},
 		{
 			name: "error - ensuring schema mapping",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
 					return testSearchResponse, nil
 				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) { return false, nil },
@@ -212,15 +237,18 @@ func TestStore_SendDocuments(t *testing.T) {
 
 	tests := []struct {
 		name   string
-		client es.SearchClient
+		client searchstore.Client
 
 		wantErrDocs []search.DocumentError
 		wantErr     error
 	}{
 		{
 			name: "ok - no failed documents",
-			client: &esmocks.Client{
-				SendBulkRequestFn: func(ctx context.Context, items []es.BulkItem) ([]es.BulkItem, error) {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SendBulkRequestFn: func(ctx context.Context, items []searchstore.BulkItem) ([]searchstore.BulkItem, error) {
 					return nil, nil
 				},
 			},
@@ -230,11 +258,14 @@ func TestStore_SendDocuments(t *testing.T) {
 		},
 		{
 			name: "ok - with failed documents",
-			client: &esmocks.Client{
-				SendBulkRequestFn: func(ctx context.Context, items []es.BulkItem) ([]es.BulkItem, error) {
-					return []es.BulkItem{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SendBulkRequestFn: func(ctx context.Context, items []searchstore.BulkItem) ([]searchstore.BulkItem, error) {
+					return []searchstore.BulkItem{
 						{
-							Index: &es.BulkIndex{
+							Index: &searchstore.BulkIndex{
 								Index: testSchemaName,
 								ID:    "doc-1",
 							},
@@ -259,8 +290,11 @@ func TestStore_SendDocuments(t *testing.T) {
 		},
 		{
 			name: "error - sending bulk request",
-			client: &esmocks.Client{
-				SendBulkRequestFn: func(ctx context.Context, items []es.BulkItem) ([]es.BulkItem, error) {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SendBulkRequestFn: func(ctx context.Context, items []searchstore.BulkItem) ([]searchstore.BulkItem, error) {
 					return nil, errTest
 				},
 			},
@@ -293,13 +327,16 @@ func TestStore_DeleteSchema(t *testing.T) {
 
 	tests := []struct {
 		name   string
-		client es.SearchClient
+		client searchstore.Client
 
 		wantErr error
 	}{
 		{
 			name: "ok",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) {
 					require.Equal(t, testIndexWithVersion, index)
 					return true, nil
@@ -308,7 +345,7 @@ func TestStore_DeleteSchema(t *testing.T) {
 					require.Equal(t, []string{testIndexWithVersion}, index)
 					return nil
 				},
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					require.Equal(t, []string{schemalogIndexName}, req.Index)
 					require.Equal(t, map[string]any{
 						"query": map[string]any{
@@ -326,7 +363,10 @@ func TestStore_DeleteSchema(t *testing.T) {
 		},
 		{
 			name: "ok - index doesn't exist",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) {
 					require.Equal(t, testIndexWithVersion, index)
 					return false, nil
@@ -334,7 +374,7 @@ func TestStore_DeleteSchema(t *testing.T) {
 				DeleteIndexFn: func(ctx context.Context, index []string) error {
 					return errors.New("DeleteIndexFn: should not be called")
 				},
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					require.Equal(t, []string{schemalogIndexName}, req.Index)
 					require.Equal(t, map[string]any{
 						"query": map[string]any{
@@ -352,14 +392,17 @@ func TestStore_DeleteSchema(t *testing.T) {
 		},
 		{
 			name: "error - checking index exists",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) {
 					return false, errTest
 				},
 				DeleteIndexFn: func(ctx context.Context, index []string) error {
 					return errors.New("DeleteIndexFn: should not be called")
 				},
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					return errors.New("DeleteByQueryFn: should not be called")
 				},
 			},
@@ -368,14 +411,17 @@ func TestStore_DeleteSchema(t *testing.T) {
 		},
 		{
 			name: "error - deleting index",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) {
 					return true, nil
 				},
 				DeleteIndexFn: func(ctx context.Context, index []string) error {
 					return errTest
 				},
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					return errors.New("DeleteByQueryFn: should not be called")
 				},
 			},
@@ -384,14 +430,17 @@ func TestStore_DeleteSchema(t *testing.T) {
 		},
 		{
 			name: "error - deleting schema from schema log",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) {
 					return false, nil
 				},
 				DeleteIndexFn: func(ctx context.Context, index []string) error {
 					return errTest
 				},
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					return errTest
 				},
 			},
@@ -421,15 +470,18 @@ func TestStore_DeleteTableDocuments(t *testing.T) {
 
 	tests := []struct {
 		name     string
-		client   es.SearchClient
+		client   searchstore.Client
 		tableIDs []string
 
 		wantErr error
 	}{
 		{
 			name: "ok",
-			client: &esmocks.Client{
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					require.Equal(t, []string{testSchemaName}, req.Index)
 					require.Equal(t, map[string]any{
 						"query": map[string]any{
@@ -448,8 +500,11 @@ func TestStore_DeleteTableDocuments(t *testing.T) {
 		},
 		{
 			name: "ok - no tables",
-			client: &esmocks.Client{
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					require.Equal(t, []string{testSchemaName}, req.Index)
 					require.Equal(t, map[string]any{
 						"query": map[string]any{
@@ -468,8 +523,11 @@ func TestStore_DeleteTableDocuments(t *testing.T) {
 		},
 		{
 			name: "error - deleting by query",
-			client: &esmocks.Client{
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					return errTest
 				},
 			},
@@ -505,7 +563,7 @@ func TestStore_getLastSchemaLogEntry(t *testing.T) {
 
 	tests := []struct {
 		name      string
-		client    es.SearchClient
+		client    searchstore.Client
 		adapter   SearchAdapter
 		marshaler func(any) ([]byte, error)
 
@@ -514,17 +572,20 @@ func TestStore_getLastSchemaLogEntry(t *testing.T) {
 	}{
 		{
 			name: "ok",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
-					require.Equal(t, &es.SearchRequest{
-						Index: es.Ptr(schemalogIndexName),
-						Size:  es.Ptr(1),
-						Sort:  es.Ptr("version:desc"),
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
+					require.Equal(t, &searchstore.SearchRequest{
+						Index: searchstore.Ptr(schemalogIndexName),
+						Size:  searchstore.Ptr(1),
+						Sort:  searchstore.Ptr("version:desc"),
 						Query: bytes.NewBuffer(testBody),
 					}, req)
-					return &es.SearchResponse{
-						Hits: es.Hits{
-							Hits: []es.Hit{
+					return &searchstore.SearchResponse{
+						Hits: searchstore.Hits{
+							Hits: []searchstore.Hit{
 								{ID: "doc-1"},
 							},
 						},
@@ -543,8 +604,11 @@ func TestStore_getLastSchemaLogEntry(t *testing.T) {
 		},
 		{
 			name: "error - marshaling search query",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
 					return nil, errors.New("SearchFn: should not be called")
 				},
 			},
@@ -560,10 +624,13 @@ func TestStore_getLastSchemaLogEntry(t *testing.T) {
 		},
 		{
 			name: "error - no hits in response",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
-					return &es.SearchResponse{
-						Hits: es.Hits{},
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
+					return &searchstore.SearchResponse{
+						Hits: searchstore.Hits{},
 					}, nil
 				},
 			},
@@ -578,9 +645,12 @@ func TestStore_getLastSchemaLogEntry(t *testing.T) {
 		},
 		{
 			name: "error - schema not found",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
-					return nil, es.ErrResourceNotFound
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
+					return nil, searchstore.ErrResourceNotFound
 				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) {
 					require.Equal(t, schemalogIndexName, index)
@@ -598,8 +668,11 @@ func TestStore_getLastSchemaLogEntry(t *testing.T) {
 		},
 		{
 			name: "error - retrieving schema",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
 					return nil, errTest
 				},
 			},
@@ -614,9 +687,12 @@ func TestStore_getLastSchemaLogEntry(t *testing.T) {
 		},
 		{
 			name: "error - schema not found with pgstream index creation",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
-					return nil, es.ErrResourceNotFound
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
+					return nil, searchstore.ErrResourceNotFound
 				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) {
 					require.Equal(t, schemalogIndexName, index)
@@ -638,9 +714,12 @@ func TestStore_getLastSchemaLogEntry(t *testing.T) {
 		},
 		{
 			name: "error - schema not found, failed to create schemalog index",
-			client: &esmocks.Client{
-				SearchFn: func(ctx context.Context, req *es.SearchRequest) (*es.SearchResponse, error) {
-					return nil, es.ErrResourceNotFound
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
+				SearchFn: func(ctx context.Context, req *searchstore.SearchRequest) (*searchstore.SearchResponse, error) {
+					return nil, searchstore.ErrResourceNotFound
 				},
 				IndexExistsFn: func(ctx context.Context, index string) (bool, error) {
 					require.Equal(t, schemalogIndexName, index)
@@ -689,13 +768,16 @@ func TestStore_createSchema(t *testing.T) {
 
 	tests := []struct {
 		name   string
-		client es.SearchClient
+		client searchstore.Client
 
 		wantErr error
 	}{
 		{
 			name: "ok",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				CreateIndexFn: func(ctx context.Context, index string, body map[string]any) error {
 					return nil
 				},
@@ -710,7 +792,10 @@ func TestStore_createSchema(t *testing.T) {
 		},
 		{
 			name: "error - creating index",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				CreateIndexFn: func(ctx context.Context, index string, body map[string]any) error {
 					return errTest
 				},
@@ -723,7 +808,10 @@ func TestStore_createSchema(t *testing.T) {
 		},
 		{
 			name: "error - putting index alias",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				CreateIndexFn: func(ctx context.Context, index string, body map[string]any) error {
 					return nil
 				},
@@ -767,7 +855,7 @@ func TestStore_updateMapping(t *testing.T) {
 
 	tests := []struct {
 		name   string
-		client es.SearchClient
+		client searchstore.Client
 		diff   *schemalog.SchemaDiff
 		mapper search.Mapper
 
@@ -775,14 +863,17 @@ func TestStore_updateMapping(t *testing.T) {
 	}{
 		{
 			name: "ok - no diff",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				PutIndexMappingsFn: func(ctx context.Context, index string, body map[string]any) error {
 					return errors.New("PutIndexMappingsFn: should not be called")
 				},
-				IndexWithIDFn: func(ctx context.Context, req *es.IndexWithIDRequest) error {
+				IndexWithIDFn: func(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
 					return nil
 				},
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					return errors.New("DeleteByQueryFn: should not be called")
 				},
 			},
@@ -791,7 +882,10 @@ func TestStore_updateMapping(t *testing.T) {
 		},
 		{
 			name: "ok - diff with columns to add",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				PutIndexMappingsFn: func(ctx context.Context, index string, body map[string]any) error {
 					require.Equal(t, testIndexName, index)
 					require.Equal(t, map[string]any{
@@ -801,10 +895,10 @@ func TestStore_updateMapping(t *testing.T) {
 					}, body)
 					return nil
 				},
-				IndexWithIDFn: func(ctx context.Context, req *es.IndexWithIDRequest) error {
+				IndexWithIDFn: func(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
 					return nil
 				},
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					return errors.New("DeleteByQueryFn: should not be called")
 				},
 			},
@@ -823,14 +917,17 @@ func TestStore_updateMapping(t *testing.T) {
 		},
 		{
 			name: "ok - diff with tables to remove",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				PutIndexMappingsFn: func(ctx context.Context, index string, body map[string]any) error {
 					return errors.New("PutIndexMappingsFn: should not be called")
 				},
-				IndexWithIDFn: func(ctx context.Context, req *es.IndexWithIDRequest) error {
+				IndexWithIDFn: func(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
 					return nil
 				},
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					require.Equal(t, []string{testIndexName}, req.Index)
 					require.Equal(t, map[string]any{
 						"query": map[string]any{
@@ -854,14 +951,17 @@ func TestStore_updateMapping(t *testing.T) {
 		},
 		{
 			name: "error - updating mapping",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				PutIndexMappingsFn: func(ctx context.Context, index string, body map[string]any) error {
 					return errTest
 				},
-				IndexWithIDFn: func(ctx context.Context, req *es.IndexWithIDRequest) error {
+				IndexWithIDFn: func(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
 					return errors.New("IndexWithIDFn: should not be called")
 				},
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					return errors.New("DeleteByQueryFn: should not be called")
 				},
 			},
@@ -875,14 +975,17 @@ func TestStore_updateMapping(t *testing.T) {
 		},
 		{
 			name: "error - deleting tables",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				PutIndexMappingsFn: func(ctx context.Context, index string, body map[string]any) error {
 					return errors.New("PutIndexMappingsFn: should not be called")
 				},
-				IndexWithIDFn: func(ctx context.Context, req *es.IndexWithIDRequest) error {
+				IndexWithIDFn: func(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
 					return errors.New("IndexWithIDFn: should not be called")
 				},
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					return errTest
 				},
 			},
@@ -897,14 +1000,17 @@ func TestStore_updateMapping(t *testing.T) {
 		},
 		{
 			name: "error - inserting schemalog",
-			client: &esmocks.Client{
+			client: &searchstoremocks.Client{
+				GetMapperFn: func() searchstore.Mapper {
+					return &searchstoremocks.Mapper{}
+				},
 				PutIndexMappingsFn: func(ctx context.Context, index string, body map[string]any) error {
 					return errors.New("PutIndexMappingsFn: should not be called")
 				},
-				IndexWithIDFn: func(ctx context.Context, req *es.IndexWithIDRequest) error {
+				IndexWithIDFn: func(ctx context.Context, req *searchstore.IndexWithIDRequest) error {
 					return errTest
 				},
-				DeleteByQueryFn: func(ctx context.Context, req *es.DeleteByQueryRequest) error {
+				DeleteByQueryFn: func(ctx context.Context, req *searchstore.DeleteByQueryRequest) error {
 					return errors.New("DeleteByQueryFn: should not be called")
 				},
 			},