fix: enable streaming append for Kafka with Cayenne accelerator (spiceai#10269) (spiceai#10780)

Allow `refresh_mode: append` to use a connector-provided `append_stream`
when the accelerator engine is Cayenne. Previously, the Cayenne engine was
unconditionally excluded from the append-stream attachment path, so a Kafka
dataset with `engine: cayenne` and `refresh_mode: append` failed because no
streaming source was registered and Cayenne could not fall through to its
primary_key validation (since Kafka messages have neither time_column nor
primary_key).

The check now attempts to attach the source's `append_stream` first. If the
source provides one (Kafka, SpiceAI), it is used for all engines including
Cayenne. If it does not, Cayenne still falls back to its primary_key
validation; other engines require time_column.

Co-authored-by: Claude <noreply@anthropic.com>

Author: lukekim

crates/runtime/src/datafusion/mod.rs

@@ -2398,15 +2398,15 @@ impl DataFusion {
             }
         }
 
-        // For append mode without time_column, check if source provides append_stream
-        // Skip this check for Cayenne which has its own validation (supports primary_key or time_column)
-        if refresh_mode == RefreshMode::Append
-            && dataset.time_column.is_none()
-            && acceleration_settings.engine != Engine::Cayenne
-        {
+        // For append mode without time_column, attach the source's append_stream
+        // when available (e.g. Kafka). This enables streaming append into any
+        // accelerator engine, including Cayenne. When the source does not
+        // provide an append_stream, Cayenne falls back to its own validation
+        // (supports primary_key); other engines require time_column.
+        if refresh_mode == RefreshMode::Append && dataset.time_column.is_none() {
             if let Some(append_stream) = source.append_stream(source_table_provider) {
                 accelerated_table_builder.append_stream(append_stream);
-            } else {
+            } else if acceleration_settings.engine != Engine::Cayenne {
                 return Err(Error::AppendRequiresTimeColumn {
                     from: dataset.from.clone(),
                 });