JOINs in Elasticsearch? One SQL Statement. Zero Application Code.

SELECT o.amount, c.name, c.city
FROM orders o
JOIN customers c ON o.customer_id = c.id

- A Kafka consumer listens for order events
- For each order, it looks up the customer in another index
- It writes an enriched document to a third index
- Another consumer listens for customer updates
- When a customer changes, it re-enriches all their orders

6 months later:
- 2,000 lines of application code
- 3+ Kafka topics (inputs, outputs, dead-letter queues - often 5-6 in production)
- A race condition that causes stale customer names 1% of the time
- A backlog of 500K documents during peak hours
- An on-call rotation just for the denormalization pipeline

CREATE MATERIALIZED VIEW orders_with_customers_mv
REFRESH EVERY 8 SECONDS
WITH (delay = '1s', user_latency = '1s')
AS
SELECT
  o.id,
  o.amount,
  o.status,
  c.name AS customer_name,
  c.email AS customer_email,
  c.city AS customer_city,
  c.country AS customer_country,
  c.department.zip_code AS customer_zip,
  UPPER(c.name) AS customer_name_upper
FROM orders AS o
JOIN customers AS c ON o.customer_id = c.id
WHERE o.status = 'completed';

Step 1:  Alter source schemas        → add changelog tracking fields
Step 2:  Create intermediate indices → changelog indices, enriched indices, view index
Step 3:  Preload changelogs          → copy existing data into changelog indices
Step 4:  Create enrich policies      → define lookup enrichment from customers
Step 5:  Create watcher              → schedule enrich policy re-execution
Step 6:  Execute enrich policies     → build initial enrich indices
Step 7:  Create ingest pipelines     → enrichment + computed field processors
Step 8:  Create transforms           → changelog, enrichment, computed fields
Step 9:  Start transforms            → sequentially, with checkpoint waits
Step 10: Save metadata               → persist MV definition for management

SELECT * FROM orders_with_customers_mv
WHERE customer_name = 'Alice'
ORDER BY amount DESC
LIMIT 10;

CREATE MATERIALIZED VIEW revenue_by_city_mv
AS
SELECT
  c.city,
  c.country,
  COUNT(*) AS order_count,
  SUM(o.amount) AS total_revenue,
  AVG(o.amount) AS avg_order_value,
  MAX(o.amount) AS max_order
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE o.status = 'completed'
GROUP BY c.city, c.country
HAVING SUM(o.amount) > 10000
ORDER BY total_revenue DESC;

CREATE MATERIALIZED VIEW active_orders_mv
REFRESH EVERY 30 SECONDS
AS
SELECT id, amount, status, created_at
FROM orders
WHERE status = 'active';

CREATE TABLE IF NOT EXISTS orders (
  id INT NOT NULL,
  customer_id INT NOT NULL,
  amount DOUBLE,
  status KEYWORD DEFAULT 'pending',
  createdAt TIMESTAMP,
  PRIMARY KEY (id)
);

CREATE TABLE IF NOT EXISTS customers (
  id INT NOT NULL,
  name VARCHAR,
  email KEYWORD,
  city KEYWORD,
  country KEYWORD,
  department STRUCT FIELDS (
    name VARCHAR,
    zip_code KEYWORD
  ),
  PRIMARY KEY (id)
);

INSERT INTO customers (id, name, email, city, country, department) VALUES
  (1, 'Alice', 'alice@example.com', 'Paris', 'France', {name = 'Engineering', zip_code = '75001'}),
  (2, 'Bob', 'bob@example.com', 'Lyon', 'France', {name = 'Marketing', zip_code = '69001'}),
  (3, 'Chloe', 'chloe@example.com', 'Marseille', 'France', {name = 'Sales', zip_code = '13001'});
INSERT INTO orders (id, customer_id, amount, status) VALUES
  (101, 1, 250.00, 'completed'),
  (102, 2, 180.50, 'completed'),
  (103, 1, 420.00, 'completed'),
  (104, 3, 95.00, 'pending');

CREATE OR REPLACE MATERIALIZED VIEW orders_with_customers_mv
REFRESH EVERY 8 SECONDS
WITH (delay = '1s', user_latency = '1s')
AS
SELECT
  o.id,
  o.amount,
  c.name AS customer_name,
  c.email,
  c.city AS customer_city,
  c.country AS customer_country,
  c.department.zip_code AS customer_zip,
  UPPER(c.name) AS customer_name_upper
