Repository: ArroyoSystems/arroyo (language: rust)

Issue:
Feature request: Kafka record timestamp column in Kafka Source Connector
Currently the kafka record timestamp is used in Arroyo for default event_time purposes. Need to implement a feature wherein one would be able to access the record timestamp in Arroyo SQL as a column value.
Can reserve a keyword to implement the same.


Discussion / hints:

Fix patch:
diff --git a/crates/arroyo-api/src/pipelines.rs b/crates/arroyo-api/src/pipelines.rs
index 29d501297..52ce19b55 100644
--- a/crates/arroyo-api/src/pipelines.rs
+++ b/crates/arroyo-api/src/pipelines.rs
@@ -159,7 +159,6 @@ async fn compile_sql<'a>(
                     .unwrap_or(json!({})),
                 &table.config,
                 Some(&table.schema),
-                None,
             )
             .map_err(log_and_map)?;
 
diff --git a/crates/arroyo-connectors/src/blackhole/mod.rs b/crates/arroyo-connectors/src/blackhole/mod.rs
index 1b3f522e6..33de9e1d4 100644
--- a/crates/arroyo-connectors/src/blackhole/mod.rs
+++ b/crates/arroyo-connectors/src/blackhole/mod.rs
@@ -1,6 +1,5 @@
 use crate::blackhole::operator::BlackholeSinkFunc;
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;
 use arroyo_rpc::api_types::connections::{
@@ -79,9 +78,8 @@ impl Connector for BlackholeConnector {
         _options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
-        self.from_config(None, name, EmptyConfig {}, EmptyConfig {}, schema, None)
+        self.from_config(None, name, EmptyConfig {}, EmptyConfig {}, schema)
     }
 
     fn from_config(
@@ -91,7 +89,6 @@ impl Connector for BlackholeConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         s: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let description = "Blackhole".to_string();
 
@@ -102,7 +99,7 @@ impl Connector for BlackholeConnector {
             format: None,
             bad_data: None,
             framing: None,
-            additional_fields: None,
+            metadata_fields: vec![],
         };
 
         Ok(Connection {
diff --git a/crates/arroyo-connectors/src/confluent/mod.rs b/crates/arroyo-connectors/src/confluent/mod.rs
index aa991afbd..c694b9ef7 100644
--- a/crates/arroyo-connectors/src/confluent/mod.rs
+++ b/crates/arroyo-connectors/src/confluent/mod.rs
@@ -3,7 +3,6 @@ use crate::kafka::{
 };
 use crate::{kafka, pull_opt};
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;
 use arroyo_rpc::api_types::connections::{
@@ -162,7 +161,6 @@ impl Connector for ConfluentConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let connection = profile
             .map(|p| {
@@ -174,7 +172,7 @@ impl Connector for ConfluentConnector {
 
         let table = KafkaConnector::table_from_options(options)?;
 
-        self.from_config(None, name, connection, table, schema, None)
+        self.from_config(None, name, connection, table, schema)
     }
 
     fn from_config(
@@ -184,12 +182,11 @@ impl Connector for ConfluentConnector {
         config: Self::ProfileT,
         mut table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         table
             .client_configs
             .insert("client.id".to_string(), CLIENT_ID.to_string());
-        KafkaConnector {}.from_config(id, name, config.into(), table, schema, None)
+        KafkaConnector {}.from_config(id, name, config.into(), table, schema)
     }
 
     fn make_operator(
diff --git a/crates/arroyo-connectors/src/filesystem/delta.rs b/crates/arroyo-connectors/src/filesystem/delta.rs
index 2c62434a1..30ebde66b 100644
--- a/crates/arroyo-connectors/src/filesystem/delta.rs
+++ b/crates/arroyo-connectors/src/filesystem/delta.rs
@@ -1,5 +1,4 @@
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_operator::connector::Connection;
 use arroyo_storage::BackendConfig;
 use std::collections::HashMap;
@@ -78,7 +77,6 @@ impl Connector for DeltaLakeConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<arroyo_operator::connector::Connection> {
         let TableType::Sink {
             write_path,
@@ -125,7 +123,7 @@ impl Connector for DeltaLakeConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: schema.metadata_fields(),
         };
 
         Ok(Connection {
@@ -145,11 +143,10 @@ impl Connector for DeltaLakeConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let table = file_system_sink_from_options(options, schema, CommitStyle::DeltaLake)?;
 
-        self.from_config(None, name, EmptyConfig {}, table, schema, None)
+        self.from_config(None, name, EmptyConfig {}, table, schema)
     }
 
     fn make_operator(
diff --git a/crates/arroyo-connectors/src/filesystem/mod.rs b/crates/arroyo-connectors/src/filesystem/mod.rs
index d3b4706bd..0aecefefe 100644
--- a/crates/arroyo-connectors/src/filesystem/mod.rs
+++ b/crates/arroyo-connectors/src/filesystem/mod.rs
@@ -3,7 +3,6 @@ mod sink;
 mod source;
 
 use anyhow::{anyhow, bail, Result};
-use arrow::datatypes::DataType;
 use arroyo_storage::BackendConfig;
 use regex::Regex;
 use std::collections::HashMap;
@@ -115,7 +114,6 @@ impl Connector for FileSystemConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let (description, connection_type) = match table.table_type {
             TableType::Source { .. } => ("FileSystem".to_string(), ConnectionType::Source),
@@ -170,7 +168,7 @@ impl Connector for FileSystemConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: schema.metadata_fields(),
         };
 
         Ok(Connection {
@@ -190,7 +188,6 @@ impl Connector for FileSystemConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         match options.remove("type") {
             Some(t) if t == "source" => {
@@ -214,13 +211,12 @@ impl Connector for FileSystemConnector {
                         },
                     },
                     schema,
-                    None,
                 )
             }
             Some(t) if t == "sink" => {
                 let table = file_system_sink_from_options(options, schema, CommitStyle::Direct)?;
 
-                self.from_config(None, name, EmptyConfig {}, table, schema, None)
+                self.from_config(None, name, EmptyConfig {}, table, schema)
             }
             Some(t) => bail!("unknown type: {}", t),
             None => bail!("must have type set"),
diff --git a/crates/arroyo-connectors/src/fluvio/mod.rs b/crates/arroyo-connectors/src/fluvio/mod.rs
index 98f9aa4bb..9ff76464a 100644
--- a/crates/arroyo-connectors/src/fluvio/mod.rs
+++ b/crates/arroyo-connectors/src/fluvio/mod.rs
@@ -1,5 +1,4 @@
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_formats::ser::ArrowSerializer;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;
@@ -89,7 +88,6 @@ impl Connector for FluvioConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let endpoint = options.remove("endpoint");
         let topic = pull_opt("topic", options)?;
@@ -118,7 +116,7 @@ impl Connector for FluvioConnector {
             type_: table_type,
         };
 
-        Self::from_config(self, None, name, EmptyConfig {}, table, schema, None)
+        Self::from_config(self, None, name, EmptyConfig {}, table, schema)
     }
 
     fn from_config(
@@ -128,7 +126,6 @@ impl Connector for FluvioConnector {
         config: EmptyConfig,
         table: FluvioTable,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let (typ, desc) = match table.type_ {
             TableType::Source { .. } => (
@@ -157,7 +154,7 @@ impl Connector for FluvioConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: schema.metadata_fields(),
         };
 
         Ok(Connection {
diff --git a/crates/arroyo-connectors/src/impulse/mod.rs b/crates/arroyo-connectors/src/impulse/mod.rs
index c5382a369..a5e46981a 100644
--- a/crates/arroyo-connectors/src/impulse/mod.rs
+++ b/crates/arroyo-connectors/src/impulse/mod.rs
@@ -1,7 +1,6 @@
 mod operator;
 
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;
 use arroyo_rpc::api_types::connections::FieldType::Primitive;
@@ -102,7 +101,6 @@ impl Connector for ImpulseConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let event_rate = f64::from_str(&pull_opt("event_rate", options)?)
             .map_err(|_| anyhow!("invalid value for event_rate; expected float"))?;
@@ -136,7 +134,6 @@ impl Connector for ImpulseConnector {
                 message_count,
             },
             None,
-            None,
         )
     }
 
@@ -147,7 +144,6 @@ impl Connector for ImpulseConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         _: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let description = format!(
             "{}Impulse<{} eps{}>",
@@ -170,7 +166,7 @@ impl Connector for ImpulseConnector {
             format: None,
             bad_data: None,
             framing: None,
-            additional_fields: None,
+            metadata_fields: vec![],
         };
 
         Ok(Connection {
diff --git a/crates/arroyo-connectors/src/kafka/mod.rs b/crates/arroyo-connectors/src/kafka/mod.rs
index b3faf20a2..f03c97c10 100644
--- a/crates/arroyo-connectors/src/kafka/mod.rs
+++ b/crates/arroyo-connectors/src/kafka/mod.rs
@@ -2,7 +2,7 @@ use anyhow::{anyhow, bail};
 use arrow::datatypes::DataType;
 use arroyo_formats::de::ArrowDeserializer;
 use arroyo_formats::ser::ArrowSerializer;
-use arroyo_operator::connector::Connection;
+use arroyo_operator::connector::{Connection, MetadataDef};
 use arroyo_rpc::api_types::connections::{ConnectionProfile, ConnectionSchema, TestSourceMessage};
 use arroyo_rpc::df::ArroyoSchema;
 use arroyo_rpc::formats::{BadData, Format, JsonFormat};
@@ -189,7 +189,6 @@ impl Connector for KafkaConnector {
         config: KafkaConfig,
         table: KafkaTable,
         schema: Option<&ConnectionSchema>,
-        metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let (typ, desc) = match table.type_ {
             TableType::Source { .. } => (
@@ -209,13 +208,6 @@ impl Connector for KafkaConnector {
             .map(|t| t.to_owned())
             .ok_or_else(|| anyhow!("'format' must be set for Kafka connection"))?;
 
-        let metadata_fields = metadata_fields.map(|fields| {
-            fields
-                .into_iter()
-                .map(|(k, (v, _))| (k, v))
-                .collect::<HashMap<String, String>>()
-        });
-
         let config = OperatorConfig {
             connection: serde_json::to_value(config).unwrap(),
             table: serde_json::to_value(table).unwrap(),
@@ -223,7 +215,7 @@ impl Connector for KafkaConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: metadata_fields,
+            metadata_fields: schema.metadata_fields(),
         };
 
         Ok(Connection {
@@ -316,13 +308,33 @@ impl Connector for KafkaConnector {
         }
     }
 
+    fn metadata_defs(&self) -> &'static [MetadataDef] {
+        &[
+            MetadataDef {
+                name: "offset_id",
+                data_type: DataType::Int64,
+            },
+            MetadataDef {
+                name: "partition",
+                data_type: DataType::Int32,
+            },
+            MetadataDef {
+                name: "topic",
+                data_type: DataType::Utf8,
+            },
+            MetadataDef {
+                name: "timestamp",
+                data_type: DataType::Int64,
+            },
+        ]
+    }
+
     fn from_options(
         &self,
         name: &str,
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         profile: Option<&ConnectionProfile>,
-        metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let connection = profile
             .map(|p| {
@@ -334,37 +346,7 @@ impl Connector for KafkaConnector {
 
         let table = Self::table_from_options(options)?;
 
-        let allowed_metadata_udf_args: HashMap<&str, DataType> = [
-            ("offset_id", DataType::Int64),
-            ("partition", DataType::Int32),
-            ("topic", DataType::Utf8),
-        ]
-        .iter()
-        .cloned()
-        .collect();
-
-        if let Some(fields) = &metadata_fields {
-            for (field_name, data_type) in fields.values() {
-                match allowed_metadata_udf_args.get(field_name.as_str()) {
-                    Some(expected_type) => {
-                        if expected_type != data_type {
-                            return Err(anyhow!(
-                                "Invalid datatype for metadata field '{}': expected '{:?}', found '{:?}'",
-                                field_name, expected_type, data_type
-                            ));
-                        }
-                    }
-                    None => {
-                        return Err(anyhow!(
-                            "Invalid metadata field name for Kafka connector: '{}'",
-                            field_name
-                        ));
-                    }
-                }
-            }
-        }
-
-        Self::from_config(self, None, name, connection, table, schema, metadata_fields)
+        Self::from_config(self, None, name, connection, table, schema)
     }
 
     fn make_operator(
@@ -424,7 +406,7 @@ impl Connector for KafkaConnector {
                             .unwrap_or(u32::MAX),
                     )
                     .unwrap(),
-                    metadata_fields: config.additional_fields,
+                    metadata_fields: config.metadata_fields,
                 })))
             }
             TableType::Sink {
diff --git a/crates/arroyo-connectors/src/kafka/source/mod.rs b/crates/arroyo-connectors/src/kafka/source/mod.rs
index f95c4a479..89fa297d0 100644
--- a/crates/arroyo-connectors/src/kafka/source/mod.rs
+++ b/crates/arroyo-connectors/src/kafka/source/mod.rs
@@ -2,7 +2,7 @@ use arroyo_formats::de::FieldValueType;
 use arroyo_rpc::formats::{BadData, Format, Framing};
 use arroyo_rpc::grpc::rpc::TableConfig;
 use arroyo_rpc::schema_resolver::SchemaResolver;
-use arroyo_rpc::{grpc::rpc::StopMode, ControlMessage, ControlResp};
+use arroyo_rpc::{grpc::rpc::StopMode, ControlMessage, ControlResp, MetadataField};
 
 use arroyo_operator::context::ArrowContext;
 use arroyo_operator::operator::SourceOperator;
@@ -36,7 +36,7 @@ pub struct KafkaSourceFunc {
     pub schema_resolver: Option<Arc<dyn SchemaResolver + Sync>>,
     pub client_configs: HashMap<String, String>,
     pub messages_per_second: NonZeroU32,
-    pub metadata_fields: Option<HashMap<String, String>>,
+    pub metadata_fields: Vec<MetadataField>,
 }
 
 #[derive(Copy, Clone, Debug, Encode, Decode, PartialEq, PartialOrd)]
@@ -180,29 +180,25 @@ impl KafkaSourceFunc {
                                     .ok_or_else(|| UserError::new("Failed to read timestamp from Kafka record",
                                         "The message read from Kafka did not contain a message timestamp"))?;
 
-                                let connector_metadata = if let Some(metadata_fields) = &self.metadata_fields {
+                                let topic = msg.topic();
+
+                                let connector_metadata = if !self.metadata_fields.is_empty() {
                                     let mut connector_metadata = HashMap::new();
-                                    for (key, value) in metadata_fields {
-                                        match value.as_str() {
-                                            "offset_id" => {
-                                                connector_metadata.insert(key, FieldValueType::Int64(msg.offset()));
-                                            }
-                                            "partition" => {
-                                                connector_metadata.insert(key, FieldValueType::Int32(msg.partition()));
-                                            }
-                                            "topic" => {
-                                                connector_metadata.insert(key, FieldValueType::String(&self.topic));
-                                            }
-                                            _ => {}
-                                        }
+                                    for f in &self.metadata_fields {
+                                        connector_metadata.insert(&f.field_name, match f.key.as_str() {
+                                            "offset_id" => FieldValueType::Int64(msg.offset()),
+                                            "partition" => FieldValueType::Int32(msg.partition()),
+                                            "topic" => FieldValueType::String(topic),
+                                            "timestamp" => FieldValueType::Int64(timestamp),
+                                            k => unreachable!("Invalid metadata key '{}'", k),
+                                        });
                                     }
                                     Some(connector_metadata)
                                 } else {
                                     None
                                 };
 
-
-                                ctx.deserialize_slice(v, from_millis(timestamp as u64), connector_metadata).await?;
+                                ctx.deserialize_slice(v, from_millis(timestamp.max(0) as u64), connector_metadata.as_ref()).await?;
 
 
                                 if ctx.should_flush() {
diff --git a/crates/arroyo-connectors/src/kinesis/mod.rs b/crates/arroyo-connectors/src/kinesis/mod.rs
index 4014a7a75..1ece9279f 100644
--- a/crates/arroyo-connectors/src/kinesis/mod.rs
+++ b/crates/arroyo-connectors/src/kinesis/mod.rs
@@ -1,5 +1,4 @@
 use anyhow::{anyhow, bail, Result};
-use arrow::datatypes::DataType;
 use std::collections::HashMap;
 use typify::import_types;
 
@@ -84,7 +83,6 @@ impl Connector for KinesisConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<arroyo_operator::connector::Connection> {
         let (connection_type, description) = match table.type_ {
             TableType::Source { .. } => (
@@ -113,7 +111,7 @@ impl Connector for KinesisConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: schema.metadata_fields(),
         };
 
         Ok(Connection {
@@ -133,7 +131,6 @@ impl Connector for KinesisConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let typ = pull_opt("type", options)?;
         let table_type = match typ.as_str() {
@@ -170,7 +167,7 @@ impl Connector for KinesisConnector {
             aws_region: options.remove("aws_region").map(|s| s.to_string()),
         };
 
-        Self::from_config(self, None, name, EmptyConfig {}, table, schema, None)
+        Self::from_config(self, None, name, EmptyConfig {}, table, schema)
     }
 
     fn make_operator(
diff --git a/crates/arroyo-connectors/src/lib.rs b/crates/arroyo-connectors/src/lib.rs
index 2adb17ca0..6765a06d3 100644
--- a/crates/arroyo-connectors/src/lib.rs
+++ b/crates/arroyo-connectors/src/lib.rs
@@ -138,6 +138,7 @@ pub(crate) fn source_field(name: &str, field_type: FieldType) -> SourceField {
             r#type: field_type,
         },
         nullable: false,
+        metadata_key: None,
     }
 }
 
diff --git a/crates/arroyo-connectors/src/mqtt/mod.rs b/crates/arroyo-connectors/src/mqtt/mod.rs
index ff604d51c..d642b7c5f 100644
--- a/crates/arroyo-connectors/src/mqtt/mod.rs
+++ b/crates/arroyo-connectors/src/mqtt/mod.rs
@@ -10,7 +10,7 @@ use crate::pull_opt;
 use anyhow::{anyhow, bail};
 use arrow::datatypes::DataType;
 use arroyo_formats::ser::ArrowSerializer;
-use arroyo_operator::connector::{Connection, Connector};
+use arroyo_operator::connector::{Connection, Connector, MetadataDef};
 use arroyo_operator::operator::OperatorNode;
 use arroyo_rpc::api_types::connections::{
     ConnectionProfile, ConnectionSchema, ConnectionType, TestSourceMessage,
@@ -158,7 +158,6 @@ impl Connector for MqttConnector {
         config: MqttConfig,
         table: MqttTable,
         schema: Option<&ConnectionSchema>,
-        metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let (typ, desc) = match table.type_ {
             TableType::Source { .. } => (
@@ -178,13 +177,6 @@ impl Connector for MqttConnector {
             .map(|t| t.to_owned())
             .ok_or_else(|| anyhow!("'format' must be set for Mqtt connection"))?;
 
-        let metadata_fields = metadata_fields.map(|fields| {
-            fields
-                .into_iter()
-                .map(|(k, (v, _))| (k, v))
-                .collect::<HashMap<String, String>>()
-        });
-
         let config = OperatorConfig {
             connection: serde_json::to_value(config).unwrap(),
             table: serde_json::to_value(table).unwrap(),
@@ -192,7 +184,7 @@ impl Connector for MqttConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: metadata_fields,
+            metadata_fields: schema.metadata_fields(),
         };
 
         Ok(Connection {
@@ -247,13 +239,19 @@ impl Connector for MqttConnector {
         }
     }
 
+    fn metadata_defs(&self) -> &'static [MetadataDef] {
+        &[MetadataDef {
+            name: "topic",
+            data_type: DataType::Utf8,
+        }]
+    }
+
     fn from_options(
         &self,
         name: &str,
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         profile: Option<&ConnectionProfile>,
-        metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let connection = profile
             .map(|p| {
@@ -265,25 +263,7 @@ impl Connector for MqttConnector {
 
         let table = Self::table_from_options(options)?;
 
-        if let Some(fields) = &metadata_fields {
-            for (k, (v, t)) in fields {
-                if v != "topic" {
-                    return Err(anyhow!(
-                        "Invalid metadata field name for mqtt connector: {}",
-                        k
-                    ));
-                }
-                if *t != DataType::Utf8 {
-                    return Err(anyhow!(
-                        "Invalid datatype: {} for metadata field: {} for mqtt connector",
-                        k,
-                        v
-                    ));
-                }
-            }
-        }
-
-        Self::from_config(self, None, name, connection, table, schema, metadata_fields)
+        Self::from_config(self, None, name, connection, table, schema)
     }
 
     fn make_operator(
@@ -311,7 +291,7 @@ impl Connector for MqttConnector {
                 )
                 .unwrap(),
                 subscribed: Arc::new(AtomicBool::new(false)),
-                metadata_fields: config.additional_fields,
+                metadata_fields: config.metadata_fields,
             })),
             TableType::Sink { retain } => OperatorNode::from_operator(Box::new(MqttSinkFunc {
                 config: profile,
diff --git a/crates/arroyo-connectors/src/mqtt/source/mod.rs b/crates/arroyo-connectors/src/mqtt/source/mod.rs
index 3ef88bee4..6f8f9f19d 100644
--- a/crates/arroyo-connectors/src/mqtt/source/mod.rs
+++ b/crates/arroyo-connectors/src/mqtt/source/mod.rs
@@ -7,7 +7,7 @@ use std::sync::Arc;
 use std::time::{Duration, SystemTime};
 
 use arroyo_rpc::formats::{BadData, Format, Framing};
-use arroyo_rpc::{grpc::rpc::StopMode, ControlMessage, ControlResp};
+use arroyo_rpc::{grpc::rpc::StopMode, ControlMessage, ControlResp, MetadataField};
 use arroyo_types::{ArrowMessage, SignalMessage, UserError, Watermark};
 use governor::{Quota, RateLimiter as GovernorRateLimiter};
 use rumqttc::v5::mqttbytes::QoS;
@@ -34,7 +34,7 @@ pub struct MqttSourceFunc {
     pub bad_data: Option<BadData>,
     pub messages_per_second: NonZeroU32,
     pub subscribed: Arc<AtomicBool>,
-    pub metadata_fields: Option<HashMap<String, String>>,
+    pub metadata_fields: Vec<MetadataField>,
 }
 
 #[async_trait]
@@ -77,7 +77,7 @@ impl MqttSourceFunc {
         framing: Option<Framing>,
         bad_data: Option<BadData>,
         messages_per_second: u32,
-        metadata_fields: Option<HashMap<String, String>>,
+        metadata_fields: Vec<MetadataField>,
     ) -> Self {
         Self {
             config,
@@ -149,19 +149,21 @@ impl MqttSourceFunc {
                     match event {
                         Ok(MqttEvent::Incoming(Incoming::Publish(p))) => {
                             let topic = String::from_utf8_lossy(&p.topic).to_string();
-                            let connector_metadata: Option<HashMap<&String, FieldValueType<'_>>> =
-                            self.metadata_fields.as_ref().map(|fields| {
-                                fields.iter()
-                                    .filter_map(|(k, v)| {
-                                        if v == "topic" {
-                                            Some((k, FieldValueType::String(&topic)))
-                                        } else {
-                                            None
-                                        }
-                                    })
-                                    .collect()
-                            });
-                            ctx.deserialize_slice(&p.payload, SystemTime::now(), connector_metadata).await?;
+
+                            let connector_metadata = if !self.metadata_fields.is_empty() {
+                                let mut connector_metadata = HashMap::new();
+                                for mf in &self.metadata_fields {
+                                    connector_metadata.insert(&mf.field_name, match mf.key.as_str() {
+                                        "topic" => FieldValueType::String(&topic),
+                                        k => unreachable!("invalid metadata key '{}' for mqtt", k)
+                                    });
+                                }
+                                Some(connector_metadata)
+                            } else {
+                                None
+                            };
+
+                            ctx.deserialize_slice(&p.payload, SystemTime::now(), connector_metadata.as_ref()).await?;
                             rate_limiter.until_ready().await;
                         }
                         Ok(MqttEvent::Outgoing(Outgoing::Subscribe(_))) => {
diff --git a/crates/arroyo-connectors/src/nats/mod.rs b/crates/arroyo-connectors/src/nats/mod.rs
index 095d8fce5..6b7a98fb7 100644
--- a/crates/arroyo-connectors/src/nats/mod.rs
+++ b/crates/arroyo-connectors/src/nats/mod.rs
@@ -3,7 +3,6 @@ use crate::nats::source::NatsSourceFunc;
 use crate::pull_opt;
 use anyhow::anyhow;
 use anyhow::bail;
-use arrow::datatypes::DataType;
 use arroyo_formats::ser::ArrowSerializer;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;
@@ -247,7 +246,6 @@ impl Connector for NatsConnector {
         config: NatsConfig,
         table: NatsTable,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let stream_or_subject = match &table.connector_type {
             ConnectorType::Source { source_type, .. } => {
@@ -297,7 +295,7 @@ impl Connector for NatsConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: schema.metadata_fields(),
         };
 
         Ok(Connection {
@@ -317,7 +315,6 @@ impl Connector for NatsConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let connection = profile
             .map(|p| {
@@ -329,7 +326,7 @@ impl Connector for NatsConnector {
 
         let table = Self::table_from_options(options)?;
 
-        Self::from_config(self, None, name, connection, table, schema, None)
+        Self::from_config(self, None, name, connection, table, schema)
     }
 
     fn make_operator(
diff --git a/crates/arroyo-connectors/src/nexmark/mod.rs b/crates/arroyo-connectors/src/nexmark/mod.rs
index 5c110f936..4c9ef358d 100644
--- a/crates/arroyo-connectors/src/nexmark/mod.rs
+++ b/crates/arroyo-connectors/src/nexmark/mod.rs
@@ -3,7 +3,7 @@ mod operator;
 mod test;
 
 use anyhow::{anyhow, bail};
-use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
+use arrow::datatypes::{Field, Schema, TimeUnit};
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;
 use arroyo_rpc::api_types::connections::{
@@ -158,7 +158,6 @@ impl Connector for NexmarkConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let event_rate = f64::from_str(&pull_opt("event_rate", options)?)
             .map_err(|_| anyhow!("invalid value for event_rate; expected float"))?;
@@ -184,7 +183,6 @@ impl Connector for NexmarkConnector {
                 runtime,
             },
             None,
-            None,
         )
     }
 
@@ -195,7 +193,6 @@ impl Connector for NexmarkConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         _: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let description = format!(
             "{}Nexmark<{} eps>",
@@ -214,7 +211,7 @@ impl Connector for NexmarkConnector {
             format: None,
             bad_data: None,
             framing: None,
-            additional_fields: None,
+            metadata_fields: vec![],
         };
 
         Ok(Connection {
diff --git a/crates/arroyo-connectors/src/polling_http/mod.rs b/crates/arroyo-connectors/src/polling_http/mod.rs
index ea8ec421b..7db361e04 100644
--- a/crates/arroyo-connectors/src/polling_http/mod.rs
+++ b/crates/arroyo-connectors/src/polling_http/mod.rs
@@ -5,7 +5,6 @@ use std::str::FromStr;
 use std::time::Duration;
 
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_rpc::{var_str::VarStr, OperatorConfig};
 use arroyo_types::string_to_map;
 use reqwest::{Client, Request};
@@ -153,7 +152,6 @@ impl Connector for PollingHTTPConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let endpoint = pull_opt("endpoint", options)?;
         let headers = options.remove("headers");
@@ -185,7 +183,6 @@ impl Connector for PollingHTTPConnector {
                 emit_behavior,
             },
             schema,
-            None,
         )
     }
 
@@ -196,7 +193,6 @@ impl Connector for PollingHTTPConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<arroyo_operator::connector::Connection> {
         let description = format!("PollingHTTPSource<{}>", table.endpoint);
 
@@ -226,7 +222,7 @@ impl Connector for PollingHTTPConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: vec![],
         };
 
         Ok(Connection {
diff --git a/crates/arroyo-connectors/src/preview/mod.rs b/crates/arroyo-connectors/src/preview/mod.rs
index 72dddd7f0..6b3672af4 100644
--- a/crates/arroyo-connectors/src/preview/mod.rs
+++ b/crates/arroyo-connectors/src/preview/mod.rs
@@ -3,7 +3,6 @@ mod operator;
 use std::collections::HashMap;
 
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_rpc::OperatorConfig;
 
 use arroyo_operator::connector::Connection;
@@ -72,7 +71,6 @@ impl Connector for PreviewConnector {
         _: &mut HashMap<String, String>,
         _: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         bail!("Preview connector cannot be created in SQL");
     }
@@ -84,7 +82,6 @@ impl Connector for PreviewConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let description = "PreviewSink".to_string();
 
@@ -99,7 +96,7 @@ impl Connector for PreviewConnector {
             format: None,
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: schema.metadata_fields(),
         };
 
         Ok(Connection {
diff --git a/crates/arroyo-connectors/src/redis/mod.rs b/crates/arroyo-connectors/src/redis/mod.rs
index 84323762e..cf9d4e045 100644
--- a/crates/arroyo-connectors/src/redis/mod.rs
+++ b/crates/arroyo-connectors/src/redis/mod.rs
@@ -1,7 +1,6 @@
 mod operator;
 
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_formats::ser::ArrowSerializer;
 use arroyo_operator::connector::{Connection, Connector};
 use arroyo_operator::operator::OperatorNode;
@@ -228,7 +227,6 @@ impl Connector for RedisConnector {
         options: &mut HashMap<String, String>,
         s: Option<&ConnectionSchema>,
         profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let connection_config = match profile {
             Some(connection_profile) => {
@@ -350,7 +348,6 @@ impl Connector for RedisConnector {
                 connector_type: sink,
             },
             s,
-            None,
         )
     }
 
@@ -361,7 +358,6 @@ impl Connector for RedisConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let schema = schema
             .map(|s| s.to_owned())
@@ -382,7 +378,7 @@ impl Connector for RedisConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: vec![],
         };
 
         Ok(Connection {
diff --git a/crates/arroyo-connectors/src/single_file/mod.rs b/crates/arroyo-connectors/src/single_file/mod.rs
index 543e5ba13..c5cb0cf89 100644
--- a/crates/arroyo-connectors/src/single_file/mod.rs
+++ b/crates/arroyo-connectors/src/single_file/mod.rs
@@ -1,5 +1,4 @@
 use anyhow::{anyhow, bail, Result};
-use arrow::datatypes::DataType;
 use arroyo_formats::ser::ArrowSerializer;
 use std::collections::HashMap;
 use typify::import_types;
@@ -85,7 +84,6 @@ impl Connector for SingleFileConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<arroyo_operator::connector::Connection> {
         let schema = schema
             .map(|s| s.to_owned())
@@ -105,7 +103,7 @@ impl Connector for SingleFileConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: vec![],
         };
 
         Ok(Connection {
@@ -125,7 +123,6 @@ impl Connector for SingleFileConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let path = pull_opt("path", options)?;
         let Ok(table_type) = pull_opt("type", options)?.try_into() else {
@@ -150,7 +147,6 @@ impl Connector for SingleFileConnector {
                 wait_for_control,
             },
             schema,
-            None,
         )
     }
 
diff --git a/crates/arroyo-connectors/src/sse/mod.rs b/crates/arroyo-connectors/src/sse/mod.rs
index 52b9b3be9..0e64b5583 100644
--- a/crates/arroyo-connectors/src/sse/mod.rs
+++ b/crates/arroyo-connectors/src/sse/mod.rs
@@ -4,7 +4,6 @@ use std::collections::HashMap;
 use std::time::Duration;
 
 use anyhow::{anyhow, bail};
-use arrow::datatypes::DataType;
 use arroyo_rpc::{var_str::VarStr, OperatorConfig};
 use arroyo_types::string_to_map;
 use eventsource_client::Client;
@@ -79,7 +78,6 @@ impl Connector for SSEConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<arroyo_operator::connector::Connection> {
         let description = format!("SSESource<{}>", table.endpoint);
 
@@ -109,7 +107,7 @@ impl Connector for SSEConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: schema.metadata_fields(),
         };
 
         Ok(Connection {
@@ -129,7 +127,6 @@ impl Connector for SSEConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let endpoint = pull_opt("endpoint", options)?;
         let headers = options.remove("headers");
@@ -145,7 +142,6 @@ impl Connector for SSEConnector {
                 headers: headers.map(VarStr::new),
             },
             schema,
-            None,
         )
     }
 
diff --git a/crates/arroyo-connectors/src/stdout/mod.rs b/crates/arroyo-connectors/src/stdout/mod.rs
index 59db510a3..7fbe84f2d 100644
--- a/crates/arroyo-connectors/src/stdout/mod.rs
+++ b/crates/arroyo-connectors/src/stdout/mod.rs
@@ -3,7 +3,6 @@ mod operator;
 use std::collections::HashMap;
 
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_rpc::OperatorConfig;
 use tokio::io::BufWriter;
 
@@ -77,9 +76,8 @@ impl Connector for StdoutConnector {
         _options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
-        self.from_config(None, name, EmptyConfig {}, EmptyConfig {}, schema, None)
+        self.from_config(None, name, EmptyConfig {}, EmptyConfig {}, schema)
     }
 
     fn from_config(
@@ -89,7 +87,6 @@ impl Connector for StdoutConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let description = "StdoutSink".to_string();
 
@@ -110,7 +107,7 @@ impl Connector for StdoutConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: vec![],
         };
 
         Ok(Connection {
diff --git a/crates/arroyo-connectors/src/webhook/mod.rs b/crates/arroyo-connectors/src/webhook/mod.rs
index d5e1a2a29..b8224e392 100644
--- a/crates/arroyo-connectors/src/webhook/mod.rs
+++ b/crates/arroyo-connectors/src/webhook/mod.rs
@@ -5,7 +5,6 @@ use std::sync::Arc;
 use std::time::SystemTime;
 
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_rpc::OperatorConfig;
 
 use arroyo_formats::ser::ArrowSerializer;
@@ -143,7 +142,6 @@ impl Connector for WebhookConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<arroyo_operator::connector::Connection> {
         let description = format!("WebhookSink<{}>", table.endpoint.sub_env_vars()?);
 
@@ -164,7 +162,7 @@ impl Connector for WebhookConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: schema.metadata_fields(),
         };
 
         Ok(Connection {
@@ -184,7 +182,6 @@ impl Connector for WebhookConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let endpoint = pull_opt("endpoint", options)?;
 
@@ -205,7 +202,7 @@ impl Connector for WebhookConnector {
         )?;
         let _ = Self::construct_test_request(&client, &table)?;
 
-        self.from_config(None, name, EmptyConfig {}, table, schema, None)
+        self.from_config(None, name, EmptyConfig {}, table, schema)
     }
 
     fn make_operator(
diff --git a/crates/arroyo-connectors/src/websocket/mod.rs b/crates/arroyo-connectors/src/websocket/mod.rs
index 36fefca1f..9bf8cc967 100644
--- a/crates/arroyo-connectors/src/websocket/mod.rs
+++ b/crates/arroyo-connectors/src/websocket/mod.rs
@@ -3,7 +3,6 @@ use std::str::FromStr;
 use std::time::Duration;
 
 use anyhow::anyhow;
-use arrow::datatypes::DataType;
 use arroyo_operator::connector::Connection;
 use arroyo_rpc::api_types::connections::{
     ConnectionProfile, ConnectionSchema, ConnectionType, TestSourceMessage,
@@ -221,7 +220,6 @@ impl Connector for WebsocketConnector {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<arroyo_operator::connector::Connection> {
         let description = format!("WebsocketSource<{}>", table.endpoint);
 
@@ -251,7 +249,7 @@ impl Connector for WebsocketConnector {
             format: Some(format),
             bad_data: schema.bad_data.clone(),
             framing: schema.framing.clone(),
-            additional_fields: None,
+            metadata_fields: schema.metadata_fields(),
         };
 
         Ok(Connection {
@@ -271,7 +269,6 @@ impl Connector for WebsocketConnector {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         _profile: Option<&ConnectionProfile>,
-        _metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         let endpoint = pull_opt("endpoint", options)?;
         let headers = options.remove("headers");
@@ -308,7 +305,6 @@ impl Connector for WebsocketConnector {
                 subscription_messages,
             },
             schema,
-            None,
         )
     }
 
diff --git a/crates/arroyo-formats/src/de.rs b/crates/arroyo-formats/src/de.rs
index 0198b1193..3938fe0fc 100644
--- a/crates/arroyo-formats/src/de.rs
+++ b/crates/arroyo-formats/src/de.rs
@@ -25,7 +25,7 @@ use tokio::sync::Mutex;
 pub enum FieldValueType<'a> {
     Int64(i64),
     Int32(i32),
-    String(&'a String),
+    String(&'a str),
     // Extend with more types as needed
 }
 
@@ -177,12 +177,12 @@ impl ArrowDeserializer {
         buffer: &mut [Box<dyn ArrayBuilder>],
         msg: &[u8],
         timestamp: SystemTime,
-        additional_fields: Option<HashMap<&String, FieldValueType<'_>>>,
+        additional_fields: Option<&HashMap<&String, FieldValueType<'_>>>,
     ) -> Vec<SourceError> {
         match &*self.format {
             Format::Avro(_) => self.deserialize_slice_avro(buffer, msg, timestamp).await,
             _ => FramingIterator::new(self.framing.clone(), msg)
-                .map(|t| self.deserialize_single(buffer, t, timestamp, additional_fields.clone()))
+                .map(|t| self.deserialize_single(buffer, t, timestamp, additional_fields))
                 .filter_map(|t| t.err())
                 .collect(),
         }
@@ -247,7 +247,7 @@ impl ArrowDeserializer {
         buffer: &mut [Box<dyn ArrayBuilder>],
         msg: &[u8],
         timestamp: SystemTime,
-        additional_fields: Option<HashMap<&String, FieldValueType>>,
+        additional_fields: Option<&HashMap<&String, FieldValueType>>,
     ) -> Result<(), SourceError> {
         match &*self.format {
             Format::RawString(_)
@@ -500,7 +500,7 @@ pub(crate) fn add_additional_fields(
 }
 
 pub(crate) fn add_additional_fields_using_builder(
-    additional_fields: Option<HashMap<&String, FieldValueType<'_>>>,
+    additional_fields: Option<&HashMap<&String, FieldValueType<'_>>>,
     additional_fields_builder: &mut Option<HashMap<String, Box<dyn ArrayBuilder>>>,
 ) {
     if let Some(fields) = additional_fields {
@@ -856,7 +856,7 @@ mod tests {
                 &mut arrays,
                 json!({ "x": 5 }).to_string().as_bytes(),
                 time,
-                Some(additional_fields),
+                Some(&additional_fields),
             )
             .await;
         assert!(result.is_empty());
diff --git a/crates/arroyo-operator/src/connector.rs b/crates/arroyo-operator/src/connector.rs
index 8606de728..32745309b 100644
--- a/crates/arroyo-operator/src/connector.rs
+++ b/crates/arroyo-operator/src/connector.rs
@@ -1,10 +1,11 @@
 use crate::operator::OperatorNode;
-use anyhow::anyhow;
-use arrow::datatypes::DataType;
+use anyhow::{anyhow, bail};
+use arrow::datatypes::{DataType, Field};
 use arroyo_rpc::api_types::connections::{
     ConnectionProfile, ConnectionSchema, ConnectionType, TestSourceMessage,
 };
 use arroyo_rpc::OperatorConfig;
+use arroyo_types::DisplayAsSql;
 use serde::de::DeserializeOwned;
 use serde::ser::Serialize;
 use serde_json::value::Value;
@@ -23,6 +24,11 @@ pub struct Connection {
     pub description: String,
 }
 
+pub struct MetadataDef {
+    pub name: &'static str,
+    pub data_type: DataType,
+}
+
 #[allow(clippy::wrong_self_convention)]
 pub trait Connector: Send {
     type ProfileT: DeserializeOwned + Serialize;
@@ -45,6 +51,10 @@ pub trait Connector: Send {
 
     fn metadata(&self) -> arroyo_rpc::api_types::connections::Connector;
 
+    fn metadata_defs(&self) -> &'static [MetadataDef] {
+        &[]
+    }
+
     fn table_type(&self, config: Self::ProfileT, table: Self::TableT) -> ConnectionType;
 
     #[allow(unused)]
@@ -90,7 +100,6 @@ pub trait Connector: Send {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         profile: Option<&ConnectionProfile>,
-        metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection>;
 
     fn from_config(
@@ -100,7 +109,6 @@ pub trait Connector: Send {
         config: Self::ProfileT,
         table: Self::TableT,
         schema: Option<&ConnectionSchema>,
-        metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection>;
 
     #[allow(unused)]
@@ -118,6 +126,8 @@ pub trait ErasedConnector: Send {
 
     fn metadata(&self) -> arroyo_rpc::api_types::connections::Connector;
 
+    fn metadata_defs(&self) -> &'static [MetadataDef];
+
     fn validate_config(&self, s: &serde_json::Value) -> Result<(), serde_json::Error>;
 
     fn validate_table(&self, s: &serde_json::Value) -> Result<(), serde_json::Error>;
@@ -165,7 +175,6 @@ pub trait ErasedConnector: Send {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         profile: Option<&ConnectionProfile>,
-        metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection>;
 
     fn from_config(
@@ -175,7 +184,6 @@ pub trait ErasedConnector: Send {
         config: &serde_json::Value,
         table: &serde_json::Value,
         schema: Option<&ConnectionSchema>,
-        metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection>;
 
     fn make_operator(&self, config: OperatorConfig) -> anyhow::Result<OperatorNode>;
@@ -190,6 +198,10 @@ impl<C: Connector> ErasedConnector for C {
         self.metadata()
     }
 
+    fn metadata_defs(&self) -> &'static [MetadataDef] {
+        self.metadata_defs()
+    }
+
     fn config_description(&self, s: &serde_json::Value) -> Result<String, serde_json::Error> {
         Ok(self.config_description(self.parse_config(s)?))
     }
@@ -261,9 +273,34 @@ impl<C: Connector> ErasedConnector for C {
         options: &mut HashMap<String, String>,
         schema: Option<&ConnectionSchema>,
         profile: Option<&ConnectionProfile>,
-        metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
-        self.from_options(name, options, schema, profile, metadata_fields)
+        if let Some(schema) = schema {
+            for sf in schema.fields.iter() {
+                if let Some(key) = &sf.metadata_key {
+                    let field = self
+                        .metadata_defs()
+                        .iter()
+                        .find(|f| f.name == key)
+                        .ok_or_else(|| {
+                            anyhow!(
+                                "unknown metadata field '{}' for {} connector '{}'",
+                                key,
+                                self.name(),
+                                name
+                            )
+                        })?;
+
+                    let arrow_field: Field = sf.clone().into();
+
+                    if !field.data_type.equals_datatype(arrow_field.data_type()) {
+                        bail!("incorrect data type for metadata field '{}'; expected {}, but found {}",
+                        arrow_field.name(), DisplayAsSql(&field.data_type), DisplayAsSql(arrow_field.data_type()));
+                    }
+                }
+            }
+        }
+
+        self.from_options(name, options, schema, profile)
     }
 
     fn from_config(
@@ -273,7 +310,6 @@ impl<C: Connector> ErasedConnector for C {
         config: &serde_json::Value,
         table: &serde_json::Value,
         schema: Option<&ConnectionSchema>,
-        metadata_fields: Option<HashMap<String, (String, DataType)>>,
     ) -> anyhow::Result<Connection> {
         self.from_config(
             id,
@@ -281,7 +317,6 @@ impl<C: Connector> ErasedConnector for C {
             self.parse_config(config)?,
             self.parse_table(table)?,
             schema,
-            metadata_fields,
         )
     }
 
diff --git a/crates/arroyo-operator/src/context.rs b/crates/arroyo-operator/src/context.rs
index cf0534e32..33268923f 100644
--- a/crates/arroyo-operator/src/context.rs
+++ b/crates/arroyo-operator/src/context.rs
@@ -670,7 +670,7 @@ impl ArrowContext {
         &mut self,
         msg: &[u8],
         time: SystemTime,
-        additional_fields: Option<HashMap<&String, FieldValueType<'_>>>,
+        additional_fields: Option<&HashMap<&String, FieldValueType<'_>>>,
     ) -> Result<(), UserError> {
         let deserializer = self
             .deserializer
diff --git a/crates/arroyo-planner/src/extension/table_source.rs b/crates/arroyo-planner/src/extension/table_source.rs
index 853618c2a..1fd582547 100644
--- a/crates/arroyo-planner/src/extension/table_source.rs
+++ b/crates/arroyo-planner/src/extension/table_source.rs
@@ -8,14 +8,14 @@ use datafusion::logical_expr::{Expr, LogicalPlan, UserDefinedLogicalNodeCore};
 
 use prost::Message;
 
+use super::{ArroyoExtension, DebeziumUnrollingExtension, NodeWithIncomingEdges};
+use crate::tables::FieldSpec;
 use crate::{
     builder::{NamedNode, Planner},
     schema_from_df_fields,
     schemas::add_timestamp_field,
     tables::ConnectorTable,
 };
-
-use super::{ArroyoExtension, DebeziumUnrollingExtension, NodeWithIncomingEdges};
 pub(crate) const TABLE_SOURCE_NAME: &str = "TableSourceExtension";
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -31,10 +31,10 @@ impl TableSourceExtension {
             .fields
             .iter()
             .filter_map(|field| match field {
-                crate::tables::FieldSpec::StructField(field) => {
+                FieldSpec::Struct(field) | FieldSpec::Metadata { field, .. } => {
                     Some((Some(name.clone()), Arc::new(field.clone())).into())
                 }
-                crate::tables::FieldSpec::VirtualField { .. } => None,
+                FieldSpec::Virtual { .. } => None,
             })
             .collect::<Vec<_>>();
         let base_schema = Arc::new(schema_from_df_fields(&physical_fields).unwrap());
diff --git a/crates/arroyo-planner/src/rewriters.rs b/crates/arroyo-planner/src/rewriters.rs
index 39e7a1877..fd3b47cf6 100644
--- a/crates/arroyo-planner/src/rewriters.rs
+++ b/crates/arroyo-planner/src/rewriters.rs
@@ -49,11 +49,13 @@ impl<'a> SourceRewriter<'a> {
                 .find_map(|f| {
                     if f.field().name() == &watermark_field {
                         return match f {
-                            FieldSpec::StructField(f) => Some(Expr::Column(Column {
-                                relation: None,
-                                name: f.name().to_string(),
-                            })),
-                            FieldSpec::VirtualField { expression, .. } => Some(expression.clone()),
+                            FieldSpec::Struct(field) | FieldSpec::Metadata { field, .. } => {
+                                Some(Expr::Column(Column {
+                                    relation: None,
+                                    name: field.name().to_string(),
+                                }))
+                            }
+                            FieldSpec::Virtual { expression, .. } => Some(expression.clone()),
                         };
                     }
                     None
@@ -84,11 +86,13 @@ impl<'a> SourceRewriter<'a> {
             .fields
             .iter()
             .map(|field| match field {
-                FieldSpec::StructField(f) => Expr::Column(Column {
-                    relation: Some(qualifier.clone()),
-                    name: f.name().to_string(),
-                }),
-                FieldSpec::VirtualField { field, expression } => expression
+                FieldSpec::Struct(field) | FieldSpec::Metadata { field, .. } => {
+                    Expr::Column(Column {
+                        relation: Some(qualifier.clone()),
+                        name: field.name().to_string(),
+                    })
+                }
+                FieldSpec::Virtual { field, expression } => expression
                     .clone()
                     .alias_qualified(Some(qualifier.clone()), field.name().to_string()),
             })
@@ -106,11 +110,13 @@ impl<'a> SourceRewriter<'a> {
                 .find_map(|f| {
                     if f.field().name() == &event_time_field {
                         return match f {
-                            FieldSpec::StructField(f) => Some(Expr::Column(Column {
-                                relation: Some(qualifier.clone()),
-                                name: f.name().to_string(),
-                            })),
-                            FieldSpec::VirtualField { expression, .. } => Some(expression.clone()),
+                            FieldSpec::Struct(field) | FieldSpec::Metadata { field, .. } => {
+                                Some(Expr::Column(Column {
+                                    relation: Some(qualifier.clone()),
+                                    name: field.name().to_string(),
+                                }))
+                            }
+                            FieldSpec::Virtual { expression, .. } => Some(expression.clone()),
                         };
                     }
                     None
diff --git a/crates/arroyo-planner/src/tables.rs b/crates/arroyo-planner/src/tables.rs
index b13bcb14a..da5ae1391 100644
--- a/crates/arroyo-planner/src/tables.rs
+++ b/crates/arroyo-planner/src/tables.rs
@@ -4,7 +4,6 @@ use std::{collections::HashMap, time::Duration};
 
 use arrow_schema::{DataType, Field, FieldRef, Schema};
 use arroyo_connectors::connector_for_type;
-use datafusion::logical_expr::expr::ScalarFunction;
 
 use crate::extension::remote_table::RemoteTableExtension;
 use crate::types::convert_data_type;
@@ -21,7 +20,8 @@ use arroyo_rpc::api_types::connections::{
 use arroyo_rpc::formats::{BadData, Format, Framing, JsonFormat};
 use arroyo_rpc::grpc::api::ConnectorOp;
 use arroyo_types::ArroyoExtensionType;
-use datafusion::common::{config::ConfigOptions, DFSchema, Result};
+use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion, TreeNodeVisitor};
+use datafusion::common::{config::ConfigOptions, DFSchema, Result, ScalarValue};
 use datafusion::common::{plan_err, Column, DataFusionError};
 use datafusion::execution::context::SessionState;
 use datafusion::execution::FunctionRegistry;
@@ -53,7 +53,7 @@ use datafusion::optimizer::unwrap_cast_in_comparison::UnwrapCastInComparison;
 use datafusion::optimizer::OptimizerRule;
 use datafusion::sql::planner::PlannerContext;
 use datafusion::sql::sqlparser;
-use datafusion::sql::sqlparser::ast::{FunctionArg, FunctionArguments, Query};
+use datafusion::sql::sqlparser::ast::Query;
 use datafusion::{
     optimizer::{optimizer::Optimizer, OptimizerContext},
     sql::{
@@ -82,41 +82,37 @@ pub struct ConnectorTable {
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub enum FieldSpec {
-    StructField(Field),
-    VirtualField { field: Field, expression: Expr },
+    Struct(Field),
+    Metadata { field: Field, key: String },
+    Virtual { field: Field, expression: Expr },
 }
 
 impl FieldSpec {
     fn is_virtual(&self) -> bool {
         match self {
-            FieldSpec::StructField(_) => false,
-            FieldSpec::VirtualField { .. } => true,
+            FieldSpec::Struct(_) | FieldSpec::Metadata { .. } => false,
+            FieldSpec::Virtual { .. } => true,
         }
     }
     pub fn field(&self) -> &Field {
         match self {
-            FieldSpec::StructField(f) => f,
-            FieldSpec::VirtualField { field, .. } => field,
+            FieldSpec::Struct(f) => f,
+            FieldSpec::Metadata { field, .. } => field,
+            FieldSpec::Virtual { field, .. } => field,
         }
     }
-    fn is_metadata_virtual(&self) -> bool {
-        match self {
-            FieldSpec::VirtualField {
-                expression: Expr::ScalarFunction(ScalarFunction { func, args, .. }),
-                ..
-            } => {
-                func.name() == "metadata"
-                    && args.len() == 1
-                    && matches!(args.first(), Some(Expr::Literal(_)))
-            }
-            _ => false,
+
+    fn metadata_key(&self) -> Option<&str> {
+        match &self {
+            FieldSpec::Metadata { key, .. } => Some(key.as_str()),
+            _ => None,
         }
     }
 }
 
 impl From<Field> for FieldSpec {
     fn from(value: Field) -> Self {
-        FieldSpec::StructField(value)
+        FieldSpec::Struct(value)
     }
 }
 
@@ -202,7 +198,7 @@ impl From<Connection> for ConnectorTable {
                 .schema
                 .fields
                 .iter()
-                .map(|f| FieldSpec::StructField(f.clone().into()))
+                .map(|f| FieldSpec::Struct(f.clone().into()))
                 .collect(),
             config: value.config,
             description: value.description,
@@ -224,22 +220,21 @@ impl ConnectorTable {
         primary_keys: Vec<String>,
         options: &mut HashMap<String, String>,
         connection_profile: Option<&ConnectionProfile>,
-        connector_metadata_columns: Option<HashMap<String, (String, DataType)>>,
     ) -> Result<Self> {
         // TODO: a more principled way of letting connectors dictate types to use
         if "delta" == connector {
             fields = fields
                 .into_iter()
                 .map(|field_spec| match &field_spec {
-                    FieldSpec::StructField(struct_field) => match struct_field.data_type() {
+                    FieldSpec::Struct(struct_field) => match struct_field.data_type() {
                         DataType::Timestamp(_, None) => {
-                            FieldSpec::StructField(struct_field.clone().with_data_type(
+                            FieldSpec::Struct(struct_field.clone().with_data_type(
                                 DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, None),
                             ))
                         }
                         _ => field_spec,
                     },
-                    FieldSpec::VirtualField { .. } => {
+                    FieldSpec::Metadata { .. } | FieldSpec::Virtual { .. } => {
                         unreachable!("delta lake is only a sink, can't have virtual fields")
                     }
                 })
@@ -264,25 +259,30 @@ impl ConnectorTable {
             let df_struct_type =
                 DataType::Struct(fields.iter().map(|f| f.field().clone()).collect());
             let before_field_spec =
-                FieldSpec::StructField(Field::new("before", df_struct_type.clone(), true));
-            let after_field_spec =
-                FieldSpec::StructField(Field::new("after", df_struct_type, true));
-            let op_field_spec = FieldSpec::StructField(Field::new("op", DataType::Utf8, false));
+                FieldSpec::Struct(Field::new("before", df_struct_type.clone(), true));
+            let after_field_spec = FieldSpec::Struct(Field::new("after", df_struct_type, true));
+            let op_field_spec = FieldSpec::Struct(Field::new("op", DataType::Utf8, false));
             input_to_schema_fields = vec![before_field_spec, after_field_spec, op_field_spec];
         }
 
         let schema_fields: Vec<SourceField> = input_to_schema_fields
             .iter()
-            .filter(|f| f.is_metadata_virtual() || !f.is_virtual())
+            .filter(|f| !f.is_virtual())
             .map(|f| {
                 let struct_field = f.field();
-                struct_field.clone().try_into().map_err(|_| {
+                let mut sf: SourceField = struct_field.clone().try_into().map_err(|_| {
                     DataFusionError::Plan(format!(
                         "field '{}' has a type '{:?}' that cannot be used in a connection table",
                         struct_field.name(),
                         struct_field.data_type()
                     ))
-                })
+                })?;
+
+                if let Some(key) = f.metadata_key() {
+                    sf.metadata_key = Some(key.to_string());
+                }
+
+                Ok(sf)
             })
             .collect::<Result<_>>()?;
         let bad_data = BadData::from_opts(options)
@@ -300,13 +300,7 @@ impl ConnectorTable {
         .map_err(|e| DataFusionError::Plan(format!("could not create connection schema: {}", e)))?;
 
         let connection = connector
-            .from_options(
-                name,
-                options,
-                Some(&schema),
-                connection_profile,
-                connector_metadata_columns,
-            )
+            .from_options(name, options, Some(&schema), connection_profile)
             .map_err(|e| DataFusionError::Plan(e.to_string()))?;
 
         let mut table: ConnectorTable = connection.into();
@@ -443,10 +437,8 @@ impl ConnectorTable {
             struct_def: self
                 .fields
                 .iter()
-                .filter_map(|field| match field {
-                    FieldSpec::StructField(struct_field) => Some(Arc::new(struct_field.clone())),
-                    FieldSpec::VirtualField { .. } => None,
-                })
+                .filter(|f| !f.is_virtual())
+                .map(|f| Arc::new(f.field().clone()))
                 .collect(),
             config: self.connector_op(),
             processing_mode: self.processing_mode(),
@@ -512,11 +504,50 @@ fn value_to_inner_string(value: &Value) -> Result<String> {
     }
 }
 
+#[derive(Default)]
+struct MetadataFinder {
+    key: Option<String>,
+    depth: usize,
+}
+
+impl<'a> TreeNodeVisitor<'a> for MetadataFinder {
+    type Node = Expr;
+
+    fn f_down(&mut self, node: &'a Self::Node) -> Result<TreeNodeRecursion> {
+        if let Expr::ScalarFunction(func) = node {
+            if func.name() == "metadata" {
+                if self.depth > 0 {
+                    return plan_err!(
+                        "Metadata columns must have only a single call to 'metadata'"
+                    );
+                }
+
+                return if let &[arg] = &func.args.as_slice() {
+                    if let Expr::Literal(ScalarValue::Utf8(Some(key))) = &arg {
+                        self.key = Some(key.clone());
+                        Ok(TreeNodeRecursion::Stop)
+                    } else {
+                        plan_err!("For metadata columns, metadata call must have a literal string argument")
+                    }
+                } else {
+                    plan_err!("For metadata columns, metadata call must have a single argument")
+                };
+            }
+        }
+        self.depth += 1;
+        Ok(TreeNodeRecursion::Continue)
+    }
+
+    fn f_up(&mut self, _node: &'a Self::Node) -> Result<TreeNodeRecursion> {
+        self.depth -= 1;
+        Ok(TreeNodeRecursion::Continue)
+    }
+}
+
 impl Table {
     fn schema_from_columns(
         columns: &[ColumnDef],
         schema_provider: &ArroyoSchemaProvider,
-        connector_metadata_columns: &mut HashMap<String, (String, DataType)>,
     ) -> Result<Vec<FieldSpec>> {
         let struct_field_pairs = columns
             .iter()
@@ -537,35 +568,6 @@ impl Table {
                         generation_expr, ..
                     } = &option.option
                     {
-                        if let Some(sqlparser::ast::Expr::Function(sqlparser::ast::Function {
-                            name,
-                            args,
-                            ..
-                        })) = generation_expr
-                        {
-                            if name
-                                .0
-                                .iter()
-                                .any(|ident| ident.value.to_lowercase() == "metadata")
-                            {
-                                if let FunctionArguments::List(arg_list) = args {
-                                    if let Some(FunctionArg::Unnamed(
-                                        sqlparser::ast::FunctionArgExpr::Expr(
-                                            sqlparser::ast::Expr::Value(
-                                                sqlparser::ast::Value::SingleQuotedString(value),
-                                            ),
-                                        ),
-                                    )) = arg_list.args.first()
-                                    {
-                                        connector_metadata_columns.insert(
-                                            column.name.value.to_string(),
-                                            (value.to_string(), data_type.clone()),
-                                        );
-                                        return None;
-                                    }
-                                }
-                            }
-                        }
                         generation_expr.clone()
                     } else {
                         None
@@ -612,12 +614,22 @@ impl Table {
                         &mut PlannerContext::default(),
                     )?;
 
-                    Ok(FieldSpec::VirtualField {
-                        field: struct_field,
-                        expression: df_expr,
-                    })
+                    let mut metadata_finder = MetadataFinder::default();
+                    df_expr.visit(&mut metadata_finder)?;
+
+                    if let Some(key) = metadata_finder.key {
+                        Ok(FieldSpec::Metadata {
+                            field: struct_field,
+                            key,
+                        })
+                    } else {
+                        Ok(FieldSpec::Virtual {
+                            field: struct_field,
+                            expression: df_expr,
+                        })
+                    }
                 } else {
-                    Ok(FieldSpec::StructField(struct_field))
+                    Ok(FieldSpec::Struct(struct_field))
                 }
             })
             .collect::<Result<Vec<_>>>()
@@ -646,12 +658,7 @@ impl Table {
             }
 
             let connector = with_map.remove("connector");
-            let mut connector_metadata_columns = Some(HashMap::new());
-            let fields = Self::schema_from_columns(
-                columns,
-                schema_provider,
-                connector_metadata_columns.as_mut().unwrap(),
-            )?;
+            let fields = Self::schema_from_columns(columns, schema_provider)?;
 
             let primary_keys = columns
                 .iter()
@@ -715,7 +722,6 @@ impl Table {
                             primary_keys,
                             &mut with_map,
                             connection_profile,
-                            connector_metadata_columns,
                         )
                         .map_err(|e| e.context(format!("Failed to create table {}", name)))?,
                     )))
diff --git a/crates/arroyo-rpc/src/api_types/connections.rs b/crates/arroyo-rpc/src/api_types/connections.rs
index fccde33f9..d4ba60f87 100644
--- a/crates/arroyo-rpc/src/api_types/connections.rs
+++ b/crates/arroyo-rpc/src/api_types/connections.rs
@@ -1,5 +1,5 @@
 use crate::formats::{BadData, Format, Framing};
-use crate::primitive_to_sql;
+use crate::{primitive_to_sql, MetadataField};
 use anyhow::bail;
 use arrow_schema::{DataType, Field, Fields, TimeUnit};
 use serde::{Deserialize, Serialize};
@@ -120,6 +120,8 @@ pub struct SourceField {
     pub field_name: String,
     pub field_type: SourceFieldType,
     pub nullable: bool,
+    #[serde(default)]
+    pub metadata_key: Option<String>,
 }
 
 impl From<SourceField> for Field {
@@ -220,6 +222,7 @@ impl TryFrom<Field> for SourceField {
                 sql_name,
             },
             nullable: f.is_nullable(),
+            metadata_key: None,
         })
     }
 }
@@ -273,22 +276,28 @@ impl ConnectionSchema {
     }
 
     pub fn validate(self) -> anyhow::Result<Self> {
+        let non_metadata_fields: Vec<_> = self
+            .fields
+            .iter()
+            .filter(|f| f.metadata_key.is_none())
+            .collect();
+
         match &self.format {
             Some(Format::RawString(_)) => {
-                if self.fields.len() != 1
-                    || self.fields.first().unwrap().field_type.r#type
+                if non_metadata_fields.len() != 1
+                    || non_metadata_fields.first().unwrap().field_type.r#type
                         != FieldType::Primitive(PrimitiveType::String)
-                    || self.fields.first().unwrap().field_name != "value"
+                    || non_metadata_fields.first().unwrap().field_name != "value"
                 {
                     bail!("raw_string format requires a schema with a single field called `value` of type TEXT");
                 }
             }
             Some(Format::Json(json_format)) => {
                 if json_format.unstructured
-                    && (self.fields.len() != 1
-                        || self.fields.first().unwrap().field_type.r#type
+                    && (non_metadata_fields.len() != 1
+                        || non_metadata_fields.first().unwrap().field_type.r#type
                             != FieldType::Primitive(PrimitiveType::Json)
-                        || self.fields.first().unwrap().field_name != "value")
+                        || non_metadata_fields.first().unwrap().field_name != "value")
                 {
                     bail!("json format with unstructured flag enabled requires a schema with a single field called `value` of type JSON");
                 }
@@ -304,6 +313,18 @@ impl ConnectionSchema {
         let fields: Vec<Field> = self.fields.iter().map(|f| f.clone().into()).collect();
         Arc::new(ArroyoSchema::from_fields(fields))
     }
+
+    pub fn metadata_fields(&self) -> Vec<MetadataField> {
+        self.fields
+            .iter()
+            .filter_map(|f| {
+                Some(MetadataField {
+                    field_name: f.field_name.clone(),
+                    key: f.metadata_key.clone()?,
+                })
+            })
+            .collect()
+    }
 }
 
 impl From<ConnectionSchema> for ArroyoSchema {
diff --git a/crates/arroyo-rpc/src/lib.rs b/crates/arroyo-rpc/src/lib.rs
index 21058b0c1..36130bbe6 100644
--- a/crates/arroyo-rpc/src/lib.rs
+++ b/crates/arroyo-rpc/src/lib.rs
@@ -183,6 +183,12 @@ pub struct RateLimit {
     pub messages_per_second: u32,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct MetadataField {
+    pub field_name: String,
+    pub key: String,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct OperatorConfig {
     pub connection: Value,
@@ -191,7 +197,8 @@ pub struct OperatorConfig {
     pub bad_data: Option<BadData>,
     pub framing: Option<Framing>,
     pub rate_limit: Option<RateLimit>,
-    pub additional_fields: Option<HashMap<String, String>>,
+    #[serde(default)]
+    pub metadata_fields: Vec<MetadataField>,
 }
 
 impl Default for OperatorConfig {
@@ -203,7 +210,7 @@ impl Default for OperatorConfig {
             bad_data: None,
             framing: None,
             rate_limit: None,
-            additional_fields: None,
+            metadata_fields: vec![],
         }
     }
 }
diff --git a/crates/arroyo-types/src/lib.rs b/crates/arroyo-types/src/lib.rs
index ae3ca4039..3a0a4d98b 100644
--- a/crates/arroyo-types/src/lib.rs
+++ b/crates/arroyo-types/src/lib.rs
@@ -5,7 +5,7 @@ use serde::ser::SerializeStruct;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::convert::TryFrom;
-use std::fmt::Debug;
+use std::fmt::{Debug, Display, Formatter};
 use std::hash::Hash;
 use std::ops::{Range, RangeInclusive};
 use std::sync::Arc;
@@ -456,6 +456,37 @@ pub struct CheckpointBarrier {
     pub then_stop: bool,
 }
 
+pub struct DisplayAsSql<'a>(pub &'a DataType);
+
+impl<'a> Display for DisplayAsSql<'a> {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self.0 {
+            DataType::Boolean => write!(f, "BOOLEAN"),
+            DataType::Int8 | DataType::Int16 | DataType::Int32 => write!(f, "INT"),
+            DataType::Int64 => write!(f, "BIGINT"),
+            DataType::UInt8 | DataType::UInt16 | DataType::UInt32 => write!(f, "INT UNSIGNED"),
+            DataType::UInt64 => write!(f, "BIGINT UNSIGNED"),
+            DataType::Float16 | DataType::Float32 => write!(f, "FLOAT"),
+            DataType::Float64 => write!(f, "DOUBLE"),
+            DataType::Timestamp(_, _) => write!(f, "TIMESTAMP"),
+            DataType::Date32 => write!(f, "DATE"),
+            DataType::Date64 => write!(f, "DATETIME"),
+            DataType::Time32(_) => write!(f, "TIME"),
+            DataType::Time64(_) => write!(f, "TIME"),
+            DataType::Duration(_) => write!(f, "INTERVAL"),
+            DataType::Interval(_) => write!(f, "INTERVAL"),
+            DataType::Binary | DataType::FixedSizeBinary(_) | DataType::LargeBinary => {
+                write!(f, "BYTEA")
+            }
+            DataType::Utf8 | DataType::LargeUtf8 => write!(f, "TEXT"),
+            DataType::List(inner) => {
+                write!(f, "{}[]", DisplayAsSql(inner.data_type()))
+            }
+            dt => write!(f, "{}", dt),
+        }
+    }
+}
+
 #[derive(Debug, Clone, Copy, Eq, PartialEq, PartialOrd, Hash, Serialize)]
 pub enum DatePart {
     Year,


Test patch:
diff --git a/crates/arroyo-connectors/src/kafka/source/test.rs b/crates/arroyo-connectors/src/kafka/source/test.rs
index a8af9cb9d..a6c5f5b51 100644
--- a/crates/arroyo-connectors/src/kafka/source/test.rs
+++ b/crates/arroyo-connectors/src/kafka/source/test.rs
@@ -18,7 +18,7 @@ use arroyo_operator::operator::SourceOperator;
 use arroyo_rpc::df::ArroyoSchema;
 use arroyo_rpc::formats::{Format, RawStringFormat};
 use arroyo_rpc::grpc::rpc::{CheckpointMetadata, OperatorCheckpointMetadata, OperatorMetadata};
-use arroyo_rpc::{CheckpointCompleted, ControlMessage, ControlResp};
+use arroyo_rpc::{CheckpointCompleted, ControlMessage, ControlResp, MetadataField};
 use arroyo_types::{
     single_item_hash_map, to_micros, ArrowMessage, CheckpointBarrier, SignalMessage, TaskInfo,
 };
@@ -87,7 +87,7 @@ impl KafkaTopicTester {
             schema_resolver: None,
             client_configs: HashMap::new(),
             messages_per_second: NonZeroU32::new(100).unwrap(),
-            metadata_fields: None,
+            metadata_fields: vec![],
         });
 
         let (to_control_tx, control_rx) = channel(128);
@@ -358,10 +358,10 @@ async fn test_kafka_with_metadata_fields() {
     kafka_topic_tester.create_topic().await;
 
     // Prepare metadata fields
-    let metadata_fields = Some(HashMap::from([(
-        "offset".to_string(),
-        "offset_id".to_string(),
-    )]));
+    let metadata_fields = vec![MetadataField {
+        field_name: "offset".to_string(),
+        key: "offset_id".to_string(),
+    }];
 
     // Set metadata fields in KafkaSourceFunc
     let mut kafka = KafkaSourceFunc {
diff --git a/crates/arroyo-connectors/src/mqtt/source/test.rs b/crates/arroyo-connectors/src/mqtt/source/test.rs
index 5ab61c09d..f085a7c5e 100644
--- a/crates/arroyo-connectors/src/mqtt/source/test.rs
+++ b/crates/arroyo-connectors/src/mqtt/source/test.rs
@@ -125,7 +125,7 @@ impl MqttTopicTester {
             None,
             None,
             10,
-            None,
+            vec![],
         );
 
         let (to_control_tx, control_rx) = channel(128);
diff --git a/crates/arroyo-planner/src/test/mod.rs b/crates/arroyo-planner/src/test/mod.rs
index 111a59f00..d2ec51f9d 100644
--- a/crates/arroyo-planner/src/test/mod.rs
+++ b/crates/arroyo-planner/src/test/mod.rs
@@ -24,7 +24,6 @@ fn get_test_schema_provider() -> ArroyoSchemaProvider {
                 runtime: Some(10.0 * 1_000_000.0),
             },
             None,
-            None,
         )
         .unwrap();
 
diff --git a/crates/arroyo-planner/src/test/queries/metadata_error.sql b/crates/arroyo-planner/src/test/queries/metadata_error.sql
new file mode 100644
index 000000000..78fc4e39f
--- /dev/null
+++ b/crates/arroyo-planner/src/test/queries/metadata_error.sql
@@ -0,0 +1,14 @@
+--fail=Failed to create table mqtt caused by Error during planning: incorrect data type for metadata field 'topic'; expected TEXT, but found INT
+create table mqtt (
+    name TEXT,
+    value INT,
+    topic INT GENERATED ALWAYS AS (metadata('topic')) STORED
+) with (
+    connector = 'mqtt',
+    url = 'tcp://localhost:1883',
+    topic = 'plant/#',
+    type = 'source',
+    format = 'json'
+);
+
+select topic from mqtt;
\ No newline at end of file
diff --git a/crates/arroyo-planner/src/test/queries/metadata_fields.sql b/crates/arroyo-planner/src/test/queries/metadata_fields.sql
new file mode 100644
index 000000000..fab16eb17
--- /dev/null
+++ b/crates/arroyo-planner/src/test/queries/metadata_fields.sql
@@ -0,0 +1,13 @@
+create table mqtt (
+    name TEXT,
+    value INT,
+    my_topic TEXT GENERATED ALWAYS AS (metadata('topic')) STORED
+) with (
+    connector = 'mqtt',
+    url = 'tcp://localhost:1883',
+    topic = 'plant/#',
+    type = 'source',
+    format = 'json'
+);
+
+select my_topic from mqtt;
\ No newline at end of file
diff --git a/crates/arroyo-planner/src/test/queries/metadata_raw_string.sql b/crates/arroyo-planner/src/test/queries/metadata_raw_string.sql
new file mode 100644
index 000000000..e0d9b3022
--- /dev/null
+++ b/crates/arroyo-planner/src/test/queries/metadata_raw_string.sql
@@ -0,0 +1,12 @@
+create table mqtt (
+    value TEXT,
+    my_topic TEXT GENERATED ALWAYS AS (metadata('topic')) STORED
+) with (
+    connector = 'mqtt',
+    url = 'tcp://localhost:1883',
+    topic = 'plant/#',
+    type = 'source',
+    format = 'raw_string'
+);
+
+select my_topic from mqtt;
\ No newline at end of file