diff --git a/changelog/product-lifecycle.mdx b/changelog/product-lifecycle.mdx index 089d23d6..d9c73fa5 100644 --- a/changelog/product-lifecycle.mdx +++ b/changelog/product-lifecycle.mdx @@ -22,6 +22,7 @@ Below is a list of all features in the public preview phase: | Feature name | Start version | | :-- | :-- | +| [Ingest data from MySQL table](/integrations/sources/mysql-table)| 2.2 | | [EXPLAIN FORMAT JSON](/sql/commands/sql-explain#explain-options)| 2.2 | | [Ingest data from Postgres table](/integrations/sources/postgresql-table) | 2.1 | | [Ingest data from webhook](/integrations/sources/webhook) | 2.1 | diff --git a/get-started/quickstart.mdx b/get-started/quickstart.mdx index f4bab36e..e415b5f6 100644 --- a/get-started/quickstart.mdx +++ b/get-started/quickstart.mdx @@ -137,7 +137,7 @@ Unlike other deployment modes, for instance [Docker Compose](/deploy/risingwave- For state store, we will use the embedded `LocalFs` Object Store, eliminating the need for an external service like `minio` or `s3`; for meta store, we will use the embedded `SQLite` database, eliminating the need for an external service like `etcd`. -By default, the RisingWave standalone mode will store its data in `~/risingwave`, which includes both `Metadata` and `State Data`. +By default, the RisingWave standalone mode will store its data in `~/.risingwave`, which includes both `Metadata` and `State Data`. For a batteries-included setup, with `monitoring` tools and external services like `kafka` fully included, you can use [Docker Compose](/deploy/risingwave-docker-compose) instead. If you would like to set up these external services manually, you may check out RisingWave's [Docker Compose](https://github.com/risingwavelabs/risingwave/blob/main/docker/docker-compose.yml), and run these services using the same configurations. @@ -147,7 +147,7 @@ The instance of RisingWave standalone mode can run without any configuration. Ho The main options which new users may require would be the state store directory (`--state-store-directory`) and in-memory mode (`--in-memory`). -`--state-store-directory` specifies the new directory where the cluster's `Metadata` and `State Data` will reside. The default is to store it in the `~/risingwave` folder. +`--state-store-directory` specifies the new directory where the cluster's `Metadata` and `State Data` will reside. The default is to store it in the `~/.risingwave` folder. ```bash # Reconfigure RisingWave to be stored under 'projects' folder instead. diff --git a/get-started/rw-premium-edition-intro.mdx b/get-started/rw-premium-edition-intro.mdx index cb12b2e8..2ff06d6c 100644 --- a/get-started/rw-premium-edition-intro.mdx +++ b/get-started/rw-premium-edition-intro.mdx @@ -16,18 +16,12 @@ For RisingWave Cloud users, all Premium Edition features are available out of th ## Premium features -The premium features are carefully selected based on the following criteria: - -- Seamless integration with proprietary or licensed open-source systems. -- Advanced features that enhance development velocity and lower production deployment overhead. -- Performance improvements for non-standard deployment environments. -- Tailored features specifically requested by our paying customers. - The following are Premium Edition features, which include a "Premium Edition Feature" note in the documentation. ### SQL and security - +* [Time travel queries](/processing/time-travel-queries) +* [Secret management](/operate/manage-secrets) ### Schema management @@ -37,10 +31,25 @@ The following are Premium Edition features, which include a "Premium Edition Fea ### Connectors - +* [Sink to Snowflake](/integrations/destinations/snowflake) +* [Sink to DynamoDB](/integrations/destinations/amazon-dynamodb) +* [Sink to OpenSearch](/integrations/destinations/opensearch) +* [Sink to BigQuery](/integrations/destinations/bigquery) +* [Sink to SharedMergeTree table engine on ClickHouse Cloud](/integrations/destinations/clickhouse#supported-table-engines) +* [Sink to SQL Server](/integrations/destinations/sql-server) +* [Direct SQL Server CDC source connector](/integrations/sources/sql-server-cdc) +* [Sink to Iceberg with glue catalog](/integrations/destinations/apache-iceberg#glue-catelogs) +* [Ingest data from webhook](/integrations/sources/webhook) For users who are already using these features in 1.9.x or earlier versions, rest assured that the functionality of these features will be intact if you stay on the version. If you choose to upgrade to v2.0 or later versions, an error will show up to indicate you need a license to use the features. +The premium features are carefully selected based on the following criteria: + +- Seamless integration with proprietary or licensed open-source systems. +- Advanced features that enhance development velocity and lower production deployment overhead. +- Performance improvements for non-standard deployment environments. +- Tailored features specifically requested by our paying customers. + ## How to access Premium Edition features For RisingWave Cloud users, all Premium Edition features are available out of the box without additional cost. @@ -104,4 +113,4 @@ RisingWave provides three levels of support packages: ## Pricing -Pricing for RisingWave Premium will be based on the cluster size, measured in RisingWave Units (RWUs). The number of RWUs will be determined based on the scale of data ingestion, number of streaming jobs, and the complexity of use case. There could be additional factors as well. Please contact our sales at [sales@risingwave-labs.com](mailto:sales@risingwave-labs.com) for more details. +For pricing details, please contact our sales at [sales@risingwave-labs.com](mailto:sales@risingwave-labs.com). \ No newline at end of file diff --git a/ingestion/overview.mdx b/ingestion/overview.mdx index 64f95d1b..e100df6a 100644 --- a/ingestion/overview.mdx +++ b/ingestion/overview.mdx @@ -74,11 +74,11 @@ The statement will create a streaming job that continuously ingests data from th 4. **Stronger consistency guarantee**: When using a table with connectors, all downstream jobs will be guaranteed to have a consistent view of the data persisted in the table; while for source, different jobs may see inconsistent results due to different ingestion speed or data retention in the external system. 5. **Greater flexibility**: Like regular tables, you can use DML statements like [INSERT](/sql/commands/sql-insert), [UPDATE](/sql/commands/sql-update) and [DELETE](/sql/commands/sql-delete) to insert or modify data in tables with connectors, and use [CREATE SINK INTO TABLE](/sql/commands/sql-create-sink-into) to merge other data streams into the table. -### PostgreSQL table +### Table-valued function -RisingWave supports using the table-valued function `postgres_query` to directly query PostgreSQL databases. This function connects to a specified PostgreSQL instance, executes the provided SQL query, and returns the results as a table in RisingWave. +RisingWave supports using the table-valued function (TVF) `postgres_query` or `mysql_query` to directly query PostgreSQL or MySQL databases. This function connects to a specified instance, executes the provided SQL query, and returns the results as a table in RisingWave. -To use it, specify connection details (such as hostname, port, username, password, database name) and the desired SQL query. This makes it easier to integrate PostgreSQL data directly into RisingWave workflows without needing additional data transfer steps. For more information, see [Ingest data from Postgres tables](/integrations/sources/postgresql-table). +To use it, specify connection details (such as hostname, port, username, password, database name) and the desired SQL query. This makes it easier to integrate databases directly into RisingWave workflows without needing additional data transfer steps. For more information, see [Ingest data from Postgres tables](/integrations/sources/postgresql-table) and [Ingest data from MySQL tables](/integrations/sources/mysql-table). ## DML on tables diff --git a/integrations/sources/mysql-table.mdx b/integrations/sources/mysql-table.mdx new file mode 100644 index 00000000..5d4b0728 --- /dev/null +++ b/integrations/sources/mysql-table.mdx @@ -0,0 +1,102 @@ +--- +title: "Ingest data from MySQL table" +description: "Describes how to ingest data from MySQL table to RisingWave using table-valued function." +sidebarTitle: MySQL table +--- + +RisingWave allows you to query MySQL tables directly with the `mysql_query` table-valued function (TVF). It offers a simpler alternative to Change Data Capture (CDC) when working with MySQL data in RisingWave. + +Unlike CDC, which continuously syncs data changes, this function lets you fetch data directly from MySQL when needed. Therefore, this approach is ideal for static or infrequently updated data, as it's more resource-efficient than maintaining a constant CDC connection. + + +**PUBLIC PREVIEW** + +This feature is currently in public preview, meaning it is nearing the final product but may not yet be fully stable. If you encounter any issues or have feedback, please reach out to us via our [Slack channel](https://www.risingwave.com/slack). Your input is valuable in helping us improve this feature. For more details, see our [Public Preview Feature List](/changelog/product-lifecycle#features-in-the-public-preview-stage). + + + +Added in version 2.2. + + +## Syntax + +Define `mysql_query` as follows: + +```sql +mysql_query( + hostname varchar, -- Database hostname + port varchar, -- Database port + username varchar, -- Authentication username + password varchar, -- Authentication password + database_name varchar, -- Target database name + query varchar -- SQL query to execute +) +``` + +## Data type mapping + +The following table shows how MySQL data types are mapped to RisingWave data types: + +| MySQL Type | RisingWave Type | +|:-----------|:----------------| +| `bit(1)` | `boolean` | +| `bit(>1)` | `bytea` | +| `bool`/`boolean` | `smallint` | +| `tinyint` | `smallint` | +| `smallint` | `smallint` | +| `mediumint` | `int` | +| `int` | `int` | +| `bigint` | `bigint` | +| `float` | `float32` | +| `double` | `float64` | +| `decimal` | `decimal` | +| `numeric` | `decimal` | +| `year` | `int` | +| `date` | `date` | +| `time` | `time` | +| `datetime` | `timestamp` | +| `timestamp` | `timestamptz` | +| `varchar` | `varchar` | +| `char` | `varchar` | +| `json` | `jsonb` | +| `blob` | `bytea` | +| `tinyblob` | `bytea` | +| `mediumblob` | `bytea` | +| `longblob` | `bytea` | +| `array` | *unsupported* | +| `enum` | *unsupported* | +| `set` | *unsupported* | +| `geometry` | *unsupported* | +| `null` | *unsupported* | + +## Example + +1. In your MySQL database, create a table and populate it with sample data of various data types. + +```sql +CREATE TABLE test ( + id bigint primary key, v0 bit, v1 bool, v2 tinyint(1), + v3 tinyint(2), v4 smallint, v5 mediumint, v6 integer, + v7 bigint, v8 float, v9 double, v10 numeric(4, 2), + v11 decimal(4, 2), v12 char(255), v13 varchar(255), + v14 bit(10), v15 tinyblob, v16 blob, v17 mediumblob, + v18 longblob, v19 date, v20 time, v21 timestamp, + v22 json +); + +INSERT INTO test SELECT + 1 as id, true as v0, true as v1, 2 as v2, 3 as v3, 4 as v4, 5 as v5, + 6 as v6, 7 as v7, 1.08 as v8, 1.09 as v9, 1.10 as v10, 1.11 as v11, + 'char' as v12, 'varchar' as v13, b'1010' as v14, x'16' as v15, x'17' as v16, + x'18' as v17, x'19' as v18, '2021-01-01' as v19, '12:34:56' as v20, + '2021-01-01 12:34:56' as v21, JSON_OBJECT('key1', 1, 'key2', 'abc'); +``` + +2. In RisingWave, use `postgres_query` function to perform the query. + +```sql +SELECT * +FROM mysql_query('$MYSQL_HOST', '$MYSQL_TCP_PORT', '$RISEDEV_MYSQL_USER', '$MYSQL_PWD', 'tvf', 'select * from test;'); +----RESULT +1 t 1 2 3 4 5 6 7 1.08 1.09 1.10 1.11 char varchar \x000a \x16 \x17 \x18 \x19 2021-01-01 12:34:56 2021-01-01 12:34:56+00:00 {"key1": 1, "key2": "abc"} +``` diff --git a/integrations/sources/postgresql-table.mdx b/integrations/sources/postgresql-table.mdx index 43c08767..ff83fd01 100644 --- a/integrations/sources/postgresql-table.mdx +++ b/integrations/sources/postgresql-table.mdx @@ -6,7 +6,7 @@ sidebarTitle: PostgreSQL table RisingWave allows you to query PostgreSQL tables directly with the `postgres_query` table-valued function (TVF). It offers a simpler alternative to Change Data Capture (CDC) when working with PostgreSQL data in RisingWave. -Unlike CDC, which continuously syncs data changes, this function lets you fetch data directly from PostgreSQL when needed. Therefore, this approach is ideal static or infrequently updated data, as it's more resource-efficient than maintaining a constant CDC connection. +Unlike CDC, which continuously syncs data changes, this function lets you fetch data directly from PostgreSQL when needed. Therefore, this approach is ideal for static or infrequently updated data, as it's more resource-efficient than maintaining a constant CDC connection. **PUBLIC PREVIEW** @@ -14,6 +14,10 @@ Unlike CDC, which continuously syncs data changes, this function lets you fetch This feature is currently in public preview, meaning it is nearing the final product but may not yet be fully stable. If you encounter any issues or have feedback, please reach out to us via our [Slack channel](https://www.risingwave.com/slack). Your input is valuable in helping us improve this feature. For more details, see our [Public Preview Feature List](/changelog/product-lifecycle#features-in-the-public-preview-stage). + +Added in version 2.1. + + ## Syntax Define `postgres_query` as follows: diff --git a/mint.json b/mint.json index ebc8917f..e4291611 100644 --- a/mint.json +++ b/mint.json @@ -453,7 +453,6 @@ "serve/query-from-visualization-tools", "serve/risingwave-as-postgres-fdw", "serve/subscription" - ] }, { diff --git a/query/overview.mdx b/query/overview.mdx new file mode 100644 index 00000000..8cd42393 --- /dev/null +++ b/query/overview.mdx @@ -0,0 +1,77 @@ +--- +title: "Query data in RisingWave" +description: "RisingWave allows you to access and use insights from your streaming data immediately. It also functions like any other database, allowing you to query batch or raw data that you've inserted." +sidebarTitle: Overview +--- + +This section explains how to query and interact with data in RisingWave. + +RisingWave offers several methods for serving results, catering to various use cases, from ad-hoc analysis to application integration. + +## Query with `SELECT` statements + +Retrieve data directly from RisingWave using standard SQL SELECT queries against tables or materialized views. Use this method for ad-hoc analysis, exploring the latest results, and extracting specific data subsets. + +For syntax details, see [`SELECT`](/sql/commands/sql-select). To learn how RisingWave processes data (ad-hoc or streaming), see [Ad-hoc vs. Streaming queries](/processing/overview#ad-hoc-on-read-vs-streaming-on-write). + +Connect to RisingWave with psql or any other PostgreSQL-compatible client to execute these queries. RisingWave is compatible with [many data visualization tools](/integrations/visualization/overview). Here are a few that we have tested: + +- [Beekeeper Studio](/integrations/visualization/beekeeper-studio) +- [DBeaver](/integrations/visualization/dbeaver) +- [Grafana](/integrations/visualization/grafana) +- [Looker](/integrations/visualization/looker) +- [Metabase](/integrations/visualization/metabase) +- [Superset](/integrations/visualization/superset) + +**Key features**: + +- Uses familiar SQL syntax. +- Provides immediate access to the most up-to-date results. +- Offers flexibility to filter, aggregate, and join data. + +**Example**: Retrieve the latest aggregated results from a materialized view. + +## Integrate with PostgreSQL via foreign data wrapper (FDW) + +RisingWave seamlessly integrates with existing PostgreSQL ecosystems through its Foreign Data Wrapper (FDW) functionality. Query data in RisingWave's tables and materialized views as if it were part of your PostgreSQL database. This allows you to leverage existing PostgreSQL tools and workflows. + +For details, see [RisingWave as Postgres FDW](/query/risingwave-as-postgres-fdw). + +**Key features**: + +- Enables unified querying across RisingWave and PostgreSQL data. +- Allows you to use existing PostgreSQL tools and applications. +- Simplifies integration into existing data infrastructure. +- Performance: While FDW offers convenience, it may introduce some performance overhead compared to directly querying RisingWave. RisingWave pushes down filters in `WHERE` clauses to optimize queries. However, complex queries with joins, aggregations, or `LIMIT` clauses are processed in PostgreSQL after fetching the data from RisingWave. + +**Example**: Join data in a PostgreSQL table with a continuously updated materialized view in RisingWave. + +## Subscribe to real-time updates + +RisingWave's subscription feature allows you to receive a continuous stream of updates from a materialized view directly, without needing an external message queue. This includes both existing data in the materialized view when the subscription is created and subsequent changes. You can choose to retrieve the full dataset or only incremental changes from a specific point using a subscription cursor. + +For details, see [Subscription](/query/subscription). + +**Key features**: + +- Provides real-time data updates directly from RisingWave. +- Allows retrieving full or incremental data using a cursor. +- Requires fewer components and less maintenance than external event stores. + +**Example**: Subscribe to a materialized view that tracks website user activity to power a live dashboard, receiving updates directly from RisingWave. + +## Access programmatically via SDK and client libraries + +RisingWave provides a [Python SDK](/python-sdk/intro) [`risingwave-py`](https://pypi.org/project/risingwave-py/) (currently in public preview) to help you develop event-driven applications. The SDK offers a simple way to perform ad-hoc queries, subscribe to changes, and define event handlers for tables and materialized views. + +Additionally, since RisingWave is compatible with Postgres, you can use standard PostgreSQL drivers to interact with RisingWave from your applications. + +Client libraries in various languages allow developers to interact with RisingWave programmatically and execute `SELECT` queries within their applications. + +For the list of available client libraries, see [Client Libraries](/client-libraries/overview). + +**Example**: Use the Python client library to fetch the latest results from a materialized view and display them in a financial data analysis application. + +## Choose the right method + +From the methods described above, select the one that best fits your needs, considering factors like query complexity, integration requirements, and team expertise. RisingWave ensures consistency across all methods. \ No newline at end of file diff --git a/query/query-from-visualization-tools.mdx b/query/query-from-visualization-tools.mdx new file mode 100644 index 00000000..291d2bc1 --- /dev/null +++ b/query/query-from-visualization-tools.mdx @@ -0,0 +1,4 @@ +--- +title: "Query from visualization tools" +url: "/integrations/visualization/overview" +--- diff --git a/query/query-with-select.mdx b/query/query-with-select.mdx new file mode 100644 index 00000000..bf0c4519 --- /dev/null +++ b/query/query-with-select.mdx @@ -0,0 +1,4 @@ +--- +title: "Query with SELECT statements" +url: "/sql/commands/sql-select" +--- diff --git a/query/risingwave-as-postgres-fdw.mdx b/query/risingwave-as-postgres-fdw.mdx new file mode 100644 index 00000000..fa6a044a --- /dev/null +++ b/query/risingwave-as-postgres-fdw.mdx @@ -0,0 +1,162 @@ +--- +title: "RisingWave as a PostgreSQL foreign data wrapper" +sidebarTitle: RisingWave as Postgres FDW +description: "A foreign data wrapper in PostgreSQL allows you to directly virtualize data stored in an external database as a local external table, also known as a foreign table. This tutorial will demonstrate how to interact between PostgreSQL and RisingWave. In this example, RisingWave will use CDC (Change Data Capture) to extract data from PostgreSQL, and analyze it using a materialized view. Then PostgreSQL will directly retrieve the computation results stored in RisingWave." +--- + + +**PUBLIC PREVIEW** + +This feature is currently in public preview, meaning it is nearing the final product but may not yet be fully stable. If you encounter any issues or have feedback, please reach out to us via our [Slack channel](https://www.risingwave.com/slack). Your input is valuable in helping us improve this feature. For more details, see our [Public Preview Feature List](/changelog/product-lifecycle#features-in-the-public-preview-stage). + + +## Prerequisites + +The following demo needs to be completed under two `psql` connections. One connection links to PostgreSQL, providing the source data for analysis and obtaining the analysis results from RisingWave. In our demo, the command to connect to PostgreSQL is `psql -h localhost -p 5432 -d myd -U postgresuser`, with the password being `postgrespw`. The other connection links to RisingWave to establish a job for analyzing the data. In our demo, the command to connect to RisingWave is `psql -h localhost -p 4566 -d dev -U root`, with no password required. You can choose to exit from `psql` and log in to the other database when you need to operate in another database. Alternatively, you can use `tmux` to open two terminals simultaneously, connecting to the respective databases with `psql`. + +* The PostgreSQL used supports the `postgres_fdw` extension. +* Both PostgreSQL and RisingWave are accessible from each other. +* Both of the users in PostgreSQL (`postgresuser` in this demo) and in RisingWave (`root` in this demo) have the necessary permissions to create tables and materialized views. + +## Prepare data in PostgreSQL + +The following commands create a table in PostgreSQL and insert data into it. + +```sql +---Run in PostgreSQL +CREATE TABLE person ( + "id" int, + "name" varchar(64), + "credit_card" varchar(200), + "city" varchar(200), + PRIMARY KEY ("id") +); + +INSERT INTO person VALUES (1001, 'peter white', '1781 2313 8157 6974', 'boise'); +INSERT INTO person VALUES (1002, 'sarah spencer', '3453 4987 9481 6270', 'los angeles'); +INSERT INTO person VALUES (1004, 'julie white', '0052 8113 1582 4430', 'seattle'); +INSERT INTO person VALUES (1005, 'sarah smith', '4591 5419 7260 8350', 'los angeles'); +INSERT INTO person VALUES (1007, 'walter spencer', '5136 7504 2879 7886', 'los angeles'); +INSERT INTO person VALUES (1008, 'john abrams', '6064 8548 6057 2021', 'redmond'); +INSERT INTO person VALUES (1010, 'kate smith', '9474 6887 6463 6972', 'bend'); +INSERT INTO person VALUES (1011, 'vicky noris', '9959 4034 5717 6729', 'boise'); +INSERT INTO person VALUES (1012, 'walter jones', '8793 6517 3085 0542', 'boise'); +INSERT INTO person VALUES (1013, 'sarah walton', '2280 4209 8743 0735', 'kent'); +INSERT INTO person VALUES (1015, 'vicky jones', '3148 5012 3225 2870', 'los angeles'); +INSERT INTO person VALUES (1016, 'john walton', '0426 2682 6145 8371', 'seattle'); +INSERT INTO person VALUES (1017, 'luke jones', '9641 9352 0248 2749', 'redmond'); +``` + +## Analyze data in RisingWave + +The following command creates a table in RisingWave. This table will use the native CDC connector to synchronize the data of the Person table from PostgreSQL, and then create a materialized view to analyze the ingested data. + +```sql +---Run in RisingWave +---Create a table in RisingWave to replicate the Person table of PostgreSQL into RisingWave +CREATE TABLE pg_person ( + "id" int, + "name" varchar, + "credit_card" varchar, + "city" varchar, + PRIMARY KEY ("id") +) with ( + connector = 'postgres-cdc', + hostname = 'localhost', + port = '5432', + username = 'postgresuser', + password = 'postgrespw', + database.name = 'mydb', + schema.name = 'public', + table.name = 'person', + slot.name = 'person' +); + +---Create a materialized view to analyze the population of each city +CREATE MATERIALIZED VIEW city_population AS +SELECT + city, + COUNT(*) as population +FROM + pg_person +GROUP BY + city; +``` + +## Query result in PostgreSQL using FDW + +The following command creates a foreign table in PostgreSQL to connect to RisingWave and query the materialized view. The first four commands prepare the remote access of `postgres_fdw`. You can check the PostgreSQL's doc [here](https://www.postgresql.org/docs/current/postgres-fdw.html) for more details. + +```sql +---Run in PostgreSQL +---Enable the postgres_fdw extension +CREATE EXTENSION postgres_fdw; + +---Create a foreign table to connect to RisingWave +CREATE SERVER risingwave + FOREIGN DATA WRAPPER postgres_fdw + OPTIONS (host 'localhost', port '4566', dbname 'dev'); + +---Create a user mapping for the foreign server, mapping the RisingWave's user `root` to the PostgreSQL's user `postgresuser` +CREATE USER MAPPING FOR postgresuser + SERVER risingwave + OPTIONS (user 'root', password ''); + +---Import the definition of table and materialized view from RisingWave. +IMPORT FOREIGN SCHEMA public + FROM SERVER risingwave INTO public; + +---List the foreign table and materialized view in PostgreSQL. +SELECT * FROM pg_foreign_table; +---------+----------+------------------------------------------------- + ftrelid | ftserver | ftoptions +---------+----------+------------------------------------------------- + 16413 | 16411 | {schema_name=public,table_name=city_population} + 16416 | 16411 | {schema_name=public,table_name=pg_person} + +---Check whether the data is synchronized from PostgreSQL to RisingWave. +SELECT * FROM pg_person; +------|----------------+---------------------+------------- + id | name | credit_card | city +------+----------------+---------------------+------------- + 1005 | sarah smith | 4591 5419 7260 8350 | los angeles + 1012 | walter jones | 8793 6517 3085 0542 | boise + 1002 | sarah spencer | 3453 4987 9481 6270 | los angeles + 1007 | walter spencer | 5136 7504 2879 7886 | los angeles + 1011 | vicky noris | 9959 4034 5717 6729 | boise + 1016 | john walton | 0426 2682 6145 8371 | seattle + 1010 | kate smith | 9474 6887 6463 6972 | bend + 1015 | vicky jones | 3148 5012 3225 2870 | los angeles + 1017 | luke jones | 9641 9352 0248 2749 | redmond + 1001 | peter white | 1781 2313 8157 6974 | boise + 1004 | julie white | 0052 8113 1582 4430 | seattle + 1008 | john abrams | 6064 8548 6057 2021 | redmond + 1013 | sarah walton | 2280 4209 8743 0735 | kent + +---Query the materialized view in RisingWave with PostgreSQL's foreign table. +SELECT * FROM city_population; +-------------+------------ + city | population +-------------+------------ + boise | 3 + los angeles | 4 + bend | 1 + kent | 1 + redmond | 2 + seattle | 2 +``` + + +Currently, write operations to RisingWave through a foreign data wrapper are not supported. The data in the foreign table is read-only. + + +## Differences between sinking to Postgres and using FDW in Postgres + +There are two main methods to interact between RisingWave and PostgreSQL: sinking data to PostgreSQL and utilizing a foreign data wrapper of PostgreSQL to access data in RisingWave. The table below provides a summary of the differences between these two methods. Your choice between these methods will depend on your specific requirements, data architecture, and performance considerations. + +| Aspect | Sinking to PostgreSQL | Using PostgreSQL FDW to access data | +| :------------------------- | :------------------------------------------------------ | :----------------------------------------------------------------- | +| Data Access | Data is physically stored in PostgreSQL | Data is physically stored in RisingWave | +| Performance | Potential latency for RisingWave to write to PostgreSQL | Potential latency when reading data from RisingWave | +| Message Delivery Guarantee | At-least-once while sinking into PostgreSQL tables | Exactly-once for MVs and the data is not moved | +| Extra Requirement | None | Requires the postgres\_fdw extension and involves more setup steps | diff --git a/query/subscription.mdx b/query/subscription.mdx new file mode 100644 index 00000000..1097428c --- /dev/null +++ b/query/subscription.mdx @@ -0,0 +1,328 @@ +--- +title: "Subscription" +description: Subscription is used to pull data change records for a specific table or materialized view (MV). +--- + +The data from a subscription includes both the existing data in the table at the time of subscription creation and the incremental change records in the table after the subscription is created. You can use the method of creating a subscription cursor to retrieve the full data set or the incremental data set after a specified starting point. + +This feature allows you to monitor all data changes without relying on external event stores like Kafka. Compared to the Kafka sink or other event store sinks, a subscription requires fewer components and thus, less maintenance. + + +**PUBLIC PREVIEW** + +This feature is currently in public preview, meaning it is nearing the final product but may not yet be fully stable. If you encounter any issues or have feedback, please reach out to us via our [Slack channel](https://www.risingwave.com/slack). Your input is valuable in helping us improve this feature. For more details, see our [Public Preview Feature List](/changelog/product-lifecycle#features-in-the-public-preview-stage). + + +## Manage subscription + +Use the syntax below to create, drop or alter subscription. + +### Create subscription + +To create a subscription, use the syntax below: + +```sql +CREATE SUBSCRIPTION FROM WITH ( +retention = '' +); +``` + +The `FROM` clause must specify either a table or a materialized view (mv). + +The `retention` parameter should be provided as a string in the format of an interval. It represents the duration for which incremental data will be retained. Any incremental data that exceeds the specified retention duration will be automatically deleted and will no longer be accessible. + +### Drop subscription + +To drop a subscription, use the syntax below: + +```sql +DROP SUBSCRIPTION ; +``` + +### Alter subscription + +To rename a subscription, change the owner, or set a new schema, use the syntax below: + +```sql +ALTER SUBSCRIPTION + [ RENAME TO ] + [ OWNER TO ] + [ SET SCHEMA ] + ; +``` + +## Subscription cursor + +A subscription cursor is a unit used to consume data from a subscription. In RisingWave, it’s a tool specifically designed to work in conjunction with a subscription, differing from the general cursor. + +In RisingWave, the subscription cursor allows you to specify a specific starting point within the data of the subscription. Once the subscription cursor is created, you can use a loop to fetch and consume the data starting from that point onwards. A subscription can have multiple subscription cursors, which can be used to consume different ranges or intervals of data from the subscription. + +### Syntax[](#syntax "Direct link to Syntax") + +The syntax of creating a subscription cursor is as follows: + +```sql +DECLARE cursor_name SUBSCRIPTION CURSOR FOR subscription_name [since_clause | FULL]; +``` + +The `since_clause` is used to specify the starting point for reading data. By setting this clause, you can control the range of data that is returned, allowing you to retrieve only the incremental data or data starting from a specific time or event. + +Below are the available choices for `since_clause`. If you don’t specify the `since_clause`, the returned data will just include the incremental data after declaration, which equals to the first choice below. + +1. `since now()/proctime()` : The returned data will include only the incremental data starting from the time of declaration. +2. `since begin()` : The returned data will include the oldest incremental data available, typically starting from the beginning of the subscription's retention period. +3. `since unix_ms` : Starts reading from the first time point greater than or equal to the specified `unix_ms` value. It's important to note that the `unix_ms` value should fall within the range of `now() - subscription's retention` and `now`. + +If you specify `FULL` instead of the `since_clause`, the subscription cursor starts consuming data from stock. + +### Fetch from cursor + + +FETCH from cursor function is supported in the PSQL simple query mode and extended mode. + + +#### Non-blocking data fetch + +```sql +FETCH NEXT/n FROM cursor_name; +``` + +Fetch the next row or up to N rows from the cursor. If fewer than N rows are available, it will return whatever is available immediately without waiting. This also means that if there are no rows available (i.e., the latest data has been reached), an empty result will be returned immediately. + +```sql +FETCH NEXT FROM cur; + +----RESULT +t1.v1 | t1.v2 | t1.v3 | t1.op | rw_timestamp +-------+-------+-------+--------------+--------------- + 1 | 1 | 1 | UpdateDelete | 1715669376304 +(1 row) +``` + +In the example above, the `op` column in the result indicates the type of change operations. There are four options: `Insert`, `UpdateInsert`, `Delete`, and `UpdateDelete`. For a single UPDATE statement, the subscription log will contain two separate rows: one with `UpdateInsert` and another with `UpdateDelete`. This is because RisingWave treats an UPDATE as a delete of the old value followed by an insert of the new value. As for `rw_timestamp`, it corresponds to the Unix timestamp in milliseconds when the data was written. + +#### Blocking data fetch + +```sql +FETCH NEXT/n FROM cursor_name WITH (timeout = '1s'); +``` + +Fetch up to N rows from the cursor with a specified timeout. The `timeout` value should be a string in the interval format. In this case, the fetch statement will return when either N rows have been fetched or the timeout occurs. If the timeout occurs, whatever has been read up to that point will be returned. Here are two scenarios to trigger the timeout: + +1. The cursor has reached the latest data and has been waiting too long for new data to arrive. + +2. At least N rows are available for the cursor to read, but retrieving all of them takes an extended period. + +To avoid polling for new data frequently with the non-blocking `FETCH`, you can set a longer timeout to simulate a scenario where you want the `FETCH` to block until new data arrives. + +#### Order of the fetched data + +* For data with different `rw_timestamp`, values are returned in the order the events occurred. +* For data with the same `rw_timestamp`, the order matches the event sequence if the data belongs to the same primary key in the subscribed materialized view or table. +* For data with the same `rw_timestamp` but different primary keys, the order may not reflect the exact event sequence. + +### Show subscription cursors + +To show all subscription cursors in the current session, use the syntax below: + +```sql +SHOW SUBSCRIPTION CURSORS; + +------RESULT + Name | SubscriptionName +------+------------------ + cur2 | sub + cur | sub +(2 rows) +``` + +### Examples + +Let’s create a table `t1` and subscribe this table, then create a cursor for this subscription. + +```sql +-- Create a table and insert some data. +create table t1(v1 int, v2 int, v3 int); +insert into t1 values(1,1,1); + +-- Create a subscription. +create subscription sub from t1 with (retention = '1D'); + +-- Create a subscription cursor. +declare cur subscription cursor for sub; +``` + +After creation, we can use the `FETCH NEXT FROM cursor_name` statement to fetch data from this cursor: + +```sql +fetch next from cur; + +----RESULT + v1 | v2 | v3 | op | rw_timestamp +----+----+----+--------+-------------- + 1 | 1 | 1 | Insert | +(1 row) +``` + +Then we can update table `t1` and fetch again to view the changes: + +```sql +update t1 set v3 = 10 where v1 = 1; +fetch next from cur; + +----RESULT + t1.v1 | t1.v2 | t1.v3 | t1.op | rw_timestamp +-------+-------+-------+---------------+--------------- + 1 | 1 | 1 | UpdateDelete | 1715669376304 +(1 row) + +fetch next from cur; +----RESULT + t1.v1 | t1.v2 | t1.v3 | t1.op | rw_timestamp +-------+-------+-------+---------------+--------------- + 1 | 1 | 10 | UpdateInsert | 1715669376304 +(1 row) +``` + +We can also create another subscription cursor to specify `since_clause` . Let’s use `since unix_ms` to rebuild the cursor: + +```sql +declare cur2 subscription cursor for sub since 1715669376304; +fetch next from cur2; + +----RESULT + t1.v1 | t1.v2 | t1.v3 | t1.op | rw_timestamp +-------+-------+-------+---------------+--------------- + 1 | 1 | 1 | UpdateDelete | 1715669376304 +(1 row) + +fetch next from cur2; +----RESULT + t1.v1 | t1.v2 | t1.v3 | t1.op | rw_timestamp +-------+-------+-------+---------------+--------------- + 1 | 1 | 10 | UpdateInsert | 1715669376304 +(1 row) +``` + +## Subscribing via Postgres driver + +For this feature, you only need to use [the Postgres driver](https://docs.risingwave.com/docs/dev/client-libraries-overview/), and no extra dependencies are required. + +Here’s an example using Python and [psycopg2](https://pypi.org/project/psycopg2/). + +```py +import psycopg2 +import time + +def main(): + # Connect to the PostgreSQL database + conn = psycopg2.connect( + host="localhost", + port="4566", + user="root", + database="dev" + ) + + try: + # Create a cursor object + cur = conn.cursor() + + # Declare a cursor for the subscription + cur.execute("DECLARE cur SUBSCRIPTION CURSOR FOR sub_users;") + + while True: + # Fetch the next row from the cursor + cur.execute("FETCH NEXT FROM cur;") + row = cur.fetchone() + + if row is None: + # Sleep for 1 second if no row is fetched + time.sleep(1) + continue + + # Replace with your event handling logic + print("Row fetched:", row) + + finally: + # Close the cursor and connection + print("Terminated") + cur.close() + conn.close() + +if __name__ == "__main__": + main() +``` + +Example output: + +```sql +Row fetched: (1, 'Alice', 30, 1, 1716434906890) +Row fetched: (2, 'Bob', 25, 1, 1716434909889) +Row fetched: (3, 'Charlie', 35, 1, 1716434912889) +Row fetched: (4, 'Diana', 28, 1, 1716434920889) +``` + +## Exactly-once delivery + +The persistent nature of subscriptions allows the subscriber to resume from a specific point in time (`rw_timestamp`) without data loss after a failure recovery. We also guarantee no duplicates in subscriptions, thus ensuring exactly-once delivery. + +### Persisting the consumption progress + +To achieve exactly-once delivery, it’s required to periodically persist the timestamp in storage. We recommend using RisingWave as the store, as no extra component is needed. + +First, we need to create a table for storing the progress. + +```sql +CREATE TABLE IF NOT EXISTS subscription_progress ( + sub_name VARCHAR PRIMARY KEY, + progress BIGINT +) ON CONFLICT OVERWRITE; +``` + +Here's an example python code for retrieving and updating the consumption progress: + +```sql +def get_last_progress(conn, sub_name): + with conn.cursor() as cur: + cur.execute("SELECT progress FROM subscription_progress WHERE sub_name = %s", (sub_name,)) + result = cur.fetchone() + return result[0] if result else None + +def update_progress(conn, sub_name, progress): + with conn.cursor() as cur: + cur.execute("INSERT INTO subscription_progress (sub_name, progress)", (sub_name, progress, progress)) + cur.execute("FLUSH") + conn.commit() +``` + +The client needs to retrieve the last progress during bootstrapping and periodically store the progress. + +```sql +# Fetch the last progress from subscription_progress table +last_progress = get_last_progress(conn, sub_name) + +with conn.cursor() as cur: + if last_progress is not None: + cur.execute( + "DECLARE cur SUBSCRIPTION CURSOR FOR {} SINCE {}".format(sub_name, last_progress)) + else: + cur.execute( + "DECLARE cur SUBSCRIPTION CURSOR FOR {};".format(sub_name)) + + while True: + cur.execute("FETCH NEXT FROM cur") + row = cur.fetchone() + last_progress = row[0] # 'rw_timestamp' is the progress indicator + + ... + + if trigger_update(): + update_progress(conn, sub_name, last_progress) +``` + +## Use case + +Potential use cases for subscriptions are as follows. If you have explored more use cases, feel free to share them with us in our [Slack channel](https://www.risingwave.com/slack). + +* **Real-time alerting/notification:** Subscribers can employ sophisticated alerting rules to detect abnormal events and notify downstream applications. +* **Event-driven architectures:** Develop event-driven systems that react to changes based on specific business logic, such as synchronizing data to microservices.