fix: Don't return errors from ParsedLogPath::try_from (#1433)

## What changes are proposed in this pull request?
This change effectively makes log path parsing infallible. We no longer
return errors from `ParsedLogPath::try_from`. This fixes #1432.

This _also_ turns any valid looking files that aren't in an expected
subdir to be of "Unknown" type. (either `_delta_log` for most everything
except `_staged_commits` for staged files)

As a follow-up we need to _actually_ just return an
`Option<ParsedLogPath<Location>>` from that method. Doing so requires
numerous tricky changes in our log listing code and I wanted to get a
more targeted fix in for the issue and then we can tackle the refactor
at a more leisurely pace. At that time I think we could also clean up
the `try_from` method logic to make it a bit more clear, but again I
didn't want to make large changes close to release.

## How was this change tested?
Modified existing unit tests, added a new test in `read.rs` that we can
read a table with a `0.zip` file in the log dir.

Author: nicklan

kernel/src/checkpoint/tests.rs

@@ -181,11 +181,11 @@ fn write_commit_to_store(
         .collect();
     let content = json_lines.join("\n");
 
-    let commit_path = format!("_delta_log/{}", delta_path_for_version(version, "json"));
+    let commit_path = delta_path_for_version(version, "json");
 
     tokio::runtime::Runtime::new()
         .expect("create tokio runtime")
-        .block_on(async { store.put(&Path::from(commit_path), content.into()).await })?;
+        .block_on(async { store.put(&commit_path, content.into()).await })?;
 
     Ok(())
 }

kernel/src/log_segment/tests.rs

@@ -1760,12 +1760,18 @@ fn test_debug_assert_listed_log_file_in_order_compaction_files() {
     let _ = ListedLogFiles::try_new(
         vec![],
         vec![
-            create_log_path("file:///00000000000000000000.00000000000000000004.compacted.json"),
-            create_log_path("file:///00000000000000000001.00000000000000000002.compacted.json"),
+            create_log_path(
+                "file:///_delta_log/00000000000000000000.00000000000000000004.compacted.json",
+            ),
+            create_log_path(
+                "file:///_delta_log/00000000000000000001.00000000000000000002.compacted.json",
+            ),
         ],
         vec![],
         None,
-        Some(create_log_path("file:///00000000000000000001.json")),
+        Some(create_log_path(
+            "file:///_delta_log/00000000000000000001.json",
+        )),
     );
 }
 
@@ -1776,12 +1782,18 @@ fn test_debug_assert_listed_log_file_out_of_order_compaction_files() {
     let _ = ListedLogFiles::try_new(
         vec![],
         vec![
-            create_log_path("file:///00000000000000000000.00000000000000000004.compacted.json"),
-            create_log_path("file:///00000000000000000000.00000000000000000003.compacted.json"),
+            create_log_path(
+                "file:///_delta_log/00000000000000000000.00000000000000000004.compacted.json",
+            ),
+            create_log_path(
+                "file:///_delta_log/00000000000000000000.00000000000000000003.compacted.json",
+            ),
         ],
         vec![],
         None,
-        Some(create_log_path("file:///00000000000000000001.json")),
+        Some(create_log_path(
+            "file:///_delta_log/00000000000000000001.json",
+        )),
     );
 }
 
@@ -1793,11 +1805,17 @@ fn test_debug_assert_listed_log_file_different_multipart_checkpoint_versions() {
         vec![],
         vec![],
         vec![
-            create_log_path("00000000000000000010.checkpoint.0000000001.0000000002.parquet"),
-            create_log_path("00000000000000000011.checkpoint.0000000002.0000000002.parquet"),
+            create_log_path(
+                "file:///_delta_log/00000000000000000010.checkpoint.0000000001.0000000002.parquet",
+            ),
+            create_log_path(
+                "file:///_delta_log/00000000000000000011.checkpoint.0000000002.0000000002.parquet",
+            ),
         ],
         None,
-        Some(create_log_path("file:///00000000000000000001.json")),
+        Some(create_log_path(
+            "file:///_delta_log/00000000000000000001.json",
+        )),
     );
 }
 
@@ -1809,11 +1827,17 @@ fn test_debug_assert_listed_log_file_invalid_multipart_checkpoint() {
         vec![],
         vec![],
         vec![
-            create_log_path("00000000000000000010.checkpoint.0000000001.0000000003.parquet"),
-            create_log_path("00000000000000000011.checkpoint.0000000002.0000000003.parquet"),
+            create_log_path(
+                "file:///_delta_log/00000000000000000010.checkpoint.0000000001.0000000003.parquet",
+            ),
+            create_log_path(
+                "file:///_delta_log/00000000000000000011.checkpoint.0000000002.0000000003.parquet",
+            ),
         ],
         None,
-        Some(create_log_path("file:///00000000000000000001.json")),
+        Some(create_log_path(
+            "file:///_delta_log/00000000000000000001.json",
+        )),
     );
 }
 
@@ -2187,27 +2211,31 @@ fn test_latest_commit_file_edge_case_commit_before_checkpoint() {
 fn test_log_segment_contiguous_commit_files() {
     let res = ListedLogFiles::try_new(
         vec![
-            create_log_path("file:///00000000000000000001.json"),
-            create_log_path("file:///00000000000000000002.json"),
-            create_log_path("file:///00000000000000000003.json"),
+            create_log_path("file:///_delta_log/00000000000000000001.json"),
+            create_log_path("file:///_delta_log/00000000000000000002.json"),
+            create_log_path("file:///_delta_log/00000000000000000003.json"),
         ],
         vec![],
         vec![],
         None,
-        Some(create_log_path("file:///00000000000000000001.json")),
+        Some(create_log_path(
+            "file:///_delta_log/00000000000000000001.json",
+        )),
     );
     assert!(res.is_ok());
 
     // allow gaps in ListedLogFiles
     let listed = ListedLogFiles::try_new(
         vec![
-            create_log_path("file:///00000000000000000001.json"),
-            create_log_path("file:///00000000000000000003.json"),
+            create_log_path("file:///_delta_log/00000000000000000001.json"),
+            create_log_path("file:///_delta_log/00000000000000000003.json"),
         ],
         vec![],
         vec![],
         None,
-        Some(create_log_path("file:///00000000000000000001.json")),
+        Some(create_log_path(
+            "file:///_delta_log/00000000000000000001.json",
+        )),
     );
 
     // disallow gaps in LogSegment
@@ -2217,11 +2245,11 @@ fn test_log_segment_contiguous_commit_files() {
         "Generic delta kernel error: Expected ordered \
         contiguous commit files [ParsedLogPath { location: FileMeta { location: Url { scheme: \
         \"file\", cannot_be_a_base: false, username: \"\", password: None, host: None, port: \
-        None, path: \"/00000000000000000001.json\", query: None, fragment: None }, last_modified: \
+        None, path: \"/_delta_log/00000000000000000001.json\", query: None, fragment: None }, last_modified: \
         0, size: 0 }, filename: \"00000000000000000001.json\", extension: \"json\", version: 1, \
         file_type: Commit }, ParsedLogPath { location: FileMeta { location: Url { scheme: \
         \"file\", cannot_be_a_base: false, username: \"\", password: None, host: None, port: \
-        None, path: \"/00000000000000000003.json\", query: None, fragment: None }, last_modified: \
+        None, path: \"/_delta_log/00000000000000000003.json\", query: None, fragment: None }, last_modified: \
         0, size: 0 }, filename: \"00000000000000000003.json\", extension: \"json\", version: 3, \
         file_type: Commit }]",
     );