fix: Debezium schema evolution breaks dataset init on reload (fixes spiceai#9782) (spiceai#10144)

* fix: Debezium schema evolution breaks dataset init on runtime reload (fixes spiceai#9782)

The Debezium connector inferred the Arrow schema once from the first Kafka
message at startup and cached it persistently. When the source table schema
evolved (columns added or removed), the stale cached schema caused dataset
initialization to fail on runtime reload.

Three changes fix this:

1. Schema refresh on reload: When cached metadata exists, peek at the latest
   Kafka message via a temporary consumer to detect schema evolution. If the
   schema has changed, update the cached metadata and use the fresh schema.

2. Resilient CDC processing: Handle missing nullable fields in incoming
   messages gracefully by appending null instead of failing. This supports
   replaying older CDC events that predate newly added columns.

3. New KafkaConsumer::fetch_latest_message utility: Seeks to the highest
   watermark offset across all partitions to read the most recent message
   without affecting any existing consumer group state.

* style: Apply rustfmt formatting

* fix: Collapse nested if to satisfy clippy collapsible_if lint

* fix: Replace unwrap() with expect() in tests to satisfy clippy lint

* Address review feedback: isolate temp consumer metrics, add .to_string() for error context consistency, log peek errors

* fix: Update github workflows snapshot after features.yml removal

The `check all features` workflow (.github/workflows/features.yml) was
removed from the repository, shifting the top-10 workflows query result.

* fix: Update search snapshot for s3vectors_chunking_view_with_where

Score for id 551 shifted from 0.28 to 0.29 (consistent across retries),
changing result order when tied with id 1035. Update snapshot to match.

* fix: Make search snapshot tests robust to cross-runner score variance

model2vec similarity scores vary ±0.01 across CI runners (different
macOS versions), causing snapshot tests to fail when scores land on
different sides of truncation boundaries.

Two fixes:
1. normalize_search_response_json: use round() instead of trunc() for
   score display and sorting. Scores like 0.289 now consistently round
   to 0.29 instead of truncating to 0.28 on some runners.
2. SQL test queries: reduce trunc(_score, 3) to trunc(_score, 2) to
   avoid flakiness at the 3rd decimal place (e.g., 0.556 vs 0.557).

* fix: Apply cargo fmt to search test normalization

* fix: Update OpenAI search snapshots for embedding model score shift

OpenAI's text-embedding-3-small model scores shifted by +0.01,
causing snapshot mismatches in the openai_test_search CI check.

* fix: Scope score rounding to s3vectors tests only

The previous change to use `round` instead of `trunc` for score display
in `normalize_search_response_json` was applied globally, causing
cascading snapshot failures in OpenAI search tests (0.65→0.66, etc.).

This fix adds a `round_scores` flag to `SearchTestCase` and
`run_search_w_explain` so that only s3vectors tests (which have
non-deterministic model2vec scores that vary ±0.002 across CI runners)
use rounding for display. All other tests (OpenAI, HF, text search)
continue to use truncation, preserving their existing snapshots.

Sort comparison still uses rounding universally to stabilize ordering.

* fix: Revert OpenAI snapshots to truncated score values

The previous commit incorrectly updated these snapshots to rounded
values when the normalization was unconditionally using round(). Now
that rounding is scoped to s3vectors tests only, OpenAI tests use
truncation again - restore the original snapshot values.

* fix: Also scope sort rounding to round_scores flag

The sort comparison was unconditionally using rounded values, causing
ordering mismatches with truncated display values in OpenAI tests.
Now both sort and display use the same precision mode: raw floats when
round_scores is false, rounded when true.

* fix: Use score rounding for OpenAI search tests

OpenAI embeddings are non-deterministic — scores vary by ±0.01 across
CI runs, causing snapshot failures when truncation amplifies boundary
effects. Switch OpenAI search tests to use score rounding (same as
model2vec/s3vectors tests) for more stable comparisons.

* fix: Correct round_scores=false for OpenAI tests, remove unused builder, update github workflows snapshot

- OpenAI tests should use truncation (round_scores=false) since their embeddings are deterministic
- Remove unused round_scores() builder method that triggered lint error
- Update github workflows snapshot to reflect removed integration.yml workflow

* fix: Update snapshot expression headers to match new function signatures

All normalize_search_response and normalize_search_response_json calls
now include the round_scores parameter. Update snapshot expression lines
to match so insta doesn't flag expression mismatches.

* fix: Update snapshot column aliases from trunc(_score,3) to trunc(_score,2)

SQL test queries were changed from trunc(_score, 3) to trunc(_score, 2)
in a previous commit. Update all snapshot files that reference the old
Int64(3) column alias to use Int64(2).

* Gate schema evolution behind opt-in `schema_evolution` parameter

Address reviewer feedback: schema evolution detection is now disabled
by default and must be explicitly enabled with `schema_evolution: true`
in the dataset params. This preserves the intentional behavior of
preventing schema evolution at the accelerator level while allowing
users who need it to opt in.

* fix: Revert unrelated snapshot/test changes, keep only Debezium schema evolution fix

Remove the score rounding normalization changes and trunc precision
modifications that were unrelated to the Debezium schema evolution fix.
Restore search.rs, s3_vectors.rs, openai.rs, and all snapshot files to
trunk state so this PR only contains the Debezium schema evolution logic.

---------

Co-authored-by: Claude <claude@Mac-mini.localdomain>
Co-authored-by: Luke Kim <80174+lukekim@users.noreply.github.com>

Author: 3 people

crates/data_components/src/debezium/arrow.rs

@@ -169,14 +169,21 @@ pub fn append_value_to_struct_builder(
     builder: &mut StructBuilder,
 ) -> Result<()> {
     builder.append(true);
+    let null_value = serde_json::Value::Null;
 
     for (idx, field) in builder.fields().iter().enumerate() {
-        let Some(field_value) = value.get(field.name()) else {
-            return MissingFieldInValueSnafu {
-                field_name: field.name().clone(),
-                value,
+        // If the field is missing from the message (e.g. due to schema evolution),
+        // append null for nullable fields instead of failing.
+        let field_value = match value.get(field.name()) {
+            Some(v) => v,
+            None if field.is_nullable() => &null_value,
+            None => {
+                return MissingFieldInValueSnafu {
+                    field_name: field.name().clone(),
+                    value,
+                }
+                .fail();
             }
-            .fail();
         };
 
         let field_builder = builder.field_builder_array(idx);
@@ -698,4 +705,135 @@ mod tests {
         let result = convert_json_to_decimal(&input, 2);
         result.expect_err("Should fail for wrong JSON type");
     }
+
+    #[test]
+    fn test_append_value_missing_nullable_field_fills_null() {
+        use crate::arrow::struct_builder::StructBuilder;
+        use arrow::array::Array;
+
+        // Schema with one required and one nullable field
+        let schema = Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new("name", DataType::Utf8, true),
+        ]);
+
+        let mut builder = StructBuilder::from_fields(schema.fields().clone(), 1);
+
+        // Message is missing the nullable "name" field
+        let value = json!({"id": 42});
+        let result = append_value_to_struct_builder(value, &mut builder);
+        assert!(
+            result.is_ok(),
+            "Should succeed when nullable field is missing"
+        );
+
+        let struct_array = builder.finish();
+        let record_batch: RecordBatch = struct_array.into();
+        assert_eq!(record_batch.num_rows(), 1);
+
+        let id_col = record_batch
+            .column_by_name("id")
+            .expect("id column should exist")
+            .as_any()
+            .downcast_ref::<arrow::array::Int32Array>()
+            .expect("id column should be Int32Array");
+        assert_eq!(id_col.value(0), 42);
+
+        let name_col = record_batch
+            .column_by_name("name")
+            .expect("name column should exist")
+            .as_any()
+            .downcast_ref::<arrow::array::StringArray>()
+            .expect("name column should be StringArray");
+        assert!(name_col.is_null(0));
+    }
+
+    #[test]
+    fn test_append_value_missing_required_field_fails() {
+        use crate::arrow::struct_builder::StructBuilder;
+
+        let schema = Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new("status", DataType::Utf8, false), // not nullable
+        ]);
+
+        let mut builder = StructBuilder::from_fields(schema.fields().clone(), 1);
+
+        // Message is missing the required "status" field
+        let value = json!({"id": 42});
+        let result = append_value_to_struct_builder(value, &mut builder);
+        assert!(
+            result.is_err(),
+            "Should fail when required field is missing"
+        );
+    }
+
+    #[test]
+    fn test_append_value_extra_fields_ignored() {
+        use crate::arrow::struct_builder::StructBuilder;
+
+        // Schema only has "id"
+        let schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]);
+
+        let mut builder = StructBuilder::from_fields(schema.fields().clone(), 1);
+
+        // Message has extra field "removed_column" not in schema
+        let value = json!({"id": 42, "removed_column": "old_value"});
+        let result = append_value_to_struct_builder(value, &mut builder);
+        assert!(result.is_ok(), "Extra fields in message should be ignored");
+
+        let struct_array = builder.finish();
+        let record_batch: RecordBatch = struct_array.into();
+        assert_eq!(record_batch.num_rows(), 1);
+        assert_eq!(record_batch.num_columns(), 1);
+    }
+
+    #[test]
+    fn test_append_value_multiple_missing_nullable_fields() {
+        use crate::arrow::struct_builder::StructBuilder;
+        use arrow::array::Array;
+
+        // Schema with multiple nullable fields added via schema evolution
+        let schema = Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new("name", DataType::Utf8, true),
+            Field::new("age", DataType::Int64, true),
+            Field::new("active", DataType::Boolean, true),
+        ]);
+
+        let mut builder = StructBuilder::from_fields(schema.fields().clone(), 2);
+
+        // Old message with only "id" (before schema evolution)
+        let old_value = json!({"id": 1});
+        append_value_to_struct_builder(old_value, &mut builder)
+            .expect("old message should process successfully");
+
+        // New message with all fields
+        let new_value = json!({"id": 2, "name": "Alice", "age": 30, "active": true});
+        append_value_to_struct_builder(new_value, &mut builder)
+            .expect("new message should process successfully");
+
+        let struct_array = builder.finish();
+        let record_batch: RecordBatch = struct_array.into();
+        assert_eq!(record_batch.num_rows(), 2);
+
+        // First row: id=1, rest null
+        let id_col = record_batch
+            .column_by_name("id")
+            .expect("id column should exist")
+            .as_any()
+            .downcast_ref::<arrow::array::Int32Array>()
+            .expect("id column should be Int32Array");
+        assert_eq!(id_col.value(0), 1);
+        assert_eq!(id_col.value(1), 2);
+
+        let name_col = record_batch
+            .column_by_name("name")
+            .expect("name column should exist")
+            .as_any()
+            .downcast_ref::<arrow::array::StringArray>()
+            .expect("name column should be StringArray");
+        assert!(name_col.is_null(0));
+        assert_eq!(name_col.value(1), "Alice");
+    }
 }

