Parquet Incremental Sync

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

@@ -27,12 +27,14 @@
 import java.util.Comparator;
 import java.util.HashSet;
 import java.util.List;
+import java.util.OptionalLong;
 import java.util.Set;
 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.FileStatus;
@@ -63,6 +65,7 @@
 import org.apache.xtable.spi.extractor.ConversionSource;
 
 @Builder
+@Log4j2
 public class ParquetConversionSource implements ConversionSource<Long> {
 
   private static final ParquetSchemaExtractor schemaExtractor =
@@ -79,6 +82,7 @@ public class ParquetConversionSource implements ConversionSource<Long> {
   private final String tableName;
   private final String basePath;
   @NonNull private final Configuration hadoopConf;
+  private final ParquetDataManager parquetDataManager;
 
   private InternalTable createInternalTableFromFile(LocatedFileStatus latestFile) {
     ParquetMetadata parquetMetadata =
@@ -130,13 +134,14 @@ private Stream<InternalDataFile> getInternalDataFiles(
                 .build());
   }
 
-  private InternalDataFile createInternalDataFileFromParquetFile(
+  public InternalDataFile createInternalDataFileFromParquetFile(
       FileStatus parquetFile, InternalSchema schema) {
     return InternalDataFile.builder()
         .physicalPath(parquetFile.getPath().toString())
         .partitionValues(
             partitionValueExtractor.extractPartitionValues(
-                partitionSpecExtractor.spec(schema), basePath))
+                partitionSpecExtractor.spec(schema),
+                HudiPathUtils.getPartitionPath(new Path(basePath), parquetFile.getPath())))
         .lastModified(parquetFile.getModificationTime())
         .fileSizeBytes(parquetFile.getLen())
         .columnStats(
@@ -162,14 +167,17 @@ public TableChange getTableChangeForCommit(Long modificationTime) {
         parquetFiles
             .filter(fileStatus -> fileStatus.getModificationTime() > modificationTime)
             .collect(Collectors.toList());
-    InternalTable internalTable = getMostRecentTable(parquetFiles);
+    InternalTable internalTable = getMostRecentTable(getParquetFiles(hadoopConf, basePath));
     for (FileStatus tableStatus : tableChangesAfter) {
       InternalDataFile currentDataFile =
           createInternalDataFileFromParquetFile(tableStatus, internalTable.getReadSchema());
       addedInternalDataFiles.add(currentDataFile);
     }
 
     return TableChange.builder()
+        .sourceIdentifier(
+            getCommitIdentifier(
+                parquetDataManager.getMostRecentParquetFile().getModificationTime()))
         .tableAsOfChange(internalTable)
         .filesDiff(InternalFilesDiff.builder().filesAdded(addedInternalDataFiles).build())
         .build();
@@ -235,8 +243,32 @@ private Stream<LocatedFileStatus> getParquetFiles(Configuration hadoopConf, Stri
   }
 
   @Override
-  public boolean isIncrementalSyncSafeFrom(Instant instant) {
-    return false;
+  public boolean isIncrementalSyncSafeFrom(Instant timeInMillis) {
+    Stream<ParquetFileInfo> parquetFilesMetadata = parquetDataManager.getCurrentFilesInfo();
+    OptionalLong earliestModTimeOpt =
+        parquetFilesMetadata.mapToLong(ParquetFileInfo::getModificationTime).min();
+
+    if (!earliestModTimeOpt.isPresent()) {
+      log.warn("No parquet files found in table {}. Incremental sync is not possible.", tableName);
+      return false;
+    }
+
+    long earliestModTime = earliestModTimeOpt.getAsLong();
+
+    if (earliestModTime > timeInMillis.toEpochMilli()) {
+      log.warn(
+          "Incremental sync is not safe. Earliest available metadata (time={}) is newer "
+              + "than requested instant {}.",
+          Instant.ofEpochMilli(earliestModTime),
+          timeInMillis.toEpochMilli());
+      return false;
+    }
+
+    log.debug(
+        "Incremental sync is safe from instant {} for table {}",
+        timeInMillis.toEpochMilli(),
+        tableName);
+    return true;
   }
 
   @Override

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

@@ -32,12 +32,15 @@ public ParquetConversionSource getConversionSourceInstance(SourceTable sourceTab
     ParquetPartitionValueExtractor partitionValueExtractor =
         new ParquetPartitionValueExtractor(
             sourcePartitionSpecExtractor.getPathToPartitionFieldFormat());
+    ParquetDataManager dataManager =
+        new ParquetDataManager(this.hadoopConf, sourceTable.getBasePath());
     return ParquetConversionSource.builder()
         .tableName(sourceTable.getName())
         .basePath(sourceTable.getBasePath())
         .hadoopConf(this.hadoopConf)
         .partitionSpecExtractor(sourcePartitionSpecExtractor)
         .partitionValueExtractor(partitionValueExtractor)
+        .parquetDataManager(dataManager)
         .build();
   }
 }

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

@@ -0,0 +1,107 @@
+/*
+ * 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.parquet;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.util.functional.RemoteIterators;
+
+import org.apache.xtable.exception.ReadException;
+
+/**
+ * Manages Parquet File's Metadata
+ *
+ * <p>This class provides functions to handle Parquet metadata, creating metadata objects from
+ * parquet files and filtering the files based on the modification times.
+ */
+@Log4j2
+@RequiredArgsConstructor
+public class ParquetDataManager {
+  private final Configuration hadoopConf;
+  private final String basePath;
+  private final FileSystem fileSystem;
+
+  public ParquetDataManager(Configuration hadoopConf, String basePath) {
+    this.hadoopConf = hadoopConf;
+    this.basePath = basePath;
+    try {
+      URI uri = new Path(basePath).toUri();
+      this.fileSystem = FileSystem.get(uri, hadoopConf);
+    } catch (IOException e) {
+      throw new ReadException("Unable to initialize file system for base path: " + basePath, e);
+    }
+  }
+
+  @Getter(value = AccessLevel.PRIVATE, lazy = true)
+  private final List<LocatedFileStatus> parquetFiles = loadParquetFiles();
+
+  ParquetFileInfo getMostRecentParquetFile() {
+    LocatedFileStatus file =
+        getParquetFiles().stream()
+            .max(Comparator.comparing(LocatedFileStatus::getModificationTime))
+            .orElseThrow(() -> new IllegalStateException("No files found"));
+    return new ParquetFileInfo(hadoopConf, file);
+  }
+
+  ParquetFileInfo getParquetDataFileAt(long targetTime) {
+    return getParquetFiles().stream()
+        .filter(file -> file.getModificationTime() > targetTime)
+        .min(Comparator.comparing(LocatedFileStatus::getModificationTime))
+        .map(file -> new ParquetFileInfo(hadoopConf, file))
+        .orElseThrow(() -> new IllegalStateException("No file found at or after " + targetTime));
+  }
+
+  private List<LocatedFileStatus> loadParquetFiles() {
+    try {
+      RemoteIterator<LocatedFileStatus> iterator = fileSystem.listFiles(new Path(basePath), true);
+      return RemoteIterators.toList(iterator).stream()
+          .filter(file -> file.getPath().getName().endsWith("parquet"))
+          .collect(Collectors.toList());
+    } catch (IOException e) {
+      throw new ReadException("Unable to read files from file system", e);
+    }
+  }
+
+  Stream<ParquetFileInfo> getCurrentFilesInfo() {
+    return getParquetFiles().stream()
+        .map(fileStatus -> new ParquetFileInfo(hadoopConf, fileStatus));
+  }
+
+  List<ParquetFileInfo> getParquetFilesMetadataAfterTime(long syncTime) {
+    return getParquetFiles().stream()
+        .filter(file -> file.getModificationTime() > syncTime)
+        .map(file -> new ParquetFileInfo(hadoopConf, file))
+        .collect(Collectors.toList());
+  }
+}

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

@@ -0,0 +1,51 @@
+/*
+ * 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.parquet;
+
+import lombok.Getter;
+import lombok.Value;
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+
+@Log4j2
+@Value
+class ParquetFileInfo {
+  @Getter(lazy = true)
+  ParquetMetadata metadata = getMetadataInternal();
+
+  long modificationTime;
+  long size;
+  Path path;
+  Configuration conf;
+
+  public ParquetFileInfo(Configuration conf, FileStatus file) {
+    this.modificationTime = file.getModificationTime();
+    this.path = file.getPath();
+    this.size = file.getLen();
+    this.conf = conf;
+  }
+
+  private ParquetMetadata getMetadataInternal() {
+    return ParquetMetadataExtractor.getInstance().readParquetMetadata(conf, path);
+  }
+}