Add base representation of storage files

Author: ashvina

xtable-api/src/main/java/org/apache/xtable/model/TableChange.java

@@ -22,7 +22,7 @@
 import lombok.NonNull;
 import lombok.Value;
 
-import org.apache.xtable.model.storage.DataFilesDiff;
+import org.apache.xtable.model.storage.InternalFilesDiff;
 
 /**
  * Captures the changes in a single commit/instant from the source table.
@@ -33,7 +33,7 @@
 @Builder(toBuilder = true)
 public class TableChange {
   // Change in files at the specified instant
-  DataFilesDiff filesDiff;
+  InternalFilesDiff filesDiff;
 
   /** The {@link InternalTable} at the commit time to which this table change belongs. */
   InternalTable tableAsOfChange;

xtable-api/src/main/java/org/apache/xtable/model/storage/FilesDiff.java

@@ -92,12 +92,12 @@ public static <L, P> FilesDiff<L, P> findNewAndRemovedFiles(
    * @param <P> the type of the previous files
    * @return the set of files that are added
    */
-  public static <P> FilesDiff<InternalDataFile, P> findNewAndRemovedFiles(
+  public static <P> FilesDiff<InternalFile, P> findNewAndRemovedFiles(
       List<PartitionFileGroup> latestFileGroups, Map<String, P> previousFiles) {
-    Map<String, InternalDataFile> latestFiles =
+    Map<String, InternalFile> latestFiles =
         latestFileGroups.stream()
             .flatMap(group -> group.getFiles().stream())
-            .collect(Collectors.toMap(InternalDataFile::getPhysicalPath, Function.identity()));
+            .collect(Collectors.toMap(InternalFile::getPhysicalPath, Function.identity()));
     return findNewAndRemovedFiles(latestFiles, previousFiles);
   }
 }

xtable-api/src/main/java/org/apache/xtable/model/storage/InternalDataFile.java

@@ -22,8 +22,12 @@
 import java.util.List;
 
 import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
 import lombok.NonNull;
-import lombok.Value;
+import lombok.ToString;
+import lombok.experimental.FieldDefaults;
+import lombok.experimental.SuperBuilder;
 
 import org.apache.xtable.model.stat.ColumnStat;
 import org.apache.xtable.model.stat.PartitionValue;
@@ -33,18 +37,17 @@
  *
  * @since 0.1
  */
-@Builder(toBuilder = true)
-@Value
-public class InternalDataFile {
-  // physical path of the file (absolute with scheme)
-  @NonNull String physicalPath;
+@SuperBuilder(toBuilder = true)
+@FieldDefaults(makeFinal = true, level = lombok.AccessLevel.PRIVATE)
+@ToString(callSuper = true)
+@EqualsAndHashCode(callSuper = true)
+@Getter
+public class InternalDataFile extends InternalFile {
   // file format
   @Builder.Default @NonNull FileFormat fileFormat = FileFormat.APACHE_PARQUET;
   // partition ranges for the data file
   @Builder.Default @NonNull List<PartitionValue> partitionValues = Collections.emptyList();
 
-  long fileSizeBytes;
-  long recordCount;
   // column stats for each column in the data file
   @Builder.Default @NonNull List<ColumnStat> columnStats = Collections.emptyList();
   // last modified time in millis since epoch

xtable-api/src/main/java/org/apache/xtable/model/storage/InternalFile.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package org.apache.xtable.model.storage;
+
+import lombok.AllArgsConstructor;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NonNull;
+import lombok.ToString;
+import lombok.experimental.FieldDefaults;
+import lombok.experimental.SuperBuilder;
+
+/**
+ * This class is an internal representation of a logical storage file of a table and is the base
+ * class of all types of storage files. The most common type of storage file is a data-file, which
+ * contains the actual records of a table. Other examples of a storage file are positional delete
+ * files (containing ordinals of the records to be deleted from a data file), stat files, and index
+ * files. For completeness of the conversion process, XTable needs to recognize current storage
+ * files of a table, and also the storage files that are added and removed over time as the state of
+ * the table changes. Different table formats support different storage file types. This base file
+ * representation is generic and extensible and is needed to design stable interfaces.
+ */
+@SuperBuilder(toBuilder = true)
+@FieldDefaults(makeFinal = true, level = lombok.AccessLevel.PRIVATE)
+@ToString(callSuper = true)
+@EqualsAndHashCode
+@AllArgsConstructor
+@Getter
+public abstract class InternalFile {
+  // Absolute path of the storage file, with the scheme, that contains this logical file. Typically,
+  // one physical storage file contains only one base file, for e.g. a parquet data file. However,
+  // in some cases, one storage file can contain multiple logical storage files for optimizations.
+  @NonNull String physicalPath;
+
+  // The size of the logical file in the physical storage file.
+  long fileSizeBytes;
+
+  // The number of records in the storage file.
+  long recordCount;
+}

xtable-api/src/main/java/org/apache/xtable/model/storage/DataFilesDiff.java renamed to xtable-api/src/main/java/org/apache/xtable/model/storage/InternalFilesDiff.java

@@ -20,6 +20,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -31,17 +32,18 @@
 @Value
 @EqualsAndHashCode(callSuper = true)
 @SuperBuilder
-public class DataFilesDiff extends FilesDiff<InternalDataFile, InternalDataFile> {
+public class InternalFilesDiff extends FilesDiff<InternalFile, InternalFile> {
 
   /**
-   * Creates a DataFilesDiff from the list of files in the target table and the list of files in the
-   * source table.
+   * Creates a InternalFilesDiff from the list of files in the target table and the list of files in
+   * the source table.
    *
    * @param source list of files currently in the source table
    * @param target list of files currently in the target table
    * @return files that need to be added and removed for the target table match the source table
    */
-  public static DataFilesDiff from(List<InternalDataFile> source, List<InternalDataFile> target) {
+  public static InternalFilesDiff from(
+      List<InternalDataFile> source, List<InternalDataFile> target) {
     Map<String, InternalDataFile> targetPaths =
         target.stream()
             .collect(Collectors.toMap(InternalDataFile::getPhysicalPath, Function.identity()));
@@ -51,9 +53,35 @@ public static DataFilesDiff from(List<InternalDataFile> source, List<InternalDat
 
     FilesDiff<InternalDataFile, InternalDataFile> diff =
         findNewAndRemovedFiles(sourcePaths, targetPaths);
-    return DataFilesDiff.builder()
+    return InternalFilesDiff.builder()
         .filesAdded(diff.getFilesAdded())
         .filesRemoved(diff.getFilesRemoved())
         .build();
   }
+
+  /**
+   * Filters files of type {@link InternalDataFile} from the list of files added to the source table
+   * and returns the list.
+   */
+  public Set<InternalDataFile> dataFilesAdded() {
+    Set<InternalDataFile> result =
+        getFilesAdded().stream()
+            .filter(InternalDataFile.class::isInstance)
+            .map(file -> (InternalDataFile) file)
+            .collect(Collectors.toSet());
+    return result;
+  }
+
+  /**
+   * Filters files of type {@link InternalDataFile} from the list of files removed to the source
+   * table and returns the list.
+   */
+  public Set<InternalDataFile> dataFilesRemoved() {
+    Set<InternalDataFile> result =
+        getFilesRemoved().stream()
+            .filter(InternalDataFile.class::isInstance)
+            .map(file -> (InternalDataFile) file)
+            .collect(Collectors.toSet());
+    return result;
+  }
 }

xtable-api/src/main/java/org/apache/xtable/model/storage/PartitionFileGroup.java

@@ -28,12 +28,12 @@
 
 import org.apache.xtable.model.stat.PartitionValue;
 
-/** Represents a grouping of {@link InternalDataFile} with the same partition values. */
+/** Represents a grouping of {@link InternalFile} with the same partition values. */
 @Value
 @Builder
 public class PartitionFileGroup {
   List<PartitionValue> partitionValues;
-  List<InternalDataFile> files;
+  List<? extends InternalFile> files;
 
   public static List<PartitionFileGroup> fromFiles(List<InternalDataFile> files) {
     return fromFiles(files.stream());
@@ -51,4 +51,12 @@ public static List<PartitionFileGroup> fromFiles(Stream<InternalDataFile> files)
                     .build())
         .collect(Collectors.toList());
   }
+
+  /** Filters storage files of type {@link InternalDataFile} and returns them. */
+  public List<InternalDataFile> getDataFiles() {
+    return files.stream()
+        .filter(InternalDataFile.class::isInstance)
+        .map(file -> (InternalDataFile) file)
+        .collect(Collectors.toList());
+  }
 }

xtable-api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java

@@ -28,7 +28,7 @@
 import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.schema.InternalPartitionField;
 import org.apache.xtable.model.schema.InternalSchema;
-import org.apache.xtable.model.storage.DataFilesDiff;
+import org.apache.xtable.model.storage.InternalFilesDiff;
 import org.apache.xtable.model.storage.PartitionFileGroup;
 
 /** A client that provides the major functionality for syncing changes to a target system. */
@@ -68,9 +68,9 @@ public interface ConversionTarget {
    * Syncs the changes in files to the target system. This method is required to both add and remove
    * files.
    *
-   * @param dataFilesDiff the diff that needs to be synced
+   * @param internalFilesDiff the diff that needs to be synced
    */
-  void syncFilesForDiff(DataFilesDiff dataFilesDiff);
+  void syncFilesForDiff(InternalFilesDiff internalFilesDiff);
 
   /**
    * Starts the sync and performs any initialization required

xtable-api/src/test/java/org/apache/xtable/model/storage/TestDataFilesDiff.java

@@ -37,13 +37,13 @@ void testFrom() {
         InternalDataFile.builder().physicalPath("file://already_in_target2.parquet").build();
     InternalDataFile sourceFileInTargetAlready =
         InternalDataFile.builder().physicalPath("file://already_in_target3.parquet").build();
-    DataFilesDiff actual =
-        DataFilesDiff.from(
+    InternalFilesDiff actual =
+        InternalFilesDiff.from(
             Arrays.asList(sourceFile1, sourceFile2, sourceFileInTargetAlready),
             Arrays.asList(targetFile1, targetFile2, sourceFileInTargetAlready));
 
-    DataFilesDiff expected =
-        DataFilesDiff.builder()
+    InternalFilesDiff expected =
+        InternalFilesDiff.builder()
             .filesAdded(Arrays.asList(sourceFile1, sourceFile2))
             .filesRemoved(Arrays.asList(targetFile1, targetFile2))
             .build();

xtable-api/src/test/java/org/apache/xtable/model/storage/TestFilesDiff.java

@@ -50,7 +50,7 @@ void findDiffFromFileGroups() {
     previousFiles.put("file2NoGroup", file2);
     previousFiles.put("file2Group2", file3);
 
-    FilesDiff<InternalDataFile, File> diff =
+    FilesDiff<InternalFile, File> diff =
         FilesDiff.findNewAndRemovedFiles(latestFileGroups, previousFiles);
     assertEquals(2, diff.getFilesAdded().size());
     assertTrue(diff.getFilesAdded().contains(file1Group2));

xtable-api/src/test/java/org/apache/xtable/spi/extractor/TestExtractFromSource.java

@@ -41,8 +41,8 @@
 import org.apache.xtable.model.InternalSnapshot;
 import org.apache.xtable.model.InternalTable;
 import org.apache.xtable.model.TableChange;
-import org.apache.xtable.model.storage.DataFilesDiff;
 import org.apache.xtable.model.storage.InternalDataFile;
+import org.apache.xtable.model.storage.InternalFilesDiff;
 import org.apache.xtable.model.storage.PartitionFileGroup;
 
 public class TestExtractFromSource {
@@ -89,7 +89,7 @@ public void extractTableChanges() {
         TableChange.builder()
             .tableAsOfChange(tableAtFirstInstant)
             .filesDiff(
-                DataFilesDiff.builder().fileAdded(newFile1).fileRemoved(initialFile2).build())
+                InternalFilesDiff.builder().fileAdded(newFile1).fileRemoved(initialFile2).build())
             .sourceIdentifier("0")
             .build();
     when(mockConversionSource.getTableChangeForCommit(firstCommitToSync))
@@ -98,7 +98,7 @@ public void extractTableChanges() {
         TableChange.builder()
             .tableAsOfChange(tableAtFirstInstant)
             .filesDiff(
-                DataFilesDiff.builder().fileAdded(newFile1).fileRemoved(initialFile2).build())
+                InternalFilesDiff.builder().fileAdded(newFile1).fileRemoved(initialFile2).build())
             .sourceIdentifier("0")
             .build();
 
@@ -112,7 +112,7 @@ public void extractTableChanges() {
         TableChange.builder()
             .tableAsOfChange(tableAtSecondInstant)
             .filesDiff(
-                DataFilesDiff.builder()
+                InternalFilesDiff.builder()
                     .filesAdded(Arrays.asList(newFile2, newFile3))
                     .filesRemoved(Arrays.asList(initialFile3, newFile1))
                     .build())
@@ -124,7 +124,7 @@ public void extractTableChanges() {
         TableChange.builder()
             .tableAsOfChange(tableAtSecondInstant)
             .filesDiff(
-                DataFilesDiff.builder()
+                InternalFilesDiff.builder()
                     .filesAdded(Arrays.asList(newFile2, newFile3))
                     .filesRemoved(Arrays.asList(initialFile3, newFile1))
                     .build())