Parquet Incremental Sync

xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java

@@ -18,26 +18,20 @@
 
 package org.apache.xtable.parquet;
 
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.nio.file.Paths;
 import java.time.Instant;
 import java.util.*;
-import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import lombok.Builder;
 import lombok.NonNull;
+import lombok.extern.log4j.Log4j2;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.*;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.util.functional.RemoteIterators;
 import org.apache.parquet.hadoop.metadata.ParquetMetadata;
 import org.apache.parquet.schema.MessageType;
 
-import org.apache.xtable.exception.ReadException;
 import org.apache.xtable.hudi.*;
 import org.apache.xtable.hudi.HudiPathUtils;
 import org.apache.xtable.model.*;
@@ -52,10 +46,15 @@
 import org.apache.xtable.spi.extractor.ConversionSource;
 
 @Builder
+@Log4j2
 public class ParquetConversionSource implements ConversionSource<Long> {
 
   private static final ParquetSchemaExtractor schemaExtractor =
       ParquetSchemaExtractor.getInstance();
+  private static final ParquetMetadataExtractor metadataExtractor =
+      ParquetMetadataExtractor.getInstance();
+  private static final ParquetDataManager parquetDataManagerExtractor =
+      ParquetDataManager.getInstance();
 
   private static final ParquetMetadataExtractor parquetMetadataExtractor =
       ParquetMetadataExtractor.getInstance();
@@ -69,7 +68,17 @@ public class ParquetConversionSource implements ConversionSource<Long> {
   private final String basePath;
   @NonNull private final Configuration hadoopConf;
 
-  private InternalTable createInternalTableFromFile(LocatedFileStatus latestFile) {
+  private List<LocatedFileStatus> parquetFiles;
+
+  // helper method to ensure files are loaded once
+  private List<LocatedFileStatus> getOrLoadParquetFiles() {
+    if (this.parquetFiles == null) {
+      this.parquetFiles = parquetDataManagerExtractor.getParquetFiles(hadoopConf, basePath);
+    }
+    return this.parquetFiles;
+  }
+
+  private InternalTable createInternalTableFromFile(ParquetFileConfig latestFile) {
     ParquetMetadata parquetMetadata =
         parquetMetadataExtractor.readParquetMetadata(hadoopConf, latestFile.getPath());
     MessageType parquetSchema = parquetMetadataExtractor.getSchema(parquetMetadata);
@@ -94,49 +103,45 @@ private InternalTable createInternalTableFromFile(LocatedFileStatus latestFile)
   @Override
   public InternalTable getTable(Long modificationTime) {
     // get parquetFile at specific time modificationTime
-    Stream<LocatedFileStatus> parquetFiles = getParquetFiles(hadoopConf, basePath);
-    LocatedFileStatus file = getParquetFileAt(parquetFiles, modificationTime);
+    LocatedFileStatus parquetFile =
+        parquetDataManagerExtractor.getParquetDataFileAt(getOrLoadParquetFiles(), modificationTime);
+    ParquetFileConfig file = parquetDataManagerExtractor.getConfigFromFile(parquetFile, hadoopConf);
     return createInternalTableFromFile(file);
   }
 
-  private Stream<InternalDataFile> getInternalDataFiles(Stream<LocatedFileStatus> parquetFiles) {
+  private Stream<InternalDataFile> getInternalDataFiles(Stream<ParquetFileConfig> parquetFiles) {
     return parquetFiles.map(
-        file ->
-            InternalDataFile.builder()
-                .physicalPath(file.getPath().toString())
-                .fileFormat(FileFormat.APACHE_PARQUET)
-                .fileSizeBytes(file.getLen())
-                .partitionValues(
-                    partitionValueExtractor.extractPartitionValues(
-                        partitionSpecExtractor.spec(
-                            partitionValueExtractor.extractSchemaForParquetPartitions(
-                                parquetMetadataExtractor.readParquetMetadata(
-                                    hadoopConf, file.getPath()),
-                                file.getPath().toString())),
-                        HudiPathUtils.getPartitionPath(new Path(basePath), file.getPath())))
-                .lastModified(file.getModificationTime())
-                .columnStats(
-                    parquetStatsExtractor.getColumnStatsForaFile(
-                        parquetMetadataExtractor.readParquetMetadata(hadoopConf, file.getPath())))
-                .build());
+        file -> {
+          ParquetMetadata metadata =
+              parquetMetadataExtractor.readParquetMetadata(hadoopConf, file.getPath());
+          return InternalDataFile.builder()
+              .physicalPath(file.getPath().toString())
+              .fileFormat(FileFormat.APACHE_PARQUET)
+              .fileSizeBytes(file.getSize())
+              .partitionValues(
+                  partitionValueExtractor.extractPartitionValues(
+                      partitionSpecExtractor.spec(
+                          partitionValueExtractor.extractSchemaForParquetPartitions(
+                              metadata, file.getPath().toString())),
+                      HudiPathUtils.getPartitionPath(new Path(basePath), file.getPath())))
+              .lastModified(file.getModificationTime())
+              .columnStats(parquetStatsExtractor.getColumnStatsForaFile(metadata))
+              .build();
+        });
   }
 
-  private InternalDataFile createInternalDataFileFromParquetFile(FileStatus parquetFile) {
+  private InternalDataFile createInternalDataFileFromParquetFile(ParquetFileConfig parquetFile) {
     return InternalDataFile.builder()
         .physicalPath(parquetFile.getPath().toString())
         .partitionValues(
             partitionValueExtractor.extractPartitionValues(
                 partitionSpecExtractor.spec(
                     partitionValueExtractor.extractSchemaForParquetPartitions(
-                        parquetMetadataExtractor.readParquetMetadata(
-                            hadoopConf, parquetFile.getPath()),
-                        parquetFile.getPath().toString())),
-                basePath))
+                        parquetFile.getMetadata(), parquetFile.getPath().toString())),
+                HudiPathUtils.getPartitionPath(new Path(basePath), parquetFile.getPath())))
         .lastModified(parquetFile.getModificationTime())
-        .fileSizeBytes(parquetFile.getLen())
-        .columnStats(
-            parquetStatsExtractor.getColumnStatsForaFile(
-                parquetMetadataExtractor.readParquetMetadata(hadoopConf, parquetFile.getPath())))
+        .fileSizeBytes(parquetFile.getSize())
+        .columnStats(parquetStatsExtractor.getColumnStatsForaFile(parquetFile.getMetadata()))
         .build();
   }
 
@@ -149,87 +154,103 @@ public CommitsBacklog<Long> getCommitsBacklog(InstantsForIncrementalSync syncIns
 
   @Override
   public TableChange getTableChangeForCommit(Long modificationTime) {
-    Stream<LocatedFileStatus> parquetFiles = getParquetFiles(hadoopConf, basePath);
     Set<InternalDataFile> addedInternalDataFiles = new HashSet<>();
 
-    List<FileStatus> tableChangesAfter =
-        parquetFiles
-            .filter(fileStatus -> fileStatus.getModificationTime() > modificationTime)
-            .collect(Collectors.toList());
-    InternalTable internalTable = getMostRecentTable(parquetFiles);
-    for (FileStatus tableStatus : tableChangesAfter) {
-      InternalDataFile currentDataFile = createInternalDataFileFromParquetFile(tableStatus);
+    List<ParquetFileConfig> filesMetadata =
+        parquetDataManagerExtractor.getParquetFilesMetadataAfterTime(
+            hadoopConf, getOrLoadParquetFiles(), modificationTime);
+    List<ParquetFileConfig> tableChangesAfterMetadata =
+        parquetDataManagerExtractor.getParquetFilesMetadataAfterTime(
+            hadoopConf, getOrLoadParquetFiles(), modificationTime);
+    InternalTable internalTable = getMostRecentTableConfig(tableChangesAfterMetadata.stream());
+    for (ParquetFileConfig fileMetadata : filesMetadata) {
+      InternalDataFile currentDataFile = createInternalDataFileFromParquetFile(fileMetadata);
       addedInternalDataFiles.add(currentDataFile);
     }
 
     return TableChange.builder()
+        .sourceIdentifier(
+            getCommitIdentifier(
+                parquetDataManagerExtractor
+                    .getMostRecentParquetFile(getOrLoadParquetFiles(), hadoopConf)
+                    .getModificationTime()))
         .tableAsOfChange(internalTable)
         .filesDiff(InternalFilesDiff.builder().filesAdded(addedInternalDataFiles).build())
         .build();
   }
 
-  private InternalTable getMostRecentTable(Stream<LocatedFileStatus> parquetFiles) {
-    LocatedFileStatus latestFile = getMostRecentParquetFile(parquetFiles);
+  private InternalTable getMostRecentTable(
+      List<LocatedFileStatus> parquetFiles, Configuration conf) {
+    ParquetFileConfig latestFile =
+        parquetDataManagerExtractor.getMostRecentParquetFile(parquetFiles, conf);
+    return createInternalTableFromFile(latestFile);
+  }
+
+  private InternalTable getMostRecentTableConfig(Stream<ParquetFileConfig> parquetFiles) {
+    ParquetFileConfig latestFile =
+        parquetDataManagerExtractor.getMostRecentParquetFileConfig(parquetFiles);
     return createInternalTableFromFile(latestFile);
   }
 
   @Override
   public InternalTable getCurrentTable() {
-    Stream<LocatedFileStatus> parquetFiles = getParquetFiles(hadoopConf, basePath);
-    return getMostRecentTable(parquetFiles);
+    return getMostRecentTable(getOrLoadParquetFiles(), hadoopConf);
   }
 
-  /**
-   * get current snapshot
-   *
-   * @return
-   */
   @Override
   public InternalSnapshot getCurrentSnapshot() {
-    // to avoid consume the stream call the method twice to return the same stream of parquet files
     Stream<InternalDataFile> internalDataFiles =
-        getInternalDataFiles(getParquetFiles(hadoopConf, basePath));
-    InternalTable table = getMostRecentTable(getParquetFiles(hadoopConf, basePath));
+        getInternalDataFiles(
+            parquetDataManagerExtractor.getConfigsFromStream(getOrLoadParquetFiles(), hadoopConf));
+    InternalTable table = getMostRecentTable(getOrLoadParquetFiles(), hadoopConf);
     return InternalSnapshot.builder()
         .table(table)
         .sourceIdentifier(
             getCommitIdentifier(
-                getMostRecentParquetFile(getParquetFiles(hadoopConf, basePath))
+                parquetDataManagerExtractor
+                    .getMostRecentParquetFile(getOrLoadParquetFiles(), hadoopConf)
                     .getModificationTime()))
         .partitionedDataFiles(PartitionFileGroup.fromFiles(internalDataFiles))
         .build();
   }
 
-  private LocatedFileStatus getMostRecentParquetFile(Stream<LocatedFileStatus> parquetFiles) {
-    return parquetFiles
-        .max(Comparator.comparing(FileStatus::getModificationTime))
-        .orElseThrow(() -> new IllegalStateException("No files found"));
-  }
+  @Override
+  public boolean isIncrementalSyncSafeFrom(Instant timeInMillis) {
+    Stream<ParquetFileConfig> parquetFilesMetadata =
+        parquetDataManagerExtractor.getConfigsFromStream(getOrLoadParquetFiles(), hadoopConf);
+    LongSummaryStatistics stats =
+        parquetFilesMetadata.mapToLong(ParquetFileConfig::getModificationTime).summaryStatistics();
 
-  private LocatedFileStatus getParquetFileAt(
-      Stream<LocatedFileStatus> parquetFiles, long modificationTime) {
-    return parquetFiles
-        .filter(fileStatus -> fileStatus.getModificationTime() == modificationTime)
-        .findFirst()
-        .orElseThrow(() -> new IllegalStateException("No file found at " + modificationTime));
-  }
+    if (stats.getCount() == 0) {
+      log.warn("No parquet files found in table {}. Incremental sync is not possible.", tableName);
+      return false;
+    }
 
-  private Stream<LocatedFileStatus> getParquetFiles(Configuration hadoopConf, String basePath) {
-    try {
-      FileSystem fs = FileSystem.get(hadoopConf);
-      URI uriBasePath = new URI(basePath);
-      String parentPath = Paths.get(uriBasePath).toString();
-      RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(new Path(parentPath), true);
-      return RemoteIterators.toList(iterator).stream()
-          .filter(file -> file.getPath().getName().endsWith("parquet"));
-    } catch (IOException | URISyntaxException e) {
-      throw new ReadException("Unable to read files from file system", e);
+    long earliestModTime = stats.getMin();
+    long latestModTime = stats.getMax();
+
+    if (timeInMillis.toEpochMilli() > latestModTime) {
+      log.warn(
+          "Instant {} is in the future relative to the data. Latest file time: {}",
+          timeInMillis.toEpochMilli(),
+          Instant.ofEpochMilli(latestModTime));
+      return false;
     }
-  }
 
-  @Override
-  public boolean isIncrementalSyncSafeFrom(Instant instant) {
-    return false;
+    if (earliestModTime > timeInMillis.toEpochMilli()) {
+      log.warn(
+          "Incremental sync is not safe. Earliest available metadata (time={}) is newer "
+              + "than requested instant {}. Data history has been truncated.",
+          Instant.ofEpochMilli(earliestModTime),
+          timeInMillis.toEpochMilli());
+      return false;
+    }
+
+    log.info(
+        "Incremental sync is safe from instant {} for table {}",
+        timeInMillis.toEpochMilli(),
+        tableName);
+    return true;
   }
 
   @Override