\[, … ])
Optionally, this clause specifies [#stream\_parameters](#stream_parameters "mention").
#### select\_statement
This statement specifies the [SELECT](/reference/sql-syntax/query/select.md) statement to run.
## Table Parameters
| Parameter Name | Description |
| -------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `store` | The name of the store that hosts the entity for this stream.
Required: No
Default value: User’s default /pages/khAma1ENd997ICjDGcJW.
Type: String
Valid values: See /pages/yGCSID06tqSuDcvkGevU
|
### **Snowflake-Specific Parameters**
| Parameter Name | Description |
| ------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `snowflake.db.name` | The name of the Snowflake database that would host the Snowflake /pages/6UJYcUVtorC4Xmf9nEwB#table.
Required: Yes
Default values: None
Type: String
Valid values: Database names available from /pages/ZVNhODGeUbLHKEa7VuaP.
|
| `snowflake.schema.name` | The name of the Snowflake schema that would host the Snowflake /pages/6UJYcUVtorC4Xmf9nEwB#table.
Required: Yes
Default value: None
Type: String
Valid values: Schema names available from /pages/ZVNhODGeUbLHKEa7VuaP under the snowflake.db.name database name.
|
| `snowflake.table.name` | The name of the Snowflake table to use when creating the /pages/6UJYcUVtorC4Xmf9nEwB#table. If the table doesn't exist in the store, a table with the table\_name is created in the corresponding store.
Required: No
Default value: table\_name
Type: String
|
| `snowflake.buffer.millis` | The amount of time to buffer events with the sink /pages/6UJYcUVtorC4Xmf9nEwB#table before writing the data into the corresponding Snowflake table.
It is important to note that the longer events are buffered, the larger the internal state of the query gets. Depending on the volume and size of the events for the corresponding query, memory limitations may be reached.
Required: No
Default value: 1000
Type: Long
Valid values: (0, ...]
|
Snowflake stores provide a delivery guarantee of `at_least_once` when producing events into a sink [Database](/overview/core-concepts/databases.md#table). Snowflake stores provide an insert-only mode when writing to Snowflake tables.
### **Databricks-Specific Parameters**
| Parameter Name | Description |
| -------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `databricks.catalog.name` | The name of the Databricks catalog that would host the Databricks /pages/6UJYcUVtorC4Xmf9nEwB#table.
Required: Yes
Default values: None
Type: String
Valid values: Database names available from /pages/ZVNhODGeUbLHKEa7VuaP.
|
| `databricks.schema.name` | The name of the Databricks schema that would host the Databricks /pages/6UJYcUVtorC4Xmf9nEwB#table.
Required: Yes
Default value: None
Type: String
Valid values: Schema names available from /pages/ZVNhODGeUbLHKEa7VuaP under the databricks.catalog.name catalog name.
|
| `databricks.table.name` | The name of the Databricks table to use when creating the /pages/6UJYcUVtorC4Xmf9nEwB#table. A table with the table\_name is created in the corresponding store. If that table already exists, then an error is reported back to the user.
Required: No
Default value: table\_name
Type: String
|
| `table.data.file.location` | The S3 directory location for the Delta-formatted data to be written. This location is either a full S3 path, or a relative path to the data store's Cloud Provider Bucket (see /pages/7FWIfCeUUJkQEcKKHSRB). The credentials for writing to S3 are given during data store creation (see /pages/7FWIfCeUUJkQEcKKHSRB). Note that the S3 bucket from the location specified by this parameter must match the databricks.cloud.s3.bucket property defined in the data store.
Required: Yes
Default value: None
Type: String
|
Databricks stores provide a delivery guarantee of `exactly_once` when producing events into a sink [Database](/overview/core-concepts/databases.md#table). Databricks stores provide an insert-only mode when writing to Databricks tables.
### Postgres-Specific Parameters
| Parameter Name | Description |
| ---------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| postgresql.db.name | Required: Yes
See CDC source parameters for the same parameter name
|
| postgresql.schema.name | Required: Yes
See CDC source parameters for the same parameter name
|
| postgresql.table.name | Required: No
See CDC source parameters for the same parameter name
|
### Iceberg-Specific Parameters
| Parameter Name | Description |
| ----------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `iceberg.aws.glue.db.name` | The name of the database (namespace) in the AWS Glue instance (used as the Iceberg Catalog implementation in the Iceberg Store) containing the existing Iceberg table. Required: Yes
Type: String
|
| `iceberg.aws.glue.table.name` | The name of the existing Iceberg table in the AWS Glue instance (used as the Iceberg Catalog implementation in the Iceberg Store).
Required: Yes
Type: String
|
### S3-Specific Parameters
The S3 sink writes the results of a query to an S3 store as newline-delimited JSON (the `jsonl` format which is one JSON object per record). Records are written as part files under the configured `s3.uri` path and rolled into new files based on a configurable rolling policy (elapsed time, part-file size, and how often the sink checks whether a part file should be rolled). When neither a rollover interval nor a file size is set, the connector's built-in default rolling policy applies. The S3 sink is append-only: records are only added and never updated or deleted. Optionally, a `PARTITION BY` clause lays out the written objects into Hive-style partition directories (`column=value/`); partition columns are used only for the directory layout and are not included in the JSON payload. All access to S3 uses the credentials and region configured on the associated store.
| Parameter Name | Description |
| ------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `s3.uri` | The S3 URI, using the canonical s3:// scheme, that the query writes its results to. It must reference a sub-folder of the store's URI. The s3a:// and s3n:// schemes are not accepted.
Required: Yes
Default value: None
Type: String
|
| `value.format` | Format of the records written to S3. Records are written as newline-delimited JSON. When omitted, the format is inferred from the source relation; an error is reported if the source has no value format.
Required: No
Default value: Inferred from the source relation
Type: String
Valid values: json, jsonl (jsonl is treated as json)
|
| `s3.sink.rolling.policy.rollover.interval` | The maximum amount of time a part file can stay open before it is rolled into a new file.
Required: No
Default value: The connector's default rolling policy applies when neither this property nor s3.sink.rolling.policy.file.size is set.
Type: String (duration, e.g. '15 min')
|
| `s3.sink.rolling.policy.file.size` | The maximum size a part file can reach before it is rolled into a new file.
Required: No
Default value: The connector's default rolling policy applies when neither this property nor s3.sink.rolling.policy.rollover.interval is set.
Type: String (memory size, e.g. '128 mb')
|
| `s3.sink.rolling.policy.check.interval` | How often the sink checks whether an open part file should be rolled based on the rollover interval.
Required: No
Default value: The connector's default rolling policy applies.
Type: String (duration, e.g. '5 min')
|
## Examples — Snowflake
#### Create a copy of a stream in a Snowflake table
The following creates a replica of the source [Database](/overview/core-concepts/databases.md#stream), `pageviews` in the Snowflake [Database](/overview/core-concepts/databases.md#table), `PV_TABLE`:
```sql
CREATE TABLE "PV_TABLE" WITH (
'store' = 'sfstore',
'snowflake.db.name' = 'DELTA_STREAMING',
'snowflake.schema.name' = 'PUBLIC'
) AS SELECT * FROM pageviews;
```
#### Create a stream of changes for a changelog in a Snowflake table
The following CTAS query creates a new Snowflake [Database](/overview/core-concepts/databases.md#table) to store incremental changes resulting from a grouping aggregation on the `transactions` [Database](/overview/core-concepts/databases.md#stream):
```sql
CREATE TABLE "CC_TYPE_USAGE" WITH (
'store' = 'sfstore',
'snowflake.db.name' = 'DELTA_STREAMING',
'snowflake.schema.name' = 'PUBLIC'
) AS SELECT
cc_type AS "CREDIT_TYPE",
tx_time AS "TRANSACTION_TS",
tx_id AS "TRANSACTION_ID"
FROM transactions
GROUP BY cc_type;
```
This query stores insert and update changes to the grouping column `cc_type` to the sink table `CC_TYPE_USAGE`.
## Examples — PostgreSQL
Assumptions for the following 2 examples:
* `pageviews` is a stream you have already defined.
* `ps_store` is a PostgreSQL data store you have already created.
* For details and an example, see [CREATE STORE](https://docs.deltastream.io/reference/sql-syntax/ddl/create-store#create-a-postgresql-store).
#### Create a new table in a PostgresSQL data store — example 1
In this example, you create a new table, named `pageviews`, in the given PostgreSQL data store under the public schema. The table has 3 columns:
1. viewtime
2. uid
3. pageid.
The query writes its results into this table as its sink.
```sql
CREATE TABLE pg_table WITH (
'store'='ps_store',
'postgresql.db.name'='demo',
'postgresql.schema.name'='public',
'postgresql.table.name'='pageviews') AS
SELECT viewtime, userid AS uid, pageid
FROM pageviews;
```
#### Create a new table in a PostgresSQL data store — example 2
This query creates a new table, named `pagevisits`, in the given PostgreSQL data store under the public schema. The table has 2 columns:
1. `userid`
2. `cnt`
The query writes its results into this table.
```sql
CREATE TABLE pagevisits WITH (
'store'='ps_store',
'postgresql.db.name'='demo',
'postgresql.schema.name'='public') AS
SELECT userid, count(*) AS cnt
FROM pageviews
GROUP BY userid;
```
{% hint style="info" %}
**Note** Since the query includes a `GROUP BY` clause, the `userid` column — which is the grouping column — is considered the primary key for the results. The `pagevisits` table in PostgreSQL is created accordingly, with `userid` as its primary column.
{% endhint %}
## Examples — Databricks
#### Create a copy of a stream in a Databricks table
The following creates a replica of the source [Database](/overview/core-concepts/databases.md#stream), `pageviews` in the Databricks [Database](/overview/core-concepts/databases.md#table), `pageviews_db`:
```sql
CREATE TABLE pageviews_db WITH (
'store' = 'databricks_store',
'databricks.catalog.name' = 'catalog1',
'databricks.schema.name' = 'schema1',
'databricks.table.name' = 'pageviews',
'table.data.file.location' = 's3://mybucket/test/0/pageviews'
) AS SELECT * FROM pageviews;
```
Upon issuing this query, a Databricks table is created in `catalog1.schema1.pageviews` that uses `s3://mybucket/test/0/pageviews` as its external location. This query writes the Delta-formatted parquet files and updates the Delta log in that S3 location.
## Examples — S3
#### **Create a S3 Table**
The following writes the results of a query over the `pageviews` Database to the S3 store `s3_store` as newline-delimited JSON under the `s3://bucket_name/ctas_output/` path:
```sql
CREATE TABLE pageviews_s3 WITH (
'store' = 's3_store',
's3.uri' = 's3://bucket_name/ctas_output/',
'value.format' = 'json'
) AS SELECT viewtime, userid AS "UID", pageid
FROM pageviews;
```
#### Create a partitioned S3 Table
In this example, a partitioned Table is created in the `pgv_parts` bucket in `s3_store` . The written objects are partitioned by `userid` into Hive-style directories:
```sql
CREATE TABLE pageviews_s3_part WITH (
'store' = 's3_store',
's3.uri' = 's3://bucket_name/pgv_parts/',
'value.format' = 'json'
) AS SELECT viewtime, userid, pageid
FROM pageviews
PARTITION BY userid;
```