danny0405 commented on code in PR #18987: URL: https://github.com/apache/hudi/pull/18987#discussion_r3457876538
########## hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/SpillableLsmRecordIterator.java: ########## @@ -0,0 +1,175 @@ +/* + * 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.engine.RecordContext; +import org.apache.hudi.common.serialization.CustomSerializer; +import org.apache.hudi.common.table.read.BufferedRecord; +import org.apache.hudi.common.util.collection.ClosableIterator; +import org.apache.hudi.exception.HoodieIOException; + +import java.io.BufferedInputStream; +import java.io.BufferedOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; + +/** + * Sequential disk-backed iterator for sorted LSM inputs. + * + * <p>The source iterator is drained into a length-prefixed spill file and closed. The resulting + * iterator reads the records back sequentially, which matches the loser-tree access pattern. + */ +class SpillableLsmRecordIterator<T> implements ClosableIterator<BufferedRecord<T>> { + + private static final int BUFFER_SIZE = 128 * 1024; + private static final String SPILL_FILE_PREFIX = "hudi-lsm-"; + private static final String SPILL_FILE_SUFFIX = ".spill"; + + private final CustomSerializer<BufferedRecord<T>> serializer; + private final RecordContext<T> recordContext; + private final File spillFile; + private final long recordCount; + private DataInputStream inputStream; + private long recordsRead; + private BufferedRecord<T> nextRecord; + private boolean closed; + + SpillableLsmRecordIterator(ClosableIterator<BufferedRecord<T>> sourceIterator, + CustomSerializer<BufferedRecord<T>> serializer, + RecordContext<T> recordContext, + String spillBasePath) { + this.serializer = serializer; + this.recordContext = recordContext; + Throwable spillFailure = null; + try { + Path spillDirectory = Paths.get(spillBasePath); + Files.createDirectories(spillDirectory); + this.spillFile = Files.createTempFile(spillDirectory, SPILL_FILE_PREFIX, SPILL_FILE_SUFFIX).toFile(); + this.spillFile.deleteOnExit(); Review Comment: Done. Removed deleteOnExit(); close() and constructor failure cleanup continue to delete the spill file explicitly. ########## hudi-common/src/main/java/org/apache/hudi/common/model/HoodieLogFile.java: ########## Review Comment: Done. HoodieLogFile.isCDC() now delegates to FSUtils.isCDCLogFile(...), which parses native cdc logs like ...cdc.parquet and ...cdc.lance. Added tests for native CDC log parsing. ########## hudi-common/src/main/java/org/apache/hudi/common/table/read/lsm/LsmFileGroupRecordIterator.java: ########## @@ -0,0 +1,541 @@ +/* + * 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.BaseFile; +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"); + } + BaseFile file = baseFile.getBootstrapBaseFile().orElse(baseFile); + return createFileIterator(file.getPathInfo(), file.getStoragePath(), file.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, Review Comment: Done. InputSplit now filters with HoodieLogFile.isCDC(), and FSUtils.isCDCLogFile covers both legacy .cdc logs and native cdc.<format> logs. Added TestInputSplit coverage for legacy/native parquet/lance CDC names. -- 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]
