danny0405 commented on code in PR #18987: URL: https://github.com/apache/hudi/pull/18987#discussion_r3459576492
########## hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/LsmFileGroupRecordIterator.java: ########## @@ -0,0 +1,539 @@ +/* + * 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.common.table.read.lsm; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.engine.HoodieReaderContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.model.HoodieLogFile; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemas; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.read.BaseFileUpdateCallback; +import org.apache.hudi.common.table.read.BufferedRecord; +import org.apache.hudi.common.table.read.BufferedRecordMerger; +import org.apache.hudi.common.table.read.BufferedRecordMergerFactory; +import org.apache.hudi.common.table.read.BufferedRecords; +import org.apache.hudi.common.table.read.HoodieReadStats; +import org.apache.hudi.common.table.read.InputSplit; +import org.apache.hudi.common.table.read.ReaderParameters; +import org.apache.hudi.common.table.read.UpdateProcessor; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.common.util.VisibleForTesting; +import org.apache.hudi.common.util.collection.ClosableIterator; +import org.apache.hudi.common.util.collection.CloseableMappingIterator; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.StoragePath; +import org.apache.hudi.storage.StoragePathInfo; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.Set; +import java.util.function.UnaryOperator; + +import static org.apache.hudi.common.config.HoodieMemoryConfig.SPILLABLE_MAP_BASE_PATH; +import static org.apache.hudi.common.config.HoodieReaderConfig.LSM_SORT_MERGE_SPILL_THRESHOLD; +import static org.apache.hudi.common.schema.HoodieSchemaCompatibility.areSchemasProjectionEquivalent; +import static org.apache.hudi.io.util.FileIOUtils.getDefaultSpillableMapBasePath; + +/** + * Streaming sorted-merge iterator for RFC-103 LSM file groups. + * + * <p>The iterator merges one optional L1/base sorted run with zero or more L0/native parquet log + * runs. Every input file must already be sorted by record key; this class does not sort records + * within a file. The active head record from each participating run is tracked by a loser tree, + * which provides efficient k-way merge behavior while preserving deterministic source ordering for + * records with the same key. + * + * <p>Merge order follows the same conflict resolution model as {@code HoodieFileGroupReader}: the + * L1/base file is processed first, and L0 log files are processed in file-group log order so newer + * log instants or versions can win when ordering values tie. Native delete logs contain only delete + * metadata and are converted into {@link BufferedRecord} delete records before entering the merge. + * + * <p>To reduce open-reader memory pressure when a file group has many L0 runs, the iterator can spill + * selected L0 file iterators to sequential temporary files. Spilling changes how an input run is + * buffered, but not its merge order or merge semantics. The L1/base iterator is always kept direct, + * and native delete logs plus smaller L0 files are prioritized for direct reading. + */ +public class LsmFileGroupRecordIterator<T> implements ClosableIterator<BufferedRecord<T>> { + + private static final String DELETE_LOG_RECORD_KEY_FIELD = "record_key"; + private static final List<String> DELETE_LOG_ORDERING_FIELD_NAMES = Arrays.asList("ordering_val"); + + private final HoodieReaderContext<T> readerContext; + private final HoodieStorage storage; + private final InputSplit inputSplit; + private final HoodieSchema readerSchema; + private final List<String> orderingFieldNames; + private final boolean includeBaseFile; + private final BufferedRecordMerger<T> bufferedRecordMerger; + private final UpdateProcessor<T> updateProcessor; + private final LoserTree<T> readers; + private final int spillThreshold; + private final String spillBasePath; + private BufferedRecord<T> nextRecord; + + /** + * Creates an iterator that merges both the L1/base file, when present, and all L0 native log files. + */ + public LsmFileGroupRecordIterator(HoodieReaderContext<T> readerContext, + HoodieStorage storage, + InputSplit inputSplit, + List<String> orderingFieldNames, + HoodieTableMetaClient metaClient, + TypedProperties props, + ReaderParameters readerParameters, + HoodieReadStats readStats, + Option<BaseFileUpdateCallback<T>> fileGroupUpdateCallback) throws IOException { + this(readerContext, storage, inputSplit, orderingFieldNames, metaClient, props, readerParameters, readStats, fileGroupUpdateCallback, true); + } + + /** + * Creates an iterator over an LSM file group. + * + * @param includeBaseFile whether the L1/base file should be included in the merge. Passing + * {@code false} produces a log-only view for callers that only need L0 data. + */ + public LsmFileGroupRecordIterator(HoodieReaderContext<T> readerContext, + HoodieStorage storage, + InputSplit inputSplit, + List<String> orderingFieldNames, + HoodieTableMetaClient metaClient, + TypedProperties props, + ReaderParameters readerParameters, + HoodieReadStats readStats, + Option<BaseFileUpdateCallback<T>> fileGroupUpdateCallback, + boolean includeBaseFile) throws IOException { + this.readerContext = readerContext; + this.storage = storage; + this.inputSplit = inputSplit; + this.readerSchema = readerContext.getSchemaHandler().getRequiredSchema(); + this.orderingFieldNames = orderingFieldNames; + this.includeBaseFile = includeBaseFile; + this.bufferedRecordMerger = BufferedRecordMergerFactory.create( + readerContext, readerContext.getMergeMode(), false, readerContext.getRecordMerger(), + readerSchema, readerContext.getPayloadClasses(props), props, metaClient.getTableConfig().getPartialUpdateMode()); + this.updateProcessor = UpdateProcessor.create(readStats, readerContext, readerParameters.isEmitDeletes(), fileGroupUpdateCallback, props); + this.spillThreshold = Math.max(0, props.getInteger(LSM_SORT_MERGE_SPILL_THRESHOLD.key(), LSM_SORT_MERGE_SPILL_THRESHOLD.defaultValue())); + this.spillBasePath = props.getString(SPILLABLE_MAP_BASE_PATH.key(), getDefaultSpillableMapBasePath()); + this.readers = new LoserTree<>(initializeReaders()); + } + + /** + * Builds one sorted-run reader for each sorted run that has at least one record. + * + * <p>The assigned {@code mergeOrder} is the stable source precedence used when multiple runs expose + * the same key. It is assigned before spill selection so direct and spilled iterators remain + * semantically identical during the loser-tree merge. + */ + private List<SortedRunReader<T>> initializeReaders() throws IOException { + List<SortedRunReader<T>> sortedRunReaders = new ArrayList<>(); + int mergeOrder = 0; + boolean hasBaseFileReader = includeBaseFile && inputSplit.getBaseFileOption().isPresent(); + if (hasBaseFileReader) { + addReader(sortedRunReaders, mergeOrder++, createBaseFileIterator(inputSplit.getBaseFileOption().get())); + } + + List<LogReaderSpec> logReaderSpecs = new ArrayList<>(); + for (HoodieLogFile logFile : inputSplit.getLogFiles()) { + logReaderSpecs.add(new LogReaderSpec(mergeOrder++, logFile)); + } + Set<Integer> directLogMergeOrders = selectDirectLogMergeOrders(logReaderSpecs, hasBaseFileReader); + for (LogReaderSpec spec : logReaderSpecs) { + ClosableIterator<BufferedRecord<T>> iterator = createFileIterator(spec.logFile.getPathInfo(), spec.logFile.getPath(), spec.logFile.getFileSize()); + addReader(sortedRunReaders, spec.mergeOrder, maybeSpillIterator(directLogMergeOrders.contains(spec.mergeOrder), iterator)); + } + return sortedRunReaders; + } + + /** + * Selects which L0 log readers stay direct under the configured spill threshold. + * + * <p>The base/L1 reader consumes one direct-reader slot when present and is never spilled. Remaining + * direct-reader budget is spent on native delete logs first, then smaller log files, because those + * readers tend to be cheaper to keep open while avoiding unnecessary spill materialization. + */ + private Set<Integer> selectDirectLogMergeOrders(List<LogReaderSpec> logReaderSpecs, boolean hasBaseFileReader) { + return selectDirectLogMergeOrders(logReaderSpecs, hasBaseFileReader, spillThreshold); + } + + @VisibleForTesting + static Set<Integer> selectDirectLogMergeOrders(List<LogReaderSpec> logReaderSpecs, + boolean hasBaseFileReader, + int spillThreshold) { + int directLogBudget = spillThreshold - (hasBaseFileReader ? 1 : 0); + if (directLogBudget <= 0) { + return new HashSet<>(); + } + Set<Integer> directMergeOrders = new HashSet<>(); + logReaderSpecs.stream() + .sorted(Comparator + .comparing((LogReaderSpec spec) -> !spec.nativeDeleteLog) + .thenComparingLong(spec -> spec.fileSize) + .thenComparingInt(spec -> spec.mergeOrder)) + .limit(directLogBudget) + .forEach(spec -> directMergeOrders.add(spec.mergeOrder)); + return directMergeOrders; + } + + /** + * Returns the original iterator when it is selected for direct reading, otherwise materializes it + * into a sequential spill iterator. + */ + private ClosableIterator<BufferedRecord<T>> maybeSpillIterator(boolean directReader, + ClosableIterator<BufferedRecord<T>> iterator) { + if (directReader) { + return iterator; + } + return new SpillableLsmRecordIterator<>(iterator, readerContext.getRecordSerializer(), readerContext.getRecordContext(), spillBasePath); + } + + /** + * Metadata used only for choosing the direct-versus-spilled L0 reader plan. + */ + @VisibleForTesting + static class LogReaderSpec { + final int mergeOrder; + final HoodieLogFile logFile; + final boolean nativeDeleteLog; + final long fileSize; + + LogReaderSpec(int mergeOrder, HoodieLogFile logFile) { + this.mergeOrder = mergeOrder; + this.logFile = logFile; + this.nativeDeleteLog = FSUtils.isNativeDeleteLogFile(logFile.getFileName()); + this.fileSize = logFile.getFileSize() >= 0 ? logFile.getFileSize() : Long.MAX_VALUE; + } + } + + /** + * Adds a reader to the merge only when the underlying sorted run contains at least one record. + */ + private void addReader(List<SortedRunReader<T>> sortedRunReaders, int mergeOrder, ClosableIterator<BufferedRecord<T>> iterator) { + SortedRunReader<T> sortedRunReader = new SortedRunReader<>(mergeOrder, iterator); + if (sortedRunReader.advance()) { + sortedRunReaders.add(sortedRunReader); + } else { + sortedRunReader.close(); + } + } + + /** + * Creates the L1/base sorted-run iterator. + */ + private ClosableIterator<BufferedRecord<T>> createBaseFileIterator(HoodieBaseFile baseFile) throws IOException { + if (baseFile.getBootstrapBaseFile().isPresent()) { + // Bootstrap base files require joining the skeleton file with the external data file. + // Keep that path on HoodieFileGroupReader until the LSM reader implements the same merge. + throw new UnsupportedOperationException("LSM file group reader does not support bootstrap base files"); + } + return createFileIterator(baseFile.getPathInfo(), baseFile.getStoragePath(), baseFile.getFileSize()); + } + + /** + * Creates a sorted-run iterator for a parquet data file or a native parquet log file. + * + * <p>Native delete logs use a specialized schema and are routed through + * {@link #createNativeDeleteLogIterator(StoragePathInfo, StoragePath, long)}. + */ + private ClosableIterator<BufferedRecord<T>> createFileIterator(StoragePathInfo pathInfo, + StoragePath path, + long fileSize) throws IOException { + StoragePath storagePath = pathInfo != null ? pathInfo.getPath() : path; + if (FSUtils.isNativeDeleteLogFile(storagePath.getName())) { + return createNativeDeleteLogIterator(pathInfo, storagePath, fileSize); + } + Pair<HoodieSchema, Map<String, String>> requiredSchemaAndRenamedColumns = + readerContext.getSchemaHandler().getRequiredSchemaForFileAndRenamedColumns(storagePath); + HoodieSchema fileRequiredSchema = requiredSchemaAndRenamedColumns.getLeft(); + ClosableIterator<T> recordIterator; + if (pathInfo != null) { + recordIterator = readerContext.getFileRecordIterator( + pathInfo, 0, pathInfo.getLength(), readerContext.getSchemaHandler().getTableSchema(), fileRequiredSchema, storage); + } else { + long length = fileSize >= 0 ? fileSize : storage.getPathInfo(storagePath).getLength(); + recordIterator = readerContext.getFileRecordIterator( + storagePath, 0, length, readerContext.getSchemaHandler().getTableSchema(), fileRequiredSchema, storage); + } + if (!areSchemasProjectionEquivalent(fileRequiredSchema, readerSchema) || !requiredSchemaAndRenamedColumns.getRight().isEmpty()) { + UnaryOperator<T> projector = readerContext.getRecordContext() + .projectRecord(fileRequiredSchema, readerSchema, requiredSchemaAndRenamedColumns.getRight()); + recordIterator = new CloseableMappingIterator<>(recordIterator, projector); + } + if (readerContext.getInstantRange().isPresent()) { + recordIterator = readerContext.applyInstantRangeFilter(recordIterator); + } + return new CloseableMappingIterator<>(recordIterator, record -> BufferedRecords.fromEngineRecord( + readerContext.getRecordContext().seal(record), + readerSchema, + readerContext.getRecordContext(), + orderingFieldNames, + readerContext.getRecordContext().isDeleteRecord(record, readerContext.getSchemaHandler().getDeleteContext().withReaderSchema(readerSchema)))); + } + + /** + * Creates delete records from an RFC-103 native delete log. + * + * <p>The delete log schema intentionally contains only the record key and ordering value fields; + * partition path and full data columns are not read for delete-only logs. + */ + private ClosableIterator<BufferedRecord<T>> createNativeDeleteLogIterator(StoragePathInfo pathInfo, + StoragePath storagePath, + long fileSize) throws IOException { + ClosableIterator<T> recordIterator; + if (pathInfo != null) { + recordIterator = readerContext.getFileRecordIterator( + pathInfo, 0, pathInfo.getLength(), HoodieSchemas.DELETE_LOG_SCHEMA, HoodieSchemas.DELETE_LOG_SCHEMA, storage); + } else { + long length = fileSize >= 0 ? fileSize : storage.getPathInfo(storagePath).getLength(); + recordIterator = readerContext.getFileRecordIterator( + storagePath, 0, length, HoodieSchemas.DELETE_LOG_SCHEMA, HoodieSchemas.DELETE_LOG_SCHEMA, storage); + } + return new CloseableMappingIterator<>(recordIterator, record -> { + return createNativeDeleteRecord(readerContext, record); + }); + } + + @VisibleForTesting + static <T> BufferedRecord<T> createNativeDeleteRecord(HoodieReaderContext<T> readerContext, T record) { + Object recordKey = readerContext.getRecordContext().getValue(record, HoodieSchemas.DELETE_LOG_SCHEMA, DELETE_LOG_RECORD_KEY_FIELD); + // Preserve the delete log's ordering value so event-time/custom merge modes can compare + // deletes against data records instead of treating every native delete as commit-time ordered. + Comparable orderingValue = + readerContext.getRecordContext().getOrderingValue(record, HoodieSchemas.DELETE_LOG_SCHEMA, DELETE_LOG_ORDERING_FIELD_NAMES); Review Comment: Addressed in ab0f6180551a. Native delete logs now decode `ordering_val` from the bytes field before creating the buffered delete record: `LsmFileGroupRecordIterator` calls `RecordContext.getBytes(...)` and then `DeleteRecord.decodeOrderingValue(...)`. The encode/decode helpers use Kryo serialization for the Java `Comparable` ordering value. I also added `JavaTypeConverter.castToBytes(...)` so the bytes conversion stays engine-specific: default handles Java `byte[]`, Avro handles `ByteBuffer`, and Hive handles `BytesWritable.copyBytes()` to avoid returning the backing buffer. Added focused coverage in `TestDeleteRecord`, `TestLsmFileGroupRecordIterator`, `TestAvroJavaTypeConverter`, `TestAvroRecordContext`, `TestDefaultJavaTypeConverter`, `TestHiveJavaTypeConverter`, and `TestHiveHoodieReaderContext`. -- 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]
