Merge pull request #7558 from lichuang/add_table_stage_file_meta

feat: save table stage file info into meta, remove these data when truncate table

Author: mergify[bot]

src/meta/api/src/kv_api_key.rs

@@ -30,6 +30,9 @@ pub enum KVApiKeyError {
     #[error("Expect {expect} segments, but: '{got}'")]
     WrongNumberOfSegments { expect: usize, got: String },
 
+    #[error("Expect at least {expect} segments, but {actual} segments found")]
+    AtleastSegments { expect: usize, actual: usize },
+
     #[error("Invalid id string: '{s}': {reason}")]
     InvalidId { s: String, reason: String },
 }

src/meta/api/src/kv_api_test_suite.rs

@@ -483,6 +483,7 @@ impl KVApiTestSuite {
                 key: txn_key.clone(),
                 value: b"new_v1".to_vec(),
                 prev_value: true,
+                expire_at: None,
             })),
         }];
 
@@ -640,6 +641,7 @@ impl KVApiTestSuite {
                     key: txn_key.clone(),
                     value: b"new_v1".to_vec(),
                     prev_value: true,
+                    expire_at: None,
                 })),
             }];
 
@@ -691,6 +693,7 @@ impl KVApiTestSuite {
                     key: txn_key1.clone(),
                     value: b"new_v1".to_vec(),
                     prev_value: true,
+                    expire_at: None,
                 })),
             }];
 
@@ -745,6 +748,7 @@ impl KVApiTestSuite {
                         key: txn_key1.clone(),
                         value: val1_new.to_vec(),
                         prev_value: true,
+                        expire_at: None,
                     })),
                 },
                 // change k2
@@ -753,6 +757,7 @@ impl KVApiTestSuite {
                         key: txn_key2.clone(),
                         value: b"new_v2".to_vec(),
                         prev_value: true,
+                        expire_at: None,
                     })),
                 },
                 // get k1
@@ -852,6 +857,7 @@ impl KVApiTestSuite {
                         key: txn_key1.clone(),
                         value: val1_new.to_vec(),
                         prev_value: true,
+                        expire_at: None,
                     })),
                 },
                 // get k1

src/meta/api/src/kv_api_utils.rs

@@ -226,6 +226,18 @@ pub fn txn_op_put(key: &impl KVApiKey, value: Vec<u8>) -> TxnOp {
             key: key.to_key(),
             value,
             prev_value: true,
+            expire_at: None,
+        })),
+    }
+}
+
+pub fn txn_op_put_with_expire(key: &impl KVApiKey, value: Vec<u8>, expire_at: u64) -> TxnOp {
+    TxnOp {
+        request: Some(Request::Put(TxnPutRequest {
+            key: key.to_key(),
+            value,
+            prev_value: true,
+            expire_at: Some(expire_at),
         })),
     }
 }

src/meta/api/src/lib.rs

@@ -64,6 +64,7 @@ pub use kv_api_utils::table_has_to_exist;
 pub use kv_api_utils::txn_cond_seq;
 pub use kv_api_utils::txn_op_del;
 pub use kv_api_utils::txn_op_put;
+pub use kv_api_utils::txn_op_put_with_expire;
 pub use kv_api_utils::TXN_MAX_RETRY_TIMES;
 pub use schema_api::SchemaApi;
 pub(crate) use schema_api_impl::get_db_or_err;

src/meta/api/src/schema_api.rs

@@ -26,6 +26,8 @@ use common_meta_app::schema::DropDatabaseReq;
 use common_meta_app::schema::DropTableReply;
 use common_meta_app::schema::DropTableReq;
 use common_meta_app::schema::GetDatabaseReq;
+use common_meta_app::schema::GetTableCopiedFileReply;
+use common_meta_app::schema::GetTableCopiedFileReq;
 use common_meta_app::schema::GetTableReq;
 use common_meta_app::schema::ListDatabaseReq;
 use common_meta_app::schema::ListTableReq;
@@ -36,12 +38,16 @@ use common_meta_app::schema::RenameTableReq;
 use common_meta_app::schema::TableIdent;
 use common_meta_app::schema::TableInfo;
 use common_meta_app::schema::TableMeta;
+use common_meta_app::schema::TruncateTableReply;
+use common_meta_app::schema::TruncateTableReq;
 use common_meta_app::schema::UndropDatabaseReply;
 use common_meta_app::schema::UndropDatabaseReq;
 use common_meta_app::schema::UndropTableReply;
 use common_meta_app::schema::UndropTableReq;
 use common_meta_app::schema::UpdateTableMetaReply;
 use common_meta_app::schema::UpdateTableMetaReq;
+use common_meta_app::schema::UpsertTableCopiedFileReply;
+use common_meta_app::schema::UpsertTableCopiedFileReq;
 use common_meta_app::schema::UpsertTableOptionReply;
 use common_meta_app::schema::UpsertTableOptionReq;
 use common_meta_types::GCDroppedDataReply;
@@ -104,6 +110,18 @@ pub trait SchemaApi: Send + Sync {
         table_id: MetaId,
     ) -> Result<(TableIdent, Arc<TableMeta>), MetaError>;
 
+    async fn get_table_copied_file_info(
+        &self,
+        req: GetTableCopiedFileReq,
+    ) -> Result<GetTableCopiedFileReply, MetaError>;
+
+    async fn upsert_table_copied_file_info(
+        &self,
+        req: UpsertTableCopiedFileReq,
+    ) -> Result<UpsertTableCopiedFileReply, MetaError>;
+
+    async fn truncate_table(&self, req: TruncateTableReq) -> Result<TruncateTableReply, MetaError>;
+
     async fn upsert_table_option(
         &self,
         req: UpsertTableOptionReq,