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