Add etl_type telemetry dimension (events/changes) (#218)

* Add etl_type telemetry dimension (events/changes)

Emit an etl_type dimension on all benchmark telemetry to distinguish
append-only (events) from mutation (changes) runs.

- Derive etl_type from VersionMetadata mutation ratios
- Add as OTel resource attribute and per-metric attribute
- Include in system adapter setup metadata
- Rename workflow dropdown from append-only/mutations to events/changes
- Add ETL Type dropdown to Grafana dashboard (All/events/changes)
- Filter all 20 dashboard panels by etl_type (same pattern as outcome)

* chore: auto-fix cargo fmt + clippy

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: phillipleblanc

.github/workflows/run_spicebench.yml

@@ -21,11 +21,11 @@ on:
       etl_type:
         description: 'ETL type'
         required: true
-        default: 'append-only'
+        default: 'events'
         type: choice
         options:
-          - append-only
-          - mutations
+          - events
+          - changes
       scale_factor:
         description: 'Scale Factor'
         required: true
@@ -223,7 +223,7 @@ jobs:
           SYSTEM_ADAPTER: ${{ github.event.inputs.system_under_test || 'spice_cloud' }}
           NUM_QUERY_CLIENTS: '8'
           ETL_BUCKET: 'spicebench'
-          ETL_PREFIX: ${{ github.event.inputs.etl_type == 'mutations' && 'data-gen-mutable' || 'data-gen' }}
+          ETL_PREFIX: ${{ github.event.inputs.etl_type == 'changes' && 'data-gen-mutable' || 'data-gen' }}
           SCALE_FACTOR: ${{ github.event.inputs.scale_factor || '1' }}
           ETL_REGION: 'us-east-1'
           ETL_SINK: 'adbc'

crates/adbc_client/src/pool.rs

@@ -72,10 +72,7 @@ impl AdbcConnectionManager {
     /// comparisons against BIGINT columns fail with `DATATYPE_MISMATCH`
     /// unless Int64 literals are explicitly suffixed with `L`.
     pub fn bigint_suffix(driver_name: &str) -> bool {
-        match driver_name {
-            d if d.eq_ignore_ascii_case("databricks") => true,
-            _ => false,
-        }
+        matches!(driver_name, d if d.eq_ignore_ascii_case("databricks"))
     }
 }
 

crates/data-generation/src/version.rs

@@ -103,6 +103,19 @@ impl VersionMetadata {
     pub fn mutation_config(&self) -> MutationConfig {
         MutationConfig::new(self.mutations.update_ratio, self.mutations.delete_ratio)
     }
+
+    /// Returns the ETL type based on the mutation configuration.
+    ///
+    /// - `"events"` — append-only data (no updates or deletes).
+    /// - `"changes"` — data with mutations (updates and/or deletes).
+    #[must_use]
+    pub fn etl_type(&self) -> &'static str {
+        if self.mutations.update_ratio == 0.0 && self.mutations.delete_ratio == 0.0 {
+            "events"
+        } else {
+            "changes"
+        }
+    }
 }
 
 /// Mutation configuration stored in version metadata.

crates/etl/src/sink/adbc.rs

@@ -67,7 +67,7 @@ const ADBC_BULK_INGEST_STREAM_BUFFER_ENV: &str = "SPICEBENCH_ADBC_BULK_INGEST_ST
 /// - `statement`          — row-by-row `UPDATE … SET … WHERE …` statements (default)
 /// - `staging_table`      — bulk ingest into temp staging table + single `MERGE INTO`
 /// - `bulk_ingest_upsert` — bulk ingest directly into the target table (relies on the
-///                          target system's `on_conflict: upsert` or equivalent to merge)
+///   target system's `on_conflict: upsert` or equivalent to merge)
 const ADBC_UPDATE_STRATEGY_ENV: &str = "SPICEBENCH_ADBC_UPDATE_STRATEGY";
 
 /// Strategy for executing UPDATE operations.
@@ -482,9 +482,11 @@ impl AdbcSink {
             .collect::<anyhow::Result<Vec<_>>>()?
             .join(", ");
 
-        let partition_clause = (!partition_by.is_empty())
-            .then(|| format!("PARTITION BY ({})", partition_by.join(", ")))
-            .unwrap_or_default();
+        let partition_clause = if !partition_by.is_empty() {
+            format!("PARTITION BY ({})", partition_by.join(", "))
+        } else {
+            String::new()
+        };
 
         let primary_key_statement = if !primary_keys.is_empty() {
             let key_idents: Vec<String> = primary_keys
@@ -510,7 +512,7 @@ impl AdbcSink {
             .iter()
             .map(|(table_name, config)| {
                 self.create_table_sql(
-                    &table_name,
+                    table_name,
                     config.schema.as_ref(),
                     config.partition_columns.clone(),
                     &config.primary_key_columns,
@@ -924,6 +926,7 @@ impl AdbcSink {
         ))
     }
 
+    #[expect(clippy::too_many_arguments)]
     fn update_sql_for_row(
         &self,
         table_name: &str,