diff --git a/src/sinks/clickhouse/config.rs b/src/sinks/clickhouse/config.rs
index 0efffb97ef3ce..a6fad6a11cdc3 100644
--- a/src/sinks/clickhouse/config.rs
+++ b/src/sinks/clickhouse/config.rs
@@ -25,11 +25,11 @@ pub struct ClickhouseConfig {
 
     /// The table that data is inserted into.
     #[configurable(metadata(docs::examples = "mytable"))]
-    pub table: String,
+    pub table: Template,
 
     /// The database that contains the table that data is inserted into.
     #[configurable(metadata(docs::examples = "mydatabase"))]
-    pub database: Option<String>,
+    pub database: Option<Template>,
 
     /// Sets `input_format_skip_unknown_fields`, allowing ClickHouse to discard fields not present in the table schema.
     #[serde(default)]
@@ -90,12 +90,10 @@ impl SinkConfig for ClickhouseConfig {
         let service = ClickhouseService::new(
             client.clone(),
             auth.clone(),
-            &endpoint,
-            self.database.as_deref(),
-            self.table.as_str(),
+            endpoint.clone(),
             self.skip_unknown_fields,
             self.date_time_best_effort,
-        )?;
+        );
 
         let request_limits = self.request.unwrap_with(&Default::default());
         let service = ServiceBuilder::new()
@@ -103,12 +101,19 @@ impl SinkConfig for ClickhouseConfig {
             .service(service);
 
         let batch_settings = self.batch.into_batcher_settings()?;
+        let database = self.database.clone().unwrap_or_else(|| {
+            "default"
+                .try_into()
+                .expect("'default' should be a valid template")
+        });
         let sink = ClickhouseSink::new(
             batch_settings,
             self.compression,
             self.encoding.clone(),
             service,
             protocol,
+            database,
+            self.table.clone(),
         );
 
         let healthcheck = Box::pin(healthcheck(client, endpoint, auth));
@@ -126,7 +131,6 @@ impl SinkConfig for ClickhouseConfig {
 }
 
 async fn healthcheck(client: HttpClient, endpoint: Uri, auth: Option<Auth>) -> crate::Result<()> {
-    // TODO: check if table exists?
     let uri = format!("{}/?query=SELECT%201", endpoint);
     let mut request = Request::get(uri).body(Body::empty()).unwrap();
 
diff --git a/src/sinks/clickhouse/integration_tests.rs b/src/sinks/clickhouse/integration_tests.rs
index f16b19ffcc7ed..0e18a7bd2a375 100644
--- a/src/sinks/clickhouse/integration_tests.rs
+++ b/src/sinks/clickhouse/integration_tests.rs
@@ -45,7 +45,7 @@ async fn insert_events() {
 
     let config = ClickhouseConfig {
         endpoint: host.parse().unwrap(),
-        table: table.clone(),
+        table: table.clone().try_into().unwrap(),
         compression: Compression::None,
         batch,
         request: TowerRequestConfig {
@@ -94,7 +94,7 @@ async fn skip_unknown_fields() {
 
     let config = ClickhouseConfig {
         endpoint: host.parse().unwrap(),
-        table: table.clone(),
+        table: table.clone().try_into().unwrap(),
         skip_unknown_fields: true,
         compression: Compression::None,
         batch,
@@ -140,7 +140,7 @@ async fn insert_events_unix_timestamps() {
 
     let config = ClickhouseConfig {
         endpoint: host.parse().unwrap(),
-        table: table.clone(),
+        table: table.clone().try_into().unwrap(),
         compression: Compression::None,
         encoding: Transformer::new(None, None, Some(TimestampFormat::Unix)).unwrap(),
         batch,
@@ -269,7 +269,7 @@ async fn no_retry_on_incorrect_data() {
 
     let config = ClickhouseConfig {
         endpoint: host.parse().unwrap(),
-        table: table.clone(),
+        table: table.clone().try_into().unwrap(),
         compression: Compression::None,
         batch,
         ..Default::default()
@@ -320,7 +320,7 @@ async fn no_retry_on_incorrect_data_warp() {
 
     let config = ClickhouseConfig {
         endpoint: host.parse().unwrap(),
-        table: gen_table(),
+        table: gen_table().try_into().unwrap(),
         batch,
         ..Default::default()
     };
@@ -338,6 +338,70 @@ async fn no_retry_on_incorrect_data_warp() {
     assert_eq!(receiver.try_recv(), Ok(BatchStatus::Rejected));
 }
 
+#[tokio::test]
+async fn templated_table() {
+    trace_init();
+
+    let n_tables = 2;
+    let table_events: Vec<(String, Event, BatchStatusReceiver)> = (0..n_tables)
+        .map(|_| {
+            let table = gen_table();
+            let (mut event, receiver) = make_event();
+            event.as_mut_log().insert("table", table.as_str());
+            (table, event, receiver)
+        })
+        .collect();
+
+    let host = clickhouse_address();
+
+    let mut batch = BatchConfig::default();
+    batch.max_events = Some(1);
+
+    let config = ClickhouseConfig {
+        endpoint: host.parse().unwrap(),
+        table: "{{ .table }}".try_into().unwrap(),
+        batch,
+        ..Default::default()
+    };
+
+    let client = ClickhouseClient::new(host);
+    for (table, _, _) in &table_events {
+        client
+            .create_table(
+                table,
+                "host String, timestamp String, message String, table String",
+            )
+            .await;
+    }
+
+    let (sink, _hc) = config.build(SinkContext::default()).await.unwrap();
+
+    let events: Vec<Event> = table_events
+        .iter()
+        .map(|(_, event, _)| event.clone())
+        .collect();
+    run_and_assert_sink_compliance(sink, stream::iter(events), &SINK_TAGS).await;
+
+    for (table, event, mut receiver) in table_events {
+        let output = client.select_all(&table).await;
+        assert_eq!(1, output.rows, "table {} should have 1 row", table);
+
+        let expected = serde_json::to_value(event.into_log()).unwrap();
+        assert_eq!(
+            expected, output.data[0],
+            "table \"{}\"'s one row should have the correct data",
+            table
+        );
+
+        assert_eq!(
+            receiver.try_recv(),
+            Ok(BatchStatus::Delivered),
+            "table \"{}\"'s event should have been delivered",
+            table
+        );
+    }
+}
+
 fn make_event() -> (Event, BatchStatusReceiver) {
     let (batch, receiver) = BatchNotifier::new_with_receiver();
     let mut event = LogEvent::from("raw log line").with_batch_notifier(&batch);
diff --git a/src/sinks/clickhouse/service.rs b/src/sinks/clickhouse/service.rs
index ecce62e537290..f912b99e6df22 100644
--- a/src/sinks/clickhouse/service.rs
+++ b/src/sinks/clickhouse/service.rs
@@ -19,6 +19,8 @@ use crate::{
 
 #[derive(Debug, Clone)]
 pub struct ClickhouseRequest {
+    pub database: String,
+    pub table: String,
     pub body: Bytes,
     pub compression: Compression,
     pub finalizers: EventFinalizers,
@@ -107,31 +109,28 @@ impl RetryLogic for ClickhouseRetryLogic {
 #[derive(Debug, Clone)]
 pub struct ClickhouseService {
     client: HttpClient,
-    uri: Uri,
     auth: Option<Auth>,
+    endpoint: Uri,
+    skip_unknown_fields: bool,
+    date_time_best_effort: bool,
 }
 
 impl ClickhouseService {
     /// Creates a new `ClickhouseService`.
-    pub fn new(
+    pub const fn new(
         client: HttpClient,
         auth: Option<Auth>,
-        endpoint: &Uri,
-        database: Option<&str>,
-        table: &str,
+        endpoint: Uri,
         skip_unknown_fields: bool,
         date_time_best_effort: bool,
-    ) -> crate::Result<Self> {
-        // Set the URI query once during initialization, as it won't change throughout the lifecycle
-        // of the service.
-        let uri = set_uri_query(
+    ) -> Self {
+        Self {
+            client,
+            auth,
             endpoint,
-            database.unwrap_or("default"),
-            table,
             skip_unknown_fields,
             date_time_best_effort,
-        )?;
-        Ok(Self { client, auth, uri })
+        }
     }
 }
 
@@ -148,22 +147,32 @@ impl Service<ClickhouseRequest> for ClickhouseService {
     // Emission of Error internal event is handled upstream by the caller.
     fn call(&mut self, request: ClickhouseRequest) -> Self::Future {
         let mut client = self.client.clone();
+        let auth = self.auth.clone();
 
-        let mut builder = Request::post(&self.uri)
-            .header(CONTENT_TYPE, "application/x-ndjson")
-            .header(CONTENT_LENGTH, request.body.len());
-        if let Some(ce) = request.compression.content_encoding() {
-            builder = builder.header(CONTENT_ENCODING, ce);
-        }
-        if let Some(auth) = &self.auth {
-            builder = auth.apply_builder(builder);
-        }
-
-        let http_request = builder
-            .body(Body::from(request.body))
-            .expect("building HTTP request failed unexpectedly");
+        // Build the URI outside of the boxed future to avoid unnecessary clones.
+        let uri = set_uri_query(
+            &self.endpoint,
+            &request.database,
+            &request.table,
+            self.skip_unknown_fields,
+            self.date_time_best_effort,
+        );
 
         Box::pin(async move {
+            let mut builder = Request::post(&uri?)
+                .header(CONTENT_TYPE, "application/x-ndjson")
+                .header(CONTENT_LENGTH, request.body.len());
+            if let Some(ce) = request.compression.content_encoding() {
+                builder = builder.header(CONTENT_ENCODING, ce);
+            }
+            if let Some(auth) = auth {
+                builder = auth.apply_builder(builder);
+            }
+
+            let http_request = builder
+                .body(Body::from(request.body))
+                .expect("building HTTP request failed unexpectedly");
+
             let response = client.call(http_request).in_current_span().await?;
             let (parts, body) = response.into_parts();
             let body = body::to_bytes(body).await?;
diff --git a/src/sinks/clickhouse/sink.rs b/src/sinks/clickhouse/sink.rs
index 805cc50bcf4bc..44013b85b2e6e 100644
--- a/src/sinks/clickhouse/sink.rs
+++ b/src/sinks/clickhouse/sink.rs
@@ -2,7 +2,7 @@ use bytes::Bytes;
 use codecs::{encoding::Framer, JsonSerializerConfig, NewlineDelimitedEncoderConfig};
 
 use super::service::{ClickhouseRequest, ClickhouseRetryLogic, ClickhouseService};
-use crate::{internal_events::SinkRequestBuildError, sinks::prelude::*};
+use crate::sinks::prelude::*;
 
 pub struct ClickhouseSink {
     batch_settings: BatcherSettings,
@@ -10,6 +10,8 @@ pub struct ClickhouseSink {
     encoding: (Transformer, Encoder<Framer>),
     service: Svc<ClickhouseService, ClickhouseRetryLogic>,
     protocol: &'static str,
+    database: Template,
+    table: Template,
 }
 
 impl ClickhouseSink {
@@ -19,6 +21,8 @@ impl ClickhouseSink {
         transformer: Transformer,
         service: Svc<ClickhouseService, ClickhouseRetryLogic>,
         protocol: &'static str,
+        database: Template,
+        table: Template,
     ) -> Self {
         Self {
             batch_settings,
@@ -32,12 +36,18 @@ impl ClickhouseSink {
             ),
             service,
             protocol,
+            database,
+            table,
         }
     }
 
     async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
         input
-            .batched(self.batch_settings.into_byte_size_config())
+            .batched_partitioned(
+                KeyPartitioner::new(self.database, self.table),
+                self.batch_settings,
+            )
+            .filter_map(|(key, batch)| async move { key.map(move |k| (k, batch)) })
             .request_builder(
                 None,
                 ClickhouseRequestBuilder {
@@ -76,8 +86,8 @@ struct ClickhouseRequestBuilder {
     encoding: (Transformer, Encoder<Framer>),
 }
 
-impl RequestBuilder<Vec<Event>> for ClickhouseRequestBuilder {
-    type Metadata = EventFinalizers;
+impl RequestBuilder<(PartitionKey, Vec<Event>)> for ClickhouseRequestBuilder {
+    type Metadata = (PartitionKey, EventFinalizers);
     type Events = Vec<Event>;
     type Encoder = (Transformer, Encoder<Framer>);
     type Payload = Bytes;
@@ -94,11 +104,13 @@ impl RequestBuilder<Vec<Event>> for ClickhouseRequestBuilder {
 
     fn split_input(
         &self,
-        mut events: Vec<Event>,
+        input: (PartitionKey, Vec<Event>),
     ) -> (Self::Metadata, RequestMetadataBuilder, Self::Events) {
+        let (key, mut events) = input;
+
         let finalizers = events.take_finalizers();
         let builder = RequestMetadataBuilder::from_events(&events);
-        (finalizers, builder, events)
+        ((key, finalizers), builder, events)
     }
 
     fn build_request(
@@ -107,11 +119,57 @@ impl RequestBuilder<Vec<Event>> for ClickhouseRequestBuilder {
         request_metadata: RequestMetadata,
         payload: EncodeResult<Self::Payload>,
     ) -> Self::Request {
+        let (key, finalizers) = metadata;
         ClickhouseRequest {
+            database: key.database,
+            table: key.table,
             body: payload.into_payload(),
             compression: self.compression,
-            finalizers: metadata,
+            finalizers,
             metadata: request_metadata,
         }
     }
 }
+
+/// PartitionKey used to partition events by (database, table) pair.
+#[derive(Hash, Eq, PartialEq, Clone, Debug)]
+struct PartitionKey {
+    database: String,
+    table: String,
+}
+
+/// KeyPartitioner that partitions events by (database, table) pair.
+struct KeyPartitioner {
+    database: Template,
+    table: Template,
+}
+
+impl KeyPartitioner {
+    const fn new(database: Template, table: Template) -> Self {
+        Self { database, table }
+    }
+
+    fn render(template: &Template, item: &Event, field: &'static str) -> Option<String> {
+        template
+            .render_string(item)
+            .map_err(|error| {
+                emit!(TemplateRenderingError {
+                    error,
+                    field: Some(field),
+                    drop_event: true,
+                });
+            })
+            .ok()
+    }
+}
+
+impl Partitioner for KeyPartitioner {
+    type Item = Event;
+    type Key = Option<PartitionKey>;
+
+    fn partition(&self, item: &Self::Item) -> Self::Key {
+        let database = Self::render(&self.database, item, "database_key")?;
+        let table = Self::render(&self.table, item, "table_key")?;
+        Some(PartitionKey { database, table })
+    }
+}
diff --git a/src/sinks/prelude.rs b/src/sinks/prelude.rs
index 775b530c41002..e3b3adc4327b6 100644
--- a/src/sinks/prelude.rs
+++ b/src/sinks/prelude.rs
@@ -5,7 +5,7 @@ pub use crate::{
     codecs::{Encoder, EncodingConfig, Transformer},
     config::{DataType, GenerateConfig, SinkConfig, SinkContext},
     event::{Event, LogEvent},
-    internal_events::TemplateRenderingError,
+    internal_events::{SinkRequestBuildError, TemplateRenderingError},
     sinks::util::retries::RetryLogic,
     sinks::{
         util::{
diff --git a/vdev/src/testing/integration.rs b/vdev/src/testing/integration.rs
index 9c70543ec4937..8b371853cb37b 100644
--- a/vdev/src/testing/integration.rs
+++ b/vdev/src/testing/integration.rs
@@ -179,7 +179,10 @@ impl Compose {
                 // Inject the networks block
                 config.networks.insert(
                     "default".to_string(),
-                    BTreeMap::from_iter([("name".to_string(), network.clone())]),
+                    BTreeMap::from_iter([
+                        ("name".to_string(), network.clone()),
+                        ("external".to_string(), "true".to_string()),
+                    ]),
                 );
 
                 // Create a named tempfile, there may be resource leakage here in case of SIGINT
diff --git a/website/cue/reference/components/sinks/base/clickhouse.cue b/website/cue/reference/components/sinks/base/clickhouse.cue
index d030debc352aa..6bccc5dfe1968 100644
--- a/website/cue/reference/components/sinks/base/clickhouse.cue
+++ b/website/cue/reference/components/sinks/base/clickhouse.cue
@@ -138,7 +138,10 @@ base: components: sinks: clickhouse: configuration: {
 	database: {
 		description: "The database that contains the table that data is inserted into."
 		required:    false
-		type: string: examples: ["mydatabase"]
+		type: string: {
+			examples: ["mydatabase"]
+			syntax: "template"
+		}
 	}
 	date_time_best_effort: {
 		description: "Sets `date_time_input_format` to `best_effort`, allowing ClickHouse to properly parse RFC3339/ISO 8601."
@@ -325,7 +328,10 @@ base: components: sinks: clickhouse: configuration: {
 	table: {
 		description: "The table that data is inserted into."
 		required:    true
-		type: string: examples: ["mytable"]
+		type: string: {
+			examples: ["mytable"]
+			syntax: "template"
+		}
 	}
 	tls: {
 		description: "TLS configuration."