Re: [PR] Parquet Incremental Sync [incubator-xtable]

["vinishjail97 (via GitHub)" <gi...@apache.org>]

vinishjail97 commented on code in PR #768:
URL: https://github.com/apache/incubator-xtable/pull/768#discussion_r3364543748


##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -162,14 +182,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));

Review Comment:
   `getTableChangeForCommit` lists the filesystem twice here (L178 and L185), and `getCurrentCommitState`/`getCurrentSnapshot` repeat the pattern (L208/220/222/227). Each `getParquetFiles(hadoopConf, basePath)` is a full recursive `listFiles` — on S3/GCS that is many round-trips. This is exactly @the-other-tim-brown's earlier "fetch the list once" comment, and it's still open. You already inject `parquetDataManager`, which caches the listing lazily (`@Getter(lazy=true)`) — please route all listings through it and drop the static `getParquetFiles` duplicate so there's a single source of truth.



##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetDataManager.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * 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 com.google.common.annotations.VisibleForTesting;
+
+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(
+    access = AccessLevel.PACKAGE,
+    onConstructor_ = {@VisibleForTesting})
+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) {

Review Comment:
   `getParquetDataFileAt(long)` (and `getParquetFilesMetadataAfterTime(long)` at L105) are only referenced from `TestParquetDataManager` — no production caller. Either wire them into the incremental path (they look like the intended replacement for the duplicate listing in `ParquetConversionSource`) or remove them so we don't ship test-only API.



##########
xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java:
##########
@@ -162,14 +182,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()))

Review Comment:
   The `sourceIdentifier` is derived from `parquetDataManager.getMostRecentParquetFile()`, but the table snapshot and `tableChangesAfter` come from a *separate* `getParquetFiles(...)` listing above. If a file lands between the two listings, the committed source identifier can advance past a file that wasn't included in `filesAdded`, marking that file permanently synced. Take one snapshot of the file list and derive both the changes and the identifier from it.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@xtable.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org