Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[HUDI-5336] Fixing log file pattern match to ignore extraneous files #7612

Merged
merged 5 commits into from
Jan 20, 2023
Merged

[HUDI-5336] Fixing log file pattern match to ignore extraneous files #7612

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
82e6b63
Fixing log file pattern match to ignore extraneous files
nsivabalan Jan 5, 2023
66370e1
minor fixes
nsivabalan Jan 6, 2023
1dc0a07
addressing comments
nsivabalan Jan 18, 2023
18c64ac
addressing more comments
nsivabalan Jan 19, 2023
3ea3fec
Fix nit
yihua Jan 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 39 additions & 18 deletions hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,13 @@ public class FSUtils {
// Log files are of this pattern - .b5068208-e1a4-11e6-bf01-fe55135034f3_20170101134598.log.1_1-0-1
// Archive log files are of this pattern - .commits_.archive.1_1-0-1
public static final Pattern LOG_FILE_PATTERN =
Pattern.compile("\\.(.+)_(.*)\\.(.+)\\.(\\d+)(_((\\d+)-(\\d+)-(\\d+))(.cdc)?)?");
Pattern.compile("^\\.(.+)_(.*)\\.(log)\\.(\\d+)(_((\\d+)-(\\d+)-(\\d+))(.cdc)?)?");
public static final String ARCHIVE_STR = "archive";
public static final String COMMITS_STR = "commits";
public static final String EMPTY_STR = "";
public static final String ARCHIVED_LOG_PREFIX = "." + COMMITS_STR + "_." + ARCHIVE_STR;
public static final Pattern ARCHIVED_LOG_FILE_PATTERN =
Pattern.compile(ARCHIVED_LOG_PREFIX + ".(\\d+)(_((\\d+)-(\\d+)-(\\d+)))");
private static final int MAX_ATTEMPTS_RECOVER_LEASE = 10;
private static final long MIN_CLEAN_TO_KEEP = 10;
private static final long MIN_ROLLBACK_TO_KEEP = 10;
Expand Down Expand Up @@ -358,23 +364,27 @@ public static String createNewFileId(String idPfx, int id) {
* Get the file extension from the log file.
*/
public static String getFileExtensionFromLog(Path logPath) {
Matcher matcher = LOG_FILE_PATTERN.matcher(logPath.getName());
boolean isArchivedLog = logPath.getName().contains(ARCHIVED_LOG_PREFIX);
Matcher matcher = isArchivedLog ? ARCHIVED_LOG_FILE_PATTERN.matcher(logPath.getName()) :
LOG_FILE_PATTERN.matcher(logPath.getName());
if (!matcher.find()) {
throw new InvalidHoodiePathException(logPath, "LogFile");
}
return matcher.group(3);
return isArchivedLog ? ARCHIVE_STR : matcher.group(3);
nsivabalan marked this conversation as resolved.
Show resolved Hide resolved
}

/**
* Get the first part of the file name in the log file. That will be the fileId. Log file do not have instantTime in
* the file name.
*/
public static String getFileIdFromLogPath(Path path) {
Matcher matcher = LOG_FILE_PATTERN.matcher(path.getName());
boolean isArchivedLog = path.getName().contains(ARCHIVED_LOG_PREFIX);
Matcher matcher = isArchivedLog ? ARCHIVED_LOG_FILE_PATTERN.matcher(path.getName())
: LOG_FILE_PATTERN.matcher(path.getName());
if (!matcher.find()) {
throw new InvalidHoodiePathException(path, "LogFile");
}
return matcher.group(1);
return isArchivedLog ? COMMITS_STR : matcher.group(1);
nsivabalan marked this conversation as resolved.
Show resolved Hide resolved
}

/**
Expand All @@ -392,57 +402,66 @@ public static String getFileIdFromFilePath(Path filePath) {
* the file name.
*/
public static String getBaseCommitTimeFromLogPath(Path path) {
Matcher matcher = LOG_FILE_PATTERN.matcher(path.getName());
boolean isArchivedLog = path.getName().contains(ARCHIVED_LOG_PREFIX);
Matcher matcher = isArchivedLog ? ARCHIVED_LOG_FILE_PATTERN.matcher(path.getName()) : LOG_FILE_PATTERN.matcher(path.getName());
if (!matcher.find()) {
throw new InvalidHoodiePathException(path, "LogFile");
}
return matcher.group(2);
return isArchivedLog ? EMPTY_STR : matcher.group(2);
}

/**
* Get TaskPartitionId used in log-path.
*/
public static Integer getTaskPartitionIdFromLogPath(Path path) {
Matcher matcher = LOG_FILE_PATTERN.matcher(path.getName());
boolean isArchivedLog = path.getName().contains(ARCHIVED_LOG_PREFIX);
Matcher matcher = isArchivedLog ? ARCHIVED_LOG_FILE_PATTERN.matcher(path.getName()) :
LOG_FILE_PATTERN.matcher(path.getName());
if (!matcher.find()) {
throw new InvalidHoodiePathException(path, "LogFile");
}
String val = matcher.group(7);
String val = isArchivedLog ? matcher.group(4) : matcher.group(7);
return val == null ? null : Integer.parseInt(val);
}

/**
* Get Write-Token used in log-path.
*/
public static String getWriteTokenFromLogPath(Path path) {
Matcher matcher = LOG_FILE_PATTERN.matcher(path.getName());
boolean isArchivedLog = path.getName().contains(ARCHIVED_LOG_PREFIX);
Matcher matcher = isArchivedLog ? ARCHIVED_LOG_FILE_PATTERN.matcher(path.getName()) :
LOG_FILE_PATTERN.matcher(path.getName());
if (!matcher.find()) {
throw new InvalidHoodiePathException(path, "LogFile");
}
return matcher.group(6);
return isArchivedLog ? matcher.group(3) : matcher.group(6);
}

/**
* Get StageId used in log-path.
*/
public static Integer getStageIdFromLogPath(Path path) {
Matcher matcher = LOG_FILE_PATTERN.matcher(path.getName());
boolean isArchivedLog = path.getName().contains(ARCHIVED_LOG_PREFIX);
Matcher matcher = isArchivedLog ? ARCHIVED_LOG_FILE_PATTERN.matcher(path.getName()) :
LOG_FILE_PATTERN.matcher(path.getName());
if (!matcher.find()) {
throw new InvalidHoodiePathException(path, "LogFile");
}
String val = matcher.group(8);
String val = isArchivedLog ? matcher.group(5) : matcher.group(8);
return val == null ? null : Integer.parseInt(val);
}

/**
* Get Task Attempt Id used in log-path.
*/
public static Integer getTaskAttemptIdFromLogPath(Path path) {
Matcher matcher = LOG_FILE_PATTERN.matcher(path.getName());
boolean isArchivedLog = path.getName().contains(ARCHIVED_LOG_PREFIX);
Matcher matcher = isArchivedLog ? ARCHIVED_LOG_FILE_PATTERN.matcher(path.getName()) :
LOG_FILE_PATTERN.matcher(path.getName());
if (!matcher.find()) {
throw new InvalidHoodiePathException(path, "LogFile");
}
String val = matcher.group(9);
String val = isArchivedLog ? matcher.group(6) : matcher.group(9);
return val == null ? null : Integer.parseInt(val);
}

Expand All @@ -454,11 +473,13 @@ public static int getFileVersionFromLog(Path logPath) {
}

public static int getFileVersionFromLog(String logFileName) {
Matcher matcher = LOG_FILE_PATTERN.matcher(logFileName);
boolean isArchivedLog = logFileName.contains(ARCHIVED_LOG_PREFIX);
Matcher matcher = isArchivedLog ? ARCHIVED_LOG_FILE_PATTERN.matcher(logFileName) :
LOG_FILE_PATTERN.matcher(logFileName);
if (!matcher.find()) {
throw new HoodieIOException("Invalid log file name: " + logFileName);
}
return Integer.parseInt(matcher.group(4));
return Integer.parseInt(isArchivedLog ? matcher.group(1) : matcher.group(4));
}

public static String makeLogFileName(String fileId, String logFileExtension, String baseCommitTime, int version,
Expand All @@ -479,7 +500,7 @@ public static boolean isLogFile(Path logPath) {
}

public static boolean isLogFile(String fileName) {
Matcher matcher = LOG_FILE_PATTERN.matcher(fileName);
Matcher matcher = fileName.contains(ARCHIVED_LOG_PREFIX) ? ARCHIVED_LOG_FILE_PATTERN.matcher(fileName) : LOG_FILE_PATTERN.matcher(fileName);
return matcher.find() && fileName.contains(".log");
}

Expand Down
2 changes: 1 addition & 1 deletion hudi-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -343,7 +343,7 @@ public void testAppendNotSupported(@TempDir java.nio.file.Path tempDir) throws I

for (int i = 0; i < 2; i++) {
Writer writer = HoodieLogFormat.newWriterBuilder().onParentPath(testPath)
.withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION).withFileId("commits.archive").overBaseCommit("")
.withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION).withFileId("commits").overBaseCommit("")
.withFs(localFs).build();
writer.appendBlock(dataBlock);
writer.close();
Expand Down
4 changes: 2 additions & 2 deletions ...mon/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormatAppendFailure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ public void testFailedToGetAppendStreamFromHDFSNameNode()
HoodieAvroDataBlock dataBlock = new HoodieAvroDataBlock(records, header, HoodieRecord.RECORD_KEY_METADATA_FIELD);

Writer writer = HoodieLogFormat.newWriterBuilder().onParentPath(testPath)
.withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION).withFileId("commits.archive")
.withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION).withFileId("commits")
.overBaseCommit("").withFs(fs).build();

