fix(rust): read only checkpoints that match _last_checkpoint version

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

@@ -449,7 +449,7 @@ async fn list_log_files_with_checkpoint(
     let checkpoint_files = files
         .iter()
         .filter_map(|f| {
-            if f.location.is_checkpoint_file() {
+            if f.location.is_checkpoint_file() && f.location.commit_version() == Some(cp.version) {
                 Some(f.clone())
             } else {
                 None

crates/test/tests/data/table_failed_last_checkpoint_update/_delta_log/00000000000000000000.json

@@ -0,0 +1,4 @@
+{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
+{"metaData":{"id":"98c9faeb-7940-43eb-9898-50b2a99c0a7e","name":null,"description":null,"format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"price\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}},{\"name\":\"sold\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"deleted\",\"type\":\"boolean\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"createdTime":1709986334419,"configuration":{}}}
+{"add":{"path":"part-00001-9c90a84d-6999-463c-bd2d-f68333e6d03d-c000.snappy.parquet","partitionValues":{},"size":1432,"modificationTime":1709986334424,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"deleted\":false,\"id\":\"1\",\"sold\":0,\"price\":0},\"maxValues\":{\"price\":4,\"id\":\"5\",\"deleted\":false,\"sold\":4},\"nullCount\":{\"deleted\":0,\"id\":0,\"price\":0,\"sold\":0}}","tags":null,"deletionVector":null,"baseRowId":null,"defaultRowCommitVersion":null,"clusteringProvider":null}}
+{"commitInfo":{"timestamp":1709986334424,"operation":"WRITE","operationParameters":{"mode":"Append"},"clientVersion":"delta-rs.0.17.1"}}

crates/test/tests/data/table_failed_last_checkpoint_update/_delta_log/00000000000000000001.json

@@ -0,0 +1,2 @@
+{"add":{"path":"part-00001-fed6d112-d244-4c54-810d-25ba3f0a4016-c000.snappy.parquet","partitionValues":{},"size":1432,"modificationTime":1709986334474,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"deleted\":false,\"price\":0,\"id\":\"1\",\"sold\":0},\"maxValues\":{\"price\":4,\"id\":\"5\",\"sold\":4,\"deleted\":false},\"nullCount\":{\"id\":0,\"sold\":0,\"deleted\":0,\"price\":0}}","tags":null,"deletionVector":null,"baseRowId":null,"defaultRowCommitVersion":null,"clusteringProvider":null}}
+{"commitInfo":{"timestamp":1709986334475,"operation":"WRITE","operationParameters":{"mode":"Append"},"clientVersion":"delta-rs.0.17.1"}}

crates/test/tests/data/table_failed_last_checkpoint_update/_delta_log/00000000000000000002.json

@@ -0,0 +1,2 @@
+{"add":{"path":"part-00001-6791b37e-f318-4d2b-87a0-89be205c338b-c000.snappy.parquet","partitionValues":{},"size":1432,"modificationTime":1709986423857,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"deleted\":false,\"price\":0,\"id\":\"1\",\"sold\":0},\"maxValues\":{\"sold\":4,\"id\":\"5\",\"price\":4,\"deleted\":false},\"nullCount\":{\"id\":0,\"price\":0,\"sold\":0,\"deleted\":0}}","tags":null,"deletionVector":null,"baseRowId":null,"defaultRowCommitVersion":null,"clusteringProvider":null}}
+{"commitInfo":{"timestamp":1709986423857,"operation":"WRITE","operationParameters":{"mode":"Append"},"clientVersion":"delta-rs.0.17.1"}}

crates/test/tests/data/table_failed_last_checkpoint_update/_delta_log/00000000000000000003.json

@@ -0,0 +1,2 @@
+{"add":{"path":"part-00001-bea93a33-9112-41a5-aca6-c2d1f2c43873-c000.snappy.parquet","partitionValues":{},"size":1432,"modificationTime":1709986423962,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"id\":\"1\",\"sold\":0,\"deleted\":false,\"price\":0},\"maxValues\":{\"deleted\":false,\"id\":\"5\",\"price\":4,\"sold\":4},\"nullCount\":{\"id\":0,\"price\":0,\"sold\":0,\"deleted\":0}}","tags":null,"deletionVector":null,"baseRowId":null,"defaultRowCommitVersion":null,"clusteringProvider":null}}
+{"commitInfo":{"timestamp":1709986423962,"operation":"WRITE","operationParameters":{"mode":"Append"},"clientVersion":"delta-rs.0.17.1"}}

crates/test/tests/data/table_failed_last_checkpoint_update/_delta_log/_last_checkpoint

@@ -0,0 +1 @@
+{"size":6,"size_in_bytes":23521,"version":1}

python/tests/test_table_read.py

@@ -740,3 +740,9 @@ def test_encode_partition_value(input_value: Any, expected: str) -> None:
         assert [encode_partition_value(val) for val in input_value] == expected
     else:
         assert encode_partition_value(input_value) == expected
+
+
+def test_read_table_last_checkpoint_not_updated():
+    dt = DeltaTable("../crates/test/tests/data/table_failed_last_checkpoint_update")
+
+    assert dt.version() == 3