This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 790414fe8ce5 refactor: Rename dataFile/df variables to baseFile/bf
(#19944)
790414fe8ce5 is described below
commit 790414fe8ce530a8ad7e604c5d7dc0492178fc73
Author: Nanjesh Ramesh <[email protected]>
AuthorDate: Tue Oct 13 00:23:17 2026 -0700
refactor: Rename dataFile/df variables to baseFile/bf (#19944)
* [HUDI-541] Rename dataFile/df variables to baseFile/bf
HoodieBaseFile was previously named HoodieDataFile, and many call sites
across the codebase still name variables, parameters, and lambda
arguments of this type dataFile or df rather than baseFile or bf,
per the terminology convergence the cWiki design and arch page calls
for.
This renames every local variable, field, method parameter, and
lambda argument whose declared or inferred type is HoodieBaseFile (or
a direct List/Stream/Option/Collection of it) from dataFile/df to
baseFile/bf, across 35 files in hudi-common, hudi-hadoop-common,
hudi-hadoop-mr, hudi-sync/hudi-hive-sync, hudi-client-common,
hudi-spark-client, hudi-java-client, hudi-flink-client, and
hudi-spark-datasource.
Deliberately out of scope:
- Method names such as HoodieFileGroup#getLatestDataFile and
TestHoodieBaseFile#newDataFile are unchanged. Renaming a method,
especially a public one, is a materially different, higher-risk
change than renaming a local variable or parameter, and the issue's
own wording ("all variables of this type") does not ask for it.
- Identifiers that are not of type HoodieBaseFile even though they
contain "dataFile" in the name, for example dataFileName (String),
skipCreatingDataFile (boolean), and the several FileSlice-typed
orphan/inflightFileSliceWithDataFile locals in
TestHoodieTableFileSystemView. Each was checked individually against
its actual declared or inferred type before being included or
excluded.
- df/dataFile identifiers that are unrelated Spark DataFrames or other
types entirely, such as StreamSync's transformed.map(df -> ...) and
IncrementalTimelineSyncFileSystemView's
deltaFileGroups.stream().map(df -> df.getTimeline()), the latter
being a HoodieFileGroup, not a HoodieBaseFile, despite sitting one
line away from two genuine HoodieBaseFile df usages that were
renamed.
An earlier attempt at this same issue (#1650) was opened in 2022 and
closed unmerged after review stalled; a good number of the dataFile/df
sites it touched have since been renamed independently in other
changes, which is why this diff is smaller in file count than that
one.
Pure rename, no behavior change: 281 insertions, 281 deletions.
Verified by test-compiling every touched module (hudi-common,
hudi-hadoop-common, hudi-hadoop-mr, hudi-hive-sync, hudi-client-common,
hudi-spark-client, hudi-java-client, hudi-flink-client,
hudi-spark-datasource/hudi-spark and its spark3.5 dependency chain)
against JDK 11 with zero compile errors.
* [HUDI-541] Use baseFile consistently instead of bf
Addresses review nits: the rename previously used the abbreviated bf
in a number of lambda parameters and local variables while using the
full baseFile elsewhere. Converts every remaining bf introduced by
this PR to baseFile so the whole change uses one convention.
Files touched: CompactionAdminClient, AbstractTableFileSystemView,
IncrementalTimelineSyncFileSystemView, RocksDbBasedFileSystemView,
TestCleanerInsertAndCleanByVersions, TestHoodieCompactionStrategy,
TestHoodieTableFileSystemView, TestCompactionUtils.
Verified by test-compiling hudi-common, hudi-hadoop-common,
hudi-client-common, and hudi-spark-client against JDK 11 with no
errors, including no lambda-parameter shadowing conflicts.
* [HUDI-541] Wrap a line pushed over the checkstyle limit by the bf ->
baseFile rename
getLatestBaseFile's filter/findFirst chain was 196 characters with the
abbreviated bf; renaming both lambda parameters to baseFile pushed it
to 208, over the 200-character checkstyle limit and failing every CI
job that depends on hudi-common. Wraps the chain across three lines,
matching the wrapping style already used for the same filter pattern
elsewhere in this file (e.g. getLatestBaseFilesBeforeOrOnFromCache).
Verified with mvn checkstyle:check on hudi-common.
---
.../apache/hudi/client/CompactionAdminClient.java | 8 +-
.../org/apache/hudi/io/HoodieConcatHandle.java | 4 +-
.../apache/hudi/io/HoodieMergeHandleFactory.java | 4 +-
.../hudi/io/HoodieMergeHandleWithChangeLog.java | 4 +-
.../apache/hudi/io/HoodieSortedMergeHandle.java | 4 +-
.../io/HoodieSortedMergeHandleWithChangeLog.java | 4 +-
.../org/apache/hudi/io/HoodieWriteMergeHandle.java | 4 +-
.../apache/hudi/table/HoodieCompactionHandler.java | 2 +-
.../hudi/table/action/clean/CleanPlanner.java | 8 +-
.../BaseHoodieCompactionPlanGenerator.java | 4 +-
.../hudi/table/HoodieFlinkCopyOnWriteTable.java | 8 +-
.../hudi/table/HoodieJavaCopyOnWriteTable.java | 8 +-
.../org/apache/hudi/client/SparkRDDReadClient.java | 4 +-
.../index/bloom/HoodieFileProbingFunction.java | 6 +-
.../hudi/table/HoodieSparkCopyOnWriteTable.java | 8 +-
.../org/apache/hudi/client/TestClientRollback.java | 48 +++----
.../clean/TestCleanerInsertAndCleanByVersions.java | 16 +--
.../table/action/compact/CompactionTestBase.java | 4 +-
.../strategy/TestHoodieCompactionStrategy.java | 24 ++--
.../SparkClientFunctionalTestHarness.java | 8 +-
.../hudi/common/model/CompactionOperation.java | 14 +-
.../apache/hudi/common/model/HoodieBaseFile.java | 10 +-
.../apache/hudi/common/model/HoodieFileGroup.java | 8 +-
.../table/view/AbstractTableFileSystemView.java | 31 ++--
.../IncrementalTimelineSyncFileSystemView.java | 4 +-
.../table/view/RocksDbBasedFileSystemView.java | 8 +-
.../table/view/TestHoodieTableFileSystemView.java | 156 ++++++++++-----------
.../hudi/common/util/TestCompactionUtils.java | 4 +-
.../hudi/hadoop/utils/HoodieInputFormatUtils.java | 14 +-
.../TestHoodieSparkMergeOnReadTableClustering.java | 20 +--
.../hudi/table/TestHoodieMergeOnReadTable.java | 24 ++--
...HoodieSparkMergeOnReadTableIncrementalRead.java | 8 +-
...dieSparkMergeOnReadTableInsertUpdateDelete.java | 28 ++--
.../TestHoodieSparkMergeOnReadTableRollback.java | 32 ++---
.../apache/hudi/hive/testutils/HiveTestUtil.java | 24 ++--
35 files changed, 284 insertions(+), 281 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
index 6abe10c6cca7..d0990a6c1c12 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/CompactionAdminClient.java
@@ -232,17 +232,17 @@ public class CompactionAdminClient extends
BaseHoodieClient {
.filter(fs ->
fs.getFileId().equals(operation.getFileId())).findFirst());
if (fileSliceOptional.isPresent()) {
FileSlice fs = fileSliceOptional.get();
- Option<HoodieBaseFile> df = fs.getBaseFile();
+ Option<HoodieBaseFile> baseFile = fs.getBaseFile();
if (operation.getDataFileName().isPresent()) {
String expPath = metaClient.getStorage()
.getPathInfo(new StoragePath(
FSUtils.constructAbsolutePath(metaClient.getBasePath(),
operation.getPartitionPath()),
operation.getDataFileName().get()))
.getPath().toString();
- ValidationUtils.checkArgument(df.isPresent(),
+ ValidationUtils.checkArgument(baseFile.isPresent(),
"Data File must be present. File Slice was : " + fs + ",
operation :" + operation);
- ValidationUtils.checkArgument(df.get().getPath().equals(expPath),
- "Base Path in operation is specified as " + expPath + " but
got path " + df.get().getPath());
+
ValidationUtils.checkArgument(baseFile.get().getPath().equals(expPath),
+ "Base Path in operation is specified as " + expPath + " but
got path " + baseFile.get().getPath());
}
Set<HoodieLogFile> logFilesInFileSlice =
fs.getLogFiles().collect(Collectors.toSet());
Set<HoodieLogFile> logFilesInCompactionOp =
operation.getDeltaFileNames().stream().map(dp -> {
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
index 1d8823c67183..4a5a13b8f851 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieConcatHandle.java
@@ -84,8 +84,8 @@ public class HoodieConcatHandle<T, I, K, O> extends
HoodieWriteMergeHandle<T, I,
public HoodieConcatHandle(HoodieWriteConfig config, String instantTime,
HoodieTable hoodieTable,
Map<String, HoodieRecord<T>> keyToNewRecords,
String partitionPath, String fileId,
- HoodieBaseFile dataFileToBeMerged,
TaskContextSupplier taskContextSupplier) {
- super(config, instantTime, hoodieTable, Collections.emptyMap(),
partitionPath, fileId, dataFileToBeMerged, taskContextSupplier, Option.empty());
+ HoodieBaseFile baseFile, TaskContextSupplier
taskContextSupplier) {
+ super(config, instantTime, hoodieTable, Collections.emptyMap(),
partitionPath, fileId, baseFile, taskContextSupplier, Option.empty());
this.recordItr = keyToNewRecords.values().iterator();
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
index 488b637b504a..c1a17e3cbf1d 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleFactory.java
@@ -85,7 +85,7 @@ public class HoodieMergeHandleFactory {
Map<String, HoodieRecord<T>> keyToNewRecords,
String partitionPath,
String fileId,
- HoodieBaseFile dataFileToBeMerged,
+ HoodieBaseFile baseFile,
TaskContextSupplier taskContextSupplier,
Option<BaseKeyGenerator> keyGeneratorOpt) {
@@ -101,7 +101,7 @@ public class HoodieMergeHandleFactory {
return instantiateMergeHandle(
isFallbackEnabled, mergeHandleClasses.getLeft(),
mergeHandleClasses.getRight(), logContext, constructorParamTypes,
- writeConfig, instantTime, table, keyToNewRecords, partitionPath,
fileId, dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
+ writeConfig, instantTime, table, keyToNewRecords, partitionPath,
fileId, baseFile, taskContextSupplier, keyGeneratorOpt);
}
/**
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleWithChangeLog.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleWithChangeLog.java
index 3dd86c8699fb..18fa4f1c4bf6 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleWithChangeLog.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandleWithChangeLog.java
@@ -63,8 +63,8 @@ public class HoodieMergeHandleWithChangeLog<T, I, K, O>
extends HoodieWriteMerge
*/
public HoodieMergeHandleWithChangeLog(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
Map<String, HoodieRecord<T>>
keyToNewRecords, String partitionPath, String fileId,
- HoodieBaseFile dataFileToBeMerged,
TaskContextSupplier taskContextSupplier, Option<BaseKeyGenerator>
keyGeneratorOpt) {
- super(config, instantTime, hoodieTable, keyToNewRecords, partitionPath,
fileId, dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
+ HoodieBaseFile baseFile,
TaskContextSupplier taskContextSupplier, Option<BaseKeyGenerator>
keyGeneratorOpt) {
+ super(config, instantTime, hoodieTable, keyToNewRecords, partitionPath,
fileId, baseFile, taskContextSupplier, keyGeneratorOpt);
this.cdcLogger = createCDCLogWriter(instantTime, config, hoodieTable,
partitionPath, taskContextSupplier);
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
index dabb71ebdd7a..6cdbfcf1ce27 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandle.java
@@ -59,8 +59,8 @@ public class HoodieSortedMergeHandle<T, I, K, O> extends
HoodieWriteMergeHandle<
*/
public HoodieSortedMergeHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
Map<String, HoodieRecord<T>>
keyToNewRecordsOrig, String partitionPath, String fileId,
- HoodieBaseFile dataFileToBeMerged,
TaskContextSupplier taskContextSupplier, Option<BaseKeyGenerator>
keyGeneratorOpt) {
- super(config, instantTime, hoodieTable, keyToNewRecordsOrig,
partitionPath, fileId, dataFileToBeMerged, taskContextSupplier,
keyGeneratorOpt);
+ HoodieBaseFile baseFile, TaskContextSupplier
taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
+ super(config, instantTime, hoodieTable, keyToNewRecordsOrig,
partitionPath, fileId, baseFile, taskContextSupplier, keyGeneratorOpt);
newRecordKeysSorted.addAll(keyToNewRecords.keySet());
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandleWithChangeLog.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandleWithChangeLog.java
index 48a32deceda6..2658989adb17 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandleWithChangeLog.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieSortedMergeHandleWithChangeLog.java
@@ -48,8 +48,8 @@ public class HoodieSortedMergeHandleWithChangeLog<T, I, K, O>
extends HoodieMerg
*/
public HoodieSortedMergeHandleWithChangeLog(HoodieWriteConfig config, String
instantTime, HoodieTable<T, I, K, O> hoodieTable,
Map<String, HoodieRecord<T>>
keyToNewRecords, String partitionPath, String fileId,
- HoodieBaseFile
dataFileToBeMerged, TaskContextSupplier taskContextSupplier,
Option<BaseKeyGenerator> keyGeneratorOpt) {
- super(config, instantTime, hoodieTable, keyToNewRecords, partitionPath,
fileId, dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
+ HoodieBaseFile baseFile,
TaskContextSupplier taskContextSupplier, Option<BaseKeyGenerator>
keyGeneratorOpt) {
+ super(config, instantTime, hoodieTable, keyToNewRecords, partitionPath,
fileId, baseFile, taskContextSupplier, keyGeneratorOpt);
}
protected boolean writeRecord(HoodieRecord<T> newRecord, HoodieRecord
insertRecord, HoodieSchema schema, Properties props)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
index 4a426b2496d2..37718fdbbd08 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteMergeHandle.java
@@ -147,9 +147,9 @@ public class HoodieWriteMergeHandle<T, I, K, O> extends
HoodieAbstractMergeHandl
*/
public HoodieWriteMergeHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
Map<String, HoodieRecord<T>> keyToNewRecords,
String partitionPath, String fileId,
- HoodieBaseFile dataFileToBeMerged,
TaskContextSupplier taskContextSupplier,
+ HoodieBaseFile baseFile, TaskContextSupplier
taskContextSupplier,
Option<BaseKeyGenerator> keyGeneratorOpt) {
- super(config, instantTime, hoodieTable,
MergeContext.create(Collections.emptyIterator()), partitionPath, fileId,
taskContextSupplier, dataFileToBeMerged, keyGeneratorOpt,
+ super(config, instantTime, hoodieTable,
MergeContext.create(Collections.emptyIterator()), partitionPath, fileId,
taskContextSupplier, baseFile, keyGeneratorOpt,
// preserveMetadata is disabled by default for MDT but enabled
otherwise
!HoodieTableMetadata.isMetadataTable(config.getBasePath()));
this.keyToNewRecords = keyToNewRecords;
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieCompactionHandler.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieCompactionHandler.java
index 4d6a216cf231..2c240fc4c602 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieCompactionHandler.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieCompactionHandler.java
@@ -37,7 +37,7 @@ import java.util.Map;
*/
public interface HoodieCompactionHandler<T> {
Iterator<List<WriteStatus>> handleUpdate(String instantTime, String
partitionPath, String fileId,
- Map<String, HoodieRecord<T>>
keyToNewRecords, HoodieBaseFile oldDataFile) throws IOException;
+ Map<String, HoodieRecord<T>>
keyToNewRecords, HoodieBaseFile baseFile) throws IOException;
Iterator<List<WriteStatus>> handleInsert(String instantTime, String
partitionPath, String fileId,
Map<String, HoodieRecord<?>>
recordMap);
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java
index f230f32dc366..ac3c40614a83 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java
@@ -578,10 +578,10 @@ public class CleanPlanner<T, I, K, O> implements
Serializable {
private List<CleanFileInfo> getCleanFileInfoForSlice(FileSlice nextSlice) {
List<CleanFileInfo> cleanPaths = new ArrayList<>();
if (nextSlice.getBaseFile().isPresent()) {
- HoodieBaseFile dataFile = nextSlice.getBaseFile().get();
- cleanPaths.add(new CleanFileInfo(dataFile.getPath(), false));
- if (dataFile.getBootstrapBaseFile().isPresent() &&
config.shouldCleanBootstrapBaseFile()) {
- cleanPaths.add(new
CleanFileInfo(dataFile.getBootstrapBaseFile().get().getPath(), true));
+ HoodieBaseFile baseFile = nextSlice.getBaseFile().get();
+ cleanPaths.add(new CleanFileInfo(baseFile.getPath(), false));
+ if (baseFile.getBootstrapBaseFile().isPresent() &&
config.shouldCleanBootstrapBaseFile()) {
+ cleanPaths.add(new
CleanFileInfo(baseFile.getBootstrapBaseFile().get().getPath(), true));
}
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/plan/generators/BaseHoodieCompactionPlanGenerator.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/plan/generators/BaseHoodieCompactionPlanGenerator.java
index ac3d2d31da7b..9c629d076f87 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/plan/generators/BaseHoodieCompactionPlanGenerator.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/plan/generators/BaseHoodieCompactionPlanGenerator.java
@@ -154,8 +154,8 @@ public abstract class BaseHoodieCompactionPlanGenerator<T
extends HoodieRecordPa
// Avro generated classes are not inheriting Serializable. Using
CompactionOperation POJO
// for Map operations and collecting them finally in Avro generated
classes for storing
// into meta files.
- Option<HoodieBaseFile> dataFile = s.getBaseFile();
- return new CompactionOperation(dataFile, partitionPath, logFiles,
+ Option<HoodieBaseFile> baseFile = s.getBaseFile();
+ return new CompactionOperation(baseFile, partitionPath, logFiles,
writeConfig.getCompactionStrategy().captureMetrics(writeConfig,
s));
}), parallelism).stream()
.map(CompactionUtils::buildHoodieCompactionOperation).collect(toList());
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkCopyOnWriteTable.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkCopyOnWriteTable.java
index 3143a37bd99a..07c8c99536b2 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkCopyOnWriteTable.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/table/HoodieFlinkCopyOnWriteTable.java
@@ -408,16 +408,16 @@ public class HoodieFlinkCopyOnWriteTable<T>
@Override
public Iterator<List<WriteStatus>> handleUpdate(
String instantTime, String partitionPath, String fileId,
- Map<String, HoodieRecord<T>> keyToNewRecords, HoodieBaseFile
oldDataFile) throws IOException {
+ Map<String, HoodieRecord<T>> keyToNewRecords, HoodieBaseFile baseFile)
throws IOException {
// always using avro record merger for legacy compaction since log scanner
do not support rowdata reading yet.
config.setRecordMergerClass(HoodieAvroRecordMerger.class.getName());
// these are updates
- HoodieMergeHandle mergeHandle = getUpdateHandle(instantTime,
partitionPath, fileId, keyToNewRecords, oldDataFile);
+ HoodieMergeHandle mergeHandle = getUpdateHandle(instantTime,
partitionPath, fileId, keyToNewRecords, baseFile);
return MergeUtils.runMerge(mergeHandle, instantTime, fileId);
}
protected HoodieMergeHandle getUpdateHandle(String instantTime, String
partitionPath, String fileId,
- Map<String, HoodieRecord<T>>
keyToNewRecords, HoodieBaseFile dataFileToBeMerged) {
+ Map<String, HoodieRecord<T>>
keyToNewRecords, HoodieBaseFile baseFile) {
Option<BaseKeyGenerator> keyGeneratorOpt = Option.empty();
if (!config.populateMetaFields()) {
try {
@@ -428,7 +428,7 @@ public class HoodieFlinkCopyOnWriteTable<T>
}
}
return HoodieMergeHandleFactory.create(config, instantTime, this,
keyToNewRecords, partitionPath, fileId,
- dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
+ baseFile, taskContextSupplier, keyGeneratorOpt);
}
@Override
diff --git
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaCopyOnWriteTable.java
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaCopyOnWriteTable.java
index 01dc850319ba..7118e645f0f7 100644
---
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaCopyOnWriteTable.java
+++
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/HoodieJavaCopyOnWriteTable.java
@@ -278,15 +278,15 @@ public class HoodieJavaCopyOnWriteTable<T>
@Override
public Iterator<List<WriteStatus>> handleUpdate(
String instantTime, String partitionPath, String fileId,
- Map<String, HoodieRecord<T>> keyToNewRecords, HoodieBaseFile oldDataFile)
+ Map<String, HoodieRecord<T>> keyToNewRecords, HoodieBaseFile baseFile)
throws IOException {
// these are updates
- HoodieMergeHandle mergeHandle = getUpdateHandle(instantTime,
partitionPath, fileId, keyToNewRecords, oldDataFile);
+ HoodieMergeHandle mergeHandle = getUpdateHandle(instantTime,
partitionPath, fileId, keyToNewRecords, baseFile);
return MergeUtils.runMerge(mergeHandle, instantTime, fileId);
}
protected HoodieMergeHandle getUpdateHandle(String instantTime, String
partitionPath, String fileId,
- Map<String, HoodieRecord<T>>
keyToNewRecords, HoodieBaseFile dataFileToBeMerged) {
+ Map<String, HoodieRecord<T>>
keyToNewRecords, HoodieBaseFile baseFile) {
Option<BaseKeyGenerator> keyGeneratorOpt = Option.empty();
if (!config.populateMetaFields()) {
try {
@@ -297,7 +297,7 @@ public class HoodieJavaCopyOnWriteTable<T>
}
}
return HoodieMergeHandleFactory.create(config, instantTime, this,
keyToNewRecords, partitionPath, fileId,
- dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
+ baseFile, taskContextSupplier, keyGeneratorOpt);
}
@Override
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDReadClient.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDReadClient.java
index a568c4855ef9..b4a443edb55c 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDReadClient.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDReadClient.java
@@ -134,9 +134,9 @@ public class SparkRDDReadClient<T> implements Serializable {
private Option<String> convertToDataFilePath(Option<Pair<String, String>>
partitionPathFileIDPair) {
if (partitionPathFileIDPair.isPresent()) {
- HoodieBaseFile dataFile = hoodieTable.getBaseFileOnlyView()
+ HoodieBaseFile baseFile = hoodieTable.getBaseFileOnlyView()
.getLatestBaseFile(partitionPathFileIDPair.get().getLeft(),
partitionPathFileIDPair.get().getRight()).get();
- return Option.of(dataFile.getPath());
+ return Option.of(baseFile.getPath());
} else {
return Option.empty();
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieFileProbingFunction.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieFileProbingFunction.java
index e08f378bd9e4..7b941a493fe7 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieFileProbingFunction.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bloom/HoodieFileProbingFunction.java
@@ -136,16 +136,16 @@ public class HoodieFileProbingFunction implements
// TODO add assertion that file is checked only once
- final HoodieBaseFile dataFile = fileIDBaseFileMap.get(fileId);
+ final HoodieBaseFile baseFile = fileIDBaseFileMap.get(fileId);
Collection<Pair<String, Long>> matchingKeysAndPositions =
HoodieIndexUtils.filterKeysFromFile(
- dataFile.getStoragePath(), candidateRecordKeys,
HoodieStorageUtils.getStorage(dataFile.getStoragePath(), storageConf));
+ baseFile.getStoragePath(), candidateRecordKeys,
HoodieStorageUtils.getStorage(baseFile.getStoragePath(), storageConf));
log.debug(
"Bloom filter candidates ({}) / false positives ({}), actual
matches ({})",
candidateRecordKeys.size(), candidateRecordKeys.size() -
matchingKeysAndPositions.size(),
matchingKeysAndPositions.size());
- return new HoodieKeyLookupResult(fileId, partitionPath,
dataFile.getCommitTime(), matchingKeysAndPositions);
+ return new HoodieKeyLookupResult(fileId, partitionPath,
baseFile.getCommitTime(), matchingKeysAndPositions);
})
.iterator();
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java
index 50a5ddf7936b..79cd4755c29c 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/HoodieSparkCopyOnWriteTable.java
@@ -237,14 +237,14 @@ public class HoodieSparkCopyOnWriteTable<T>
@Override
public Iterator<List<WriteStatus>> handleUpdate(
String instantTime, String partitionPath, String fileId,
- Map<String, HoodieRecord<T>> keyToNewRecords, HoodieBaseFile
oldDataFile) throws IOException {
+ Map<String, HoodieRecord<T>> keyToNewRecords, HoodieBaseFile baseFile)
throws IOException {
// these are updates
- HoodieMergeHandle mergeHandle = getUpdateHandle(instantTime,
partitionPath, fileId, keyToNewRecords, oldDataFile);
+ HoodieMergeHandle mergeHandle = getUpdateHandle(instantTime,
partitionPath, fileId, keyToNewRecords, baseFile);
return MergeUtils.runMerge(mergeHandle, instantTime, fileId);
}
protected HoodieMergeHandle getUpdateHandle(String instantTime, String
partitionPath, String fileId,
- Map<String, HoodieRecord<T>>
keyToNewRecords, HoodieBaseFile dataFileToBeMerged) {
+ Map<String, HoodieRecord<T>>
keyToNewRecords, HoodieBaseFile baseFile) {
Option<BaseKeyGenerator> keyGeneratorOpt = Option.empty();
if (!config.populateMetaFields()) {
try {
@@ -255,7 +255,7 @@ public class HoodieSparkCopyOnWriteTable<T>
}
}
HoodieMergeHandle mergeHandle = HoodieMergeHandleFactory.create(config,
instantTime, this, keyToNewRecords, partitionPath, fileId,
- dataFileToBeMerged, taskContextSupplier, keyGeneratorOpt);
+ baseFile, taskContextSupplier, keyGeneratorOpt);
if (mergeHandle.getOldFilePath() != null &&
mergeHandle.baseFileForMerge().getBootstrapBaseFile().isPresent()) {
Option<String[]> partitionFields =
getMetaClient().getTableConfig().getPartitionFields();
Object[] partitionValues =
SparkPartitionUtils.getPartitionFieldVals(getMetaClient().getTableConfig(),
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestClientRollback.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestClientRollback.java
index 475b418c265d..b0f8bd3d2036 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestClientRollback.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestClientRollback.java
@@ -156,15 +156,15 @@ public class TestClientRollback extends
HoodieClientTestBase {
HoodieSparkTable table = HoodieSparkTable.create(getConfig(), context,
metaClient);
final BaseFileOnlyView view1 = table.getBaseFileOnlyView();
- List<HoodieBaseFile> dataFiles = partitionPaths.stream().flatMap(s -> {
+ List<HoodieBaseFile> baseFiles = partitionPaths.stream().flatMap(s -> {
return view1.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("003"));
}).collect(Collectors.toList());
- assertEquals(3, dataFiles.size(), "The data files for commit 003 should
be present");
+ assertEquals(3, baseFiles.size(), "The data files for commit 003 should
be present");
- dataFiles = partitionPaths.stream().flatMap(s -> {
+ baseFiles = partitionPaths.stream().flatMap(s -> {
return view1.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("002"));
}).collect(Collectors.toList());
- assertEquals(3, dataFiles.size(), "The data files for commit 002 should
be present");
+ assertEquals(3, baseFiles.size(), "The data files for commit 002 should
be present");
/**
* Write 4 (updates)
@@ -180,8 +180,8 @@ public class TestClientRollback extends
HoodieClientTestBase {
table = HoodieSparkTable.create(getConfig(), context, metaClient);
final BaseFileOnlyView view2 = table.getBaseFileOnlyView();
- dataFiles = partitionPaths.stream().flatMap(s ->
view2.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("004"))).collect(Collectors.toList());
- assertEquals(3, dataFiles.size(), "The data files for commit 004 should
be present");
+ baseFiles = partitionPaths.stream().flatMap(s ->
view2.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("004"))).collect(Collectors.toList());
+ assertEquals(3, baseFiles.size(), "The data files for commit 004 should
be present");
// rolling back to a non existent savepoint must not succeed
assertThrows(HoodieRollbackException.class, () -> {
@@ -195,14 +195,14 @@ public class TestClientRollback extends
HoodieClientTestBase {
metaClient = HoodieTableMetaClient.reload(metaClient);
table = HoodieSparkTable.create(getConfig(), context, metaClient);
final BaseFileOnlyView view3 = table.getBaseFileOnlyView();
- dataFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("002"))).collect(Collectors.toList());
- assertEquals(3, dataFiles.size(), "The data files for commit 002 be
available");
+ baseFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("002"))).collect(Collectors.toList());
+ assertEquals(3, baseFiles.size(), "The data files for commit 002 be
available");
- dataFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("003"))).collect(Collectors.toList());
- assertEquals(0, dataFiles.size(), "The data files for commit 003 should
be rolled back");
+ baseFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("003"))).collect(Collectors.toList());
+ assertEquals(0, baseFiles.size(), "The data files for commit 003 should
be rolled back");
- dataFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("004"))).collect(Collectors.toList());
- assertEquals(0, dataFiles.size(), "The data files for commit 004 should
be rolled back");
+ baseFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("004"))).collect(Collectors.toList());
+ assertEquals(0, baseFiles.size(), "The data files for commit 004 should
be rolled back");
if (testFailedRestore) {
//test to make sure that restore commit is reused when the restore
fails and is re-ran
@@ -339,15 +339,15 @@ public class TestClientRollback extends
HoodieClientTestBase {
HoodieSparkTable table = HoodieSparkTable.create(getConfig(), context,
metaClient);
final BaseFileOnlyView view1 = table.getBaseFileOnlyView();
- List<HoodieBaseFile> dataFiles = partitionPaths.stream().flatMap(s -> {
+ List<HoodieBaseFile> baseFiles = partitionPaths.stream().flatMap(s -> {
return view1.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("003"));
}).collect(Collectors.toList());
- assertEquals(3, dataFiles.size(), "The data files for commit 003 should
be present");
+ assertEquals(3, baseFiles.size(), "The data files for commit 003 should
be present");
- dataFiles = partitionPaths.stream().flatMap(s -> {
+ baseFiles = partitionPaths.stream().flatMap(s -> {
return view1.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("002"));
}).collect(Collectors.toList());
- assertEquals(3, dataFiles.size(), "The data files for commit 002 should
be present");
+ assertEquals(3, baseFiles.size(), "The data files for commit 002 should
be present");
/**
* Write 4 (updates)
@@ -364,8 +364,8 @@ public class TestClientRollback extends
HoodieClientTestBase {
table = HoodieSparkTable.create(getConfig(), context, metaClient);
final BaseFileOnlyView view2 = table.getBaseFileOnlyView();
- dataFiles = partitionPaths.stream().flatMap(s ->
view2.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("004"))).collect(Collectors.toList());
- assertEquals(3, dataFiles.size(), "The data files for commit 004 should
be present");
+ baseFiles = partitionPaths.stream().flatMap(s ->
view2.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("004"))).collect(Collectors.toList());
+ assertEquals(3, baseFiles.size(), "The data files for commit 004 should
be present");
// rollback to savepoint 002
HoodieInstant savepoint =
table.getCompletedSavepointTimeline().getInstantsAsStream().findFirst().get();
@@ -374,14 +374,14 @@ public class TestClientRollback extends
HoodieClientTestBase {
metaClient = HoodieTableMetaClient.reload(metaClient);
table = HoodieSparkTable.create(getConfig(), context, metaClient);
final BaseFileOnlyView view3 = table.getBaseFileOnlyView();
- dataFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("002"))).collect(Collectors.toList());
- assertEquals(3, dataFiles.size(), "The data files for commit 002 be
available");
+ baseFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("002"))).collect(Collectors.toList());
+ assertEquals(3, baseFiles.size(), "The data files for commit 002 be
available");
- dataFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("003"))).collect(Collectors.toList());
- assertEquals(0, dataFiles.size(), "The data files for commit 003 should
be rolled back");
+ baseFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("003"))).collect(Collectors.toList());
+ assertEquals(0, baseFiles.size(), "The data files for commit 003 should
be rolled back");
- dataFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("004"))).collect(Collectors.toList());
- assertEquals(0, dataFiles.size(), "The data files for commit 004 should
be rolled back");
+ baseFiles = partitionPaths.stream().flatMap(s ->
view3.getAllBaseFiles(s).filter(f ->
f.getCommitTime().equals("004"))).collect(Collectors.toList());
+ assertEquals(0, baseFiles.size(), "The data files for commit 004 should
be rolled back");
}
}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByVersions.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByVersions.java
index aa26921c174f..71dea86d3a69 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByVersions.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/clean/TestCleanerInsertAndCleanByVersions.java
@@ -213,25 +213,25 @@ public class TestCleanerInsertAndCleanByVersions extends
SparkClientFunctionalTe
for (HoodieFileGroup fileGroup : fileGroups) {
if
(compactionFileIdToLatestFileSlice.containsKey(fileGroup.getFileGroupId())) {
// Ensure latest file-slice selected for compaction is retained
- Option<HoodieBaseFile> dataFileForCompactionPresent =
-
Option.fromJavaOptional(fileGroup.getAllBaseFiles().filter(df -> {
+ Option<HoodieBaseFile> baseFileForCompactionPresent =
+
Option.fromJavaOptional(fileGroup.getAllBaseFiles().filter(baseFile -> {
return
compactionFileIdToLatestFileSlice.get(fileGroup.getFileGroupId()).getBaseInstantTime()
- .equals(df.getCommitTime());
+ .equals(baseFile.getCommitTime());
}).findAny());
- assertTrue(dataFileForCompactionPresent.isPresent(),
+ assertTrue(baseFileForCompactionPresent.isPresent(),
"Data File selected for compaction is retained");
} else {
// file has no more than max versions
String fileId = fileGroup.getFileGroupId().getFileId();
- List<HoodieBaseFile> dataFiles =
fileGroup.getAllBaseFiles().collect(Collectors.toList());
+ List<HoodieBaseFile> baseFiles =
fileGroup.getAllBaseFiles().collect(Collectors.toList());
- assertTrue(dataFiles.size() <= maxVersions,
+ assertTrue(baseFiles.size() <= maxVersions,
"fileId " + fileId + " has more than " + maxVersions + "
versions");
// Each file, has the latest N versions (i.e cleaning gets rid
of older versions)
List<String> commitedVersions = new
ArrayList<>(fileIdToVersions.get(fileId));
- for (int i = 0; i < dataFiles.size(); i++) {
- assertEquals((dataFiles.get(i)).getCommitTime(),
+ for (int i = 0; i < baseFiles.size(); i++) {
+ assertEquals((baseFiles.get(i)).getCommitTime(),
commitedVersions.get(commitedVersions.size() - 1 - i),
"File " + fileId + " does not have latest versions on
commits" + commitedVersions);
}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/CompactionTestBase.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/CompactionTestBase.java
index 677ece46d801..ffcce707595c 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/CompactionTestBase.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/CompactionTestBase.java
@@ -140,8 +140,8 @@ public class CompactionTestBase extends
HoodieClientTestBase {
assertNoWriteErrors(statusList);
metaClient = createMetaClient(cfg.getBasePath());
HoodieTable hoodieTable = getHoodieTable(metaClient, cfg);
- List<HoodieBaseFile> dataFilesToRead =
getCurrentLatestBaseFiles(hoodieTable);
- assertTrue(dataFilesToRead.stream().findAny().isPresent(),
+ List<HoodieBaseFile> baseFilesToRead =
getCurrentLatestBaseFiles(hoodieTable);
+ assertTrue(baseFilesToRead.stream().findAny().isPresent(),
"should list the base files we wrote in the delta commit");
validateDeltaCommit(firstInstant, fgIdToCompactionOperation, cfg);
}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/strategy/TestHoodieCompactionStrategy.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/strategy/TestHoodieCompactionStrategy.java
index 73bb9e08b5ee..d35b8df2bc2f 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/strategy/TestHoodieCompactionStrategy.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/strategy/TestHoodieCompactionStrategy.java
@@ -517,17 +517,17 @@ public class TestHoodieCompactionStrategy {
List<HoodieCompactionOperation> operations = new
ArrayList<>(sizesMap.size());
sizesMap.forEach((k, v) -> {
- HoodieBaseFile df = TestHoodieBaseFile.newDataFile(k);
+ HoodieBaseFile baseFile = TestHoodieBaseFile.newDataFile(k);
String partitionPath = keyToPartitionMap.get(k);
List<HoodieLogFile> logFiles =
v.stream().map(TestHoodieLogFile::newLogFile).collect(Collectors.toList());
- FileSlice slice = new FileSlice(new HoodieFileGroupId(partitionPath,
df.getFileId()), df.getCommitTime());
- slice.setBaseFile(df);
+ FileSlice slice = new FileSlice(new HoodieFileGroupId(partitionPath,
baseFile.getFileId()), baseFile.getCommitTime());
+ slice.setBaseFile(baseFile);
logFiles.stream().forEach(f -> slice.addLogFile(f));
- operations.add(new HoodieCompactionOperation(df.getCommitTime(),
- logFiles.stream().map(s ->
s.getPath().toString()).collect(Collectors.toList()), df.getPath(),
df.getFileId(),
+ operations.add(new HoodieCompactionOperation(baseFile.getCommitTime(),
+ logFiles.stream().map(s ->
s.getPath().toString()).collect(Collectors.toList()), baseFile.getPath(),
baseFile.getFileId(),
partitionPath,
config.getCompactionStrategy().captureMetrics(config, slice),
- df.getBootstrapBaseFile().map(BaseFile::getPath).orElse(null))
+ baseFile.getBootstrapBaseFile().map(BaseFile::getPath).orElse(null))
);
});
return operations;
@@ -538,20 +538,20 @@ public class TestHoodieCompactionStrategy {
List<HoodieCompactionOperation> operations = new
ArrayList<>(sizesMap.size());
sizesMap.forEach((k, v) -> {
- HoodieBaseFile df = TestHoodieBaseFile.newDataFile(k);
+ HoodieBaseFile baseFile = TestHoodieBaseFile.newDataFile(k);
String partitionPath = keyToPartitionMap.get(k);
// create operation for target partition
if (filterPartitions.contains(partitionPath)) {
List<HoodieLogFile> logFiles =
v.stream().map(TestHoodieLogFile::newLogFile).collect(Collectors.toList());
- FileSlice slice = new FileSlice(new HoodieFileGroupId(partitionPath,
df.getFileId()), df.getCommitTime());
- slice.setBaseFile(df);
+ FileSlice slice = new FileSlice(new HoodieFileGroupId(partitionPath,
baseFile.getFileId()), baseFile.getCommitTime());
+ slice.setBaseFile(baseFile);
logFiles.stream().forEach(f -> slice.addLogFile(f));
- operations.add(new HoodieCompactionOperation(df.getCommitTime(),
+ operations.add(new HoodieCompactionOperation(baseFile.getCommitTime(),
logFiles.stream().map(s ->
s.getPath().toString()).collect(Collectors.toList()),
- df.getPath(), df.getFileId(),
+ baseFile.getPath(), baseFile.getFileId(),
partitionPath,
config.getCompactionStrategy().captureMetrics(config, slice),
- df.getBootstrapBaseFile().map(BaseFile::getPath).orElse(null))
+
baseFile.getBootstrapBaseFile().map(BaseFile::getPath).orElse(null))
);
}
});
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/SparkClientFunctionalTestHarness.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/SparkClientFunctionalTestHarness.java
index 1133168ad374..920fe9c5e812 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/SparkClientFunctionalTestHarness.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/SparkClientFunctionalTestHarness.java
@@ -311,12 +311,12 @@ public class SparkClientFunctionalTestHarness implements
SparkProvider, HoodieMe
TableFileSystemView.BaseFileOnlyView roView =
getHoodieTableFileSystemView(reloadedMetaClient,
reloadedMetaClient.getCommitTimeline().filterCompletedInstants(),
allFiles);
- Stream<HoodieBaseFile> dataFilesToRead = roView.getLatestBaseFiles();
- assertTrue(!dataFilesToRead.findAny().isPresent());
+ Stream<HoodieBaseFile> baseFilesToRead = roView.getLatestBaseFiles();
+ assertTrue(!baseFilesToRead.findAny().isPresent());
roView = getHoodieTableFileSystemView(reloadedMetaClient,
hoodieTable.getCompletedCommitsTimeline(), allFiles);
- dataFilesToRead = roView.getLatestBaseFiles();
- return dataFilesToRead;
+ baseFilesToRead = roView.getLatestBaseFiles();
+ return baseFilesToRead;
}
protected void updateRecordsInMORTable(HoodieTableMetaClient metaClient,
List<HoodieRecord> records, SparkRDDWriteClient client, HoodieWriteConfig cfg,
String commitTime) throws IOException {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/model/CompactionOperation.java
b/hudi-common/src/main/java/org/apache/hudi/common/model/CompactionOperation.java
index 87a991bcde2f..2b86ba20a238 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/model/CompactionOperation.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/model/CompactionOperation.java
@@ -66,15 +66,15 @@ public class CompactionOperation implements Serializable {
this.metrics = metrics;
}
- public CompactionOperation(Option<HoodieBaseFile> dataFile, String
partitionPath, List<HoodieLogFile> logFiles,
+ public CompactionOperation(Option<HoodieBaseFile> baseFile, String
partitionPath, List<HoodieLogFile> logFiles,
Map<String, Double> metrics) {
ValidationUtils.checkArgument(!logFiles.isEmpty(), "log files should not
be empty.");
- if (dataFile.isPresent()) {
- this.baseInstantTime = dataFile.get().getCommitTime();
- this.dataFileName = Option.of(dataFile.get().getFileName());
- this.fileGroupId = new HoodieFileGroupId(partitionPath,
dataFile.get().getFileId());
- this.dataFileCommitTime = Option.of(dataFile.get().getCommitTime());
- this.bootstrapFilePath =
dataFile.get().getBootstrapBaseFile().map(BaseFile::getPath);
+ if (baseFile.isPresent()) {
+ this.baseInstantTime = baseFile.get().getCommitTime();
+ this.dataFileName = Option.of(baseFile.get().getFileName());
+ this.fileGroupId = new HoodieFileGroupId(partitionPath,
baseFile.get().getFileId());
+ this.dataFileCommitTime = Option.of(baseFile.get().getCommitTime());
+ this.bootstrapFilePath =
baseFile.get().getBootstrapBaseFile().map(BaseFile::getPath);
} else {
this.dataFileName = Option.empty();
this.baseInstantTime = logFiles.get(0).getDeltaCommitTime();
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieBaseFile.java
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieBaseFile.java
index 2ee0385ca5e6..61825e80faa7 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieBaseFile.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieBaseFile.java
@@ -41,11 +41,11 @@ public class HoodieBaseFile extends BaseFile {
private Option<BaseFile> bootstrapBaseFile;
- public HoodieBaseFile(HoodieBaseFile dataFile) {
- super(dataFile);
- this.bootstrapBaseFile = dataFile.bootstrapBaseFile;
- this.fileId = dataFile.getFileId();
- this.commitTime = dataFile.getCommitTime();
+ public HoodieBaseFile(HoodieBaseFile baseFile) {
+ super(baseFile);
+ this.bootstrapBaseFile = baseFile.bootstrapBaseFile;
+ this.fileId = baseFile.getFileId();
+ this.commitTime = baseFile.getCommitTime();
}
public HoodieBaseFile(StoragePathInfo pathInfo) {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileGroup.java
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileGroup.java
index 5b31db849504..f86d461c195d 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileGroup.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/model/HoodieFileGroup.java
@@ -106,11 +106,11 @@ public class HoodieFileGroup implements Serializable {
/**
* Add a new datafile into the file group.
*/
- public void addBaseFile(HoodieBaseFile dataFile) {
- if (!fileSlices.containsKey(dataFile.getCommitTime())) {
- fileSlices.put(dataFile.getCommitTime(), new FileSlice(fileGroupId,
dataFile.getCommitTime()));
+ public void addBaseFile(HoodieBaseFile baseFile) {
+ if (!fileSlices.containsKey(baseFile.getCommitTime())) {
+ fileSlices.put(baseFile.getCommitTime(), new FileSlice(fileGroupId,
baseFile.getCommitTime()));
}
- fileSlices.get(dataFile.getCommitTime()).setBaseFile(dataFile);
+ fileSlices.get(baseFile.getCommitTime()).setBaseFile(baseFile);
}
/**
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
index f768c6880def..7c155ff4e635 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
@@ -661,9 +661,9 @@ public abstract class AbstractTableFileSystemView
implements SyncableFileSystemV
protected FileSlice addBootstrapBaseFileIfPresent(FileSlice fileSlice,
Function<HoodieFileGroupId, Option<BootstrapBaseFileMapping>>
bootstrapBaseFileMappingFunc) {
if (fileSlice.getBaseInstantTime().equals(METADATA_BOOTSTRAP_INSTANT_TS)) {
FileSlice copy = new FileSlice(fileSlice);
- copy.getBaseFile().ifPresent(dataFile -> {
+ copy.getBaseFile().ifPresent(baseFile -> {
Option<BootstrapBaseFileMapping> edf =
getBootstrapBaseFile(copy.getFileGroupId());
- bootstrapBaseFileMappingFunc.apply(copy.getFileGroupId()).ifPresent(e
-> dataFile.setBootstrapBaseFile(e.getBootstrapBaseFile()));
+ bootstrapBaseFileMappingFunc.apply(copy.getFileGroupId()).ifPresent(e
-> baseFile.setBootstrapBaseFile(e.getBootstrapBaseFile()));
});
return copy;
}
@@ -740,8 +740,8 @@ public abstract class AbstractTableFileSystemView
implements SyncableFileSystemV
String partitionPath = formatPartitionKey(partitionStr);
ensurePartitionLoadedCorrectly(partitionPath);
return fetchLatestBaseFiles(partitionPath)
- .filter(df -> !isFileGroupReplaced(partitionPath, df.getFileId()))
- .map(df -> addBootstrapBaseFileIfPresent(new
HoodieFileGroupId(partitionPath, df.getFileId()), df));
+ .filter(baseFile -> !isFileGroupReplaced(partitionPath,
baseFile.getFileId()))
+ .map(baseFile -> addBootstrapBaseFileIfPresent(new
HoodieFileGroupId(partitionPath, baseFile.getFileId()), baseFile));
} finally {
readLock.unlock();
}
@@ -789,9 +789,9 @@ public abstract class AbstractTableFileSystemView
implements SyncableFileSystemV
.map(fileGroup -> Option.fromJavaOptional(fileGroup.getAllBaseFiles()
.filter(baseFile -> compareTimestamps(baseFile.getCommitTime(),
LESSER_THAN_OR_EQUALS, maxCommitTime
))
- .filter(df -> !isBaseFileDueToPendingCompaction(partitionPath, df)
&& !isBaseFileDueToPendingClustering(df)).findFirst()))
+ .filter(baseFile ->
!isBaseFileDueToPendingCompaction(partitionPath, baseFile) &&
!isBaseFileDueToPendingClustering(baseFile)).findFirst()))
.filter(Option::isPresent).map(Option::get)
- .map(df -> addBootstrapBaseFileIfPresent(new
HoodieFileGroupId(partitionPath, df.getFileId()), df));
+ .map(baseFile -> addBootstrapBaseFileIfPresent(new
HoodieFileGroupId(partitionPath, baseFile.getFileId()), baseFile));
}
@Override
@@ -805,8 +805,8 @@ public abstract class AbstractTableFileSystemView
implements SyncableFileSystemV
} else {
return fetchHoodieFileGroup(partitionPath, fileId).map(fileGroup ->
fileGroup.getAllBaseFiles()
.filter(baseFile ->
compareTimestamps(baseFile.getCommitTime(), EQUALS,
- instantTime)).filter(df ->
!isBaseFileDueToPendingCompaction(partitionPath, df) &&
!isBaseFileDueToPendingClustering(df)).findFirst().orElse(null))
- .map(df -> addBootstrapBaseFileIfPresent(new
HoodieFileGroupId(partitionPath, fileId), df));
+ instantTime)).filter(baseFile ->
!isBaseFileDueToPendingCompaction(partitionPath, baseFile) &&
!isBaseFileDueToPendingClustering(baseFile)).findFirst().orElse(null))
+ .map(baseFile -> addBootstrapBaseFileIfPresent(new
HoodieFileGroupId(partitionPath, fileId), baseFile));
}
} finally {
readLock.unlock();
@@ -826,7 +826,7 @@ public abstract class AbstractTableFileSystemView
implements SyncableFileSystemV
return Option.empty();
} else {
return fetchLatestBaseFile(partitionPath, fileId)
- .map(df -> addBootstrapBaseFileIfPresent(new
HoodieFileGroupId(partitionPath, fileId), df));
+ .map(baseFile -> addBootstrapBaseFileIfPresent(new
HoodieFileGroupId(partitionPath, fileId), baseFile));
}
} finally {
readLock.unlock();
@@ -875,10 +875,10 @@ public abstract class AbstractTableFileSystemView
implements SyncableFileSystemV
String partitionPath = formatPartitionKey(partitionStr);
ensurePartitionLoadedCorrectly(partitionPath);
return fetchAllBaseFiles(partitionPath)
- .filter(df -> !isFileGroupReplaced(partitionPath, df.getFileId()))
- .filter(df ->
visibleCommitsAndCompactionTimeline.containsOrBeforeTimelineStarts(df.getCommitTime()))
- .filter(df -> !isBaseFileDueToPendingCompaction(partitionPath, df)
&& !isBaseFileDueToPendingClustering(df))
- .map(df -> addBootstrapBaseFileIfPresent(new
HoodieFileGroupId(partitionPath, df.getFileId()), df));
+ .filter(baseFile -> !isFileGroupReplaced(partitionPath,
baseFile.getFileId()))
+ .filter(baseFile ->
visibleCommitsAndCompactionTimeline.containsOrBeforeTimelineStarts(baseFile.getCommitTime()))
+ .filter(baseFile -> !isBaseFileDueToPendingCompaction(partitionPath,
baseFile) && !isBaseFileDueToPendingClustering(baseFile))
+ .map(baseFile -> addBootstrapBaseFileIfPresent(new
HoodieFileGroupId(partitionPath, baseFile.getFileId()), baseFile));
} finally {
readLock.unlock();
}
@@ -1525,7 +1525,10 @@ public abstract class AbstractTableFileSystemView
implements SyncableFileSystemV
protected Option<HoodieBaseFile> getLatestBaseFile(HoodieFileGroup
fileGroup) {
return Option
- .fromJavaOptional(fileGroup.getAllBaseFiles().filter(df ->
!isBaseFileDueToPendingCompaction(fileGroup.getPartitionPath(), df) &&
!isBaseFileDueToPendingClustering(df)).findFirst());
+ .fromJavaOptional(fileGroup.getAllBaseFiles()
+ .filter(baseFile ->
!isBaseFileDueToPendingCompaction(fileGroup.getPartitionPath(), baseFile)
+ && !isBaseFileDueToPendingClustering(baseFile))
+ .findFirst());
}
/**
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java
index 997fbdea224b..af44b9fe533c 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java
@@ -417,12 +417,12 @@ public abstract class
IncrementalTimelineSyncFileSystemView extends AbstractTabl
*/
Map<String, HoodieBaseFile> viewDataFiles =
fileGroups.stream().flatMap(HoodieFileGroup::getAllRawFileSlices)
.map(FileSlice::getBaseFile).filter(Option::isPresent).map(Option::get)
- .map(df -> Pair.of(FSUtils.getPathWithoutSchemeAndAuthority(new
StoragePath(df.getPath())).toString(), df))
+ .map(baseFile -> Pair.of(FSUtils.getPathWithoutSchemeAndAuthority(new
StoragePath(baseFile.getPath())).toString(), baseFile))
.collect(Collectors.toMap(Pair::getKey, Pair::getValue));
// Note: Delta Log Files and Data Files can be empty when adding/removing
pending compactions
Map<String, HoodieBaseFile> deltaDataFiles =
deltaFileGroups.stream().flatMap(HoodieFileGroup::getAllRawFileSlices)
.map(FileSlice::getBaseFile).filter(Option::isPresent).map(Option::get)
- .map(df -> Pair.of(FSUtils.getPathWithoutSchemeAndAuthority(new
StoragePath(df.getPath())).toString(), df))
+ .map(baseFile -> Pair.of(FSUtils.getPathWithoutSchemeAndAuthority(new
StoragePath(baseFile.getPath())).toString(), baseFile))
.collect(Collectors.toMap(Pair::getKey, Pair::getValue));
Map<String, HoodieLogFile> viewLogFiles =
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java
index c8df4063abf0..67f224b4eb5b 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/view/RocksDbBasedFileSystemView.java
@@ -294,8 +294,8 @@ public class RocksDbBasedFileSystemView extends
IncrementalTimelineSyncFileSyste
rocksDB.writeBatch(batch ->
fg.getAllFileSlicesIncludingInflight().forEach(fs -> {
rocksDB.putInBatch(batch, schemaHelper.getColFamilyForView(),
schemaHelper.getKeyForSliceView(fg, fs), fs);
- fs.getBaseFile().ifPresent(df ->
- rocksDB.putInBatch(batch,
schemaHelper.getColFamilyForView(), schemaHelper.getKeyForDataFileView(fg, fs),
df)
+ fs.getBaseFile().ifPresent(baseFile ->
+ rocksDB.putInBatch(batch,
schemaHelper.getColFamilyForView(), schemaHelper.getKeyForDataFileView(fg, fs),
baseFile)
);
})
)
@@ -368,8 +368,8 @@ public class RocksDbBasedFileSystemView extends
IncrementalTimelineSyncFileSyste
}
}).filter(Objects::nonNull).forEach(fs -> {
rocksDB.putInBatch(batch, schemaHelper.getColFamilyForView(),
schemaHelper.getKeyForSliceView(fg, fs), fs);
- fs.getBaseFile().ifPresent(df ->
- rocksDB.putInBatch(batch,
schemaHelper.getColFamilyForView(), schemaHelper.getKeyForDataFileView(fg, fs),
df)
+ fs.getBaseFile().ifPresent(baseFile ->
+ rocksDB.putInBatch(batch,
schemaHelper.getColFamilyForView(), schemaHelper.getKeyForDataFileView(fg, fs),
baseFile)
);
})
)
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
index f1f3437afa13..a00f149e9495 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
@@ -267,8 +267,8 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
refreshFsView();
- List<HoodieBaseFile> dataFiles =
roView.getLatestBaseFiles().collect(Collectors.toList());
- assertTrue(dataFiles.isEmpty(), "No data file expected");
+ List<HoodieBaseFile> baseFiles =
roView.getLatestBaseFiles().collect(Collectors.toList());
+ assertTrue(baseFiles.isEmpty(), "No data file expected");
List<FileSlice> fileSliceList =
rtView.getLatestFileSlices(partitionPath).collect(Collectors.toList());
assertEquals(1, fileSliceList.size());
FileSlice fileSlice = fileSliceList.get(0);
@@ -577,8 +577,8 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
refreshFsView();
- List<HoodieBaseFile> dataFiles =
roView.getLatestBaseFiles().collect(Collectors.toList());
- assertTrue(dataFiles.isEmpty(), "No data file expected");
+ List<HoodieBaseFile> baseFiles =
roView.getLatestBaseFiles().collect(Collectors.toList());
+ assertTrue(baseFiles.isEmpty(), "No data file expected");
List<FileSlice> fileSliceList =
rtView.getLatestFileSlices(partitionPath).collect(Collectors.toList());
assertEquals(1, fileSliceList.size());
FileSlice fileSlice = fileSliceList.get(0);
@@ -1048,12 +1048,12 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
refreshFsView(preTableVersion8);
- List<HoodieBaseFile> dataFiles =
roView.getAllBaseFiles(partitionPath).collect(Collectors.toList());
+ List<HoodieBaseFile> baseFiles =
roView.getAllBaseFiles(partitionPath).collect(Collectors.toList());
if (skipCreatingDataFile) {
- assertTrue(dataFiles.isEmpty(), "No data file expected");
+ assertTrue(baseFiles.isEmpty(), "No data file expected");
} else {
- assertEquals(1, dataFiles.size(), "One data-file is expected as there is
only one file-group");
- assertEquals(dataFileName, dataFiles.get(0).getFileName(), "Expect only
valid data-file");
+ assertEquals(1, baseFiles.size(), "One data-file is expected as there is
only one file-group");
+ assertEquals(dataFileName, baseFiles.get(0).getFileName(), "Expect only
valid data-file");
}
// Merge API Tests
@@ -1091,40 +1091,40 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
assertEquals(fileName3, logFiles.get(1).getFileName(), "Log File Order
check");
// Data Files API tests
- dataFiles = roView.getLatestBaseFiles().collect(Collectors.toList());
+ baseFiles = roView.getLatestBaseFiles().collect(Collectors.toList());
if (skipCreatingDataFile) {
- assertEquals(0, dataFiles.size(), "Expect no data file to be returned");
+ assertEquals(0, baseFiles.size(), "Expect no data file to be returned");
} else {
- assertEquals(1, dataFiles.size(), "Expect only one data-file to be
sent");
- dataFiles.forEach(df -> assertEquals(df.getCommitTime(), instantTime1,
"Expect data-file for instant 1 be returned"));
- checkExternalFile(srcFileStatus,
dataFiles.get(0).getBootstrapBaseFile(), testBootstrap);
+ assertEquals(1, baseFiles.size(), "Expect only one data-file to be
sent");
+ baseFiles.forEach(baseFile -> assertEquals(baseFile.getCommitTime(),
instantTime1, "Expect data-file for instant 1 be returned"));
+ checkExternalFile(srcFileStatus,
baseFiles.get(0).getBootstrapBaseFile(), testBootstrap);
}
- dataFiles =
roView.getLatestBaseFiles(partitionPath).collect(Collectors.toList());
+ baseFiles =
roView.getLatestBaseFiles(partitionPath).collect(Collectors.toList());
if (skipCreatingDataFile) {
- assertEquals(0, dataFiles.size(), "Expect no data file to be returned");
+ assertEquals(0, baseFiles.size(), "Expect no data file to be returned");
} else {
- assertEquals(1, dataFiles.size(), "Expect only one data-file to be
sent");
- dataFiles.forEach(df -> assertEquals(df.getCommitTime(), instantTime1,
"Expect data-file for instant 1 be returned"));
- checkExternalFile(srcFileStatus,
dataFiles.get(0).getBootstrapBaseFile(), testBootstrap);
+ assertEquals(1, baseFiles.size(), "Expect only one data-file to be
sent");
+ baseFiles.forEach(baseFile -> assertEquals(baseFile.getCommitTime(),
instantTime1, "Expect data-file for instant 1 be returned"));
+ checkExternalFile(srcFileStatus,
baseFiles.get(0).getBootstrapBaseFile(), testBootstrap);
}
- dataFiles = roView.getLatestBaseFilesBeforeOrOn(partitionPath,
deltaInstantTime5).collect(Collectors.toList());
+ baseFiles = roView.getLatestBaseFilesBeforeOrOn(partitionPath,
deltaInstantTime5).collect(Collectors.toList());
if (skipCreatingDataFile) {
- assertEquals(0, dataFiles.size(), "Expect no data file to be returned");
+ assertEquals(0, baseFiles.size(), "Expect no data file to be returned");
} else {
- assertEquals(1, dataFiles.size(), "Expect only one data-file to be
sent");
- dataFiles.forEach(df -> assertEquals(df.getCommitTime(), instantTime1,
"Expect data-file for instant 1 be returned"));
- checkExternalFile(srcFileStatus,
dataFiles.get(0).getBootstrapBaseFile(), testBootstrap);
+ assertEquals(1, baseFiles.size(), "Expect only one data-file to be
sent");
+ baseFiles.forEach(baseFile -> assertEquals(baseFile.getCommitTime(),
instantTime1, "Expect data-file for instant 1 be returned"));
+ checkExternalFile(srcFileStatus,
baseFiles.get(0).getBootstrapBaseFile(), testBootstrap);
}
- dataFiles =
roView.getLatestBaseFilesInRange(allInstantTimes).collect(Collectors.toList());
+ baseFiles =
roView.getLatestBaseFilesInRange(allInstantTimes).collect(Collectors.toList());
if (skipCreatingDataFile) {
- assertEquals(0, dataFiles.size(), "Expect no data file to be returned");
+ assertEquals(0, baseFiles.size(), "Expect no data file to be returned");
} else {
- assertEquals(1, dataFiles.size(), "Expect only one data-file to be
sent");
- dataFiles.forEach(df -> assertEquals(df.getCommitTime(), instantTime1,
"Expect data-file for instant 1 be returned"));
- checkExternalFile(srcFileStatus,
dataFiles.get(0).getBootstrapBaseFile(), testBootstrap);
+ assertEquals(1, baseFiles.size(), "Expect only one data-file to be
sent");
+ baseFiles.forEach(baseFile -> assertEquals(baseFile.getCommitTime(),
instantTime1, "Expect data-file for instant 1 be returned"));
+ checkExternalFile(srcFileStatus,
baseFiles.get(0).getBootstrapBaseFile(), testBootstrap);
}
// Inflight/Orphan File-groups needs to be in the view
@@ -1162,7 +1162,7 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
List<FileSlice> allRawFileSlices =
getAllRawFileSlices(partitionPath).collect(Collectors.toList());
- dataFiles = allRawFileSlices.stream().flatMap(slice -> {
+ baseFiles = allRawFileSlices.stream().flatMap(slice -> {
if (slice.getBaseFile().isPresent()) {
return Stream.of(slice.getBaseFile().get());
}
@@ -1171,10 +1171,10 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
if (includeInvalidAndInflight) {
assertEquals(2 + (isCompactionInFlight ? 1 : 0) + (skipCreatingDataFile
? 0 : 1),
- dataFiles.size(),
+ baseFiles.size(),
"Inflight/Orphan data-file is also expected");
Set<String> fileNames =
-
dataFiles.stream().map(HoodieBaseFile::getFileName).collect(Collectors.toSet());
+
baseFiles.stream().map(HoodieBaseFile::getFileName).collect(Collectors.toSet());
assertTrue(fileNames.contains(orphanDataFileName), "Expect orphan
data-file to be present");
assertTrue(fileNames.contains(inflightDataFileName),
"Expect inflight data-file to be present");
@@ -1243,12 +1243,12 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
fileSliceList =
rtView.getLatestFileSlices(partitionPath).collect(Collectors.toList());
log.info("FILESLICE LIST={}", fileSliceList);
- dataFiles =
fileSliceList.stream().map(FileSlice::getBaseFile).filter(Option::isPresent).map(Option::get)
+ baseFiles =
fileSliceList.stream().map(FileSlice::getBaseFile).filter(Option::isPresent).map(Option::get)
.collect(Collectors.toList());
- assertEquals(1, dataFiles.size(), "Expect only one data-files in latest
view as there is only one file-group");
- assertEquals(compactDataFileName, dataFiles.get(0).getFileName(), "Data
Filename must match");
+ assertEquals(1, baseFiles.size(), "Expect only one data-files in latest
view as there is only one file-group");
+ assertEquals(compactDataFileName, baseFiles.get(0).getFileName(), "Data
Filename must match");
assertEquals(1, fileSliceList.size(), "Only one latest file-slice in the
partition");
- assertFalse(dataFiles.get(0).getBootstrapBaseFile().isPresent(), "No
external data file must be present");
+ assertFalse(baseFiles.get(0).getBootstrapBaseFile().isPresent(), "No
external data file must be present");
fileSlice = fileSliceList.get(0);
assertEquals(fileId, fileSlice.getFileId(), "Check file-Id is set
correctly");
assertEquals(compactDataFileName,
fileSlice.getBaseFile().get().getFileName(),
@@ -1261,31 +1261,31 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
assertEquals(fileName3, logFiles.get(1).getFileName(), "Log File Order
check");
// Data Files API tests
- dataFiles = roView.getLatestBaseFiles().collect(Collectors.toList());
- assertEquals(1, dataFiles.size(), "Expect only one data-file to be sent");
- assertFalse(dataFiles.get(0).getBootstrapBaseFile().isPresent(), "No
external data file must be present");
+ baseFiles = roView.getLatestBaseFiles().collect(Collectors.toList());
+ assertEquals(1, baseFiles.size(), "Expect only one data-file to be sent");
+ assertFalse(baseFiles.get(0).getBootstrapBaseFile().isPresent(), "No
external data file must be present");
- dataFiles.forEach(df -> {
- assertEquals(df.getCommitTime(), compactionRequestedTime, "Expect
data-file created by compaction be returned");
- assertFalse(df.getBootstrapBaseFile().isPresent(), "No external data
file must be present");
+ baseFiles.forEach(baseFile -> {
+ assertEquals(baseFile.getCommitTime(), compactionRequestedTime, "Expect
data-file created by compaction be returned");
+ assertFalse(baseFile.getBootstrapBaseFile().isPresent(), "No external
data file must be present");
});
- dataFiles =
roView.getLatestBaseFiles(partitionPath).collect(Collectors.toList());
- assertEquals(1, dataFiles.size(), "Expect only one data-file to be sent");
- dataFiles.forEach(df -> {
- assertEquals(df.getCommitTime(), compactionRequestedTime, "Expect
data-file created by compaction be returned");
- assertFalse(df.getBootstrapBaseFile().isPresent(), "No external data
file must be present");
+ baseFiles =
roView.getLatestBaseFiles(partitionPath).collect(Collectors.toList());
+ assertEquals(1, baseFiles.size(), "Expect only one data-file to be sent");
+ baseFiles.forEach(baseFile -> {
+ assertEquals(baseFile.getCommitTime(), compactionRequestedTime, "Expect
data-file created by compaction be returned");
+ assertFalse(baseFile.getBootstrapBaseFile().isPresent(), "No external
data file must be present");
});
- dataFiles = roView.getLatestBaseFilesBeforeOrOn(partitionPath,
deltaInstantTime5).collect(Collectors.toList());
- assertEquals(1, dataFiles.size(), "Expect only one data-file to be sent");
- dataFiles.forEach(df -> {
- assertEquals(df.getCommitTime(), compactionRequestedTime, "Expect
data-file created by compaction be returned");
- assertFalse(df.getBootstrapBaseFile().isPresent(), "No external data
file must be present");
+ baseFiles = roView.getLatestBaseFilesBeforeOrOn(partitionPath,
deltaInstantTime5).collect(Collectors.toList());
+ assertEquals(1, baseFiles.size(), "Expect only one data-file to be sent");
+ baseFiles.forEach(baseFile -> {
+ assertEquals(baseFile.getCommitTime(), compactionRequestedTime, "Expect
data-file created by compaction be returned");
+ assertFalse(baseFile.getBootstrapBaseFile().isPresent(), "No external
data file must be present");
});
- dataFiles =
roView.getLatestBaseFilesInRange(allInstantTimes).collect(Collectors.toList());
- assertEquals(1, dataFiles.size(), "Expect only one data-file to be sent");
- dataFiles.forEach(df -> {
- assertEquals(df.getCommitTime(), compactionRequestedTime, "Expect
data-file created by compaction be returned");
- assertFalse(df.getBootstrapBaseFile().isPresent(), "No external data
file must be present");
+ baseFiles =
roView.getLatestBaseFilesInRange(allInstantTimes).collect(Collectors.toList());
+ assertEquals(1, baseFiles.size(), "Expect only one data-file to be sent");
+ baseFiles.forEach(baseFile -> {
+ assertEquals(baseFile.getCommitTime(), compactionRequestedTime, "Expect
data-file created by compaction be returned");
+ assertFalse(baseFile.getBootstrapBaseFile().isPresent(), "No external
data file must be present");
});
assertEquals(expTotalFileSlices,
rtView.getAllFileSlices(partitionPath).count(),
@@ -1430,11 +1430,11 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
assertEquals(isLatestFileSliceOnly ? 1 : 2,
fileSliceMap.get(fileId3).longValue());
assertEquals(1, fileSliceMap.get(fileId4).longValue());
- List<HoodieBaseFile> dataFileList =
+ List<HoodieBaseFile> baseFileList =
roView.getLatestBaseFilesBeforeOrOn("2016/05/01",
commitTime5).collect(Collectors.toList());
- assertEquals(3, dataFileList.size());
+ assertEquals(3, baseFileList.size());
Set<String> filenames = new HashSet<>();
- for (HoodieBaseFile status : dataFileList) {
+ for (HoodieBaseFile status : baseFileList) {
filenames.add(status.getFileName());
}
assertTrue(filenames.contains(FSUtils.makeBaseFileName(commitTime3,
TEST_WRITE_TOKEN, fileId1, BASE_FILE_EXTENSION)));
@@ -1465,19 +1465,19 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
TEST_WRITE_TOKEN)));
// Reset the max commit time
- List<HoodieBaseFile> dataFiles =
+ List<HoodieBaseFile> baseFiles =
roView.getLatestBaseFilesBeforeOrOn("2016/05/01",
commitTime4).collect(Collectors.toList());
filenames = new HashSet<>();
- for (HoodieBaseFile status : dataFiles) {
+ for (HoodieBaseFile status : baseFiles) {
filenames.add(status.getFileName());
}
if (!isLatestFileSliceOnly) {
- assertEquals(3, dataFiles.size());
+ assertEquals(3, baseFiles.size());
assertTrue(filenames.contains(FSUtils.makeBaseFileName(commitTime3,
TEST_WRITE_TOKEN, fileId1, BASE_FILE_EXTENSION)));
assertTrue(filenames.contains(FSUtils.makeBaseFileName(commitTime4,
TEST_WRITE_TOKEN, fileId2, BASE_FILE_EXTENSION)));
assertTrue(filenames.contains(FSUtils.makeBaseFileName(commitTime4,
TEST_WRITE_TOKEN, fileId3, BASE_FILE_EXTENSION)));
} else {
- assertEquals(1, dataFiles.size());
+ assertEquals(1, baseFiles.size());
assertTrue(filenames.contains(FSUtils.makeBaseFileName(commitTime4,
TEST_WRITE_TOKEN, fileId2, BASE_FILE_EXTENSION)));
}
@@ -1530,9 +1530,9 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
for (HoodieFileGroup fileGroup : fileGroups) {
String fileId = fileGroup.getFileGroupId().getFileId();
Set<String> filenames = new HashSet<>();
- fileGroup.getAllBaseFiles().forEach(dataFile -> {
- assertEquals(fileId, dataFile.getFileId(), "All same fileId should be
grouped");
- filenames.add(dataFile.getFileName());
+ fileGroup.getAllBaseFiles().forEach(baseFile -> {
+ assertEquals(fileId, baseFile.getFileId(), "All same fileId should be
grouped");
+ filenames.add(baseFile.getFileName());
});
Set<String> expFileNames = new HashSet<>();
if (fileId.equals(fileId1)) {
@@ -1606,12 +1606,12 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
// Populate view for partition
roView.getAllBaseFiles("2016/05/01/");
- List<HoodieBaseFile> dataFiles =
+ List<HoodieBaseFile> baseFiles =
roView.getLatestBaseFilesInRange(Arrays.asList(commitTime2,
commitTime3))
.collect(Collectors.toList());
- assertEquals(3, dataFiles.size());
+ assertEquals(3, baseFiles.size());
Set<String> filenames = new HashSet<>();
- for (HoodieBaseFile status : dataFiles) {
+ for (HoodieBaseFile status : baseFiles) {
filenames.add(status.getFileName());
}
@@ -1675,12 +1675,12 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
assertEquals(7, partitionFileList.size());
refreshFsView(preTableVersion8);
- List<HoodieBaseFile> dataFiles =
+ List<HoodieBaseFile> baseFiles =
roView.getLatestBaseFilesBeforeOrOn(partitionPath, commitTime2)
.collect(Collectors.toList());
- assertEquals(2, dataFiles.size());
+ assertEquals(2, baseFiles.size());
Set<String> filenames = new HashSet<>();
- for (HoodieBaseFile status : dataFiles) {
+ for (HoodieBaseFile status : baseFiles) {
filenames.add(status.getFileName());
}
assertTrue(filenames.contains(FSUtils.makeBaseFileName(commitTime1,
TEST_WRITE_TOKEN, fileIdA, BASE_FILE_EXTENSION)));
@@ -1889,12 +1889,12 @@ public class TestHoodieTableFileSystemView extends
HoodieCommonTestHarness {
refreshFsView(preTableVersion8);
// Test Data Files
- List<HoodieBaseFile> dataFiles =
roView.getAllBaseFiles(partitionPath1).collect(Collectors.toList());
- assertEquals(1, dataFiles.size(), "One data-file is expected as there is
only one file-group");
- assertEquals(instantTime1, dataFiles.get(0).getCommitTime(), "Expect only
valid commit");
- dataFiles =
roView.getAllBaseFiles(partitionPath2).collect(Collectors.toList());
- assertEquals(1, dataFiles.size(), "One data-file is expected as there is
only one file-group");
- assertEquals(instantTime1, dataFiles.get(0).getCommitTime(), "Expect only
valid commit");
+ List<HoodieBaseFile> baseFiles =
roView.getAllBaseFiles(partitionPath1).collect(Collectors.toList());
+ assertEquals(1, baseFiles.size(), "One data-file is expected as there is
only one file-group");
+ assertEquals(instantTime1, baseFiles.get(0).getCommitTime(), "Expect only
valid commit");
+ baseFiles =
roView.getAllBaseFiles(partitionPath2).collect(Collectors.toList());
+ assertEquals(1, baseFiles.size(), "One data-file is expected as there is
only one file-group");
+ assertEquals(instantTime1, baseFiles.get(0).getCommitTime(), "Expect only
valid commit");
// Merge API Tests
Arrays.asList(partitionPath1, partitionPath2,
partitionPath3).forEach(partitionPath -> {
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
index 3e74ec204b76..82823eece7e8 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
@@ -730,8 +730,8 @@ public class TestCompactionUtils extends
HoodieCommonTestHarness {
assertEquals(slice.getBaseInstantTime(), op.getBaseInstantTime(), "Same
base-instant");
assertEquals(slice.getFileId(), op.getFileId(), "Same file-id");
if (slice.getBaseFile().isPresent()) {
- HoodieBaseFile df = slice.getBaseFile().get();
- assertEquals(version == COMPACTION_METADATA_VERSION_1 ? df.getPath() :
df.getFileName(),
+ HoodieBaseFile baseFile = slice.getBaseFile().get();
+ assertEquals(version == COMPACTION_METADATA_VERSION_1 ?
baseFile.getPath() : baseFile.getFileName(),
op.getDataFilePath(), "Same data-file");
}
List<String> paths = slice.getLogFiles().map(l ->
l.getPath().toString()).collect(Collectors.toList());
diff --git
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
index 88e04c377bd9..3d9265bbf8bc 100644
---
a/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
+++
b/hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieInputFormatUtils.java
@@ -495,19 +495,19 @@ public class HoodieInputFormatUtils {
* 3. Generation of splits looks at FileStatus size to create splits, which
skips this file
*
* @param conf
- * @param dataFile
+ * @param baseFile
* @return
*/
- private static HoodieBaseFile refreshFileStatus(Configuration conf,
HoodieBaseFile dataFile) {
- StoragePath dataPath = dataFile.getPathInfo().getPath();
+ private static HoodieBaseFile refreshFileStatus(Configuration conf,
HoodieBaseFile baseFile) {
+ StoragePath dataPath = baseFile.getPathInfo().getPath();
try {
- if (dataFile.getFileSize() == 0) {
+ if (baseFile.getFileSize() == 0) {
HoodieStorage storage = HoodieStorageUtils.getStorage(dataPath,
HadoopFSUtils.getStorageConf(conf));
- LOG.info("Refreshing file status {}", dataFile.getPath());
+ LOG.info("Refreshing file status {}", baseFile.getPath());
return new HoodieBaseFile(storage.getPathInfo(dataPath),
- dataFile.getBootstrapBaseFile().orElse(null));
+ baseFile.getBootstrapBaseFile().orElse(null));
}
- return dataFile;
+ return baseFile;
} catch (IOException e) {
throw new HoodieIOException("Could not get FileStatus on path " +
dataPath);
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieSparkMergeOnReadTableClustering.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieSparkMergeOnReadTableClustering.java
index 560f811cb27a..dda980ee9ebe 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieSparkMergeOnReadTableClustering.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestHoodieSparkMergeOnReadTableClustering.java
@@ -89,8 +89,8 @@ class TestHoodieSparkMergeOnReadTableClustering extends
SparkClientFunctionalTes
WriteClientTestUtils.startCommitWithTime(client, newCommitTime);
List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 400);
- Stream<HoodieBaseFile> dataFiles = insertRecordsToMORTable(metaClient,
records.subList(0, 200), client, cfg, newCommitTime);
- assertTrue(dataFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
+ Stream<HoodieBaseFile> baseFiles = insertRecordsToMORTable(metaClient,
records.subList(0, 200), client, cfg, newCommitTime);
+ assertTrue(baseFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
/*
* Write 2 (more inserts to create new files)
@@ -98,8 +98,8 @@ class TestHoodieSparkMergeOnReadTableClustering extends
SparkClientFunctionalTes
// we already set small file size to small number to force inserts to go
into new file.
newCommitTime = "002";
WriteClientTestUtils.startCommitWithTime(client, newCommitTime);
- dataFiles = insertRecordsToMORTable(metaClient, records.subList(200,
400), client, cfg, newCommitTime);
- assertTrue(dataFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
+ baseFiles = insertRecordsToMORTable(metaClient, records.subList(200,
400), client, cfg, newCommitTime);
+ assertTrue(baseFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
if (doUpdates) {
/*
@@ -185,12 +185,12 @@ class TestHoodieSparkMergeOnReadTableClustering extends
SparkClientFunctionalTes
String newCommitTime = "001";
WriteClientTestUtils.startCommitWithTime(client, newCommitTime);
List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 400);
- Stream<HoodieBaseFile> dataFiles = insertRecordsToMORTable(metaClient,
records.subList(0, 200), client, cfg, newCommitTime);
- assertTrue(!dataFiles.findAny().isPresent(), "should not have any base
files");
+ Stream<HoodieBaseFile> baseFiles = insertRecordsToMORTable(metaClient,
records.subList(0, 200), client, cfg, newCommitTime);
+ assertTrue(!baseFiles.findAny().isPresent(), "should not have any base
files");
newCommitTime = "002";
WriteClientTestUtils.startCommitWithTime(client, newCommitTime);
- dataFiles = insertRecordsToMORTable(metaClient, records.subList(200,
400), client, cfg, newCommitTime);
- assertTrue(!dataFiles.findAny().isPresent(), "should not have any base
files");
+ baseFiles = insertRecordsToMORTable(metaClient, records.subList(200,
400), client, cfg, newCommitTime);
+ assertTrue(!baseFiles.findAny().isPresent(), "should not have any base
files");
// run updates
if (doUpdates) {
newCommitTime = "003";
@@ -260,9 +260,9 @@ class TestHoodieSparkMergeOnReadTableClustering extends
SparkClientFunctionalTes
metaClient = HoodieTableMetaClient.reload(metaClient);
final HoodieTable clusteredTable = HoodieSparkTable.create(cfg, context(),
metaClient);
clusteredTable.getHoodieView().sync();
- Stream<HoodieBaseFile> dataFilesToRead =
Arrays.stream(dataGen.getPartitionPaths())
+ Stream<HoodieBaseFile> baseFilesToRead =
Arrays.stream(dataGen.getPartitionPaths())
.flatMap(p ->
clusteredTable.getBaseFileOnlyView().getLatestBaseFiles(p));
- assertEquals(dataGen.getPartitionPaths().length, dataFilesToRead.count());
+ assertEquals(dataGen.getPartitionPaths().length, baseFilesToRead.count());
HoodieTimeline timeline =
metaClient.getCommitTimeline().filterCompletedInstants();
assertEquals(1, timeline.findInstantsAfter("003",
Integer.MAX_VALUE).countInstants(),
"Expecting a single commit.");
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
index 846136bacff5..bdfb3b57dcd7 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
@@ -189,14 +189,14 @@ public class TestHoodieMergeOnReadTable extends
SparkClientFunctionalTestHarness
List<StoragePathInfo> allFiles = listAllBaseFilesInPath(hoodieTable);
BaseFileOnlyView roView = getHoodieTableFileSystemView(metaClient,
metaClient.getCommitsTimeline().filterCompletedInstants(), allFiles);
- Stream<HoodieBaseFile> dataFilesToRead = roView.getLatestBaseFiles();
+ Stream<HoodieBaseFile> baseFilesToRead = roView.getLatestBaseFiles();
Map<String, Long> fileIdToSize =
- dataFilesToRead.collect(Collectors.toMap(HoodieBaseFile::getFileId,
HoodieBaseFile::getFileSize));
+ baseFilesToRead.collect(Collectors.toMap(HoodieBaseFile::getFileId,
HoodieBaseFile::getFileSize));
roView = getHoodieTableFileSystemView(metaClient,
hoodieTable.getCompletedCommitsTimeline(), allFiles);
- dataFilesToRead = roView.getLatestBaseFiles();
- List<HoodieBaseFile> dataFilesList =
dataFilesToRead.collect(Collectors.toList());
- assertTrue(dataFilesList.size() > 0,
+ baseFilesToRead = roView.getLatestBaseFiles();
+ List<HoodieBaseFile> baseFilesList =
baseFilesToRead.collect(Collectors.toList());
+ assertTrue(baseFilesList.size() > 0,
"Should list the base files we wrote in the delta commit");
/**
@@ -224,10 +224,10 @@ public class TestHoodieMergeOnReadTable extends
SparkClientFunctionalTestHarness
allFiles = listAllBaseFilesInPath(hoodieTable);
roView = getHoodieTableFileSystemView(metaClient,
hoodieTable.getActiveTimeline().reload().getCommitsTimeline().filterCompletedInstants(),
allFiles);
- dataFilesToRead = roView.getLatestBaseFiles();
- List<HoodieBaseFile> newDataFilesList =
dataFilesToRead.collect(Collectors.toList());
+ baseFilesToRead = roView.getLatestBaseFiles();
+ List<HoodieBaseFile> newBaseFilesList =
baseFilesToRead.collect(Collectors.toList());
Map<String, Long> fileIdToNewSize =
-
newDataFilesList.stream().collect(Collectors.toMap(HoodieBaseFile::getFileId,
HoodieBaseFile::getFileSize));
+
newBaseFilesList.stream().collect(Collectors.toMap(HoodieBaseFile::getFileId,
HoodieBaseFile::getFileSize));
assertTrue(fileIdToNewSize.entrySet().stream().anyMatch(entry ->
fileIdToSize.get(entry.getKey()) < entry.getValue()));
@@ -866,12 +866,12 @@ public class TestHoodieMergeOnReadTable extends
SparkClientFunctionalTestHarness
List<StoragePathInfo> allFiles = listAllBaseFilesInPath(hoodieTable);
BaseFileOnlyView roView =
getHoodieTableFileSystemView(metaClient,
metaClient.getCommitTimeline().filterCompletedInstants(), allFiles);
- Stream<HoodieBaseFile> dataFilesToRead = roView.getLatestBaseFiles();
- assertFalse(dataFilesToRead.findAny().isPresent());
+ Stream<HoodieBaseFile> baseFilesToRead = roView.getLatestBaseFiles();
+ assertFalse(baseFilesToRead.findAny().isPresent());
roView = getHoodieTableFileSystemView(metaClient,
hoodieTable.getCompletedCommitsTimeline(), allFiles);
- dataFilesToRead = roView.getLatestBaseFiles();
- assertTrue(dataFilesToRead.findAny().isPresent(),
+ baseFilesToRead = roView.getLatestBaseFiles();
+ assertTrue(baseFilesToRead.findAny().isPresent(),
"should list the base files we wrote in the delta commit");
/**
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableIncrementalRead.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableIncrementalRead.java
index 1e731adeed98..ba2892a18c4c 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableIncrementalRead.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableIncrementalRead.java
@@ -98,8 +98,8 @@ public class TestHoodieSparkMergeOnReadTableIncrementalRead
extends SparkClientF
WriteClientTestUtils.startCommitWithTime(client, commitTime1);
List<HoodieRecord> records001 = dataGen.generateInserts(commitTime1,
200);
- Stream<HoodieBaseFile> dataFiles = insertRecordsToMORTable(metaClient,
records001, client, cfg, commitTime1);
- assertTrue(dataFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
+ Stream<HoodieBaseFile> baseFiles = insertRecordsToMORTable(metaClient,
records001, client, cfg, commitTime1);
+ assertTrue(baseFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
// verify only one base file shows up with commit time 001
FileStatus[] snapshotROFiles = getROSnapshotFiles(partitionPath);
@@ -147,8 +147,8 @@ public class TestHoodieSparkMergeOnReadTableIncrementalRead
extends SparkClientF
String insertsTime = "006";
List<HoodieRecord> records006 = dataGen.generateInserts(insertsTime,
200);
WriteClientTestUtils.startCommitWithTime(client, insertsTime);
- dataFiles = insertRecordsToMORTable(metaClient, records006, client, cfg,
insertsTime);
- assertTrue(dataFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
+ baseFiles = insertRecordsToMORTable(metaClient, records006, client, cfg,
insertsTime);
+ assertTrue(baseFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
// verify new write shows up in snapshot mode even though there is
pending compaction
snapshotROFiles = getROSnapshotFiles(partitionPath);
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java
index 8f2ba3a4947f..eb68d9abfd0c 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java
@@ -126,8 +126,8 @@ public class
TestHoodieSparkMergeOnReadTableInsertUpdateDelete extends SparkClie
WriteClientTestUtils.startCommitWithTime(client, newCommitTime);
List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 200);
- Stream<HoodieBaseFile> dataFiles = insertRecordsToMORTable(metaClient,
records, client, cfg, newCommitTime);
- assertTrue(dataFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
+ Stream<HoodieBaseFile> baseFiles = insertRecordsToMORTable(metaClient,
records, client, cfg, newCommitTime);
+ assertTrue(baseFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
/*
* Write 2 (updates)
@@ -147,8 +147,8 @@ public class
TestHoodieSparkMergeOnReadTableInsertUpdateDelete extends SparkClie
List<StoragePathInfo> allFiles = listAllBaseFilesInPath(hoodieTable);
HoodieTableFileSystemView tableView =
getHoodieTableFileSystemView(metaClient,
hoodieTable.getCompletedCommitsTimeline(), allFiles);
- Stream<HoodieBaseFile> dataFilesToRead = tableView.getLatestBaseFiles();
- assertTrue(dataFilesToRead.findAny().isPresent());
+ Stream<HoodieBaseFile> baseFilesToRead = tableView.getLatestBaseFiles();
+ assertTrue(baseFilesToRead.findAny().isPresent());
// verify that there is a commit
metaClient = HoodieTableMetaClient.reload(metaClient);
@@ -189,8 +189,8 @@ public class
TestHoodieSparkMergeOnReadTableInsertUpdateDelete extends SparkClie
WriteClientTestUtils.startCommitWithTime(client, newCommitTime);
List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 200);
- Stream<HoodieBaseFile> dataFiles = insertRecordsToMORTable(metaClient,
records, client, cfg, newCommitTime, true);
- assertTrue(dataFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
+ Stream<HoodieBaseFile> baseFiles = insertRecordsToMORTable(metaClient,
records, client, cfg, newCommitTime, true);
+ assertTrue(baseFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
/*
* Write 2 (updates)
@@ -231,8 +231,8 @@ public class
TestHoodieSparkMergeOnReadTableInsertUpdateDelete extends SparkClie
WriteClientTestUtils.startCommitWithTime(client, newCommitTime);
List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 200);
- Stream<HoodieBaseFile> dataFiles = insertRecordsToMORTable(metaClient,
records, client, cfg, newCommitTime, true);
- assertTrue(dataFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
+ Stream<HoodieBaseFile> baseFiles = insertRecordsToMORTable(metaClient,
records, client, cfg, newCommitTime, true);
+ assertTrue(baseFiles.findAny().isPresent(), "should list the base files
we wrote in the delta commit");
/*
* Write 2 (updates)
@@ -309,12 +309,12 @@ public class
TestHoodieSparkMergeOnReadTableInsertUpdateDelete extends SparkClie
HoodieTableFileSystemView tableView =
getHoodieTableFileSystemView(metaClient,
metaClient.getCommitTimeline().filterCompletedInstants(),
allFiles);
- Stream<HoodieBaseFile> dataFilesToRead = tableView.getLatestBaseFiles();
- assertFalse(dataFilesToRead.findAny().isPresent());
+ Stream<HoodieBaseFile> baseFilesToRead = tableView.getLatestBaseFiles();
+ assertFalse(baseFilesToRead.findAny().isPresent());
tableView = getHoodieTableFileSystemView(metaClient,
hoodieTable.getCompletedCommitsTimeline(), allFiles);
- dataFilesToRead = tableView.getLatestBaseFiles();
- assertTrue(dataFilesToRead.findAny().isPresent(),
+ baseFilesToRead = tableView.getLatestBaseFiles();
+ assertTrue(baseFilesToRead.findAny().isPresent(),
"should list the base files we wrote in the delta commit");
/*
@@ -352,8 +352,8 @@ public class
TestHoodieSparkMergeOnReadTableInsertUpdateDelete extends SparkClie
allFiles = listAllBaseFilesInPath(hoodieTable);
tableView = getHoodieTableFileSystemView(metaClient,
hoodieTable.getCompletedCommitsTimeline(), allFiles);
- dataFilesToRead = tableView.getLatestBaseFiles();
- assertTrue(dataFilesToRead.findAny().isPresent());
+ baseFilesToRead = tableView.getLatestBaseFiles();
+ assertTrue(baseFilesToRead.findAny().isPresent());
List<String> inputPaths = tableView.getLatestBaseFiles()
.map(baseFile -> new Path(baseFile.getPath()).getParent().toString())
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
index c61aeaf9cdac..53a49fd960ad 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
@@ -224,12 +224,12 @@ public class TestHoodieSparkMergeOnReadTableRollback
extends TestHoodieSparkRoll
List<StoragePathInfo> allFiles = listAllBaseFilesInPath(hoodieTable);
HoodieTableFileSystemView tableView =
getHoodieTableFileSystemView(metaClient,
metaClient.getCommitTimeline().filterCompletedInstants(), allFiles);
- Stream<HoodieBaseFile> dataFilesToRead = tableView.getLatestBaseFiles();
- assertFalse(dataFilesToRead.findAny().isPresent());
+ Stream<HoodieBaseFile> baseFilesToRead = tableView.getLatestBaseFiles();
+ assertFalse(baseFilesToRead.findAny().isPresent());
tableView = getHoodieTableFileSystemView(metaClient,
hoodieTable.getCompletedCommitsTimeline(), allFiles);
- dataFilesToRead = tableView.getLatestBaseFiles();
- assertTrue(dataFilesToRead.findAny().isPresent(),
+ baseFilesToRead = tableView.getLatestBaseFiles();
+ assertTrue(baseFilesToRead.findAny().isPresent(),
"should list the base files we wrote in the delta commit");
HoodieWriteConfig secondCfg =
getHoodieWriteConfigWithSmallFileHandlingOff(true);
@@ -424,12 +424,12 @@ public class TestHoodieSparkMergeOnReadTableRollback
extends TestHoodieSparkRoll
List<StoragePathInfo> allFiles = listAllBaseFilesInPath(hoodieTable);
HoodieTableFileSystemView tableView =
getHoodieTableFileSystemView(metaClient,
metaClient.getCommitTimeline().filterCompletedInstants(), allFiles);
- Stream<HoodieBaseFile> dataFilesToRead = tableView.getLatestBaseFiles();
- assertFalse(dataFilesToRead.findAny().isPresent());
+ Stream<HoodieBaseFile> baseFilesToRead = tableView.getLatestBaseFiles();
+ assertFalse(baseFilesToRead.findAny().isPresent());
tableView = getHoodieTableFileSystemView(metaClient,
hoodieTable.getCompletedCommitsTimeline(), allFiles);
- dataFilesToRead = tableView.getLatestBaseFiles();
- assertTrue(dataFilesToRead.findAny().isPresent(),
+ baseFilesToRead = tableView.getLatestBaseFiles();
+ assertTrue(baseFilesToRead.findAny().isPresent(),
"should list the base files we wrote in the delta commit");
/*
@@ -541,14 +541,14 @@ public class TestHoodieSparkMergeOnReadTableRollback
extends TestHoodieSparkRoll
List<StoragePathInfo> allFiles = listAllBaseFilesInPath(hoodieTable);
HoodieTableFileSystemView tableView =
getHoodieTableFileSystemView(metaClient,
metaClient.getCommitTimeline().filterCompletedInstants(), allFiles);
- Stream<HoodieBaseFile> dataFilesToRead = tableView.getLatestBaseFiles();
- assertFalse(dataFilesToRead.findAny().isPresent());
+ Stream<HoodieBaseFile> baseFilesToRead = tableView.getLatestBaseFiles();
+ assertFalse(baseFilesToRead.findAny().isPresent());
tableView =
getHoodieTableFileSystemView(metaClient,
hoodieTable.getCompletedCommitsTimeline(),
allFiles);
- dataFilesToRead = tableView.getLatestBaseFiles();
- assertTrue(dataFilesToRead.findAny().isPresent(),
+ baseFilesToRead = tableView.getLatestBaseFiles();
+ assertTrue(baseFilesToRead.findAny().isPresent(),
"Should list the base files we wrote in the delta commit");
/*
@@ -663,8 +663,8 @@ public class TestHoodieSparkMergeOnReadTableRollback
extends TestHoodieSparkRoll
metaClient.reloadActiveTimeline();
allFiles = listAllBaseFilesInPath(hoodieTable);
tableView = getHoodieTableFileSystemView(metaClient,
metaClient.reloadActiveTimeline().getCommitsTimeline().filterCompletedInstants(),
allFiles);
- dataFilesToRead = tableView.getLatestBaseFiles();
- assertFalse(dataFilesToRead.filter(hoodieBaseFile ->
hoodieBaseFile.getCommitTime().compareTo("003") > 0).findAny().isPresent());
+ baseFilesToRead = tableView.getLatestBaseFiles();
+ assertFalse(baseFilesToRead.filter(hoodieBaseFile ->
hoodieBaseFile.getCommitTime().compareTo("003") > 0).findAny().isPresent());
TableFileSystemView.SliceView rtView =
getHoodieTableFileSystemView(metaClient,
metaClient.getCommitTimeline().filterCompletedInstants(), allFiles);
List<HoodieFileGroup> fileGroups =
((HoodieTableFileSystemView)
rtView).getAllFileGroups().collect(Collectors.toList());
@@ -749,8 +749,8 @@ public class TestHoodieSparkMergeOnReadTableRollback
extends TestHoodieSparkRoll
List<StoragePathInfo> allFiles = listAllBaseFilesInPath(hoodieTable);
HoodieTableFileSystemView tableView =
getHoodieTableFileSystemView(metaClient,
metaClient.getCommitTimeline().filterCompletedInstants(), allFiles);
- Stream<HoodieBaseFile> dataFilesToRead = tableView.getLatestBaseFiles();
- assertFalse(dataFilesToRead.anyMatch(
+ Stream<HoodieBaseFile> baseFilesToRead = tableView.getLatestBaseFiles();
+ assertFalse(baseFilesToRead.anyMatch(
file -> compareTimestamps(commit2, GREATER_THAN,
file.getCommitTime())));
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java
index 08ce27d99e8f..50a79e0012ed 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.java
@@ -536,10 +536,10 @@ public class HiveTestUtil {
String partitionPath = wEntry.getKey();
for (HoodieWriteStat wStat : wEntry.getValue()) {
StoragePath path = new StoragePath(wStat.getPath());
- HoodieBaseFile dataFile = new
HoodieBaseFile(storage.getPathInfo(path));
+ HoodieBaseFile baseFile = new
HoodieBaseFile(storage.getPathInfo(path));
HoodieLogFile logFile = generateLogData(path, isLogSchemaSimple);
HoodieDeltaWriteStat writeStat = new HoodieDeltaWriteStat();
- writeStat.setFileId(dataFile.getFileId());
+ writeStat.setFileId(baseFile.getFileId());
writeStat.setPath(logFile.getPath().toString());
writeStat.setNumUpdateWrites(10);
commitMetadata.addWriteStat(partitionPath, writeStat);
@@ -557,10 +557,10 @@ public class HiveTestUtil {
String partitionPath = wEntry.getKey();
for (HoodieWriteStat wStat : wEntry.getValue()) {
StoragePath path = new StoragePath(wStat.getPath());
- HoodieBaseFile dataFile = new
HoodieBaseFile(storage.getPathInfo(path));
+ HoodieBaseFile baseFile = new
HoodieBaseFile(storage.getPathInfo(path));
HoodieLogFile logFile = generateLogData(path, logSchemaPath, dataPath);
HoodieDeltaWriteStat writeStat = new HoodieDeltaWriteStat();
- writeStat.setFileId(dataFile.getFileId());
+ writeStat.setFileId(baseFile.getFileId());
writeStat.setPath(logFile.getPath().toString());
writeStat.setNumUpdateWrites(10);
commitMetadata.addWriteStat(partitionPath, writeStat);
@@ -727,18 +727,18 @@ public class HiveTestUtil {
boolean isLogSchemaSimple)
throws IOException, InterruptedException, URISyntaxException {
HoodieSchema schema = getTestDataSchema(isLogSchemaSimple);
- HoodieBaseFile dataFile = new
HoodieBaseFile(storage.getPathInfo(parquetFilePath));
+ HoodieBaseFile baseFile = new
HoodieBaseFile(storage.getPathInfo(parquetFilePath));
// Write a log file for this parquet file
Writer logWriter = HoodieLogFormatWriter.builder()
.withParentPath(parquetFilePath.getParent())
.withFileExtension(HoodieLogFile.DELTA_EXTENSION)
- .withLogFileId(dataFile.getFileId())
-
.withInstantTime(dataFile.getCommitTime()).withStorage(storage).build();
+ .withLogFileId(baseFile.getFileId())
+
.withInstantTime(baseFile.getCommitTime()).withStorage(storage).build();
List<HoodieRecord> records = (isLogSchemaSimple ?
SchemaTestUtil.generateTestRecords(0, 100)
: SchemaTestUtil.generateEvolvedTestRecords(100, 100)).stream()
.map(HoodieAvroIndexedRecord::new).collect(Collectors.toList());
Map<HeaderMetadataType, String> header = new HashMap<>(2);
- header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME,
dataFile.getCommitTime());
+ header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME,
baseFile.getCommitTime());
header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString());
HoodieAvroDataBlock dataBlock = new HoodieAvroDataBlock(records, header,
HoodieRecord.RECORD_KEY_METADATA_FIELD);
logWriter.appendBlock(dataBlock);
@@ -749,18 +749,18 @@ public class HiveTestUtil {
private static HoodieLogFile generateLogData(StoragePath parquetFilePath,
String logSchemaPath, String dataPath)
throws IOException, InterruptedException, URISyntaxException {
HoodieSchema schema = SchemaTestUtil.getSchema(logSchemaPath);
- HoodieBaseFile dataFile = new
HoodieBaseFile(storage.getPathInfo(parquetFilePath));
+ HoodieBaseFile baseFile = new
HoodieBaseFile(storage.getPathInfo(parquetFilePath));
// Write a log file for this parquet file
Writer logWriter = HoodieLogFormatWriter.builder()
.withParentPath(parquetFilePath.getParent())
.withFileExtension(HoodieLogFile.DELTA_EXTENSION)
- .withLogFileId(dataFile.getFileId())
- .withInstantTime(dataFile.getCommitTime())
+ .withLogFileId(baseFile.getFileId())
+ .withInstantTime(baseFile.getCommitTime())
.withStorage(storage)
.build();
List<HoodieRecord> records =
SchemaTestUtil.generateTestRecords(logSchemaPath,
dataPath).stream().map(HoodieAvroIndexedRecord::new).collect(Collectors.toList());
Map<HeaderMetadataType, String> header = new HashMap<>(2);
- header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME,
dataFile.getCommitTime());
+ header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME,
baseFile.getCommitTime());
header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema.toString());
HoodieAvroDataBlock dataBlock = new HoodieAvroDataBlock(records, header,
HoodieRecord.RECORD_KEY_METADATA_FIELD);
logWriter.appendBlock(dataBlock);
