This is an automated email from the ASF dual-hosted git repository.
vinishjail97 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
The following commit(s) were added to refs/heads/main by this push:
new 754fd270 Parquet Incremental Sync (#768)
754fd270 is described below
commit 754fd270e6f0f74c089f6c64f5c386a6d67b05f1
Author: Selim S. <[email protected]>
AuthorDate: Fri Jun 5 23:21:29 2026 +0200
Parquet Incremental Sync (#768)
* given a parquet file return data from a certain modification time
* create the path based on the partition then inject the file to append
into the parquet table
* Handle case of path construction with file partitioned over many fields,
interfacing with ConversionSource
* test append Parquet file into table init
* add function to test schema equivalence before appending
* construct path to inject to based on partitions
* fix imports
* refactoring (lombok, logs, javadocs and function and approach commenting)
* use appendFile to append a file into a table while tracking the appends
using the FileStatus' modifTime attribute
* find the files that satisfy to the time condition
* treat appends as separate files to add in the target partition folder
* update approach: selective block compaction
* update approach: added a basic test to check data selection using
modificationTime selector
* fix append based on partition value
* fix test with basic example where partitions are not considered
* fix test with basic example where partitions are not considered2
* fix test with basic example where partitions are not considered3
* test with time of last append is now
* test appendFile with Parquet: TODO test with multiple partitions 1)
append and 2) filter for sync
* merge recursively one partition files
* fix paths for files to append
* fix bug of appending file path
* fix bug of schema
* make sure returned files are non empty when iterating in the partitions
* make sure returned files are non empty when iterating in the partitions 2
* discard empty files when appending
* discard empty files when appending
* one spark session for both tests
* bug fix
* bug fix
* cleanups + TODO: partitions match and merge
* added test for many partitions, TODO integrate functions into
ParquetConversionSource
* selecting data bug fix, TODO integrate functions into
ParquetConversionSource
* run all tests, TODO integrate functions into ParquetConversionSource
* run all tests, TODO integrate functions into ParquetConversionSource
* spotless:apply, TODO integrate functions into ParquetConversionSource
* bug fixes, TODO integrate functions into ParquetConversionSource
* bug fixes, TODO integrate functions into ParquetConversionSource
* handle lightweight metadata in the conversionSource
* fix CI
* fix CI, refactoring and isIncrementalSyncSafeFrom implementation
* added test
* added test: init conversionSource instance
* added test: init conversionSource instance: fix error
* added test: fix error
* more tests + bug fixes and reformatting
* CI fix
* CI fix
* CI fix
* CI fix
* CI fix
* CI fix
* merge ready fixes & tests + refactoring & reformatting
* optimization: read metadata once for target file, read parquet files
once, reformatting
* Lazily fetch info, move state into ParquetDataManager
* add TestParquetDataManager, simplify data manager methods
* make IT parameterized, add incremental sync
* fix schema extraction to fix snapshot sync
* add optimizations
* fix for ParquetDataManager Test (mock())
* before resycing delete the first synced files metadata
* in order to append parquet files, perform union then spark overwrite (not
append)
* solving directory issue when overwriting data using spark
* solving directory issue when overwriting data using spark2
* solving directory issue when overwriting data using spark3
* solving directory issue when overwriting data using spark4
* solving directory issue when overwriting data using spark5
* solving directory issue when overwriting data using spark6
* solving directory issue when overwriting data using spark7
* revert writeData changes
* solving directory issue when overwriting data using spark8
* solving directory issue when overwriting data using spark9
* solving directory issue when overwriting data using spark9
* revert changes
* testing without cleaning metadata for the second sync
* testing without cleaning metadata for the second sync
* testing without cleaning metadata for the second sync
* testing without cleaning metadata for the second sync
* testing without cleaning metadata for the second sync
* revert changes
* revert changes
* revert changes
* use no tempDir
* revert changes
* use union() to append data for the second sync()
* use union() to append data for the second sync()
* use union() to append data for the second sync()
* use union() to append data for the second sync()
* use union() to append data for the second sync() reverted
* fix second sync error
* fix second sync error
* fix second sync error
* fix second sync error
* fix second sync error
* fix second sync error
* fix second sync error
* fix second sync error
* fix second sync error
* fix second sync error
* fix second sync error
* remove old blocking tests
* spotless:apply and more tests
* spotless:apply and test fix
* spotless:apply and test fix
* spotless:apply and test fix
* test fix
* spotless
* spotless imports
* spotless imports
* add syncMode Incr
* revert changes
* add syncMode Incr
* spotless:apply
* update naming
* minimize diff with main
* added test for tableChangeAddedFiles and changed path construction of
ParquetDataManager and adjusted operator for mostRecentFile + spotless
* added test for tableChangeAddedFiles and changed path construction of
ParquetDataManager and adjusted operator for mostRecentFile + spotless + import
error fixed
* tests fixed
* tests fixed
* tests fixed
* log for actual changes
* log for actual changes
* tests fixed
* spotless
* removed comment
* spotless
* address PR comments
* fix path
* add safety check around handling of out of sync targets with parquet
source
* Address review feedback: fetch parquet listing once; drop unused
ParquetDataManager methods
- ParquetConversionSource: materialize the file listing once per operation
in
getTableChangeForCommit and getCurrentSnapshot instead of re-listing the
filesystem 2-3 times (addresses the 'fetch the list once' review comment).
- ParquetDataManager: remove getParquetDataFileAt and
getParquetFilesMetadataAfterTime; both had no production caller (only
tests).
- TestParquetDataManager: drop tests for the removed methods and repoint the
real-filesystem tests to getCurrentFilesInfo.
* Derive source identifier from the same listing as the table change
getTableChangeForCommit computed filesAdded and the table from one
materialized listing but derived the committed sourceIdentifier from a
separate parquetDataManager.getMostRecentParquetFile() listing. A file
landing between the two listings could advance the committed identifier
past a file that was never included in filesAdded, permanently skipping
it on the next incremental sync.
Use the most recent file from the same snapshot for both the table and
the source identifier. Add an assertion in ITParquetConversionSource
that the committed identifier matches that snapshot's latest mod time.
---------
Co-authored-by: Selim Soufargi <[email protected]~>
Co-authored-by: Timothy Brown <[email protected]>
Co-authored-by: Tim Brown <[email protected]>
Co-authored-by: Vinish Reddy <[email protected]>
---
.../xtable/model/InstantsForIncrementalSync.java | 9 +
.../xtable/conversion/ConversionController.java | 20 +-
.../xtable/parquet/ParquetConversionSource.java | 76 ++++-
.../parquet/ParquetConversionSourceProvider.java | 3 +
.../apache/xtable/parquet/ParquetDataManager.java | 96 +++++++
.../org/apache/xtable/parquet/ParquetFileInfo.java | 51 ++++
.../conversion/TestConversionController.java | 31 ++-
.../xtable/parquet/ITParquetConversionSource.java | 196 ++++++++++++-
.../xtable/parquet/TestParquetDataManager.java | 305 +++++++++++++++++++++
9 files changed, 757 insertions(+), 30 deletions(-)
diff --git
a/xtable-api/src/main/java/org/apache/xtable/model/InstantsForIncrementalSync.java
b/xtable-api/src/main/java/org/apache/xtable/model/InstantsForIncrementalSync.java
index 7c00bf21..af5022dd 100644
---
a/xtable-api/src/main/java/org/apache/xtable/model/InstantsForIncrementalSync.java
+++
b/xtable-api/src/main/java/org/apache/xtable/model/InstantsForIncrementalSync.java
@@ -48,6 +48,15 @@ import lombok.Value;
@Builder
public class InstantsForIncrementalSync {
Instant lastSyncInstant;
+ // last synced instants by target format for all targets considered for this
incremental sync.
+ @Builder.Default List<TargetSyncInstant> targetSyncInstants =
Collections.emptyList();
// pending commits that are not yet synced and to be considered for next
incremental sync.
@Builder.Default List<Instant> pendingCommits = Collections.emptyList();
+
+ @Value
+ @Builder
+ public static class TargetSyncInstant {
+ String tableFormat;
+ Instant lastSyncInstant;
+ }
}
diff --git
a/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionController.java
b/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionController.java
index 1c138aeb..73bc531a 100644
---
a/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionController.java
+++
b/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionController.java
@@ -24,6 +24,7 @@ import java.io.IOException;
import java.time.Instant;
import java.util.Collection;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -43,6 +44,7 @@ import org.apache.xtable.catalog.CatalogConversionFactory;
import org.apache.xtable.exception.ReadException;
import org.apache.xtable.model.IncrementalTableChanges;
import org.apache.xtable.model.InstantsForIncrementalSync;
+import org.apache.xtable.model.InstantsForIncrementalSync.TargetSyncInstant;
import org.apache.xtable.model.InternalSnapshot;
import org.apache.xtable.model.catalog.CatalogTableIdentifier;
import org.apache.xtable.model.metadata.TableSyncMetadata;
@@ -351,10 +353,21 @@ public class ConversionController {
private InstantsForIncrementalSync getMostOutOfSyncCommitAndPendingCommits(
Map<ConversionTarget, TableSyncMetadata> lastSyncMetadataByFormat) {
+ List<TargetSyncInstant> targetSyncInstants =
+ lastSyncMetadataByFormat.entrySet().stream()
+ .map(
+ entry ->
+ TargetSyncInstant.builder()
+ .tableFormat(entry.getKey().getTableFormat())
+
.lastSyncInstant(entry.getValue().getLastInstantSynced())
+ .build())
+ .sorted(
+ Comparator.comparing(TargetSyncInstant::getLastSyncInstant)
+ .thenComparing(TargetSyncInstant::getTableFormat))
+ .collect(Collectors.toList());
Instant mostOutOfSyncCommit =
- lastSyncMetadataByFormat.values().stream()
- .map(TableSyncMetadata::getLastInstantSynced)
- .sorted()
+ targetSyncInstants.stream()
+ .map(TargetSyncInstant::getLastSyncInstant)
.findFirst()
.orElseThrow(
() ->
@@ -368,6 +381,7 @@ public class ConversionController {
.collect(Collectors.toList());
return InstantsForIncrementalSync.builder()
.lastSyncInstant(mostOutOfSyncCommit)
+ .targetSyncInstants(targetSyncInstants)
.pendingCommits(allPendingInstants)
.build();
}
diff --git
a/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java
b/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java
index 61a2ba8f..3a505bb8 100644
---
a/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java
+++
b/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSource.java
@@ -27,12 +27,14 @@ import java.util.Collections;
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;
@@ -44,6 +46,7 @@ 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.NotSupportedException;
import org.apache.xtable.exception.ReadException;
import org.apache.xtable.hudi.HudiPathUtils;
import org.apache.xtable.hudi.PathBasedPartitionSpecExtractor;
@@ -63,6 +66,7 @@ import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.spi.extractor.ConversionSource;
@Builder
+@Log4j2
public class ParquetConversionSource implements ConversionSource<Long> {
private static final ParquetSchemaExtractor schemaExtractor =
@@ -79,6 +83,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 +135,14 @@ public class ParquetConversionSource implements
ConversionSource<Long> {
.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(
@@ -148,6 +154,20 @@ public class ParquetConversionSource implements
ConversionSource<Long> {
@Override
public CommitsBacklog<Long> getCommitsBacklog(InstantsForIncrementalSync
syncInstants) {
+ if (syncInstants.getTargetSyncInstants().stream()
+
.map(InstantsForIncrementalSync.TargetSyncInstant::getLastSyncInstant)
+ .distinct()
+ .count()
+ > 1) {
+ String targetSyncInstants =
+ syncInstants.getTargetSyncInstants().stream()
+ .map(target -> target.getTableFormat() + "=" +
target.getLastSyncInstant().toString())
+ .collect(Collectors.joining(", "));
+ throw new NotSupportedException(
+ "Parquet incremental sync does not support target formats synced to
different instants. "
+ + "Run a full sync to realign the target tables. Target
instants: "
+ + targetSyncInstants);
+ }
List<Long> commitsToProcess =
Collections.singletonList(syncInstants.getLastSyncInstant().toEpochMilli());
return
CommitsBacklog.<Long>builder().commitsToProcess(commitsToProcess).build();
@@ -155,14 +175,20 @@ public class ParquetConversionSource implements
ConversionSource<Long> {
@Override
public TableChange getTableChangeForCommit(Long modificationTime) {
- Stream<LocatedFileStatus> parquetFiles = getParquetFiles(hadoopConf,
basePath);
+ List<LocatedFileStatus> parquetFiles =
+ getParquetFiles(hadoopConf, basePath).collect(Collectors.toList());
Set<InternalDataFile> addedInternalDataFiles = new HashSet<>();
List<FileStatus> tableChangesAfter =
- parquetFiles
+ parquetFiles.stream()
.filter(fileStatus -> fileStatus.getModificationTime() >
modificationTime)
.collect(Collectors.toList());
- InternalTable internalTable = getMostRecentTable(parquetFiles);
+ // Derive both the table and the source identifier from the same snapshot.
Using a separate
+ // listing for the identifier risks committing a modification time for a
file that landed after
+ // the snapshot and was not included in filesAdded, which would
permanently skip that file on
+ // the next incremental sync.
+ LocatedFileStatus mostRecentFile =
getMostRecentParquetFile(parquetFiles.stream());
+ InternalTable internalTable = createInternalTableFromFile(mostRecentFile);
for (FileStatus tableStatus : tableChangesAfter) {
InternalDataFile currentDataFile =
createInternalDataFileFromParquetFile(tableStatus,
internalTable.getReadSchema());
@@ -170,6 +196,7 @@ public class ParquetConversionSource implements
ConversionSource<Long> {
}
return TableChange.builder()
+
.sourceIdentifier(getCommitIdentifier(mostRecentFile.getModificationTime()))
.tableAsOfChange(internalTable)
.filesDiff(InternalFilesDiff.builder().filesAdded(addedInternalDataFiles).build())
.build();
@@ -193,16 +220,17 @@ public class ParquetConversionSource implements
ConversionSource<Long> {
*/
@Override
public InternalSnapshot getCurrentSnapshot() {
- // to avoid consume the stream call the method twice to return the same
stream of parquet files
- InternalTable table = getMostRecentTable(getParquetFiles(hadoopConf,
basePath));
+ // Fetch the file listing once and reuse it across the table, data files,
and source identifier.
+ List<LocatedFileStatus> parquetFiles =
+ getParquetFiles(hadoopConf, basePath).collect(Collectors.toList());
+ InternalTable table = getMostRecentTable(parquetFiles.stream());
Stream<InternalDataFile> internalDataFiles =
- getInternalDataFiles(getParquetFiles(hadoopConf, basePath),
table.getReadSchema());
+ getInternalDataFiles(parquetFiles.stream(), table.getReadSchema());
return InternalSnapshot.builder()
.table(table)
.sourceIdentifier(
getCommitIdentifier(
- getMostRecentParquetFile(getParquetFiles(hadoopConf, basePath))
- .getModificationTime()))
+
getMostRecentParquetFile(parquetFiles.stream()).getModificationTime()))
.partitionedDataFiles(PartitionFileGroup.fromFiles(internalDataFiles))
.build();
}
@@ -235,8 +263,32 @@ public class ParquetConversionSource implements
ConversionSource<Long> {
}
@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
diff --git
a/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSourceProvider.java
b/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSourceProvider.java
index 0d508823..81035cea 100644
---
a/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSourceProvider.java
+++
b/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetConversionSourceProvider.java
@@ -32,12 +32,15 @@ public class ParquetConversionSourceProvider extends
ConversionSourceProvider<Lo
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();
}
}
diff --git
a/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetDataManager.java
b/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetDataManager.java
new file mode 100644
index 00000000..b66b7491
--- /dev/null
+++
b/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetDataManager.java
@@ -0,0 +1,96 @@
+/*
+ * 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);
+ }
+
+ 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));
+ }
+}
diff --git
a/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetFileInfo.java
b/xtable-core/src/main/java/org/apache/xtable/parquet/ParquetFileInfo.java
new file mode 100644
index 00000000..7dfd587f
--- /dev/null
+++ b/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);
+ }
+}
diff --git
a/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
b/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
index 75323af8..6569c4e8 100644
---
a/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
+++
b/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
@@ -45,6 +45,7 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.hadoop.conf.Configuration;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentMatcher;
@@ -54,6 +55,7 @@ import org.apache.xtable.catalog.CatalogConversionFactory;
import org.apache.xtable.model.CommitsBacklog;
import org.apache.xtable.model.IncrementalTableChanges;
import org.apache.xtable.model.InstantsForIncrementalSync;
+import org.apache.xtable.model.InstantsForIncrementalSync.TargetSyncInstant;
import org.apache.xtable.model.InternalSnapshot;
import org.apache.xtable.model.InternalTable;
import org.apache.xtable.model.TableChange;
@@ -90,6 +92,12 @@ public class TestConversionController {
private final CatalogSyncClient mockCatalogSyncClient1 =
mock(CatalogSyncClient.class);
private final CatalogSyncClient mockCatalogSyncClient2 =
mock(CatalogSyncClient.class);
+ @BeforeEach
+ void setupTargetFormats() {
+ when(mockConversionTarget1.getTableFormat()).thenReturn(ICEBERG);
+ when(mockConversionTarget2.getTableFormat()).thenReturn(DELTA);
+ }
+
@Test
void testAllSnapshotSyncAsPerConfig() {
SyncMode syncMode = SyncMode.FULL;
@@ -165,6 +173,10 @@ public class TestConversionController {
InstantsForIncrementalSync instantsForIncrementalSync =
InstantsForIncrementalSync.builder()
.lastSyncInstant(icebergLastSyncInstant)
+ .targetSyncInstants(
+ Arrays.asList(
+ targetSyncInstant(ICEBERG, icebergLastSyncInstant),
+ targetSyncInstant(DELTA, deltaLastSyncInstant)))
.pendingCommits(combinedPendingInstants)
.build();
List<Instant> instantsToProcess =
@@ -307,6 +319,8 @@ public class TestConversionController {
InstantsForIncrementalSync instantsForIncrementalSync =
InstantsForIncrementalSync.builder()
.lastSyncInstant(deltaLastSyncInstant)
+ .targetSyncInstants(
+ Collections.singletonList(targetSyncInstant(DELTA,
deltaLastSyncInstant)))
.pendingCommits(pendingInstantsForDelta)
.build();
List<Instant> instantsToProcess = Arrays.asList(instantAt8, instantAt2);
@@ -389,10 +403,16 @@ public class TestConversionController {
// Iceberg last synced at instantAt5, the last instant in the source
Instant icebergLastSyncInstant = instantAt5;
- // Delta last synced at instantAt10
+ // Delta last synced at instantAt5
Instant deltaLastSyncInstant = instantAt5;
InstantsForIncrementalSync instantsForIncrementalSync =
-
InstantsForIncrementalSync.builder().lastSyncInstant(deltaLastSyncInstant).build();
+ InstantsForIncrementalSync.builder()
+ .lastSyncInstant(deltaLastSyncInstant)
+ .targetSyncInstants(
+ Arrays.asList(
+ targetSyncInstant(DELTA, deltaLastSyncInstant),
+ targetSyncInstant(ICEBERG, icebergLastSyncInstant)))
+ .build();
List<Instant> instantsToProcess = Collections.emptyList();
CommitsBacklog<Instant> commitsBacklog =
CommitsBacklog.<Instant>builder().commitsToProcess(instantsToProcess).build();
@@ -640,4 +660,11 @@ public class TestConversionController {
}
return !first.hasNext() && !second.hasNext();
}
+
+ private static TargetSyncInstant targetSyncInstant(String tableFormat,
Instant lastSyncInstant) {
+ return TargetSyncInstant.builder()
+ .tableFormat(tableFormat)
+ .lastSyncInstant(lastSyncInstant)
+ .build();
+ }
}
diff --git
a/xtable-core/src/test/java/org/apache/xtable/parquet/ITParquetConversionSource.java
b/xtable-core/src/test/java/org/apache/xtable/parquet/ITParquetConversionSource.java
index 2ffa579a..be3d0b43 100644
---
a/xtable-core/src/test/java/org/apache/xtable/parquet/ITParquetConversionSource.java
+++
b/xtable-core/src/test/java/org/apache/xtable/parquet/ITParquetConversionSource.java
@@ -18,6 +18,7 @@
package org.apache.xtable.parquet;
+import static org.apache.spark.sql.functions.expr;
import static org.apache.xtable.GenericTable.getTableName;
import static org.apache.xtable.model.storage.TableFormat.DELTA;
import static org.apache.xtable.model.storage.TableFormat.HUDI;
@@ -25,14 +26,19 @@ import static
org.apache.xtable.model.storage.TableFormat.ICEBERG;
import static org.apache.xtable.model.storage.TableFormat.PARQUET;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.io.IOException;
import java.nio.file.Path;
-import java.nio.file.Paths;
import java.sql.Timestamp;
import java.time.Duration;
+import java.time.Instant;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -45,6 +51,11 @@ import java.util.stream.Stream;
import lombok.Builder;
import lombok.Value;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.RemoteIterator;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
@@ -59,6 +70,7 @@ import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
@@ -73,7 +85,15 @@ import org.apache.xtable.conversion.ConversionController;
import org.apache.xtable.conversion.ConversionSourceProvider;
import org.apache.xtable.conversion.SourceTable;
import org.apache.xtable.conversion.TargetTable;
+import org.apache.xtable.exception.NotSupportedException;
import org.apache.xtable.hudi.HudiTestUtil;
+import org.apache.xtable.model.InstantsForIncrementalSync;
+import org.apache.xtable.model.InstantsForIncrementalSync.TargetSyncInstant;
+import org.apache.xtable.model.InternalSnapshot;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.TableChange;
+import org.apache.xtable.model.storage.InternalDataFile;
+import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.model.sync.SyncMode;
public class ITParquetConversionSource {
@@ -167,7 +187,7 @@ public class ITParquetConversionSource {
return partitionConfigs.stream()
.flatMap(
partitionConfig ->
- Stream.of(SyncMode.FULL) // Incremental sync is not yet
supported
+ Stream.of(SyncMode.FULL, SyncMode.INCREMENTAL)
.map(
syncMode ->
Arguments.of(
@@ -226,20 +246,17 @@ public class ITParquetConversionSource {
new MetadataBuilder().putString("precision",
"millis").build())
});
Dataset<Row> df = sparkSession.createDataFrame(data, schema);
- String dataPath =
- tempDir
- .resolve(
- (xTablePartitionConfig == null ? "non_partitioned_data_" :
"partitioned_data_")
- + tableFormatPartitionDataHolder.getSyncMode())
- .toString();
-
- writeData(df, dataPath, xTablePartitionConfig);
+ Path dataPath =
+ tempDir.resolve(
+ (xTablePartitionConfig == null ? "non_partitioned_data_" :
"partitioned_data_")
+ + tableFormatPartitionDataHolder.getSyncMode());
+
+ writeData(df, dataPath.toString(), xTablePartitionConfig);
boolean isPartitioned = xTablePartitionConfig != null;
- Path pathForXTable = Paths.get(dataPath);
try (GenericTable table =
GenericTable.getInstance(
- tableName, pathForXTable, sparkSession, jsc, sourceTableFormat,
isPartitioned)) {
+ tableName, dataPath, sparkSession, jsc, sourceTableFormat,
isPartitioned)) {
ConversionConfig conversionConfig =
getTableSyncConfig(
sourceTableFormat,
@@ -266,7 +283,7 @@ public class ITParquetConversionSource {
new Timestamp(System.currentTimeMillis() + 1500)));
Dataset<Row> dfAppend = sparkSession.createDataFrame(dataToAppend,
schema);
- writeData(dfAppend, dataPath, xTablePartitionConfig);
+ writeData(dfAppend, dataPath.toString(), xTablePartitionConfig);
ConversionConfig conversionConfigAppended =
getTableSyncConfig(
sourceTableFormat,
@@ -418,6 +435,159 @@ public class ITParquetConversionSource {
}
}
+ @Test
+ void testIncrementalSyncWithMultiplePartitions() throws IOException {
+
+ Configuration conf = sparkSession.sparkContext().hadoopConfiguration();
+
+ StructType schema =
+ DataTypes.createStructType(
+ new StructField[] {
+ DataTypes.createStructField("id", DataTypes.IntegerType, false),
+ DataTypes.createStructField("value", DataTypes.StringType,
false),
+ DataTypes.createStructField("year", DataTypes.IntegerType,
false),
+ DataTypes.createStructField("month", DataTypes.IntegerType,
false)
+ });
+ List<Row> data =
+ Arrays.asList(
+ RowFactory.create(100, "A", 2026, 12),
+ RowFactory.create(101, "AA", 2026, 12),
+ RowFactory.create(102, "CB", 2027, 11),
+ RowFactory.create(103, "BA", 2027, 11));
+
+ Dataset<Row> dfInit = sparkSession.createDataFrame(data, schema);
+ Path fixedPath =
tempDir.resolve("fixed-parquet-data/parquet_table_test_2");
+
+ Dataset<Row> df = dfInit.withColumn("full_date", expr("make_date(year,
month, 1)"));
+ String outputPath = fixedPath.toString();
+ df.coalesce(1).write().partitionBy("year",
"month").mode("overwrite").parquet(outputPath);
+
+ // test find files to sync
+ FileSystem fs = FileSystem.get(fixedPath.toUri(), conf);
+ // set the modification time to the table file
+ // update modificationTime for file to append
+ // many partitions case
+ List<String> newPartitions = Arrays.asList("year=2026/month=12",
"year=2027/month=11");
+ long targetModificationTime = System.currentTimeMillis() - 360000;
+ long newModificationTime = System.currentTimeMillis() - 50000;
+ long testTime = System.currentTimeMillis() - 90000; // between two prev
times
+ for (String partition : newPartitions) {
+ org.apache.hadoop.fs.Path partitionPath =
+ new org.apache.hadoop.fs.Path(outputPath, partition);
+ if (fs.exists(partitionPath)) {
+ updateModificationTimeRecursive(fs, partitionPath,
targetModificationTime);
+ }
+ }
+ // create new file to append using Spark
+ List<Row> futureDataToSync =
+ Arrays.asList(
+ RowFactory.create(101, "A", 2026, 12),
+ RowFactory.create(301, "D", 2027, 11),
+ RowFactory.create(302, "DA", 2027, 11));
+ Dataset<Row> dfToSyncInit = sparkSession.createDataFrame(futureDataToSync,
schema);
+ Dataset<Row> dfToSync = dfToSyncInit.withColumn("full_date",
expr("make_date(year, month, 1)"));
+ dfToSync.coalesce(1).write().partitionBy("year",
"month").mode("append").parquet(outputPath);
+
+ // conversionSource operations
+ Properties sourceProperties = new Properties();
+ String partitionConfig = "full_date:MONTH:year=yyyy/month=MM";
+ sourceProperties.put(PARTITION_FIELD_SPEC_CONFIG, partitionConfig);
+
+ SourceTable tableConfig =
+ SourceTable.builder()
+ .name("parquet_table_test_2")
+ .basePath(fixedPath.toAbsolutePath().toString())
+ .additionalProperties(sourceProperties)
+ .formatName(TableFormat.PARQUET)
+ .build();
+
+ ParquetConversionSourceProvider conversionSourceProvider =
+ new ParquetConversionSourceProvider();
+ conversionSourceProvider.init(conf);
+ ParquetConversionSource conversionSource =
+ conversionSourceProvider.getConversionSourceInstance(tableConfig);
+ Set<InternalDataFile> expectedAddedFiles = new HashSet<>();
+
+ for (String partition : newPartitions) {
+ org.apache.hadoop.fs.Path partitionPath =
+ new org.apache.hadoop.fs.Path(outputPath, partition);
+
+ RemoteIterator<LocatedFileStatus> it = fs.listFiles(partitionPath, true);
+ while (it.hasNext()) {
+ LocatedFileStatus fileStatus = it.next();
+
+ if (fileStatus.getModificationTime() > testTime) {
+ fs.setTimes(fileStatus.getPath(), newModificationTime, -1);
+ FileStatus refreshedStatus = fs.getFileStatus(fileStatus.getPath());
+ expectedAddedFiles.add(
+ conversionSource.createInternalDataFileFromParquetFile(
+ refreshedStatus,
conversionSource.getCurrentTable().getReadSchema()));
+ } else {
+ fs.setTimes(fileStatus.getPath(), targetModificationTime, -1);
+ }
+ }
+ fs.setTimes(partitionPath, newModificationTime, -1);
+ }
+ InternalTable result = conversionSource.getTable(newModificationTime);
+ assertEquals(
+ Instant.ofEpochMilli(newModificationTime).toString(),
+ result.getLatestCommitTime().toString());
+ assertEquals("parquet_table_test_2", result.getName());
+ assertEquals(TableFormat.PARQUET, result.getTableFormat());
+ assertNotNull(result.getReadSchema());
+ InternalSnapshot snapshot = conversionSource.getCurrentSnapshot();
+ assertNotNull(snapshot);
+ TableChange changes = conversionSource.getTableChangeForCommit(testTime);
+ assertNotNull(changes);
+ assertFalse(changes.getFilesDiff().dataFilesAdded().isEmpty(), "Should
have found added files");
+ assertEquals(expectedAddedFiles, changes.getFilesDiff().dataFilesAdded());
+ // The committed source identifier must be the most-recent modification
time from the same
+ // listing used to compute filesAdded, otherwise a file could be marked
synced without being
+ // included in the changes.
+ assertEquals(String.valueOf(newModificationTime),
changes.getSourceIdentifier());
+ Instant instantBeforeFirstSnapshot =
+
Instant.ofEpochMilli(snapshot.getTable().getLatestCommitTime().toEpochMilli());
+ assertEquals(instantBeforeFirstSnapshot.toEpochMilli(),
newModificationTime);
+
assertTrue(conversionSource.isIncrementalSyncSafeFrom(Instant.ofEpochMilli(testTime)));
+ }
+
+ @Test
+ void testGetCommitsBacklog_rejectsDifferentTargetSyncInstants() {
+ ParquetConversionSource source =
+ ParquetConversionSource.builder().hadoopConf(new
Configuration()).build();
+ InstantsForIncrementalSync syncInstants =
+ InstantsForIncrementalSync.builder()
+ .lastSyncInstant(Instant.ofEpochMilli(1000L))
+ .targetSyncInstants(
+ Arrays.asList(
+ TargetSyncInstant.builder()
+ .tableFormat(ICEBERG)
+ .lastSyncInstant(Instant.ofEpochMilli(1000L))
+ .build(),
+ TargetSyncInstant.builder()
+ .tableFormat(DELTA)
+ .lastSyncInstant(Instant.ofEpochMilli(2000L))
+ .build()))
+ .build();
+
+ NotSupportedException exception =
+ assertThrows(NotSupportedException.class, () ->
source.getCommitsBacklog(syncInstants));
+ assertTrue(exception.getMessage().contains("target formats synced to
different instants"));
+
assertTrue(exception.getMessage().contains("ICEBERG=1970-01-01T00:00:01Z"));
+ assertTrue(exception.getMessage().contains("DELTA=1970-01-01T00:00:02Z"));
+ }
+
+ private void updateModificationTimeRecursive(
+ FileSystem fs, org.apache.hadoop.fs.Path path, long time) throws
IOException {
+ RemoteIterator<LocatedFileStatus> it = fs.listFiles(path, true);
+ while (it.hasNext()) {
+ LocatedFileStatus status = it.next();
+ if (status.getPath().getName().endsWith(".parquet")) {
+ fs.setTimes(status.getPath(), time, -1);
+ }
+ }
+ }
+
@Builder
@Value
private static class TableFormatPartitionDataHolder {
diff --git
a/xtable-core/src/test/java/org/apache/xtable/parquet/TestParquetDataManager.java
b/xtable-core/src/test/java/org/apache/xtable/parquet/TestParquetDataManager.java
new file mode 100644
index 00000000..cce32fb8
--- /dev/null
+++
b/xtable-core/src/test/java/org/apache/xtable/parquet/TestParquetDataManager.java
@@ -0,0 +1,305 @@
+/*
+ * 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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.FileTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class TestParquetDataManager {
+
+ private static final Configuration CONF = new Configuration();
+
+ // Helper method to create a mock LocatedFileStatus
+ private static LocatedFileStatus createMockFileStatus(String name, long
modTime) {
+ LocatedFileStatus mockStatus = mock(LocatedFileStatus.class);
+ org.apache.hadoop.fs.Path mockPath = mock(org.apache.hadoop.fs.Path.class);
+ when(mockPath.getName()).thenReturn(name);
+ when(mockStatus.getPath()).thenReturn(mockPath);
+ when(mockStatus.getModificationTime()).thenReturn(modTime);
+ when(mockStatus.getLen()).thenReturn(1024L);
+ return mockStatus;
+ }
+
+ private static class StubbedRemoteIterator implements
RemoteIterator<LocatedFileStatus> {
+ private final Iterator<LocatedFileStatus> files;
+
+ public StubbedRemoteIterator(List<LocatedFileStatus> files) {
+ this.files = files.iterator();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return files.hasNext();
+ }
+
+ @Override
+ public LocatedFileStatus next() {
+ return files.next();
+ }
+ }
+
+ // Helper method to create a mock FileSystem
+ private static FileSystem
createMockFileSystem(RemoteIterator<LocatedFileStatus> iterator)
+ throws IOException {
+ FileSystem mockFs = mock(FileSystem.class);
+ when(mockFs.listFiles(any(org.apache.hadoop.fs.Path.class),
anyBoolean())).thenReturn(iterator);
+ return mockFs;
+ }
+
+ private static Set<Long> getModificationTimes(List<ParquetFileInfo> files) {
+ return
files.stream().map(ParquetFileInfo::getModificationTime).collect(Collectors.toSet());
+ }
+
+ private static Set<Long> modificationTimes(Long... modificationTimes) {
+ return Arrays.stream(modificationTimes).collect(Collectors.toSet());
+ }
+
+ @Test
+ void testGetMostRecentParquetFile_withMultipleFiles() throws IOException {
+ LocatedFileStatus file1 = createMockFileStatus("file1.parquet", 1000L);
+ LocatedFileStatus file2 = createMockFileStatus("file2.parquet", 3000L);
+ LocatedFileStatus file3 = createMockFileStatus("file3.parquet", 2000L);
+
+ RemoteIterator<LocatedFileStatus> iterator =
+ new StubbedRemoteIterator(Arrays.asList(file1, file2, file3));
+ FileSystem mockFs = createMockFileSystem(iterator);
+
+ ParquetDataManager manager = new ParquetDataManager(CONF, "test-path",
mockFs);
+
+ ParquetFileInfo result = manager.getMostRecentParquetFile();
+
+ assertNotNull(result);
+ assertEquals(3000L, result.getModificationTime());
+ }
+
+ @Test
+ void testGetMostRecentParquetFile_noFiles() throws IOException {
+ RemoteIterator<LocatedFileStatus> iterator = new StubbedRemoteIterator(new
ArrayList<>());
+ FileSystem mockFs = createMockFileSystem(iterator);
+
+ ParquetDataManager manager = new ParquetDataManager(CONF, "test-path",
mockFs);
+
+ IllegalStateException exception =
+ assertThrows(IllegalStateException.class,
manager::getMostRecentParquetFile);
+ assertEquals("No files found", exception.getMessage());
+ }
+
+ @Test
+ void testGetCurrentFilesInfo_multipleFiles() throws IOException {
+ LocatedFileStatus file1 = createMockFileStatus("file1.parquet", 1000L);
+ LocatedFileStatus file2 = createMockFileStatus("file2.parquet", 2000L);
+ LocatedFileStatus file3 = createMockFileStatus("file3.parquet", 3000L);
+
+ RemoteIterator<LocatedFileStatus> iterator =
+ new StubbedRemoteIterator(Arrays.asList(file1, file2, file3));
+ FileSystem mockFs = createMockFileSystem(iterator);
+
+ ParquetDataManager manager = new ParquetDataManager(CONF, "test-path",
mockFs);
+
+ Stream<ParquetFileInfo> result = manager.getCurrentFilesInfo();
+
+ assertNotNull(result);
+ List<ParquetFileInfo> fileList = result.collect(Collectors.toList());
+ assertEquals(3, fileList.size());
+ assertEquals(modificationTimes(1000L, 2000L, 3000L),
getModificationTimes(fileList));
+ }
+
+ @Test
+ void testGetCurrentFilesInfo_emptyList() throws IOException {
+ RemoteIterator<LocatedFileStatus> iterator = new
StubbedRemoteIterator(Collections.emptyList());
+ FileSystem mockFs = createMockFileSystem(iterator);
+
+ ParquetDataManager manager = new ParquetDataManager(CONF, "test-path",
mockFs);
+
+ Stream<ParquetFileInfo> result = manager.getCurrentFilesInfo();
+
+ assertNotNull(result);
+ assertEquals(0, result.count());
+ }
+
+ @Test
+ void testGetCurrentFilesInfo_streamCharacteristics() throws IOException {
+ LocatedFileStatus file1 = createMockFileStatus("file1.parquet", 1000L);
+ LocatedFileStatus file2 = createMockFileStatus("file2.parquet", 2000L);
+
+ RemoteIterator<LocatedFileStatus> iterator =
+ new StubbedRemoteIterator(Arrays.asList(file1, file2));
+ FileSystem mockFs = createMockFileSystem(iterator);
+
+ ParquetDataManager manager = new ParquetDataManager(CONF, "test-path",
mockFs);
+
+ Stream<ParquetFileInfo> result = manager.getCurrentFilesInfo();
+
+ assertNotNull(result);
+ assertTrue(result.allMatch(info -> info.getModificationTime() > 0));
+ }
+
+ @Test
+ void testGetParquetFiles_caching() throws IOException {
+ LocatedFileStatus file = createMockFileStatus("file.parquet", 1000L);
+ RemoteIterator<LocatedFileStatus> iterator =
+ new StubbedRemoteIterator(Collections.singletonList(file));
+ FileSystem mockFs = createMockFileSystem(iterator);
+
+ ParquetDataManager manager = new ParquetDataManager(CONF, "test-path",
mockFs);
+
+ // No filesystem access should happen yet
+ verify(mockFs, never()).listFiles(any(org.apache.hadoop.fs.Path.class),
anyBoolean());
+
+ // Access multiple times
+ manager.getCurrentFilesInfo();
+ manager.getMostRecentParquetFile();
+ manager.getCurrentFilesInfo();
+
+ // Verify filesystem was accessed only once
+ verify(mockFs, times(1)).listFiles(any(org.apache.hadoop.fs.Path.class),
anyBoolean());
+ }
+
+ @Test
+ void testOnlyParquetFilesIncluded() throws IOException {
+ LocatedFileStatus parquetFile = createMockFileStatus("data.parquet",
1000L);
+ LocatedFileStatus txtFile = createMockFileStatus("readme.txt", 2000L);
+ LocatedFileStatus jsonFile = createMockFileStatus("config.json", 3000L);
+
+ RemoteIterator<LocatedFileStatus> iterator =
+ new StubbedRemoteIterator(Arrays.asList(parquetFile, txtFile,
jsonFile));
+ FileSystem mockFs = createMockFileSystem(iterator);
+
+ ParquetDataManager manager = new ParquetDataManager(CONF, "test-path",
mockFs);
+
+ List<ParquetFileInfo> result =
manager.getCurrentFilesInfo().collect(Collectors.toList());
+
+ assertEquals(1, result.size());
+ assertEquals(Collections.singleton(1000L), getModificationTimes(result));
+ }
+
+ @Test
+ void testWithRealFileSystem_multipleFiles(@TempDir Path tempDir) throws
IOException {
+ // Create multiple parquet files with different modification times
+ Path file1 = tempDir.resolve("data1.parquet");
+ Path file2 = tempDir.resolve("data2.parquet");
+ Path file3 = tempDir.resolve("data3.parquet");
+
+ Files.createFile(file1);
+ Files.createFile(file2);
+ Files.createFile(file3);
+
+ // Set different modification times
+ Files.setLastModifiedTime(file1, FileTime.fromMillis(1000L));
+ Files.setLastModifiedTime(file2, FileTime.fromMillis(3000L));
+ Files.setLastModifiedTime(file3, FileTime.fromMillis(2000L));
+
+ ParquetDataManager manager = new ParquetDataManager(CONF,
tempDir.toString());
+
+ ParquetFileInfo mostRecent = manager.getMostRecentParquetFile();
+ assertNotNull(mostRecent);
+ assertEquals(3000L, mostRecent.getModificationTime());
+
+ long afterTime =
+ manager.getCurrentFilesInfo().filter(f -> f.getModificationTime() >
2000L).count();
+ assertEquals(1, afterTime);
+ }
+
+ @Test
+ void testWithRealFileSystem_nestedDirectories(@TempDir Path tempDir) throws
IOException {
+ // Create nested directory structure
+ Path subDir1 = tempDir.resolve("subdir1");
+ Path subDir2 = tempDir.resolve("subdir2");
+ Files.createDirectories(subDir1);
+ Files.createDirectories(subDir2);
+
+ // Create parquet files in different directories
+ Path file1 = tempDir.resolve("root.parquet");
+ Path file2 = subDir1.resolve("nested1.parquet");
+ Path file3 = subDir2.resolve("nested2.parquet");
+
+ Files.createFile(file1);
+ Files.createFile(file2);
+ Files.createFile(file3);
+
+ Files.setLastModifiedTime(file1, FileTime.fromMillis(1000L));
+ Files.setLastModifiedTime(file2, FileTime.fromMillis(2000L));
+ Files.setLastModifiedTime(file3, FileTime.fromMillis(3000L));
+ ParquetDataManager manager = new ParquetDataManager(CONF,
tempDir.toString());
+
+ List<ParquetFileInfo> allFiles =
manager.getCurrentFilesInfo().collect(Collectors.toList());
+ assertEquals(3, allFiles.size());
+
+ ParquetFileInfo mostRecent = manager.getMostRecentParquetFile();
+ assertEquals(3000L, mostRecent.getModificationTime());
+ }
+
+ @Test
+ void testWithRealFileSystem_mixedFileTypes(@TempDir Path tempDir) throws
IOException {
+ // Create mix of parquet and non-parquet files
+ Path parquetFile1 = tempDir.resolve("data1.parquet");
+ Path parquetFile2 = tempDir.resolve("data2.parquet");
+ Path txtFile = tempDir.resolve("readme.txt");
+ Path jsonFile = tempDir.resolve("config.json");
+
+ Files.createFile(parquetFile1);
+ Files.createFile(parquetFile2);
+ Files.createFile(txtFile);
+ Files.createFile(jsonFile);
+
+ Files.setLastModifiedTime(parquetFile1, FileTime.fromMillis(1000L));
+ Files.setLastModifiedTime(parquetFile2, FileTime.fromMillis(2000L));
+ Files.setLastModifiedTime(txtFile, FileTime.fromMillis(3000L));
+ Files.setLastModifiedTime(jsonFile, FileTime.fromMillis(4000L));
+
+ ParquetDataManager manager = new ParquetDataManager(CONF,
tempDir.toString());
+
+ List<ParquetFileInfo> allFiles =
manager.getCurrentFilesInfo().collect(Collectors.toList());
+ // Only parquet files should be included
+ assertEquals(2, allFiles.size());
+
+ ParquetFileInfo mostRecent = manager.getMostRecentParquetFile();
+ // Most recent parquet file, not the txt or json
+ assertEquals(2000L, mostRecent.getModificationTime());
+ }
+}
