[Kernel] Load the protocol and metadata from the CRC files when available

kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/FileNames.java

@@ -52,6 +52,9 @@ private FileNames() {}
   private static final Pattern CLASSIC_CHECKPOINT_FILE_PATTERN =
       Pattern.compile("\\d+\\.checkpoint\\.parquet");
 
+    /** Example: 00000000000000000001.crc */
+    private static final Pattern CHECK_SUM_FILE_PATTERN = Pattern.compile("(\\d+)\\.crc")
+
   /**
    * Examples:
    *
@@ -69,11 +72,27 @@ private FileNames() {}
 
   public static final String SIDECAR_DIRECTORY = "_sidecars";
 
-  private static final Pattern checksumFileRegex = Pattern.compile("(\\d+)\\.crc");
+  ////////////////////////
+  // Version extractors //
+  ////////////////////////
 
-  /** Returns the delta (json format) path for a given delta file. */
-  public static String deltaFile(Path path, long version) {
-    return String.format("%s/%020d.json", path, version);
+  /**
+   * Get the version of the checkpoint, checksum or delta file. Throws an error if an unexpected
+   * file type is seen. These unexpected files should be filtered out to ensure forward
+   * compatibility in cases where new file types are added, but without an explicit protocol
+   * upgrade.
+   */
+  public static long getFileVersion(Path path) {
+    if (isCheckpointFile(path.getName())) {
+      return checkpointVersion(path);
+    } else if (isCommitFile(path.getName())) {
+      return deltaVersion(path);
+       } else if (isChecksumFile(path.getName())) {
+       return   checksumVersion(path);
+    } else {
+      throw new IllegalArgumentException(
+          String.format("Unexpected file type found in transaction log: %s", path));
+    }
   }
 
   /** Returns the version for the given delta path. */
@@ -152,7 +171,7 @@ public static Path topLevelV2CheckpointFile(
 
   /** Returns the path for a V2 sidecar file with a given UUID. */
   public static Path v2CheckpointSidecarFile(Path path, String uuid) {
-    return new Path(String.format("%s/%s/%s.parquet", path.toString(), SIDECAR_DIRECTORY, uuid));
+    return new Path(String.format("%s/_sidecars/%s.parquet", path.toString(), uuid));
   }
 
   /**
@@ -194,32 +213,13 @@ public static boolean isV2CheckpointFile(String path) {
     return V2_CHECKPOINT_FILE_PATTERN.matcher(new Path(path).getName()).matches();
   }
 
-  public static boolean isCommitFile(String fileName) {
-    String filename = new Path(fileName).getName();
-    return DELTA_FILE_PATTERN.matcher(filename).matches()
-        || UUID_DELTA_FILE_REGEX.matcher(filename).matches();
-  }
-
-  public static boolean isChecksumFile(String checksumFilePath) {
-    return checksumFileRegex.matcher(new Path(checksumFilePath).getName()).matches();
+  public static boolean isCommitFile(String path) {
+    final String fileName = new Path(path).getName();
+    return DELTA_FILE_PATTERN.matcher(fileName).matches()
+        || UUID_DELTA_FILE_REGEX.matcher(fileName).matches();
   }
 
-  /**
-   * Get the version of the checkpoint, checksum or delta file. Throws an error if an unexpected
-   * file type is seen. These unexpected files should be filtered out to ensure forward
-   * compatibility in cases where new file types are added, but without an explicit protocol
-   * upgrade.
-   */
-  public static long getFileVersion(Path path) {
-    if (isCheckpointFile(path.getName())) {
-      return checkpointVersion(path);
-    } else if (isCommitFile(path.getName())) {
-      return deltaVersion(path);
-    } else if (isChecksumFile(path.getName())) {
-      return checksumVersion(path);
-    } else {
-      throw new IllegalArgumentException(
-          String.format("Unexpected file type found in transaction log: %s", path));
+    public static boolean isChecksumFile(String checksumFilePath) {
+        return CHECK_SUM_FILE_PATTERN.matcher(new Path(checksumFilePath).getName()).matches();
     }
-  }
 }