FROM orders AS o
JOIN customers AS c ON o.customer_id = c.id
WHERE o.status = 'completed';

-- View the schema
DESCRIBE MATERIALIZED VIEW orders_with_customers_mv;

| Field                | Type    | Null | Key | Default | Comment | Script | Extra |
|----------------------|---------|------|-----|---------|---------|--------|-------|
| id                   | INT     | yes  |     | NULL    |         |        | ()    |
| amount               | DOUBLE  | yes  |     | NULL    |         |        | ()    |
| customer_name        | VARCHAR | yes  |     | NULL    |         |        | ()    |
| email                | KEYWORD | yes  |     | NULL    |         |        | ()    |
| customer_city        | KEYWORD | yes  |     | NULL    |         |        | ()    |
| customer_country     | KEYWORD | yes  |     | NULL    |         |        | ()    |
| customer_zip         | KEYWORD | yes  |     | NULL    |         |        | ()    |
| customer_name_upper  | KEYWORD | yes  |     | NULL    |         |        | ()    |

-- View the original SQL
SHOW CREATE MATERIALIZED VIEW orders_with_customers_mv;

-- Check transform status
SHOW MATERIALIZED VIEW STATUS orders_with_customers_mv;

-- List all materialized views
SHOW MATERIALIZED VIEWS;

SELECT * FROM orders_with_customers_mv
WHERE customer_name = 'Alice'
ORDER BY amount DESC;

| id  | amount | customer_name | email             | customer_city | customer_country | customer_zip | customer_name_upper |
|-----|--------|---------------|-------------------|---------------|------------------|--------------|---------------------|
| 103 | 420.00 | Alice         | alice@example.com | Paris         | France           | 75001        | ALICE               |
| 101 | 250.00 | Alice         | alice@example.com | Paris         | France           | 75001        | ALICE               |

REFRESH MATERIALIZED VIEW orders_with_customers_mv WITH SCHEDULE NOW;

DROP MATERIALIZED VIEW IF EXISTS orders_with_customers_mv;

Step 1:  Alter source schemas         ✅
Step 2:  Create intermediate indices  ✅
Step 3:  Preload changelogs           ✅
Step 4:  Create enrich policies       ✅
Step 5:  Create watcher               ✅
Step 6:  Execute enrich policies      ✅
Step 7:  Create ingest pipelines      ❌ FAILED

→ Automatic rollback initiated
→ Steps 6, 5, 4, 3, 2, 1 reversed
→ Original state restored
→ Error message returned with details

-- Fast refresh for real-time dashboards
REFRESH EVERY 8 SECONDS
WITH (delay = '1s', user_latency = '1s')

-- Moderate refresh for operational views
REFRESH EVERY 1 MINUTE

-- Slow refresh for analytical views
REFRESH EVERY 1 HOUR

CREATE MATERIALIZED VIEW fast_view_mv
REFRESH EVERY 5 SECONDS
WITH (delay = '2s', user_latency = '500ms')
AS
SELECT ...

-- Cron job, Kubernetes CronJob, or Airflow task
EXECUTE ENRICH POLICY orders_with_customers_mv_customers_enrich_policy;

-- Or trigger a full refresh
REFRESH MATERIALIZED VIEW orders_with_customers_mv;

-- Create
CREATE [OR REPLACE] MATERIALIZED VIEW [IF NOT EXISTS] name
  [REFRESH EVERY n time_unit]
  [WITH (delay = 'interval', user_latency = 'interval')]
  AS SELECT ...
  [FROM t1 JOIN t2 ON t1.col = t2.col]
  [WHERE ...]
  [GROUP BY ... HAVING ...]
  [ORDER BY ... LIMIT n]

-- Drop (cleans up all artifacts)
DROP MATERIALIZED VIEW [IF EXISTS] name;

-- Force refresh
REFRESH MATERIALIZED VIEW [IF EXISTS] name [WITH SCHEDULE NOW];

-- Inspect
DESCRIBE MATERIALIZED VIEW name;
SHOW MATERIALIZED VIEW name;
SHOW CREATE MATERIALIZED VIEW name;
SHOW MATERIALIZED VIEW STATUS name;
SHOW MATERIALIZED VIEWS;