writer.appendBlock(dataBlock);
Expand Down Expand Up @@ -134,7 +134,7 @@ public void testFailedToGetAppendStreamFromHDFSNameNode()
// Opening a new Writer right now will throw IOException. The code should handle this, rollover the logfile and
// return a new writer with a bumped up logVersion
writer = HoodieLogFormat.newWriterBuilder().onParentPath(testPath)
.withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION).withFileId("commits.archive")
.withFileExtension(HoodieArchivedLogFile.ARCHIVE_EXTENSION).withFileId("commits")
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

does this mean the original tests do not properly construct the archived log file name?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes.

.overBaseCommit("").withFs(fs).build();
header = new HashMap<>();
header.put(HoodieLogBlock.HeaderMetadataType.COMMAND_BLOCK_TYPE,
Expand Down
2 changes: 1 addition & 1 deletion ...common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -306,7 +306,7 @@ protected void testInvalidLogFiles() throws Exception {
String fileName2 =
FSUtils.makeLogFileName(fileId, HoodieLogFile.DELTA_EXTENSION, instantTime1, 1, TEST_WRITE_TOKEN);
// create a dummy log file mimicing cloud stores marker files
String fileName3 = "_DUMMY_" + fileName1.substring(1, fileName1.length());
String fileName3 = "_GCS_SYNCABLE_TEMPFILE_" + fileName1;
nsivabalan marked this conversation as resolved.
Show resolved Hide resolved
// this file should not be deduced as a log file.

Paths.get(basePath, partitionPath, fileName1).toFile().createNewFile();
Expand Down