Add output format and table format labels to filesystem sink trace events (ArroyoSystems#1019)

Author: garvit-gupta

crates/arroyo-connectors/src/filesystem/mod.rs

@@ -40,6 +40,14 @@ pub enum TableFormat {
 }
 
 impl TableFormat {
+    pub fn name(&self) -> &'static str {
+        match self {
+            TableFormat::None => "none",
+            TableFormat::Delta => "delta",
+            TableFormat::Iceberg(_) => "iceberg",
+        }
+    }
+
     pub async fn get_storage_provider(
         &mut self,
         task_info: Arc<TaskInfo>,

crates/arroyo-connectors/src/filesystem/sink/mod.rs

@@ -80,6 +80,8 @@ impl<R: BatchBufferingWriter + Send + 'static> FileSystemSink<R> {
         partitioner_mode: PartitionerMode,
         connection_id: Option<String>,
     ) -> TwoPhaseCommitterOperator<Self> {
+        let output_format = format.name();
+        let table_format_name = table_format.name();
         TwoPhaseCommitterOperator::new(Self {
             sender: None,
             checkpoint_receiver: None,
@@ -95,6 +97,8 @@ impl<R: BatchBufferingWriter + Send + 'static> FileSystemSink<R> {
             event_logger: FsEventLogger {
                 task_info: None,
                 connection_id: connection_id.unwrap_or_default().into(),
+                output_format,
+                table_format: table_format_name,
             },
             _ts: Default::default(),
         })
@@ -186,6 +190,8 @@ fn map_object_store_error(obj_err: &object_store::Error) -> DataflowError {
 pub struct FsEventLogger {
     task_info: Option<Arc<TaskInfo>>,
     connection_id: Arc<String>,
+    output_format: &'static str,
+    table_format: &'static str,
 }
 
 impl FsEventLogger {
@@ -209,6 +215,8 @@ impl FsEventLogger {
             "connection_id": self.connection_id.as_str(),
             "write_error_reason": failure_message.as_deref().unwrap_or(""),
             "subtask_idx": task_info.task_index,
+            "output_format": self.output_format,
+            "table_format": self.table_format,
         }, [
             "bytes_written" => bytes as f64,
             "files_written" => files as f64,

crates/arroyo-connectors/src/filesystem/sink/v2/mod.rs

@@ -304,6 +304,8 @@ impl<BBW: BatchBufferingWriter + Send + 'static> FileSystemSinkV2<BBW> {
         }
 
         let connection_id_str = connection_id.clone().unwrap_or_default();
+        let output_format = format.name();
+        let table_format_name = table_format.name();
 
         Self {
             config: SinkConfig {
@@ -328,6 +330,8 @@ impl<BBW: BatchBufferingWriter + Send + 'static> FileSystemSinkV2<BBW> {
             event_logger: FsEventLogger {
                 task_info: None,
                 connection_id: connection_id_str.into(),
+                output_format,
+                table_format: table_format_name,
             },
             watermark: None,
         }

crates/arroyo-rpc/src/formats.rs

@@ -376,22 +376,22 @@ pub enum Format {
 
 impl Display for Format {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "{}",
-            match self {
-                Format::Json(_) => "json",
-                Format::Avro(_) => "avro",
-                Format::Protobuf(_) => "protobuf",
-                Format::Parquet(_) => "parquet",
-                Format::RawString(_) => "raw_string",
-                Format::RawBytes(_) => "raw_bytes",
-            }
-        )
+        f.write_str(self.name())
     }
 }
 
 impl Format {
+    pub fn name(&self) -> &'static str {
+        match self {
+            Format::Json(_) => "json",
+            Format::Avro(_) => "avro",
+            Format::Protobuf(_) => "protobuf",
+            Format::Parquet(_) => "parquet",
+            Format::RawString(_) => "raw_string",
+            Format::RawBytes(_) => "raw_bytes",
+        }
+    }
+
     pub fn from_opts(opts: &mut ConnectorOptions) -> DFResult<Option<Self>> {
         let Some(name) = opts.pull_opt_str("format")? else {
             return Ok(None);