Support SQS Queue URL for continuous ingestion in S3

examples/clickpipe/object_storage_s3_sqs_iam_role/README.md

@@ -0,0 +1,9 @@
+## ClickPipe Object Storage: Unordered Mode with IAM role example
+
+This example demonstrates how to deploy a ClickPipe with S3 continuous ingestion using unordered mode (event-based via SQS) with IAM role authentication.
+
+## How to run
+
+- Rename `variables.sample.tfvars` to `variables.tfvars` and fill in all needed data.
+- Run `terraform init`
+- Run `terraform <plan|apply> -var-file=variables.tfvars`

examples/clickpipe/object_storage_s3_sqs_iam_role/main.tf

@@ -0,0 +1,106 @@
+variable "organization_id" {}
+variable "token_key" {}
+variable "token_secret" {}
+
+variable "service_id" {
+  description = "ClickHouse Cloud service ID"
+}
+
+variable "bucket_url" {
+  description = "S3 bucket URL pattern (e.g., s3://my-bucket/path/*.json)"
+}
+
+variable "sqs_queue_url" {
+  description = "SQS queue URL for S3 event notifications (e.g., https://sqs.us-east-1.amazonaws.com/123456789012/my-queue)"
+}
+
+variable "iam_role" {
+  description = "ARN of the IAM role with permissions to read from S3 and receive SQS messages"
+  sensitive   = true
+}
+
+# S3 ClickPipe with continuous ingestion using SQS event notifications
+# This example demonstrates event-based continuous ingestion where new files
+# are detected via S3 event notifications sent to an SQS queue, rather than
+# polling S3 for new files in lexicographical order.
+resource "clickhouse_clickpipe" "s3_sqs_continuous" {
+  name       = "S3 Continuous ClickPipe with SQS (IAM Role)"
+  service_id = var.service_id
+
+  source = {
+    object_storage = {
+      type   = "s3"
+      format = "JSONEachRow"
+      url    = var.bucket_url
+
+      # Enable continuous ingestion with event-based processing
+      is_continuous = true
+      queue_url     = var.sqs_queue_url
+
+      # IAM role authentication - recommended for AWS services
+      authentication = "IAM_ROLE"
+      iam_role       = var.iam_role
+    }
+  }
+
+  destination = {
+    table         = "s3_events_data"
+    managed_table = true
+
+    table_definition = {
+      engine = {
+        type = "MergeTree"
+      }
+
+      sorting_key = ["timestamp"]
+    }
+
+    columns = [
+      {
+        name = "id"
+        type = "String"
+      },
+      {
+        name = "timestamp"
+        type = "DateTime64(3)"
+      },
+      {
+        name = "event_type"
+        type = "String"
+      },
+      {
+        name = "data"
+        type = "String"
+      }
+    ]
+  }
+
+  field_mappings = [
+    {
+      source_field      = "id"
+      destination_field = "id"
+    },
+    {
+      source_field      = "timestamp"
+      destination_field = "timestamp"
+    },
+    {
+      source_field      = "event_type"
+      destination_field = "event_type"
+    },
+    {
+      source_field      = "data"
+      destination_field = "data"
+    }
+  ]
+}
+
+output "clickpipe_id" {
+  value       = clickhouse_clickpipe.s3_sqs_continuous.id
+  description = "The ID of the created ClickPipe"
+}
+
+output "clickpipe_state" {
+  value       = clickhouse_clickpipe.s3_sqs_continuous.state
+  description = "The current state of the ClickPipe"
+}

examples/clickpipe/object_storage_s3_sqs_iam_role/provider.tf

@@ -0,0 +1,13 @@
+terraform {
+  required_providers {
+    clickhouse = {
+      source = "ClickHouse/clickhouse"
+    }
+  }
+}
+
+provider "clickhouse" {
+  organization_id = var.organization_id
+  token_key       = var.token_key
+  token_secret    = var.token_secret
+}

examples/clickpipe/object_storage_s3_sqs_iam_role/provider.tf.template.alpha

@@ -0,0 +1,14 @@
+terraform {
+  required_providers {
+    clickhouse = {
+      version = "${CLICKHOUSE_TERRAFORM_PROVIDER_VERSION}"
+      source  = "ClickHouse/clickhouse"
+    }
+  }
+}
+
+provider "clickhouse" {
+  organization_id = var.organization_id
+  token_key       = var.token_key
+  token_secret    = var.token_secret
+}

examples/clickpipe/object_storage_s3_sqs_iam_role/variables.sample.tfvars

@@ -0,0 +1,13 @@
+# These keys are for example only and won't work when pointed to a deployed ClickHouse OpenAPI server
+organization_id = "aee076c1-3f83-4637-95b1-ad5a0a825b71"
+token_key       = "avhj1U5QCdWAE9CA9"
+token_secret    = "4b1dROiHQEuSXJHlV8zHFd0S7WQj7CGxz5kGJeJnca"
+service_id      = "aee076c1-3f83-4637-95b1-ad5a0a825b71"
+
+bucket_url = "s3://mybucket/path/*.json"
+
+# SQS queue URL must follow the format: https://sqs.{region}.amazonaws.com/{account-id}/{queue-name}
+sqs_queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/my-clickpipe-queue"
+
+# IAM role ARN with permissions to read S3 and receive SQS messages
+iam_role = "arn:aws:iam::123456789012:role/ClickPipeRole"

examples/clickpipe/object_storage_s3_sqs_iam_user/README.md

@@ -0,0 +1,9 @@
+## ClickPipe Object Storage: Unordered Mode with IAM user example
+
+This example demonstrates how to deploy a ClickPipe with S3 continuous ingestion using unordered mode (event-based via SQS) with IAM user authentication.
+
+## How to run
+
+- Rename `variables.sample.tfvars` to `variables.tfvars` and fill in all needed data.
+- Run `terraform init`
+- Run `terraform <plan|apply> -var-file=variables.tfvars`

