# GENERATE STREAM DDL

## Syntax <a href="#synopsis" id="synopsis"></a>

{% code overflow="wrap" %}

```sql
GENERATE STREAM DDL stream_name WITH (stream_parameter = value [, ...]);
```

{% endcode %}

## Description

This command generates a CREATE DDL for STREAM from a given schema or data — for example, [#entity](https://docs.deltastream.io/overview/core-concepts/store#entity "mention") — that has corresponding [#protocol-buffers-and-descriptors](https://docs.deltastream.io/reference/data-format-serialization#protocol-buffers-and-descriptors "mention"). The command specially simplifies creating DDL for complex descriptors.

You can generate DDL if the current role has [`USAGE`](https://docs.deltastream.io/overview/core-concepts/access-control#privilege) privileges on the [store](https://docs.deltastream.io/overview/core-concepts/store "mention").

### Stream Parameters <a href="#parameters" id="parameters"></a>

| Parameter Name                | Description                                                                                                                                                                                                                                     |
| ----------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **`stream_name`**             | The name of the new stream.                                                                                                                                                                                                                     |
| **`topic`**                   | <p>Name of the Kafka topic whose assigned data schema you're using (for Protobuf descriptor and Avro schema from SchemaRegistry).<br></p><p><strong>Required?</strong>  Yes, for <code>AVRO</code>, <code>PROTOBUF</code></p>                   |
| **`value. format`**           | <p><code>AVRO</code>, <code>PROTOBUF</code>, <code>JSON</code></p><p></p><p><strong>Required?</strong>  Yes.</p>                                                                                                                                |
| **`key.format`**              | <p>AVRO, PROTOBUF, JSON</p><p></p><p><strong>Required?</strong>  No.</p>                                                                                                                                                                        |
| **`data.json.value.content`** | <p>A string containing one or more JSON objects separated by commas. It serves as a data sample you use to infer the schema of the value field in Kafka.</p><p></p><p><strong>Required?</strong>  Required only for <code>JSON</code></p>       |
| **`data.json.key.content`**   | <p>A string containing one or more <code>JSON</code> objects separated by commas. It serves as a data sample you used to infer the schema of the key in Kafka.</p><p></p><p><strong>Required?</strong>  Required only for <code>JSON</code></p> |
| **`postgresql.table.name`**   | <p>Name of the source table in the <code>POSTGRESQL</code> store.</p><p></p><p><strong>Required:</strong> only for  <code>POSTGRESQL</code> store.</p><p><strong>Type:</strong> String</p>                                                      |
| **`postgresql.schema.name`**  | <p>Name of the schema in the <code>POSTGRESQL</code> store containing the source table.<br></p><p><strong>Required:</strong> only for  <code>POSTGRESQL</code> store.</p><p><strong>Type:</strong> String</p>                                   |
| **`postgresql.db.name`**      | <p>Name of the database in the <code>POSTGRESQL</code> store containing the source table.</p><p></p><p><strong>Required:</strong> only for  <code>POSTGRESQL</code> store.</p><p><strong>Type:</strong> String</p>                              |

## Generating "Create Stream" DDL examples <a href="#examples" id="examples"></a>

**Generate a "create stream" ddl for an entity in the current store, which includes a protobuf descriptor.**

{% code overflow="wrap" %}

```sql
GENERATE STREAM DDL myStream with('topic'='pageviews_pb', 'value.format'='protobuf', 
'key.format'='protobuf');
```

{% endcode %}

The output would be:

{% code overflow="wrap" lineNumbers="true" fullWidth="false" %}

```sql
CREATE STREAM myStream(
 "viewtime" BIGINT NOT NULL,
  "userid" VARCHAR NOT NULL,
  "pageid" VARCHAR NOT NULL
 ) WITH ('key.format'='protobuf', 'key.type'='STRUCT<"userid" VARCHAR>', 'topic'='pageviews_pb', 'value.format'='protobuf');
```

{% endcode %}

**Generate a "create stream" ddl for an entity in the current store, which includes AVRO in the Confluent Schema Registry.**

{% code overflow="wrap" %}

```sql
GENERATE STREAM DDL myStream with('topic'='pageviews_avro', 'value.format'='avro', 
'key.format'='avro');
```

{% endcode %}

The output would be:

{% code overflow="wrap" lineNumbers="true" fullWidth="false" %}

```sql
CREATE STREAM myStream(
 "viewtime" BIGINT ,
  "userid" VARCHAR ,
  "pageid" VARCHAR 
) WITH ('key.format'='avro', 'key.type'='STRUCT<"userid" VARCHAR>', 'topic'='pageviews_avro', 'value.format'='avro');     
```

{% endcode %}

**Generate a "create stream" ddl for an entity whose content is JSON.**

{% code overflow="wrap" %}

```sql
GENERATE STREAM DDL myStream with ('value.format'='json', 'data.json.value.content'='{"viewtime":1629453600000,"userid":"user_123","pageid":"page_1"},{"viewtime":1629457200000,"userid":"user_456","pageid":"page_2"},{"viewtime":1629460800000,"userid":"user_789","pageid":"page_3"}' );
```

{% endcode %}

The output would be:

{% code overflow="wrap" fullWidth="false" %}

```sql
CREATE STREAM myStream(
 "viewtime" BIGINT,
  "userid" VARCHAR,
  "pageid" VARCHAR 
) WITH ('value.format'='json'); 
```

{% endcode %}

**Generate a "create stream" ddl for a Postgres table in a CDC Pipeline**

{% code overflow="wrap" %}

```sql
GENERATE STREAM DDL myStream with('value.format'='json', 'store'='pgStore', 'postgresql.table.name'='pgTb', 'postgresql.schema.name'='pgSchema', 'postgresql.db.name'='pgDb');
```

{% endcode %}

{% code overflow="wrap" %}

```sql
CREATE STREAM mystream( op VARCHAR,
 ts_ms BIGINT,
 "before" STRUCT<"id" BIGINT, "first_name" VARCHAR, "last_name" VARCHAR, "email" VARCHAR, "biography" VARCHAR>,
 "after" STRUCT<"id" BIGINT, "first_name" VARCHAR, "last_name" VARCHAR, "email" VARCHAR, "biography" VARCHAR>,
 "source" STRUCT<"db" VARCHAR, "schema" VARCHAR, "table" VARCHAR, "lsn" BIGINT> 
) WITH ('postgresql.db.name'='pgDb', 'postgresql.schema.name'='pgSchema', 'postgresql.table.name'='pgTb', 'store'='pgStore');
```

{% endcode %}