lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-402]: code clean up in test cases

Browse Source
This commit is contained in:
Pratyaksh Sharma
2019-12-29 15:00:06 +05:30
committed by vinoth chandar
parent 98c0d8cf60
commit dde21e7315
62 changed files with 320 additions and 619 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
hudi-client/src/test/java/org/apache/hudi/HoodieClientTestHarness.java
View File

@@ -190,18 +190,16 @@ public abstract class HoodieClientTestHarness extends HoodieCommonTestHarness im
/**
* Initializes a test data generator which used to generate test datas.
*
* @throws IOException
*/
protected void initTestDataGenerator() throws IOException {
protected void initTestDataGenerator() {
dataGen = new HoodieTestDataGenerator();
}
/**
* Cleanups test data generator.
*
* @throws IOException
*/
protected void cleanupTestDataGenerator() throws IOException {
protected void cleanupTestDataGenerator() {
dataGen = null;
}

59
hudi-client/src/test/java/org/apache/hudi/TestCleaner.java
View File

@@ -365,7 +365,7 @@ public class TestCleaner extends TestHoodieClientBase {
insertFirstBigBatchForClientCleanerTest(cfg, client, recordInsertGenWrappedFunction, insertFn);
// Keep doing some writes and clean inline. Make sure we have expected number of files remaining.
HoodieTestUtils.monotonicIncreasingCommitTimestamps(8, 1).stream().forEach(newCommitTime -> {
HoodieTestUtils.monotonicIncreasingCommitTimestamps(8, 1).forEach(newCommitTime -> {
try {
client.startCommitWithTime(newCommitTime);
List<HoodieRecord> records = recordUpsertGenWrappedFunction.apply(newCommitTime, 100);
@@ -441,7 +441,7 @@ public class TestCleaner extends TestHoodieClientBase {
Assert.assertEquals(new Integer(0), cleanMetadata2.getTotalFilesDeleted());
Assert.assertEquals(cleanMetadata1.getPartitionMetadata().keySet(),
cleanMetadata2.getPartitionMetadata().keySet());
cleanMetadata1.getPartitionMetadata().keySet().stream().forEach(k -> {
cleanMetadata1.getPartitionMetadata().keySet().forEach(k -> {
HoodieCleanPartitionMetadata p1 = cleanMetadata1.getPartitionMetadata().get(k);
HoodieCleanPartitionMetadata p2 = cleanMetadata2.getPartitionMetadata().get(k);
Assert.assertEquals(p1.getDeletePathPatterns(), p2.getDeletePathPatterns());
@@ -450,7 +450,8 @@ public class TestCleaner extends TestHoodieClientBase {
Assert.assertEquals(k, p1.getPartitionPath());
});
}
List<HoodieCleanStat> stats = cleanMetadata1.getPartitionMetadata().values().stream()
return cleanMetadata1.getPartitionMetadata().values().stream()
.map(x -> new HoodieCleanStat.Builder().withPartitionPath(x.getPartitionPath())
.withFailedDeletes(x.getFailedDeleteFiles()).withSuccessfulDeletes(x.getSuccessDeleteFiles())
.withPolicy(HoodieCleaningPolicy.valueOf(x.getPolicy())).withDeletePathPattern(x.getDeletePathPatterns())
@@ -459,8 +460,6 @@ public class TestCleaner extends TestHoodieClientBase {
: null))
.build())
.collect(Collectors.toList());
return stats;
}
/**
@@ -482,7 +481,6 @@ public class TestCleaner extends TestHoodieClientBase {
String file1P1C0 =
HoodieTestUtils.createNewDataFile(basePath, HoodieTestDataGenerator.DEFAULT_SECOND_PARTITION_PATH, "000");
metaClient = HoodieTableMetaClient.reload(metaClient);
HoodieTable table = HoodieTable.getHoodieTable(metaClient, config, jsc);
List<HoodieCleanStat> hoodieCleanStatsOne = runCleaner(config);
assertEquals("Must not clean any files", 0,
@@ -499,7 +497,6 @@ public class TestCleaner extends TestHoodieClientBase {
// make next commit, with 1 insert & 1 update per partition
HoodieTestUtils.createCommitFiles(basePath, "001");
metaClient = HoodieTableMetaClient.reload(metaClient);
table = HoodieTable.getHoodieTable(metaClient, config, jsc);
String file2P0C1 =
HoodieTestUtils.createNewDataFile(basePath, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, "001"); // insert
@@ -527,7 +524,6 @@ public class TestCleaner extends TestHoodieClientBase {
// make next commit, with 2 updates to existing files, and 1 insert
HoodieTestUtils.createCommitFiles(basePath, "002");
metaClient = HoodieTableMetaClient.reload(metaClient);
table = HoodieTable.getHoodieTable(metaClient, config, jsc);
HoodieTestUtils.createDataFile(basePath, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, "002", file1P0C0); // update
HoodieTestUtils.createDataFile(basePath, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, "002", file2P0C1); // update
@@ -567,31 +563,25 @@ public class TestCleaner extends TestHoodieClientBase {
.withCleanerPolicy(HoodieCleaningPolicy.KEEP_LATEST_FILE_VERSIONS).retainFileVersions(1).build())
.build();
HoodieTableMetaClient metaClient =
HoodieTestUtils.init(jsc.hadoopConfiguration(), basePath, HoodieTableType.MERGE_ON_READ);
HoodieTestUtils.init(jsc.hadoopConfiguration(), basePath, HoodieTableType.MERGE_ON_READ);
// Make 3 files, one base file and 2 log files associated with base file
String file1P0 =
HoodieTestUtils.createNewDataFile(basePath, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, "000");
String file2P0L0 = HoodieTestUtils.createNewLogFile(fs, basePath,
HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, "000", file1P0, Option.empty());
String file2P0L1 = HoodieTestUtils.createNewLogFile(fs, basePath,
HoodieTestUtils.createNewLogFile(fs, basePath,
HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, "000", file1P0, Option.of(2));
// make 1 compaction commit
HoodieTestUtils.createCompactionCommitFiles(fs, basePath, "000");
// Make 4 files, one base file and 3 log files associated with base file
HoodieTestUtils.createDataFile(basePath, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, "001", file1P0);
file2P0L0 = HoodieTestUtils.createNewLogFile(fs, basePath, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH,
"001", file1P0, Option.empty());
file2P0L0 = HoodieTestUtils.createNewLogFile(fs, basePath, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH,
"001", file1P0, Option.of(2));
file2P0L0 = HoodieTestUtils.createNewLogFile(fs, basePath, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH,
"001", file1P0, Option.of(3));
// make 1 compaction commit
HoodieTestUtils.createCompactionCommitFiles(fs, basePath, "001");
HoodieTable table = HoodieTable.getHoodieTable(metaClient, config, jsc);
List<HoodieCleanStat> hoodieCleanStats = runCleaner(config);
assertEquals("Must clean three files, one parquet and 2 log files", 3,
getCleanStat(hoodieCleanStats, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH).getSuccessDeleteFiles()
@@ -679,14 +669,14 @@ public class TestCleaner extends TestHoodieClientBase {
Assert.assertEquals(map1.keySet(), map2.keySet());
List<String> partitions1 = map1.values().stream().map(m -> m.getPartitionPath()).collect(
List<String> partitions1 = map1.values().stream().map(HoodieCleanPartitionMetadata::getPartitionPath).collect(
Collectors.toList());
List<String> partitions2 = map2.values().stream().map(m -> m.getPartitionPath()).collect(
List<String> partitions2 = map2.values().stream().map(HoodieCleanPartitionMetadata::getPartitionPath).collect(
Collectors.toList());
Assert.assertEquals(partitions1, partitions2);
List<String> policies1 = map1.values().stream().map(m -> m.getPolicy()).collect(Collectors.toList());
List<String> policies2 = map2.values().stream().map(m -> m.getPolicy()).collect(Collectors.toList());
List<String> policies1 = map1.values().stream().map(HoodieCleanPartitionMetadata::getPolicy).collect(Collectors.toList());
List<String> policies2 = map2.values().stream().map(HoodieCleanPartitionMetadata::getPolicy).collect(Collectors.toList());
Assert.assertEquals(policies1, policies2);
}
@@ -758,7 +748,6 @@ public class TestCleaner extends TestHoodieClientBase {
Option.of(commitMetadata.toJsonString().getBytes(StandardCharsets.UTF_8)));
metaClient = HoodieTableMetaClient.reload(metaClient);
HoodieTable table = HoodieTable.getHoodieTable(metaClient, config, jsc);
List<HoodieCleanStat> hoodieCleanStatsOne = runCleaner(config, simulateFailureRetry);
assertEquals("Must not clean any files", 0,
@@ -775,7 +764,6 @@ public class TestCleaner extends TestHoodieClientBase {
// make next commit, with 1 insert & 1 update per partition
HoodieTestUtils.createInflightCommitFiles(basePath, "001");
metaClient = HoodieTableMetaClient.reload(metaClient);
table = HoodieTable.getHoodieTable(metaClient, config, jsc);
String file2P0C1 =
HoodieTestUtils
@@ -815,7 +803,6 @@ public class TestCleaner extends TestHoodieClientBase {
// make next commit, with 2 updates to existing files, and 1 insert
HoodieTestUtils.createInflightCommitFiles(basePath, "002");
metaClient = HoodieTableMetaClient.reload(metaClient);
table = HoodieTable.getHoodieTable(metaClient, config, jsc);
HoodieTestUtils
.createDataFile(basePath, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, "002", file1P0C0); // update
@@ -843,7 +830,6 @@ public class TestCleaner extends TestHoodieClientBase {
// make next commit, with 2 updates to existing files, and 1 insert
HoodieTestUtils.createInflightCommitFiles(basePath, "003");
metaClient = HoodieTableMetaClient.reload(metaClient);
table = HoodieTable.getHoodieTable(metaClient, config, jsc);
HoodieTestUtils
.createDataFile(basePath, HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, "003", file1P0C0); // update
@@ -938,7 +924,6 @@ public class TestCleaner extends TestHoodieClientBase {
HoodieTestUtils.createCommitFiles(basePath, "000");
metaClient = HoodieTableMetaClient.reload(metaClient);
HoodieTable table = HoodieTable.getHoodieTable(metaClient, config, jsc);
List<HoodieCleanStat> hoodieCleanStatsOne = runCleaner(config);
assertTrue("HoodieCleanStats should be empty for a table with empty partitionPaths", hoodieCleanStatsOne.isEmpty());
@@ -1078,19 +1063,17 @@ public class TestCleaner extends TestHoodieClientBase {
// Clean now
metaClient = HoodieTableMetaClient.reload(metaClient);
HoodieTable table = HoodieTable.getHoodieTable(metaClient, config, jsc);
List<HoodieCleanStat> hoodieCleanStats = runCleaner(config, retryFailure);
// Test for safety
final HoodieTableMetaClient newMetaClient = HoodieTableMetaClient.reload(metaClient);
final HoodieTable hoodieTable = HoodieTable.getHoodieTable(metaClient, config, jsc);
expFileIdToPendingCompaction.entrySet().stream().forEach(entry -> {
String fileId = entry.getKey();
expFileIdToPendingCompaction.forEach((fileId, value) -> {
String baseInstantForCompaction = fileIdToLatestInstantBeforeCompaction.get(fileId);
Option<FileSlice> fileSliceForCompaction = Option.fromJavaOptional(hoodieTable.getRTFileSystemView()
.getLatestFileSlicesBeforeOrOn(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, baseInstantForCompaction,
true)
true)
.filter(fs -> fs.getFileId().equals(fileId)).findFirst());
Assert.assertTrue("Base Instant for Compaction must be preserved", fileSliceForCompaction.isPresent());
Assert.assertTrue("FileSlice has data-file", fileSliceForCompaction.get().getDataFile().isPresent());
@@ -1112,7 +1095,7 @@ public class TestCleaner extends TestHoodieClientBase {
});
}).filter(x -> x).count();
long numDeleted =
hoodieCleanStats.stream().flatMap(cleanStat -> cleanStat.getDeletePathPatterns().stream()).count();
hoodieCleanStats.stream().mapToLong(cleanStat -> cleanStat.getDeletePathPatterns().size()).sum();
// Tighter check for regression
Assert.assertEquals("Correct number of files deleted", expNumFilesDeleted, numDeleted);
Assert.assertEquals("Correct number of files under compaction deleted", expNumFilesUnderCompactionDeleted,
@@ -1170,15 +1153,13 @@ public class TestCleaner extends TestHoodieClientBase {
private static HoodieCommitMetadata generateCommitMetadata(Map<String, List<String>> partitionToFilePaths) {
HoodieCommitMetadata metadata = new HoodieCommitMetadata();
partitionToFilePaths.entrySet().forEach(e -> {
e.getValue().forEach(f -> {
HoodieWriteStat writeStat = new HoodieWriteStat();
writeStat.setPartitionPath(e.getKey());
writeStat.setPath(f);
writeStat.setFileId(f);
metadata.addWriteStat(e.getKey(), writeStat);
});
});
partitionToFilePaths.forEach((key, value) -> value.forEach(f -> {
HoodieWriteStat writeStat = new HoodieWriteStat();
writeStat.setPartitionPath(key);
writeStat.setPath(f);
writeStat.setFileId(f);
metadata.addWriteStat(key, writeStat);
}));
return metadata;
}
}

6
hudi-client/src/test/java/org/apache/hudi/TestClientRollback.java
View File

@@ -127,9 +127,7 @@ public class TestClientRollback extends TestHoodieClientBase {
table = HoodieTable.getHoodieTable(metaClient, getConfig(), jsc);
final ReadOptimizedView view2 = table.getROFileSystemView();
dataFiles = partitionPaths.stream().flatMap(s -> {
return view2.getAllDataFiles(s).filter(f -> f.getCommitTime().equals("004"));
}).collect(Collectors.toList());
dataFiles = partitionPaths.stream().flatMap(s -> view2.getAllDataFiles(s).filter(f -> f.getCommitTime().equals("004"))).collect(Collectors.toList());
assertEquals("The data files for commit 004 should be present", 3, dataFiles.size());
// rolling back to a non existent savepoint must not succeed
@@ -205,7 +203,7 @@ public class TestClientRollback extends TestHoodieClientBase {
// Rollback commit 1 (this should fail, since commit2 is still around)
try {
client.rollback(commitTime1);
assertTrue("Should have thrown an exception ", false);
fail("Should have thrown an exception ");
} catch (HoodieRollbackException hrbe) {
// should get here
}

41
hudi-client/src/test/java/org/apache/hudi/TestCompactionAdminClient.java
View File

@@ -20,6 +20,8 @@ package org.apache.hudi;
import org.apache.hudi.CompactionAdminClient.ValidationOpResult;
import org.apache.hudi.common.model.CompactionOperation;
import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieFileGroup;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieTestUtils;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -31,6 +33,7 @@ import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.func.OperationResult;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.junit.After;
@@ -39,12 +42,12 @@ import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.apache.hudi.common.model.HoodieTableType.MERGE_ON_READ;
@@ -91,7 +94,7 @@ public class TestCompactionAdminClient extends TestHoodieClientBase {
CompactionTestUtils.setupAndValidateCompactionOperations(metaClient, false, numEntriesPerInstant,
numEntriesPerInstant, numEntriesPerInstant, numEntriesPerInstant);
Map<String, CompactionOperation> instantsWithOp =
Arrays.asList("001", "003", "005", "007").stream().map(instant -> {
Stream.of("001", "003", "005", "007").map(instant -> {
try {
return Pair.of(instant, CompactionUtils.getCompactionPlan(metaClient, instant));
} catch (IOException ioe) {
@@ -132,7 +135,7 @@ public class TestCompactionAdminClient extends TestHoodieClientBase {
metaClient = new HoodieTableMetaClient(metaClient.getHadoopConf(), basePath, true);
List<ValidationOpResult> result = client.validateCompactionPlan(metaClient, compactionInstant, 1);
if (expNumRepairs > 0) {
Assert.assertTrue("Expect some failures in validation", result.stream().filter(r -> !r.isSuccess()).count() > 0);
Assert.assertTrue("Expect some failures in validation", result.stream().anyMatch(r -> !r.isSuccess()));
}
// Now repair
List<Pair<HoodieLogFile, HoodieLogFile>> undoFiles =
@@ -154,15 +157,13 @@ public class TestCompactionAdminClient extends TestHoodieClientBase {
} else {
Assert.assertTrue("Rename Files must be empty", renameFiles.isEmpty());
}
expRenameFiles.entrySet().stream().forEach(r -> {
LOG.info("Key :" + r.getKey() + " renamed to " + r.getValue() + " rolled back to "
+ renameFilesFromUndo.get(r.getKey()));
});
expRenameFiles.forEach((key, value) -> LOG.info("Key :" + key + " renamed to " + value + " rolled back to "
+ renameFilesFromUndo.get(key)));
Assert.assertEquals("Undo must completely rollback renames", expRenameFiles, renameFilesFromUndo);
// Now expect validation to succeed
result = client.validateCompactionPlan(metaClient, compactionInstant, 1);
Assert.assertTrue("Expect no failures in validation", result.stream().filter(r -> !r.isSuccess()).count() == 0);
Assert.assertTrue("Expect no failures in validation", result.stream().allMatch(OperationResult::isSuccess));
Assert.assertEquals("Expected Num Repairs", expNumRepairs, undoFiles.size());
}
@@ -176,7 +177,7 @@ public class TestCompactionAdminClient extends TestHoodieClientBase {
// Ensure compaction-plan is good to begin with
List<ValidationOpResult> validationResults = client.validateCompactionPlan(metaClient, compactionInstant, 1);
Assert.assertFalse("Some validations failed",
validationResults.stream().filter(v -> !v.isSuccess()).findAny().isPresent());
validationResults.stream().anyMatch(v -> !v.isSuccess()));
}
private void validateRenameFiles(List<Pair<HoodieLogFile, HoodieLogFile>> renameFiles, String ingestionInstant,
@@ -185,14 +186,14 @@ public class TestCompactionAdminClient extends TestHoodieClientBase {
Set<HoodieLogFile> uniqNewLogFiles = new HashSet<>();
Set<HoodieLogFile> uniqOldLogFiles = new HashSet<>();
renameFiles.stream().forEach(lfPair -> {
renameFiles.forEach(lfPair -> {
Assert.assertFalse("Old Log File Names do not collide", uniqOldLogFiles.contains(lfPair.getKey()));
Assert.assertFalse("New Log File Names do not collide", uniqNewLogFiles.contains(lfPair.getValue()));
uniqOldLogFiles.add(lfPair.getKey());
uniqNewLogFiles.add(lfPair.getValue());
});
renameFiles.stream().forEach(lfPair -> {
renameFiles.forEach(lfPair -> {
HoodieLogFile oldLogFile = lfPair.getLeft();
HoodieLogFile newLogFile = lfPair.getValue();
Assert.assertEquals("Base Commit time is expected", ingestionInstant, newLogFile.getBaseCommitTime());
@@ -234,18 +235,18 @@ public class TestCompactionAdminClient extends TestHoodieClientBase {
// Log files belonging to file-slices created because of compaction request must be renamed
Set<HoodieLogFile> gotLogFilesToBeRenamed = renameFiles.stream().map(p -> p.getLeft()).collect(Collectors.toSet());
Set<HoodieLogFile> gotLogFilesToBeRenamed = renameFiles.stream().map(Pair::getLeft).collect(Collectors.toSet());
final HoodieTableFileSystemView fsView =
new HoodieTableFileSystemView(metaClient, metaClient.getCommitsAndCompactionTimeline());
Set<HoodieLogFile> expLogFilesToBeRenamed = fsView.getLatestFileSlices(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0])
.filter(fs -> fs.getBaseInstantTime().equals(compactionInstant)).flatMap(fs -> fs.getLogFiles())
.filter(fs -> fs.getBaseInstantTime().equals(compactionInstant)).flatMap(FileSlice::getLogFiles)
.collect(Collectors.toSet());
Assert.assertEquals("Log files belonging to file-slices created because of compaction request must be renamed",
expLogFilesToBeRenamed, gotLogFilesToBeRenamed);
if (skipUnSchedule) {
// Do the renaming only but do not touch the compaction plan - Needed for repair tests
renameFiles.stream().forEach(lfPair -> {
renameFiles.forEach(lfPair -> {
try {
client.renameLogFile(metaClient, lfPair.getLeft(), lfPair.getRight());
} catch (IOException e) {
@@ -273,12 +274,12 @@ public class TestCompactionAdminClient extends TestHoodieClientBase {
newFsView.getLatestFileSlicesBeforeOrOn(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0], compactionInstant, true)
.filter(fs -> fs.getBaseInstantTime().equals(compactionInstant)).forEach(fs -> {
Assert.assertFalse("No Data file must be present", fs.getDataFile().isPresent());
Assert.assertTrue("No Log Files", fs.getLogFiles().count() == 0);
Assert.assertEquals("No Log Files", 0, fs.getLogFiles().count());
});
// Ensure same number of log-files before and after renaming per fileId
Map<String, Long> fileIdToCountsAfterRenaming =
newFsView.getAllFileGroups(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0]).flatMap(fg -> fg.getAllFileSlices())
newFsView.getAllFileGroups(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0]).flatMap(HoodieFileGroup::getAllFileSlices)
.filter(fs -> fs.getBaseInstantTime().equals(ingestionInstant))
.map(fs -> Pair.of(fs.getFileId(), fs.getLogFiles().count()))
.collect(Collectors.toMap(Pair::getKey, Pair::getValue));
@@ -305,12 +306,12 @@ public class TestCompactionAdminClient extends TestHoodieClientBase {
// Log files belonging to file-slices created because of compaction request must be renamed
Set<HoodieLogFile> gotLogFilesToBeRenamed = renameFiles.stream().map(p -> p.getLeft()).collect(Collectors.toSet());
Set<HoodieLogFile> gotLogFilesToBeRenamed = renameFiles.stream().map(Pair::getLeft).collect(Collectors.toSet());
final HoodieTableFileSystemView fsView =
new HoodieTableFileSystemView(metaClient, metaClient.getCommitsAndCompactionTimeline());
Set<HoodieLogFile> expLogFilesToBeRenamed = fsView.getLatestFileSlices(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0])
.filter(fs -> fs.getBaseInstantTime().equals(compactionInstant))
.filter(fs -> fs.getFileId().equals(op.getFileId())).flatMap(fs -> fs.getLogFiles())
.filter(fs -> fs.getFileId().equals(op.getFileId())).flatMap(FileSlice::getLogFiles)
.collect(Collectors.toSet());
Assert.assertEquals("Log files belonging to file-slices created because of compaction request must be renamed",
expLogFilesToBeRenamed, gotLogFilesToBeRenamed);
@@ -334,12 +335,12 @@ public class TestCompactionAdminClient extends TestHoodieClientBase {
.filter(fs -> fs.getBaseInstantTime().equals(compactionInstant))
.filter(fs -> fs.getFileId().equals(op.getFileId())).forEach(fs -> {
Assert.assertFalse("No Data file must be present", fs.getDataFile().isPresent());
Assert.assertTrue("No Log Files", fs.getLogFiles().count() == 0);
Assert.assertEquals("No Log Files", 0, fs.getLogFiles().count());
});
// Ensure same number of log-files before and after renaming per fileId
Map<String, Long> fileIdToCountsAfterRenaming =
newFsView.getAllFileGroups(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0]).flatMap(fg -> fg.getAllFileSlices())
newFsView.getAllFileGroups(HoodieTestUtils.DEFAULT_PARTITION_PATHS[0]).flatMap(HoodieFileGroup::getAllFileSlices)
.filter(fs -> fs.getBaseInstantTime().equals(ingestionInstant))
.filter(fs -> fs.getFileId().equals(op.getFileId()))
.map(fs -> Pair.of(fs.getFileId(), fs.getLogFiles().count()))

55
hudi-client/src/test/java/org/apache/hudi/TestHoodieClientBase.java
View File

@@ -26,7 +26,6 @@ import org.apache.hudi.common.model.EmptyHoodieRecordPayload;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodiePartitionMetadata;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTestUtils;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTimeline;
import org.apache.hudi.common.table.SyncableFileSystemView;
@@ -54,12 +53,12 @@ import org.junit.Assert;
import org.junit.Before;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -217,8 +216,7 @@ public class TestHoodieClientBase extends HoodieClientTestHarness {
if (!partitionToKeys.containsKey(partitionPath)) {
partitionToKeys.put(partitionPath, new HashSet<>());
}
assertTrue("key " + key + " is duplicate within partition " + partitionPath,
!partitionToKeys.get(partitionPath).contains(key));
assertFalse("key " + key + " is duplicate within partition " + partitionPath, partitionToKeys.get(partitionPath).contains(key));
partitionToKeys.get(partitionPath).add(key);
}
}
@@ -253,11 +251,11 @@ public class TestHoodieClientBase extends HoodieClientTestHarness {
* @param keyGenFunction Keys Generation function
* @return Wrapped function
*/
private Function2<List<HoodieKey>, String, Integer> wrapDeleteKeysGenFunctionForPreppedCalls(
final HoodieWriteConfig writeConfig, final Function2<List<HoodieKey>, String, Integer> keyGenFunction) {
return (commit, numRecords) -> {
private Function<Integer, List<HoodieKey>> wrapDeleteKeysGenFunctionForPreppedCalls(
final HoodieWriteConfig writeConfig, final Function<Integer, List<HoodieKey>> keyGenFunction) {
return (numRecords) -> {
final HoodieIndex index = HoodieIndex.createIndex(writeConfig, jsc);
List<HoodieKey> records = keyGenFunction.apply(commit, numRecords);
List<HoodieKey> records = keyGenFunction.apply(numRecords);
final HoodieTableMetaClient metaClient = new HoodieTableMetaClient(jsc.hadoopConfiguration(), basePath, true);
HoodieTable table = HoodieTable.getHoodieTable(metaClient, writeConfig, jsc);
JavaRDD<HoodieRecord> recordsToDelete = jsc.parallelize(records, 1)
@@ -292,8 +290,8 @@ public class TestHoodieClientBase extends HoodieClientTestHarness {
* @param wrapped Actual Records Generation function
* @return Wrapped Function
*/
Function2<List<HoodieKey>, String, Integer> generateWrapDeleteKeysFn(boolean isPreppedAPI,
HoodieWriteConfig writeConfig, Function2<List<HoodieKey>, String, Integer> wrapped) {
Function<Integer, List<HoodieKey>> generateWrapDeleteKeysFn(boolean isPreppedAPI,
HoodieWriteConfig writeConfig, Function<Integer, List<HoodieKey>> wrapped) {
if (isPreppedAPI) {
return wrapDeleteKeysGenFunctionForPreppedCalls(writeConfig, wrapped);
} else {
@@ -381,7 +379,7 @@ public class TestHoodieClientBase extends HoodieClientTestHarness {
int numRecordsInThisCommit,
Function3<JavaRDD<WriteStatus>, HoodieWriteClient, JavaRDD<HoodieKey>, String> deleteFn, boolean isPreppedAPI,
boolean assertForCommit, int expRecordsInThisCommit, int expTotalRecords) throws Exception {
final Function2<List<HoodieKey>, String, Integer> keyGenFunction =
final Function<Integer, List<HoodieKey>> keyGenFunction =
generateWrapDeleteKeysFn(isPreppedAPI, writeConfig, dataGen::generateUniqueDeletes);
return deleteBatch(client, newCommitTime, prevCommitTime, initCommitTime, numRecordsInThisCommit,
@@ -476,14 +474,14 @@ public class TestHoodieClientBase extends HoodieClientTestHarness {
*/
JavaRDD<WriteStatus> deleteBatch(HoodieWriteClient client, String newCommitTime, String prevCommitTime,
String initCommitTime, int numRecordsInThisCommit,
Function2<List<HoodieKey>, String, Integer> keyGenFunction,
Function<Integer, List<HoodieKey>> keyGenFunction,
Function3<JavaRDD<WriteStatus>, HoodieWriteClient, JavaRDD<HoodieKey>, String> deleteFn,
boolean assertForCommit, int expRecordsInThisCommit, int expTotalRecords) throws Exception {
// Delete 1 (only deletes)
client.startCommitWithTime(newCommitTime);
List<HoodieKey> keysToDelete = keyGenFunction.apply(newCommitTime, numRecordsInThisCommit);
List<HoodieKey> keysToDelete = keyGenFunction.apply(numRecordsInThisCommit);
JavaRDD<HoodieKey> deleteRecords = jsc.parallelize(keysToDelete, 1);
JavaRDD<WriteStatus> result = deleteFn.apply(client, deleteRecords, newCommitTime);
@@ -533,37 +531,6 @@ public class TestHoodieClientBase extends HoodieClientTestHarness {
return hoodieCleanStatsTwo.stream().filter(e -> e.getPartitionPath().equals(partitionPath)).findFirst().orElse(null);
}
/**
* Utility to simulate commit touching files in a partition.
*
* @param files List of file-Ids to be touched
* @param partitionPath Partition
* @param commitTime Commit Timestamp
* @throws IOException in case of error
*/
void updateAllFilesInPartition(List<String> files, String partitionPath, String commitTime) throws IOException {
for (String fileId : files) {
HoodieTestUtils.createDataFile(basePath, partitionPath, commitTime, fileId);
}
}
/**
* Helper methods to create new data files in a partition.
*
* @param partitionPath Partition
* @param commitTime Commit Timestamp
* @param numFiles Number of files to be added
* @return Created files
* @throws IOException in case of error
*/
List<String> createFilesInPartition(String partitionPath, String commitTime, int numFiles) throws IOException {
List<String> files = new ArrayList<>();
for (int i = 0; i < numFiles; i++) {
files.add(HoodieTestUtils.createNewDataFile(basePath, partitionPath, commitTime));
}
return files;
}
// Functional Interfaces for passing lambda and Hoodie Write API contexts
@FunctionalInterface

8
hudi-client/src/test/java/org/apache/hudi/TestHoodieClientOnCopyOnWriteStorage.java
View File

@@ -237,7 +237,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends TestHoodieClientBase {
* @return Hoodie Write Client
* @throws Exception in case of error
*/
private HoodieWriteClient getWriteClientWithDummyIndex(final boolean isGlobal) throws Exception {
private HoodieWriteClient getWriteClientWithDummyIndex(final boolean isGlobal) {
HoodieIndex index = mock(HoodieIndex.class);
when(index.isGlobal()).thenReturn(isGlobal);
return getHoodieWriteClient(getConfigBuilder().build(), false, index);
@@ -637,7 +637,6 @@ public class TestHoodieClientOnCopyOnWriteStorage extends TestHoodieClientBase {
public void testDeletesWithDeleteApi() throws Exception {
final String testPartitionPath = "2016/09/26";
final int insertSplitLimit = 100;
List<String> keysSoFar = new ArrayList<>();
// setup the small file handling params
HoodieWriteConfig config = getSmallInsertWriteConfig(insertSplitLimit); // hold upto 200 records max
dataGen = new HoodieTestDataGenerator(new String[] {testPartitionPath});
@@ -649,7 +648,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends TestHoodieClientBase {
client.startCommitWithTime(commitTime1);
List<HoodieRecord> inserts1 = dataGen.generateInserts(commitTime1, insertSplitLimit); // this writes ~500kb
Set<String> keys1 = HoodieClientTestUtils.getRecordKeys(inserts1);
keysSoFar.addAll(keys1);
List<String> keysSoFar = new ArrayList<>(keys1);
JavaRDD<HoodieRecord> insertRecordsRDD1 = jsc.parallelize(inserts1, 1);
List<WriteStatus> statuses = client.upsert(insertRecordsRDD1, commitTime1).collect();
@@ -763,7 +762,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends TestHoodieClientBase {
* Test delete with delete api.
*/
@Test
public void testDeletesWithoutInserts() throws Exception {
public void testDeletesWithoutInserts() {
final String testPartitionPath = "2016/09/26";
final int insertSplitLimit = 100;
// setup the small file handling params
@@ -844,7 +843,6 @@ public class TestHoodieClientOnCopyOnWriteStorage extends TestHoodieClientBase {
HoodieWriteConfig cfg = getConfigBuilder().withAutoCommit(false).build();
HoodieWriteClient client = getHoodieWriteClient(cfg);
HoodieTableMetaClient metaClient = new HoodieTableMetaClient(jsc.hadoopConfiguration(), basePath);
HoodieTable table = HoodieTable.getHoodieTable(metaClient, cfg, jsc);
String commitTime = "000";
client.startCommitWithTime(commitTime);

7
hudi-client/src/test/java/org/apache/hudi/TestHoodieReadClient.java
View File

@@ -23,7 +23,6 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.spark.api.java.JavaRDD;
import org.junit.Assert;
import org.junit.Test;
import java.util.Arrays;
@@ -31,7 +30,7 @@ import java.util.Collection;
import java.util.List;
import java.util.stream.Collectors;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertEquals;
@SuppressWarnings("unchecked")
/**
@@ -93,7 +92,7 @@ public class TestHoodieReadClient extends TestHoodieClientBase {
JavaRDD<HoodieRecord> filteredRDD = readClient.filterExists(recordsRDD);
// Should not find any files
assertTrue(filteredRDD.collect().size() == 100);
assertEquals(100, filteredRDD.collect().size());
JavaRDD<HoodieRecord> smallRecordsRDD = jsc.parallelize(records.subList(0, 75), 1);
// We create three parquet file, each having one record. (3 different partitions)
@@ -105,7 +104,7 @@ public class TestHoodieReadClient extends TestHoodieClientBase {
filteredRDD = anotherReadClient.filterExists(recordsRDD);
List<HoodieRecord> result = filteredRDD.collect();
// Check results
Assert.assertEquals(25, result.size());
assertEquals(25, result.size());
}
}
}

4
hudi-client/src/test/java/org/apache/hudi/TestWriteStatus.java
View File

@@ -23,14 +23,12 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.junit.Test;
import org.mockito.Mockito;
import java.io.IOException;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class TestWriteStatus {
@Test
public void testFailureFraction() throws IOException {
public void testFailureFraction() {
WriteStatus status = new WriteStatus(true, 0.1);
Throwable t = new Exception("some error in writing");
for (int i = 0; i < 1000; i++) {

2
hudi-client/src/test/java/org/apache/hudi/common/HoodieClientTestUtils.java
View File

@@ -164,7 +164,7 @@ public class HoodieClientTestUtils {
String commitTime) {
HoodieInstant commitInstant = new HoodieInstant(false, HoodieTimeline.COMMIT_ACTION, commitTime);
if (!commitTimeline.containsInstant(commitInstant)) {
new HoodieException("No commit exists at " + commitTime);
throw new HoodieException("No commit exists at " + commitTime);
}
try {
HashMap<String, String> paths =

3
hudi-client/src/test/java/org/apache/hudi/common/HoodieMergeOnReadTestUtils.java
View File

@@ -45,8 +45,7 @@ import java.util.stream.Collectors;
*/
public class HoodieMergeOnReadTestUtils {
public static List<GenericRecord> getRecordsUsingInputFormat(List<String> inputPaths, String basePath)
throws IOException {
public static List<GenericRecord> getRecordsUsingInputFormat(List<String> inputPaths, String basePath) {
JobConf jobConf = new JobConf();
Schema schema = HoodieAvroUtils.addMetadataFields(
new Schema.Parser().parse(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA));

22
hudi-client/src/test/java/org/apache/hudi/common/HoodieTestDataGenerator.java
View File

@@ -24,7 +24,6 @@ import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodiePartitionMetadata;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTestUtils;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
@@ -121,7 +120,7 @@ public class HoodieTestDataGenerator {
/**
* Generates a new avro record of the above schema format, retaining the key if optionally provided.
*/
public static HoodieAvroPayload generateAvroPayload(HoodieKey key, String commitTime) throws IOException {
public static HoodieAvroPayload generateAvroPayload(HoodieKey key, String commitTime) {
GenericRecord rec = generateGenericRecord(key.getRecordKey(), "rider-" + commitTime, "driver-" + commitTime, 0.0);
return new HoodieAvroPayload(Option.of(rec));
}
@@ -141,10 +140,6 @@ public class HoodieTestDataGenerator {
return rec;
}
public static void createCommitFile(String basePath, String commitTime) throws IOException {
createCommitFile(basePath, commitTime, HoodieTestUtils.getDefaultHadoopConf());
}
public static void createCommitFile(String basePath, String commitTime, Configuration configuration) {
Arrays.asList(HoodieTimeline.makeCommitFileName(commitTime), HoodieTimeline.makeInflightCommitFileName(commitTime),
HoodieTimeline.makeRequestedCommitFileName(commitTime)).forEach(f -> {
@@ -213,7 +208,7 @@ public class HoodieTestDataGenerator {
/**
* Generates new inserts, uniformly across the partition paths above. It also updates the list of existing keys.
*/
public List<HoodieRecord> generateInserts(String commitTime, Integer n) throws IOException {
public List<HoodieRecord> generateInserts(String commitTime, Integer n) {
return generateInsertsStream(commitTime, n).collect(Collectors.toList());
}
@@ -249,7 +244,7 @@ public class HoodieTestDataGenerator {
return copy;
}
public List<HoodieRecord> generateInsertsWithHoodieAvroPayload(String commitTime, int limit) throws IOException {
public List<HoodieRecord> generateInsertsWithHoodieAvroPayload(String commitTime, int limit) {
List<HoodieRecord> inserts = new ArrayList<>();
int currSize = getNumExistingKeys();
for (int i = 0; i < limit; i++) {
@@ -267,8 +262,7 @@ public class HoodieTestDataGenerator {
return inserts;
}
public List<HoodieRecord> generateUpdatesWithHoodieAvroPayload(String commitTime, List<HoodieRecord> baseRecords)
throws IOException {
public List<HoodieRecord> generateUpdatesWithHoodieAvroPayload(String commitTime, List<HoodieRecord> baseRecords) {
List<HoodieRecord> updates = new ArrayList<>();
for (HoodieRecord baseRecord : baseRecords) {
HoodieRecord record = new HoodieRecord(baseRecord.getKey(), generateAvroPayload(baseRecord.getKey(), commitTime));
@@ -365,12 +359,11 @@ public class HoodieTestDataGenerator {
/**
* Generates deduped delete of keys previously inserted, randomly distributed across the keys above.
*
* @param commitTime Commit Timestamp
* @param n Number of unique records
* @return list of hoodie record updates
*/
public List<HoodieKey> generateUniqueDeletes(String commitTime, Integer n) {
return generateUniqueDeleteStream(commitTime, n).collect(Collectors.toList());
public List<HoodieKey> generateUniqueDeletes(Integer n) {
return generateUniqueDeleteStream(n).collect(Collectors.toList());
}
/**
@@ -407,11 +400,10 @@ public class HoodieTestDataGenerator {
/**
* Generates deduped delete of keys previously inserted, randomly distributed across the keys above.
*
* @param commitTime Commit Timestamp
* @param n Number of unique records
* @return stream of hoodie record updates
*/
public Stream<HoodieKey> generateUniqueDeleteStream(String commitTime, Integer n) {
public Stream<HoodieKey> generateUniqueDeleteStream(Integer n) {
final Set<KeyPartition> used = new HashSet<>();
if (n > numExistingKeys) {

2
hudi-client/src/test/java/org/apache/hudi/func/TestBoundedInMemoryExecutor.java
View File

@@ -56,7 +56,7 @@ public class TestBoundedInMemoryExecutor extends HoodieClientTestHarness {
}
@Test
public void testExecutor() throws Exception {
public void testExecutor() {
final List<HoodieRecord> hoodieRecords = dataGen.generateInserts(commitTime, 100);

2
hudi-client/src/test/java/org/apache/hudi/func/TestBoundedInMemoryQueue.java
View File

@@ -221,7 +221,7 @@ public class TestBoundedInMemoryQueue extends HoodieClientTestHarness {
new BoundedInMemoryQueue(memoryLimitInBytes, getTransformFunction(HoodieTestDataGenerator.avroSchema));
// Produce
Future<Boolean> resFuture = executorService.submit(() -> {
executorService.submit(() -> {
new IteratorBasedQueueProducer<>(hoodieRecords.iterator()).produce(queue);
return true;
});

2
hudi-client/src/test/java/org/apache/hudi/func/TestUpdateMapFunction.java
View File

@@ -35,7 +35,6 @@ import org.apache.hudi.io.HoodieCreateHandle;
import org.apache.hudi.io.HoodieMergeHandle;
import org.apache.hudi.table.HoodieCopyOnWriteTable;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
@@ -102,7 +101,6 @@ public class TestUpdateMapFunction extends HoodieClientTestHarness {
// Now try an update with an evolved schema
// Evolved schema does not have guarantee on preserving the original field ordering
final HoodieWriteConfig config2 = makeHoodieClientConfig("/exampleEvolvedSchema.txt");
final Schema schema = new Schema.Parser().parse(config2.getSchema());
final WriteStatus insertResult = statuses.get(0);
String fileId = insertResult.getFileId();

28
hudi-client/src/test/java/org/apache/hudi/index/TestHbaseIndex.java
View File

@@ -62,6 +62,7 @@ import java.util.List;
import scala.Tuple2;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.anyObject;
@@ -80,7 +81,7 @@ public class TestHbaseIndex extends HoodieClientTestHarness {
private static Configuration hbaseConfig;
private static String tableName = "test_table";
public TestHbaseIndex() throws Exception {}
public TestHbaseIndex() {}
@AfterClass
public static void clean() throws Exception {
@@ -155,10 +156,10 @@ public class TestHbaseIndex extends HoodieClientTestHarness {
metaClient = HoodieTableMetaClient.reload(metaClient);
hoodieTable = HoodieTable.getHoodieTable(metaClient, config, jsc);
javaRDD = index.tagLocation(writeRecords, jsc, hoodieTable);
assertTrue(javaRDD.filter(record -> record.isCurrentLocationKnown()).collect().size() == 200);
assertTrue(javaRDD.map(record -> record.getKey().getRecordKey()).distinct().count() == 200);
assertTrue(javaRDD.filter(record -> (record.getCurrentLocation() != null
&& record.getCurrentLocation().getInstantTime().equals(newCommitTime))).distinct().count() == 200);
assertEquals(200, javaRDD.filter(record -> record.isCurrentLocationKnown()).collect().size());
assertEquals(200, javaRDD.map(record -> record.getKey().getRecordKey()).distinct().count());
assertEquals(200, javaRDD.filter(record -> (record.getCurrentLocation() != null
&& record.getCurrentLocation().getInstantTime().equals(newCommitTime))).distinct().count());
}
}
@@ -194,10 +195,10 @@ public class TestHbaseIndex extends HoodieClientTestHarness {
metaClient = HoodieTableMetaClient.reload(metaClient);
hoodieTable = HoodieTable.getHoodieTable(metaClient, config, jsc);
JavaRDD<HoodieRecord> javaRDD = index.tagLocation(writeRecords, jsc, hoodieTable);
assertTrue(javaRDD.filter(record -> record.isCurrentLocationKnown()).collect().size() == 10);
assertTrue(javaRDD.map(record -> record.getKey().getRecordKey()).distinct().count() == 10);
assertTrue(javaRDD.filter(record -> (record.getCurrentLocation() != null
&& record.getCurrentLocation().getInstantTime().equals(newCommitTime))).distinct().count() == 10);
assertEquals(10, javaRDD.filter(HoodieRecord::isCurrentLocationKnown).collect().size());
assertEquals(10, javaRDD.map(record -> record.getKey().getRecordKey()).distinct().count());
assertEquals(10, javaRDD.filter(record -> (record.getCurrentLocation() != null
&& record.getCurrentLocation().getInstantTime().equals(newCommitTime))).distinct().count());
}
@Test
@@ -211,7 +212,6 @@ public class TestHbaseIndex extends HoodieClientTestHarness {
List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 200);
JavaRDD<HoodieRecord> writeRecords = jsc.parallelize(records, 1);
metaClient = HoodieTableMetaClient.reload(metaClient);
HoodieTable hoodieTable = HoodieTable.getHoodieTable(metaClient, config, jsc);
// Insert 200 records
JavaRDD<WriteStatus> writeStatues = writeClient.upsert(writeRecords, newCommitTime);
@@ -219,13 +219,13 @@ public class TestHbaseIndex extends HoodieClientTestHarness {
// commit this upsert
writeClient.commit(newCommitTime, writeStatues);
hoodieTable = HoodieTable.getHoodieTable(metaClient, config, jsc);
HoodieTable hoodieTable = HoodieTable.getHoodieTable(metaClient, config, jsc);
// Now tagLocation for these records, hbaseIndex should tag them
JavaRDD<HoodieRecord> javaRDD = index.tagLocation(writeRecords, jsc, hoodieTable);
assert (javaRDD.filter(record -> record.isCurrentLocationKnown()).collect().size() == 200);
assert (javaRDD.filter(HoodieRecord::isCurrentLocationKnown).collect().size() == 200);
// check tagged records are tagged with correct fileIds
List<String> fileIds = writeStatues.map(status -> status.getFileId()).collect();
List<String> fileIds = writeStatues.map(WriteStatus::getFileId).collect();
assert (javaRDD.filter(record -> record.getCurrentLocation().getFileId() == null).collect().size() == 0);
List<String> taggedFileIds = javaRDD.map(record -> record.getCurrentLocation().getFileId()).distinct().collect();
@@ -238,7 +238,7 @@ public class TestHbaseIndex extends HoodieClientTestHarness {
// Now tagLocation for these records, hbaseIndex should not tag them since it was a rolled
// back commit
javaRDD = index.tagLocation(writeRecords, jsc, hoodieTable);
assert (javaRDD.filter(record -> record.isCurrentLocationKnown()).collect().size() == 0);
assert (javaRDD.filter(HoodieRecord::isCurrentLocationKnown).collect().size() == 0);
assert (javaRDD.filter(record -> record.getCurrentLocation() != null).collect().size() == 0);
}

2
hudi-client/src/test/java/org/apache/hudi/index/TestHoodieIndex.java
View File

@@ -49,7 +49,7 @@ public class TestHoodieIndex extends HoodieClientTestHarness {
}
@Test
public void testCreateIndex() throws Exception {
public void testCreateIndex() {
HoodieWriteConfig.Builder clientConfigBuilder = HoodieWriteConfig.newBuilder();
HoodieIndexConfig.Builder indexConfigBuilder = HoodieIndexConfig.newBuilder();
// Different types

5
hudi-client/src/test/java/org/apache/hudi/index/bloom/TestHoodieBloomIndex.java
View File

@@ -110,12 +110,11 @@ public class TestHoodieBloomIndex extends HoodieClientTestHarness {
}
private HoodieWriteConfig makeConfig() {
HoodieWriteConfig config = HoodieWriteConfig.newBuilder().withPath(basePath)
return HoodieWriteConfig.newBuilder().withPath(basePath)
.withIndexConfig(HoodieIndexConfig.newBuilder().bloomIndexPruneByRanges(rangePruning)
.bloomIndexTreebasedFilter(treeFiltering).bloomIndexBucketizedChecking(bucketizedChecking)
.bloomIndexKeysPerBucket(2).build())
.build();
return config;
}
@Test
@@ -278,7 +277,7 @@ public class TestHoodieBloomIndex extends HoodieClientTestHarness {
}
@Test
public void testTagLocationWithEmptyRDD() throws Exception {
public void testTagLocationWithEmptyRDD() {
// We have some records to be tagged (two different partitions)
JavaRDD<HoodieRecord> recordRDD = jsc.emptyRDD();
// Also create the metadata and config

2
hudi-client/src/test/java/org/apache/hudi/index/bloom/TestHoodieGlobalBloomIndex.java
View File

@@ -61,7 +61,7 @@ public class TestHoodieGlobalBloomIndex extends HoodieClientTestHarness {
private String schemaStr;
private Schema schema;
public TestHoodieGlobalBloomIndex() throws Exception {
public TestHoodieGlobalBloomIndex() {
}
@Before

3
hudi-client/src/test/java/org/apache/hudi/index/bloom/TestKeyRangeLookupTree.java
View File

@@ -28,7 +28,6 @@ import java.util.Random;
import java.util.UUID;
import static junit.framework.TestCase.assertEquals;
import static junit.framework.TestCase.assertTrue;
/**
* Tests {@link KeyRangeLookupTree}.
@@ -152,7 +151,7 @@ public class TestKeyRangeLookupTree {
if (!expectedMatches.containsKey(iStr)) {
assertEquals(Collections.EMPTY_SET, keyRangeLookupTree.getMatchingIndexFiles(iStr));
} else {
assertTrue(expectedMatches.get(iStr).equals(keyRangeLookupTree.getMatchingIndexFiles(iStr)));
assertEquals(expectedMatches.get(iStr), keyRangeLookupTree.getMatchingIndexFiles(iStr));
}
}
}

2
hudi-client/src/test/java/org/apache/hudi/io/TestHoodieCommitArchiveLog.java
View File

@@ -147,7 +147,7 @@ public class TestHoodieCommitArchiveLog extends HoodieClientTestHarness {
HoodieTestUtils.createCleanFiles(metaClient, basePath, "103", dfs.getConf());
HoodieTestUtils.createCleanFiles(metaClient, basePath, "104", dfs.getConf());
HoodieTestUtils.createCleanFiles(metaClient, basePath, "105", dfs.getConf());
HoodieTestUtils.createPendingCleanFiles(metaClient, dfs.getConf(), "106", "107");
HoodieTestUtils.createPendingCleanFiles(metaClient, "106", "107");
// reload the timeline and get all the commmits before archive
timeline = metaClient.getActiveTimeline().reload().getAllCommitsTimeline().filterCompletedInstants();

2
hudi-client/src/test/java/org/apache/hudi/io/TestHoodieMergeHandle.java
View File

@@ -34,7 +34,6 @@ import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieStorageConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.table.HoodieTable;
import org.apache.hadoop.fs.FileSystem;
import org.apache.spark.api.java.JavaRDD;
@@ -250,7 +249,6 @@ public class TestHoodieMergeHandle extends HoodieClientTestHarness {
// Update all the 100 records
metaClient = HoodieTableMetaClient.reload(metaClient);
HoodieTable table = HoodieTable.getHoodieTable(metaClient, config, jsc);
newCommitTime = "101";
writeClient.startCommitWithTime(newCommitTime);

8
hudi-client/src/test/java/org/apache/hudi/io/strategy/TestHoodieCompactionStrategy.java
View File

@@ -87,7 +87,7 @@ public class TestHoodieCompactionStrategy {
assertEquals("BoundedIOCompaction should have resulted in 2 compactions being chosen", 2, returned.size());
// Total size of all the log files
Long returnedSize = returned.stream().map(s -> s.getMetrics().get(BoundedIOCompactionStrategy.TOTAL_IO_MB))
.map(s -> s.longValue()).reduce((size1, size2) -> size1 + size2).orElse(0L);
.map(Double::longValue).reduce(Long::sum).orElse(0L);
assertEquals("Should chose the first 2 compactions which should result in a total IO of 690 MB", 610,
(long) returnedSize);
}
@@ -111,7 +111,7 @@ public class TestHoodieCompactionStrategy {
assertEquals("LogFileSizeBasedCompactionStrategy should have resulted in 1 compaction", 1, returned.size());
// Total size of all the log files
Long returnedSize = returned.stream().map(s -> s.getMetrics().get(BoundedIOCompactionStrategy.TOTAL_IO_MB))
.map(s -> s.longValue()).reduce((size1, size2) -> size1 + size2).orElse(0L);
.map(Double::longValue).reduce(Long::sum).orElse(0L);
assertEquals("Should chose the first 2 compactions which should result in a total IO of 690 MB", 1204,
(long) returnedSize);
}
@@ -227,8 +227,8 @@ public class TestHoodieCompactionStrategy {
private List<HoodieCompactionOperation> createCompactionOperations(HoodieWriteConfig config,
Map<Long, List<Long>> sizesMap) {
Map<Long, String> keyToPartitionMap = sizesMap.entrySet().stream()
.map(e -> Pair.of(e.getKey(), partitionPaths[new Random().nextInt(partitionPaths.length - 1)]))
Map<Long, String> keyToPartitionMap = sizesMap.keySet().stream()
.map(e -> Pair.of(e, partitionPaths[new Random().nextInt(partitionPaths.length - 1)]))
.collect(Collectors.toMap(Pair::getKey, Pair::getValue));
return createCompactionOperations(config, sizesMap, keyToPartitionMap);
}

6
hudi-client/src/test/java/org/apache/hudi/metrics/TestHoodieJmxMetrics.java
View File

@@ -24,7 +24,7 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.junit.Test;
import static org.apache.hudi.metrics.Metrics.registerGauge;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -46,7 +46,7 @@ public class TestHoodieJmxMetrics extends TestHoodieMetrics {
@Test
public void testRegisterGauge() {
registerGauge("jmx_metric", 123L);
assertTrue(Metrics.getInstance().getRegistry().getGauges()
.get("jmx_metric").getValue().toString().equals("123"));
assertEquals("123", Metrics.getInstance().getRegistry().getGauges()
.get("jmx_metric").getValue().toString());
}
}

4
hudi-client/src/test/java/org/apache/hudi/metrics/TestHoodieMetrics.java
View File

@@ -24,7 +24,7 @@ import org.junit.Before;
import org.junit.Test;
import static org.apache.hudi.metrics.Metrics.registerGauge;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -41,6 +41,6 @@ public class TestHoodieMetrics {
@Test
public void testRegisterGauge() {
registerGauge("metric1", 123L);
assertTrue(Metrics.getInstance().getRegistry().getGauges().get("metric1").getValue().toString().equals("123"));
assertEquals("123", Metrics.getInstance().getRegistry().getGauges().get("metric1").getValue().toString());
}
}

9
hudi-client/src/test/java/org/apache/hudi/table/TestCopyOnWriteTable.java
View File

@@ -64,6 +64,7 @@ import java.util.UUID;
import scala.Tuple2;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -224,7 +225,7 @@ public class TestCopyOnWriteTable extends HoodieClientTestHarness {
}
}
}
assertTrue(updatedParquetFile != null);
assertNotNull(updatedParquetFile);
// Check whether the record has been updated
Path updatedParquetFilePath = new Path(updatedParquetFile.getAbsolutePath());
BloomFilter updatedFilter =
@@ -240,16 +241,16 @@ public class TestCopyOnWriteTable extends HoodieClientTestHarness {
ParquetReader updatedReader = ParquetReader.builder(new AvroReadSupport<>(), updatedParquetFilePath).build();
index = 0;
while ((newRecord = (GenericRecord) updatedReader.read()) != null) {
assertTrue(newRecord.get("_row_key").toString().equals(records.get(index).getRecordKey()));
assertEquals(newRecord.get("_row_key").toString(), records.get(index).getRecordKey());
if (index == 0) {
assertTrue(newRecord.get("number").toString().equals("15"));
assertEquals("15", newRecord.get("number").toString());
}
index++;
}
updatedReader.close();
// Also check the numRecordsWritten
WriteStatus writeStatus = statuses.get(0);
assertTrue("Should be only one file generated", statuses.size() == 1);
assertEquals("Should be only one file generated", 1, statuses.size());
assertEquals(4, writeStatus.getStat().getNumWrites());// 3 rewritten records + 1 new record
}

133
hudi-client/src/test/java/org/apache/hudi/table/TestMergeOnReadTable.java
View File

@@ -241,7 +241,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
ReadOptimizedView roView =
new HoodieTableFileSystemView(metaClient, metaClient.getCommitTimeline().filterCompletedInstants(), allFiles);
Stream<HoodieDataFile> dataFilesToRead = roView.getLatestDataFiles();
assertTrue(!dataFilesToRead.findAny().isPresent());
assertFalse(dataFilesToRead.findAny().isPresent());
roView = new HoodieTableFileSystemView(metaClient, hoodieTable.getCompletedCommitsTimeline(), allFiles);
dataFilesToRead = roView.getLatestDataFiles();
@@ -284,7 +284,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
dataFilesToRead = roView.getLatestDataFiles();
assertTrue(dataFilesToRead.findAny().isPresent());
List<String> dataFiles = roView.getLatestDataFiles().map(hf -> hf.getPath()).collect(Collectors.toList());
List<String> dataFiles = roView.getLatestDataFiles().map(HoodieDataFile::getPath).collect(Collectors.toList());
List<GenericRecord> recordsRead = HoodieMergeOnReadTestUtils.getRecordsUsingInputFormat(dataFiles, basePath);
// Wrote 20 records and deleted 20 records, so remaining 20-20 = 0
assertEquals("Must contain 0 records", 0, recordsRead.size());
@@ -343,13 +343,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
new HoodieTableFileSystemView(metaClient, hoodieTable.getCompletedCommitsTimeline(), allFiles);
final String absentCommit = newCommitTime;
assertFalse(roView.getLatestDataFiles().filter(file -> {
if (absentCommit.equals(file.getCommitTime())) {
return true;
} else {
return false;
}
}).findAny().isPresent());
assertFalse(roView.getLatestDataFiles().anyMatch(file -> absentCommit.equals(file.getCommitTime())));
}
}
@@ -407,7 +401,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
copyOfRecords = dataGen.generateUpdates(commitTime1, copyOfRecords);
copyOfRecords.addAll(dataGen.generateInserts(commitTime1, 200));
List<String> dataFiles = roView.getLatestDataFiles().map(hf -> hf.getPath()).collect(Collectors.toList());
List<String> dataFiles = roView.getLatestDataFiles().map(HoodieDataFile::getPath).collect(Collectors.toList());
List<GenericRecord> recordsRead = HoodieMergeOnReadTestUtils.getRecordsUsingInputFormat(dataFiles, basePath);
assertEquals(recordsRead.size(), 200);
@@ -419,9 +413,9 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
secondClient.rollback(commitTime1);
allFiles = HoodieTestUtils.listAllDataFilesInPath(metaClient.getFs(), cfg.getBasePath());
// After rollback, there should be no parquet file with the failed commit time
Assert.assertEquals(Arrays.asList(allFiles).stream()
.filter(file -> file.getPath().getName().contains(commitTime1)).collect(Collectors.toList()).size(), 0);
dataFiles = roView.getLatestDataFiles().map(hf -> hf.getPath()).collect(Collectors.toList());
Assert.assertEquals(Arrays.stream(allFiles)
.filter(file -> file.getPath().getName().contains(commitTime1)).count(), 0);
dataFiles = roView.getLatestDataFiles().map(HoodieDataFile::getPath).collect(Collectors.toList());
recordsRead = HoodieMergeOnReadTestUtils.getRecordsUsingInputFormat(dataFiles, basePath);
assertEquals(recordsRead.size(), 200);
}
@@ -437,7 +431,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
copyOfRecords = dataGen.generateUpdates(commitTime2, copyOfRecords);
copyOfRecords.addAll(dataGen.generateInserts(commitTime2, 200));
List<String> dataFiles = roView.getLatestDataFiles().map(hf -> hf.getPath()).collect(Collectors.toList());
List<String> dataFiles = roView.getLatestDataFiles().map(HoodieDataFile::getPath).collect(Collectors.toList());
List<GenericRecord> recordsRead = HoodieMergeOnReadTestUtils.getRecordsUsingInputFormat(dataFiles, basePath);
assertEquals(recordsRead.size(), 200);
@@ -452,13 +446,13 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
thirdClient.rollback(commitTime2);
allFiles = HoodieTestUtils.listAllDataFilesInPath(metaClient.getFs(), cfg.getBasePath());
// After rollback, there should be no parquet file with the failed commit time
Assert.assertEquals(Arrays.asList(allFiles).stream()
.filter(file -> file.getPath().getName().contains(commitTime2)).collect(Collectors.toList()).size(), 0);
Assert.assertEquals(Arrays.stream(allFiles)
.filter(file -> file.getPath().getName().contains(commitTime2)).count(), 0);
metaClient = HoodieTableMetaClient.reload(metaClient);
hoodieTable = HoodieTable.getHoodieTable(metaClient, cfg, jsc);
roView = new HoodieTableFileSystemView(metaClient, hoodieTable.getCompletedCommitsTimeline(), allFiles);
dataFiles = roView.getLatestDataFiles().map(hf -> hf.getPath()).collect(Collectors.toList());
dataFiles = roView.getLatestDataFiles().map(HoodieDataFile::getPath).collect(Collectors.toList());
recordsRead = HoodieMergeOnReadTestUtils.getRecordsUsingInputFormat(dataFiles, basePath);
// check that the number of records read is still correct after rollback operation
assertEquals(recordsRead.size(), 200);
@@ -470,8 +464,6 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
newCommitTime = "003";
thirdClient.startCommitWithTime(newCommitTime);
records = dataGen.generateUpdates(newCommitTime, records);
writeStatusJavaRDD = thirdClient.upsert(writeRecords, newCommitTime);
thirdClient.commit(newCommitTime, writeStatusJavaRDD);
statuses = writeStatusJavaRDD.collect();
@@ -486,35 +478,20 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
allFiles = HoodieTestUtils.listAllDataFilesInPath(metaClient.getFs(), cfg.getBasePath());
metaClient = HoodieTableMetaClient.reload(metaClient);
hoodieTable = HoodieTable.getHoodieTable(metaClient, cfg, jsc);
roView = new HoodieTableFileSystemView(metaClient, metaClient.getCommitsTimeline(), allFiles);
List<HoodieDataFile> dataFiles2 = roView.getLatestDataFiles().collect(Collectors.toList());
final String compactedCommitTime =
metaClient.getActiveTimeline().reload().getCommitsTimeline().lastInstant().get().getTimestamp();
assertTrue(roView.getLatestDataFiles().filter(file -> {
if (compactedCommitTime.equals(file.getCommitTime())) {
return true;
} else {
return false;
}
}).findAny().isPresent());
assertTrue(roView.getLatestDataFiles().anyMatch(file -> compactedCommitTime.equals(file.getCommitTime())));
thirdClient.rollback(compactedCommitTime);
allFiles = HoodieTestUtils.listAllDataFilesInPath(metaClient.getFs(), cfg.getBasePath());
metaClient = HoodieTableMetaClient.reload(metaClient);
hoodieTable = HoodieTable.getHoodieTable(metaClient, cfg, jsc);
roView = new HoodieTableFileSystemView(metaClient, metaClient.getCommitsTimeline(), allFiles);
assertFalse(roView.getLatestDataFiles().filter(file -> {
if (compactedCommitTime.equals(file.getCommitTime())) {
return true;
} else {
return false;
}
}).findAny().isPresent());
assertFalse(roView.getLatestDataFiles().anyMatch(file -> compactedCommitTime.equals(file.getCommitTime())));
}
}
}
@@ -524,12 +501,10 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
HoodieWriteConfig cfg = getConfig(false);
try (final HoodieWriteClient client = getWriteClient(cfg);) {
List<String> allCommits = new ArrayList<>();
/**
* Write 1 (only inserts)
*/
String newCommitTime = "001";
allCommits.add(newCommitTime);
client.startCommitWithTime(newCommitTime);
List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 200);
@@ -554,7 +529,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
ReadOptimizedView roView =
new HoodieTableFileSystemView(metaClient, metaClient.getCommitTimeline().filterCompletedInstants(), allFiles);
Stream<HoodieDataFile> dataFilesToRead = roView.getLatestDataFiles();
assertTrue(!dataFilesToRead.findAny().isPresent());
assertFalse(dataFilesToRead.findAny().isPresent());
roView = new HoodieTableFileSystemView(metaClient, hoodieTable.getCompletedCommitsTimeline(), allFiles);
dataFilesToRead = roView.getLatestDataFiles();
@@ -565,7 +540,6 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
* Write 2 (inserts + updates)
*/
newCommitTime = "002";
allCommits.add(newCommitTime);
// WriteClient with custom config (disable small file handling)
HoodieWriteClient nClient = getWriteClient(getHoodieWriteConfigWithSmallFileHandlingOff());
nClient.startCommitWithTime(newCommitTime);
@@ -589,7 +563,6 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
* Write 3 (inserts + updates)
*/
newCommitTime = "003";
allCommits.add(newCommitTime);
client.startCommitWithTime(newCommitTime);
List<HoodieRecord> newInserts = dataGen.generateInserts(newCommitTime, 100);
@@ -606,7 +579,6 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
metaClient = HoodieTableMetaClient.reload(metaClient);
String compactionInstantTime = "004";
allCommits.add(compactionInstantTime);
client.scheduleCompactionAtInstant(compactionInstantTime, Option.empty());
// Compaction commit
@@ -614,7 +586,6 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
* Write 4 (updates)
*/
newCommitTime = "005";
allCommits.add(newCommitTime);
client.startCommitWithTime(newCommitTime);
records = dataGen.generateUpdates(newCommitTime, records);
@@ -629,7 +600,6 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
metaClient = HoodieTableMetaClient.reload(metaClient);
compactionInstantTime = "006";
allCommits.add(compactionInstantTime);
client.scheduleCompactionAtInstant(compactionInstantTime, Option.empty());
JavaRDD<WriteStatus> ws = client.compact(compactionInstantTime);
client.commitCompaction(compactionInstantTime, ws, Option.empty());
@@ -641,19 +611,12 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
final String compactedCommitTime =
metaClient.getActiveTimeline().reload().getCommitsTimeline().lastInstant().get().getTimestamp();
assertTrue(roView.getLatestDataFiles().filter(file -> {
if (compactedCommitTime.equals(file.getCommitTime())) {
return true;
} else {
return false;
}
}).findAny().isPresent());
assertTrue(roView.getLatestDataFiles().anyMatch(file -> compactedCommitTime.equals(file.getCommitTime())));
/**
* Write 5 (updates)
*/
newCommitTime = "007";
allCommits.add(newCommitTime);
client.startCommitWithTime(newCommitTime);
copyOfRecords = new ArrayList<>(records);
copyOfRecords = dataGen.generateUpdates(newCommitTime, copyOfRecords);
@@ -673,7 +636,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
roView =
new HoodieTableFileSystemView(metaClient, metaClient.getCommitTimeline().filterCompletedInstants(), allFiles);
dataFilesToRead = roView.getLatestDataFiles();
assertTrue(!dataFilesToRead.findAny().isPresent());
assertFalse(dataFilesToRead.findAny().isPresent());
RealtimeView rtView =
new HoodieTableFileSystemView(metaClient, metaClient.getCommitTimeline().filterCompletedInstants(), allFiles);
List<HoodieFileGroup> fileGroups =
@@ -681,9 +644,9 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
assertTrue(fileGroups.isEmpty());
// make sure there are no log files remaining
assertTrue(((HoodieTableFileSystemView) rtView).getAllFileGroups()
.filter(fileGroup -> fileGroup.getAllRawFileSlices().filter(f -> f.getLogFiles().count() == 0).count() == 0)
.count() == 0L);
assertEquals(0L, ((HoodieTableFileSystemView) rtView).getAllFileGroups()
.filter(fileGroup -> fileGroup.getAllRawFileSlices().noneMatch(f -> f.getLogFiles().count() == 0))
.count());
}
}
@@ -691,10 +654,10 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
protected HoodieWriteConfig getHoodieWriteConfigWithSmallFileHandlingOff() {
return HoodieWriteConfig.newBuilder().withPath(basePath).withSchema(TRIP_EXAMPLE_SCHEMA).withParallelism(2, 2)
.withAutoCommit(false).withAssumeDatePartitioning(true)
.withCompactionConfig(HoodieCompactionConfig.newBuilder().compactionSmallFileSize(1 * 1024)
.withCompactionConfig(HoodieCompactionConfig.newBuilder().compactionSmallFileSize(1024)
.withInlineCompaction(false).withMaxNumDeltaCommitsBeforeCompaction(1).build())
.withEmbeddedTimelineServerEnabled(true)
.withStorageConfig(HoodieStorageConfig.newBuilder().limitFileSize(1 * 1024).build()).forTable("test-trip-table")
.withStorageConfig(HoodieStorageConfig.newBuilder().limitFileSize(1024).build()).forTable("test-trip-table")
.build();
}
@@ -767,10 +730,9 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
Map<String, Long> parquetFileIdToNewSize =
newDataFilesList.stream().collect(Collectors.toMap(HoodieDataFile::getFileId, HoodieDataFile::getFileSize));
assertTrue(parquetFileIdToNewSize.entrySet().stream()
.filter(entry -> parquetFileIdToSize.get(entry.getKey()) < entry.getValue()).count() > 0);
assertTrue(parquetFileIdToNewSize.entrySet().stream().anyMatch(entry -> parquetFileIdToSize.get(entry.getKey()) < entry.getValue()));
List<String> dataFiles = roView.getLatestDataFiles().map(hf -> hf.getPath()).collect(Collectors.toList());
List<String> dataFiles = roView.getLatestDataFiles().map(HoodieDataFile::getPath).collect(Collectors.toList());
List<GenericRecord> recordsRead = HoodieMergeOnReadTestUtils.getRecordsUsingInputFormat(dataFiles, basePath);
// Wrote 20 records in 2 batches
assertEquals("Must contain 40 records", 40, recordsRead.size());
@@ -787,13 +749,11 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 100);
JavaRDD<HoodieRecord> recordsRDD = jsc.parallelize(records, 1);
List<WriteStatus> statuses = writeClient.insert(recordsRDD, newCommitTime).collect();
writeClient.insert(recordsRDD, newCommitTime).collect();
// Update all the 100 records
HoodieTableMetaClient metaClient = new HoodieTableMetaClient(jsc.hadoopConfiguration(), basePath);
HoodieTable table = HoodieTable.getHoodieTable(metaClient, config, jsc);
HoodieTimeline timeline2 = metaClient.getActiveTimeline();
newCommitTime = "101";
writeClient.startCommitWithTime(newCommitTime);
@@ -808,7 +768,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
// Verify that all data file has one log file
metaClient = HoodieTableMetaClient.reload(metaClient);
table = HoodieTable.getHoodieTable(metaClient, config, jsc);
HoodieTable table = HoodieTable.getHoodieTable(metaClient, config, jsc);
// In writeRecordsToLogFiles, no commit files are getting added, so resetting file-system view state
((SyncableFileSystemView) (table.getRTFileSystemView())).reset();
@@ -842,13 +802,10 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
List<FileSlice> groupedLogFiles =
table.getRTFileSystemView().getLatestFileSlices(partitionPath).collect(Collectors.toList());
for (FileSlice slice : groupedLogFiles) {
assertTrue("After compaction there should be no log files visiable on a Realtime view",
slice.getLogFiles().collect(Collectors.toList()).isEmpty());
assertEquals("After compaction there should be no log files visiable on a Realtime view", 0, slice.getLogFiles().count());
}
List<WriteStatus> writeStatuses = result.collect();
assertTrue(writeStatuses.stream()
.filter(writeStatus -> writeStatus.getStat().getPartitionPath().contentEquals(partitionPath))
.count() > 0);
assertTrue(writeStatuses.stream().anyMatch(writeStatus -> writeStatus.getStat().getPartitionPath().contentEquals(partitionPath)));
}
}
}
@@ -874,10 +831,9 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
long numLogFiles = 0;
for (String partitionPath : dataGen.getPartitionPaths()) {
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getDataFile().isPresent()).count() == 0);
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getLogFiles().count() > 0).count() > 0);
assertEquals(0, tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getDataFile().isPresent()).count());
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath).anyMatch(fileSlice -> fileSlice.getLogFiles().count() > 0));
numLogFiles += tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getLogFiles().count() > 0).count();
}
@@ -886,7 +842,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
// Do a compaction
String commitTime = writeClient.scheduleCompaction(Option.empty()).get().toString();
statuses = writeClient.compact(commitTime);
Assert.assertTrue(statuses.map(status -> status.getStat().getPath().contains("parquet")).count() == numLogFiles);
assertEquals(statuses.map(status -> status.getStat().getPath().contains("parquet")).count(), numLogFiles);
Assert.assertEquals(statuses.count(), numLogFiles);
writeClient.commitCompaction(commitTime, statuses, Option.empty());
}
@@ -911,7 +867,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
Assert.assertEquals(
writeStatuses.stream().filter(writeStatus -> !writeStatus.getStat().getPath().contains("log")).count(), 0);
Assert.assertTrue(
writeStatuses.stream().filter(writeStatus -> writeStatus.getStat().getPath().contains("log")).count() > 0);
writeStatuses.stream().anyMatch(writeStatus -> writeStatus.getStat().getPath().contains("log")));
// rollback a failed commit
boolean rollback = writeClient.rollback(newCommitTime);
@@ -951,14 +907,12 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
long numLogFiles = 0;
for (String partitionPath : dataGen.getPartitionPaths()) {
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getDataFile().isPresent()).count() == 0);
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getLogFiles().count() > 0).count() == 0);
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath).noneMatch(fileSlice -> fileSlice.getDataFile().isPresent()));
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath).noneMatch(fileSlice -> fileSlice.getLogFiles().count() > 0));
numLogFiles += tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getLogFiles().count() > 0).count();
}
Assert.assertTrue(numLogFiles == 0);
assertEquals(0, numLogFiles);
metaClient.getFs().copyFromLocalFile(new Path(file.getAbsolutePath()),
new Path(metaClient.getMetaPath(), fileName));
Thread.sleep(1000);
@@ -990,10 +944,8 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
long numLogFiles = 0;
for (String partitionPath : dataGen.getPartitionPaths()) {
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getDataFile().isPresent()).count() == 0);
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getLogFiles().count() > 0).count() > 0);
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath).noneMatch(fileSlice -> fileSlice.getDataFile().isPresent()));
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath).anyMatch(fileSlice -> fileSlice.getLogFiles().count() > 0));
numLogFiles += tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getLogFiles().count() > 0).count();
}
@@ -1003,7 +955,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
newCommitTime = writeClient.scheduleCompaction(Option.empty()).get().toString();
statuses = writeClient.compact(newCommitTime);
// Ensure all log files have been compacted into parquet files
Assert.assertTrue(statuses.map(status -> status.getStat().getPath().contains("parquet")).count() == numLogFiles);
assertEquals(statuses.map(status -> status.getStat().getPath().contains("parquet")).count(), numLogFiles);
Assert.assertEquals(statuses.count(), numLogFiles);
writeClient.commitCompaction(newCommitTime, statuses, Option.empty());
// Trigger a rollback of compaction
@@ -1014,10 +966,8 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
Option<HoodieInstant> lastInstant = ((SyncableFileSystemView) tableRTFileSystemView).getLastInstant();
System.out.println("Last Instant =" + lastInstant);
for (String partitionPath : dataGen.getPartitionPaths()) {
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getDataFile().isPresent()).count() == 0);
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath)
.filter(fileSlice -> fileSlice.getLogFiles().count() > 0).count() > 0);
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath).noneMatch(fileSlice -> fileSlice.getDataFile().isPresent()));
Assert.assertTrue(tableRTFileSystemView.getLatestFileSlices(partitionPath).anyMatch(fileSlice -> fileSlice.getLogFiles().count() > 0));
}
}
}
@@ -1134,7 +1084,6 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
HoodieTableMetaClient metaClient = new HoodieTableMetaClient(jsc.hadoopConfiguration(), basePath);
Map<String, Long> fileIdToInsertsMap = new HashMap<>();
Map<String, Long> fileIdToUpsertsMap = new HashMap<>();
HoodieTable table = HoodieTable.getHoodieTable(metaClient, cfg, jsc);
String commitTime = "000";
client.startCommitWithTime(commitTime);
@@ -1146,7 +1095,7 @@ public class TestMergeOnReadTable extends HoodieClientTestHarness {
assertTrue("Commit should succeed", client.commit(commitTime, statuses));
// Read from commit file
table = HoodieTable.getHoodieTable(metaClient, cfg, jsc);
HoodieTable table = HoodieTable.getHoodieTable(metaClient, cfg, jsc);
HoodieCommitMetadata metadata = HoodieCommitMetadata.fromBytes(
table.getActiveTimeline()
.getInstantDetails(table.getActiveTimeline().getDeltaCommitTimeline().lastInstant().get()).get(),