examples/clickpipe/object_storage_s3_sqs_iam_user/main.tf

@@ -0,0 +1,114 @@
+variable "organization_id" {}
+variable "token_key" {}
+variable "token_secret" {}
+
+variable "service_id" {
+  description = "ClickHouse Cloud service ID"
+}
+
+variable "bucket_url" {
+  description = "S3 bucket URL pattern (e.g., s3://my-bucket/path/*.json)"
+}
+
+variable "sqs_queue_url" {
+  description = "SQS queue URL for S3 event notifications (e.g., https://sqs.us-east-1.amazonaws.com/123456789012/my-queue)"
+}
+
+variable "iam_access_key_id" {
+  description = "AWS IAM access key ID with permissions to read from S3 and receive SQS messages"
+  sensitive   = true
+}
+
+variable "iam_secret_key" {
+  description = "AWS IAM secret access key"
+  sensitive   = true
+}
+
+# S3 ClickPipe with continuous ingestion using SQS event notifications
+# This example demonstrates event-based continuous ingestion where new files
+# are detected via S3 event notifications sent to an SQS queue, rather than
+# polling S3 for new files in lexicographical order.
+resource "clickhouse_clickpipe" "s3_sqs_continuous" {
+  name       = "S3 Continuous ClickPipe with SQS (IAM User)"
+  service_id = var.service_id
+
+  source = {
+    object_storage = {
+      type   = "s3"
+      format = "JSONEachRow"
+      url    = var.bucket_url
+
+      # Enable continuous ingestion with event-based processing
+      is_continuous = true
+      queue_url     = var.sqs_queue_url
+
+      # IAM user authentication
+      authentication = "IAM_USER"
+      access_key = {
+        access_key_id = var.iam_access_key_id
+        secret_key    = var.iam_secret_key
+      }
+    }
+  }
+
+  destination = {
+    table         = "s3_events_data"
+    managed_table = true
+
+    table_definition = {
+      engine = {
+        type = "MergeTree"
+      }
+
+      sorting_key = ["timestamp"]
+    }
+
+    columns = [
+      {
+        name = "id"
+        type = "String"
+      },
+      {
+        name = "timestamp"
+        type = "DateTime64(3)"
+      },
+      {
+        name = "event_type"
+        type = "String"
+      },
+      {
+        name = "data"
+        type = "String"
+      }
+    ]
+  }
+
+  field_mappings = [
+    {
+      source_field      = "id"
+      destination_field = "id"
+    },
+    {
+      source_field      = "timestamp"
+      destination_field = "timestamp"
+    },
+    {
+      source_field      = "event_type"
+      destination_field = "event_type"
+    },
+    {
+      source_field      = "data"
+      destination_field = "data"
+    }
+  ]
+}
+
+output "clickpipe_id" {
+  value       = clickhouse_clickpipe.s3_sqs_continuous.id
+  description = "The ID of the created ClickPipe"
+}
+
+output "clickpipe_state" {
+  value       = clickhouse_clickpipe.s3_sqs_continuous.state
+  description = "The current state of the ClickPipe"
+}

examples/clickpipe/object_storage_s3_sqs_iam_user/provider.tf

@@ -0,0 +1,13 @@
+terraform {
+  required_providers {
+    clickhouse = {
+      source = "ClickHouse/clickhouse"
+    }
+  }
+}
+
+provider "clickhouse" {
+  organization_id = var.organization_id
+  token_key       = var.token_key
+  token_secret    = var.token_secret
+}

examples/clickpipe/object_storage_s3_sqs_iam_user/provider.tf.template.alpha

@@ -0,0 +1,14 @@
+terraform {
+  required_providers {
+    clickhouse = {
+      version = "${CLICKHOUSE_TERRAFORM_PROVIDER_VERSION}"
+      source  = "ClickHouse/clickhouse"
+    }
+  }
+}
+
+provider "clickhouse" {
+  organization_id = var.organization_id
+  token_key       = var.token_key
+  token_secret    = var.token_secret
+}

examples/clickpipe/object_storage_s3_sqs_iam_user/variables.sample.tfvars

@@ -0,0 +1,13 @@
+# These keys are for example only and won't work when pointed to a deployed ClickHouse OpenAPI server
+organization_id = "aee076c1-3f83-4637-95b1-ad5a0a825b71"
+token_key       = "avhj1U5QCdWAE9CA9"
+token_secret    = "4b1dROiHQEuSXJHlV8zHFd0S7WQj7CGxz5kGJeJnca"
+service_id      = "aee076c1-3f83-4637-95b1-ad5a0a825b71"
+
+bucket_url = "s3://mybucket/path/*.json"
+
+# SQS queue URL must follow the format: https://sqs.{region}.amazonaws.com/{account-id}/{queue-name}
+sqs_queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/my-clickpipe-queue"
+
+iam_access_key_id = "AKIAIOSFODNN7EXAMPLE"
+iam_secret_key    = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"

pkg/internal/api/clickpipe_models.go

@@ -114,7 +114,8 @@ type ClickPipeObjectStorageSource struct {
 	Delimiter   *string `json:"delimiter,omitempty"`
 	Compression *string `json:"compression,omitempty"`
 
-	IsContinuous bool `json:"isContinuous"`
+	IsContinuous bool    `json:"isContinuous"`
+	QueueURL     *string `json:"queueUrl,omitempty"`
 
 	Authentication *string                   `json:"authentication,omitempty"`
 	AccessKey      *ClickPipeSourceAccessKey `json:"accessKey,omitempty"`