You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by as...@apache.org on 2024/03/27 20:28:11 UTC
(incubator-xtable) branch main updated: Refactor, rename data file classes with One in prefix
This is an automated email from the ASF dual-hosted git repository.
ashvin 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 8a4dc1af Refactor, rename data file classes with One in prefix
8a4dc1af is described below
commit 8a4dc1af64815d5b5192591aef4b505e896cadbc
Author: Ashvin Agrawal <as...@apache.org>
AuthorDate: Tue Mar 26 10:19:07 2024 -0700
Refactor, rename data file classes with One in prefix
The primary goal of this change is to improve the code readability. The change replaces the class
names's old product-specific prefix (One) with a prefix that more accurately reflects the class's
function. The change targets OneDataFile and OneDataFileDiff classes. These are internal classes
that hold and transfer information about a table's data file.
This change focuses on code refactoring and does not introduce new tests.
Addresses #394, #396
---
.../java/org/apache/xtable/model/TableChange.java | 4 +-
.../java/org/apache/xtable/model/stat/Range.java | 6 +-
.../apache/xtable/model/storage/DataFilesDiff.java | 94 +++++------------
.../storage/{DataFilesDiff.java => FilesDiff.java} | 18 ++--
.../{OneDataFile.java => InternalDataFile.java} | 2 +-
.../xtable/model/storage/OneDataFilesDiff.java | 58 -----------
.../apache/xtable/model/storage/OneFileGroup.java | 12 +--
.../xtable/spi/extractor/DataFileIterator.java | 4 +-
.../org/apache/xtable/spi/sync/TargetClient.java | 6 +-
.../xtable/model/storage/TestDataFilesDiff.java | 111 ++++-----------------
.../{TestDataFilesDiff.java => TestFilesDiff.java} | 23 ++---
.../xtable/model/storage/TestOneDataFilesDiff.java | 52 ----------
.../spi/extractor/TestExtractFromSource.java | 26 ++---
.../xtable/spi/sync/TestTableFormatSync.java | 26 ++---
.../apache/xtable/delta/DeltaActionsConverter.java | 10 +-
.../java/org/apache/xtable/delta/DeltaClient.java | 6 +-
.../xtable/delta/DeltaDataFileExtractor.java | 8 +-
.../delta/DeltaDataFileUpdatesExtractor.java | 20 ++--
.../xtable/delta/DeltaPartitionExtractor.java | 9 +-
.../org/apache/xtable/delta/DeltaSourceClient.java | 22 ++--
.../xtable/hudi/BaseFileUpdatesExtractor.java | 27 ++---
.../apache/xtable/hudi/HudiDataFileExtractor.java | 48 ++++-----
.../apache/xtable/hudi/HudiFileStatsExtractor.java | 18 ++--
.../org/apache/xtable/hudi/HudiTargetClient.java | 6 +-
.../org/apache/xtable/iceberg/IcebergClient.java | 6 +-
.../xtable/iceberg/IcebergDataFileExtractor.java | 10 +-
.../xtable/iceberg/IcebergDataFileUpdatesSync.java | 19 ++--
.../apache/xtable/iceberg/IcebergSourceClient.java | 25 ++---
.../org/apache/xtable/ValidationTestHelper.java | 10 +-
.../apache/xtable/delta/ITDeltaSourceClient.java | 20 ++--
.../org/apache/xtable/delta/TestDeltaSync.java | 44 ++++----
.../org/apache/xtable/hudi/ITHudiTargetClient.java | 24 ++---
.../xtable/hudi/TestBaseFileUpdatesExtractor.java | 42 ++++----
.../xtable/hudi/TestHudiFileStatsExtractor.java | 18 ++--
.../apache/xtable/hudi/TestHudiTargetClient.java | 4 +-
.../xtable/iceberg/ITIcebergSourceClient.java | 4 +-
.../xtable/iceberg/TestIcebergSourceClient.java | 29 +++---
.../org/apache/xtable/iceberg/TestIcebergSync.java | 66 ++++++------
38 files changed, 358 insertions(+), 579 deletions(-)
diff --git a/api/src/main/java/org/apache/xtable/model/TableChange.java b/api/src/main/java/org/apache/xtable/model/TableChange.java
index 76627c9f..55996770 100644
--- a/api/src/main/java/org/apache/xtable/model/TableChange.java
+++ b/api/src/main/java/org/apache/xtable/model/TableChange.java
@@ -21,7 +21,7 @@ package org.apache.xtable.model;
import lombok.Builder;
import lombok.Value;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.DataFilesDiff;
/**
* Captures the changes in a single commit/instant from the source table.
@@ -32,7 +32,7 @@ import org.apache.xtable.model.storage.OneDataFilesDiff;
@Builder(toBuilder = true)
public class TableChange {
// Change in files at the specified instant
- OneDataFilesDiff filesDiff;
+ DataFilesDiff filesDiff;
/** The {@link OneTable} at the commit time to which this table change belongs. */
OneTable tableAsOfChange;
diff --git a/api/src/main/java/org/apache/xtable/model/stat/Range.java b/api/src/main/java/org/apache/xtable/model/stat/Range.java
index dbba7db0..80e56424 100644
--- a/api/src/main/java/org/apache/xtable/model/stat/Range.java
+++ b/api/src/main/java/org/apache/xtable/model/stat/Range.java
@@ -25,14 +25,14 @@ import lombok.Value;
import org.apache.xtable.model.schema.OneSchema;
import org.apache.xtable.model.schema.OneType;
import org.apache.xtable.model.schema.PartitionTransformType;
-import org.apache.xtable.model.storage.OneDataFile;
+import org.apache.xtable.model.storage.InternalDataFile;
/**
* Represents a range of values in the specified data type. Can represent a scalar value when the
* {@link #minValue} and {@link #maxValue} are the same.
*
- * <p>For the ranges stored in {@link OneDataFile#getPartitionValues()}, the values will be based
- * off the {@link PartitionTransformType}. {@link PartitionTransformType#HOUR}, {@link
+ * <p>For the ranges stored in {@link InternalDataFile#getPartitionValues()}, the values will be
+ * based off the {@link PartitionTransformType}. {@link PartitionTransformType#HOUR}, {@link
* PartitionTransformType#DAY}, {@link PartitionTransformType#MONTH}, {@link
* PartitionTransformType#YEAR} are all stored as a long representing a point in time as
* milliseconds since epoch. {@link PartitionTransformType#VALUE} will match the rules below.
diff --git a/api/src/main/java/org/apache/xtable/model/storage/DataFilesDiff.java b/api/src/main/java/org/apache/xtable/model/storage/DataFilesDiff.java
index 4a42cc3a..9025ec31 100644
--- a/api/src/main/java/org/apache/xtable/model/storage/DataFilesDiff.java
+++ b/api/src/main/java/org/apache/xtable/model/storage/DataFilesDiff.java
@@ -18,86 +18,42 @@
package org.apache.xtable.model.storage;
-import java.util.*;
+import java.util.List;
+import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
-import lombok.Data;
-import lombok.Singular;
+import lombok.EqualsAndHashCode;
+import lombok.Value;
import lombok.experimental.SuperBuilder;
-/**
- * Holds the collection of files that represent the difference between two states/commits/snapshots
- * of a table with respect to the data files. Between any two states of a table, the newer/latest
- * state may contain new files not present in the older state and may have removed files that were
- * present in the older state. In most cases the data files included in the newer state are derived
- * from a new commit in a source table format that has not been applied to a target table format
- * yet. Hence, the collection of data files in the newer state are typically {@link OneDataFile}s,
- * whereas the files in the older state are represented using a generic type P which can be a data
- * file type in specific to the target table format.
- *
- * @param <L> the type of the files in the latest state
- * @param <P> the type of the files in the target table format
- */
-@Data
+/** Container for holding the list of files added and files removed between source and target. */
+@Value
+@EqualsAndHashCode(callSuper = true)
@SuperBuilder
-public class DataFilesDiff<L, P> {
- @Singular("fileAdded")
- private Set<L> filesAdded;
-
- @Singular("fileRemoved")
- private Set<P> filesRemoved;
+public class DataFilesDiff extends FilesDiff<InternalDataFile, InternalDataFile> {
/**
- * Compares the latest files with the previous files and identifies the files that are new, i.e.
- * are present in latest files buy not present in the previously known files, and the files that
- * are removed, i.e. present in the previously known files but not present in the latest files.
+ * Creates a DataFilesDiff from the list of files in the target table and the list of files in the
+ * source table.
*
- * @param latestFiles a map of file path and file object representing files in the latest snapshot
- * of a table
- * @param previousFiles a map of file path and file object representing files in a previously
- * synced snapshot of a table.
- * @param <P> the type of the previous files
- * @return the diff of the files
+ * @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 <L, P> DataFilesDiff<L, P> findNewAndRemovedFiles(
- Map<String, L> latestFiles, Map<String, P> previousFiles) {
- Set<L> newFiles = new HashSet<>();
- Map<String, P> removedFiles = new HashMap<>(previousFiles);
+ public static DataFilesDiff from(List<InternalDataFile> source, List<InternalDataFile> target) {
+ Map<String, InternalDataFile> targetPaths =
+ target.stream()
+ .collect(Collectors.toMap(InternalDataFile::getPhysicalPath, Function.identity()));
+ Map<String, InternalDataFile> sourcePaths =
+ source.stream()
+ .collect(Collectors.toMap(InternalDataFile::getPhysicalPath, Function.identity()));
- // if a file in latest files is also present in previous files, then it is neither new nor
- // removed.
- latestFiles.forEach(
- (key, value) -> {
- boolean notAKnownFile = removedFiles.remove(key) == null;
- if (notAKnownFile) {
- newFiles.add(value);
- }
- });
- return DataFilesDiff.<L, P>builder()
- .filesAdded(newFiles)
- .filesRemoved(removedFiles.values())
+ FilesDiff<InternalDataFile, InternalDataFile> diff =
+ findNewAndRemovedFiles(sourcePaths, targetPaths);
+ return DataFilesDiff.builder()
+ .filesAdded(diff.getFilesAdded())
+ .filesRemoved(diff.getFilesRemoved())
.build();
}
-
- /**
- * This method wraps the {@link #findNewAndRemovedFiles(Map, Map)} method, to compare the latest
- * file groups with the previous files and identifies the files that are new, i.e. are present in
- * latest files buy not present in the previously known files, and the files that are removed,
- * i.e. present in the previously known files but not present in the latest files.
- *
- * @param latestFileGroups a list of file groups representing the latest snapshot of a table
- * @param previousFiles a map of file path and file object representing files in a previously
- * synced snapshot of a table
- * @param <P> the type of the previous files
- * @return the set of files that are added
- */
- public static <P> DataFilesDiff<OneDataFile, P> findNewAndRemovedFiles(
- List<OneFileGroup> latestFileGroups, Map<String, P> previousFiles) {
- Map<String, OneDataFile> latestFiles =
- latestFileGroups.stream()
- .flatMap(group -> group.getFiles().stream())
- .collect(Collectors.toMap(OneDataFile::getPhysicalPath, Function.identity()));
- return findNewAndRemovedFiles(latestFiles, previousFiles);
- }
}
diff --git a/api/src/main/java/org/apache/xtable/model/storage/DataFilesDiff.java b/api/src/main/java/org/apache/xtable/model/storage/FilesDiff.java
similarity index 88%
copy from api/src/main/java/org/apache/xtable/model/storage/DataFilesDiff.java
copy to api/src/main/java/org/apache/xtable/model/storage/FilesDiff.java
index 4a42cc3a..0657b9f9 100644
--- a/api/src/main/java/org/apache/xtable/model/storage/DataFilesDiff.java
+++ b/api/src/main/java/org/apache/xtable/model/storage/FilesDiff.java
@@ -32,16 +32,16 @@ import lombok.experimental.SuperBuilder;
* state may contain new files not present in the older state and may have removed files that were
* present in the older state. In most cases the data files included in the newer state are derived
* from a new commit in a source table format that has not been applied to a target table format
- * yet. Hence, the collection of data files in the newer state are typically {@link OneDataFile}s,
- * whereas the files in the older state are represented using a generic type P which can be a data
- * file type in specific to the target table format.
+ * yet. Hence, the collection of data files in the newer state are typically {@link
+ * InternalDataFile}s, whereas the files in the older state are represented using a generic type P
+ * which can be a data file type in specific to the target table format.
*
* @param <L> the type of the files in the latest state
* @param <P> the type of the files in the target table format
*/
@Data
@SuperBuilder
-public class DataFilesDiff<L, P> {
+public class FilesDiff<L, P> {
@Singular("fileAdded")
private Set<L> filesAdded;
@@ -60,7 +60,7 @@ public class DataFilesDiff<L, P> {
* @param <P> the type of the previous files
* @return the diff of the files
*/
- public static <L, P> DataFilesDiff<L, P> findNewAndRemovedFiles(
+ public static <L, P> FilesDiff<L, P> findNewAndRemovedFiles(
Map<String, L> latestFiles, Map<String, P> previousFiles) {
Set<L> newFiles = new HashSet<>();
Map<String, P> removedFiles = new HashMap<>(previousFiles);
@@ -74,7 +74,7 @@ public class DataFilesDiff<L, P> {
newFiles.add(value);
}
});
- return DataFilesDiff.<L, P>builder()
+ return FilesDiff.<L, P>builder()
.filesAdded(newFiles)
.filesRemoved(removedFiles.values())
.build();
@@ -92,12 +92,12 @@ public class DataFilesDiff<L, P> {
* @param <P> the type of the previous files
* @return the set of files that are added
*/
- public static <P> DataFilesDiff<OneDataFile, P> findNewAndRemovedFiles(
+ public static <P> FilesDiff<InternalDataFile, P> findNewAndRemovedFiles(
List<OneFileGroup> latestFileGroups, Map<String, P> previousFiles) {
- Map<String, OneDataFile> latestFiles =
+ Map<String, InternalDataFile> latestFiles =
latestFileGroups.stream()
.flatMap(group -> group.getFiles().stream())
- .collect(Collectors.toMap(OneDataFile::getPhysicalPath, Function.identity()));
+ .collect(Collectors.toMap(InternalDataFile::getPhysicalPath, Function.identity()));
return findNewAndRemovedFiles(latestFiles, previousFiles);
}
}
diff --git a/api/src/main/java/org/apache/xtable/model/storage/OneDataFile.java b/api/src/main/java/org/apache/xtable/model/storage/InternalDataFile.java
similarity index 98%
rename from api/src/main/java/org/apache/xtable/model/storage/OneDataFile.java
rename to api/src/main/java/org/apache/xtable/model/storage/InternalDataFile.java
index b89e066b..52effbe2 100644
--- a/api/src/main/java/org/apache/xtable/model/storage/OneDataFile.java
+++ b/api/src/main/java/org/apache/xtable/model/storage/InternalDataFile.java
@@ -36,7 +36,7 @@ import org.apache.xtable.model.stat.PartitionValue;
*/
@Builder(toBuilder = true)
@Value
-public class OneDataFile {
+public class InternalDataFile {
// written schema version
SchemaVersion schemaVersion;
// physical path of the file (absolute with scheme)
diff --git a/api/src/main/java/org/apache/xtable/model/storage/OneDataFilesDiff.java b/api/src/main/java/org/apache/xtable/model/storage/OneDataFilesDiff.java
deleted file mode 100644
index f01ba7f4..00000000
--- a/api/src/main/java/org/apache/xtable/model/storage/OneDataFilesDiff.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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 java.util.List;
-import java.util.Map;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-import lombok.EqualsAndHashCode;
-import lombok.Value;
-import lombok.experimental.SuperBuilder;
-
-/** Container for holding the list of files added and files removed between source and target. */
-@Value
-@EqualsAndHashCode(callSuper = true)
-@SuperBuilder
-public class OneDataFilesDiff extends DataFilesDiff<OneDataFile, OneDataFile> {
-
- /**
- * Creates a OneDataFilesDiff 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 OneDataFilesDiff from(List<OneDataFile> source, List<OneDataFile> target) {
- Map<String, OneDataFile> targetPaths =
- target.stream()
- .collect(Collectors.toMap(OneDataFile::getPhysicalPath, Function.identity()));
- Map<String, OneDataFile> sourcePaths =
- source.stream()
- .collect(Collectors.toMap(OneDataFile::getPhysicalPath, Function.identity()));
-
- DataFilesDiff<OneDataFile, OneDataFile> diff = findNewAndRemovedFiles(sourcePaths, targetPaths);
- return OneDataFilesDiff.builder()
- .filesAdded(diff.getFilesAdded())
- .filesRemoved(diff.getFilesRemoved())
- .build();
- }
-}
diff --git a/api/src/main/java/org/apache/xtable/model/storage/OneFileGroup.java b/api/src/main/java/org/apache/xtable/model/storage/OneFileGroup.java
index 219a97a9..32c73f25 100644
--- a/api/src/main/java/org/apache/xtable/model/storage/OneFileGroup.java
+++ b/api/src/main/java/org/apache/xtable/model/storage/OneFileGroup.java
@@ -28,20 +28,20 @@ import lombok.Value;
import org.apache.xtable.model.stat.PartitionValue;
-/** Represents a grouping of {@link OneDataFile} with the same partition values. */
+/** Represents a grouping of {@link InternalDataFile} with the same partition values. */
@Value
@Builder
public class OneFileGroup {
List<PartitionValue> partitionValues;
- List<OneDataFile> files;
+ List<InternalDataFile> files;
- public static List<OneFileGroup> fromFiles(List<OneDataFile> files) {
+ public static List<OneFileGroup> fromFiles(List<InternalDataFile> files) {
return fromFiles(files.stream());
}
- public static List<OneFileGroup> fromFiles(Stream<OneDataFile> files) {
- Map<List<PartitionValue>, List<OneDataFile>> filesGrouped =
- files.collect(Collectors.groupingBy(OneDataFile::getPartitionValues));
+ public static List<OneFileGroup> fromFiles(Stream<InternalDataFile> files) {
+ Map<List<PartitionValue>, List<InternalDataFile>> filesGrouped =
+ files.collect(Collectors.groupingBy(InternalDataFile::getPartitionValues));
return filesGrouped.entrySet().stream()
.map(
entry ->
diff --git a/api/src/main/java/org/apache/xtable/spi/extractor/DataFileIterator.java b/api/src/main/java/org/apache/xtable/spi/extractor/DataFileIterator.java
index 767650e9..d5ece6ec 100644
--- a/api/src/main/java/org/apache/xtable/spi/extractor/DataFileIterator.java
+++ b/api/src/main/java/org/apache/xtable/spi/extractor/DataFileIterator.java
@@ -20,7 +20,7 @@ package org.apache.xtable.spi.extractor;
import java.util.Iterator;
-import org.apache.xtable.model.storage.OneDataFile;
+import org.apache.xtable.model.storage.InternalDataFile;
/** DataFileIterator lets the consumer iterate over the files of a table. */
-public interface DataFileIterator extends Iterator<OneDataFile>, AutoCloseable {}
+public interface DataFileIterator extends Iterator<InternalDataFile>, AutoCloseable {}
diff --git a/api/src/main/java/org/apache/xtable/spi/sync/TargetClient.java b/api/src/main/java/org/apache/xtable/spi/sync/TargetClient.java
index ad647649..db216246 100644
--- a/api/src/main/java/org/apache/xtable/spi/sync/TargetClient.java
+++ b/api/src/main/java/org/apache/xtable/spi/sync/TargetClient.java
@@ -28,7 +28,7 @@ import org.apache.xtable.model.OneTable;
import org.apache.xtable.model.OneTableMetadata;
import org.apache.xtable.model.schema.OnePartitionField;
import org.apache.xtable.model.schema.OneSchema;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.OneFileGroup;
/** A client that provides the major functionality for syncing changes to a target system. */
@@ -68,9 +68,9 @@ public interface TargetClient {
* Syncs the changes in files to the target system. This method is required to both add and remove
* files.
*
- * @param oneDataFilesDiff the diff that needs to be synced
+ * @param dataFilesDiff the diff that needs to be synced
*/
- void syncFilesForDiff(OneDataFilesDiff oneDataFilesDiff);
+ void syncFilesForDiff(DataFilesDiff dataFilesDiff);
/**
* Starts the sync and performs any initialization required
diff --git a/api/src/test/java/org/apache/xtable/model/storage/TestDataFilesDiff.java b/api/src/test/java/org/apache/xtable/model/storage/TestDataFilesDiff.java
index 22e6ffeb..6896dce8 100644
--- a/api/src/test/java/org/apache/xtable/model/storage/TestDataFilesDiff.java
+++ b/api/src/test/java/org/apache/xtable/model/storage/TestDataFilesDiff.java
@@ -19,101 +19,34 @@
package org.apache.xtable.model.storage;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.Mockito.mock;
-import java.io.File;
import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
import org.junit.jupiter.api.Test;
public class TestDataFilesDiff {
@Test
- void findDiffFromFileGroups() {
- OneDataFile file1Group1 = OneDataFile.builder().physicalPath("file1Group1").build();
- OneDataFile file2Group1 = OneDataFile.builder().physicalPath("file2Group1").build();
- OneDataFile file1Group2 = OneDataFile.builder().physicalPath("file1Group2").build();
- OneDataFile file2Group2 = OneDataFile.builder().physicalPath("file2Group2").build();
-
- List<OneFileGroup> latestFileGroups =
- OneFileGroup.fromFiles(Arrays.asList(file1Group1, file2Group1, file1Group2, file2Group2));
-
- Map<String, File> previousFiles = new HashMap<>();
- File file1 = mock(File.class);
- File file2 = mock(File.class);
- File file3 = mock(File.class);
- previousFiles.put("file1Group1", file1);
- previousFiles.put("file2NoGroup", file2);
- previousFiles.put("file2Group2", file3);
-
- DataFilesDiff<OneDataFile, File> diff =
- DataFilesDiff.findNewAndRemovedFiles(latestFileGroups, previousFiles);
- assertEquals(2, diff.getFilesAdded().size());
- assertTrue(diff.getFilesAdded().contains(file1Group2));
- assertTrue(diff.getFilesAdded().contains(file2Group1));
- assertEquals(1, diff.getFilesRemoved().size());
- assertTrue(diff.getFilesRemoved().contains(file2));
- }
-
- @Test
- void findDiffFromFilesNoPrevious() {
- File file1 = mock(File.class);
- File file2 = mock(File.class);
-
- Map<String, File> previousFiles = new HashMap<>();
- Map<String, File> latestFiles = new HashMap<>();
- latestFiles.put("file1", file1);
- latestFiles.put("file2", file2);
-
- DataFilesDiff<File, File> diff =
- DataFilesDiff.findNewAndRemovedFiles(latestFiles, previousFiles);
- assertEquals(0, diff.getFilesRemoved().size());
- assertEquals(2, diff.getFilesAdded().size());
- assertTrue(diff.getFilesAdded().contains(file1));
- assertTrue(diff.getFilesAdded().contains(file2));
- }
-
- @Test
- void findDiffFromFilesNoNew() {
- File file1 = mock(File.class);
- File file2 = mock(File.class);
-
- Map<String, File> previousFiles = new HashMap<>();
- previousFiles.put("file1", file1);
- previousFiles.put("file2", file2);
-
- Map<String, File> latestFiles = new HashMap<>();
- latestFiles.put("file1", file1);
- latestFiles.put("file2", file2);
-
- DataFilesDiff<File, File> diff =
- DataFilesDiff.findNewAndRemovedFiles(latestFiles, previousFiles);
- assertEquals(0, diff.getFilesRemoved().size());
- assertEquals(0, diff.getFilesAdded().size());
- }
-
- @Test
- void findDiffFromFiles() {
- File file1 = mock(File.class);
- File file2 = mock(File.class);
- File file3 = mock(File.class);
-
- Map<String, File> previousFiles = new HashMap<>();
- previousFiles.put("file1", file1);
- previousFiles.put("file2", file2);
-
- Map<String, File> latestFiles = new HashMap<>();
- latestFiles.put("file2", file2);
- latestFiles.put("file3", file3);
-
- DataFilesDiff<File, File> diff =
- DataFilesDiff.findNewAndRemovedFiles(latestFiles, previousFiles);
- assertEquals(1, diff.getFilesAdded().size());
- assertTrue(diff.getFilesAdded().contains(file3));
- assertEquals(1, diff.getFilesRemoved().size());
- assertTrue(diff.getFilesRemoved().contains(file1));
+ void testFrom() {
+ InternalDataFile sourceFile1 =
+ InternalDataFile.builder().physicalPath("file://new_source_file1.parquet").build();
+ InternalDataFile sourceFile2 =
+ InternalDataFile.builder().physicalPath("file://new_source_file2.parquet").build();
+ InternalDataFile targetFile1 =
+ InternalDataFile.builder().physicalPath("file://already_in_target1.parquet").build();
+ InternalDataFile targetFile2 =
+ InternalDataFile.builder().physicalPath("file://already_in_target2.parquet").build();
+ InternalDataFile sourceFileInTargetAlready =
+ InternalDataFile.builder().physicalPath("file://already_in_target3.parquet").build();
+ DataFilesDiff actual =
+ DataFilesDiff.from(
+ Arrays.asList(sourceFile1, sourceFile2, sourceFileInTargetAlready),
+ Arrays.asList(targetFile1, targetFile2, sourceFileInTargetAlready));
+
+ DataFilesDiff expected =
+ DataFilesDiff.builder()
+ .filesAdded(Arrays.asList(sourceFile1, sourceFile2))
+ .filesRemoved(Arrays.asList(targetFile1, targetFile2))
+ .build();
+ assertEquals(expected, actual);
}
}
diff --git a/api/src/test/java/org/apache/xtable/model/storage/TestDataFilesDiff.java b/api/src/test/java/org/apache/xtable/model/storage/TestFilesDiff.java
similarity index 80%
copy from api/src/test/java/org/apache/xtable/model/storage/TestDataFilesDiff.java
copy to api/src/test/java/org/apache/xtable/model/storage/TestFilesDiff.java
index 22e6ffeb..c52f7e03 100644
--- a/api/src/test/java/org/apache/xtable/model/storage/TestDataFilesDiff.java
+++ b/api/src/test/java/org/apache/xtable/model/storage/TestFilesDiff.java
@@ -30,13 +30,13 @@ import java.util.Map;
import org.junit.jupiter.api.Test;
-public class TestDataFilesDiff {
+public class TestFilesDiff {
@Test
void findDiffFromFileGroups() {
- OneDataFile file1Group1 = OneDataFile.builder().physicalPath("file1Group1").build();
- OneDataFile file2Group1 = OneDataFile.builder().physicalPath("file2Group1").build();
- OneDataFile file1Group2 = OneDataFile.builder().physicalPath("file1Group2").build();
- OneDataFile file2Group2 = OneDataFile.builder().physicalPath("file2Group2").build();
+ InternalDataFile file1Group1 = InternalDataFile.builder().physicalPath("file1Group1").build();
+ InternalDataFile file2Group1 = InternalDataFile.builder().physicalPath("file2Group1").build();
+ InternalDataFile file1Group2 = InternalDataFile.builder().physicalPath("file1Group2").build();
+ InternalDataFile file2Group2 = InternalDataFile.builder().physicalPath("file2Group2").build();
List<OneFileGroup> latestFileGroups =
OneFileGroup.fromFiles(Arrays.asList(file1Group1, file2Group1, file1Group2, file2Group2));
@@ -49,8 +49,8 @@ public class TestDataFilesDiff {
previousFiles.put("file2NoGroup", file2);
previousFiles.put("file2Group2", file3);
- DataFilesDiff<OneDataFile, File> diff =
- DataFilesDiff.findNewAndRemovedFiles(latestFileGroups, previousFiles);
+ FilesDiff<InternalDataFile, File> diff =
+ FilesDiff.findNewAndRemovedFiles(latestFileGroups, previousFiles);
assertEquals(2, diff.getFilesAdded().size());
assertTrue(diff.getFilesAdded().contains(file1Group2));
assertTrue(diff.getFilesAdded().contains(file2Group1));
@@ -68,8 +68,7 @@ public class TestDataFilesDiff {
latestFiles.put("file1", file1);
latestFiles.put("file2", file2);
- DataFilesDiff<File, File> diff =
- DataFilesDiff.findNewAndRemovedFiles(latestFiles, previousFiles);
+ FilesDiff<File, File> diff = FilesDiff.findNewAndRemovedFiles(latestFiles, previousFiles);
assertEquals(0, diff.getFilesRemoved().size());
assertEquals(2, diff.getFilesAdded().size());
assertTrue(diff.getFilesAdded().contains(file1));
@@ -89,8 +88,7 @@ public class TestDataFilesDiff {
latestFiles.put("file1", file1);
latestFiles.put("file2", file2);
- DataFilesDiff<File, File> diff =
- DataFilesDiff.findNewAndRemovedFiles(latestFiles, previousFiles);
+ FilesDiff<File, File> diff = FilesDiff.findNewAndRemovedFiles(latestFiles, previousFiles);
assertEquals(0, diff.getFilesRemoved().size());
assertEquals(0, diff.getFilesAdded().size());
}
@@ -109,8 +107,7 @@ public class TestDataFilesDiff {
latestFiles.put("file2", file2);
latestFiles.put("file3", file3);
- DataFilesDiff<File, File> diff =
- DataFilesDiff.findNewAndRemovedFiles(latestFiles, previousFiles);
+ FilesDiff<File, File> diff = FilesDiff.findNewAndRemovedFiles(latestFiles, previousFiles);
assertEquals(1, diff.getFilesAdded().size());
assertTrue(diff.getFilesAdded().contains(file3));
assertEquals(1, diff.getFilesRemoved().size());
diff --git a/api/src/test/java/org/apache/xtable/model/storage/TestOneDataFilesDiff.java b/api/src/test/java/org/apache/xtable/model/storage/TestOneDataFilesDiff.java
deleted file mode 100644
index bfe38f0a..00000000
--- a/api/src/test/java/org/apache/xtable/model/storage/TestOneDataFilesDiff.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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 static org.junit.jupiter.api.Assertions.assertEquals;
-
-import java.util.Arrays;
-
-import org.junit.jupiter.api.Test;
-
-public class TestOneDataFilesDiff {
- @Test
- void testFrom() {
- OneDataFile sourceFile1 =
- OneDataFile.builder().physicalPath("file://new_source_file1.parquet").build();
- OneDataFile sourceFile2 =
- OneDataFile.builder().physicalPath("file://new_source_file2.parquet").build();
- OneDataFile targetFile1 =
- OneDataFile.builder().physicalPath("file://already_in_target1.parquet").build();
- OneDataFile targetFile2 =
- OneDataFile.builder().physicalPath("file://already_in_target2.parquet").build();
- OneDataFile sourceFileInTargetAlready =
- OneDataFile.builder().physicalPath("file://already_in_target3.parquet").build();
- OneDataFilesDiff actual =
- OneDataFilesDiff.from(
- Arrays.asList(sourceFile1, sourceFile2, sourceFileInTargetAlready),
- Arrays.asList(targetFile1, targetFile2, sourceFileInTargetAlready));
-
- OneDataFilesDiff expected =
- OneDataFilesDiff.builder()
- .filesAdded(Arrays.asList(sourceFile1, sourceFile2))
- .filesRemoved(Arrays.asList(targetFile1, targetFile2))
- .build();
- assertEquals(expected, actual);
- }
-}
diff --git a/api/src/test/java/org/apache/xtable/spi/extractor/TestExtractFromSource.java b/api/src/test/java/org/apache/xtable/spi/extractor/TestExtractFromSource.java
index 1f97b48a..9be69b56 100644
--- a/api/src/test/java/org/apache/xtable/spi/extractor/TestExtractFromSource.java
+++ b/api/src/test/java/org/apache/xtable/spi/extractor/TestExtractFromSource.java
@@ -42,8 +42,8 @@ import org.apache.xtable.model.OneSnapshot;
import org.apache.xtable.model.OneTable;
import org.apache.xtable.model.TableChange;
import org.apache.xtable.model.schema.SchemaCatalog;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.DataFilesDiff;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.OneFileGroup;
public class TestExtractFromSource {
@@ -66,8 +66,8 @@ public class TestExtractFromSource {
@Test
public void extractTableChanges() {
- OneDataFile initialFile2 = getOneDataFile("file2.parquet");
- OneDataFile initialFile3 = getOneDataFile("file3.parquet");
+ InternalDataFile initialFile2 = getDataFile("file2.parquet");
+ InternalDataFile initialFile3 = getDataFile("file3.parquet");
Instant lastSyncTime = Instant.now().minus(2, ChronoUnit.DAYS);
TestCommit firstCommitToSync = TestCommit.of("first_commit");
@@ -84,14 +84,14 @@ public class TestExtractFromSource {
.thenReturn(commitsBacklogToReturn);
// drop a file and add a file
- OneDataFile newFile1 = getOneDataFile("file4.parquet");
+ InternalDataFile newFile1 = getDataFile("file4.parquet");
OneTable tableAtFirstInstant =
OneTable.builder().latestCommitTime(Instant.now().minus(1, ChronoUnit.DAYS)).build();
TableChange tableChangeToReturnAtFirstInstant =
TableChange.builder()
.tableAsOfChange(tableAtFirstInstant)
.filesDiff(
- OneDataFilesDiff.builder().fileAdded(newFile1).fileRemoved(initialFile2).build())
+ DataFilesDiff.builder().fileAdded(newFile1).fileRemoved(initialFile2).build())
.build();
when(mockSourceClient.getTableChangeForCommit(firstCommitToSync))
.thenReturn(tableChangeToReturnAtFirstInstant);
@@ -99,19 +99,19 @@ public class TestExtractFromSource {
TableChange.builder()
.tableAsOfChange(tableAtFirstInstant)
.filesDiff(
- OneDataFilesDiff.builder().fileAdded(newFile1).fileRemoved(initialFile2).build())
+ DataFilesDiff.builder().fileAdded(newFile1).fileRemoved(initialFile2).build())
.build();
// add 2 new files, remove 2 files
- OneDataFile newFile2 = getOneDataFile("file5.parquet");
- OneDataFile newFile3 = getOneDataFile("file6.parquet");
+ InternalDataFile newFile2 = getDataFile("file5.parquet");
+ InternalDataFile newFile3 = getDataFile("file6.parquet");
OneTable tableAtSecondInstant = OneTable.builder().latestCommitTime(Instant.now()).build();
TableChange tableChangeToReturnAtSecondInstant =
TableChange.builder()
.tableAsOfChange(tableAtSecondInstant)
.filesDiff(
- OneDataFilesDiff.builder()
+ DataFilesDiff.builder()
.filesAdded(Arrays.asList(newFile2, newFile3))
.filesRemoved(Arrays.asList(initialFile3, newFile1))
.build())
@@ -122,7 +122,7 @@ public class TestExtractFromSource {
TableChange.builder()
.tableAsOfChange(tableAtSecondInstant)
.filesDiff(
- OneDataFilesDiff.builder()
+ DataFilesDiff.builder()
.filesAdded(Arrays.asList(newFile2, newFile3))
.filesRemoved(Arrays.asList(initialFile3, newFile1))
.build())
@@ -137,8 +137,8 @@ public class TestExtractFromSource {
Arrays.asList(expectedFirstTableChange, expectedSecondTableChange), actualTableChanges);
}
- private OneDataFile getOneDataFile(String physicalPath) {
- return OneDataFile.builder().physicalPath(physicalPath).build();
+ private InternalDataFile getDataFile(String physicalPath) {
+ return InternalDataFile.builder().physicalPath(physicalPath).build();
}
@AllArgsConstructor(staticName = "of")
diff --git a/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java b/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java
index e0e57cee..98363b3f 100644
--- a/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java
+++ b/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java
@@ -48,8 +48,8 @@ import org.apache.xtable.model.schema.OneField;
import org.apache.xtable.model.schema.OnePartitionField;
import org.apache.xtable.model.schema.OneSchema;
import org.apache.xtable.model.schema.PartitionTransformType;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.DataFilesDiff;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.OneFileGroup;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.model.sync.SyncMode;
@@ -68,7 +68,7 @@ public class TestTableFormatSync {
OneFileGroup.builder()
.files(
Collections.singletonList(
- OneDataFile.builder().physicalPath("/tmp/path/file.parquet").build()))
+ InternalDataFile.builder().physicalPath("/tmp/path/file.parquet").build()))
.build());
List<Instant> pendingCommitInstants = Collections.singletonList(Instant.now());
OneSnapshot snapshot =
@@ -119,15 +119,15 @@ public class TestTableFormatSync {
void syncChangesWithFailureForOneFormat() {
Instant start = Instant.now();
OneTable tableState1 = getTableState(1);
- OneDataFilesDiff dataFilesDiff1 = getFilesDiff(1);
+ DataFilesDiff dataFilesDiff1 = getFilesDiff(1);
TableChange tableChange1 =
TableChange.builder().tableAsOfChange(tableState1).filesDiff(dataFilesDiff1).build();
OneTable tableState2 = getTableState(2);
- OneDataFilesDiff dataFilesDiff2 = getFilesDiff(2);
+ DataFilesDiff dataFilesDiff2 = getFilesDiff(2);
TableChange tableChange2 =
TableChange.builder().tableAsOfChange(tableState2).filesDiff(dataFilesDiff2).build();
OneTable tableState3 = getTableState(3);
- OneDataFilesDiff dataFilesDiff3 = getFilesDiff(3);
+ DataFilesDiff dataFilesDiff3 = getFilesDiff(3);
TableChange tableChange3 =
TableChange.builder().tableAsOfChange(tableState3).filesDiff(dataFilesDiff3).build();
@@ -206,15 +206,15 @@ public class TestTableFormatSync {
void syncChangesWithDifferentFormatsAndMetadata() {
Instant start = Instant.now();
OneTable tableState1 = getTableState(1);
- OneDataFilesDiff dataFilesDiff1 = getFilesDiff(1);
+ DataFilesDiff dataFilesDiff1 = getFilesDiff(1);
TableChange tableChange1 =
TableChange.builder().tableAsOfChange(tableState1).filesDiff(dataFilesDiff1).build();
OneTable tableState2 = getTableState(2);
- OneDataFilesDiff dataFilesDiff2 = getFilesDiff(2);
+ DataFilesDiff dataFilesDiff2 = getFilesDiff(2);
TableChange tableChange2 =
TableChange.builder().tableAsOfChange(tableState2).filesDiff(dataFilesDiff2).build();
OneTable tableState3 = getTableState(3);
- OneDataFilesDiff dataFilesDiff3 = getFilesDiff(3);
+ DataFilesDiff dataFilesDiff3 = getFilesDiff(3);
TableChange tableChange3 =
TableChange.builder().tableAsOfChange(tableState3).filesDiff(dataFilesDiff3).build();
@@ -292,7 +292,7 @@ public class TestTableFormatSync {
void syncChangesOneFormatWithNoRequiredChanges() {
Instant start = Instant.now();
OneTable tableState1 = getTableState(1);
- OneDataFilesDiff dataFilesDiff1 = getFilesDiff(1);
+ DataFilesDiff dataFilesDiff1 = getFilesDiff(1);
TableChange tableChange1 =
TableChange.builder().tableAsOfChange(tableState1).filesDiff(dataFilesDiff1).build();
@@ -358,11 +358,11 @@ public class TestTableFormatSync {
.build();
}
- private OneDataFilesDiff getFilesDiff(int id) {
- return OneDataFilesDiff.builder()
+ private DataFilesDiff getFilesDiff(int id) {
+ return DataFilesDiff.builder()
.filesAdded(
Collections.singletonList(
- OneDataFile.builder()
+ InternalDataFile.builder()
.physicalPath(String.format("/tmp/path/file_%d.parquet", id))
.build()))
.build();
diff --git a/core/src/main/java/org/apache/xtable/delta/DeltaActionsConverter.java b/core/src/main/java/org/apache/xtable/delta/DeltaActionsConverter.java
index 7bd3f663..8262b40b 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaActionsConverter.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaActionsConverter.java
@@ -35,7 +35,7 @@ import org.apache.xtable.model.schema.OneField;
import org.apache.xtable.model.schema.OnePartitionField;
import org.apache.xtable.model.stat.ColumnStat;
import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
+import org.apache.xtable.model.storage.InternalDataFile;
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class DeltaActionsConverter {
@@ -46,7 +46,7 @@ public class DeltaActionsConverter {
return INSTANCE;
}
- public OneDataFile convertAddActionToOneDataFile(
+ public InternalDataFile convertAddActionToInternalDataFile(
AddFile addFile,
Snapshot deltaSnapshot,
FileFormat fileFormat,
@@ -62,7 +62,7 @@ public class DeltaActionsConverter {
long recordCount =
columnStats.stream().map(ColumnStat::getNumValues).max(Long::compareTo).orElse(0L);
// TODO(https://github.com/apache/incubator-xtable/issues/102): removed record count.
- return OneDataFile.builder()
+ return InternalDataFile.builder()
.physicalPath(getFullPathToFile(deltaSnapshot, addFile.path()))
.fileFormat(fileFormat)
.fileSizeBytes(addFile.size())
@@ -74,13 +74,13 @@ public class DeltaActionsConverter {
.build();
}
- public OneDataFile convertRemoveActionToOneDataFile(
+ public InternalDataFile convertRemoveActionToInternalDataFile(
RemoveFile removeFile,
Snapshot deltaSnapshot,
FileFormat fileFormat,
List<OnePartitionField> partitionFields,
DeltaPartitionExtractor partitionExtractor) {
- return OneDataFile.builder()
+ return InternalDataFile.builder()
.physicalPath(getFullPathToFile(deltaSnapshot, removeFile.path()))
.fileFormat(fileFormat)
.partitionValues(
diff --git a/core/src/main/java/org/apache/xtable/delta/DeltaClient.java b/core/src/main/java/org/apache/xtable/delta/DeltaClient.java
index f1c62561..cee8568b 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaClient.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaClient.java
@@ -60,7 +60,7 @@ import org.apache.xtable.model.OneTable;
import org.apache.xtable.model.OneTableMetadata;
import org.apache.xtable.model.schema.OnePartitionField;
import org.apache.xtable.model.schema.OneSchema;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.OneFileGroup;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.spi.sync.TargetClient;
@@ -186,10 +186,10 @@ public class DeltaClient implements TargetClient {
}
@Override
- public void syncFilesForDiff(OneDataFilesDiff oneDataFilesDiff) {
+ public void syncFilesForDiff(DataFilesDiff dataFilesDiff) {
transactionState.setActions(
dataFileUpdatesExtractor.applyDiff(
- oneDataFilesDiff,
+ dataFilesDiff,
transactionState.getLatestSchemaInternal(),
deltaLog.dataPath().toString()));
}
diff --git a/core/src/main/java/org/apache/xtable/delta/DeltaDataFileExtractor.java b/core/src/main/java/org/apache/xtable/delta/DeltaDataFileExtractor.java
index 89e22fe0..95507826 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaDataFileExtractor.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaDataFileExtractor.java
@@ -29,7 +29,7 @@ import org.apache.xtable.model.schema.OneField;
import org.apache.xtable.model.schema.OnePartitionField;
import org.apache.xtable.model.schema.OneSchema;
import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.spi.extractor.DataFileIterator;
/** DeltaDataFileExtractor lets the consumer iterate over partitions. */
@@ -58,7 +58,7 @@ public class DeltaDataFileExtractor {
private final FileFormat fileFormat;
private final List<OneField> fields;
private final List<OnePartitionField> partitionFields;
- private final Iterator<OneDataFile> dataFilesIterator;
+ private final Iterator<InternalDataFile> dataFilesIterator;
private DeltaDataFileIterator(Snapshot snapshot, OneSchema schema, boolean includeColumnStats) {
this.fileFormat =
@@ -71,7 +71,7 @@ public class DeltaDataFileExtractor {
snapshot.allFiles().collectAsList().stream()
.map(
addFile ->
- actionsConverter.convertAddActionToOneDataFile(
+ actionsConverter.convertAddActionToInternalDataFile(
addFile,
snapshot,
fileFormat,
@@ -92,7 +92,7 @@ public class DeltaDataFileExtractor {
}
@Override
- public OneDataFile next() {
+ public InternalDataFile next() {
return dataFilesIterator.next();
}
}
diff --git a/core/src/main/java/org/apache/xtable/delta/DeltaDataFileUpdatesExtractor.java b/core/src/main/java/org/apache/xtable/delta/DeltaDataFileUpdatesExtractor.java
index 5ad7dd29..5d7de074 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaDataFileUpdatesExtractor.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaDataFileUpdatesExtractor.java
@@ -40,8 +40,8 @@ import org.apache.xtable.collectors.CustomCollectors;
import org.apache.xtable.model.schema.OneSchema;
import org.apache.xtable.model.stat.ColumnStat;
import org.apache.xtable.model.storage.DataFilesDiff;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.FilesDiff;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.OneFileGroup;
import org.apache.xtable.paths.PathUtils;
@@ -72,25 +72,25 @@ public class DeltaDataFileUpdatesExtractor {
file -> DeltaActionsConverter.getFullPathToFile(snapshot, file.path()),
file -> file));
- DataFilesDiff<OneDataFile, Action> diff =
- OneDataFilesDiff.findNewAndRemovedFiles(partitionedDataFiles, previousFiles);
+ FilesDiff<InternalDataFile, Action> diff =
+ DataFilesDiff.findNewAndRemovedFiles(partitionedDataFiles, previousFiles);
return applyDiff(
diff.getFilesAdded(), diff.getFilesRemoved(), tableSchema, deltaLog.dataPath().toString());
}
public Seq<Action> applyDiff(
- OneDataFilesDiff oneDataFilesDiff, OneSchema tableSchema, String tableBasePath) {
+ DataFilesDiff dataFilesDiff, OneSchema tableSchema, String tableBasePath) {
List<Action> removeActions =
- oneDataFilesDiff.getFilesRemoved().stream()
+ dataFilesDiff.getFilesRemoved().stream()
.flatMap(dFile -> createAddFileAction(dFile, tableSchema, tableBasePath))
.map(AddFile::remove)
- .collect(CustomCollectors.toList(oneDataFilesDiff.getFilesRemoved().size()));
- return applyDiff(oneDataFilesDiff.getFilesAdded(), removeActions, tableSchema, tableBasePath);
+ .collect(CustomCollectors.toList(dataFilesDiff.getFilesRemoved().size()));
+ return applyDiff(dataFilesDiff.getFilesAdded(), removeActions, tableSchema, tableBasePath);
}
private Seq<Action> applyDiff(
- Set<OneDataFile> filesAdded,
+ Set<InternalDataFile> filesAdded,
Collection<Action> removeFileActions,
OneSchema tableSchema,
String tableBasePath) {
@@ -105,7 +105,7 @@ public class DeltaDataFileUpdatesExtractor {
}
private Stream<AddFile> createAddFileAction(
- OneDataFile dataFile, OneSchema schema, String tableBasePath) {
+ InternalDataFile dataFile, OneSchema schema, String tableBasePath) {
return Stream.of(
new AddFile(
// Delta Lake supports relative and absolute paths in theory but relative paths seem
diff --git a/core/src/main/java/org/apache/xtable/delta/DeltaPartitionExtractor.java b/core/src/main/java/org/apache/xtable/delta/DeltaPartitionExtractor.java
index b7ba5f23..31eb0752 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaPartitionExtractor.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaPartitionExtractor.java
@@ -56,7 +56,7 @@ import org.apache.xtable.model.schema.OneSchema;
import org.apache.xtable.model.schema.PartitionTransformType;
import org.apache.xtable.model.stat.PartitionValue;
import org.apache.xtable.model.stat.Range;
-import org.apache.xtable.model.storage.OneDataFile;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.schema.SchemaFieldFinder;
/**
@@ -247,12 +247,13 @@ public class DeltaPartitionExtractor {
return nameToStructFieldMap;
}
- public Map<String, String> partitionValueSerialization(OneDataFile oneDataFile) {
+ public Map<String, String> partitionValueSerialization(InternalDataFile internalDataFile) {
Map<String, String> partitionValuesSerialized = new HashMap<>();
- if (oneDataFile.getPartitionValues() == null || oneDataFile.getPartitionValues().isEmpty()) {
+ if (internalDataFile.getPartitionValues() == null
+ || internalDataFile.getPartitionValues().isEmpty()) {
return partitionValuesSerialized;
}
- for (PartitionValue partitionValue : oneDataFile.getPartitionValues()) {
+ for (PartitionValue partitionValue : internalDataFile.getPartitionValues()) {
OnePartitionField partitionField = partitionValue.getPartitionField();
PartitionTransformType transformType = partitionField.getTransformType();
String partitionValueSerialized;
diff --git a/core/src/main/java/org/apache/xtable/delta/DeltaSourceClient.java b/core/src/main/java/org/apache/xtable/delta/DeltaSourceClient.java
index bb344be7..caff2641 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaSourceClient.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaSourceClient.java
@@ -53,9 +53,9 @@ import org.apache.xtable.model.TableChange;
import org.apache.xtable.model.schema.OneSchema;
import org.apache.xtable.model.schema.SchemaCatalog;
import org.apache.xtable.model.schema.SchemaVersion;
+import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.OneFileGroup;
import org.apache.xtable.spi.extractor.DataFileIterator;
import org.apache.xtable.spi.extractor.SourceClient;
@@ -102,7 +102,7 @@ public class DeltaSourceClient implements SourceClient<Long> {
return OneSnapshot.builder()
.table(table)
.schemaCatalog(getSchemaCatalog(table, snapshot.version()))
- .partitionedDataFiles(getOneDataFiles(snapshot, table.getReadSchema()))
+ .partitionedDataFiles(getInternalDataFiles(snapshot, table.getReadSchema()))
.build();
}
@@ -115,12 +115,12 @@ public class DeltaSourceClient implements SourceClient<Long> {
FileFormat fileFormat =
actionsConverter.convertToOneTableFileFormat(
snapshotAtVersion.metadata().format().provider());
- Set<OneDataFile> addedFiles = new HashSet<>();
- Set<OneDataFile> removedFiles = new HashSet<>();
+ Set<InternalDataFile> addedFiles = new HashSet<>();
+ Set<InternalDataFile> removedFiles = new HashSet<>();
for (Action action : actionsForVersion) {
if (action instanceof AddFile) {
addedFiles.add(
- actionsConverter.convertAddActionToOneDataFile(
+ actionsConverter.convertAddActionToInternalDataFile(
(AddFile) action,
snapshotAtVersion,
fileFormat,
@@ -131,7 +131,7 @@ public class DeltaSourceClient implements SourceClient<Long> {
DeltaStatsExtractor.getInstance()));
} else if (action instanceof RemoveFile) {
removedFiles.add(
- actionsConverter.convertRemoveActionToOneDataFile(
+ actionsConverter.convertRemoveActionToInternalDataFile(
(RemoveFile) action,
snapshotAtVersion,
fileFormat,
@@ -139,8 +139,8 @@ public class DeltaSourceClient implements SourceClient<Long> {
DeltaPartitionExtractor.getInstance()));
}
}
- OneDataFilesDiff dataFilesDiff =
- OneDataFilesDiff.builder().filesAdded(addedFiles).filesRemoved(removedFiles).build();
+ DataFilesDiff dataFilesDiff =
+ DataFilesDiff.builder().filesAdded(addedFiles).filesRemoved(removedFiles).build();
return TableChange.builder().tableAsOfChange(tableAtVersion).filesDiff(dataFilesDiff).build();
}
@@ -188,9 +188,9 @@ public class DeltaSourceClient implements SourceClient<Long> {
.build());
}
- private List<OneFileGroup> getOneDataFiles(Snapshot snapshot, OneSchema schema) {
+ private List<OneFileGroup> getInternalDataFiles(Snapshot snapshot, OneSchema schema) {
try (DataFileIterator fileIterator = dataFileExtractor.iterator(snapshot, schema)) {
- List<OneDataFile> dataFiles = new ArrayList<>();
+ List<InternalDataFile> dataFiles = new ArrayList<>();
fileIterator.forEachRemaining(dataFiles::add);
return OneFileGroup.fromFiles(dataFiles);
} catch (Exception e) {
diff --git a/core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java b/core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java
index 7d7f7558..10eb95a3 100644
--- a/core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java
+++ b/core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java
@@ -54,8 +54,8 @@ import org.apache.hudi.metadata.HoodieMetadataFileSystemView;
import org.apache.xtable.collectors.CustomCollectors;
import org.apache.xtable.model.schema.OneType;
import org.apache.xtable.model.stat.ColumnStat;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.DataFilesDiff;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.OneFileGroup;
@AllArgsConstructor(staticName = "of")
@@ -94,17 +94,18 @@ public class BaseFileUpdatesExtractor {
partitionedDataFiles.stream()
.map(
partitionFileGroup -> {
- List<OneDataFile> dataFiles = partitionFileGroup.getFiles();
+ List<InternalDataFile> dataFiles = partitionFileGroup.getFiles();
String partitionPath = getPartitionPath(tableBasePath, dataFiles);
// remove the partition from the set of partitions to drop since it is present in
// the snapshot
partitionPathsToDrop.remove(partitionPath);
// create a map of file path to the data file, any entries not in the hudi table
// will be added
- Map<String, OneDataFile> physicalPathToFile =
+ Map<String, InternalDataFile> physicalPathToFile =
dataFiles.stream()
.collect(
- Collectors.toMap(OneDataFile::getPhysicalPath, Function.identity()));
+ Collectors.toMap(
+ InternalDataFile::getPhysicalPath, Function.identity()));
List<HoodieBaseFile> baseFiles =
isTableInitialized
? fsView.getLatestBaseFiles(partitionPath).collect(Collectors.toList())
@@ -160,16 +161,16 @@ public class BaseFileUpdatesExtractor {
}
/**
- * Converts the provided {@link OneDataFilesDiff}.
+ * Converts the provided {@link DataFilesDiff}.
*
- * @param oneDataFilesDiff the diff to apply to the Hudi table
+ * @param dataFilesDiff the diff to apply to the Hudi table
* @param commit The current commit started by the Hudi client
* @return The information needed to create a "replace" commit for the Hudi table
*/
- ReplaceMetadata convertDiff(@NonNull OneDataFilesDiff oneDataFilesDiff, @NonNull String commit) {
+ ReplaceMetadata convertDiff(@NonNull DataFilesDiff dataFilesDiff, @NonNull String commit) {
// For all removed files, group by partition and extract the file id
Map<String, List<String>> partitionToReplacedFileIds =
- oneDataFilesDiff.getFilesRemoved().stream()
+ dataFilesDiff.getFilesRemoved().stream()
.map(file -> new CachingPath(file.getPhysicalPath()))
.collect(
Collectors.groupingBy(
@@ -177,9 +178,9 @@ public class BaseFileUpdatesExtractor {
Collectors.mapping(this::getFileId, Collectors.toList())));
// For all added files, group by partition and extract the file id
List<WriteStatus> writeStatuses =
- oneDataFilesDiff.getFilesAdded().stream()
+ dataFilesDiff.getFilesAdded().stream()
.map(file -> toWriteStatus(tableBasePath, commit, file, Optional.empty()))
- .collect(CustomCollectors.toList(oneDataFilesDiff.getFilesAdded().size()));
+ .collect(CustomCollectors.toList(dataFilesDiff.getFilesAdded().size()));
return ReplaceMetadata.of(partitionToReplacedFileIds, writeStatuses);
}
@@ -206,7 +207,7 @@ public class BaseFileUpdatesExtractor {
private WriteStatus toWriteStatus(
Path tableBasePath,
String commitTime,
- OneDataFile file,
+ InternalDataFile file,
Optional<String> partitionPathOptional) {
WriteStatus writeStatus = new WriteStatus();
Path path = new CachingPath(file.getPhysicalPath());
@@ -269,7 +270,7 @@ public class BaseFileUpdatesExtractor {
}
}
- private String getPartitionPath(Path tableBasePath, List<OneDataFile> files) {
+ private String getPartitionPath(Path tableBasePath, List<InternalDataFile> files) {
return HudiPathUtils.getPartitionPath(
tableBasePath, new CachingPath(files.get(0).getPhysicalPath()));
}
diff --git a/core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java b/core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java
index ee8cbe63..ccefc1c0 100644
--- a/core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java
+++ b/core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java
@@ -64,9 +64,9 @@ import org.apache.xtable.model.OneTable;
import org.apache.xtable.model.schema.OnePartitionField;
import org.apache.xtable.model.schema.SchemaVersion;
import org.apache.xtable.model.stat.PartitionValue;
+import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.OneFileGroup;
/** Extracts all the files for Hudi table represented by {@link OneTable}. */
@@ -115,14 +115,14 @@ public class HudiDataFileExtractor implements AutoCloseable {
tableMetadata != null
? tableMetadata.getAllPartitionPaths()
: FSUtils.getAllPartitionPaths(engineContext, metadataConfig, basePath.toString());
- return getOneDataFilesForPartitions(allPartitionPaths, table);
+ return getInternalDataFilesForPartitions(allPartitionPaths, table);
} catch (IOException ex) {
throw new OneIOException(
"Unable to read partitions for table " + metaClient.getTableConfig().getTableName(), ex);
}
}
- public OneDataFilesDiff getDiffForCommit(
+ public DataFilesDiff getDiffForCommit(
HoodieInstant hoodieInstantForDiff,
OneTable table,
HoodieInstant instant,
@@ -132,14 +132,14 @@ public class HudiDataFileExtractor implements AutoCloseable {
getAddedAndRemovedPartitionInfo(
visibleTimeline, instant, fsView, hoodieInstantForDiff, table.getPartitioningFields());
- Stream<OneDataFile> filesAddedWithoutStats = allInfo.getAdded().stream();
- List<OneDataFile> filesAdded =
+ Stream<InternalDataFile> filesAddedWithoutStats = allInfo.getAdded().stream();
+ List<InternalDataFile> filesAdded =
fileStatsExtractor
.addStatsToFiles(tableMetadata, filesAddedWithoutStats, table.getReadSchema())
.collect(Collectors.toList());
- List<OneDataFile> filesRemoved = allInfo.getRemoved();
+ List<InternalDataFile> filesRemoved = allInfo.getRemoved();
- return OneDataFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesRemoved).build();
+ return DataFilesDiff.builder().filesAdded(filesAdded).filesRemoved(filesRemoved).build();
}
private AddedAndRemovedFiles getAddedAndRemovedPartitionInfo(
@@ -149,8 +149,8 @@ public class HudiDataFileExtractor implements AutoCloseable {
HoodieInstant instantToConsider,
List<OnePartitionField> partitioningFields) {
try {
- List<OneDataFile> addedFiles = new ArrayList<>();
- List<OneDataFile> removedFiles = new ArrayList<>();
+ List<InternalDataFile> addedFiles = new ArrayList<>();
+ List<InternalDataFile> removedFiles = new ArrayList<>();
switch (instant.getAction()) {
case HoodieTimeline.COMMIT_ACTION:
case HoodieTimeline.DELTA_COMMIT_ACTION:
@@ -253,7 +253,7 @@ public class HudiDataFileExtractor implements AutoCloseable {
}
}
- private List<OneDataFile> getRemovedFiles(
+ private List<InternalDataFile> getRemovedFiles(
String partitionPath, List<String> deletedPaths, List<OnePartitionField> partitioningFields) {
List<PartitionValue> partitionValues =
partitionValuesExtractor.extractPartitionValues(partitioningFields, partitionPath);
@@ -282,8 +282,8 @@ public class HudiDataFileExtractor implements AutoCloseable {
String partitionPath,
Set<String> affectedFileIds,
List<OnePartitionField> partitioningFields) {
- List<OneDataFile> filesToAdd = new ArrayList<>(affectedFileIds.size());
- List<OneDataFile> filesToRemove = new ArrayList<>(affectedFileIds.size());
+ List<InternalDataFile> filesToAdd = new ArrayList<>(affectedFileIds.size());
+ List<InternalDataFile> filesToRemove = new ArrayList<>(affectedFileIds.size());
List<PartitionValue> partitionValues =
partitionValuesExtractor.extractPartitionValues(partitioningFields, partitionPath);
Stream<HoodieFileGroup> fileGroups =
@@ -318,8 +318,8 @@ public class HudiDataFileExtractor implements AutoCloseable {
Set<String> replacedFileIds,
Set<String> newFileIds,
List<OnePartitionField> partitioningFields) {
- List<OneDataFile> filesToAdd = new ArrayList<>(newFileIds.size());
- List<OneDataFile> filesToRemove = new ArrayList<>(replacedFileIds.size());
+ List<InternalDataFile> filesToAdd = new ArrayList<>(newFileIds.size());
+ List<InternalDataFile> filesToRemove = new ArrayList<>(replacedFileIds.size());
List<PartitionValue> partitionValues =
partitionValuesExtractor.extractPartitionValues(partitioningFields, partitionPath);
Stream<HoodieFileGroup> fileGroups =
@@ -341,11 +341,11 @@ public class HudiDataFileExtractor implements AutoCloseable {
return AddedAndRemovedFiles.builder().added(filesToAdd).removed(filesToRemove).build();
}
- private List<OneFileGroup> getOneDataFilesForPartitions(
+ private List<OneFileGroup> getInternalDataFilesForPartitions(
List<String> partitionPaths, OneTable table) {
SyncableFileSystemView fsView = fileSystemViewManager.getFileSystemView(metaClient);
- Stream<OneDataFile> filesWithoutStats =
+ Stream<InternalDataFile> filesWithoutStats =
partitionPaths.stream()
.parallel()
.flatMap(
@@ -357,7 +357,7 @@ public class HudiDataFileExtractor implements AutoCloseable {
.getLatestBaseFiles(partitionPath)
.map(baseFile -> buildFileWithoutStats(partitionValues, baseFile));
});
- Stream<OneDataFile> files =
+ Stream<InternalDataFile> files =
fileStatsExtractor.addStatsToFiles(tableMetadata, filesWithoutStats, table.getReadSchema());
return OneFileGroup.fromFiles(files);
}
@@ -378,21 +378,21 @@ public class HudiDataFileExtractor implements AutoCloseable {
@Builder
@Value
private static class AddedAndRemovedFiles {
- List<OneDataFile> added;
- List<OneDataFile> removed;
+ List<InternalDataFile> added;
+ List<InternalDataFile> removed;
}
/**
- * Builds a {@link OneDataFile} without any statistics or rowCount value set.
+ * Builds a {@link InternalDataFile} without any statistics or rowCount value set.
*
* @param partitionValues values extracted from the partition path
* @param hoodieBaseFile the base file from Hudi
- * @return {@link OneDataFile} without any statistics or rowCount value set.
+ * @return {@link InternalDataFile} without any statistics or rowCount value set.
*/
- private OneDataFile buildFileWithoutStats(
+ private InternalDataFile buildFileWithoutStats(
List<PartitionValue> partitionValues, HoodieBaseFile hoodieBaseFile) {
long rowCount = 0L;
- return OneDataFile.builder()
+ return InternalDataFile.builder()
.schemaVersion(DEFAULT_SCHEMA_VERSION)
.physicalPath(hoodieBaseFile.getPath())
.fileFormat(getFileFormat(FSUtils.getFileExtension(hoodieBaseFile.getPath())))
diff --git a/core/src/main/java/org/apache/xtable/hudi/HudiFileStatsExtractor.java b/core/src/main/java/org/apache/xtable/hudi/HudiFileStatsExtractor.java
index cdb96d04..1d6c1c7d 100644
--- a/core/src/main/java/org/apache/xtable/hudi/HudiFileStatsExtractor.java
+++ b/core/src/main/java/org/apache/xtable/hudi/HudiFileStatsExtractor.java
@@ -53,7 +53,7 @@ import org.apache.xtable.model.schema.OneSchema;
import org.apache.xtable.model.schema.OneType;
import org.apache.xtable.model.stat.ColumnStat;
import org.apache.xtable.model.stat.Range;
-import org.apache.xtable.model.storage.OneDataFile;
+import org.apache.xtable.model.storage.InternalDataFile;
/** Responsible for Column stats extraction for Hudi. */
@AllArgsConstructor
@@ -83,8 +83,8 @@ public class HudiFileStatsExtractor {
* @param schema the schema of the files (assumed to be the same for all files in stream)
* @return a stream of files with column stats and row count information
*/
- public Stream<OneDataFile> addStatsToFiles(
- HoodieTableMetadata metadataTable, Stream<OneDataFile> files, OneSchema schema) {
+ public Stream<InternalDataFile> addStatsToFiles(
+ HoodieTableMetadata metadataTable, Stream<InternalDataFile> files, OneSchema schema) {
boolean useMetadataTableColStats =
metadataTable != null
&& metaClient
@@ -101,8 +101,8 @@ public class HudiFileStatsExtractor {
: computeColumnStatsFromParquetFooters(files, nameFieldMap);
}
- private Stream<OneDataFile> computeColumnStatsFromParquetFooters(
- Stream<OneDataFile> files, Map<String, OneField> nameFieldMap) {
+ private Stream<InternalDataFile> computeColumnStatsFromParquetFooters(
+ Stream<InternalDataFile> files, Map<String, OneField> nameFieldMap) {
return files.map(
file -> {
HudiFileStats fileStats =
@@ -120,11 +120,11 @@ public class HudiFileStatsExtractor {
return Pair.of(partitionPath, filePath.getName());
}
- private Stream<OneDataFile> computeColumnStatsFromMetadataTable(
+ private Stream<InternalDataFile> computeColumnStatsFromMetadataTable(
HoodieTableMetadata metadataTable,
- Stream<OneDataFile> files,
+ Stream<InternalDataFile> files,
Map<String, OneField> nameFieldMap) {
- Map<Pair<String, String>, OneDataFile> filePathsToDataFile =
+ Map<Pair<String, String>, InternalDataFile> filePathsToDataFile =
files.collect(
Collectors.toMap(
file -> getPartitionAndFileName(file.getPhysicalPath()), Function.identity()));
@@ -154,7 +154,7 @@ public class HudiFileStatsExtractor {
.map(
pathToDataFile -> {
Pair<String, String> filePath = pathToDataFile.getKey();
- OneDataFile file = pathToDataFile.getValue();
+ InternalDataFile file = pathToDataFile.getValue();
List<Pair<OneField, HoodieMetadataColumnStats>> fileStats =
stats.getOrDefault(filePath, Collections.emptyList());
List<ColumnStat> columnStats =
diff --git a/core/src/main/java/org/apache/xtable/hudi/HudiTargetClient.java b/core/src/main/java/org/apache/xtable/hudi/HudiTargetClient.java
index b7eb4c11..816d05ae 100644
--- a/core/src/main/java/org/apache/xtable/hudi/HudiTargetClient.java
+++ b/core/src/main/java/org/apache/xtable/hudi/HudiTargetClient.java
@@ -87,7 +87,7 @@ import org.apache.xtable.model.OneTableMetadata;
import org.apache.xtable.model.schema.OneField;
import org.apache.xtable.model.schema.OnePartitionField;
import org.apache.xtable.model.schema.OneSchema;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.OneFileGroup;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.spi.sync.TargetClient;
@@ -251,9 +251,9 @@ public class HudiTargetClient implements TargetClient {
}
@Override
- public void syncFilesForDiff(OneDataFilesDiff oneDataFilesDiff) {
+ public void syncFilesForDiff(DataFilesDiff dataFilesDiff) {
BaseFileUpdatesExtractor.ReplaceMetadata replaceMetadata =
- baseFileUpdatesExtractor.convertDiff(oneDataFilesDiff, commitState.getInstantTime());
+ baseFileUpdatesExtractor.convertDiff(dataFilesDiff, commitState.getInstantTime());
commitState.setReplaceMetadata(replaceMetadata);
}
diff --git a/core/src/main/java/org/apache/xtable/iceberg/IcebergClient.java b/core/src/main/java/org/apache/xtable/iceberg/IcebergClient.java
index 9c72c09e..6e3eb2eb 100644
--- a/core/src/main/java/org/apache/xtable/iceberg/IcebergClient.java
+++ b/core/src/main/java/org/apache/xtable/iceberg/IcebergClient.java
@@ -48,7 +48,7 @@ import org.apache.xtable.model.OneTable;
import org.apache.xtable.model.OneTableMetadata;
import org.apache.xtable.model.schema.OnePartitionField;
import org.apache.xtable.model.schema.OneSchema;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.OneFileGroup;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.spi.sync.TargetClient;
@@ -210,9 +210,9 @@ public class IcebergClient implements TargetClient {
}
@Override
- public void syncFilesForDiff(OneDataFilesDiff oneDataFilesDiff) {
+ public void syncFilesForDiff(DataFilesDiff dataFilesDiff) {
dataFileUpdatesExtractor.applyDiff(
- transaction, oneDataFilesDiff, transaction.table().schema(), transaction.table().spec());
+ transaction, dataFilesDiff, transaction.table().schema(), transaction.table().spec());
}
@Override
diff --git a/core/src/main/java/org/apache/xtable/iceberg/IcebergDataFileExtractor.java b/core/src/main/java/org/apache/xtable/iceberg/IcebergDataFileExtractor.java
index f68da230..c6d0d3ad 100644
--- a/core/src/main/java/org/apache/xtable/iceberg/IcebergDataFileExtractor.java
+++ b/core/src/main/java/org/apache/xtable/iceberg/IcebergDataFileExtractor.java
@@ -30,7 +30,7 @@ import org.apache.xtable.model.schema.OneSchema;
import org.apache.xtable.model.stat.ColumnStat;
import org.apache.xtable.model.stat.PartitionValue;
import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
+import org.apache.xtable.model.storage.InternalDataFile;
/** Extractor of data files for Iceberg */
@Builder
@@ -41,19 +41,19 @@ public class IcebergDataFileExtractor {
IcebergPartitionValueConverter.getInstance();
/**
- * Builds {@link OneDataFile} representation from Iceberg {@link DataFile}.
+ * Builds {@link InternalDataFile} representation from Iceberg {@link DataFile}.
*
* @param dataFile Iceberg data file
* @param partitionValues representation of partition fields and ranges
* @param schema current schema for the table, used for mapping field IDs to stats
* @return corresponding OneTable data file
*/
- OneDataFile fromIceberg(
+ InternalDataFile fromIceberg(
DataFile dataFile, List<PartitionValue> partitionValues, OneSchema schema) {
return fromIceberg(dataFile, partitionValues, schema, true);
}
- private OneDataFile fromIceberg(
+ private InternalDataFile fromIceberg(
DataFile dataFile,
List<PartitionValue> partitionValues,
OneSchema schema,
@@ -74,7 +74,7 @@ public class IcebergDataFileExtractor {
if (!filePath.contains(":")) {
filePath = "file:" + filePath;
}
- return OneDataFile.builder()
+ return InternalDataFile.builder()
.physicalPath(filePath)
.fileFormat(fromIcebergFileFormat(dataFile.format()))
.fileSizeBytes(dataFile.fileSizeInBytes())
diff --git a/core/src/main/java/org/apache/xtable/iceberg/IcebergDataFileUpdatesSync.java b/core/src/main/java/org/apache/xtable/iceberg/IcebergDataFileUpdatesSync.java
index f0cff732..25bfb185 100644
--- a/core/src/main/java/org/apache/xtable/iceberg/IcebergDataFileUpdatesSync.java
+++ b/core/src/main/java/org/apache/xtable/iceberg/IcebergDataFileUpdatesSync.java
@@ -31,8 +31,8 @@ import org.apache.xtable.exception.NotSupportedException;
import org.apache.xtable.exception.OneIOException;
import org.apache.xtable.model.OneTable;
import org.apache.xtable.model.storage.DataFilesDiff;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.FilesDiff;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.OneFileGroup;
@AllArgsConstructor(staticName = "of")
@@ -57,29 +57,29 @@ public class IcebergDataFileUpdatesSync {
throw new OneIOException("Failed to iterate through Iceberg data files", e);
}
- DataFilesDiff<OneDataFile, DataFile> diff =
- OneDataFilesDiff.findNewAndRemovedFiles(partitionedDataFiles, previousFiles);
+ FilesDiff<InternalDataFile, DataFile> diff =
+ DataFilesDiff.findNewAndRemovedFiles(partitionedDataFiles, previousFiles);
applyDiff(transaction, diff.getFilesAdded(), diff.getFilesRemoved(), schema, partitionSpec);
}
public void applyDiff(
Transaction transaction,
- OneDataFilesDiff oneDataFilesDiff,
+ DataFilesDiff dataFilesDiff,
Schema schema,
PartitionSpec partitionSpec) {
Collection<DataFile> filesRemoved =
- oneDataFilesDiff.getFilesRemoved().stream()
+ dataFilesDiff.getFilesRemoved().stream()
.map(file -> getDataFile(partitionSpec, schema, file))
.collect(Collectors.toList());
- applyDiff(transaction, oneDataFilesDiff.getFilesAdded(), filesRemoved, schema, partitionSpec);
+ applyDiff(transaction, dataFilesDiff.getFilesAdded(), filesRemoved, schema, partitionSpec);
}
private void applyDiff(
Transaction transaction,
- Collection<OneDataFile> filesAdded,
+ Collection<InternalDataFile> filesAdded,
Collection<DataFile> filesRemoved,
Schema schema,
PartitionSpec partitionSpec) {
@@ -89,7 +89,8 @@ public class IcebergDataFileUpdatesSync {
overwriteFiles.commit();
}
- private DataFile getDataFile(PartitionSpec partitionSpec, Schema schema, OneDataFile dataFile) {
+ private DataFile getDataFile(
+ PartitionSpec partitionSpec, Schema schema, InternalDataFile dataFile) {
DataFiles.Builder builder =
DataFiles.builder(partitionSpec)
.withPath(dataFile.getPhysicalPath())
diff --git a/core/src/main/java/org/apache/xtable/iceberg/IcebergSourceClient.java b/core/src/main/java/org/apache/xtable/iceberg/IcebergSourceClient.java
index a7e49584..5ff4a8ed 100644
--- a/core/src/main/java/org/apache/xtable/iceberg/IcebergSourceClient.java
+++ b/core/src/main/java/org/apache/xtable/iceberg/IcebergSourceClient.java
@@ -48,11 +48,8 @@ import org.apache.xtable.model.schema.OneSchema;
import org.apache.xtable.model.schema.SchemaCatalog;
import org.apache.xtable.model.schema.SchemaVersion;
import org.apache.xtable.model.stat.PartitionValue;
-import org.apache.xtable.model.storage.DataLayoutStrategy;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
-import org.apache.xtable.model.storage.OneFileGroup;
-import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.spi.extractor.SourceClient;
@Log4j2
@@ -145,10 +142,10 @@ public class IcebergSourceClient implements SourceClient<Snapshot> {
PartitionSpec partitionSpec = iceTable.spec();
List<OneFileGroup> partitionedDataFiles;
try (CloseableIterable<FileScanTask> files = scan.planFiles()) {
- List<OneDataFile> irFiles = new ArrayList<>();
+ List<InternalDataFile> irFiles = new ArrayList<>();
for (FileScanTask fileScanTask : files) {
DataFile file = fileScanTask.file();
- OneDataFile irDataFile = fromIceberg(file, partitionSpec, irTable);
+ InternalDataFile irDataFile = fromIceberg(file, partitionSpec, irTable);
irFiles.add(irDataFile);
}
partitionedDataFiles = OneFileGroup.fromFiles(irFiles);
@@ -164,7 +161,8 @@ public class IcebergSourceClient implements SourceClient<Snapshot> {
.build();
}
- private OneDataFile fromIceberg(DataFile file, PartitionSpec partitionSpec, OneTable oneTable) {
+ private InternalDataFile fromIceberg(
+ DataFile file, PartitionSpec partitionSpec, OneTable oneTable) {
List<PartitionValue> partitionValues =
partitionConverter.toOneTable(oneTable, file.partition(), partitionSpec);
return dataFileExtractor.fromIceberg(file, partitionValues, oneTable.getReadSchema());
@@ -177,21 +175,18 @@ public class IcebergSourceClient implements SourceClient<Snapshot> {
PartitionSpec partitionSpec = iceTable.spec();
OneTable irTable = getTable(snapshot);
- Set<OneDataFile> dataFilesAdded =
+ Set<InternalDataFile> dataFilesAdded =
StreamSupport.stream(snapshot.addedDataFiles(fileIO).spliterator(), false)
.map(dataFile -> fromIceberg(dataFile, partitionSpec, irTable))
.collect(Collectors.toSet());
- Set<OneDataFile> dataFilesRemoved =
+ Set<InternalDataFile> dataFilesRemoved =
StreamSupport.stream(snapshot.removedDataFiles(fileIO).spliterator(), false)
.map(dataFile -> fromIceberg(dataFile, partitionSpec, irTable))
.collect(Collectors.toSet());
- OneDataFilesDiff filesDiff =
- OneDataFilesDiff.builder()
- .filesAdded(dataFilesAdded)
- .filesRemoved(dataFilesRemoved)
- .build();
+ DataFilesDiff filesDiff =
+ DataFilesDiff.builder().filesAdded(dataFilesAdded).filesRemoved(dataFilesRemoved).build();
OneTable table = getTable(snapshot);
return TableChange.builder().tableAsOfChange(table).filesDiff(filesDiff).build();
diff --git a/core/src/test/java/org/apache/xtable/ValidationTestHelper.java b/core/src/test/java/org/apache/xtable/ValidationTestHelper.java
index 49b50b39..aa65e9ca 100644
--- a/core/src/test/java/org/apache/xtable/ValidationTestHelper.java
+++ b/core/src/test/java/org/apache/xtable/ValidationTestHelper.java
@@ -29,7 +29,7 @@ import java.util.stream.IntStream;
import org.apache.xtable.model.OneSnapshot;
import org.apache.xtable.model.TableChange;
-import org.apache.xtable.model.storage.OneDataFile;
+import org.apache.xtable.model.storage.InternalDataFile;
public class ValidationTestHelper {
@@ -39,7 +39,7 @@ public class ValidationTestHelper {
List<String> onetablePaths =
oneSnapshot.getPartitionedDataFiles().stream()
.flatMap(group -> group.getFiles().stream())
- .map(OneDataFile::getPhysicalPath)
+ .map(InternalDataFile::getPhysicalPath)
.collect(Collectors.toList());
replaceFileScheme(allActivePaths);
replaceFileScheme(onetablePaths);
@@ -90,12 +90,12 @@ public class ValidationTestHelper {
public static List<String> getAllFilePaths(OneSnapshot oneSnapshot) {
return oneSnapshot.getPartitionedDataFiles().stream()
.flatMap(oneFileGroup -> oneFileGroup.getFiles().stream())
- .map(oneDataFile -> oneDataFile.getPhysicalPath())
+ .map(InternalDataFile::getPhysicalPath)
.collect(Collectors.toList());
}
- private static Set<String> extractPathsFromDataFile(Set<OneDataFile> dataFiles) {
- return dataFiles.stream().map(OneDataFile::getPhysicalPath).collect(Collectors.toSet());
+ private static Set<String> extractPathsFromDataFile(Set<InternalDataFile> dataFiles) {
+ return dataFiles.stream().map(InternalDataFile::getPhysicalPath).collect(Collectors.toSet());
}
private static void replaceFileScheme(List<String> filePaths) {
diff --git a/core/src/test/java/org/apache/xtable/delta/ITDeltaSourceClient.java b/core/src/test/java/org/apache/xtable/delta/ITDeltaSourceClient.java
index 72d46378..03cf602b 100644
--- a/core/src/test/java/org/apache/xtable/delta/ITDeltaSourceClient.java
+++ b/core/src/test/java/org/apache/xtable/delta/ITDeltaSourceClient.java
@@ -70,11 +70,8 @@ import org.apache.xtable.model.schema.SchemaVersion;
import org.apache.xtable.model.stat.ColumnStat;
import org.apache.xtable.model.stat.PartitionValue;
import org.apache.xtable.model.stat.Range;
-import org.apache.xtable.model.storage.DataLayoutStrategy;
-import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneFileGroup;
-import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.storage.*;
+import org.apache.xtable.model.storage.InternalDataFile;
public class ITDeltaSourceClient {
@@ -193,7 +190,7 @@ public class ITDeltaSourceClient {
OneFileGroup.builder()
.files(
Collections.singletonList(
- OneDataFile.builder()
+ InternalDataFile.builder()
.physicalPath("file:/fake/path")
.fileFormat(FileFormat.APACHE_PARQUET)
.partitionValues(Collections.emptyList())
@@ -277,7 +274,7 @@ public class ITDeltaSourceClient {
.partitionValues(partitionValue)
.files(
Collections.singletonList(
- OneDataFile.builder()
+ InternalDataFile.builder()
.physicalPath("file:/fake/path")
.fileFormat(FileFormat.APACHE_PARQUET)
.partitionValues(partitionValue)
@@ -679,17 +676,18 @@ public class ITDeltaSourceClient {
validateDataFiles(expectedPartitionFiles.getFiles(), actualPartitionFiles.getFiles());
}
- private void validateDataFiles(List<OneDataFile> expectedFiles, List<OneDataFile> actualFiles)
+ private void validateDataFiles(
+ List<InternalDataFile> expectedFiles, List<InternalDataFile> actualFiles)
throws URISyntaxException {
Assertions.assertEquals(expectedFiles.size(), actualFiles.size());
for (int i = 0; i < expectedFiles.size(); i++) {
- OneDataFile expected = expectedFiles.get(i);
- OneDataFile actual = actualFiles.get(i);
+ InternalDataFile expected = expectedFiles.get(i);
+ InternalDataFile actual = actualFiles.get(i);
validatePropertiesDataFile(expected, actual);
}
}
- private void validatePropertiesDataFile(OneDataFile expected, OneDataFile actual)
+ private void validatePropertiesDataFile(InternalDataFile expected, InternalDataFile actual)
throws URISyntaxException {
Assertions.assertEquals(expected.getSchemaVersion(), actual.getSchemaVersion());
Assertions.assertTrue(
diff --git a/core/src/test/java/org/apache/xtable/delta/TestDeltaSync.java b/core/src/test/java/org/apache/xtable/delta/TestDeltaSync.java
index 807e6d2d..5bd352cf 100644
--- a/core/src/test/java/org/apache/xtable/delta/TestDeltaSync.java
+++ b/core/src/test/java/org/apache/xtable/delta/TestDeltaSync.java
@@ -88,7 +88,7 @@ import org.apache.xtable.model.stat.PartitionValue;
import org.apache.xtable.model.stat.Range;
import org.apache.xtable.model.storage.DataLayoutStrategy;
import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.OneFileGroup;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.schema.SchemaFieldFinder;
@@ -149,9 +149,9 @@ public class TestDeltaSync {
OneTable table1 = getOneTable(tableName, basePath, schema1, null, LAST_COMMIT_TIME);
OneTable table2 = getOneTable(tableName, basePath, schema2, null, LAST_COMMIT_TIME);
- OneDataFile dataFile1 = getOneDataFile(1, Collections.emptyList(), basePath);
- OneDataFile dataFile2 = getOneDataFile(2, Collections.emptyList(), basePath);
- OneDataFile dataFile3 = getOneDataFile(3, Collections.emptyList(), basePath);
+ InternalDataFile dataFile1 = getDataFile(1, Collections.emptyList(), basePath);
+ InternalDataFile dataFile2 = getDataFile(2, Collections.emptyList(), basePath);
+ InternalDataFile dataFile3 = getDataFile(3, Collections.emptyList(), basePath);
OneSnapshot snapshot1 = buildSnapshot(table1, dataFile1, dataFile2);
OneSnapshot snapshot2 = buildSnapshot(table2, dataFile2, dataFile3);
@@ -195,9 +195,9 @@ public class TestDeltaSync {
.partitionField(onePartitionField)
.range(Range.scalar("warning"))
.build());
- OneDataFile dataFile1 = getOneDataFile(1, partitionValues1, basePath);
- OneDataFile dataFile2 = getOneDataFile(2, partitionValues1, basePath);
- OneDataFile dataFile3 = getOneDataFile(3, partitionValues2, basePath);
+ InternalDataFile dataFile1 = getDataFile(1, partitionValues1, basePath);
+ InternalDataFile dataFile2 = getDataFile(2, partitionValues1, basePath);
+ InternalDataFile dataFile3 = getDataFile(3, partitionValues2, basePath);
EqualTo equalToExpr =
new EqualTo(new Column("string_field", new StringType()), Literal.of("warning"));
@@ -268,9 +268,9 @@ public class TestDeltaSync {
.range(Range.scalar(20))
.build());
- OneDataFile dataFile1 = getOneDataFile(1, partitionValues1, basePath);
- OneDataFile dataFile2 = getOneDataFile(2, partitionValues2, basePath);
- OneDataFile dataFile3 = getOneDataFile(3, partitionValues3, basePath);
+ InternalDataFile dataFile1 = getDataFile(1, partitionValues1, basePath);
+ InternalDataFile dataFile2 = getDataFile(2, partitionValues2, basePath);
+ InternalDataFile dataFile3 = getDataFile(3, partitionValues3, basePath);
EqualTo equalToExpr1 =
new EqualTo(new Column("string_field", new StringType()), Literal.of("level"));
@@ -315,9 +315,9 @@ public class TestDeltaSync {
.partitionField(partitionField)
.range(Range.scalar(Instant.parse("2022-10-03T00:00:00.00Z").toEpochMilli()))
.build());
- OneDataFile dataFile1 = getOneDataFile(1, partitionValues1, basePath);
- OneDataFile dataFile2 = getOneDataFile(2, partitionValues1, basePath);
- OneDataFile dataFile3 = getOneDataFile(3, partitionValues2, basePath);
+ InternalDataFile dataFile1 = getDataFile(1, partitionValues1, basePath);
+ InternalDataFile dataFile2 = getDataFile(2, partitionValues1, basePath);
+ InternalDataFile dataFile3 = getDataFile(3, partitionValues2, basePath);
OneSnapshot snapshot1 = buildSnapshot(table, dataFile1, dataFile2, dataFile3);
@@ -357,7 +357,7 @@ public class TestDeltaSync {
}
private void validateDeltaTable(
- Path basePath, Set<OneDataFile> oneDataFiles, Expression filterExpression)
+ Path basePath, Set<InternalDataFile> internalDataFiles, Expression filterExpression)
throws IOException {
DeltaLog deltaLog = DeltaLog.forTable(new Configuration(), basePath.toString());
assertTrue(deltaLog.tableExists());
@@ -370,26 +370,26 @@ public class TestDeltaSync {
} else {
deltaScan = snapshot.scan(filterExpression);
}
- Map<String, OneDataFile> pathToFile =
- oneDataFiles.stream()
- .collect(Collectors.toMap(OneDataFile::getPhysicalPath, Function.identity()));
+ Map<String, InternalDataFile> pathToFile =
+ internalDataFiles.stream()
+ .collect(Collectors.toMap(InternalDataFile::getPhysicalPath, Function.identity()));
int count = 0;
try (CloseableIterator<AddFile> fileItr = deltaScan.getFiles()) {
for (CloseableIterator<AddFile> it = fileItr; it.hasNext(); ) {
AddFile addFile = it.next();
String fullPath =
new org.apache.hadoop.fs.Path(basePath.resolve(addFile.getPath()).toUri()).toString();
- OneDataFile expected = pathToFile.get(fullPath);
+ InternalDataFile expected = pathToFile.get(fullPath);
assertNotNull(expected);
assertEquals(addFile.getSize(), expected.getFileSizeBytes());
count++;
}
}
assertEquals(
- oneDataFiles.size(), count, "Number of files from DeltaScan don't match expectation");
+ internalDataFiles.size(), count, "Number of files from DeltaScan don't match expectation");
}
- private OneSnapshot buildSnapshot(OneTable table, OneDataFile... dataFiles) {
+ private OneSnapshot buildSnapshot(OneTable table, InternalDataFile... dataFiles) {
return OneSnapshot.builder()
.table(table)
.partitionedDataFiles(OneFileGroup.fromFiles(Arrays.asList(dataFiles)))
@@ -413,12 +413,12 @@ public class TestDeltaSync {
.build();
}
- private OneDataFile getOneDataFile(
+ private InternalDataFile getDataFile(
int index, List<PartitionValue> partitionValues, Path basePath) {
String physicalPath =
new org.apache.hadoop.fs.Path(basePath.toUri() + "physical" + index + ".parquet")
.toString();
- return OneDataFile.builder()
+ return InternalDataFile.builder()
.fileFormat(FileFormat.APACHE_PARQUET)
.fileSizeBytes(RANDOM.nextInt(10000))
.physicalPath(physicalPath)
diff --git a/core/src/test/java/org/apache/xtable/hudi/ITHudiTargetClient.java b/core/src/test/java/org/apache/xtable/hudi/ITHudiTargetClient.java
index 0cf012a2..8b4fc533 100644
--- a/core/src/test/java/org/apache/xtable/hudi/ITHudiTargetClient.java
+++ b/core/src/test/java/org/apache/xtable/hudi/ITHudiTargetClient.java
@@ -81,10 +81,10 @@ import org.apache.xtable.model.schema.PartitionTransformType;
import org.apache.xtable.model.stat.ColumnStat;
import org.apache.xtable.model.stat.PartitionValue;
import org.apache.xtable.model.stat.Range;
+import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.DataLayoutStrategy;
import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.OneFileGroup;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.spi.sync.TargetClient;
@@ -175,8 +175,8 @@ public class ITHudiTargetClient {
Collections.emptyMap());
}
- OneDataFile fileToRemove =
- OneDataFile.builder()
+ InternalDataFile fileToRemove =
+ InternalDataFile.builder()
.fileFormat(FileFormat.APACHE_PARQUET)
.lastModified(System.currentTimeMillis())
.fileSizeBytes(100L)
@@ -189,8 +189,8 @@ public class ITHudiTargetClient {
String fileName = "file_1.parquet";
String filePath = getFilePath(partitionPath, fileName);
- OneDataFilesDiff dataFilesDiff =
- OneDataFilesDiff.builder()
+ DataFilesDiff dataFilesDiff =
+ DataFilesDiff.builder()
.fileAdded(getTestFile(partitionPath, fileName))
.fileRemoved(fileToRemove)
.build();
@@ -371,11 +371,11 @@ public class ITHudiTargetClient {
private OneTableMetadata incrementalSync(
TargetClient targetClient,
- List<OneDataFile> filesToAdd,
- List<OneDataFile> filesToRemove,
+ List<InternalDataFile> filesToAdd,
+ List<InternalDataFile> filesToRemove,
Instant commitStart) {
- OneDataFilesDiff dataFilesDiff2 =
- OneDataFilesDiff.builder().filesAdded(filesToAdd).filesRemoved(filesToRemove).build();
+ DataFilesDiff dataFilesDiff2 =
+ DataFilesDiff.builder().filesAdded(filesToAdd).filesRemoved(filesToRemove).build();
OneTable state3 = getState(commitStart);
targetClient.beginSync(state3);
targetClient.syncFilesForDiff(dataFilesDiff2);
@@ -527,7 +527,7 @@ public class ITHudiTargetClient {
assertEquals(-1, columnStats.getTotalUncompressedSize());
}
- private OneDataFile getTestFile(String partitionPath, String fileName) {
+ private InternalDataFile getTestFile(String partitionPath, String fileName) {
List<ColumnStat> columnStats =
Arrays.asList(
ColumnStat.builder()
@@ -551,7 +551,7 @@ public class ITHudiTargetClient {
.numValues(2)
.totalSize(5)
.build());
- return OneDataFile.builder()
+ return InternalDataFile.builder()
.schemaVersion(SCHEMA_VERSION)
.physicalPath(String.format("file://%s/%s/%s", tableBasePath, partitionPath, fileName))
.fileSizeBytes(FILE_SIZE)
diff --git a/core/src/test/java/org/apache/xtable/hudi/TestBaseFileUpdatesExtractor.java b/core/src/test/java/org/apache/xtable/hudi/TestBaseFileUpdatesExtractor.java
index f183d418..83efd1ba 100644
--- a/core/src/test/java/org/apache/xtable/hudi/TestBaseFileUpdatesExtractor.java
+++ b/core/src/test/java/org/apache/xtable/hudi/TestBaseFileUpdatesExtractor.java
@@ -60,9 +60,9 @@ import org.apache.xtable.model.schema.PartitionTransformType;
import org.apache.xtable.model.stat.ColumnStat;
import org.apache.xtable.model.stat.PartitionValue;
import org.apache.xtable.model.stat.Range;
+import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.OneFileGroup;
import org.apache.xtable.testutil.ColumnStatMapUtil;
@@ -90,38 +90,38 @@ public class TestBaseFileUpdatesExtractor {
String partitionPath1 = "partition1";
String fileName1 = "file1.parquet";
// create file with empty stats to test edge case
- OneDataFile addedFile1 =
+ InternalDataFile addedFile1 =
createFile(
String.format("%s/%s/%s", tableBasePath, partitionPath1, fileName1),
Collections.emptyList());
// create file with stats
String partitionPath2 = "partition2";
String fileName2 = "file2.parquet";
- OneDataFile addedFile2 =
+ InternalDataFile addedFile2 =
createFile(
String.format("%s/%s/%s", tableBasePath, partitionPath2, fileName2), getColumnStats());
// remove files 3 files from two different partitions
String fileName3 = "file3.parquet";
- OneDataFile removedFile1 =
+ InternalDataFile removedFile1 =
createFile(
String.format("%s/%s/%s", tableBasePath, partitionPath1, fileName3), getColumnStats());
// create file that matches hudi format to mimic that a file create by hudi is now being removed
// by another system
String fileIdForFile4 = "d1cf0980-445c-4c74-bdeb-b7e5d18779f5-0";
String fileName4 = fileIdForFile4 + "_0-1116-142216_20231003013807542.parquet";
- OneDataFile removedFile2 =
+ InternalDataFile removedFile2 =
createFile(
String.format("%s/%s/%s", tableBasePath, partitionPath1, fileName4),
Collections.emptyList());
String fileName5 = "file5.parquet";
- OneDataFile removedFile3 =
+ InternalDataFile removedFile3 =
createFile(
String.format("%s/%s/%s", tableBasePath, partitionPath2, fileName5),
Collections.emptyList());
- OneDataFilesDiff diff =
- OneDataFilesDiff.builder()
+ DataFilesDiff diff =
+ DataFilesDiff.builder()
.filesAdded(Arrays.asList(addedFile1, addedFile2))
.filesRemoved(Arrays.asList(removedFile1, removedFile2, removedFile3))
.build();
@@ -161,19 +161,19 @@ public class TestBaseFileUpdatesExtractor {
String partitionPath1 = "partition1";
String fileName1 = "file1.parquet";
// create file with empty stats to test edge case
- OneDataFile addedFile1 =
+ InternalDataFile addedFile1 =
createFile(
String.format("%s/%s/%s", tableBasePath, partitionPath1, fileName1),
Collections.emptyList());
// create file with stats
String fileName2 = "file2.parquet";
- OneDataFile addedFile2 =
+ InternalDataFile addedFile2 =
createFile(
String.format("%s/%s/%s", tableBasePath, partitionPath1, fileName2), getColumnStats());
// create file in a second partition
String partitionPath2 = "partition2";
String fileName3 = "file3.parquet";
- OneDataFile addedFile3 =
+ InternalDataFile addedFile3 =
createFile(
String.format("%s/%s/%s", tableBasePath, partitionPath2, fileName3), getColumnStats());
@@ -257,17 +257,17 @@ public class TestBaseFileUpdatesExtractor {
// create a snapshot without partition1 (dropped partition), a new file added to partition 2
// along with one of the existing files, and a new file in partition 3
String newFileName1 = "new_file_1.parquet";
- OneDataFile addedFile1 =
+ InternalDataFile addedFile1 =
createFile(
String.format("%s/%s/%s", tableBasePath, partitionPath2, newFileName1),
Collections.emptyList());
String newFileName2 = "new_file_2.parquet";
- OneDataFile addedFile2 =
+ InternalDataFile addedFile2 =
createFile(
String.format("%s/%s/%s", tableBasePath, partitionPath3, newFileName2),
getColumnStats());
- // OneDataFile for one of the existing files in partition2
- OneDataFile existingFile =
+ // InternalDataFile for one of the existing files in partition2
+ InternalDataFile existingFile =
createFile(
String.format("%s/%s/%s", tableBasePath, partitionPath2, existingFileName2),
Collections.emptyList());
@@ -350,10 +350,10 @@ public class TestBaseFileUpdatesExtractor {
HoodieTableMetaClient metaClient = HoodieTableMetaClient.reload(setupMetaClient);
// create a snapshot with a new file added along with one of the existing files
String newFileName1 = "new_file_1.parquet";
- OneDataFile addedFile1 =
+ InternalDataFile addedFile1 =
createFile(String.format("%s/%s", tableBasePath, newFileName1), getColumnStats());
- // OneDataFile for one of the existing files in partition2
- OneDataFile existingFile =
+ // InternalDataFile for one of the existing files in partition2
+ InternalDataFile existingFile =
createFile(
String.format("%s/%s", tableBasePath, existingFileName2), Collections.emptyList());
List<OneFileGroup> partitionedDataFiles =
@@ -388,8 +388,8 @@ public class TestBaseFileUpdatesExtractor {
actual.stream().map(WriteStatus::toString).collect(Collectors.toSet()));
}
- private OneDataFile createFile(String physicalPath, List<ColumnStat> columnStats) {
- return OneDataFile.builder()
+ private InternalDataFile createFile(String physicalPath, List<ColumnStat> columnStats) {
+ return InternalDataFile.builder()
.schemaVersion(SCHEMA_VERSION)
.physicalPath(physicalPath)
.fileSizeBytes(FILE_SIZE)
diff --git a/core/src/test/java/org/apache/xtable/hudi/TestHudiFileStatsExtractor.java b/core/src/test/java/org/apache/xtable/hudi/TestHudiFileStatsExtractor.java
index 994b239d..bb0d08a4 100644
--- a/core/src/test/java/org/apache/xtable/hudi/TestHudiFileStatsExtractor.java
+++ b/core/src/test/java/org/apache/xtable/hudi/TestHudiFileStatsExtractor.java
@@ -72,7 +72,7 @@ import org.apache.xtable.model.schema.OneType;
import org.apache.xtable.model.schema.SchemaVersion;
import org.apache.xtable.model.stat.ColumnStat;
import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
+import org.apache.xtable.model.storage.InternalDataFile;
public class TestHudiFileStatsExtractor {
private static final Schema AVRO_SCHEMA =
@@ -142,8 +142,8 @@ public class TestHudiFileStatsExtractor {
.filter(path -> path.toString().endsWith(".parquet"))
.findFirst()
.orElseThrow(() -> new RuntimeException("No files found"));
- OneDataFile inputFile =
- OneDataFile.builder()
+ InternalDataFile inputFile =
+ InternalDataFile.builder()
.physicalPath(parquetFile.toString())
.schemaVersion(new SchemaVersion(1, null))
.columnStats(Collections.emptyList())
@@ -155,7 +155,7 @@ public class TestHudiFileStatsExtractor {
HoodieTableMetaClient metaClient =
HoodieTableMetaClient.builder().setBasePath(basePath).setConf(configuration).build();
HudiFileStatsExtractor fileStatsExtractor = new HudiFileStatsExtractor(metaClient);
- List<OneDataFile> output =
+ List<InternalDataFile> output =
fileStatsExtractor
.addStatsToFiles(tableMetadata, Stream.of(inputFile), schema)
.collect(Collectors.toList());
@@ -179,8 +179,8 @@ public class TestHudiFileStatsExtractor {
}
}
- OneDataFile inputFile =
- OneDataFile.builder()
+ InternalDataFile inputFile =
+ InternalDataFile.builder()
.physicalPath(file.toString())
.schemaVersion(new SchemaVersion(1, null))
.columnStats(Collections.emptyList())
@@ -193,16 +193,16 @@ public class TestHudiFileStatsExtractor {
HoodieTableMetaClient mockMetaClient = mock(HoodieTableMetaClient.class);
when(mockMetaClient.getHadoopConf()).thenReturn(configuration);
HudiFileStatsExtractor fileStatsExtractor = new HudiFileStatsExtractor(mockMetaClient);
- List<OneDataFile> output =
+ List<InternalDataFile> output =
fileStatsExtractor
.addStatsToFiles(null, Stream.of(inputFile), schema)
.collect(Collectors.toList());
validateOutput(output);
}
- private void validateOutput(List<OneDataFile> output) {
+ private void validateOutput(List<InternalDataFile> output) {
assertEquals(1, output.size());
- OneDataFile fileWithStats = output.get(0);
+ InternalDataFile fileWithStats = output.get(0);
assertEquals(2, fileWithStats.getRecordCount());
List<ColumnStat> columnStats = fileWithStats.getColumnStats();
diff --git a/core/src/test/java/org/apache/xtable/hudi/TestHudiTargetClient.java b/core/src/test/java/org/apache/xtable/hudi/TestHudiTargetClient.java
index 2c132ead..00b28112 100644
--- a/core/src/test/java/org/apache/xtable/hudi/TestHudiTargetClient.java
+++ b/core/src/test/java/org/apache/xtable/hudi/TestHudiTargetClient.java
@@ -49,7 +49,7 @@ import org.apache.xtable.model.schema.OnePartitionField;
import org.apache.xtable.model.schema.OneSchema;
import org.apache.xtable.model.schema.OneType;
import org.apache.xtable.model.schema.PartitionTransformType;
-import org.apache.xtable.model.storage.OneDataFilesDiff;
+import org.apache.xtable.model.storage.DataFilesDiff;
import org.apache.xtable.model.storage.OneFileGroup;
/**
@@ -203,7 +203,7 @@ public class TestHudiTargetClient {
HudiTargetClient targetClient = getTargetClient(null);
HudiTargetClient.CommitState mockCommitState = initMocksForBeginSync(targetClient).getLeft();
String instant = "commit";
- OneDataFilesDiff input = OneDataFilesDiff.builder().build();
+ DataFilesDiff input = DataFilesDiff.builder().build();
BaseFileUpdatesExtractor.ReplaceMetadata output =
BaseFileUpdatesExtractor.ReplaceMetadata.of(
Collections.emptyMap(), Collections.emptyList());
diff --git a/core/src/test/java/org/apache/xtable/iceberg/ITIcebergSourceClient.java b/core/src/test/java/org/apache/xtable/iceberg/ITIcebergSourceClient.java
index ec19baec..22f8d034 100644
--- a/core/src/test/java/org/apache/xtable/iceberg/ITIcebergSourceClient.java
+++ b/core/src/test/java/org/apache/xtable/iceberg/ITIcebergSourceClient.java
@@ -21,7 +21,9 @@ package org.apache.xtable.iceberg;
import static org.apache.xtable.GenericTable.getTableName;
import static org.apache.xtable.ValidationTestHelper.validateOneSnapshot;
import static org.apache.xtable.ValidationTestHelper.validateTableChanges;
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import java.nio.file.Path;
import java.time.Instant;
diff --git a/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSourceClient.java b/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSourceClient.java
index 9d799884..f8076209 100644
--- a/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSourceClient.java
+++ b/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSourceClient.java
@@ -26,7 +26,10 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Instant;
-import java.util.*;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
@@ -59,11 +62,9 @@ import org.apache.xtable.model.schema.PartitionTransformType;
import org.apache.xtable.model.schema.SchemaCatalog;
import org.apache.xtable.model.schema.SchemaVersion;
import org.apache.xtable.model.stat.PartitionValue;
-import org.apache.xtable.model.storage.DataLayoutStrategy;
+import org.apache.xtable.model.storage.*;
import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
-import org.apache.xtable.model.storage.OneFileGroup;
-import org.apache.xtable.model.storage.TableFormat;
+import org.apache.xtable.model.storage.InternalDataFile;
class TestIcebergSourceClient {
@@ -176,20 +177,20 @@ class TestIcebergSourceClient {
List<OneFileGroup> dataFileChunks = oneSnapshot.getPartitionedDataFiles();
assertEquals(5, dataFileChunks.size());
for (OneFileGroup dataFilesChunk : dataFileChunks) {
- List<OneDataFile> oneDataFiles = dataFilesChunk.getFiles();
- assertEquals(1, oneDataFiles.size());
- OneDataFile oneDataFile = oneDataFiles.get(0);
- assertEquals(FileFormat.APACHE_PARQUET, oneDataFile.getFileFormat());
- assertEquals(1, oneDataFile.getRecordCount());
- Assertions.assertTrue(oneDataFile.getPhysicalPath().startsWith("file:" + workingDir));
-
- List<PartitionValue> partitionValues = oneDataFile.getPartitionValues();
+ List<InternalDataFile> internalDataFiles = dataFilesChunk.getFiles();
+ assertEquals(1, internalDataFiles.size());
+ InternalDataFile internalDataFile = internalDataFiles.get(0);
+ assertEquals(FileFormat.APACHE_PARQUET, internalDataFile.getFileFormat());
+ assertEquals(1, internalDataFile.getRecordCount());
+ Assertions.assertTrue(internalDataFile.getPhysicalPath().startsWith("file:" + workingDir));
+
+ List<PartitionValue> partitionValues = internalDataFile.getPartitionValues();
assertEquals(1, partitionValues.size());
PartitionValue partitionEntry = partitionValues.iterator().next();
assertEquals(
"cs_sold_date_sk", partitionEntry.getPartitionField().getSourceField().getName());
// TODO generate test with column stats
- assertEquals(0, oneDataFile.getColumnStats().size());
+ assertEquals(0, internalDataFile.getColumnStats().size());
}
}
diff --git a/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java b/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java
index 3cc9dd95..9fc8ef11 100644
--- a/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java
+++ b/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java
@@ -97,7 +97,7 @@ import org.apache.xtable.model.stat.PartitionValue;
import org.apache.xtable.model.stat.Range;
import org.apache.xtable.model.storage.DataLayoutStrategy;
import org.apache.xtable.model.storage.FileFormat;
-import org.apache.xtable.model.storage.OneDataFile;
+import org.apache.xtable.model.storage.InternalDataFile;
import org.apache.xtable.model.storage.OneFileGroup;
import org.apache.xtable.model.storage.TableFormat;
import org.apache.xtable.schema.SchemaFieldFinder;
@@ -224,9 +224,9 @@ public class TestIcebergSync {
SchemaVersion schemaVersion2 = new SchemaVersion(2, "");
schemas.put(schemaVersion2, schema2);
- OneDataFile dataFile1 = getOneDataFile(schemaVersion1, 1, Collections.emptyList());
- OneDataFile dataFile2 = getOneDataFile(schemaVersion1, 2, Collections.emptyList());
- OneDataFile dataFile3 = getOneDataFile(schemaVersion2, 3, Collections.emptyList());
+ InternalDataFile dataFile1 = getDataFile(schemaVersion1, 1, Collections.emptyList());
+ InternalDataFile dataFile2 = getDataFile(schemaVersion1, 2, Collections.emptyList());
+ InternalDataFile dataFile3 = getDataFile(schemaVersion2, 3, Collections.emptyList());
OneSnapshot snapshot1 = buildSnapshot(table1, schemas, dataFile1, dataFile2);
OneSnapshot snapshot2 = buildSnapshot(table2, schemas, dataFile2, dataFile3);
when(mockSchemaExtractor.toIceberg(oneSchema)).thenReturn(icebergSchema);
@@ -320,10 +320,10 @@ public class TestIcebergSync {
SchemaVersion schemaVersion2 = new SchemaVersion(2, "");
schemas.put(schemaVersion2, schema2);
- OneDataFile dataFile1 = getOneDataFile(schemaVersion1, 1, Collections.emptyList());
- OneDataFile dataFile2 = getOneDataFile(schemaVersion1, 2, Collections.emptyList());
- OneDataFile dataFile3 = getOneDataFile(schemaVersion2, 3, Collections.emptyList());
- OneDataFile dataFile4 = getOneDataFile(schemaVersion2, 4, Collections.emptyList());
+ InternalDataFile dataFile1 = getDataFile(schemaVersion1, 1, Collections.emptyList());
+ InternalDataFile dataFile2 = getDataFile(schemaVersion1, 2, Collections.emptyList());
+ InternalDataFile dataFile3 = getDataFile(schemaVersion2, 3, Collections.emptyList());
+ InternalDataFile dataFile4 = getDataFile(schemaVersion2, 4, Collections.emptyList());
OneSnapshot snapshot1 = buildSnapshot(table1, schemas, dataFile1, dataFile2);
OneSnapshot snapshot2 = buildSnapshot(table2, schemas, dataFile2, dataFile3);
OneSnapshot snapshot3 = buildSnapshot(table2, schemas, dataFile3, dataFile4);
@@ -397,9 +397,9 @@ public class TestIcebergSync {
.partitionField(partitionField)
.range(Range.scalar(Instant.parse("2022-10-03T00:00:00.00Z").toEpochMilli()))
.build());
- OneDataFile dataFile1 = getOneDataFile(schemaVersion, 1, partitionValues1);
- OneDataFile dataFile2 = getOneDataFile(schemaVersion, 2, partitionValues1);
- OneDataFile dataFile3 = getOneDataFile(schemaVersion, 3, partitionValues2);
+ InternalDataFile dataFile1 = getDataFile(schemaVersion, 1, partitionValues1);
+ InternalDataFile dataFile2 = getDataFile(schemaVersion, 2, partitionValues1);
+ InternalDataFile dataFile3 = getDataFile(schemaVersion, 3, partitionValues2);
OneSnapshot snapshot = buildSnapshot(table, schemas, dataFile1, dataFile2, dataFile3);
when(mockSchemaExtractor.toIceberg(oneSchema))
@@ -461,9 +461,9 @@ public class TestIcebergSync {
.partitionField(partitionField)
.range(Range.scalar(Instant.parse("2022-10-03T00:00:00.00Z").toEpochMilli()))
.build());
- OneDataFile dataFile1 = getOneDataFile(schemaVersion, 1, partitionValues1);
- OneDataFile dataFile2 = getOneDataFile(schemaVersion, 2, partitionValues1);
- OneDataFile dataFile3 = getOneDataFile(schemaVersion, 3, partitionValues2);
+ InternalDataFile dataFile1 = getDataFile(schemaVersion, 1, partitionValues1);
+ InternalDataFile dataFile2 = getDataFile(schemaVersion, 2, partitionValues1);
+ InternalDataFile dataFile3 = getDataFile(schemaVersion, 3, partitionValues2);
OneSnapshot snapshot = buildSnapshot(table, schemas, dataFile1, dataFile2, dataFile3);
when(mockSchemaExtractor.toIceberg(oneSchema)).thenReturn(icebergSchema);
@@ -516,9 +516,9 @@ public class TestIcebergSync {
List<PartitionValue> partitionValues2 =
Collections.singletonList(
PartitionValue.builder().partitionField(partitionField).range(Range.scalar(2)).build());
- OneDataFile dataFile1 = getOneDataFile(schemaVersion, 1, partitionValues1);
- OneDataFile dataFile2 = getOneDataFile(schemaVersion, 2, partitionValues1);
- OneDataFile dataFile3 = getOneDataFile(schemaVersion, 3, partitionValues2);
+ InternalDataFile dataFile1 = getDataFile(schemaVersion, 1, partitionValues1);
+ InternalDataFile dataFile2 = getDataFile(schemaVersion, 2, partitionValues1);
+ InternalDataFile dataFile3 = getDataFile(schemaVersion, 3, partitionValues2);
OneSnapshot snapshot = buildSnapshot(table, schemas, dataFile1, dataFile2, dataFile3);
when(mockSchemaExtractor.toIceberg(oneSchema)).thenReturn(icebergSchema);
@@ -592,9 +592,9 @@ public class TestIcebergSync {
.partitionField(partitionField2)
.range(Range.scalar(Instant.parse("2022-10-03T00:00:00.00Z").toEpochMilli()))
.build());
- OneDataFile dataFile1 = getOneDataFile(schemaVersion, 1, partitionValues1);
- OneDataFile dataFile2 = getOneDataFile(schemaVersion, 2, partitionValues2);
- OneDataFile dataFile3 = getOneDataFile(schemaVersion, 3, partitionValues3);
+ InternalDataFile dataFile1 = getDataFile(schemaVersion, 1, partitionValues1);
+ InternalDataFile dataFile2 = getDataFile(schemaVersion, 2, partitionValues2);
+ InternalDataFile dataFile3 = getDataFile(schemaVersion, 3, partitionValues3);
OneSnapshot snapshot = buildSnapshot(table, schemas, dataFile1, dataFile2, dataFile3);
when(mockSchemaExtractor.toIceberg(oneSchema)).thenReturn(icebergSchema);
@@ -659,9 +659,9 @@ public class TestIcebergSync {
.partitionField(partitionField)
.range(Range.scalar("value2"))
.build());
- OneDataFile dataFile1 = getOneDataFile(schemaVersion, 1, partitionValues1);
- OneDataFile dataFile2 = getOneDataFile(schemaVersion, 2, partitionValues1);
- OneDataFile dataFile3 = getOneDataFile(schemaVersion, 3, partitionValues2);
+ InternalDataFile dataFile1 = getDataFile(schemaVersion, 1, partitionValues1);
+ InternalDataFile dataFile2 = getDataFile(schemaVersion, 2, partitionValues1);
+ InternalDataFile dataFile3 = getDataFile(schemaVersion, 3, partitionValues2);
OneSnapshot snapshot = buildSnapshot(table, schemas, dataFile1, dataFile2, dataFile3);
when(mockSchemaExtractor.toIceberg(oneSchema)).thenReturn(icebergSchema);
@@ -687,7 +687,7 @@ public class TestIcebergSync {
}
private OneSnapshot buildSnapshot(
- OneTable table, Map<SchemaVersion, OneSchema> schemas, OneDataFile... dataFiles) {
+ OneTable table, Map<SchemaVersion, OneSchema> schemas, InternalDataFile... dataFiles) {
return OneSnapshot.builder()
.table(table)
.schemaCatalog(SchemaCatalog.builder().schemas(schemas).build())
@@ -695,10 +695,10 @@ public class TestIcebergSync {
.build();
}
- private OneDataFile getOneDataFile(
+ private InternalDataFile getDataFile(
SchemaVersion schemaVersion, int index, List<PartitionValue> partitionValues) {
String physicalPath = "file:/physical" + index + ".parquet";
- return OneDataFile.builder()
+ return InternalDataFile.builder()
.fileFormat(FileFormat.APACHE_PARQUET)
.fileSizeBytes(RANDOM.nextInt(10000))
.physicalPath(physicalPath)
@@ -727,7 +727,10 @@ public class TestIcebergSync {
}
private void validateIcebergTable(
- String tableName, OneTable table, Set<OneDataFile> expectedFiles, Expression filterExpression)
+ String tableName,
+ OneTable table,
+ Set<InternalDataFile> expectedFiles,
+ Expression filterExpression)
throws IOException {
Path warehouseLocation = Paths.get(table.getBasePath()).getParent();
try (HadoopCatalog catalog = new HadoopCatalog(CONFIGURATION, warehouseLocation.toString())) {
@@ -741,12 +744,13 @@ public class TestIcebergSync {
assertEquals(1, Iterables.size(tasks), "1 combined scan task should be generated");
for (CombinedScanTask combinedScanTask : tasks) {
assertEquals(expectedFiles.size(), combinedScanTask.files().size());
- Map<String, OneDataFile> pathToFile =
+ Map<String, InternalDataFile> pathToFile =
expectedFiles.stream()
- .collect(Collectors.toMap(OneDataFile::getPhysicalPath, Function.identity()));
+ .collect(
+ Collectors.toMap(InternalDataFile::getPhysicalPath, Function.identity()));
for (FileScanTask fileScanTask : combinedScanTask.files()) {
// check that path and other stats match
- OneDataFile expected = pathToFile.get(fileScanTask.file().path());
+ InternalDataFile expected = pathToFile.get(fileScanTask.file().path());
assertNotNull(expected);
assertEquals(expected.getFileSizeBytes(), fileScanTask.file().fileSizeInBytes());
assertEquals(expected.getRecordCount(), fileScanTask.file().recordCount());
@@ -766,7 +770,7 @@ public class TestIcebergSync {
}
}
- private void mockColStatsForFile(OneDataFile dataFile, int times) {
+ private void mockColStatsForFile(InternalDataFile dataFile, int times) {
Metrics response = new Metrics(dataFile.getRecordCount(), null, null, null, null);
Metrics[] responses =
IntStream.of(times - 1).mapToObj(unused -> response).toArray(Metrics[]::new);