[REST] enforce backward compatibility before shipping (#2057)

Author: vvvdwbvvv

src/moonlink_service/src/rest_api.rs

@@ -55,10 +55,11 @@ impl ApiState {
 /// ====================
 ///
 /// Request mode.
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
 #[serde(rename_all = "lowercase")]
 pub enum RequestMode {
     /// Only issues request, but not block wait its completion.
+    #[default]
     Async,
     /// Block wait request completion.
     Sync,
@@ -67,6 +68,7 @@ pub enum RequestMode {
 /// Error response structure
 #[derive(Debug, Serialize)]
 pub struct ErrorResponse {
+    #[serde(rename = "message")]
     pub message: String,
 }
 
@@ -87,18 +89,32 @@ pub struct GetTableSchemaResponse {
 /// Request structure for table creation
 #[derive(Debug, Serialize, Deserialize)]
 pub struct CreateTableRequest {
+    #[serde(rename = "database")]
     pub database: String,
+
+    #[serde(rename = "table")]
     pub table: String,
+
+    #[serde(rename = "schema")]
     pub schema: Option<Vec<FieldSchema>>,
+
+    #[serde(rename = "avro_schema")]
     pub avro_schema: Option<serde_json::Value>,
+
+    #[serde(rename = "table_config")]
     pub table_config: TableConfig,
 }
 
 /// Response structure for table creation
 #[derive(Debug, Serialize, Deserialize)]
 pub struct CreateTableResponse {
+    #[serde(rename = "database")]
     pub database: String,
+
+    #[serde(rename = "table")]
     pub table: String,
+
+    #[serde(rename = "lsn")]
     pub lsn: u64,
 }
 
@@ -109,18 +125,29 @@ pub struct CreateTableResponse {
 /// Request structure for kafka schema creation.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct SetAvroSchemaRequest {
+    #[serde(rename = "database")]
     pub database: String,
+
+    #[serde(rename = "table")]
     pub table: String,
     /// Avro schema JSON object
+    #[serde(rename = "kafka_schema")]
     pub kafka_schema: serde_json::Value,
+
+    #[serde(rename = "schema_id")]
     pub schema_id: u64,
 }
 
 /// Response structure for kafka schema creation.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct SetAvroSchemaResponse {
+    #[serde(rename = "database")]
     pub database: String,
+
+    #[serde(rename = "table")]
     pub table: String,
+
+    #[serde(rename = "schema_id")]
     pub schema_id: u64,
 }
 
@@ -131,18 +158,32 @@ pub struct SetAvroSchemaResponse {
 /// Request structure for creating table from PostgreSQL source
 #[derive(Debug, Serialize, Deserialize)]
 pub struct CreateTableFromPostgresRequest {
+    #[serde(rename = "database")]
     pub database: String,
+
+    #[serde(rename = "table")]
     pub table: String,
+
+    #[serde(rename = "src_uri")]
     pub src_uri: String,
+
+    #[serde(rename = "src_table_name")]
     pub src_table_name: String,
+
+    #[serde(rename = "table_config")]
     pub table_config: TableConfig,
 }
 
 /// Response structure for creating table from PostgreSQL source
 #[derive(Debug, Serialize, Deserialize)]
 pub struct CreateTableFromPostgresResponse {
+    #[serde(rename = "database")]
     pub database: String,
+
+    #[serde(rename = "table")]
     pub table: String,
+
+    #[serde(rename = "lsn")]
     pub lsn: u64,
 }
 
@@ -153,7 +194,12 @@ pub struct CreateTableFromPostgresResponse {
 /// Request structure for table drop.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct DropTableRequest {
+    #[serde(rename = "database")]
+    #[serde(default)]
     pub database: String,
+
+    #[serde(rename = "table")]
+    #[serde(default)]
     pub table: String,
 }
 
@@ -168,6 +214,7 @@ pub struct DropTableResponse {}
 /// Response structure for table list.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct ListTablesResponse {
+    #[serde(rename = "tables")]
     pub tables: Vec<TableStatus>,
 }
 
@@ -178,8 +225,13 @@ pub struct ListTablesResponse {
 /// Request structure for table optimization.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct OptimizeTableRequest {
+    #[serde(rename = "database")]
     pub database: String,
+
+    #[serde(rename = "table")]
     pub table: String,
+
+    #[serde(rename = "mode")]
     pub mode: String,
 }
 
@@ -194,8 +246,13 @@ pub struct OptimizeTableResponse {}
 /// Request structure for snapshot creation.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct CreateSnapShotRequest {
+    #[serde(rename = "database")]
     pub database: String,
+
+    #[serde(rename = "table")]
     pub table: String,
+
+    #[serde(rename = "lsn")]
     pub lsn: u64,
 }
 
@@ -210,27 +267,42 @@ pub struct CreateSnapShotResponse {}
 /// Request structure for data ingestion
 #[derive(Debug, Serialize, Deserialize)]
 pub struct IngestRequest {
+    #[serde(rename = "operation")]
     pub operation: String,
+
+    #[serde(rename = "data")]
     pub data: serde_json::Value,
     /// Whether to enable synchronous mode.
+    #[serde(rename = "request_mode")]
+    #[serde(default)]
     pub request_mode: RequestMode,
 }
 
 /// Request structure for data ingestion with protobuf
 #[derive(Debug, Serialize, Deserialize)]
 pub struct IngestProtobufRequest {
+    #[serde(rename = "operation")]
     pub operation: String,
+
+    #[serde(rename = "data")]
     pub data: Vec<u8>,
     /// Whether to enable synchronous mode.
+    #[serde(rename = "request_mode")]
+    #[serde(default)]
     pub request_mode: RequestMode,
 }
 
 /// Response structure for data ingestion
 #[derive(Debug, Serialize, Deserialize)]
 pub struct IngestResponse {
+    #[serde(rename = "table")]
     pub table: String,
+
+    #[serde(rename = "operation")]
     pub operation: String,
+
     /// Assigned for synchronous mode.
+    #[serde(rename = "lsn")]
     pub lsn: Option<u64>,
 }
 
@@ -241,18 +313,26 @@ pub struct IngestResponse {
 #[derive(Debug, Serialize, Deserialize)]
 pub struct FileUploadRequest {
     /// Ingestion operation.
+    #[serde(rename = "operation")]
     pub operation: String,
+
     /// Files to ingest into mooncake table.
+    #[serde(rename = "files")]
     pub files: Vec<String>,
+
     /// Storage configuration to access files.
+    #[serde(rename = "storage_config")]
     pub storage_config: StorageConfig,
     /// Whether to enable synchronous mode.
+    #[serde(rename = "request_mode")]
+    #[serde(default)]
     pub request_mode: RequestMode,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct FileUploadResponse {
     /// Assigned for synchronous mode.
+    #[serde(rename = "lsn")]
     pub lsn: Option<u64>,
 }
 
@@ -262,8 +342,13 @@ pub struct FileUploadResponse {
 ///
 #[derive(Debug, Serialize, Deserialize)]
 pub struct SyncFlushRequest {
+    #[serde(rename = "database")]
     pub database: String,
+
+    #[serde(rename = "table")]
     pub table: String,
+
+    #[serde(rename = "lsn")]
     pub lsn: u64,
 }
 
@@ -277,8 +362,13 @@ pub struct SyncFlushResponse {}
 /// Health check response
 #[derive(Debug, Serialize, Deserialize)]
 pub struct HealthResponse {
+    #[serde(rename = "service")]
     pub service: String,
+
+    #[serde(rename = "status")]
     pub status: String,
+
+    #[serde(rename = "timestamp")]
     pub timestamp: u64,
 }
 

src/moonlink_service/src/test.rs

@@ -4,7 +4,9 @@ use serde_json::json;
 use serial_test::serial;
 use tokio::net::TcpStream;
 
-use crate::rest_api::{CreateTableResponse, FileUploadResponse, HealthResponse, IngestResponse};
+use crate::rest_api::{
+    CreateTableResponse, DropTableRequest, FileUploadResponse, HealthResponse, IngestResponse,
+};
 use crate::start_with_config;
 use crate::test_guard::TestGuard;
 use crate::test_utils::*;
@@ -497,6 +499,24 @@ async fn test_drop_and_recreate_table() {
     assert_eq!(list_results.len(), 1);
 }
 
+#[tokio::test]
+async fn test_drop_uses_defaults() {
+    let s = r#"{}"#;
+    let req: DropTableRequest = serde_json::from_str(s).unwrap();
+    assert_eq!(req.database, "");
+    assert_eq!(req.table, "");
+}
+
+#[tokio::test]
+async fn test_drop_serialize_uses_renamed_fields() {
+    let req = DropTableRequest {
+        database: "database".into(),
+        table: "users".into(),
+    };
+    let out = serde_json::to_string(&req).unwrap();
+    assert!(out.contains("\"table\":\"users\""));
+}
+
 /// Testing scenario: create multiple tables, and check list table result.
 #[tokio::test]
 #[serial]