fix. check for all known valid delta files in is_deltatable

Signed-off-by: Martin Andersson <u.martin.andersson@gmail.com>

Author: umartin

crates/core/src/kernel/snapshot/log_segment.rs

@@ -28,6 +28,12 @@ static CHECKPOINT_FILE_PATTERN: LazyLock<Regex> =
 static DELTA_FILE_PATTERN: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"^\d+\.json$").unwrap());
 static CRC_FILE_PATTERN: LazyLock<Regex> =
     LazyLock::new(|| Regex::new(r"^(\.\d+(\.crc|\.json)|\d+)\.crc$").unwrap());
+static LAST_CHECKPOINT_FILE_PATTERN: LazyLock<Regex> =
+    LazyLock::new(|| Regex::new(r"^_last_checkpoint$").unwrap());
+static LAST_VACUUM_INFO_FILE_PATTERN: LazyLock<Regex> =
+    LazyLock::new(|| Regex::new(r"^_last_vacuum_info$").unwrap());
+static DELETION_VECTOR_FILE_PATTERN: LazyLock<Regex> =
+    LazyLock::new(|| Regex::new(r".*\.bin$").unwrap());
 pub(super) static TOMBSTONE_SCHEMA: LazyLock<StructType> =
     LazyLock::new(|| StructType::new(vec![ActionType::Remove.schema_field().clone()]));
 
@@ -66,6 +72,24 @@ pub(crate) trait PathExt {
             .map(|name| CRC_FILE_PATTERN.captures(name).is_some())
             .unwrap()
     }
+
+    fn is_last_checkpoint_file(&self) -> bool {
+        self.filename()
+            .map(|name| LAST_CHECKPOINT_FILE_PATTERN.captures(name).is_some())
+            .unwrap_or(false)
+    }
+
+    fn is_last_vacuum_info_file(&self) -> bool {
+        self.filename()
+            .map(|name| LAST_VACUUM_INFO_FILE_PATTERN.captures(name).is_some())
+            .unwrap_or(false)
+    }
+
+    fn is_deletion_vector_file(&self) -> bool {
+        self.filename()
+            .map(|name| DELETION_VECTOR_FILE_PATTERN.captures(name).is_some())
+            .unwrap_or(false)
+    }
 }
 
 impl PathExt for Path {

crates/core/src/logstore/mod.rs

@@ -267,11 +267,23 @@ pub trait LogStore: Send + Sync + AsAny {
         while let Some(res) = stream.next().await {
             match res {
                 Ok(meta) => {
-                    // crc files are valid files according to the protocol
-                    if meta.location.is_crc_file() {
+                    // Valid but optional files.
+                    if meta.location.is_crc_file()
+                        || meta.location.is_last_checkpoint_file()
+                        || meta.location.is_last_vacuum_info_file()
+                        || meta.location.is_deletion_vector_file()
+                    {
                         continue;
                     }
-                    return Ok(meta.location.is_commit_file() || meta.location.is_checkpoint_file());
+                    let is_valid =
+                        meta.location.is_commit_file() || meta.location.is_checkpoint_file();
+                    if !is_valid {
+                        warn!(
+                            "Expected a valid delta file. Found {}",
+                            meta.location.filename().unwrap_or("<empty>")
+                        )
+                    }
+                    return Ok(is_valid);
                 }
                 Err(ObjectStoreError::NotFound { .. }) => return Ok(false),
                 Err(err) => return Err(err.into()),