> ## Documentation Index
> Fetch the complete documentation index at: https://deepline.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Clickhouse Click Pipe Create: Inputs, Cost & CLI Example
> Create ClickPipe. Clickhouse Clickhouse Click Pipe Create docs include request fields, response shape, pricing notes, and Deepline CLI examples.
## Run in Enrichment Spreadsheet
Use this function as a column step in `deepline enrich`.
```bash theme={null}
deepline enrich --input leads.csv --output leads.enriched.csv --with 'result=clickhouse_click_pipe_create:{"organizationId":"{{organizationId}}","serviceId":"{{serviceId}}"}' --json
```
Map payload values to spreadsheet columns with `{{column_name}}` placeholders.
## Input Schema
| Name | Type | Required | Default | Description |
| ------------------------ | -------- | -------- | ------- | ---------------------------------------------------------------------------------------------------- |
| `payload.organizationId` | `string` | Yes | | ID of the organization that owns the service. |
| `payload.serviceId` | `string` | Yes | | ID of the service to create the ClickPipe for. |
| `payload.name` | `string` | No | | Name of the ClickPipe. |
| `payload.source` | `object` | No | | |
| `payload.destination` | `object` | No | | |
| `payload.fieldMappings` | `array` | No | | Field mappings of the ClickPipe. Note that all destination columns must be included in the mappings. |
| `payload.scaling` | `object` | No | | |
| `payload.settings` | `object` | No | | |
Show raw input schema
### Input JSON Schema
```json theme={null}
{
"type": "object",
"description": "Create a new ClickPipe.",
"properties": {
"organizationId": {
"type": "string",
"description": "ID of the organization that owns the service.",
"format": "uuid"
},
"serviceId": {
"type": "string",
"description": "ID of the service to create the ClickPipe for.",
"format": "uuid"
},
"name": {
"type": "string",
"description": "Name of the ClickPipe."
},
"source": {
"properties": {
"kafka": {
"type": "object",
"additionalProperties": true
},
"objectStorage": {
"type": "object",
"additionalProperties": true
},
"kinesis": {
"type": "object",
"additionalProperties": true
},
"pubsub": {
"type": "object",
"additionalProperties": true
},
"postgres": {
"properties": {
"type": {
"type": [
"string",
"null"
],
"description": "Type of the Postgres source. Defaults to \"postgres\" if not specified.",
"enum": [
"postgres",
"supabase",
"neon",
"alloydb",
"planetscale",
"rdspostgres",
"aurorapostgres",
"cloudsqlpostgres",
"azurepostgres",
"crunchybridge",
"tigerdata"
]
},
"credentials": {
"properties": {
"username": {
"type": "string",
"description": "Database username."
},
"password": {
"type": "string",
"description": "Database password.",
"format": "password"
}
},
"additionalProperties": false
},
"host": {
"type": "string",
"description": "PostgreSQL server hostname or IP address. To use a reverse private endpoint, pass the endpoint hostname here.",
"format": "hostname"
},
"port": {
"type": "integer",
"description": "PostgreSQL server port.",
"minimum": 1,
"maximum": 65535
},
"database": {
"type": "string",
"description": "PostgreSQL database name to replicate from."
},
"settings": {
"properties": {
"syncIntervalSeconds": {
"type": "integer",
"description": "Interval in seconds to sync data from Postgres during CDC replication.",
"minimum": 1
},
"pullBatchSize": {
"type": "integer",
"description": "Number of rows to pull in each batch during CDC replication.",
"minimum": 1
},
"publicationName": {
"type": "string",
"description": "PostgreSQL publication name to use for CDC replication. If not provided, ClickPipes will create one automatically."
},
"replicationMode": {
"type": "string",
"description": "Replication mode: \"cdc\" (change data capture with initial snapshot), \"snapshot\" (one-time snapshot only), or \"cdc_only\" (CDC without initial snapshot).",
"enum": [
"cdc",
"snapshot",
"cdc_only"
]
},
"replicationSlotName": {
"type": "string",
"description": "PostgreSQL replication slot name. Only valid for \"cdc_only\" mode. For \"cdc\" mode, ClickPipes creates the slot automatically."
},
"allowNullableColumns": {
"type": "boolean",
"description": "Preserve nullability from Postgres in the destination ClickHouse table. When true, columns without NOT NULL constraints are created as Nullable(...). When false, all columns are non-nullable and NULL values are replaced with the default value for the type. Note: Nullable types have performance overhead in ClickHouse."
},
"initialLoadParallelism": {
"type": "integer",
"description": "Number of parallel workers to use per table in the initial snapshot phase.",
"minimum": 1
},
"snapshotNumRowsPerPartition": {
"type": "integer",
"description": "Number of rows per partition during the snapshot phase.",
"minimum": 1000
},
"snapshotNumberOfParallelTables": {
"type": "integer",
"description": "Number of tables to snapshot in parallel during the initial load phase.",
"minimum": 1
},
"enableFailoverSlots": {
"type": "boolean",
"description": "Enable failover support for the replication slot on PG17 and newer. Only applicable when ClickPipes creates the replication slot (i.e., replicationSlotName is NOT provided)."
},
"deleteOnMerge": {
"type": "boolean",
"description": "Enable hard delete behavior in ReplacingMergeTree for PostgreSQL DELETE operations."
}
},
"additionalProperties": false
},
"authentication": {
"type": "string",
"description": "Authentication method for Postgres connection.",
"enum": [
"basic",
"IAM_ROLE"
]
},
"iamRole": {
"type": "string",
"description": "IAM role ARN for IAM authentication (required for IAM_ROLE authentication)."
},
"tlsHost": {
"type": "string",
"description": "TLS/SSL host for secure connections."
},
"caCertificate": {
"type": "string",
"description": "PEM encoded CA certificate to validate the Postgres server certificate."
},
"disableTls": {
"type": "boolean",
"description": "Disable TLS for the Postgres connection. Use with caution in production environments."
},
"skipCertVerification": {
"type": "boolean",
"description": "Skip TLS certificate verification for the Postgres connection. Use with caution in production environments."
},
"tableMappings": {
"type": "array",
"description": "List of table mappings defining which PostgreSQL tables to replicate and how they map to ClickHouse tables.",
"items": {
"properties": {
"sourceSchemaName": {
"type": "string",
"description": "PostgreSQL source schema name."
},
"sourceTable": {
"type": "string",
"description": "PostgreSQL source table name."
},
"targetTable": {
"type": "string",
"description": "ClickHouse target table name, optionally prefixed with schema name (e.g., \"my_schema_my_table\"). The table will be created automatically if it does not exist. For snapshot mode, the target table must be empty."
},
"excludedColumns": {
"type": "array",
"description": "List of column names to exclude from replication. Column names must be unique within this list.",
"uniqueItems": true,
"items": {
"type": "string"
}
},
"useCustomSortingKey": {
"type": "boolean",
"description": "Whether to use a custom sorting key. If true, sortingKeys must be provided. If false or omitted, the default sorting key is the PostgreSQL primary key."
},
"sortingKeys": {
"type": "array",
"description": "Ordered list of column names to use as the sorting (ORDER BY) key in ClickHouse. Only used when useCustomSortingKey is true. Column names must be unique within this list.",
"uniqueItems": true,
"items": {
"type": "string"
}
},
"tableEngine": {
"type": "string",
"description": "ClickHouse table engine: \"ReplacingMergeTree\" (handles updates/deletes), \"MergeTree\" (append-only), or \"Null\" (forward data to materialized views without storing it).",
"enum": [
"MergeTree",
"ReplacingMergeTree",
"Null"
]
},
"partitionKey": {
"type": "string",
"description": "Custom partitioning column used for parallel snapshotting. Only beneficial for PostgreSQL 13 (no benefit for PG14+, which supports indexed ctid scans). Must be an indexed column of type: `smallint`, `integer`, `bigint`, `timestamp without time zone`, or `timestamp with time zone`. Unrelated to ClickHouse partitioning."
},
"partitionByExpr": {
"type": "string",
"description": "ClickHouse PARTITION BY expression applied to the destination table when ClickPipes creates it."
}
},
"additionalProperties": false
}
}
},
"additionalProperties": false
},
"mysql": {
"type": "object",
"additionalProperties": true
},
"bigquery": {
"type": "object",
"additionalProperties": true
},
"mongodb": {
"type": "object",
"additionalProperties": true
},
"validateSamples": {
"type": "boolean",
"description": "Validate data samples received from data source. It will validate the connection and data availability and correctness. If not enabled, only connection will be validated. This has no effect on Postgres or MySQL pipes, they always only validate the connection and table definitions. This is experimental and can be removed in the future."
}
},
"additionalProperties": false
},
"destination": {
"properties": {
"database": {
"type": "string",
"description": "Destination database."
},
"table": {
"type": "string",
"description": "Destination table. Required field for all pipe types except database pipes (Postgres, MySQL, BigQuery)."
},
"managedTable": {
"type": "boolean",
"description": "Is the table managed by ClickPipes? Required field for all pipe types except database pipes (Postgres, MySQL, BigQuery)."
},
"tableDefinition": {
"properties": {
"engine": {
"properties": {
"type": {
"type": "string",
"description": "Engine type of the destination table.",
"enum": [
"MergeTree",
"ReplacingMergeTree",
"SummingMergeTree",
"Null"
]
},
"versionColumnId": {
"type": [
"string",
"null"
],
"description": "Column name to use as version for ReplacingMergeTree engine."
},
"columnIds": {
"type": "array",
"description": "Column names to sum for SummingMergeTree engine.",
"items": {
"type": "string"
}
}
},
"additionalProperties": false
},
"sortingKey": {
"type": "array",
"description": "Sorting key of the destination table. List of columns.",
"items": {
"type": "string"
}
},
"partitionBy": {
"type": "string",
"description": "Partition key SQL expression."
},
"primaryKey": {
"type": "string",
"description": "Primary key of SQL expression."
}
},
"additionalProperties": false
},
"columns": {
"type": "array",
"description": "Columns of the destination table. Required field for all pipe types except database pipes (Postgres, MySQL, BigQuery).",
"items": {
"properties": {
"name": {
"type": "string",
"description": "Name of the column."
},
"type": {
"type": "string",
"description": "Type of the column."
}
},
"additionalProperties": false
}
},
"roles": {
"type": "array",
"description": "Optional. Roles to grant to the ClickHouse user that ClickPipe creates. If omitted, the user is granted the default role (`default_role`). Add your custom roles here if required.",
"items": {
"type": "string"
}
}
},
"additionalProperties": false
},
"fieldMappings": {
"type": "array",
"description": "Field mappings of the ClickPipe. Note that all destination columns must be included in the mappings.",
"items": {
"properties": {
"sourceField": {
"type": "string",
"description": "Source field name."
},
"destinationField": {
"type": "string",
"description": "Destination field name."
}
},
"additionalProperties": false
}
},
"scaling": {
"properties": {
"replicas": {
"type": "integer",
"description": "Desired number of replicas. Only for scalable pipes.",
"minimum": 1,
"maximum": 40
},
"concurrency": {
"type": "integer",
"description": "Desired number of concurrency. Only for S3 pipes. If set to 0, concurrency is auto-scaled based on the cluster memory."
},
"replicaCpuMillicores": {
"type": "integer",
"description": "CPU in millicores for each replica. Only for streaming pipes.",
"minimum": 125,
"maximum": 2000
},
"replicaMemoryGb": {
"type": "number",
"description": "Memory in GB for each replica. Only for streaming pipes.",
"minimum": 0.5,
"maximum": 8
}
},
"additionalProperties": false
},
"settings": {
"properties": {
"streaming_max_insert_wait_ms": {
"type": [
"integer",
"null"
],
"description": "Streaming max insert wait time. Configures the max wait period before inserting data into the ClickHouse.",
"minimum": 500,
"maximum": 60000
},
"object_storage_concurrency": {
"type": [
"integer",
"null"
],
"description": "Object storage concurrency. Number of concurrent file processing threads",
"minimum": 1,
"maximum": 35
},
"object_storage_polling_interval_ms": {
"type": [
"integer",
"null"
],
"description": "Object storage polling interval. Configures the refresh interval for querying continuous ingest for new object storage data",
"minimum": 100,
"maximum": 3600000
},
"object_storage_max_insert_bytes": {
"type": [
"integer",
"null"
],
"description": "Max insert bytes. Number of bytes to process in a single insert batch",
"minimum": 10485760,
"maximum": 53687091200
},
"object_storage_max_file_count": {
"type": [
"integer",
"null"
],
"description": "Max file count. Maximum number of files to process in a single insert batch",
"minimum": 1,
"maximum": 10000
},
"clickhouse_max_threads": {
"type": [
"integer",
"null"
],
"description": "Max threads. Maximum number of concurrent threads for file processing",
"minimum": 0,
"maximum": 64
},
"clickhouse_max_insert_threads": {
"type": [
"integer",
"null"
],
"description": "Max insert threads. Maximum number of concurrent insert threads",
"minimum": 0,
"maximum": 16
},
"clickhouse_min_insert_block_size_bytes": {
"type": [
"integer",
"null"
],
"description": "Min insert block size bytes. Minimum size of data block for insert (in bytes)",
"minimum": 0,
"maximum": 10737418240
},
"clickhouse_max_download_threads": {
"type": [
"integer",
"null"
],
"description": "Max download threads. Maximum number of concurrent download threads",
"minimum": 0,
"maximum": 32
},
"clickhouse_parallel_distributed_insert_select": {
"type": [
"integer",
"null"
],
"description": "Parallel distributed insert select. Parallel distributed insert select setting",
"minimum": 0,
"maximum": 2
},
"object_storage_use_cluster_function": {
"type": [
"boolean",
"null"
],
"description": "use cluster function. Whether to use ClickHouse cluster function for distributed processing"
},
"clickhouse_parallel_view_processing": {
"type": [
"boolean",
"null"
],
"description": "parallel view processing. Whether to enable pushing to attached views concurrently instead of sequentially"
}
},
"additionalProperties": false
}
},
"required": [
"organizationId",
"serviceId"
],
"additionalProperties": false
}
```
## Output Schema
| Name | Type | Required | Default | Description |
| ------------- | -------- | -------- | ------- | ---------------------------------------------- |
| `result.data` | `object` | Yes | | Provider response payload. |
| `result.meta` | `object` | No | | Additional response metadata (status, paging). |
Show raw output schema
### Output JSON Schema
```json theme={null}
{
"type": "object",
"description": "Standard tool result payload.",
"properties": {
"data": {
"type": "object",
"description": "Provider response payload.",
"properties": {
"status": {
"type": "number",
"description": "HTTP status code."
},
"requestId": {
"type": "string",
"description": "Unique id assigned to every request. UUIDv4",
"format": "uuid"
},
"result": {
"properties": {
"id": {
"type": "string",
"description": "Unique ClickPipe ID.",
"format": "uuid"
},
"serviceId": {
"type": "string",
"description": "ID of the service this ClickPipe belongs to.",
"format": "uuid"
},
"name": {
"type": "string",
"description": "Name of the ClickPipe.",
"minLength": 1,
"maxLength": 255
},
"state": {
"type": "string",
"description": "Current lifecycle state of the ClickPipe. For database pipes: \"Provisioning\" (initial setup), \"Setup\" (configuring replication), \"Snapshot\" (initial data load), \"Running\" (actively replicating), \"Pausing\" (transitioning to paused state), \"Paused\" (temporarily paused), \"Modifying\" (applying configuration updates), \"Resync\" (swapping resync tables with original tables), \"Failed\" (error occurred), \"Unknown\". For streaming/object storage pipes (Kafka, Kinesis, S3): \"Unknown\" (initial state), \"Provisioning\" (setting up resources), \"Running\" (actively ingesting data), \"Stopping\" (transitioning to stopped state), \"Stopped\" (manually stopped, can be restarted), \"Completed\" (batch ingestion finished for object storage), \"Failed\" (error occurred, pipe stopped), \"InternalError\" (internal system error).",
"enum": [
"Unknown",
"Provisioning",
"Running",
"Stopping",
"Stopped",
"Failed",
"Completed",
"InternalError",
"Setup",
"Snapshot",
"Paused",
"Pausing",
"Modifying",
"Resync"
]
},
"scaling": {
"properties": {
"replicas": {
"type": "integer",
"description": "Desired number of replicas. Only for scalable pipes.",
"minimum": 1,
"maximum": 40
},
"concurrency": {
"type": "integer",
"description": "Desired number of concurrency. Only for S3 pipes. If set to 0, concurrency is auto-scaled based on the cluster memory."
},
"replicaCpuMillicores": {
"type": "integer",
"description": "CPU in millicores for each replica. Only for streaming pipes.",
"minimum": 125,
"maximum": 2000
},
"replicaMemoryGb": {
"type": "number",
"description": "Memory in GB for each replica. Only for streaming pipes.",
"minimum": 0.5,
"maximum": 8
}
},
"additionalProperties": false
},
"source": {
"properties": {
"kafka": {
"type": "object",
"additionalProperties": true
},
"objectStorage": {
"type": "object",
"additionalProperties": true
},
"kinesis": {
"type": "object",
"additionalProperties": true
},
"pubsub": {
"type": "object",
"additionalProperties": true
},
"postgres": {
"type": "object",
"additionalProperties": true
},
"mysql": {
"type": "object",
"additionalProperties": true
},
"bigquery": {
"type": "object",
"additionalProperties": true
},
"mongodb": {
"type": "object",
"additionalProperties": true
}
},
"additionalProperties": false
},
"destination": {
"properties": {
"database": {
"type": "string",
"description": "Destination database."
},
"table": {
"type": "string",
"description": "Destination table. Required field for all pipe types except database pipes (Postgres, MySQL, BigQuery)."
},
"managedTable": {
"type": "boolean",
"description": "Is the table managed by ClickPipes? Required field for all pipe types except database pipes (Postgres, MySQL, BigQuery)."
},
"tableDefinition": {
"properties": {
"engine": {
"properties": {
"type": {
"type": "string",
"description": "Engine type of the destination table.",
"enum": [
"MergeTree",
"ReplacingMergeTree",
"SummingMergeTree",
"Null"
]
},
"versionColumnId": {
"type": [
"string",
"null"
],
"description": "Column name to use as version for ReplacingMergeTree engine."
},
"columnIds": {
"type": "array",
"description": "Column names to sum for SummingMergeTree engine.",
"items": {
"type": "string"
}
}
},
"additionalProperties": false
},
"sortingKey": {
"type": "array",
"description": "Sorting key of the destination table. List of columns.",
"items": {
"type": "string"
}
},
"partitionBy": {
"type": "string",
"description": "Partition key SQL expression."
},
"primaryKey": {
"type": "string",
"description": "Primary key of SQL expression."
}
},
"additionalProperties": false
},
"columns": {
"type": "array",
"description": "Columns of the destination table. Required field for all pipe types except database pipes (Postgres, MySQL, BigQuery).",
"items": {
"properties": {
"name": {
"type": "string",
"description": "Name of the column."
},
"type": {
"type": "string",
"description": "Type of the column."
}
},
"additionalProperties": false
}
}
},
"additionalProperties": false
},
"fieldMappings": {
"type": "array",
"description": "Field mappings of the ClickPipe. Note that all destination columns must be included in the mappings.",
"items": {
"properties": {
"sourceField": {
"type": "string",
"description": "Source field name."
},
"destinationField": {
"type": "string",
"description": "Destination field name."
}
},
"additionalProperties": false
}
},
"settings": {
"properties": {
"streaming_max_insert_wait_ms": {
"type": [
"integer",
"null"
],
"description": "Streaming max insert wait time. Configures the max wait period before inserting data into the ClickHouse.",
"minimum": 500,
"maximum": 60000
},
"object_storage_concurrency": {
"type": [
"integer",
"null"
],
"description": "Object storage concurrency. Number of concurrent file processing threads",
"minimum": 1,
"maximum": 35
},
"object_storage_polling_interval_ms": {
"type": [
"integer",
"null"
],
"description": "Object storage polling interval. Configures the refresh interval for querying continuous ingest for new object storage data",
"minimum": 100,
"maximum": 3600000
},
"object_storage_max_insert_bytes": {
"type": [
"integer",
"null"
],
"description": "Max insert bytes. Number of bytes to process in a single insert batch",
"minimum": 10485760,
"maximum": 53687091200
},
"object_storage_max_file_count": {
"type": [
"integer",
"null"
],
"description": "Max file count. Maximum number of files to process in a single insert batch",
"minimum": 1,
"maximum": 10000
},
"clickhouse_max_threads": {
"type": [
"integer",
"null"
],
"description": "Max threads. Maximum number of concurrent threads for file processing",
"minimum": 0,
"maximum": 64
},
"clickhouse_max_insert_threads": {
"type": [
"integer",
"null"
],
"description": "Max insert threads. Maximum number of concurrent insert threads",
"minimum": 0,
"maximum": 16
},
"clickhouse_min_insert_block_size_bytes": {
"type": [
"integer",
"null"
],
"description": "Min insert block size bytes. Minimum size of data block for insert (in bytes)",
"minimum": 0,
"maximum": 10737418240
},
"clickhouse_max_download_threads": {
"type": [
"integer",
"null"
],
"description": "Max download threads. Maximum number of concurrent download threads",
"minimum": 0,
"maximum": 32
},
"clickhouse_parallel_distributed_insert_select": {
"type": [
"integer",
"null"
],
"description": "Parallel distributed insert select. Parallel distributed insert select setting",
"minimum": 0,
"maximum": 2
},
"object_storage_use_cluster_function": {
"type": [
"boolean",
"null"
],
"description": "use cluster function. Whether to use ClickHouse cluster function for distributed processing"
},
"clickhouse_parallel_view_processing": {
"type": [
"boolean",
"null"
],
"description": "parallel view processing. Whether to enable pushing to attached views concurrently instead of sequentially"
}
},
"additionalProperties": false
},
"createdAt": {
"type": "string",
"description": "Creation timestamp of the ClickPipe in ISO 8601 format.",
"format": "date-time"
},
"updatedAt": {
"type": "string",
"description": "Last update timestamp of the ClickPipe in ISO 8601 format.",
"format": "date-time"
}
},
"additionalProperties": false
}
},
"additionalProperties": false
},
"meta": {
"type": "object",
"description": "Additional response metadata (status, paging).",
"additionalProperties": true
}
},
"required": [
"data"
],
"additionalProperties": false
}
```
## Advanced: Direct CLI
Use direct execution for single payload debugging.
```bash theme={null}
deepline tools execute clickhouse_click_pipe_create --payload '{
"organizationId": "string",
"serviceId": "string"
}' --json
```
### CLI flags
| Flag | Description |
| --------------------------- | --------------------------------------------------- |
| `--json` | Print machine-readable output. |
| `--wait` | Wait for terminal provider status when supported. |
| `--debug` | Enable wait mode with additional status/log output. |
| `--wait-timeout SECONDS` | Max seconds to wait in wait mode. |
| `--poll-interval SECONDS` | Polling interval in seconds during wait mode. |
| `--timeout SECONDS` | Request timeout in seconds. |
| `--connect-timeout SECONDS` | Connection timeout in seconds. |
## Cost
* Pricing model: `fixed` (per call).
* Estimated Deepline credits: `0` per pricing unit.
* Provider-native pricing may still exist outside Deepline credit billing.
* Billing mode: `no_bill`.