BsoBird commented on code in PR #10623:
URL: https://github.com/apache/iceberg/pull/10623#discussion_r1669894107
##########
core/src/main/java/org/apache/iceberg/hadoop/HadoopTableOperations.java:
##########
@@ -291,137 +407,171 @@ Path versionHintFile() {
return metadataPath(Util.VERSION_HINT_FILENAME);
}
- private void writeVersionHint(int versionToWrite) {
+ @VisibleForTesting
+ void writeVersionHint(FileSystem fs, Integer versionToWrite) throws
Exception {
Path versionHintFile = versionHintFile();
- FileSystem fs = getFileSystem(versionHintFile, conf);
-
+ Path tempVersionHintFile = metadataPath(UUID.randomUUID() +
"-version-hint.temp");
try {
- Path tempVersionHintFile = metadataPath(UUID.randomUUID().toString() +
"-version-hint.temp");
writeVersionToPath(fs, tempVersionHintFile, versionToWrite);
- fs.delete(versionHintFile, false /* recursive delete */);
fs.rename(tempVersionHintFile, versionHintFile);
} catch (IOException e) {
- LOG.warn("Failed to update version hint", e);
+ // Cleaning up temporary files.
+ if (fs.exists(tempVersionHintFile)) {
+ io().deleteFile(tempVersionHintFile.toString());
+ }
+ throw e;
}
}
- private void writeVersionToPath(FileSystem fs, Path path, int
versionToWrite) throws IOException {
+ @VisibleForTesting
+ boolean nextVersionIsLatest(int nextVersion, int currentMaxVersion) {
+ return nextVersion == (currentMaxVersion + 1);
+ }
+
+ private void writeVersionToPath(FileSystem fs, Path path, int
versionToWrite) {
try (FSDataOutputStream out = fs.create(path, false /* overwrite */)) {
out.write(String.valueOf(versionToWrite).getBytes(StandardCharsets.UTF_8));
+ } catch (IOException e) {
+ throw new RuntimeIOException(e);
}
}
@VisibleForTesting
- int findVersion() {
- Path versionHintFile = versionHintFile();
- FileSystem fs = getFileSystem(versionHintFile, conf);
-
+ int findVersionByUsingVersionHint(FileSystem fs, Path versionHintFile)
throws IOException {
try (InputStreamReader fsr =
new InputStreamReader(fs.open(versionHintFile),
StandardCharsets.UTF_8);
BufferedReader in = new BufferedReader(fsr)) {
return Integer.parseInt(in.readLine().replace("\n", ""));
+ }
+ }
- } catch (Exception e) {
- try {
- if (fs.exists(metadataRoot())) {
- LOG.warn("Error reading version hint file {}", versionHintFile, e);
- } else {
- LOG.debug("Metadata for table not found in directory {}",
metadataRoot(), e);
- return 0;
- }
-
- // List the metadata directory to find the version files, and try to
recover the max
- // available version
- FileStatus[] files =
- fs.listStatus(
- metadataRoot(), name ->
VERSION_PATTERN.matcher(name.getName()).matches());
- int maxVersion = 0;
-
- for (FileStatus file : files) {
- int currentVersion = version(file.getPath().getName());
- if (currentVersion > maxVersion && getMetadataFile(currentVersion)
!= null) {
- maxVersion = currentVersion;
- }
- }
-
- return maxVersion;
- } catch (IOException io) {
- LOG.warn("Error trying to recover version-hint.txt data for {}",
versionHintFile, e);
+ @VisibleForTesting
+ int findVersionWithOutVersionHint(FileSystem fs) {
+ try {
+ if (!fs.exists(metadataRoot())) {
+ // Either the table has just been created, or it has been corrupted,
but either way, we have
+ // to start at version 0.
+ LOG.warn("Metadata for table not found in directory [{}]",
metadataRoot());
return 0;
}
+ // List the metadata directory to find the version files, and try to
recover the max
+ // available version
+ FileStatus[] files =
+ fs.listStatus(metadataRoot(), name ->
VERSION_PATTERN.matcher(name.getName()).matches());
+ int maxVersion = 0;
+ for (FileStatus file : files) {
+ int currentVersion = version(file.getPath().getName());
+ if (currentVersion > maxVersion && getMetadataFile(currentVersion) !=
null) {
+ maxVersion = currentVersion;
+ }
+ }
+ return maxVersion;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @VisibleForTesting
+ int findVersion() {
+ Path versionHintFile = versionHintFile();
+ FileSystem fs = getFileSystem(versionHintFile, conf);
+ try {
+ return fs.exists(versionHintFile)
+ ? findVersionByUsingVersionHint(fs, versionHintFile)
+ : findVersionWithOutVersionHint(fs);
+ } catch (Exception e) {
+ // try one last time
+ return findVersionWithOutVersionHint(fs);
}
}
/**
- * Renames the source file to destination, using the provided file system.
If the rename failed,
- * an attempt will be made to delete the source file.
+ * Renames the source file to destination, using the provided file system.
*
* @param fs the filesystem used for the rename
* @param src the source file
* @param dst the destination file
+ * @return If it returns true, then the commit was successful.
*/
- private void renameToFinal(FileSystem fs, Path src, Path dst, int
nextVersion) {
- try {
- if (!lockManager.acquire(dst.toString(), src.toString())) {
- throw new CommitFailedException(
- "Failed to acquire lock on file: %s with owner: %s", dst, src);
+ @VisibleForTesting
+ boolean commitNewVersion(
+ FileSystem fs, Path src, Path dst, Integer nextVersion, boolean
supportGlobalLocking)
+ throws IOException {
+ if (fs.exists(dst)) {
+ throw new CommitFailedException("Version %d already exists: %s",
nextVersion, dst);
+ }
+ int maxVersion = supportGlobalLocking ? findVersion() :
findVersionWithOutVersionHint(fs);
+ if (!nextVersionIsLatest(nextVersion, maxVersion)) {
+ if (!supportGlobalLocking) {
+ io().deleteFile(versionHintFile().toString());
Review Comment:
For cases where the global locking service is not used, we always try to
remove the wrong versionHintFile before throwing an exception.
Because versionHintFile is unreliable, instead of keeping it, it should be
removed so that all remaining clients use the findVersionWithOutVersionHint
method to get the exact version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
