> ## Documentation Index
> Fetch the complete documentation index at: https://docs.risingwave.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Use SQLMesh to streamline pipelines

> Introduces how SQLMesh simplifies managing RisingWave streaming SQL pipelines.

SQLMesh provides a structured and maintainable approach to managing end-to-end streaming pipelines. As the control plane for RisingWave, it simplifies the workflow by enabling local definition of transformations, planning and reviewing changes before deployment, and running optional audits to ensure data quality.

This guide explains how to use RisingWave for streaming data and SQLMesh for managing transformations to aggregate website click events. Each event contains a `timestamp`, `event_type`, and `value`.

It aims to calculate a rolling 5-minute count and sum of `value` for each `event_type` using a tumbling window. Then, the logic will be modified to calculate an average, showcasing how SQLMesh handles the change.

## Set up RisingWave and ingest sample data

Begin by starting RisingWave and manually inserting sample data into a base table.

1. Start RisingWave via Docker.

   ```bash theme={null}
   docker run -it --pull=always -p 4566:4566 -p 5691:5691 risingwavelabs/risingwave:latest single_node
   ```

   The message of `RisingWave standalone mode is ready` means that RisingWave has started successfully.

2. Connect via `psql`  (or another Postgres-compatible SQL client) and prepare the data.

   ```bash theme={null}
   # Use localhost when connecting from your machine to the mapped port
   psql -h localhost -p 4566 -d dev -U root
   ```

   Inside `psql`, run the following:

   ```sql theme={null}
   -- Create the source table to hold incoming data
   CREATE TABLE click_events (
       event_id INT PRIMARY KEY,
       event_type VARCHAR,
       event_value INT,
       event_timestamp TIMESTAMPTZ
   );

   -- Insert seven rows of sample data
   INSERT INTO click_events (event_id, event_type, event_value, event_timestamp) VALUES
   (1, 'page_view', 10, '2023-10-27 10:00:30+00'),
   (2, 'add_to_cart', 150, '2023-10-27 10:01:15+00'),
   (3, 'page_view', 12, '2023-10-27 10:02:05+00'),
   (4, 'page_view', 11, '2023-10-27 10:05:40+00'),
   (5, 'purchase', 250, '2023-10-27 10:06:20+00'),
   (6, 'page_view', 15, '2023-10-27 10:11:00+00'),
   (7, 'add_to_cart', 99, '2023-10-27 10:12:30+00');

   -- Optional: Verify inserts
   SELECT COUNT(*) FROM click_events;

   -- Exit psql
   \q
   ```

## Set up SQLMesh project

Set up the SQLMesh environment to manage our RisingWave transformations.

1. Create a project directory and Python virtual environment.

   ```bash theme={null}
   mkdir sqlmesh-risingwave-demo
   cd sqlmesh-risingwave-demo
   python3 -m venv venv
   source venv/bin/activate # Adjust activation for your OS/shell
   ```

2. Install the SQLMesh RisingWave adapter:

   ```bash theme={null}
   pip install "sqlmesh[risingwave]"
   ```

3. Initialize the SQLMesh Project specifically for RisingWave:

   ```bash theme={null}
   sqlmesh init risingwave
   ```

   This creates necessary folders and configuration files.

4. Configure the connection in `config.yaml` file (created by `init`). Ensure the following matches or update your `config.yaml`.

   ```yaml theme={null}
   gateways:
   risingwave: # Gateway name used by default
       connection:
       type: risingwave
       host: localhost
       user: root
       port: 4566 # Match the Docker port mapping
       database: dev

   default_gateway: risingwave

   model_defaults:
   dialect: risingwave
   # start: <YYYY-MM-DD> # Optional: Default start date for backfills if needed
   ```

## Define and deploy a streaming query (v1)

Create the first transformation model and deploy it.

1. Create the SQLMesh model file `models/event_summary_tumbling.sql`:

   ```sql theme={null}
   -- models/event_summary_tumbling.sql
   MODEL (
     name reporting.event_summary_tumbling,
     kind VIEW (materialized = true), -- This makes it a RisingWave MV
     owner data_team,
     description 'Summarizes event counts and total value in 5-minute tumbling windows.'
   );

   SELECT
       window_start,
       window_end,
       event_type,
       COUNT(*) AS event_count,
       SUM(event_value) AS total_value
   FROM TUMBLE(
       click_events,         -- Read from the base table
       event_timestamp,            -- Time column for windowing
       INTERVAL '5 minutes'        -- Tumbling window size
   )
   GROUP BY
       window_start,
       window_end,
       event_type;
   ```

