feat: support write native format log files#19067
Conversation
There was a problem hiding this comment.
HoodieTableConfig.TABLE_STORAGE_LAYOUT is declared with .sinceVersion("1.3.0"), but HoodieTableVersion only maps releases through 1.1.0. During table config creation, validateConfigVersion(...) calls HoodieTableVersion.fromReleaseVersion("1.3.0"), which throws before the table config can be persisted.
This means the new config that enables the LSM/native-log write path fails at creation time, so the writer selection based on isLSMTreeStorageLayout() is not reachable.
Maybe we should add the corresponding table version/release mapping before using sinceVersion("1.3.0"), or avoid using an unmapped sinceVersion value for this config until table version support is added.
There was a problem hiding this comment.
Addressed in aa48c8e: added HoodieTableVersion.TEN (version code 10) mapped to release 1.3.0, while keeping HoodieTableVersion.current() at NINE. TABLE_STORAGE_LAYOUT stays .sinceVersion("1.3.0"), and the validation test now covers NINE as invalid and TEN as valid.
| String keyField = config.populateMetaFields() | ||
| ? HoodieRecord.RECORD_KEY_METADATA_FIELD | ||
| : hoodieTable.getMetaClient().getTableConfig().getRecordKeyFieldProp(); | ||
| if (!writer.canWriteDataFile()) { |
There was a problem hiding this comment.
HoodieNativeLogAppendHandle.writeInsertAndUpdate(...) flushes only the data native log when !writer.canWriteDataFile(). HoodieNativeLogFormatWriter.flushDataAppend(...) closes dataFileWriter and resets currentAppendVersion, but leaves an already-open deleteFileWriter/deleteLogFile under the old version.
A single append can therefore produce:
v1.log.parquetv2.log.parquetv1.deletes.parquet
HoodieLogFile.LogFileComparator sorts by log version before extension precedence, so v1.deletes.parquet sorts before v2.log.parquet. That breaks the intended same-key ordering where deletes should be applied after all data records for the same append/version sequence.
There was a problem hiding this comment.
Addressed in aa48c8e: the data-file rollover path now calls flushAppend() instead of the old data-only flush. That closes any pending data and delete native writers together and advances the append version as one unit, so .log.parquet and .deletes.parquet stay aligned for comparator ordering.
| record, recordSchema, recordProperties); | ||
| } | ||
|
|
||
| public void appendDeleteRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException { |
There was a problem hiding this comment.
The existing inline log append path has rollover at the append/log-writer level. Records and deletes are buffered in the same handle, then HoodieInlineLogAppendHandle.appendDataAndDeleteBlocks(...) writes both data blocks and HoodieDeleteBlocks through HoodieLogFormat.Writer. That writer is created with config.getLogFileMaxSize(), so delete blocks participate in the same log-file size/rollover behavior as data blocks.
While HoodieNativeLogFormatWriter.appendDeleteRecord(...) always writes to the current delete writer. There is no delete-side equivalent of canWriteDataFile(), so delete-heavy workloads can keep appending to one .deletes.parquet file past the configured log/native file size threshold. Should we also add delete-writer size/rollover handling for native logs as well?
There was a problem hiding this comment.
Addressed in aa48c8e: added canWriteDeleteFile() and the native append handle now flushes via flushAppend() before appending a delete when the current delete writer reaches its write limit. Delete-heavy workloads should roll to the next native log version instead of growing one delete file indefinitely.
| return fieldValues; | ||
| } | ||
|
|
||
| public AppendResult flushAppend() throws IOException { |
There was a problem hiding this comment.
Never used, can be removed?
There was a problem hiding this comment.
Addressed in aa48c8e: removed the unused no-arg flushAppend() helper along with the data-only flushDataAppend() path.
| logFormatMetadata.put(HeaderMetadataType.VERSION.name(), String.valueOf(NATIVE_LOG_FORMAT_VERSION)); | ||
| header.forEach((key, value) -> { | ||
| if (value != null && key != HeaderMetadataType.SCHEMA) { | ||
| // filter out the redundant schema since the footer already carries it. |
There was a problem hiding this comment.
Which schema does "the footer already carries" mean — the physical Parquet MessageType, or a full logical schema? HoodieSchema carries type info that doesn't round-trip losslessly through MessageType, for example VECTOR type. Dropping SCHEMA unconditionally may lose info for native log files.
There was a problem hiding this comment.
Addressed in aa48c8e: restored the logical schema metadata by putting HeaderMetadataType.SCHEMA back into the header and no longer filtering it out when building native data-log footer metadata. Per the follow-up decision, footer metadata is only added to native data logs; native delete logs do not add a footer.
| try { | ||
| closeFileWriters(); | ||
| } catch (IOException e) { | ||
| throw new RuntimeException("Failed to close native log file writers", e); |
There was a problem hiding this comment.
HereIOException is wrapped as RuntimeException; the codebase convention is HoodieIOException.
There was a problem hiding this comment.
Addressed in aa48c8e: changed the close-path wrapper from RuntimeException to HoodieIOException.
| public List<WriteStatus> close() { | ||
| try { | ||
| if (isClosed()) { | ||
| return statuses.isEmpty() ? java.util.Collections.singletonList(writeStatus) : statuses; |
There was a problem hiding this comment.
Uses fully-qualified java.util.Collections.singletonList(...) since the import was dropped from the base class; prefer re-adding the import.
There was a problem hiding this comment.
Addressed in aa48c8e: re-added the java.util.Collections import and switched the call back to Collections.singletonList(...).
3 participants
Describe the issue this Pull Request addresses
This closes #19058 .
RFC-103 introduces an LSM-tree table storage layout that needs MOR append writes to produce native-format log files instead of inline Hudi log blocks. The existing append path was centered on
HoodieAppendHandlebuffering records into inline log blocks, which made it hard to support engine-native record types, native file writers, native delete logs, footer metadata, and per-rolled-file column stats.This PR adds the opt-in table config
hoodie.table.storage.layout=lsm_treewhile keeping the default layout unchanged. When the layout islsm_tree, MOR append/update/log-compaction paths route to native log append handles that write native data and delete log files, including native log footer metadata with format version2.Summary and Changelog
This PR refactors append handling into a shared append base plus inline/native implementations, then wires Spark, Java, and Flink MOR write paths to choose inline or native log handles based on the table storage layout.
Working tree: RFC-103 native log append support
hoodie.table.storage.layoutinHoodieTableConfigwith valuesdefaultandlsm_tree;defaultpreserves the current layout.HoodieAppendHandleinto an abstract common append lifecycle andHoodieInlineLogAppendHandle, preserving the existing inline log behavior behind the inline implementation.HoodieNativeLogAppendHandleandHoodieNativeLogFormatWriterto stream records through engine-native file writers instead of collecting records into inline data/delete blocks..log.parquetand.deletes.parquetfiles, native delete-log row construction, native footer metadata underhudi.log.format.metadata, and native log formatVERSION=2.AppendHandleFactory, Spark/Java MOR delta executors, Flink write handle factories, and log compaction to select native append handles whenisLSMTreeStorageLayout()is enabled.FileGroupReaderBasedNativeLogAppendHandle,FlinkNativeLogAppendHandle, andRowDataNativeLogWriteHandle.FileGroupReaderBasedInlineLogAppendHandle,FlinkInlineLogAppendHandle, andRowDataInlineLogWriteHandle.Working tree: log extension constants and comparator coverage
LogExtensionswith sharedDATA_LOG_EXTENSION,DELETE_LOG_EXTENSION,CDC_LOG_EXTENSION, andARCHIVE_LOG_EXTENSIONconstants.HoodieLogFileextension precedence andHoodieNativeLogFormatWriterto useLogExtensionsinstead of duplicated string literals.TestHoodieLogFile.logFileComparatorOrdersNativeLogExtensionsByPrecedenceForSameVersionto pin native log ordering aslog < deletes < cdc < archive.Impact
This is an opt-in storage-layout change. Existing tables continue to use the default inline log layout unless
hoodie.table.storage.layout=lsm_treeis configured.The PR affects core MOR append/update/log-compaction paths across Spark, Java, and Flink, plus log file naming/comparison and metadata-column-stats collection for native logs. Native logs introduce a new physical log representation and footer metadata contract, but the default layout remains backward compatible.
Risk Level
medium
The risk is medium because the change touches core MOR write paths, append-handle lifecycle accounting, log file naming/comparison, Flink RowData write handles, log compaction, native file-writer metadata, and column-stats collection. The main compatibility mitigation is that the new behavior is gated behind
hoodie.table.storage.layout=lsm_tree, withdefaultretaining the current inline log layout.Validation evidence:
git diff --checkpassed for the latest touched files.mvn -pl hudi-client/hudi-client-common -DskipTests compile, but it is currently blocked by an existing compile error inHoodieNativeLogFormatWriter.javaresolvingorg.apache.hudi.common.schema.HoodieSchemas.mvn -pl hudi-common -DskipTests compileandmvn -pl hudi-common -DskipITs -DfailIfNoTests=false -Dtest=TestHoodieLogFile test, but both are currently blocked by an existing compile error inHoodieSchema.javaresolvingStringUtils.splitTopLevelCommas(...).Documentation Update
RFC-103 is updated in this PR. Additional user-facing docs may be needed for the new
hoodie.table.storage.layout=lsm_treeconfig before enabling or advertising the layout outside the RFC context.Contributor's checklist