feat: add etag field to StageFile

1. add an `Option<String>` etag field to StageFile
2. updated proto-conv version
3. added parse_stage_file_v7 and parse_stage_file_latest test

Signed-off-by: ClSlaid <cailue@bupt.edu.cn>

Author: ClSlaid

src/meta/proto-conv/src/user_from_to_protobuf_impl.rs

@@ -657,6 +657,7 @@ impl FromToProto for mt::StageFile {
                 Some(c) => Some(mt::UserIdentity::from_pb(c)?),
                 None => None,
             },
+            etag: p.etag.clone(),
         })
     }
 
@@ -672,6 +673,7 @@ impl FromToProto for mt::StageFile {
                 Some(c) => Some(mt::UserIdentity::to_pb(c)?),
                 None => None,
             },
+            etag: self.etag.clone(),
         })
     }
 }

src/meta/proto-conv/src/util.rs

@@ -42,6 +42,7 @@ const META_CHANGE_LOG: &[(u64, &str)] = &[
         7,
         "2022-09-09: Add: table.proto/{TableCopiedFileInfo,TableCopiedFileLock} type",
     ),
+    (8, "2022-09-16: Add: users.proto/StageFile::entity_tag"),
 ];
 
 pub const VER: u64 = META_CHANGE_LOG.last().unwrap().0;

src/meta/proto-conv/tests/it/user_proto_conv.rs

@@ -14,6 +14,11 @@
 
 use std::collections::HashSet;
 
+use common_datavalues::chrono::DateTime;
+use common_datavalues::chrono::NaiveDate;
+use common_datavalues::chrono::NaiveDateTime;
+use common_datavalues::chrono::NaiveTime;
+use common_datavalues::chrono::Utc;
 use common_meta_types as mt;
 use common_meta_types::UserInfo;
 use common_meta_types::UserPrivilegeType;
@@ -148,6 +153,22 @@ pub(crate) fn test_gcs_stage_info() -> mt::UserStageInfo {
     }
 }
 
+pub(crate) fn test_stage_file() -> mt::StageFile {
+    let dt = NaiveDateTime::new(
+        NaiveDate::from_ymd(2022, 9, 16),
+        NaiveTime::from_hms(0, 1, 2),
+    );
+    let user_id = mt::UserIdentity::new("datafuselabs", "datafuselabs.rs");
+    mt::StageFile {
+        path: "/path/to/stage".to_string(),
+        size: 233,
+        md5: None,
+        last_modified: DateTime::from_utc(dt, Utc),
+        creator: Some(user_id),
+        etag: None,
+    }
+}
+
 #[test]
 fn test_user_pb_from_to() -> anyhow::Result<()> {
     let test_user_info = test_user_info();
@@ -253,6 +274,15 @@ fn test_build_user_pb_buf() -> anyhow::Result<()> {
         println!("{:?}", buf);
     }
 
+    // StageFile
+    {
+        let stage_file = test_stage_file();
+        let p = stage_file.to_pb()?;
+        let mut buf = vec![];
+        common_protos::prost::Message::encode(&p, &mut buf)?;
+        println!("{:?}", buf);
+    }
+
     // Stage on local file system
     {
         let fs_stage_info = test_fs_stage_info();

src/meta/proto-conv/tests/it/user_stage.rs

@@ -14,6 +14,11 @@
 
 //! Test UserStageInfo
 
+use common_datavalues::chrono::DateTime;
+use common_datavalues::chrono::NaiveDate;
+use common_datavalues::chrono::NaiveDateTime;
+use common_datavalues::chrono::NaiveTime;
+use common_datavalues::chrono::Utc;
 use common_meta_types as mt;
 use common_storage::StorageFsConfig;
 use common_storage::StorageGcsConfig;
@@ -24,6 +29,13 @@ use crate::common;
 use crate::user_proto_conv::test_fs_stage_info;
 use crate::user_proto_conv::test_gcs_stage_info;
 use crate::user_proto_conv::test_s3_stage_info;
+use crate::user_proto_conv::test_stage_file;
+
+#[test]
+fn test_stage_file_latest() -> anyhow::Result<()> {
+    common::test_pb_from_to("stage_file", test_stage_file())?;
+    Ok(())
+}
 
 #[test]
 fn test_user_stage_fs_latest() -> anyhow::Result<()> {
@@ -43,6 +55,36 @@ fn test_user_stage_gcs_latest() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[test]
+fn test_stage_file_v7() -> anyhow::Result<()> {
+    // Encoded data of version 6 of StageFile:
+    // Generated with common::test_pb_from_to.
+    let stage_file_v7 = vec![
+        10, 14, 47, 112, 97, 116, 104, 47, 116, 111, 47, 115, 116, 97, 103, 101, 16, 233, 1, 34,
+        23, 50, 48, 50, 50, 45, 48, 57, 45, 49, 54, 32, 48, 48, 58, 48, 49, 58, 48, 50, 32, 85, 84,
+        67, 42, 37, 10, 12, 100, 97, 116, 97, 102, 117, 115, 101, 108, 97, 98, 115, 18, 15, 100,
+        97, 116, 97, 102, 117, 115, 101, 108, 97, 98, 115, 46, 114, 115, 160, 6, 8, 168, 6, 1, 160,
+        6, 8, 168, 6, 1,
+    ];
+
+    let dt = NaiveDateTime::new(
+        NaiveDate::from_ymd(2022, 9, 16),
+        NaiveTime::from_hms(0, 1, 2),
+    );
+    let user_id = mt::UserIdentity::new("datafuselabs", "datafuselabs.rs");
+    let want = mt::StageFile {
+        path: "/path/to/stage".to_string(),
+        size: 233,
+        md5: None,
+        last_modified: DateTime::from_utc(dt, Utc),
+        creator: Some(user_id),
+        ..Default::default()
+    };
+
+    common::test_load_old(func_name!(), stage_file_v7.as_slice(), want)?;
+    Ok(())
+}
+
 #[test]
 fn test_user_stage_fs_v6() -> anyhow::Result<()> {
     // Encoded data of version 6 of user_stage_fs:

src/meta/protos/proto/user.proto

@@ -222,4 +222,5 @@ message StageFile {
   optional string md5 = 3;
   string last_modified = 4;
   optional UserIdentity creator = 5;
+  optional string etag = 6;
 }

src/meta/types/src/user_stage.rs

@@ -258,11 +258,12 @@ impl UserStageInfo {
     }
 }
 
-#[derive(Default, Debug, Clone)]
+#[derive(Default, Debug, Clone, PartialEq, Eq)]
 pub struct StageFile {
     pub path: String,
     pub size: u64,
     pub md5: Option<String>,
     pub last_modified: DateTime<Utc>,
     pub creator: Option<UserIdentity>,
+    pub etag: Option<String>,
 }

src/query/service/src/interpreters/interpreter_common.rs

@@ -232,6 +232,7 @@ pub async fn list_files_from_dal(
                 .last_modified()
                 .map_or(Utc::now(), |t| Utc.timestamp(t.unix_timestamp(), 0)),
             creator: None,
+            etag: meta.etag().map(str::to_string),
         })
         .collect::<Vec<StageFile>>();