This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit d994e06674d594eab5d701b361541373241f6f76 Author: voonhous <[email protected]> AuthorDate: Tue Jun 16 11:52:34 2026 +0800 perf(io): Derive log file size from AppendResult on append-handle close (#19002) * perf(io): Derive log file size from AppendResult on append-handle close HoodieAppendHandle.close() called storage.getPathInfo(<logFile>).getLength() for every WriteStatus to record the final log file size -- a remote HEAD per log file per file group on object stores, purely to read a size the handle already knows. (cherry picked from commit e1193c385be66f46921e2e6ed3c45d4d5f54366a) --- .../org/apache/hudi/io/HoodieAppendHandle.java | 16 +++--- .../hudi/common/table/log/TestLogReaderUtils.java | 58 ++++++++++++++++++++++ 2 files changed, 67 insertions(+), 7 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java index 3b825bbf1ade..b6389d9122f0 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java @@ -558,14 +558,16 @@ public class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O writer = null; } - // update final size, once for all log files - // TODO we can actually deduce file size purely from AppendResult (based on offset and size - // of the appended block) + // Set the final on-disk size of each log file. Appends within an append handle are contiguous, + // so a log file's length equals its start offset plus the total bytes appended to it. That is + // exactly what fs.getFileStatus().getLength() returns, and both values are already captured by + // the AppendResult stats (logOffset and the accumulated fileSizeInBytes). Deriving the size this + // way avoids a getPathInfo/HEAD per log file, which is a remote round trip per file group on + // object stores. for (WriteStatus status : statuses) { - long logFileSize = storage.getPathInfo( - new StoragePath(config.getBasePath(), status.getStat().getPath())) - .getLength(); - status.getStat().setFileSizeInBytes(logFileSize); + HoodieDeltaWriteStat stat = (HoodieDeltaWriteStat) status.getStat(); + long appendedBytes = stat.getFileSizeInBytes(); + stat.setFileSizeInBytes(stat.getLogOffset() + appendedBytes); } // generate Secondary index stats if streaming writes is enabled. diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java index 2ddffc27d875..f1a3f877ab1b 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java @@ -43,6 +43,8 @@ import org.apache.hudi.testutils.SparkClientFunctionalTestHarness; import org.apache.spark.api.java.JavaRDD; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.util.Arrays; import java.util.List; @@ -50,6 +52,7 @@ import java.util.Map; import java.util.Properties; import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA; +import static org.apache.hudi.testutils.Assertions.assertFileSizesEqual; import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -131,6 +134,61 @@ public class TestLogReaderUtils extends SparkClientFunctionalTestHarness { } } + @ParameterizedTest + @ValueSource(ints = {6, 9}) // SIX appends to the existing log file (logOffset > 0); NINE writes fresh files (logOffset == 0) + public void testLogFileWriteStatSizeMatchesOnDisk(int writeTableVersion) throws Exception { + // HoodieAppendHandle derives each log file's on-disk size from the AppendResult + // (logOffset + accumulated appended bytes) instead of a getPathInfo per file. Validate that the + // derived size in the write stat matches the actual on-disk log file length. + // + // The "logOffset +" term only contributes when a delta commit appends to a pre-existing log file: + // - table version >= EIGHT (e.g. NINE): each delta commit writes a fresh instant-named log file + // from offset 0, so logOffset is always 0; + // - table version SIX: the second upsert appends to the first commit's log file, so logOffset + // is > 0 and the derived sum is actually exercised. + // Running both covers the derivation with and without a non-zero offset. + Properties props = new Properties(); + props.setProperty(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), String.valueOf(writeTableVersion)); + HoodieTableMetaClient metaClient = getHoodieMetaClient( + storageConf(), basePath(), props, HoodieTableType.MERGE_ON_READ); + + HoodieWriteConfig config = getConfigBuilder(true) + .withPath(basePath()) + .withWriteTableVersion(writeTableVersion) + .withAutoUpgradeVersion(false) + .withCompactionConfig(HoodieCompactionConfig.newBuilder() + .withInlineCompaction(false) + .compactionSmallFileSize(0) + .build()) + .build(); + + HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator(); + + try (SparkRDDWriteClient client = getHoodieWriteClient(config)) { + // First commit - insert data (base files) + String firstCommit = "001"; + WriteClientTestUtils.startCommitWithTime(client, firstCommit); + JavaRDD<WriteStatus> insertRdd = client.insert(jsc().parallelize(dataGen.generateInserts(firstCommit, 100), 1), firstCommit); + assertNoWriteErrors(insertRdd.collect()); + client.commit(firstCommit, insertRdd); + + // Two upsert commits. Under version SIX the second commit appends to the first's log file + // (logOffset > 0); under version >= EIGHT each commit writes a fresh log file (logOffset == 0). + for (String commitTime : new String[] {"002", "003"}) { + WriteClientTestUtils.startCommitWithTime(client, commitTime); + JavaRDD<WriteStatus> upsertRdd = client.upsert(jsc().parallelize(dataGen.generateUpdates(commitTime, 50), 1), commitTime); + List<WriteStatus> statuses = upsertRdd.collect(); + assertNoWriteErrors(statuses); + assertLogFilesProduced(statuses); + client.commit(commitTime, upsertRdd); + // Derived log file size (logOffset + appended bytes) must equal the actual on-disk length + assertFileSizesEqual(statuses, status -> FSUtils.getFileSize( + metaClient.getStorage(), + new StoragePath(config.getBasePath(), status.getStat().getPath()))); + } + } + } + @Test public void testGetAllLogFilesWithMaxCommitEmptyPartitions() throws Exception { HoodieTableMetaClient metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, new Properties());
