Add format field to DataSyncCommand (#764)

clade/proto/sync.proto

@@ -38,6 +38,9 @@ message DataSyncCommand {
   // Monotonically-increasing transaction number.
   // Only specified in the last message of a transaction
   optional uint64 sequence_number = 5;
+
+  // Table format
+  schema.TableFormat format = 6;
 }
 
 message DataSyncResponse {

src/frontend/flight/handler.rs

@@ -4,6 +4,7 @@ use arrow::record_batch::RecordBatch;
 use arrow_flight::sql::metadata::{SqlInfoData, SqlInfoDataBuilder};
 use arrow_flight::sql::{ProstMessageExt, SqlInfo, TicketStatementQuery};
 use arrow_flight::{FlightDescriptor, FlightEndpoint, FlightInfo, Ticket};
+use clade::schema::TableFormat;
 use clade::sync::{DataSyncCommand, DataSyncResponse};
 use dashmap::DashMap;
 use datafusion::common::Result;
@@ -22,7 +23,7 @@ use url::Url;
 use crate::context::SeafowlContext;
 use crate::sync::schema::SyncSchema;
 use crate::sync::writer::SeafowlDataSyncWriter;
-use crate::sync::SyncResult;
+use crate::sync::{SyncError, SyncResult};
 
 lazy_static! {
     pub static ref SEAFOWL_SQL_DATA: SqlInfoData = {
@@ -162,6 +163,10 @@ impl SeafowlFlightHandler {
             });
         }
 
+        if cmd.format != TableFormat::Delta as i32 {
+            return Err(SyncError::NotImplemented);
+        }
+
         let log_store = match cmd.store {
             None => self
                 .context

src/sync/mod.rs

@@ -22,6 +22,9 @@ pub enum SyncError {
     #[error("Invalid sync message: {reason}")]
     InvalidMessage { reason: String },
 
+    #[error("Not implemented")]
+    NotImplemented,
+
     #[error(transparent)]
     ArrowError(#[from] arrow_schema::ArrowError),
 

tests/flight/sync.rs

@@ -1,6 +1,6 @@
 use crate::fixtures::minio_options;
 use crate::flight::*;
-use clade::schema::StorageLocation;
+use clade::schema::{StorageLocation, TableFormat};
 use clade::sync::{ColumnDescriptor, ColumnRole};
 use deltalake::DeltaTable;
 use std::collections::HashMap;
@@ -118,6 +118,7 @@ async fn test_sync_happy_path() -> std::result::Result<(), Box<dyn std::error::E
         column_descriptors,
         origin: "42".to_string(),
         sequence_number: None,
+        format: TableFormat::Delta.into(),
     };
 
     // Changes are still in memory
@@ -569,6 +570,7 @@ async fn test_sync_custom_store(
         column_descriptors,
         origin: "42".to_string(),
         sequence_number: Some(1000),
+        format: TableFormat::Delta.into(),
     };
 
     let sync_result = do_put_sync(cmd.clone(), batch.clone(), &mut client).await?;

tests/flight/sync_fail.rs

@@ -1,5 +1,8 @@
 use crate::flight::*;
-use clade::sync::{ColumnDescriptor, ColumnRole};
+use clade::{
+    schema::TableFormat,
+    sync::{ColumnDescriptor, ColumnRole},
+};
 
 async fn assert_sync_error(
     cmd: DataSyncCommand,
@@ -36,6 +39,7 @@ async fn test_sync_errors() -> std::result::Result<(), Box<dyn std::error::Error
         column_descriptors: vec![],
         origin: "1".to_string(),
         sequence_number: Some(42),
+        format: TableFormat::Delta.into(),
     };
 
     // No column descriptors provided