danny0405 commented on code in PR #19293:
URL: https://github.com/apache/hudi/pull/19293#discussion_r3592766421
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java:
##########
@@ -123,10 +123,9 @@ private boolean isSmallFile(FileSlice fileSlice) {
}
private double logFileToParquetCompressionRatio() {
- if (config.getLogDataBlockFormat().isPresent()
- && config.getLogDataBlockFormat().get() ==
HoodieLogBlock.HoodieLogBlockType.PARQUET_DATA_BLOCK) {
- return 1D;
+ if (config.getWriteVersion().lesserThan(HoodieTableVersion.TEN)) {
+ return config.getLogFileToParquetCompressionRatio();
}
- return config.getLogFileToParquetCompressionRatio();
+ return 1D;
Review Comment:
I aligned the per-file Spark and Flink paths: all callers now pass the write
config to `FileSlice`, which applies the ratio only to inline Avro logs and
uses physical size for native and inline non-Avro logs. There is no
table-version gate in that per-file path, so upgraded slices with legacy inline
Avro logs behave consistently. The version gate remains only for delta commit
metadata because it does not identify log-file formats individually.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java:
##########
@@ -123,10 +125,10 @@ private boolean isSmallFile(FileSlice fileSlice) {
}
private double logFileToParquetCompressionRatio() {
Review Comment:
This method is still used by `averageBytesPerRecord()` when calling
`calculateRecordSizeThroughCommitMetadata(deltaCommitTimeline,
logFileToParquetCompressionRatio())`. The config-based `FileSlice` path is
authoritative for per-file sizing, while this helper is retained only for delta
commit metadata, which cannot distinguish inline from native log files. I added
a comment to make that distinction explicit.
##########
hudi-common/src/main/java/org/apache/hudi/common/model/FileSlice.java:
##########
@@ -193,24 +196,42 @@ public boolean isEmpty() {
}
/**
- * Get the total file size of a file slice similar on the base file.
- * For the log file, we need to convert its size to the estimated size
similar on the base file in a certain proportion
+ * Gets the estimated total size of a file slice in the base Parquet format.
+ *
+ * <p>Only inline Avro log files need size calibration. Native log files and
inline non-Avro log files
+ * use their physical sizes directly.</p>
+ *
+ * @param writeConfig write config containing the log data block format and
log-to-Parquet compression ratio
*/
- public long getTotalFileSizeAsParquetFormat(double logFileFraction) {
- long logFileSize =
convertLogFilesSizeToExpectedParquetSize(logFileFraction);
+ public long getTotalFileSizeAsParquetFormat(HoodieConfig writeConfig) {
+ long logFileSize = convertLogFilesSizeToExpectedParquetSize(writeConfig);
return getBaseFile().isPresent() ? getBaseFile().get().getFileSize() +
logFileSize : logFileSize;
}
- private long convertLogFilesSizeToExpectedParquetSize(double
logFileFraction) {
- long totalSizeOfLogFiles =
+ private long convertLogFilesSizeToExpectedParquetSize(HoodieConfig
writeConfig) {
+ long totalSizeOfInlineLogFiles =
+ logFiles.stream()
+ .filter(logFile -> !logFile.isNativeLogFile())
+ .map(HoodieLogFile::getFileSize)
+ .filter(size -> size > 0)
+ .reduce(Long::sum)
+ .orElse(0L);
+ long totalSizeOfNativeLogFiles =
logFiles.stream()
+ .filter(HoodieLogFile::isNativeLogFile)
.map(HoodieLogFile::getFileSize)
.filter(size -> size > 0)
.reduce(Long::sum)
.orElse(0L);
- // Here we assume that if there is no base parquet file, all log files
contain only inserts.
+ // Here we assume that if there is no base parquet file, all calibrated
log files contain only inserts.
// We can then just get the parquet equivalent size of these log files,
compare that with
// {@link config.getParquetMaxFileSize()} and decide if there is scope to
insert more rows
- return (long) (totalSizeOfLogFiles * logFileFraction);
+ boolean isAvroDataBlocks =
HoodieLogBlockType.fromId(writeConfig.getStringOrDefault(
Review Comment:
Yes, this is an accepted limitation of the common-model path. `FileSlice`
cannot consult the table-config fallback, so the restored config-based API
reads the explicitly supplied inline block format and preserves Avro as the
legacy default when it is unset. I added a comment documenting that assumption.
Native log files are still identified from their file names and always use
their physical size.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java:
##########
@@ -123,10 +123,9 @@ private boolean isSmallFile(FileSlice fileSlice) {
}
private double logFileToParquetCompressionRatio() {
- if (config.getLogDataBlockFormat().isPresent()
- && config.getLogDataBlockFormat().get() ==
HoodieLogBlock.HoodieLogBlockType.PARQUET_DATA_BLOCK) {
- return 1D;
+ if (config.getWriteVersion().lesserThan(HoodieTableVersion.TEN)) {
Review Comment:
I added a comment documenting that the commit-metadata version gate assumes
the write version has been reconciled with the table version. The per-file
`FileSlice` path intentionally has no version gate: it distinguishes native and
inline files directly, so legacy inline Avro logs in an upgraded v10 slice are
still calibrated. The version gate now applies only to delta commit metadata,
where that per-file distinction is unavailable.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java:
##########
@@ -123,10 +123,9 @@ private boolean isSmallFile(FileSlice fileSlice) {
}
private double logFileToParquetCompressionRatio() {
- if (config.getLogDataBlockFormat().isPresent()
- && config.getLogDataBlockFormat().get() ==
HoodieLogBlock.HoodieLogBlockType.PARQUET_DATA_BLOCK) {
- return 1D;
+ if (config.getWriteVersion().lesserThan(HoodieTableVersion.TEN)) {
+ return config.getLogFileToParquetCompressionRatio();
Review Comment:
Yes. I restored the effective block-type guard in
`logFileToParquetCompressionRatio()` using
`CommonClientUtils.getLogBlockType(...)`, so the ratio is returned only for
pre-v10 Avro log blocks. I also restored
`testDeltaWriteProfileRecordsPerBucketSkipsCompressionRatioForParquetLogBlocks`
to cover the v9 Parquet case.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/DeltaWriteProfile.java:
##########
@@ -123,10 +123,9 @@ private boolean isSmallFile(FileSlice fileSlice) {
}
private double logFileToParquetCompressionRatio() {
- if (config.getLogDataBlockFormat().isPresent()
- && config.getLogDataBlockFormat().get() ==
HoodieLogBlock.HoodieLogBlockType.PARQUET_DATA_BLOCK) {
- return 1D;
+ if (config.getWriteVersion().lesserThan(HoodieTableVersion.TEN)) {
Review Comment:
Good catch. I restored the write config parameter on
`FileSlice.getTotalFileSizeAsParquetFormat` so it can inspect the configured
inline block format. It now calibrates only inline Avro logs; inline
Parquet/HFile and native logs use their physical size. I also restored
`testDeltaWriteProfileRecordsPerBucketSkipsCompressionRatioForParquetLogBlocks`
and added FileSlice coverage for Avro, Parquet, and HFile.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
