voonhous commented on code in PR #19293:
URL: https://github.com/apache/hudi/pull/19293#discussion_r3593195104
##########
hudi-common/src/main/java/org/apache/hudi/common/model/FileSlice.java:
##########
@@ -193,24 +196,44 @@ 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);
+ // The effective inline block type cannot be inferred from the log file
name, and hudi-common does not have access
+ // to the table config fallback used by write clients. Legacy callers with
no explicit block format default to Avro.
+ boolean isAvroDataBlocks =
HoodieLogBlockType.fromId(writeConfig.getStringOrDefault(
Review Comment:
`fromId` is a plain map lookup, so a user who sets `AVRO` gets null here and
silently skips calibration. Worth lowercasing the configured value before the
lookup?
##########
hudi-common/src/main/java/org/apache/hudi/common/model/FileSlice.java:
##########
@@ -193,24 +196,44 @@ 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);
+ // The effective inline block type cannot be inferred from the log file
name, and hudi-common does not have access
+ // to the table config fallback used by write clients. Legacy callers with
no explicit block format default to Avro.
Review Comment:
The `avro` fallback diverges from `CommonClientUtils.getLogBlockType`, which
infers the block type from the base format when unset (HFILE base -> HFILE
blocks). None of the current callers pass MDT slices so this is latent, but
since `FileSlice` carries its base file we could infer from the base file
extension when one is present, or at least call out the parquet-base-table
assumption in the javadoc.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestPartitionDirectoryConverter.scala:
##########
@@ -49,6 +51,7 @@ class TestPartitionDirectoryConverter extends
SparkAdapterSupport {
val options = Map(
s"${HoodieStorageConfig.LOGFILE_TO_PARQUET_COMPRESSION_RATIO_FRACTION.key()}"
-> logFraction.toString
)
+ val config = new HoodieConfig(TypedProperties.fromMap(options.asJava))
Review Comment:
Could we add one slice with a native log file here too? `TestFileSlice`
covers the math, but this is the Spark split-sizing path that benefits from the
fix.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala:
##########
@@ -106,6 +106,9 @@ case class HoodieFileIndex(spark: SparkSession,
@transient protected var hasPushedDownPartitionPredicates: Boolean = false
+ private lazy val hoodieConfig =
Review Comment:
nit: the neighboring derived fields are `@transient`; probably worth marking
this one too since it is derivable from `options`.
##########
hudi-common/src/main/java/org/apache/hudi/common/model/FileSlice.java:
##########
@@ -193,24 +196,44 @@ 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) {
Review Comment:
Since this is public API in hudi-common, keeping a `@Deprecated` overload
taking the old `double` (delegating with a synthetic config) would give
external callers a release to migrate. The description already flags the break;
this would make it painless.
--
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]