2. Plan and apply the model using SQLMesh:

   ```sql theme={null}
   sqlmesh plan
   ```

   SQLMesh will detect the new model (`reporting.event_summary_tumbling`) and show a plan to create it in the prod environment (which maps to a schema like `sqlmesh_prod` or `prod` in RisingWave, often including the model name itself, e.g., `sqlmesh__reporting`). It will also detect the default models created by init; you can choose to apply changes only for your new model if desired, or apply all.

   ```bash theme={null}
   # Example Output Snippet:
   # Summary of differences:
   # Models:
   # └── Added:
   #     └── reporting.event_summary_tumbling
   #     └── sqlmesh_example.full_model (if applying init models)
   #     └── ... (other init models)
   #
   # Apply - Apply the plan. [y/n]: y  <-- Type 'y' and press Enter
   ```

   SQLMesh executes the `CREATE MATERIALIZED VIEW` statement(s) in RisingWave.

3. Verify the MV creation and content via `psql` .

   The exact schema and MV name includes a hash for versioning. You can use `SHOW` to find it. Since `reporting` is used as model schema, the materialized view should be in the `sqlmesh__reporting` schema.

   ```sql theme={null}
   show materialized views from sqlmesh__reporting;
   -- Result below
                       Name                      
   -----------------------------------------------
   reporting__event_summary_tumbling__3036268106
   (1 row)
   ```

   You should see the aggregated results similar to below.

   <img src="https://mintcdn.com/risingwavelabs/vysvNtKAV5NsTnkZ/images/sqlmesh01.png?fit=max&auto=format&n=vysvNtKAV5NsTnkZ&q=85&s=124da9d6b64a1922283396ff0d4914f7" alt="SQLMesh aggregated results" width="1402" height="312" data-path="images/sqlmesh01.png" />

## Modify the MV schema with SQLMesh (v2)

Now we can change the aggregation logic and see how SQLMesh manages the update safely.

1. Modify the model file `models/event_summary_tumbling.sql` to calculate the average value and distinct count:

   ```sql theme={null}
   -- models/event_summary_tumbling.sql
   MODEL (
   name reporting.event_summary_tumbling, -- Keep the same name
   kind VIEW (materialized true),
   owner data_team,
   -- Update description to reflect change
   description 'Calculates average event value and distinct event count in 5-minute tumbling windows.'
   );

   SELECT
       window_start,
       window_end,
       event_type,
       COUNT(DISTINCT event_id) AS distinct_event_count, -- Changed aggregation
       AVG(event_value) AS average_value          -- Changed aggregation
   FROM TUMBLE(
       click_events,
       event_timestamp,
       INTERVAL '5 minutes'
   )
   GROUP BY
       window_start,
       window_end,
       event_type;
   ```

2. Run `sqlmesh plan` again to see the impact.

   SQLMesh detects the change in the *definition* of `reporting.event_summary_tumbling`. Because the logic changed, it plans to:

   * Create a **new version** of the Materialized View in RisingWave with a *different* physical name (a new hash suffix).
   * It does **not** immediately drop the old version, allowing for validation or zero-downtime promotion strategies (though in this simple apply, the old one might eventually be cleaned up depending on settings).

   <img src="https://mintcdn.com/risingwavelabs/vysvNtKAV5NsTnkZ/images/sqlmesh02.png?fit=max&auto=format&n=vysvNtKAV5NsTnkZ&q=85&s=c7a3ca8cd70f2b6436f8e2c603a999be" alt="Modified results" width="1632" height="1260" data-path="images/sqlmesh02.png" />

3. Verify the new MV version via `psql`:

   ```sql theme={null}
   -- List MVs again in the schema
   SHOW MATERIALIZED VIEWS FROM sqlmesh__reporting; -- Replace schema if needed
   -- You should now see TWO MVs related to event_summary_tumbling,
   -- one with the old hash and one with a NEW hash.

   -- Query the NEWER MV version (find its exact name)
   SELECT * FROM sqlmesh__reporting."reporting__event_summary_tumbling__<new_hash>" ORDER BY window_start, event_type;
   ```

   The results should show the `distinct_event_count` and `average_value columns`, reflecting the v2 logic. The original v1 MV still exists momentarily. This demonstrates SQLMesh's safe, versioned deployment approach.

## Integrate sink and source management

This guide manually created the `click_events` table in RisingWave for simplicity. However, SQLMesh provides pre- and post-statements within model definitions, allowing you to embed DDL statements that run before or after the main model logic executes.

For example, you could place a `CREATE SOURCE IF NOT EXISTS ...` statement in pre-statement (that is, a SQL query prior to the `SELECT` statement), and / or a `CREATE SINK IF NOT EXISTS ...` in a post-statement. This approach keeps the setup/teardown logic closer to the relevant transformation step within your SQLMesh project.

SQLMesh's core focus is managing the transformation logic (the `SELECT` statements) and dependencies across your entire pipeline (often multiple models). While pre- and post-statements offer integration points, managing complex `CREATE SOURCE/SINK` statements with intricate connector configurations might still be cleaner handled outside the model definitions (e.g., separate setup scripts or specialized infrastructure tools) to maintain clarity.

## Additional resources

* [SQLMesh Documentation](https://sqlmesh.readthedocs.io/en/stable/)

* [Learn more about SQLMesh](https://www.tobikodata.com/sqlmesh)