cshuo commented on code in PR #19067:
URL: https://github.com/apache/hudi/pull/19067#discussion_r3473478197
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java:
##########
@@ -35,182 +34,251 @@
import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.model.MetadataValues;
import org.apache.hudi.common.schema.HoodieSchema;
-import org.apache.hudi.common.schema.HoodieSchemaField;
-import org.apache.hudi.common.schema.HoodieSchemaUtils;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.log.AppendResult;
-import org.apache.hudi.common.table.log.HoodieLogFormat;
-import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
-import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
-import org.apache.hudi.common.table.log.block.HoodieHFileDataBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import
org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
-import org.apache.hudi.common.table.log.block.HoodieParquetDataBlock;
import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator;
import org.apache.hudi.common.table.view.TableFileSystemView;
-import org.apache.hudi.common.util.ConfigUtils;
-import org.apache.hudi.common.util.DefaultSizeEstimator;
import org.apache.hudi.common.util.HoodieRecordUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
-import org.apache.hudi.common.util.SizeEstimator;
-import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieAppendException;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieUpsertException;
-import org.apache.hudi.metadata.HoodieIndexVersion;
-import org.apache.hudi.metadata.HoodieTableMetadataUtil;
-import org.apache.hudi.stats.HoodieColumnRangeMetadata;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieTable;
-import org.apache.hudi.util.CommonClientUtils;
-import org.apache.hudi.util.Lazy;
import lombok.extern.slf4j.Slf4j;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
-import static
org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_DYNAMIC_MAX_ENTRIES;
-import static
org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_FPP_VALUE;
-import static
org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_NUM_ENTRIES_VALUE;
-import static
org.apache.hudi.common.config.HoodieStorageConfig.BLOOM_FILTER_TYPE;
-import static
org.apache.hudi.common.config.HoodieStorageConfig.HFILE_COMPRESSION_ALGORITHM_NAME;
-import static
org.apache.hudi.common.config.HoodieStorageConfig.HFILE_WITH_BLOOM_FILTER_ENABLED;
-import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS;
-import static
org.apache.hudi.metadata.HoodieTableMetadataUtil.collectColumnRangeMetadata;
-
/**
- * IO Operation to append data onto an existing file.
+ * Shared base for append handles writing merge-on-read log files.
+ *
+ * <p>This class owns the common append lifecycle:
+ * <ol>
+ * <li>Initialize write status and log writer state for the target file
group.</li>
+ * <li>Iterate incoming {@link HoodieRecord}s and route each record as an
insert/update or delete.</li>
+ * <li>Track write counters, record locations, write-status success/failure
state, and runtime stats.</li>
+ * <li>Flush pending records to one or more physical log files and update
{@link HoodieDeltaWriteStat}s.</li>
+ * <li>Close the record iterator and log writer, then publish
secondary-index streaming stats when enabled.</li>
+ * </ol>
+ *
+ * <p>Concrete subclasses provide only the physical log representation.
Inline-log handles buffer records
+ * into Hudi log blocks, while native-log handles stream records into
engine-native file writers. Both variants
+ * share the same accounting and status update logic through this class.
*/
@Slf4j
-public class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K,
O> {
- // This acts as the sequenceID for records written
+public abstract class HoodieAppendHandle<T, I, K, O> extends
HoodieWriteHandle<T, I, K, O> {
+
private static final AtomicLong RECORD_COUNTER = new AtomicLong(1);
- private static final int NUMBER_OF_RECORDS_TO_ESTIMATE_RECORD_SIZE = 100;
- // Buffer for holding records in memory before they are flushed to disk
- protected final List<HoodieRecord> recordList = new ArrayList<>();
- // Buffer for holding records (to be deleted), along with their position in
log block, in memory before they are flushed to disk
- protected final List<Pair<DeleteRecord, Long>> recordsToDeleteWithPositions
= new ArrayList<>();
- // Base file instant time of the record positions
- protected final Option<String> baseFileInstantTimeOfPositions;
- // Incoming records to be written to logs.
protected Iterator<HoodieRecord<T>> recordItr;
- // Writer to log into the file group's latest slice.
- protected HoodieLogFormat.Writer writer;
-
- protected final List<WriteStatus> statuses;
- // Total number of records written during appending
+ protected final List<WriteStatus> statuses = new ArrayList<>();
+ protected final Map<HeaderMetadataType, String> header = new HashMap<>();
+ protected final Properties recordProperties = new Properties();
+ protected final Option<String> baseFileInstantTimeOfPositions;
+ protected boolean isLogCompaction = false;
+ protected boolean useWriterSchema = false;
+ protected boolean doInit = true;
protected long recordsWritten = 0;
- // Total number of records deleted during appending
protected long recordsDeleted = 0;
- // Total number of records updated during appending
protected long updatedRecordsWritten = 0;
- // Total number of new records inserted into the delta file
protected long insertRecordsWritten = 0;
+ protected Option<HoodieLogBlock> appendDataBlock = Option.empty();
- // Average record size for a HoodieRecord. This size is updated at the end
of every log block flushed to disk
- private long averageRecordSize = 0;
- // Flag used to initialize some metadata
- private boolean doInit = true;
- // Total number of bytes written during this append phase (an estimation)
- protected long estimatedNumberOfBytesWritten;
- // Number of records that must be written to meet the max block size for a
log block
- private long numberOfRecords = 0;
- // Max block size to limit to for a log block
- private final long maxBlockSize = config.getLogFileDataBlockMaxSize();
- // Header metadata for a log block
- protected final Map<HeaderMetadataType, String> header = new HashMap<>();
- private final SizeEstimator<HoodieRecord> sizeEstimator;
- // This is used to distinguish between normal append and logcompaction's
append operation.
- private boolean isLogCompaction = false;
- // use writer schema for log compaction.
- private boolean useWriterSchema = false;
-
- private final Properties recordProperties = new Properties();
- private final String[] orderingFields;
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId,
Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr,
taskContextSupplier, false);
+ }
- /**
- * This is used by log compaction only.
- */
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId,
Iterator<HoodieRecord<T>> recordItr,
- TaskContextSupplier taskContextSupplier,
Map<HeaderMetadataType, String> header) {
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId,
Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier,
Map<HeaderMetadataType, String> header) {
this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr,
taskContextSupplier, true);
- this.useWriterSchema = true;
- this.isLogCompaction = true;
this.header.putAll(header);
}
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId,
Iterator<HoodieRecord<T>> recordItr, TaskContextSupplier taskContextSupplier) {
- this(config, instantTime, hoodieTable, partitionPath, fileId, recordItr,
taskContextSupplier, false);
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId,
TaskContextSupplier taskContextSupplier) {
+ this(config, instantTime, hoodieTable, partitionPath, fileId, null,
taskContextSupplier);
}
- private HoodieAppendHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId,
Iterator<HoodieRecord<T>> recordItr, TaskContextSupplier taskContextSupplier,
boolean preserveMetadata) {
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
+ String partitionPath, String fileId,
Iterator<HoodieRecord<T>> recordItr,
+ TaskContextSupplier taskContextSupplier,
boolean preserveMetadata) {
super(config, instantTime, partitionPath, fileId, hoodieTable,
config.shouldWritePartialUpdates()
- // When enabling writing partial updates to the data blocks in log
files,
- // i.e., partial update schema is set, the writer schema is the
partial
- // schema containing the updated fields only
? Option.of(HoodieSchema.parse(config.getPartialUpdateSchema()))
: Option.empty(),
taskContextSupplier,
preserveMetadata);
this.recordItr = recordItr;
- this.sizeEstimator = getSizeEstimator();
- this.statuses = new ArrayList<>();
+ this.isLogCompaction = preserveMetadata;
+ this.useWriterSchema = preserveMetadata;
this.recordProperties.putAll(config.getProps());
- this.orderingFields = ConfigUtils.getOrderingFields(recordProperties);
- boolean shouldWriteRecordPositions = config.shouldWriteRecordPositions()
- // record positions supported only from table version 8
- &&
config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT);
- this.baseFileInstantTimeOfPositions = shouldWriteRecordPositions
+ this.baseFileInstantTimeOfPositions = config.shouldWriteRecordPositions()
+ &&
config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT)
? getBaseFileInstantTimeOfPositions()
: Option.empty();
}
- public HoodieAppendHandle(HoodieWriteConfig config, String instantTime,
HoodieTable<T, I, K, O> hoodieTable,
- String partitionPath, String fileId,
TaskContextSupplier sparkTaskContextSupplier) {
- this(config, instantTime, hoodieTable, partitionPath, fileId, null,
sparkTaskContextSupplier);
+ protected HoodieAppendHandle(HoodieWriteConfig config, String instantTime,
String partitionPath, String fileId,
+ HoodieTable<T, I, K, O> hoodieTable,
Option<HoodieSchema> writerSchema,
+ TaskContextSupplier taskContextSupplier,
boolean preserveMetadata) {
+ super(config, instantTime, partitionPath, fileId, hoodieTable,
writerSchema, taskContextSupplier, preserveMetadata);
+ this.baseFileInstantTimeOfPositions = config.shouldWriteRecordPositions()
+ &&
config.getWriteVersion().greaterThanOrEquals(HoodieTableVersion.EIGHT)
+ ? getBaseFileInstantTimeOfPositions()
+ : Option.empty();
}
- protected SizeEstimator<HoodieRecord> getSizeEstimator() {
- return new DefaultSizeEstimator<>();
+ /**
+ * Writes all records from the incoming iterator and flushes remaining
pending data.
+ */
+ public void doAppend() {
+ while (recordItr.hasNext()) {
+ HoodieRecord<T> record = recordItr.next();
+ writeRecord(record);
+ }
+ flushAppend();
}
- protected Option<String> getBaseFileInstantTimeOfPositions() {
- return hoodieTable.getHoodieView().getLatestBaseFile(partitionPath, fileId)
- .map(HoodieBaseFile::getCommitTime);
+ /**
+ * Writes one record for APIs that call the handle directly instead of using
{@link #doAppend()}.
+ */
+ @Override
+ protected void doWrite(HoodieRecord record, HoodieSchema schema,
TypedProperties props) {
+ writeRecord(record);
+ }
+
+ /**
+ * Writes a keyed record map, primarily used by log compaction paths.
+ */
+ @Override
+ public void write(Map<String, HoodieRecord<T>> recordMap) {
+ try {
+ for (Map.Entry<String, HoodieRecord<T>> entry : recordMap.entrySet()) {
+ HoodieRecord<T> record = entry.getValue();
+ writeRecord(record);
+ }
+ flushAppend();
+ } catch (Exception e) {
+ throw new HoodieUpsertException("Failed to compact blocks for fileId " +
fileId, e);
+ }
+ }
+
+ /**
+ * Flushes pending data, closes resources, finalizes log file sizes, and
publishes secondary-index stats.
+ */
+ @Override
+ public List<WriteStatus> close() {
+ try {
+ if (isClosed()) {
+ return statuses.isEmpty() ?
java.util.Collections.singletonList(writeStatus) : statuses;
Review Comment:
Uses fully-qualified java.util.Collections.singletonList(...) since the
import was dropped from the base class; prefer re-adding the import.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
