You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "danny0405 (via GitHub)" <gi...@apache.org> on 2026/06/25 06:30:44 UTC
[PR] feat: support write native format log files [hudi]
danny0405 opened a new pull request, #19067:
URL: https://github.com/apache/hudi/pull/19067
### 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 `HoodieAppendHandle` buffering 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_tree` while keeping the default layout unchanged. When the layout is `lsm_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 version `2`.
### 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
- Added `hoodie.table.storage.layout` in `HoodieTableConfig` with values `default` and `lsm_tree`; `default` preserves the current layout.
- Split `HoodieAppendHandle` into an abstract common append lifecycle and `HoodieInlineLogAppendHandle`, preserving the existing inline log behavior behind the inline implementation.
- Added `HoodieNativeLogAppendHandle` and `HoodieNativeLogFormatWriter` to stream records through engine-native file writers instead of collecting records into inline data/delete blocks.
- Added native data and delete log handling, including `.log.parquet` and `.deletes.parquet` files, native delete-log row construction, native footer metadata under `hudi.log.format.metadata`, and native log format `VERSION=2`.
- Routed `AppendHandleFactory`, Spark/Java MOR delta executors, Flink write handle factories, and log compaction to select native append handles when `isLSMTreeStorageLayout()` is enabled.
- Added native variants for file-group-reader compaction and Flink/RowData append paths: `FileGroupReaderBasedNativeLogAppendHandle`, `FlinkNativeLogAppendHandle`, and `RowDataNativeLogWriteHandle`.
- Renamed existing append classes to make inline/native behavior explicit: `FileGroupReaderBasedInlineLogAppendHandle`, `FlinkInlineLogAppendHandle`, and `RowDataInlineLogWriteHandle`.
- Added file-writer footer metadata plumbing so native log column stats can be collected from native file footers rather than iterating buffered records.
- Updated RFC-103 documentation and existing tests touched by the append-handle split.
#### Working tree: log extension constants and comparator coverage
- Added `LogExtensions` with shared `DATA_LOG_EXTENSION`, `DELETE_LOG_EXTENSION`, `CDC_LOG_EXTENSION`, and `ARCHIVE_LOG_EXTENSION` constants.
- Updated `HoodieLogFile` extension precedence and `HoodieNativeLogFormatWriter` to use `LogExtensions` instead of duplicated string literals.
- Added `TestHoodieLogFile.logFileComparatorOrdersNativeLogExtensionsByPrecedenceForSameVersion` to pin native log ordering as `log < 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_tree` is 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`, with `default` retaining the current inline log layout.
Validation evidence:
- `git diff --check` passed for the latest touched files.
- Attempted `mvn -pl hudi-client/hudi-client-common -DskipTests compile`, but it is currently blocked by an existing compile error in `HoodieNativeLogFormatWriter.java` resolving `org.apache.hudi.common.schema.HoodieSchemas`.
- Attempted `mvn -pl hudi-common -DskipTests compile` and `mvn -pl hudi-common -DskipITs -DfailIfNoTests=false -Dtest=TestHoodieLogFile test`, but both are currently blocked by an existing compile error in `HoodieSchema.java` resolving `StringUtils.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_tree` config before enabling or advertising the layout outside the RFC context.
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Enough context is provided in the sections above
- [ ] Adequate tests were added if applicable
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #19067:
URL: https://github.com/apache/hudi/pull/19067#discussion_r3473971109
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java:
##########
@@ -35,182 +34,251 @@
import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.model.MetadataValues;
import org.apache.hudi.common.schema.HoodieSchema;
-import org.apache.hudi.common.schema.HoodieSchemaField;
-import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.log.AppendResult;
-import org.apache.hudi.common.table.log.HoodieLogFormat;
-import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
-import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
-import org.apache.hudi.common.table.log.block.HoodieHFileDataBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
-import org.apache.hudi.common.table.log.block.HoodieParquetDataBlock;
import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator;
import org.apache.hudi.common.table.view.TableFileSystemView;
-import org.apache.hudi.common.util.ConfigUtils;
-import org.apache.hudi.common.util.DefaultSizeEstimator;
import org.apache.hudi.common.util.HoodieRecordUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
-import org.apache.hudi.common.util.SizeEstimator;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieAppendException;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieUpsertException;
-import org.apache.hudi.metadata.HoodieIndexVersion;
-import org.apache.hudi.metadata.HoodieTableMetadataUtil;
-import org.apache.hudi.stats.HoodieColumnRangeMetadata;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieTable;
-import org.apache.hudi.util.CommonClientUtils;
-import org.apache.hudi.util.Lazy;
import lombok.extern.slf4j.Slf4j;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_DYNAMIC_MAX_ENTRIES;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_FPP_VALUE;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_NUM_ENTRIES_VALUE;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_TYPE;
-import static org.apache.hudi.common.config.HoodieStorageConfig.HFILE_COMPRESSION_ALGORITHM_NAME;
-import static org.apache.hudi.common.config.HoodieStorageConfig.HFILE_WITH_BLOOM_FILTER_ENABLED;
-import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
-import static org.apache.hudi.metadata.HoodieTableMetadataUtil.collectColumnRangeMetadata;
-
/**
- * IO Operation to append data onto an existing file.
+ * Shared base for append handles writing merge-on-read log files.
+ *
+ * <p>This class owns the common append lifecycle:
+ * <ol>
+ * <li>Initialize write status and log writer state for the target file group.</li>
+ * <li>Iterate incoming {@link HoodieRecord}s and route each record as an insert/update or delete.</li>
+ * <li>Track write counters, record locations, write-status success/failure state, and runtime stats.</li>
+ * <li>Flush pending records to one or more physical log files and update {@link HoodieDeltaWriteStat}s.</li>
+ * <li>Close the record iterator and log writer, then publish secondary-index streaming stats when enabled.</li>
+ * </ol>
+ *
+ * <p>Concrete subclasses provide only the physical log representation. Inline-log handles buffer records
+ * into Hudi log blocks, while native-log handles stream records into engine-native file writers. Both variants
+ * share the same accounting and status update logic through this class.
*/
@Slf4j
-public class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O> {
- // This acts as the sequenceID for records written
+public abstract class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O> {
+
private static final AtomicLong RECORD_COUNTER = new AtomicLong(1);
- private static final int NUMBER_OF_RECORDS_TO_ESTIMATE_RECORD_SIZE = 100;
- // Buffer for holding records in memory before they are flushed to disk
- protected final List<HoodieRecord> recordList = new ArrayList<>();
- // Buffer for holding records (to be deleted), along with their position in log block, in memory before they are flushed to disk
- protected final List<Pair<DeleteRecord, Long>> recordsToDeleteWithPositions = new ArrayList<>();
- // Base file instant time of the record positions
- protected final Option<String> baseFileInstantTimeOfPositions;
- // Incoming records to be written to logs.
protected Iterator<HoodieRecord<T>> recordItr;
- // Writer to log into the file group's latest slice.
- protected HoodieLogFormat.Writer writer;
-
- protected final List<WriteStatus> statuses;
- // Total number of records written during appending
+ protected final List<WriteStatus> statuses = new ArrayList<>();
+ protected final Map<HeaderMetadataType, String> header = new HashMap<>();
+ protected final Properties recordProperties = new Properties();
+ protected final Option<String> baseFileInstantTimeOfPositions;
+ protected boolean isLogCompaction = false;
+ protected boolean useWriterSchema = false;
+ protected boolean doInit = true;
protected long recordsWritten = 0;
- // Total number of records deleted during appending
protected long recordsDeleted = 0;
- // Total number of records updated during appending
protected long updatedRecordsWritten = 0;
- // Total number of new records inserted into the delta file
protected long insertRecordsWritten = 0;
+ protected Option<HoodieLogBlock> appendDataBlock = Option.empty();
- // Average record size for a HoodieRecord. This size is updated at the end of every log block flushed to disk
- private long averageRecordSize = 0;
- // Flag used to initialize some metadata
- private boolean doInit = true;
- // Total number of bytes written during this append phase (an estimation)
- protected long estimatedNumberOfBytesWritten;
- // Number of records that must be written to meet the max block size for a log block
- private long numberOfRecords = 0;
- // Max block size to limit to for a log block
- private final long maxBlockSize = config.getLogFileDataBlockMaxSize();
- // Header metadata for a log block
- protected final Map<HeaderMetadataType, String> header = new HashMap<>();
- private final SizeEstimator<HoodieRecord> sizeEstimator;
- // This is used to distinguish between normal append and logcompaction's append operation.
- private boolean isLogCompaction = false;
- // use writer schema for log compaction.
- private boolean useWriterSchema = false;
-
- private final Properties recordProperties = new Properties();
- private final String[] orderingFields;
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, false);
+ }
- /**
- * This is used by log compaction only.
- */
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
- TaskContextSupplier taskContextSupplier, Map<HeaderMetadataType, String> header) {
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier, Map<HeaderMetadataType, String> header) {
this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, true);
- this.useWriterSchema = true;
- this.isLogCompaction = true;
this.header.putAll(header);
}
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr, TaskContextSupplier taskContextSupplier) {
- this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, false);
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, null, taskContextSupplier);
}
- private HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr, TaskContextSupplier taskContextSupplier, boolean preserveMetadata) {
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier, boolean preserveMetadata) {
super(config, instantTime, partitionPath, fileId, hoodieTable,
config.shouldWritePartialUpdates()
- // When enabling writing partial updates to the data blocks in log files,
- // i.e., partial update schema is set, the writer schema is the partial
- // schema containing the updated fields only
? Option.of(HoodieSchema.parse(config.getPartialUpdateSchema()))
: Option.empty(),
taskContextSupplier,
preserveMetadata);
this.recordItr = recordItr;
- this.sizeEstimator = getSizeEstimator();
- this.statuses = new ArrayList<>();
+ this.isLogCompaction = preserveMetadata;
+ this.useWriterSchema = preserveMetadata;
this.recordProperties.putAll(config.getProps());
- this.orderingFields = ConfigUtils.getOrderingFields(recordProperties);
- boolean shouldWriteRecordPositions = config.shouldWriteRecordPositions()
- // record positions supported only from table version 8
- && config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT);
- this.baseFileInstantTimeOfPositions = shouldWriteRecordPositions
+ this.baseFileInstantTimeOfPositions = config.shouldWriteRecordPositions()
+ && config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT)
? getBaseFileInstantTimeOfPositions()
: Option.empty();
}
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, TaskContextSupplier sparkTaskContextSupplier) {
- this(config, instantTime, hoodieTable, partitionPath, fileId, null, sparkTaskContextSupplier);
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, String partitionPath, String fileId,
+ HoodieTable<T, I, K, O> hoodieTable, Option<HoodieSchema> writerSchema,
+ TaskContextSupplier taskContextSupplier, boolean preserveMetadata) {
+ super(config, instantTime, partitionPath, fileId, hoodieTable, writerSchema, taskContextSupplier, preserveMetadata);
+ this.baseFileInstantTimeOfPositions = config.shouldWriteRecordPositions()
+ && config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT)
+ ? getBaseFileInstantTimeOfPositions()
+ : Option.empty();
}
- protected SizeEstimator<HoodieRecord> getSizeEstimator() {
- return new DefaultSizeEstimator<>();
+ /**
+ * Writes all records from the incoming iterator and flushes remaining pending data.
+ */
+ public void doAppend() {
+ while (recordItr.hasNext()) {
+ HoodieRecord<T> record = recordItr.next();
+ writeRecord(record);
+ }
+ flushAppend();
}
- protected Option<String> getBaseFileInstantTimeOfPositions() {
- return hoodieTable.getHoodieView().getLatestBaseFile(partitionPath, fileId)
- .map(HoodieBaseFile::getCommitTime);
+ /**
+ * Writes one record for APIs that call the handle directly instead of using {@link #doAppend()}.
+ */
+ @Override
+ protected void doWrite(HoodieRecord record, HoodieSchema schema, TypedProperties props) {
+ writeRecord(record);
+ }
+
+ /**
+ * Writes a keyed record map, primarily used by log compaction paths.
+ */
+ @Override
+ public void write(Map<String, HoodieRecord<T>> recordMap) {
+ try {
+ for (Map.Entry<String, HoodieRecord<T>> entry : recordMap.entrySet()) {
+ HoodieRecord<T> record = entry.getValue();
+ writeRecord(record);
+ }
+ flushAppend();
+ } catch (Exception e) {
+ throw new HoodieUpsertException("Failed to compact blocks for fileId " + fileId, e);
+ }
+ }
+
+ /**
+ * Flushes pending data, closes resources, finalizes log file sizes, and publishes secondary-index stats.
+ */
+ @Override
+ public List<WriteStatus> close() {
+ try {
+ if (isClosed()) {
+ return statuses.isEmpty() ? java.util.Collections.singletonList(writeStatus) : statuses;
Review Comment:
Addressed in aa48c8eb5fc1: re-added the `java.util.Collections` import and switched the call back to `Collections.singletonList(...)`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java:
##########
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io;
+
+import org.apache.hudi.common.engine.RecordContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemas;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.log.AppendResult;
+import org.apache.hudi.common.table.log.HoodieLogFormat;
+import org.apache.hudi.common.table.log.LogFileCreationCallback;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.table.read.BufferedRecords;
+import org.apache.hudi.common.util.JsonUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.OrderingValues;
+import org.apache.hudi.common.util.collection.ArrayComparable;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.io.storage.HoodieFileWriter;
+import org.apache.hudi.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.hudi.common.model.LogExtensions.DATA_LOG_EXTENSION;
+import static org.apache.hudi.common.model.LogExtensions.DELETE_LOG_EXTENSION;
+
+/**
+ * Writes MOR log blocks as native files, for example {@code .log.parquet} and {@code .deletes.parquet}.
+ */
+public class HoodieNativeLogFormatWriter extends HoodieLogFormat.Writer {
+
+ private static final int NATIVE_LOG_FORMAT_VERSION = 2;
+ private static final String LOG_FORMAT_METADATA_FOOTER_KEY = "hudi.log.format.metadata";
+
+ private final HoodieWriteConfig writeConfig;
+ private final HoodieSchema tableSchema;
+ private final TaskContextSupplier taskContextSupplier;
+ private final RecordContext recordContext;
+ private final List<String> orderingFieldNames;
+ private final Properties recordProperties;
+ private HoodieFileWriter dataFileWriter;
+ private HoodieFileWriter deleteFileWriter;
+ private HoodieLogFile dataLogFile;
+ private HoodieLogFile deleteLogFile;
+ private HoodieSchema deleteLogSchema;
+ private int currentAppendVersion = -1;
+ private List<AppendResult> lastAppendResults = new ArrayList<>();
+ private Option<Object> lastDataFileFormatMetadata = Option.empty();
+
+ public HoodieNativeLogFormatWriter(Integer bufferSize,
+ HoodieStorage storage,
+ StoragePath parentPath,
+ String logFileId,
+ String instantTime,
+ Integer logVersion,
+ String logWriteToken,
+ Long sizeThreshold,
+ LogFileCreationCallback fileCreationCallback,
+ HoodieTableVersion tableVersion,
+ HoodieWriteConfig writeConfig,
+ HoodieSchema tableSchema,
+ TaskContextSupplier taskContextSupplier,
+ RecordContext recordContext,
+ List<String> orderingFieldNames) throws IOException {
+ super(bufferSize, storage, parentPath, logFileId, DATA_LOG_EXTENSION, instantTime, logVersion, logWriteToken,
+ null, 0L, sizeThreshold, fileCreationCallback, tableVersion);
+ this.writeConfig = writeConfig;
+ this.tableSchema = tableSchema;
+ this.taskContextSupplier = taskContextSupplier;
+ this.recordContext = recordContext;
+ this.orderingFieldNames = orderingFieldNames;
+ this.recordProperties = new Properties();
+ this.recordProperties.putAll(writeConfig.getProps());
+ this.logFile = new HoodieLogFile(makeNativeLogPath(this.logVersion, DATA_LOG_EXTENSION), this.fileSize);
+ }
+
+ public List<AppendResult> getLastAppendResults() {
+ return lastAppendResults;
+ }
+
+ public Option<Object> getLastDataFileFormatMetadata() {
+ return lastDataFileFormatMetadata;
+ }
+
+ @Override
+ public AppendResult appendBlock(HoodieLogBlock block) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public AppendResult appendBlocks(List<HoodieLogBlock> blocks) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public long getCurrentSize() {
+ return lastAppendResults.stream().mapToLong(AppendResult::size).sum();
+ }
+
+ @Override
+ public void sync() {
+ // Native log files are write-once; closing the file writer publishes the file.
+ }
+
+ @Override
+ public void close() {
+ try {
+ closeFileWriters();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to close native log file writers", e);
Review Comment:
Addressed in aa48c8eb5fc1: changed the close-path wrapper from `RuntimeException` to `HoodieIOException`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogAppendHandle.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io;
+
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieDeltaWriteStat;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.log.AppendResult;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.util.ConfigUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ParquetUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieAppendException;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.metadata.HoodieIndexVersion;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.stats.HoodieColumnRangeMetadata;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.util.Lazy;
+
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
+
+/**
+ * Append handle for native log files. Unlike {@link HoodieInlineLogAppendHandle}, this handle streams
+ * records directly into native format writers and does not buffer records or build inline log blocks.
+ */
+public class HoodieNativeLogAppendHandle<T, I, K, O> extends HoodieAppendHandle<T, I, K, O> {
+
+ private HoodieNativeLogFormatWriter writer;
+
+ public HoodieNativeLogAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, false, Collections.emptyMap());
+ }
+
+ public HoodieNativeLogAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier, Map<HeaderMetadataType, String> header) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, true, header);
+ }
+
+ public HoodieNativeLogAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, null, taskContextSupplier);
+ }
+
+ private HoodieNativeLogAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier, boolean preserveMetadata,
+ Map<HeaderMetadataType, String> header) {
+ super(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, preserveMetadata);
+ this.header.putAll(header);
+ }
+
+ @Override
+ protected void createLogWriterForAppend(String instantTime, Option<FileSlice> fileSliceOpt) {
+ try {
+ this.writer = new HoodieNativeLogFormatWriter(
+ storage.getDefaultBufferSize(),
+ storage,
+ FSUtils.constructAbsolutePath(hoodieTable.getMetaClient().getBasePath(), partitionPath),
+ fileId,
+ instantTime,
+ null,
+ writeToken,
+ config.getLogFileMaxSize(),
+ getLogCreationCallback(),
+ config.getWriteVersion(),
+ config,
+ writeSchemaWithMetaFields,
+ taskContextSupplier,
+ hoodieTable.getReaderContextFactoryForWrite().getContext().getRecordContext(),
+ Arrays.stream(ConfigUtils.getOrderingFields(recordProperties)).collect(Collectors.toList()));
+ } catch (IOException e) {
+ throw new HoodieException("Creating native log writer with fileId: " + fileId + ", "
+ + "delta commit time: " + instantTime + " error", e);
+ }
+ }
+
+ @Override
+ protected void writeInsertAndUpdate(HoodieSchema schema, HoodieRecord<T> hoodieRecord, boolean isUpdateRecord) throws IOException {
+ if (hoodieRecord.shouldIgnore(schema, recordProperties)) {
+ return;
+ }
+ HoodieRecord populatedRecord = hoodieRecord.prependMetaFields(
+ schema, writeSchemaWithMetaFields, populateMetadataFields(hoodieRecord), recordProperties);
+ String keyField = config.populateMetaFields()
+ ? HoodieRecord.RECORD_KEY_METADATA_FIELD
+ : hoodieTable.getMetaClient().getTableConfig().getRecordKeyFieldProp();
+ if (!writer.canWriteDataFile()) {
Review Comment:
Addressed in aa48c8eb5fc1: 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.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -240,6 +240,12 @@ public class HoodieTableConfig extends HoodieConfig {
.withAlternatives("hoodie.table.rt.file.format")
.withDocumentation("Log format used for the delta logs.");
+ public static final ConfigProperty<String> TABLE_STORAGE_LAYOUT = ConfigProperty
+ .key("hoodie.table.storage.layout")
+ .defaultValue(TableStorageLayout.DEFAULT.configValue)
+ .sinceVersion("1.3.0")
Review Comment:
Addressed in aa48c8eb5fc1: 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.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java:
##########
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io;
+
+import org.apache.hudi.common.engine.RecordContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemas;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.log.AppendResult;
+import org.apache.hudi.common.table.log.HoodieLogFormat;
+import org.apache.hudi.common.table.log.LogFileCreationCallback;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.table.read.BufferedRecords;
+import org.apache.hudi.common.util.JsonUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.OrderingValues;
+import org.apache.hudi.common.util.collection.ArrayComparable;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.io.storage.HoodieFileWriter;
+import org.apache.hudi.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.hudi.common.model.LogExtensions.DATA_LOG_EXTENSION;
+import static org.apache.hudi.common.model.LogExtensions.DELETE_LOG_EXTENSION;
+
+/**
+ * Writes MOR log blocks as native files, for example {@code .log.parquet} and {@code .deletes.parquet}.
+ */
+public class HoodieNativeLogFormatWriter extends HoodieLogFormat.Writer {
+
+ private static final int NATIVE_LOG_FORMAT_VERSION = 2;
+ private static final String LOG_FORMAT_METADATA_FOOTER_KEY = "hudi.log.format.metadata";
+
+ private final HoodieWriteConfig writeConfig;
+ private final HoodieSchema tableSchema;
+ private final TaskContextSupplier taskContextSupplier;
+ private final RecordContext recordContext;
+ private final List<String> orderingFieldNames;
+ private final Properties recordProperties;
+ private HoodieFileWriter dataFileWriter;
+ private HoodieFileWriter deleteFileWriter;
+ private HoodieLogFile dataLogFile;
+ private HoodieLogFile deleteLogFile;
+ private HoodieSchema deleteLogSchema;
+ private int currentAppendVersion = -1;
+ private List<AppendResult> lastAppendResults = new ArrayList<>();
+ private Option<Object> lastDataFileFormatMetadata = Option.empty();
+
+ public HoodieNativeLogFormatWriter(Integer bufferSize,
+ HoodieStorage storage,
+ StoragePath parentPath,
+ String logFileId,
+ String instantTime,
+ Integer logVersion,
+ String logWriteToken,
+ Long sizeThreshold,
+ LogFileCreationCallback fileCreationCallback,
+ HoodieTableVersion tableVersion,
+ HoodieWriteConfig writeConfig,
+ HoodieSchema tableSchema,
+ TaskContextSupplier taskContextSupplier,
+ RecordContext recordContext,
+ List<String> orderingFieldNames) throws IOException {
+ super(bufferSize, storage, parentPath, logFileId, DATA_LOG_EXTENSION, instantTime, logVersion, logWriteToken,
+ null, 0L, sizeThreshold, fileCreationCallback, tableVersion);
+ this.writeConfig = writeConfig;
+ this.tableSchema = tableSchema;
+ this.taskContextSupplier = taskContextSupplier;
+ this.recordContext = recordContext;
+ this.orderingFieldNames = orderingFieldNames;
+ this.recordProperties = new Properties();
+ this.recordProperties.putAll(writeConfig.getProps());
+ this.logFile = new HoodieLogFile(makeNativeLogPath(this.logVersion, DATA_LOG_EXTENSION), this.fileSize);
+ }
+
+ public List<AppendResult> getLastAppendResults() {
+ return lastAppendResults;
+ }
+
+ public Option<Object> getLastDataFileFormatMetadata() {
+ return lastDataFileFormatMetadata;
+ }
+
+ @Override
+ public AppendResult appendBlock(HoodieLogBlock block) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public AppendResult appendBlocks(List<HoodieLogBlock> blocks) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public long getCurrentSize() {
+ return lastAppendResults.stream().mapToLong(AppendResult::size).sum();
+ }
+
+ @Override
+ public void sync() {
+ // Native log files are write-once; closing the file writer publishes the file.
+ }
+
+ @Override
+ public void close() {
+ try {
+ closeFileWriters();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to close native log file writers", e);
+ }
+ }
+
+ public boolean hasPendingWrites() {
+ return dataFileWriter != null || deleteFileWriter != null;
+ }
+
+ public boolean canWriteDataFile() {
+ return dataFileWriter == null || dataFileWriter.canWrite();
+ }
+
+ public void appendRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
+ ensureDataFileWriter(recordSchema);
+ dataFileWriter.write(record.getRecordKey(recordSchema, keyFieldName),
+ record, recordSchema, recordProperties);
+ }
+
+ public void appendDeleteRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
+ ensureDeleteFileWriter();
+ String recordKey = record.getRecordKey(recordSchema, keyFieldName);
+ Comparable orderingValue = record.getOrderingValue(
+ recordSchema, recordProperties, orderingFieldNames.toArray(new String[0]));
+ Object deleteEngineRecord = recordContext.constructEngineRecord(
+ deleteLogSchema, createDeleteLogFieldValues(recordKey, orderingValue));
+ // Keep isDelete=false here so RecordContext constructs a data-bearing HoodieRecord
+ // with the native delete-log row. The delete semantics come from the delete log file
+ // itself; setting isDelete=true would create a HoodieEmptyRecord and lose the row.
+ BufferedRecord deleteRecord = BufferedRecords.fromEngineRecord(
+ deleteEngineRecord, deleteLogSchema, recordContext, orderingValue, recordKey, false);
+ deleteFileWriter.write(recordKey, recordContext.constructHoodieRecord(deleteRecord, record.getPartitionPath()),
+ deleteLogSchema, recordProperties);
+ }
+
+ private Object[] createDeleteLogFieldValues(String recordKey, Comparable orderingValue) {
+ Object[] fieldValues = new Object[deleteLogSchema.getFields().size()];
+ fieldValues[0] = recordContext.convertValueToEngineType(recordKey);
+ if (!OrderingValues.isCommitTimeOrderingValue(orderingValue)) {
+ if (orderingFieldNames.size() == 1) {
+ fieldValues[1] = orderingValue;
+ } else {
+ List<Comparable> orderingValues = OrderingValues.getValues((ArrayComparable) orderingValue);
+ for (int i = 0; i < orderingValues.size(); i++) {
+ fieldValues[i + 1] = orderingValues.get(i);
+ }
+ }
+ }
+ return fieldValues;
+ }
+
+ public AppendResult flushAppend() throws IOException {
Review Comment:
Addressed in aa48c8eb5fc1: removed the unused no-arg `flushAppend()` helper along with the data-only `flushDataAppend()` path.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java:
##########
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io;
+
+import org.apache.hudi.common.engine.RecordContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemas;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.log.AppendResult;
+import org.apache.hudi.common.table.log.HoodieLogFormat;
+import org.apache.hudi.common.table.log.LogFileCreationCallback;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.table.read.BufferedRecords;
+import org.apache.hudi.common.util.JsonUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.OrderingValues;
+import org.apache.hudi.common.util.collection.ArrayComparable;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.io.storage.HoodieFileWriter;
+import org.apache.hudi.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.hudi.common.model.LogExtensions.DATA_LOG_EXTENSION;
+import static org.apache.hudi.common.model.LogExtensions.DELETE_LOG_EXTENSION;
+
+/**
+ * Writes MOR log blocks as native files, for example {@code .log.parquet} and {@code .deletes.parquet}.
+ */
+public class HoodieNativeLogFormatWriter extends HoodieLogFormat.Writer {
+
+ private static final int NATIVE_LOG_FORMAT_VERSION = 2;
+ private static final String LOG_FORMAT_METADATA_FOOTER_KEY = "hudi.log.format.metadata";
+
+ private final HoodieWriteConfig writeConfig;
+ private final HoodieSchema tableSchema;
+ private final TaskContextSupplier taskContextSupplier;
+ private final RecordContext recordContext;
+ private final List<String> orderingFieldNames;
+ private final Properties recordProperties;
+ private HoodieFileWriter dataFileWriter;
+ private HoodieFileWriter deleteFileWriter;
+ private HoodieLogFile dataLogFile;
+ private HoodieLogFile deleteLogFile;
+ private HoodieSchema deleteLogSchema;
+ private int currentAppendVersion = -1;
+ private List<AppendResult> lastAppendResults = new ArrayList<>();
+ private Option<Object> lastDataFileFormatMetadata = Option.empty();
+
+ public HoodieNativeLogFormatWriter(Integer bufferSize,
+ HoodieStorage storage,
+ StoragePath parentPath,
+ String logFileId,
+ String instantTime,
+ Integer logVersion,
+ String logWriteToken,
+ Long sizeThreshold,
+ LogFileCreationCallback fileCreationCallback,
+ HoodieTableVersion tableVersion,
+ HoodieWriteConfig writeConfig,
+ HoodieSchema tableSchema,
+ TaskContextSupplier taskContextSupplier,
+ RecordContext recordContext,
+ List<String> orderingFieldNames) throws IOException {
+ super(bufferSize, storage, parentPath, logFileId, DATA_LOG_EXTENSION, instantTime, logVersion, logWriteToken,
+ null, 0L, sizeThreshold, fileCreationCallback, tableVersion);
+ this.writeConfig = writeConfig;
+ this.tableSchema = tableSchema;
+ this.taskContextSupplier = taskContextSupplier;
+ this.recordContext = recordContext;
+ this.orderingFieldNames = orderingFieldNames;
+ this.recordProperties = new Properties();
+ this.recordProperties.putAll(writeConfig.getProps());
+ this.logFile = new HoodieLogFile(makeNativeLogPath(this.logVersion, DATA_LOG_EXTENSION), this.fileSize);
+ }
+
+ public List<AppendResult> getLastAppendResults() {
+ return lastAppendResults;
+ }
+
+ public Option<Object> getLastDataFileFormatMetadata() {
+ return lastDataFileFormatMetadata;
+ }
+
+ @Override
+ public AppendResult appendBlock(HoodieLogBlock block) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public AppendResult appendBlocks(List<HoodieLogBlock> blocks) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public long getCurrentSize() {
+ return lastAppendResults.stream().mapToLong(AppendResult::size).sum();
+ }
+
+ @Override
+ public void sync() {
+ // Native log files are write-once; closing the file writer publishes the file.
+ }
+
+ @Override
+ public void close() {
+ try {
+ closeFileWriters();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to close native log file writers", e);
+ }
+ }
+
+ public boolean hasPendingWrites() {
+ return dataFileWriter != null || deleteFileWriter != null;
+ }
+
+ public boolean canWriteDataFile() {
+ return dataFileWriter == null || dataFileWriter.canWrite();
+ }
+
+ public void appendRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
+ ensureDataFileWriter(recordSchema);
+ dataFileWriter.write(record.getRecordKey(recordSchema, keyFieldName),
+ record, recordSchema, recordProperties);
+ }
+
+ public void appendDeleteRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
+ ensureDeleteFileWriter();
+ String recordKey = record.getRecordKey(recordSchema, keyFieldName);
+ Comparable orderingValue = record.getOrderingValue(
+ recordSchema, recordProperties, orderingFieldNames.toArray(new String[0]));
+ Object deleteEngineRecord = recordContext.constructEngineRecord(
+ deleteLogSchema, createDeleteLogFieldValues(recordKey, orderingValue));
+ // Keep isDelete=false here so RecordContext constructs a data-bearing HoodieRecord
+ // with the native delete-log row. The delete semantics come from the delete log file
+ // itself; setting isDelete=true would create a HoodieEmptyRecord and lose the row.
+ BufferedRecord deleteRecord = BufferedRecords.fromEngineRecord(
+ deleteEngineRecord, deleteLogSchema, recordContext, orderingValue, recordKey, false);
+ deleteFileWriter.write(recordKey, recordContext.constructHoodieRecord(deleteRecord, record.getPartitionPath()),
+ deleteLogSchema, recordProperties);
+ }
+
+ private Object[] createDeleteLogFieldValues(String recordKey, Comparable orderingValue) {
+ Object[] fieldValues = new Object[deleteLogSchema.getFields().size()];
+ fieldValues[0] = recordContext.convertValueToEngineType(recordKey);
+ if (!OrderingValues.isCommitTimeOrderingValue(orderingValue)) {
+ if (orderingFieldNames.size() == 1) {
+ fieldValues[1] = orderingValue;
+ } else {
+ List<Comparable> orderingValues = OrderingValues.getValues((ArrayComparable) orderingValue);
+ for (int i = 0; i < orderingValues.size(); i++) {
+ fieldValues[i + 1] = orderingValues.get(i);
+ }
+ }
+ }
+ return fieldValues;
+ }
+
+ public AppendResult flushAppend() throws IOException {
+ return flushAppend(Collections.emptyMap());
+ }
+
+ public AppendResult flushDataAppend(Map<HeaderMetadataType, String> header) throws IOException {
+ int appendVersion = currentAppendVersion;
+ lastAppendResults = new ArrayList<>();
+ lastDataFileFormatMetadata = Option.empty();
+ addFooterMetadataToDataFile(header);
+ if (dataFileWriter != null) {
+ dataFileWriter.close();
+ lastDataFileFormatMetadata = Option.ofNullable(dataFileWriter.getFileFormatMetadata());
+ dataFileWriter = null;
+ }
+ if (dataLogFile != null) {
+ lastAppendResults.add(toAppendResult(dataLogFile));
+ }
+ dataLogFile = null;
+ currentAppendVersion = -1;
+ return completeAppend(appendVersion);
+ }
+
+ public AppendResult flushAppend(Map<HeaderMetadataType, String> header) throws IOException {
+ int appendVersion = currentAppendVersion;
+ lastAppendResults = new ArrayList<>();
+ lastDataFileFormatMetadata = Option.empty();
+ addFooterMetadata(header);
+ closeFileWriters();
+ if (dataLogFile != null) {
+ lastAppendResults.add(toAppendResult(dataLogFile));
+ }
+ if (deleteLogFile != null) {
+ lastAppendResults.add(toAppendResult(deleteLogFile));
+ }
+ dataLogFile = null;
+ deleteLogFile = null;
+ currentAppendVersion = -1;
+ return completeAppend(appendVersion);
+ }
+
+ private void addFooterMetadata(Map<HeaderMetadataType, String> header) throws IOException {
+ Map<String, String> footerMetadata = getFooterMetadata(header);
+ if (dataFileWriter != null) {
+ dataFileWriter.addFooterMetadata(footerMetadata);
+ }
+ if (deleteFileWriter != null) {
+ deleteFileWriter.addFooterMetadata(footerMetadata);
+ }
+ }
+
+ private void addFooterMetadataToDataFile(Map<HeaderMetadataType, String> header) throws IOException {
+ if (dataFileWriter != null) {
+ dataFileWriter.addFooterMetadata(getFooterMetadata(header));
+ }
+ }
+
+ private Map<String, String> getFooterMetadata(Map<HeaderMetadataType, String> header) throws IOException {
+ Map<String, String> logFormatMetadata = new LinkedHashMap<>();
+ 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.
Review Comment:
Addressed in aa48c8eb5fc1: 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.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java:
##########
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io;
+
+import org.apache.hudi.common.engine.RecordContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemas;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.log.AppendResult;
+import org.apache.hudi.common.table.log.HoodieLogFormat;
+import org.apache.hudi.common.table.log.LogFileCreationCallback;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.table.read.BufferedRecords;
+import org.apache.hudi.common.util.JsonUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.OrderingValues;
+import org.apache.hudi.common.util.collection.ArrayComparable;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.io.storage.HoodieFileWriter;
+import org.apache.hudi.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.hudi.common.model.LogExtensions.DATA_LOG_EXTENSION;
+import static org.apache.hudi.common.model.LogExtensions.DELETE_LOG_EXTENSION;
+
+/**
+ * Writes MOR log blocks as native files, for example {@code .log.parquet} and {@code .deletes.parquet}.
+ */
+public class HoodieNativeLogFormatWriter extends HoodieLogFormat.Writer {
+
+ private static final int NATIVE_LOG_FORMAT_VERSION = 2;
+ private static final String LOG_FORMAT_METADATA_FOOTER_KEY = "hudi.log.format.metadata";
+
+ private final HoodieWriteConfig writeConfig;
+ private final HoodieSchema tableSchema;
+ private final TaskContextSupplier taskContextSupplier;
+ private final RecordContext recordContext;
+ private final List<String> orderingFieldNames;
+ private final Properties recordProperties;
+ private HoodieFileWriter dataFileWriter;
+ private HoodieFileWriter deleteFileWriter;
+ private HoodieLogFile dataLogFile;
+ private HoodieLogFile deleteLogFile;
+ private HoodieSchema deleteLogSchema;
+ private int currentAppendVersion = -1;
+ private List<AppendResult> lastAppendResults = new ArrayList<>();
+ private Option<Object> lastDataFileFormatMetadata = Option.empty();
+
+ public HoodieNativeLogFormatWriter(Integer bufferSize,
+ HoodieStorage storage,
+ StoragePath parentPath,
+ String logFileId,
+ String instantTime,
+ Integer logVersion,
+ String logWriteToken,
+ Long sizeThreshold,
+ LogFileCreationCallback fileCreationCallback,
+ HoodieTableVersion tableVersion,
+ HoodieWriteConfig writeConfig,
+ HoodieSchema tableSchema,
+ TaskContextSupplier taskContextSupplier,
+ RecordContext recordContext,
+ List<String> orderingFieldNames) throws IOException {
+ super(bufferSize, storage, parentPath, logFileId, DATA_LOG_EXTENSION, instantTime, logVersion, logWriteToken,
+ null, 0L, sizeThreshold, fileCreationCallback, tableVersion);
+ this.writeConfig = writeConfig;
+ this.tableSchema = tableSchema;
+ this.taskContextSupplier = taskContextSupplier;
+ this.recordContext = recordContext;
+ this.orderingFieldNames = orderingFieldNames;
+ this.recordProperties = new Properties();
+ this.recordProperties.putAll(writeConfig.getProps());
+ this.logFile = new HoodieLogFile(makeNativeLogPath(this.logVersion, DATA_LOG_EXTENSION), this.fileSize);
+ }
+
+ public List<AppendResult> getLastAppendResults() {
+ return lastAppendResults;
+ }
+
+ public Option<Object> getLastDataFileFormatMetadata() {
+ return lastDataFileFormatMetadata;
+ }
+
+ @Override
+ public AppendResult appendBlock(HoodieLogBlock block) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public AppendResult appendBlocks(List<HoodieLogBlock> blocks) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public long getCurrentSize() {
+ return lastAppendResults.stream().mapToLong(AppendResult::size).sum();
+ }
+
+ @Override
+ public void sync() {
+ // Native log files are write-once; closing the file writer publishes the file.
+ }
+
+ @Override
+ public void close() {
+ try {
+ closeFileWriters();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to close native log file writers", e);
+ }
+ }
+
+ public boolean hasPendingWrites() {
+ return dataFileWriter != null || deleteFileWriter != null;
+ }
+
+ public boolean canWriteDataFile() {
+ return dataFileWriter == null || dataFileWriter.canWrite();
+ }
+
+ public void appendRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
+ ensureDataFileWriter(recordSchema);
+ dataFileWriter.write(record.getRecordKey(recordSchema, keyFieldName),
+ record, recordSchema, recordProperties);
+ }
+
+ public void appendDeleteRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
Review Comment:
Addressed in aa48c8eb5fc1: 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.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4796563386
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4d7144e4ca3aa177134d3458b90bd369e7e95220 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4796598763
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4d7144e4ca3aa177134d3458b90bd369e7e95220 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4797708743
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
}, {
"hash" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921",
"triggerID" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2af16f108cc18819cd2bb8b48f8574474fb91d91 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4798384871
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
}, {
"hash" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921",
"triggerID" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"triggerType" : "PUSH"
}, {
"hash" : "1d94ecea2b8892440d929894272a427f045f44d2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1d94ecea2b8892440d929894272a427f045f44d2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2af16f108cc18819cd2bb8b48f8574474fb91d91 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921)
* 1d94ecea2b8892440d929894272a427f045f44d2 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4798888966
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
}, {
"hash" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921",
"triggerID" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"triggerType" : "PUSH"
}, {
"hash" : "1d94ecea2b8892440d929894272a427f045f44d2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14926",
"triggerID" : "1d94ecea2b8892440d929894272a427f045f44d2",
"triggerType" : "PUSH"
}, {
"hash" : "24780bd352e96bb585736491da5563898358473a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "24780bd352e96bb585736491da5563898358473a",
"triggerType" : "PUSH"
}, {
"hash" : "aa48c8eb5fc1f93afc22f228b8c7d8af2734d852",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14927",
"triggerID" : "aa48c8eb5fc1f93afc22f228b8c7d8af2734d852",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2af16f108cc18819cd2bb8b48f8574474fb91d91 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921)
* 1d94ecea2b8892440d929894272a427f045f44d2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14926)
* 24780bd352e96bb585736491da5563898358473a UNKNOWN
* aa48c8eb5fc1f93afc22f228b8c7d8af2734d852 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14927)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4799662950
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
}, {
"hash" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921",
"triggerID" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"triggerType" : "PUSH"
}, {
"hash" : "1d94ecea2b8892440d929894272a427f045f44d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14926",
"triggerID" : "1d94ecea2b8892440d929894272a427f045f44d2",
"triggerType" : "PUSH"
}, {
"hash" : "24780bd352e96bb585736491da5563898358473a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "24780bd352e96bb585736491da5563898358473a",
"triggerType" : "PUSH"
}, {
"hash" : "aa48c8eb5fc1f93afc22f228b8c7d8af2734d852",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14927",
"triggerID" : "aa48c8eb5fc1f93afc22f228b8c7d8af2734d852",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 24780bd352e96bb585736491da5563898358473a UNKNOWN
* aa48c8eb5fc1f93afc22f228b8c7d8af2734d852 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14927)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4798512049
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
}, {
"hash" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921",
"triggerID" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"triggerType" : "PUSH"
}, {
"hash" : "1d94ecea2b8892440d929894272a427f045f44d2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14926",
"triggerID" : "1d94ecea2b8892440d929894272a427f045f44d2",
"triggerType" : "PUSH"
}, {
"hash" : "24780bd352e96bb585736491da5563898358473a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "24780bd352e96bb585736491da5563898358473a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2af16f108cc18819cd2bb8b48f8574474fb91d91 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921)
* 1d94ecea2b8892440d929894272a427f045f44d2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14926)
* 24780bd352e96bb585736491da5563898358473a UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4799079902
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
}, {
"hash" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921",
"triggerID" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"triggerType" : "PUSH"
}, {
"hash" : "1d94ecea2b8892440d929894272a427f045f44d2",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14926",
"triggerID" : "1d94ecea2b8892440d929894272a427f045f44d2",
"triggerType" : "PUSH"
}, {
"hash" : "24780bd352e96bb585736491da5563898358473a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "24780bd352e96bb585736491da5563898358473a",
"triggerType" : "PUSH"
}, {
"hash" : "aa48c8eb5fc1f93afc22f228b8c7d8af2734d852",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14927",
"triggerID" : "aa48c8eb5fc1f93afc22f228b8c7d8af2734d852",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1d94ecea2b8892440d929894272a427f045f44d2 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14926)
* 24780bd352e96bb585736491da5563898358473a UNKNOWN
* aa48c8eb5fc1f93afc22f228b8c7d8af2734d852 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14927)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4797095618
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
}, {
"hash" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921",
"triggerID" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4d7144e4ca3aa177134d3458b90bd369e7e95220 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918)
* 2af16f108cc18819cd2bb8b48f8574474fb91d91 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4798710273
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
}, {
"hash" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921",
"triggerID" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"triggerType" : "PUSH"
}, {
"hash" : "1d94ecea2b8892440d929894272a427f045f44d2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14926",
"triggerID" : "1d94ecea2b8892440d929894272a427f045f44d2",
"triggerType" : "PUSH"
}, {
"hash" : "24780bd352e96bb585736491da5563898358473a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "24780bd352e96bb585736491da5563898358473a",
"triggerType" : "PUSH"
}, {
"hash" : "aa48c8eb5fc1f93afc22f228b8c7d8af2734d852",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "aa48c8eb5fc1f93afc22f228b8c7d8af2734d852",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2af16f108cc18819cd2bb8b48f8574474fb91d91 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921)
* 1d94ecea2b8892440d929894272a427f045f44d2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14926)
* 24780bd352e96bb585736491da5563898358473a UNKNOWN
* aa48c8eb5fc1f93afc22f228b8c7d8af2734d852 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 merged PR #19067:
URL: https://github.com/apache/hudi/pull/19067
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4796530876
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4d7144e4ca3aa177134d3458b90bd369e7e95220 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4797055930
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
}, {
"hash" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4d7144e4ca3aa177134d3458b90bd369e7e95220 Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918)
* 2af16f108cc18819cd2bb8b48f8574474fb91d91 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "cshuo (via GitHub)" <gi...@apache.org>.
cshuo commented on code in PR #19067:
URL: https://github.com/apache/hudi/pull/19067#discussion_r3473211456
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -240,6 +240,12 @@ public class HoodieTableConfig extends HoodieConfig {
.withAlternatives("hoodie.table.rt.file.format")
.withDocumentation("Log format used for the delta logs.");
+ public static final ConfigProperty<String> TABLE_STORAGE_LAYOUT = ConfigProperty
+ .key("hoodie.table.storage.layout")
+ .defaultValue(TableStorageLayout.DEFAULT.configValue)
+ .sinceVersion("1.3.0")
Review 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.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java:
##########
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io;
+
+import org.apache.hudi.common.engine.RecordContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemas;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.log.AppendResult;
+import org.apache.hudi.common.table.log.HoodieLogFormat;
+import org.apache.hudi.common.table.log.LogFileCreationCallback;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.table.read.BufferedRecords;
+import org.apache.hudi.common.util.JsonUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.OrderingValues;
+import org.apache.hudi.common.util.collection.ArrayComparable;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.io.storage.HoodieFileWriter;
+import org.apache.hudi.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.hudi.common.model.LogExtensions.DATA_LOG_EXTENSION;
+import static org.apache.hudi.common.model.LogExtensions.DELETE_LOG_EXTENSION;
+
+/**
+ * Writes MOR log blocks as native files, for example {@code .log.parquet} and {@code .deletes.parquet}.
+ */
+public class HoodieNativeLogFormatWriter extends HoodieLogFormat.Writer {
+
+ private static final int NATIVE_LOG_FORMAT_VERSION = 2;
+ private static final String LOG_FORMAT_METADATA_FOOTER_KEY = "hudi.log.format.metadata";
+
+ private final HoodieWriteConfig writeConfig;
+ private final HoodieSchema tableSchema;
+ private final TaskContextSupplier taskContextSupplier;
+ private final RecordContext recordContext;
+ private final List<String> orderingFieldNames;
+ private final Properties recordProperties;
+ private HoodieFileWriter dataFileWriter;
+ private HoodieFileWriter deleteFileWriter;
+ private HoodieLogFile dataLogFile;
+ private HoodieLogFile deleteLogFile;
+ private HoodieSchema deleteLogSchema;
+ private int currentAppendVersion = -1;
+ private List<AppendResult> lastAppendResults = new ArrayList<>();
+ private Option<Object> lastDataFileFormatMetadata = Option.empty();
+
+ public HoodieNativeLogFormatWriter(Integer bufferSize,
+ HoodieStorage storage,
+ StoragePath parentPath,
+ String logFileId,
+ String instantTime,
+ Integer logVersion,
+ String logWriteToken,
+ Long sizeThreshold,
+ LogFileCreationCallback fileCreationCallback,
+ HoodieTableVersion tableVersion,
+ HoodieWriteConfig writeConfig,
+ HoodieSchema tableSchema,
+ TaskContextSupplier taskContextSupplier,
+ RecordContext recordContext,
+ List<String> orderingFieldNames) throws IOException {
+ super(bufferSize, storage, parentPath, logFileId, DATA_LOG_EXTENSION, instantTime, logVersion, logWriteToken,
+ null, 0L, sizeThreshold, fileCreationCallback, tableVersion);
+ this.writeConfig = writeConfig;
+ this.tableSchema = tableSchema;
+ this.taskContextSupplier = taskContextSupplier;
+ this.recordContext = recordContext;
+ this.orderingFieldNames = orderingFieldNames;
+ this.recordProperties = new Properties();
+ this.recordProperties.putAll(writeConfig.getProps());
+ this.logFile = new HoodieLogFile(makeNativeLogPath(this.logVersion, DATA_LOG_EXTENSION), this.fileSize);
+ }
+
+ public List<AppendResult> getLastAppendResults() {
+ return lastAppendResults;
+ }
+
+ public Option<Object> getLastDataFileFormatMetadata() {
+ return lastDataFileFormatMetadata;
+ }
+
+ @Override
+ public AppendResult appendBlock(HoodieLogBlock block) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public AppendResult appendBlocks(List<HoodieLogBlock> blocks) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public long getCurrentSize() {
+ return lastAppendResults.stream().mapToLong(AppendResult::size).sum();
+ }
+
+ @Override
+ public void sync() {
+ // Native log files are write-once; closing the file writer publishes the file.
+ }
+
+ @Override
+ public void close() {
+ try {
+ closeFileWriters();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to close native log file writers", e);
+ }
+ }
+
+ public boolean hasPendingWrites() {
+ return dataFileWriter != null || deleteFileWriter != null;
+ }
+
+ public boolean canWriteDataFile() {
+ return dataFileWriter == null || dataFileWriter.canWrite();
+ }
+
+ public void appendRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
+ ensureDataFileWriter(recordSchema);
+ dataFileWriter.write(record.getRecordKey(recordSchema, keyFieldName),
+ record, recordSchema, recordProperties);
+ }
+
+ public void appendDeleteRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
Review 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 `HoodieDeleteBlock`s 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?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java:
##########
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io;
+
+import org.apache.hudi.common.engine.RecordContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemas;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.log.AppendResult;
+import org.apache.hudi.common.table.log.HoodieLogFormat;
+import org.apache.hudi.common.table.log.LogFileCreationCallback;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.table.read.BufferedRecords;
+import org.apache.hudi.common.util.JsonUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.OrderingValues;
+import org.apache.hudi.common.util.collection.ArrayComparable;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.io.storage.HoodieFileWriter;
+import org.apache.hudi.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.hudi.common.model.LogExtensions.DATA_LOG_EXTENSION;
+import static org.apache.hudi.common.model.LogExtensions.DELETE_LOG_EXTENSION;
+
+/**
+ * Writes MOR log blocks as native files, for example {@code .log.parquet} and {@code .deletes.parquet}.
+ */
+public class HoodieNativeLogFormatWriter extends HoodieLogFormat.Writer {
+
+ private static final int NATIVE_LOG_FORMAT_VERSION = 2;
+ private static final String LOG_FORMAT_METADATA_FOOTER_KEY = "hudi.log.format.metadata";
+
+ private final HoodieWriteConfig writeConfig;
+ private final HoodieSchema tableSchema;
+ private final TaskContextSupplier taskContextSupplier;
+ private final RecordContext recordContext;
+ private final List<String> orderingFieldNames;
+ private final Properties recordProperties;
+ private HoodieFileWriter dataFileWriter;
+ private HoodieFileWriter deleteFileWriter;
+ private HoodieLogFile dataLogFile;
+ private HoodieLogFile deleteLogFile;
+ private HoodieSchema deleteLogSchema;
+ private int currentAppendVersion = -1;
+ private List<AppendResult> lastAppendResults = new ArrayList<>();
+ private Option<Object> lastDataFileFormatMetadata = Option.empty();
+
+ public HoodieNativeLogFormatWriter(Integer bufferSize,
+ HoodieStorage storage,
+ StoragePath parentPath,
+ String logFileId,
+ String instantTime,
+ Integer logVersion,
+ String logWriteToken,
+ Long sizeThreshold,
+ LogFileCreationCallback fileCreationCallback,
+ HoodieTableVersion tableVersion,
+ HoodieWriteConfig writeConfig,
+ HoodieSchema tableSchema,
+ TaskContextSupplier taskContextSupplier,
+ RecordContext recordContext,
+ List<String> orderingFieldNames) throws IOException {
+ super(bufferSize, storage, parentPath, logFileId, DATA_LOG_EXTENSION, instantTime, logVersion, logWriteToken,
+ null, 0L, sizeThreshold, fileCreationCallback, tableVersion);
+ this.writeConfig = writeConfig;
+ this.tableSchema = tableSchema;
+ this.taskContextSupplier = taskContextSupplier;
+ this.recordContext = recordContext;
+ this.orderingFieldNames = orderingFieldNames;
+ this.recordProperties = new Properties();
+ this.recordProperties.putAll(writeConfig.getProps());
+ this.logFile = new HoodieLogFile(makeNativeLogPath(this.logVersion, DATA_LOG_EXTENSION), this.fileSize);
+ }
+
+ public List<AppendResult> getLastAppendResults() {
+ return lastAppendResults;
+ }
+
+ public Option<Object> getLastDataFileFormatMetadata() {
+ return lastDataFileFormatMetadata;
+ }
+
+ @Override
+ public AppendResult appendBlock(HoodieLogBlock block) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public AppendResult appendBlocks(List<HoodieLogBlock> blocks) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public long getCurrentSize() {
+ return lastAppendResults.stream().mapToLong(AppendResult::size).sum();
+ }
+
+ @Override
+ public void sync() {
+ // Native log files are write-once; closing the file writer publishes the file.
+ }
+
+ @Override
+ public void close() {
+ try {
+ closeFileWriters();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to close native log file writers", e);
+ }
+ }
+
+ public boolean hasPendingWrites() {
+ return dataFileWriter != null || deleteFileWriter != null;
+ }
+
+ public boolean canWriteDataFile() {
+ return dataFileWriter == null || dataFileWriter.canWrite();
+ }
+
+ public void appendRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
+ ensureDataFileWriter(recordSchema);
+ dataFileWriter.write(record.getRecordKey(recordSchema, keyFieldName),
+ record, recordSchema, recordProperties);
+ }
+
+ public void appendDeleteRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
+ ensureDeleteFileWriter();
+ String recordKey = record.getRecordKey(recordSchema, keyFieldName);
+ Comparable orderingValue = record.getOrderingValue(
+ recordSchema, recordProperties, orderingFieldNames.toArray(new String[0]));
+ Object deleteEngineRecord = recordContext.constructEngineRecord(
+ deleteLogSchema, createDeleteLogFieldValues(recordKey, orderingValue));
+ // Keep isDelete=false here so RecordContext constructs a data-bearing HoodieRecord
+ // with the native delete-log row. The delete semantics come from the delete log file
+ // itself; setting isDelete=true would create a HoodieEmptyRecord and lose the row.
+ BufferedRecord deleteRecord = BufferedRecords.fromEngineRecord(
+ deleteEngineRecord, deleteLogSchema, recordContext, orderingValue, recordKey, false);
+ deleteFileWriter.write(recordKey, recordContext.constructHoodieRecord(deleteRecord, record.getPartitionPath()),
+ deleteLogSchema, recordProperties);
+ }
+
+ private Object[] createDeleteLogFieldValues(String recordKey, Comparable orderingValue) {
+ Object[] fieldValues = new Object[deleteLogSchema.getFields().size()];
+ fieldValues[0] = recordContext.convertValueToEngineType(recordKey);
+ if (!OrderingValues.isCommitTimeOrderingValue(orderingValue)) {
+ if (orderingFieldNames.size() == 1) {
+ fieldValues[1] = orderingValue;
+ } else {
+ List<Comparable> orderingValues = OrderingValues.getValues((ArrayComparable) orderingValue);
+ for (int i = 0; i < orderingValues.size(); i++) {
+ fieldValues[i + 1] = orderingValues.get(i);
+ }
+ }
+ }
+ return fieldValues;
+ }
+
+ public AppendResult flushAppend() throws IOException {
Review Comment:
Never used, can be removed?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogAppendHandle.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io;
+
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieDeltaWriteStat;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.log.AppendResult;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.util.ConfigUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ParquetUtils;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieAppendException;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.metadata.HoodieIndexVersion;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.stats.HoodieColumnRangeMetadata;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.util.Lazy;
+
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
+
+/**
+ * Append handle for native log files. Unlike {@link HoodieInlineLogAppendHandle}, this handle streams
+ * records directly into native format writers and does not buffer records or build inline log blocks.
+ */
+public class HoodieNativeLogAppendHandle<T, I, K, O> extends HoodieAppendHandle<T, I, K, O> {
+
+ private HoodieNativeLogFormatWriter writer;
+
+ public HoodieNativeLogAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, false, Collections.emptyMap());
+ }
+
+ public HoodieNativeLogAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier, Map<HeaderMetadataType, String> header) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, true, header);
+ }
+
+ public HoodieNativeLogAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, null, taskContextSupplier);
+ }
+
+ private HoodieNativeLogAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier, boolean preserveMetadata,
+ Map<HeaderMetadataType, String> header) {
+ super(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, preserveMetadata);
+ this.header.putAll(header);
+ }
+
+ @Override
+ protected void createLogWriterForAppend(String instantTime, Option<FileSlice> fileSliceOpt) {
+ try {
+ this.writer = new HoodieNativeLogFormatWriter(
+ storage.getDefaultBufferSize(),
+ storage,
+ FSUtils.constructAbsolutePath(hoodieTable.getMetaClient().getBasePath(), partitionPath),
+ fileId,
+ instantTime,
+ null,
+ writeToken,
+ config.getLogFileMaxSize(),
+ getLogCreationCallback(),
+ config.getWriteVersion(),
+ config,
+ writeSchemaWithMetaFields,
+ taskContextSupplier,
+ hoodieTable.getReaderContextFactoryForWrite().getContext().getRecordContext(),
+ Arrays.stream(ConfigUtils.getOrderingFields(recordProperties)).collect(Collectors.toList()));
+ } catch (IOException e) {
+ throw new HoodieException("Creating native log writer with fileId: " + fileId + ", "
+ + "delta commit time: " + instantTime + " error", e);
+ }
+ }
+
+ @Override
+ protected void writeInsertAndUpdate(HoodieSchema schema, HoodieRecord<T> hoodieRecord, boolean isUpdateRecord) throws IOException {
+ if (hoodieRecord.shouldIgnore(schema, recordProperties)) {
+ return;
+ }
+ HoodieRecord populatedRecord = hoodieRecord.prependMetaFields(
+ schema, writeSchemaWithMetaFields, populateMetadataFields(hoodieRecord), recordProperties);
+ String keyField = config.populateMetaFields()
+ ? HoodieRecord.RECORD_KEY_METADATA_FIELD
+ : hoodieTable.getMetaClient().getTableConfig().getRecordKeyFieldProp();
+ if (!writer.canWriteDataFile()) {
Review 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.parquet`
- `v2.log.parquet`
- `v1.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.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java:
##########
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io;
+
+import org.apache.hudi.common.engine.RecordContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemas;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.log.AppendResult;
+import org.apache.hudi.common.table.log.HoodieLogFormat;
+import org.apache.hudi.common.table.log.LogFileCreationCallback;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.table.read.BufferedRecords;
+import org.apache.hudi.common.util.JsonUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.OrderingValues;
+import org.apache.hudi.common.util.collection.ArrayComparable;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.io.storage.HoodieFileWriter;
+import org.apache.hudi.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.hudi.common.model.LogExtensions.DATA_LOG_EXTENSION;
+import static org.apache.hudi.common.model.LogExtensions.DELETE_LOG_EXTENSION;
+
+/**
+ * Writes MOR log blocks as native files, for example {@code .log.parquet} and {@code .deletes.parquet}.
+ */
+public class HoodieNativeLogFormatWriter extends HoodieLogFormat.Writer {
+
+ private static final int NATIVE_LOG_FORMAT_VERSION = 2;
+ private static final String LOG_FORMAT_METADATA_FOOTER_KEY = "hudi.log.format.metadata";
+
+ private final HoodieWriteConfig writeConfig;
+ private final HoodieSchema tableSchema;
+ private final TaskContextSupplier taskContextSupplier;
+ private final RecordContext recordContext;
+ private final List<String> orderingFieldNames;
+ private final Properties recordProperties;
+ private HoodieFileWriter dataFileWriter;
+ private HoodieFileWriter deleteFileWriter;
+ private HoodieLogFile dataLogFile;
+ private HoodieLogFile deleteLogFile;
+ private HoodieSchema deleteLogSchema;
+ private int currentAppendVersion = -1;
+ private List<AppendResult> lastAppendResults = new ArrayList<>();
+ private Option<Object> lastDataFileFormatMetadata = Option.empty();
+
+ public HoodieNativeLogFormatWriter(Integer bufferSize,
+ HoodieStorage storage,
+ StoragePath parentPath,
+ String logFileId,
+ String instantTime,
+ Integer logVersion,
+ String logWriteToken,
+ Long sizeThreshold,
+ LogFileCreationCallback fileCreationCallback,
+ HoodieTableVersion tableVersion,
+ HoodieWriteConfig writeConfig,
+ HoodieSchema tableSchema,
+ TaskContextSupplier taskContextSupplier,
+ RecordContext recordContext,
+ List<String> orderingFieldNames) throws IOException {
+ super(bufferSize, storage, parentPath, logFileId, DATA_LOG_EXTENSION, instantTime, logVersion, logWriteToken,
+ null, 0L, sizeThreshold, fileCreationCallback, tableVersion);
+ this.writeConfig = writeConfig;
+ this.tableSchema = tableSchema;
+ this.taskContextSupplier = taskContextSupplier;
+ this.recordContext = recordContext;
+ this.orderingFieldNames = orderingFieldNames;
+ this.recordProperties = new Properties();
+ this.recordProperties.putAll(writeConfig.getProps());
+ this.logFile = new HoodieLogFile(makeNativeLogPath(this.logVersion, DATA_LOG_EXTENSION), this.fileSize);
+ }
+
+ public List<AppendResult> getLastAppendResults() {
+ return lastAppendResults;
+ }
+
+ public Option<Object> getLastDataFileFormatMetadata() {
+ return lastDataFileFormatMetadata;
+ }
+
+ @Override
+ public AppendResult appendBlock(HoodieLogBlock block) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public AppendResult appendBlocks(List<HoodieLogBlock> blocks) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public long getCurrentSize() {
+ return lastAppendResults.stream().mapToLong(AppendResult::size).sum();
+ }
+
+ @Override
+ public void sync() {
+ // Native log files are write-once; closing the file writer publishes the file.
+ }
+
+ @Override
+ public void close() {
+ try {
+ closeFileWriters();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to close native log file writers", e);
Review Comment:
Here`IOException` is wrapped as RuntimeException; the codebase convention is HoodieIOException.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieNativeLogFormatWriter.java:
##########
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.io;
+
+import org.apache.hudi.common.engine.RecordContext;
+import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemas;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.log.AppendResult;
+import org.apache.hudi.common.table.log.HoodieLogFormat;
+import org.apache.hudi.common.table.log.LogFileCreationCallback;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
+import org.apache.hudi.common.table.read.BufferedRecord;
+import org.apache.hudi.common.table.read.BufferedRecords;
+import org.apache.hudi.common.util.JsonUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.OrderingValues;
+import org.apache.hudi.common.util.collection.ArrayComparable;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.io.storage.HoodieFileWriter;
+import org.apache.hudi.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.hudi.common.model.LogExtensions.DATA_LOG_EXTENSION;
+import static org.apache.hudi.common.model.LogExtensions.DELETE_LOG_EXTENSION;
+
+/**
+ * Writes MOR log blocks as native files, for example {@code .log.parquet} and {@code .deletes.parquet}.
+ */
+public class HoodieNativeLogFormatWriter extends HoodieLogFormat.Writer {
+
+ private static final int NATIVE_LOG_FORMAT_VERSION = 2;
+ private static final String LOG_FORMAT_METADATA_FOOTER_KEY = "hudi.log.format.metadata";
+
+ private final HoodieWriteConfig writeConfig;
+ private final HoodieSchema tableSchema;
+ private final TaskContextSupplier taskContextSupplier;
+ private final RecordContext recordContext;
+ private final List<String> orderingFieldNames;
+ private final Properties recordProperties;
+ private HoodieFileWriter dataFileWriter;
+ private HoodieFileWriter deleteFileWriter;
+ private HoodieLogFile dataLogFile;
+ private HoodieLogFile deleteLogFile;
+ private HoodieSchema deleteLogSchema;
+ private int currentAppendVersion = -1;
+ private List<AppendResult> lastAppendResults = new ArrayList<>();
+ private Option<Object> lastDataFileFormatMetadata = Option.empty();
+
+ public HoodieNativeLogFormatWriter(Integer bufferSize,
+ HoodieStorage storage,
+ StoragePath parentPath,
+ String logFileId,
+ String instantTime,
+ Integer logVersion,
+ String logWriteToken,
+ Long sizeThreshold,
+ LogFileCreationCallback fileCreationCallback,
+ HoodieTableVersion tableVersion,
+ HoodieWriteConfig writeConfig,
+ HoodieSchema tableSchema,
+ TaskContextSupplier taskContextSupplier,
+ RecordContext recordContext,
+ List<String> orderingFieldNames) throws IOException {
+ super(bufferSize, storage, parentPath, logFileId, DATA_LOG_EXTENSION, instantTime, logVersion, logWriteToken,
+ null, 0L, sizeThreshold, fileCreationCallback, tableVersion);
+ this.writeConfig = writeConfig;
+ this.tableSchema = tableSchema;
+ this.taskContextSupplier = taskContextSupplier;
+ this.recordContext = recordContext;
+ this.orderingFieldNames = orderingFieldNames;
+ this.recordProperties = new Properties();
+ this.recordProperties.putAll(writeConfig.getProps());
+ this.logFile = new HoodieLogFile(makeNativeLogPath(this.logVersion, DATA_LOG_EXTENSION), this.fileSize);
+ }
+
+ public List<AppendResult> getLastAppendResults() {
+ return lastAppendResults;
+ }
+
+ public Option<Object> getLastDataFileFormatMetadata() {
+ return lastDataFileFormatMetadata;
+ }
+
+ @Override
+ public AppendResult appendBlock(HoodieLogBlock block) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public AppendResult appendBlocks(List<HoodieLogBlock> blocks) {
+ throw new UnsupportedOperationException("HoodieNativeLogFormatWriter writes native records directly.");
+ }
+
+ @Override
+ public long getCurrentSize() {
+ return lastAppendResults.stream().mapToLong(AppendResult::size).sum();
+ }
+
+ @Override
+ public void sync() {
+ // Native log files are write-once; closing the file writer publishes the file.
+ }
+
+ @Override
+ public void close() {
+ try {
+ closeFileWriters();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to close native log file writers", e);
+ }
+ }
+
+ public boolean hasPendingWrites() {
+ return dataFileWriter != null || deleteFileWriter != null;
+ }
+
+ public boolean canWriteDataFile() {
+ return dataFileWriter == null || dataFileWriter.canWrite();
+ }
+
+ public void appendRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
+ ensureDataFileWriter(recordSchema);
+ dataFileWriter.write(record.getRecordKey(recordSchema, keyFieldName),
+ record, recordSchema, recordProperties);
+ }
+
+ public void appendDeleteRecord(HoodieRecord record, HoodieSchema recordSchema, String keyFieldName) throws IOException {
+ ensureDeleteFileWriter();
+ String recordKey = record.getRecordKey(recordSchema, keyFieldName);
+ Comparable orderingValue = record.getOrderingValue(
+ recordSchema, recordProperties, orderingFieldNames.toArray(new String[0]));
+ Object deleteEngineRecord = recordContext.constructEngineRecord(
+ deleteLogSchema, createDeleteLogFieldValues(recordKey, orderingValue));
+ // Keep isDelete=false here so RecordContext constructs a data-bearing HoodieRecord
+ // with the native delete-log row. The delete semantics come from the delete log file
+ // itself; setting isDelete=true would create a HoodieEmptyRecord and lose the row.
+ BufferedRecord deleteRecord = BufferedRecords.fromEngineRecord(
+ deleteEngineRecord, deleteLogSchema, recordContext, orderingValue, recordKey, false);
+ deleteFileWriter.write(recordKey, recordContext.constructHoodieRecord(deleteRecord, record.getPartitionPath()),
+ deleteLogSchema, recordProperties);
+ }
+
+ private Object[] createDeleteLogFieldValues(String recordKey, Comparable orderingValue) {
+ Object[] fieldValues = new Object[deleteLogSchema.getFields().size()];
+ fieldValues[0] = recordContext.convertValueToEngineType(recordKey);
+ if (!OrderingValues.isCommitTimeOrderingValue(orderingValue)) {
+ if (orderingFieldNames.size() == 1) {
+ fieldValues[1] = orderingValue;
+ } else {
+ List<Comparable> orderingValues = OrderingValues.getValues((ArrayComparable) orderingValue);
+ for (int i = 0; i < orderingValues.size(); i++) {
+ fieldValues[i + 1] = orderingValues.get(i);
+ }
+ }
+ }
+ return fieldValues;
+ }
+
+ public AppendResult flushAppend() throws IOException {
+ return flushAppend(Collections.emptyMap());
+ }
+
+ public AppendResult flushDataAppend(Map<HeaderMetadataType, String> header) throws IOException {
+ int appendVersion = currentAppendVersion;
+ lastAppendResults = new ArrayList<>();
+ lastDataFileFormatMetadata = Option.empty();
+ addFooterMetadataToDataFile(header);
+ if (dataFileWriter != null) {
+ dataFileWriter.close();
+ lastDataFileFormatMetadata = Option.ofNullable(dataFileWriter.getFileFormatMetadata());
+ dataFileWriter = null;
+ }
+ if (dataLogFile != null) {
+ lastAppendResults.add(toAppendResult(dataLogFile));
+ }
+ dataLogFile = null;
+ currentAppendVersion = -1;
+ return completeAppend(appendVersion);
+ }
+
+ public AppendResult flushAppend(Map<HeaderMetadataType, String> header) throws IOException {
+ int appendVersion = currentAppendVersion;
+ lastAppendResults = new ArrayList<>();
+ lastDataFileFormatMetadata = Option.empty();
+ addFooterMetadata(header);
+ closeFileWriters();
+ if (dataLogFile != null) {
+ lastAppendResults.add(toAppendResult(dataLogFile));
+ }
+ if (deleteLogFile != null) {
+ lastAppendResults.add(toAppendResult(deleteLogFile));
+ }
+ dataLogFile = null;
+ deleteLogFile = null;
+ currentAppendVersion = -1;
+ return completeAppend(appendVersion);
+ }
+
+ private void addFooterMetadata(Map<HeaderMetadataType, String> header) throws IOException {
+ Map<String, String> footerMetadata = getFooterMetadata(header);
+ if (dataFileWriter != null) {
+ dataFileWriter.addFooterMetadata(footerMetadata);
+ }
+ if (deleteFileWriter != null) {
+ deleteFileWriter.addFooterMetadata(footerMetadata);
+ }
+ }
+
+ private void addFooterMetadataToDataFile(Map<HeaderMetadataType, String> header) throws IOException {
+ if (dataFileWriter != null) {
+ dataFileWriter.addFooterMetadata(getFooterMetadata(header));
+ }
+ }
+
+ private Map<String, String> getFooterMetadata(Map<HeaderMetadataType, String> header) throws IOException {
+ Map<String, String> logFormatMetadata = new LinkedHashMap<>();
+ 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.
Review 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.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java:
##########
@@ -35,182 +34,251 @@
import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.model.MetadataValues;
import org.apache.hudi.common.schema.HoodieSchema;
-import org.apache.hudi.common.schema.HoodieSchemaField;
-import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.log.AppendResult;
-import org.apache.hudi.common.table.log.HoodieLogFormat;
-import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
-import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
-import org.apache.hudi.common.table.log.block.HoodieHFileDataBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
-import org.apache.hudi.common.table.log.block.HoodieParquetDataBlock;
import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator;
import org.apache.hudi.common.table.view.TableFileSystemView;
-import org.apache.hudi.common.util.ConfigUtils;
-import org.apache.hudi.common.util.DefaultSizeEstimator;
import org.apache.hudi.common.util.HoodieRecordUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
-import org.apache.hudi.common.util.SizeEstimator;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieAppendException;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieUpsertException;
-import org.apache.hudi.metadata.HoodieIndexVersion;
-import org.apache.hudi.metadata.HoodieTableMetadataUtil;
-import org.apache.hudi.stats.HoodieColumnRangeMetadata;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieTable;
-import org.apache.hudi.util.CommonClientUtils;
-import org.apache.hudi.util.Lazy;
import lombok.extern.slf4j.Slf4j;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_DYNAMIC_MAX_ENTRIES;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_FPP_VALUE;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_NUM_ENTRIES_VALUE;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_TYPE;
-import static org.apache.hudi.common.config.HoodieStorageConfig.HFILE_COMPRESSION_ALGORITHM_NAME;
-import static org.apache.hudi.common.config.HoodieStorageConfig.HFILE_WITH_BLOOM_FILTER_ENABLED;
-import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
-import static org.apache.hudi.metadata.HoodieTableMetadataUtil.collectColumnRangeMetadata;
-
/**
- * IO Operation to append data onto an existing file.
+ * Shared base for append handles writing merge-on-read log files.
+ *
+ * <p>This class owns the common append lifecycle:
+ * <ol>
+ * <li>Initialize write status and log writer state for the target file group.</li>
+ * <li>Iterate incoming {@link HoodieRecord}s and route each record as an insert/update or delete.</li>
+ * <li>Track write counters, record locations, write-status success/failure state, and runtime stats.</li>
+ * <li>Flush pending records to one or more physical log files and update {@link HoodieDeltaWriteStat}s.</li>
+ * <li>Close the record iterator and log writer, then publish secondary-index streaming stats when enabled.</li>
+ * </ol>
+ *
+ * <p>Concrete subclasses provide only the physical log representation. Inline-log handles buffer records
+ * into Hudi log blocks, while native-log handles stream records into engine-native file writers. Both variants
+ * share the same accounting and status update logic through this class.
*/
@Slf4j
-public class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O> {
- // This acts as the sequenceID for records written
+public abstract class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O> {
+
private static final AtomicLong RECORD_COUNTER = new AtomicLong(1);
- private static final int NUMBER_OF_RECORDS_TO_ESTIMATE_RECORD_SIZE = 100;
- // Buffer for holding records in memory before they are flushed to disk
- protected final List<HoodieRecord> recordList = new ArrayList<>();
- // Buffer for holding records (to be deleted), along with their position in log block, in memory before they are flushed to disk
- protected final List<Pair<DeleteRecord, Long>> recordsToDeleteWithPositions = new ArrayList<>();
- // Base file instant time of the record positions
- protected final Option<String> baseFileInstantTimeOfPositions;
- // Incoming records to be written to logs.
protected Iterator<HoodieRecord<T>> recordItr;
- // Writer to log into the file group's latest slice.
- protected HoodieLogFormat.Writer writer;
-
- protected final List<WriteStatus> statuses;
- // Total number of records written during appending
+ protected final List<WriteStatus> statuses = new ArrayList<>();
+ protected final Map<HeaderMetadataType, String> header = new HashMap<>();
+ protected final Properties recordProperties = new Properties();
+ protected final Option<String> baseFileInstantTimeOfPositions;
+ protected boolean isLogCompaction = false;
+ protected boolean useWriterSchema = false;
+ protected boolean doInit = true;
protected long recordsWritten = 0;
- // Total number of records deleted during appending
protected long recordsDeleted = 0;
- // Total number of records updated during appending
protected long updatedRecordsWritten = 0;
- // Total number of new records inserted into the delta file
protected long insertRecordsWritten = 0;
+ protected Option<HoodieLogBlock> appendDataBlock = Option.empty();
- // Average record size for a HoodieRecord. This size is updated at the end of every log block flushed to disk
- private long averageRecordSize = 0;
- // Flag used to initialize some metadata
- private boolean doInit = true;
- // Total number of bytes written during this append phase (an estimation)
- protected long estimatedNumberOfBytesWritten;
- // Number of records that must be written to meet the max block size for a log block
- private long numberOfRecords = 0;
- // Max block size to limit to for a log block
- private final long maxBlockSize = config.getLogFileDataBlockMaxSize();
- // Header metadata for a log block
- protected final Map<HeaderMetadataType, String> header = new HashMap<>();
- private final SizeEstimator<HoodieRecord> sizeEstimator;
- // This is used to distinguish between normal append and logcompaction's append operation.
- private boolean isLogCompaction = false;
- // use writer schema for log compaction.
- private boolean useWriterSchema = false;
-
- private final Properties recordProperties = new Properties();
- private final String[] orderingFields;
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, false);
+ }
- /**
- * This is used by log compaction only.
- */
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
- TaskContextSupplier taskContextSupplier, Map<HeaderMetadataType, String> header) {
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier, Map<HeaderMetadataType, String> header) {
this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, true);
- this.useWriterSchema = true;
- this.isLogCompaction = true;
this.header.putAll(header);
}
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr, TaskContextSupplier taskContextSupplier) {
- this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, false);
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, null, taskContextSupplier);
}
- private HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr, TaskContextSupplier taskContextSupplier, boolean preserveMetadata) {
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier, boolean preserveMetadata) {
super(config, instantTime, partitionPath, fileId, hoodieTable,
config.shouldWritePartialUpdates()
- // When enabling writing partial updates to the data blocks in log files,
- // i.e., partial update schema is set, the writer schema is the partial
- // schema containing the updated fields only
? Option.of(HoodieSchema.parse(config.getPartialUpdateSchema()))
: Option.empty(),
taskContextSupplier,
preserveMetadata);
this.recordItr = recordItr;
- this.sizeEstimator = getSizeEstimator();
- this.statuses = new ArrayList<>();
+ this.isLogCompaction = preserveMetadata;
+ this.useWriterSchema = preserveMetadata;
this.recordProperties.putAll(config.getProps());
- this.orderingFields = ConfigUtils.getOrderingFields(recordProperties);
- boolean shouldWriteRecordPositions = config.shouldWriteRecordPositions()
- // record positions supported only from table version 8
- && config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT);
- this.baseFileInstantTimeOfPositions = shouldWriteRecordPositions
+ this.baseFileInstantTimeOfPositions = config.shouldWriteRecordPositions()
+ && config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT)
? getBaseFileInstantTimeOfPositions()
: Option.empty();
}
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, TaskContextSupplier sparkTaskContextSupplier) {
- this(config, instantTime, hoodieTable, partitionPath, fileId, null, sparkTaskContextSupplier);
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, String partitionPath, String fileId,
+ HoodieTable<T, I, K, O> hoodieTable, Option<HoodieSchema> writerSchema,
+ TaskContextSupplier taskContextSupplier, boolean preserveMetadata) {
+ super(config, instantTime, partitionPath, fileId, hoodieTable, writerSchema, taskContextSupplier, preserveMetadata);
+ this.baseFileInstantTimeOfPositions = config.shouldWriteRecordPositions()
+ && config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT)
+ ? getBaseFileInstantTimeOfPositions()
+ : Option.empty();
}
- protected SizeEstimator<HoodieRecord> getSizeEstimator() {
- return new DefaultSizeEstimator<>();
+ /**
+ * Writes all records from the incoming iterator and flushes remaining pending data.
+ */
+ public void doAppend() {
+ while (recordItr.hasNext()) {
+ HoodieRecord<T> record = recordItr.next();
+ writeRecord(record);
+ }
+ flushAppend();
}
- protected Option<String> getBaseFileInstantTimeOfPositions() {
- return hoodieTable.getHoodieView().getLatestBaseFile(partitionPath, fileId)
- .map(HoodieBaseFile::getCommitTime);
+ /**
+ * Writes one record for APIs that call the handle directly instead of using {@link #doAppend()}.
+ */
+ @Override
+ protected void doWrite(HoodieRecord record, HoodieSchema schema, TypedProperties props) {
+ writeRecord(record);
+ }
+
+ /**
+ * Writes a keyed record map, primarily used by log compaction paths.
+ */
+ @Override
+ public void write(Map<String, HoodieRecord<T>> recordMap) {
+ try {
+ for (Map.Entry<String, HoodieRecord<T>> entry : recordMap.entrySet()) {
+ HoodieRecord<T> record = entry.getValue();
+ writeRecord(record);
+ }
+ flushAppend();
+ } catch (Exception e) {
+ throw new HoodieUpsertException("Failed to compact blocks for fileId " + fileId, e);
+ }
+ }
+
+ /**
+ * Flushes pending data, closes resources, finalizes log file sizes, and publishes secondary-index stats.
+ */
+ @Override
+ public List<WriteStatus> close() {
+ try {
+ if (isClosed()) {
+ return statuses.isEmpty() ? java.util.Collections.singletonList(writeStatus) : statuses;
Review Comment:
uses fully-qualified java.util.Collections.singletonList(...) since the import was dropped from the base class; prefer re-adding the import.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "cshuo (via GitHub)" <gi...@apache.org>.
cshuo commented on code in PR #19067:
URL: https://github.com/apache/hudi/pull/19067#discussion_r3473478197
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java:
##########
@@ -35,182 +34,251 @@
import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.model.MetadataValues;
import org.apache.hudi.common.schema.HoodieSchema;
-import org.apache.hudi.common.schema.HoodieSchemaField;
-import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.log.AppendResult;
-import org.apache.hudi.common.table.log.HoodieLogFormat;
-import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
-import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
-import org.apache.hudi.common.table.log.block.HoodieHFileDataBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
-import org.apache.hudi.common.table.log.block.HoodieParquetDataBlock;
import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator;
import org.apache.hudi.common.table.view.TableFileSystemView;
-import org.apache.hudi.common.util.ConfigUtils;
-import org.apache.hudi.common.util.DefaultSizeEstimator;
import org.apache.hudi.common.util.HoodieRecordUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
-import org.apache.hudi.common.util.SizeEstimator;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieAppendException;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieUpsertException;
-import org.apache.hudi.metadata.HoodieIndexVersion;
-import org.apache.hudi.metadata.HoodieTableMetadataUtil;
-import org.apache.hudi.stats.HoodieColumnRangeMetadata;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieTable;
-import org.apache.hudi.util.CommonClientUtils;
-import org.apache.hudi.util.Lazy;
import lombok.extern.slf4j.Slf4j;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_DYNAMIC_MAX_ENTRIES;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_FPP_VALUE;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_NUM_ENTRIES_VALUE;
-import static org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_TYPE;
-import static org.apache.hudi.common.config.HoodieStorageConfig.HFILE_COMPRESSION_ALGORITHM_NAME;
-import static org.apache.hudi.common.config.HoodieStorageConfig.HFILE_WITH_BLOOM_FILTER_ENABLED;
-import static org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
-import static org.apache.hudi.metadata.HoodieTableMetadataUtil.collectColumnRangeMetadata;
-
/**
- * IO Operation to append data onto an existing file.
+ * Shared base for append handles writing merge-on-read log files.
+ *
+ * <p>This class owns the common append lifecycle:
+ * <ol>
+ * <li>Initialize write status and log writer state for the target file group.</li>
+ * <li>Iterate incoming {@link HoodieRecord}s and route each record as an insert/update or delete.</li>
+ * <li>Track write counters, record locations, write-status success/failure state, and runtime stats.</li>
+ * <li>Flush pending records to one or more physical log files and update {@link HoodieDeltaWriteStat}s.</li>
+ * <li>Close the record iterator and log writer, then publish secondary-index streaming stats when enabled.</li>
+ * </ol>
+ *
+ * <p>Concrete subclasses provide only the physical log representation. Inline-log handles buffer records
+ * into Hudi log blocks, while native-log handles stream records into engine-native file writers. Both variants
+ * share the same accounting and status update logic through this class.
*/
@Slf4j
-public class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O> {
- // This acts as the sequenceID for records written
+public abstract class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O> {
+
private static final AtomicLong RECORD_COUNTER = new AtomicLong(1);
- private static final int NUMBER_OF_RECORDS_TO_ESTIMATE_RECORD_SIZE = 100;
- // Buffer for holding records in memory before they are flushed to disk
- protected final List<HoodieRecord> recordList = new ArrayList<>();
- // Buffer for holding records (to be deleted), along with their position in log block, in memory before they are flushed to disk
- protected final List<Pair<DeleteRecord, Long>> recordsToDeleteWithPositions = new ArrayList<>();
- // Base file instant time of the record positions
- protected final Option<String> baseFileInstantTimeOfPositions;
- // Incoming records to be written to logs.
protected Iterator<HoodieRecord<T>> recordItr;
- // Writer to log into the file group's latest slice.
- protected HoodieLogFormat.Writer writer;
-
- protected final List<WriteStatus> statuses;
- // Total number of records written during appending
+ protected final List<WriteStatus> statuses = new ArrayList<>();
+ protected final Map<HeaderMetadataType, String> header = new HashMap<>();
+ protected final Properties recordProperties = new Properties();
+ protected final Option<String> baseFileInstantTimeOfPositions;
+ protected boolean isLogCompaction = false;
+ protected boolean useWriterSchema = false;
+ protected boolean doInit = true;
protected long recordsWritten = 0;
- // Total number of records deleted during appending
protected long recordsDeleted = 0;
- // Total number of records updated during appending
protected long updatedRecordsWritten = 0;
- // Total number of new records inserted into the delta file
protected long insertRecordsWritten = 0;
+ protected Option<HoodieLogBlock> appendDataBlock = Option.empty();
- // Average record size for a HoodieRecord. This size is updated at the end of every log block flushed to disk
- private long averageRecordSize = 0;
- // Flag used to initialize some metadata
- private boolean doInit = true;
- // Total number of bytes written during this append phase (an estimation)
- protected long estimatedNumberOfBytesWritten;
- // Number of records that must be written to meet the max block size for a log block
- private long numberOfRecords = 0;
- // Max block size to limit to for a log block
- private final long maxBlockSize = config.getLogFileDataBlockMaxSize();
- // Header metadata for a log block
- protected final Map<HeaderMetadataType, String> header = new HashMap<>();
- private final SizeEstimator<HoodieRecord> sizeEstimator;
- // This is used to distinguish between normal append and logcompaction's append operation.
- private boolean isLogCompaction = false;
- // use writer schema for log compaction.
- private boolean useWriterSchema = false;
-
- private final Properties recordProperties = new Properties();
- private final String[] orderingFields;
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, false);
+ }
- /**
- * This is used by log compaction only.
- */
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
- TaskContextSupplier taskContextSupplier, Map<HeaderMetadataType, String> header) {
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier, Map<HeaderMetadataType, String> header) {
this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, true);
- this.useWriterSchema = true;
- this.isLogCompaction = true;
this.header.putAll(header);
}
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr, TaskContextSupplier taskContextSupplier) {
- this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier, false);
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, null, taskContextSupplier);
}
- private HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr, TaskContextSupplier taskContextSupplier, boolean preserveMetadata) {
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId, Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier, boolean preserveMetadata) {
super(config, instantTime, partitionPath, fileId, hoodieTable,
config.shouldWritePartialUpdates()
- // When enabling writing partial updates to the data blocks in log files,
- // i.e., partial update schema is set, the writer schema is the partial
- // schema containing the updated fields only
? Option.of(HoodieSchema.parse(config.getPartialUpdateSchema()))
: Option.empty(),
taskContextSupplier,
preserveMetadata);
this.recordItr = recordItr;
- this.sizeEstimator = getSizeEstimator();
- this.statuses = new ArrayList<>();
+ this.isLogCompaction = preserveMetadata;
+ this.useWriterSchema = preserveMetadata;
this.recordProperties.putAll(config.getProps());
- this.orderingFields = ConfigUtils.getOrderingFields(recordProperties);
- boolean shouldWriteRecordPositions = config.shouldWriteRecordPositions()
- // record positions supported only from table version 8
- && config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT);
- this.baseFileInstantTimeOfPositions = shouldWriteRecordPositions
+ this.baseFileInstantTimeOfPositions = config.shouldWriteRecordPositions()
+ && config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT)
? getBaseFileInstantTimeOfPositions()
: Option.empty();
}
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId, TaskContextSupplier sparkTaskContextSupplier) {
- this(config, instantTime, hoodieTable, partitionPath, fileId, null, sparkTaskContextSupplier);
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime, String partitionPath, String fileId,
+ HoodieTable<T, I, K, O> hoodieTable, Option<HoodieSchema> writerSchema,
+ TaskContextSupplier taskContextSupplier, boolean preserveMetadata) {
+ super(config, instantTime, partitionPath, fileId, hoodieTable, writerSchema, taskContextSupplier, preserveMetadata);
+ this.baseFileInstantTimeOfPositions = config.shouldWriteRecordPositions()
+ && config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT)
+ ? getBaseFileInstantTimeOfPositions()
+ : Option.empty();
}
- protected SizeEstimator<HoodieRecord> getSizeEstimator() {
- return new DefaultSizeEstimator<>();
+ /**
+ * Writes all records from the incoming iterator and flushes remaining pending data.
+ */
+ public void doAppend() {
+ while (recordItr.hasNext()) {
+ HoodieRecord<T> record = recordItr.next();
+ writeRecord(record);
+ }
+ flushAppend();
}
- protected Option<String> getBaseFileInstantTimeOfPositions() {
- return hoodieTable.getHoodieView().getLatestBaseFile(partitionPath, fileId)
- .map(HoodieBaseFile::getCommitTime);
+ /**
+ * Writes one record for APIs that call the handle directly instead of using {@link #doAppend()}.
+ */
+ @Override
+ protected void doWrite(HoodieRecord record, HoodieSchema schema, TypedProperties props) {
+ writeRecord(record);
+ }
+
+ /**
+ * Writes a keyed record map, primarily used by log compaction paths.
+ */
+ @Override
+ public void write(Map<String, HoodieRecord<T>> recordMap) {
+ try {
+ for (Map.Entry<String, HoodieRecord<T>> entry : recordMap.entrySet()) {
+ HoodieRecord<T> record = entry.getValue();
+ writeRecord(record);
+ }
+ flushAppend();
+ } catch (Exception e) {
+ throw new HoodieUpsertException("Failed to compact blocks for fileId " + fileId, e);
+ }
+ }
+
+ /**
+ * Flushes pending data, closes resources, finalizes log file sizes, and publishes secondary-index stats.
+ */
+ @Override
+ public List<WriteStatus> close() {
+ try {
+ if (isClosed()) {
+ return statuses.isEmpty() ? java.util.Collections.singletonList(writeStatus) : statuses;
Review Comment:
Uses fully-qualified java.util.Collections.singletonList(...) since the import was dropped from the base class; prefer re-adding the import.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support write native format log files [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19067:
URL: https://github.com/apache/hudi/pull/19067#issuecomment-4798429500
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14918",
"triggerID" : "4d7144e4ca3aa177134d3458b90bd369e7e95220",
"triggerType" : "PUSH"
}, {
"hash" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921",
"triggerID" : "2af16f108cc18819cd2bb8b48f8574474fb91d91",
"triggerType" : "PUSH"
}, {
"hash" : "1d94ecea2b8892440d929894272a427f045f44d2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14926",
"triggerID" : "1d94ecea2b8892440d929894272a427f045f44d2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 2af16f108cc18819cd2bb8b48f8574474fb91d91 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14921)
* 1d94ecea2b8892440d929894272a427f045f44d2 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=14926)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org