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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
