lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-781] Introduce HoodieTestTable for test preparation (#1871)

Browse Source
This commit is contained in:
Raymond Xu
2020-08-10 18:44:03 -07:00
committed by GitHub
parent 934f00b689
commit b2e703d442
19 changed files with 645 additions and 470 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java Normal file
View File

@@ -0,0 +1,113 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.hudi.common.testutils;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
public class FileCreateUtils {
private static void createMetaFile(String basePath, String instantTime, String suffix) throws IOException {
Path parentPath = Paths.get(basePath, HoodieTableMetaClient.METAFOLDER_NAME);
Files.createDirectories(parentPath);
Path metaFilePath = parentPath.resolve(instantTime + suffix);
if (Files.notExists(metaFilePath)) {
Files.createFile(metaFilePath);
}
}
public static void createCommit(String basePath, String instantTime) throws IOException {
createMetaFile(basePath, instantTime, HoodieTimeline.COMMIT_EXTENSION);
}
public static void createRequestedCommit(String basePath, String instantTime) throws IOException {
createMetaFile(basePath, instantTime, HoodieTimeline.REQUESTED_COMMIT_EXTENSION);
}
public static void createInflightCommit(String basePath, String instantTime) throws IOException {
createMetaFile(basePath, instantTime, HoodieTimeline.INFLIGHT_COMMIT_EXTENSION);
}
public static void createDeltaCommit(String basePath, String instantTime) throws IOException {
createMetaFile(basePath, instantTime, HoodieTimeline.DELTA_COMMIT_EXTENSION);
}
public static void createRequestedDeltaCommit(String basePath, String instantTime) throws IOException {
createMetaFile(basePath, instantTime, HoodieTimeline.REQUESTED_DELTA_COMMIT_EXTENSION);
}
public static void createInflightDeltaCommit(String basePath, String instantTime) throws IOException {
createMetaFile(basePath, instantTime, HoodieTimeline.INFLIGHT_DELTA_COMMIT_EXTENSION);
}
public static void createDataFile(String basePath, String partitionPath, String instantTime, String fileId)
throws Exception {
createDataFile(basePath, partitionPath, instantTime, fileId, 0);
}
public static void createDataFile(String basePath, String partitionPath, String instantTime, String fileId, long length)
throws Exception {
Path parentPath = Paths.get(basePath, partitionPath);
Files.createDirectories(parentPath);
Path dataFilePath = parentPath.resolve(FSUtils.makeDataFileName(instantTime, "1-0-1", fileId));
if (Files.notExists(dataFilePath)) {
Files.createFile(dataFilePath);
}
new RandomAccessFile(dataFilePath.toFile(), "rw").setLength(length);
}
public static void createLogFile(String basePath, String partitionPath, String baseInstantTime, String fileId, int version)
throws Exception {
createLogFile(basePath, partitionPath, baseInstantTime, fileId, version, 0);
}
public static void createLogFile(String basePath, String partitionPath, String baseInstantTime, String fileId, int version, int length)
throws Exception {
Path parentPath = Paths.get(basePath, partitionPath);
Files.createDirectories(parentPath);
Path logFilePath = parentPath.resolve(FSUtils.makeLogFileName(fileId, HoodieFileFormat.HOODIE_LOG.getFileExtension(), baseInstantTime, version, "1-0-1"));
if (Files.notExists(logFilePath)) {
Files.createFile(logFilePath);
}
new RandomAccessFile(logFilePath.toFile(), "rw").setLength(length);
}
public static String createMarkerFile(String basePath, String partitionPath, String instantTime, String fileID, IOType ioType)
throws IOException {
Path folderPath = Paths.get(basePath, HoodieTableMetaClient.TEMPFOLDER_NAME, instantTime, partitionPath);
Files.createDirectories(folderPath);
String markerFileName = String.format("%s_%s_%s%s%s.%s", fileID, "1-0-1", instantTime,
HoodieFileFormat.PARQUET.getFileExtension(), HoodieTableMetaClient.MARKER_EXTN, ioType);
Path markerFilePath = folderPath.resolve(markerFileName);
if (Files.notExists(markerFilePath)) {
Files.createFile(markerFilePath);
}
return markerFilePath.toAbsolutePath().toString();
}
}

232
hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java Normal file
View File

@@ -0,0 +1,232 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.hudi.common.testutils;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.IOException;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.stream.IntStream;
import static org.apache.hudi.common.testutils.FileCreateUtils.createCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createDeltaCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightDeltaCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createMarkerFile;
import static org.apache.hudi.common.testutils.FileCreateUtils.createRequestedCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createRequestedDeltaCommit;
public class HoodieTestTable {
private final String basePath;
private final FileSystem fs;
private HoodieTableMetaClient metaClient;
private String currentInstantTime;
private HoodieTestTable(String basePath, FileSystem fs, HoodieTableMetaClient metaClient) {
ValidationUtils.checkArgument(Objects.equals(basePath, metaClient.getBasePath()));
ValidationUtils.checkArgument(Objects.equals(fs, metaClient.getRawFs()));
this.basePath = basePath;
this.fs = fs;
this.metaClient = metaClient;
}
public static HoodieTestTable of(HoodieTableMetaClient metaClient) {
return new HoodieTestTable(metaClient.getBasePath(), metaClient.getRawFs(), metaClient);
}
public HoodieTestTable addRequestedCommit(String instantTime) throws Exception {
createRequestedCommit(basePath, instantTime);
currentInstantTime = instantTime;
metaClient = HoodieTableMetaClient.reload(metaClient);
return this;
}
public HoodieTestTable addRequestedDeltaCommit(String instantTime) throws Exception {
createRequestedDeltaCommit(basePath, instantTime);
currentInstantTime = instantTime;
metaClient = HoodieTableMetaClient.reload(metaClient);
return this;
}
public HoodieTestTable addInflightCommit(String instantTime) throws Exception {
createRequestedCommit(basePath, instantTime);
createInflightCommit(basePath, instantTime);
currentInstantTime = instantTime;
metaClient = HoodieTableMetaClient.reload(metaClient);
return this;
}
public HoodieTestTable addInflightDeltaCommit(String instantTime) throws Exception {
createRequestedDeltaCommit(basePath, instantTime);
createInflightDeltaCommit(basePath, instantTime);
currentInstantTime = instantTime;
metaClient = HoodieTableMetaClient.reload(metaClient);
return this;
}
public HoodieTestTable addCommit(String instantTime) throws Exception {
createRequestedCommit(basePath, instantTime);
createInflightCommit(basePath, instantTime);
createCommit(basePath, instantTime);
currentInstantTime = instantTime;
metaClient = HoodieTableMetaClient.reload(metaClient);
return this;
}
public HoodieTestTable addDeltaCommit(String instantTime) throws Exception {
createRequestedDeltaCommit(basePath, instantTime);
createInflightDeltaCommit(basePath, instantTime);
createDeltaCommit(basePath, instantTime);
currentInstantTime = instantTime;
metaClient = HoodieTableMetaClient.reload(metaClient);
return this;
}
public HoodieTestTable forCommit(String instantTime) {
currentInstantTime = instantTime;
return this;
}
public HoodieTestTable forDeltaCommit(String instantTime) {
currentInstantTime = instantTime;
return this;
}
public HoodieTestTable withMarkerFile(String partitionPath, IOType ioType) throws IOException {
return withMarkerFile(partitionPath, UUID.randomUUID().toString(), ioType);
}
public HoodieTestTable withMarkerFile(String partitionPath, String fileId, IOType ioType) throws IOException {
createMarkerFile(basePath, partitionPath, currentInstantTime, fileId, ioType);
return this;
}
public HoodieTestTable withMarkerFiles(String partitionPath, int num, IOType ioType) throws IOException {
String[] fileIds = IntStream.range(0, num).mapToObj(i -> UUID.randomUUID().toString()).toArray(String[]::new);
return withMarkerFiles(partitionPath, fileIds, ioType);
}
public HoodieTestTable withMarkerFiles(String partitionPath, String[] fileIds, IOType ioType) throws IOException {
for (String fileId : fileIds) {
createMarkerFile(basePath, partitionPath, currentInstantTime, fileId, ioType);
}
return this;
}
/**
* Insert one base file to each of the given distinct partitions.
*
* @return A {@link Map} of partition and its newly inserted file's id.
*/
public Map<String, String> withInserts(String... partitions) throws Exception {
Map<String, String> partitionFileIdMap = new HashMap<>();
for (String p : partitions) {
String fileId = UUID.randomUUID().toString();
FileCreateUtils.createDataFile(basePath, p, currentInstantTime, fileId);
partitionFileIdMap.put(p, fileId);
}
return partitionFileIdMap;
}
public HoodieTestTable withUpdates(String partition, String... fileIds) throws Exception {
for (String f : fileIds) {
FileCreateUtils.createDataFile(basePath, partition, currentInstantTime, f);
}
return this;
}
public String withLogFile(String partitionPath) throws Exception {
String fileId = UUID.randomUUID().toString();
withLogFile(partitionPath, fileId);
return fileId;
}
public HoodieTestTable withLogFile(String partitionPath, String fileId) throws Exception {
return withLogFile(partitionPath, fileId, 0);
}
public HoodieTestTable withLogFile(String partitionPath, String fileId, int version) throws Exception {
FileCreateUtils.createLogFile(basePath, partitionPath, currentInstantTime, fileId, version);
return this;
}
public boolean filesExist(Map<String, String> partitionAndFileId, String instantTime) {
return partitionAndFileId.entrySet().stream().allMatch(entry -> {
String partition = entry.getKey();
String fileId = entry.getValue();
return fileExists(partition, instantTime, fileId);
});
}
public boolean filesExist(String partition, String instantTime, String... fileIds) {
return Arrays.stream(fileIds).allMatch(f -> fileExists(partition, instantTime, f));
}
public boolean fileExists(String partition, String instantTime, String fileId) {
try {
return fs.exists(new Path(Paths.get(basePath, partition,
FSUtils.makeDataFileName(instantTime, "1-0-1", fileId)).toString()));
} catch (IOException e) {
throw new HoodieTestTableException(e);
}
}
public boolean logFilesExist(String partition, String instantTime, String fileId, int... versions) {
return Arrays.stream(versions).allMatch(v -> logFileExists(partition, instantTime, fileId, v));
}
public boolean logFileExists(String partition, String instantTime, String fileId, int version) {
try {
return fs.exists(new Path(Paths.get(basePath, partition,
FSUtils.makeLogFileName(fileId, HoodieFileFormat.HOODIE_LOG.getFileExtension(), instantTime, version, "1-0-1")).toString()));
} catch (IOException e) {
throw new HoodieTestTableException(e);
}
}
public List<FileStatus> listAllFiles(String partitionPath) throws IOException {
return FileSystemTestUtils.listRecursive(fs, new Path(Paths.get(basePath, partitionPath).toString()));
}
public List<FileStatus> listAllFilesInTempFolder() throws IOException {
return FileSystemTestUtils.listRecursive(fs, new Path(Paths.get(basePath, HoodieTableMetaClient.TEMPFOLDER_NAME).toString()));
}
public static class HoodieTestTableException extends RuntimeException {
public HoodieTestTableException(Throwable t) {
super(t);
}
}
}

102
hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java
View File

@@ -99,7 +99,6 @@ import static org.junit.jupiter.api.Assertions.fail;
*/
public class HoodieTestUtils {
public static final String TEST_EXTENSION = ".test";
public static final String RAW_TRIPS_TEST_NAME = "raw_trips";
public static final String DEFAULT_WRITE_TOKEN = "1-0-1";
public static final int DEFAULT_LOG_VERSION = 1;
@@ -138,7 +137,7 @@ public class HoodieTestUtils {
}
public static HoodieTableMetaClient init(Configuration hadoopConf, String basePath, HoodieTableType tableType,
String tableName)
String tableName)
throws IOException {
Properties properties = new Properties();
properties.setProperty(HoodieTableConfig.HOODIE_TABLE_NAME_PROP_NAME, tableName);
@@ -146,7 +145,7 @@ public class HoodieTestUtils {
}
public static HoodieTableMetaClient init(Configuration hadoopConf, String basePath, HoodieTableType tableType,
HoodieFileFormat baseFileFormat)
HoodieFileFormat baseFileFormat)
throws IOException {
Properties properties = new Properties();
properties.setProperty(HoodieTableConfig.HOODIE_BASE_FILE_FORMAT_PROP_NAME, baseFileFormat.toString());
@@ -154,7 +153,7 @@ public class HoodieTestUtils {
}
public static HoodieTableMetaClient init(Configuration hadoopConf, String basePath, HoodieTableType tableType,
Properties properties)
Properties properties)
throws IOException {
properties.putIfAbsent(HoodieTableConfig.HOODIE_TABLE_NAME_PROP_NAME, RAW_TRIPS_TEST_NAME);
properties.putIfAbsent(HoodieTableConfig.HOODIE_TABLE_TYPE_PROP_NAME, tableType.name());
@@ -166,6 +165,9 @@ public class HoodieTestUtils {
return COMMIT_FORMATTER.format(new Date());
}
/**
* @deprecated Use {@link HoodieTestTable} instead.
*/
public static void createCommitFiles(String basePath, String... instantTimes) throws IOException {
for (String instantTime : instantTimes) {
new File(
@@ -176,20 +178,6 @@ public class HoodieTestUtils {
+ HoodieTimeline.makeInflightCommitFileName(instantTime)).createNewFile();
new File(
basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/" + HoodieTimeline.makeCommitFileName(instantTime))
.createNewFile();
}
}
public static void createDeltaCommitFiles(String basePath, String... instantTimes) throws IOException {
for (String instantTime : instantTimes) {
new File(
basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/"
+ HoodieTimeline.makeRequestedDeltaFileName(instantTime)).createNewFile();
new File(
basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/"
+ HoodieTimeline.makeInflightDeltaFileName(instantTime)).createNewFile();
new File(
basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/" + HoodieTimeline.makeDeltaFileName(instantTime))
.createNewFile();
}
}
@@ -198,6 +186,9 @@ public class HoodieTestUtils {
new File(basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME).mkdirs();
}
/**
* @deprecated Use {@link HoodieTestTable} instead.
*/
public static void createInflightCommitFiles(String basePath, String... instantTimes) throws IOException {
for (String instantTime : instantTimes) {
@@ -211,11 +202,12 @@ public class HoodieTestUtils {
public static void createPendingCleanFiles(HoodieTableMetaClient metaClient, String... instantTimes) {
for (String instantTime : instantTimes) {
Arrays.asList(HoodieTimeline.makeRequestedCleanerFileName(instantTime),
HoodieTimeline.makeInflightCleanerFileName(instantTime)).forEach(f -> {
HoodieTimeline.makeInflightCleanerFileName(instantTime))
.forEach(f -> {
FSDataOutputStream os = null;
try {
Path commitFile = new Path(
metaClient.getBasePath() + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/" + f);
Path commitFile = new Path(Paths
.get(metaClient.getBasePath(), HoodieTableMetaClient.METAFOLDER_NAME, f).toString());
os = metaClient.getFs().create(commitFile, true);
// Write empty clean metadata
os.write(TimelineMetadataUtils.serializeCleanerPlan(
@@ -237,11 +229,12 @@ public class HoodieTestUtils {
public static void createCorruptedPendingCleanFiles(HoodieTableMetaClient metaClient, String commitTime) {
Arrays.asList(HoodieTimeline.makeRequestedCleanerFileName(commitTime),
HoodieTimeline.makeInflightCleanerFileName(commitTime)).forEach(f -> {
HoodieTimeline.makeInflightCleanerFileName(commitTime))
.forEach(f -> {
FSDataOutputStream os = null;
try {
Path commitFile = new Path(
metaClient.getBasePath() + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/" + f);
Path commitFile = new Path(Paths
.get(metaClient.getBasePath(), HoodieTableMetaClient.METAFOLDER_NAME, f).toString());
os = metaClient.getFs().create(commitFile, true);
// Write empty clean metadata
os.write(new byte[0]);
@@ -259,18 +252,18 @@ public class HoodieTestUtils {
});
}
public static String createNewDataFile(String basePath, String partitionPath, String instantTime)
throws IOException {
String fileID = UUID.randomUUID().toString();
return createDataFile(basePath, partitionPath, instantTime, fileID);
}
/**
* @deprecated Use {@link HoodieTestTable} instead.
*/
public static String createNewDataFile(String basePath, String partitionPath, String instantTime, long length)
throws IOException {
String fileID = UUID.randomUUID().toString();
return createDataFileFixLength(basePath, partitionPath, instantTime, fileID, length);
}
/**
* @deprecated Use {@link HoodieTestTable} instead.
*/
public static String createDataFile(String basePath, String partitionPath, String instantTime, String fileID)
throws IOException {
String folderPath = basePath + "/" + partitionPath + "/";
@@ -279,7 +272,7 @@ public class HoodieTestUtils {
return fileID;
}
public static String createDataFileFixLength(String basePath, String partitionPath, String instantTime, String fileID,
private static String createDataFileFixLength(String basePath, String partitionPath, String instantTime, String fileID,
long length) throws IOException {
String folderPath = basePath + "/" + partitionPath + "/";
Files.createDirectories(Paths.get(folderPath));
@@ -291,6 +284,9 @@ public class HoodieTestUtils {
return fileID;
}
/**
* @deprecated Use {@link HoodieTestTable} instead.
*/
public static String createNewLogFile(FileSystem fs, String basePath, String partitionPath, String instantTime,
String fileID, Option<Integer> version) throws IOException {
String folderPath = basePath + "/" + partitionPath + "/";
@@ -307,17 +303,6 @@ public class HoodieTestUtils {
return fileID;
}
public static void createCompactionCommitFiles(FileSystem fs, String basePath, String... instantTimes)
throws IOException {
for (String instantTime : instantTimes) {
boolean createFile = fs.createNewFile(new Path(basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/"
+ HoodieTimeline.makeCommitFileName(instantTime)));
if (!createFile) {
throw new IOException("cannot create commit file for commit " + instantTime);
}
}
}
public static void createCompactionRequest(HoodieTableMetaClient metaClient, String instant,
List<Pair<String, FileSlice>> fileSliceList) throws IOException {
HoodieCompactionPlan plan = CompactionUtils.buildFromFileSlices(fileSliceList, Option.empty(), Option.empty());
@@ -326,10 +311,16 @@ public class HoodieTestUtils {
TimelineMetadataUtils.serializeCompactionPlan(plan));
}
/**
* @deprecated Use {@link HoodieTestTable} instead.
*/
public static String getDataFilePath(String basePath, String partitionPath, String instantTime, String fileID) {
return basePath + "/" + partitionPath + "/" + FSUtils.makeDataFileName(instantTime, DEFAULT_WRITE_TOKEN, fileID);
}
/**
* @deprecated Use {@link HoodieTestTable} instead.
*/
public static String getLogFilePath(String basePath, String partitionPath, String instantTime, String fileID,
Option<Integer> version) {
return basePath + "/" + partitionPath + "/" + FSUtils.makeLogFileName(fileID, ".log", instantTime,
@@ -340,36 +331,43 @@ public class HoodieTestUtils {
return basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/" + instantTime + HoodieTimeline.COMMIT_EXTENSION;
}
/**
* @deprecated Use {@link HoodieTestTable} instead.
*/
public static String getInflightCommitFilePath(String basePath, String instantTime) {
return basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/" + instantTime
+ HoodieTimeline.INFLIGHT_COMMIT_EXTENSION;
}
/**
* @deprecated Use {@link HoodieTestTable} instead.
*/
public static String getRequestedCompactionFilePath(String basePath, String instantTime) {
return basePath + "/" + HoodieTableMetaClient.AUXILIARYFOLDER_NAME + "/" + instantTime
+ HoodieTimeline.INFLIGHT_COMMIT_EXTENSION;
}
/**
* @deprecated Use {@link HoodieTestTable} instead.
*/
public static boolean doesDataFileExist(String basePath, String partitionPath, String instantTime,
String fileID) {
return new File(getDataFilePath(basePath, partitionPath, instantTime, fileID)).exists();
}
public static boolean doesLogFileExist(String basePath, String partitionPath, String instantTime, String fileID,
Option<Integer> version) {
return new File(getLogFilePath(basePath, partitionPath, instantTime, fileID, version)).exists();
}
public static boolean doesCommitExist(String basePath, String instantTime) {
return new File(
basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/" + instantTime + HoodieTimeline.COMMIT_EXTENSION)
.exists();
.exists();
}
/**
* @deprecated Use {@link HoodieTestTable} instead.
*/
public static boolean doesInflightExist(String basePath, String instantTime) {
return new File(
basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/" + instantTime + HoodieTimeline.INFLIGHT_EXTENSION)
.exists();
.exists();
}
public static void createCleanFiles(HoodieTableMetaClient metaClient, String basePath,
@@ -419,8 +417,8 @@ public class HoodieTestUtils {
Writer logWriter;
try {
logWriter = HoodieLogFormat.newWriterBuilder().onParentPath(new Path(basePath, partitionPath))
.withFileExtension(HoodieLogFile.DELTA_EXTENSION).withFileId(location.getFileId())
.overBaseCommit(location.getInstantTime()).withFs(fs).build();
.withFileExtension(HoodieLogFile.DELTA_EXTENSION).withFileId(location.getFileId())
.overBaseCommit(location.getInstantTime()).withFs(fs).build();
Map<HoodieLogBlock.HeaderMetadataType, String> header = new HashMap<>();
header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, location.getInstantTime());
@@ -474,7 +472,7 @@ public class HoodieTestUtils {
public static FileStatus[] listAllDataFilesAndLogFilesInPath(FileSystem fs, String basePath) throws IOException {
return Stream.concat(Arrays.stream(listAllDataFilesInPath(fs, basePath)), Arrays.stream(listAllLogFilesInPath(fs, basePath)))
.toArray(FileStatus[]::new);
.toArray(FileStatus[]::new);
}
public static List<String> monotonicIncreasingCommitTimestamps(int numTimestamps, int startSecsDelta) {