crates/data_components/src/kafka.rs

@@ -496,6 +496,84 @@ impl KafkaConsumer {
         })
     }
 
+    /// Fetch the latest message from a Kafka topic without affecting any existing
+    /// consumer group state.
+    ///
+    /// Creates a temporary consumer, seeks to the latest available message across
+    /// all partitions, reads it, and returns the owned key/value pair.
+    pub async fn fetch_latest_message<K: DeserializeOwned, V: DeserializeOwned>(
+        topic: &str,
+        kafka_config: &KafkaConfig,
+        timeout: Duration,
+    ) -> Result<Option<(Option<K>, V)>> {
+        let temp_group_id = format!("spice-schema-peek-{}", uuid::Uuid::new_v4());
+        let mut peek_config = kafka_config.clone();
+        peek_config.metrics_store = None; // Avoid skewing real consumer metrics
+        let temp_consumer = Self::create(temp_group_id, &peek_config)?;
+
+        // Fetch topic metadata to discover partitions
+        let metadata = temp_consumer
+            .consumer
+            .fetch_metadata(Some(topic), timeout)
+            .context(UnableToRestartTopicSnafu {
+                message: "Failed to fetch topic metadata".to_string(),
+            })?;
+
+        let topic_metadata = metadata
+            .topics()
+            .iter()
+            .find(|t| t.name() == topic)
+            .context(MetadataTopicNotFoundSnafu {
+                topic: topic.to_string(),
+            })?;
+
+        // Find the partition with the highest watermark (most recent data)
+        let mut best_partition: Option<(i32, i64)> = None;
+        for partition in topic_metadata.partitions() {
+            let (low, high) = temp_consumer
+                .consumer
+                .fetch_watermarks(topic, partition.id(), timeout)
+                .context(UnableToRestartTopicSnafu {
+                    message: format!(
+                        "Failed to fetch watermarks for partition {}",
+                        partition.id()
+                    ),
+                })?;
+
+            if high > low {
+                match &best_partition {
+                    Some((_, best_high)) if high <= *best_high => {}
+                    _ => best_partition = Some((partition.id(), high)),
+                }
+            }
+        }
+
+        let Some((partition_id, high_watermark)) = best_partition else {
+            return Ok(None); // No messages available
+        };
+
+        // Manually assign the consumer to read from the latest offset
+        let mut tpl = rdkafka::TopicPartitionList::new();
+        tpl.add_partition_offset(topic, partition_id, Offset::Offset(high_watermark - 1))
+            .context(UnableToRestartTopicSnafu {
+                message: "Failed to configure partition offset".to_string(),
+            })?;
+
+        temp_consumer
+            .consumer
+            .assign(&tpl)
+            .context(UnableToRestartTopicSnafu {
+                message: "Failed to assign partition".to_string(),
+            })?;
+
+        // Read the message with a timeout
+        match tokio::time::timeout(timeout, temp_consumer.next_json::<K, V>()).await {
+            Ok(Ok(Some(msg))) => Ok(Some(msg.into_key_value())),
+            Ok(Ok(None)) | Err(_) => Ok(None),
+            Ok(Err(e)) => Err(e),
+        }
+    }
+
     fn generate_group_id(dataset: &str) -> String {
         format!("spice.ai-{dataset}-{}", uuid::Uuid::new_v4())
     }
@@ -548,6 +626,11 @@ impl<'a, K, V> KafkaMessage<'a, K, V> {
             .store_offset_from_message(&self.msg)
             .context(UnableToCommitMessageSnafu)
     }
+
+    /// Consume the message and return owned key/value data.
+    pub fn into_key_value(self) -> (Option<K>, V) {
+        (self.key, self.value)
+    }
 }
 
 #[async_trait]