Merge pull request #7652 from ClSlaid/add-etag-to-stage-file

feat(query): Add etag to stage file

Author: mergify[bot]

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();
@@ -158,6 +179,16 @@ fn test_user_pb_from_to() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[test]
+fn test_stage_file_pb_from_to() -> anyhow::Result<()> {
+    let test_stage_file = test_stage_file();
+    let test_stage_file_pb = test_stage_file.to_pb()?;
+    let got = mt::StageFile::from_pb(test_stage_file_pb)?;
+    assert_eq!(got, test_stage_file);
+
+    Ok(())
+}
+
 #[test]
 fn test_user_incompatible() -> anyhow::Result<()> {
     {
@@ -179,6 +210,25 @@ fn test_user_incompatible() -> anyhow::Result<()> {
         );
     }
 
+    {
+        let stage_file = test_stage_file();
+        let mut p = stage_file.to_pb()?;
+        p.ver = VER + 1;
+        p.min_compatible = VER + 1;
+
+        let res = mt::StageFile::from_pb(p);
+        assert_eq!(
+            Incompatible {
+                reason: format!(
+                    "executable ver={} is smaller than the message min compatible ver: {}",
+                    VER,
+                    VER + 1
+                )
+            },
+            res.unwrap_err()
+        )
+    }
+
     {
         let fs_stage_info = test_fs_stage_info();
         let mut p = fs_stage_info.to_pb()?;
@@ -250,7 +300,16 @@ fn test_build_user_pb_buf() -> anyhow::Result<()> {
 
         let mut buf = vec![];
         common_protos::prost::Message::encode(&p, &mut buf)?;
-        println!("{:?}", buf);
+        println!("user_info: {:?}", 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!("stage_file: {:?}", buf);
     }
 
     // Stage on local file system
@@ -261,7 +320,7 @@ fn test_build_user_pb_buf() -> anyhow::Result<()> {
 
         let mut buf = vec![];
         common_protos::prost::Message::encode(&p, &mut buf)?;
-        println!("{:?}", buf);
+        println!("fs_stage_info: {:?}", buf);
     }
 
     // Stage on S3
@@ -272,7 +331,7 @@ fn test_build_user_pb_buf() -> anyhow::Result<()> {
 
         let mut buf = vec![];
         common_protos::prost::Message::encode(&p, &mut buf)?;
-        println!("{:?}", buf);
+        println!("s3_stage_info: {:?}", buf);
     }
 
     // Stage on GCS, supported in version >=4.
@@ -281,7 +340,7 @@ fn test_build_user_pb_buf() -> anyhow::Result<()> {
         let p = gcs_stage_info.to_pb()?;
         let mut buf = vec![];
         common_protos::prost::Message::encode(&p, &mut buf)?;
-        println!("{:?}", buf);
+        println!("gcs_stage_info: {:?}", buf);
     }
 
     Ok(())
@@ -360,3 +419,39 @@ fn test_load_old_user() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[test]
+fn test_old_stage_file() -> anyhow::Result<()> {
+    // Encoded data of version 7 of StageFile:
+    // Generated with `test_build_user_pb_buf`
+    {
+        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 p: pb::StageFile =
+            common_protos::prost::Message::decode(stage_file_v7.as_slice()).map_err(print_err)?;
+        let got = mt::StageFile::from_pb(p).map_err(print_err)?;
+
+        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()
+        };
+
+        assert_eq!(got, want);
+    }
+
+    Ok(())
+}

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>>();