This has been my experience too. Instead of going the logical replication route I tend to leverage the transactional outbox to achieve consistency in the application layer instead.
So when I transact data into tables I immediately fetch the latest outbox id.
And then when query from Elasticsearch I first fetch what the last outbox id of the processed data is.
This way I know if the transaction was already processed into Elasticsearch or not. Repeat. Until outbox id of Elasticsearch is equal or higher than the outbox id of the mutation.
This way I don't have to use logical replication, no k/v store and I can just use a script that fetches and processes the latest outbox changes on a loop.
Looked in the source of Materialize and it looks like this is exactly what they are doing.
They are using Debezium + Kafka for receiving the WAL changes. And using send the processed WAL offsets back using a Kafka topic to Debezium + Postges.
This way they can achieve consistency
What you are describing is having the data in elasticsearch in the same format as the data in postgresql. Which is easy.
You could also for instance create a script and use a postgresql logical replication connection ( just like Debezium ) stream the changes into elasticsearch. Without having a full Kafka connect setup. And all the training an maintenance that comes with it.
What I am describing is, before storing the data in Elasticsearch computing the data in a materialized format. So it's more efficient for Elasticsearch to work with. And no longer having any needs for joins.
So instead of recomputing the materialized data every time in it's entirety you want to be more smart about it.
The root table of the materialized data depends on maybe 5 other tables. So if data in those 5 other tables change you need to know if they have a relationship with a row in the "root table". And then only re-materialize those rows.
Materialize does this by having it's own SQL language where you define your materialized view. Which compiles to clever algorithms and uses an execution engine to get to this result.
What I am doing is just having a lookup graph + queries to see what tables, id entries are invalidated. And I re-materialize using normal sql and some extra processing using a nodejs script to make it more optimal for Elasticsearch.
The poor man's version of Materialize that I implemented is the following:
Step 1) Find all "paths" between tables
* Use the postgres information schema to get all relations
* Use npm library graph-cycles to see if there are any graph cycles. If so.. some relations go on a blacklist.
* Use npm library topopsort to sort the graph
* Traverse the graph and find all possible paths from and to tables
* Generate SQL queries to look up "affected" other tables. Input: table name + id Output: list of table names + ids
Step 2) Track changes in database using a transactional outbox
* Create outbox table fields: id, timestamp, event, table_name, table id, jsonb payload
* After running migrations have a script that ensures every table has triggers on insert,update,delete that would insert a row on each of these events in the outbox
Step 3) Compute ( You want to do reporting for a certain table ( target table ) which has relationships with other tables and for this table you want a "materialized" view )
* Have a script on a loop that takes all the outbox entries since last time processed
* Use the queries from step1 to find out which ids of "target table" are affected
* Then only materialize / compute those ids
* Store in Elasticsearch ( in our case )
This is not a solution if you are Facebook scale. But as a small SAAS company with not too many transactions this works brilliantly.
And with more tenants you can just scale this up by sharding the inbox.
Bonus points: have elasticsearch (ingest node) and a postgres replica on the same host as the "compute" script. So if you have a lot of queries calculating the "dependencies" you get better performance.
So when I transact data into tables I immediately fetch the latest outbox id.
And then when query from Elasticsearch I first fetch what the last outbox id of the processed data is.
This way I know if the transaction was already processed into Elasticsearch or not. Repeat. Until outbox id of Elasticsearch is equal or higher than the outbox id of the mutation.
This way I don't have to use logical replication, no k/v store and I can just use a script that fetches and processes the latest outbox changes on a loop.