JingsongLi commented on code in PR #9442: URL: https://github.com/apache/paimon/pull/9442#discussion_r3878426469
########## paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexWriter.java: ########## @@ -0,0 +1,419 @@ +/* + * 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.paimon.globalindex.fmindex; + +import org.apache.paimon.compression.BlockCompressionFactory; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.ResultEntry; +import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; +import org.apache.paimon.utils.Preconditions; + +import javax.annotation.Nullable; + +import java.io.ByteArrayOutputStream; +import java.io.Closeable; +import java.io.DataOutputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +/** Streaming, bounded-partition writer for an exact byte-oriented FM index. */ +public class FMGlobalIndexWriter implements GlobalIndexSingleColumnWriter, Closeable { + + private static final int VERIFICATION_PAGE_ROW_COUNT = 128; + private static final int TARGET_VERIFICATION_PAGE_SIZE = 64 * 1024; + + private final GlobalIndexFileWriter fileWriter; + private final int maxPartitionTextLength; + private final int maxPartitionRowCount; + private final int sampleRate; + @Nullable private final BlockCompressionFactory compressionFactory; + private final List<ResultEntry> results = new ArrayList<>(); + + private CharBuilder text = new CharBuilder(); + private boolean[] nullRows = new boolean[128]; + private long partitionFirstRowId; + private int partitionRowCount; + private long lastRowId; + private boolean hasLastRowId; + private boolean finished; + + FMGlobalIndexWriter( + GlobalIndexFileWriter fileWriter, + int maxPartitionTextLength, + int maxPartitionRowCount, + int sampleRate, + @Nullable BlockCompressionFactory compressionFactory) { + this.fileWriter = fileWriter; + this.maxPartitionTextLength = maxPartitionTextLength; + this.maxPartitionRowCount = maxPartitionRowCount; + this.sampleRate = sampleRate; + this.compressionFactory = compressionFactory; + } + + @Override + public void write(@Nullable Object key, long relativeRowId) { + Preconditions.checkState(!finished, "Cannot write after the FM index writer is finished."); + Preconditions.checkArgument(relativeRowId >= 0, "FM index row ID must be non-negative."); + Preconditions.checkArgument( + !hasLastRowId || relativeRowId == lastRowId + 1, Review Comment: Thanks for raising this. I revalidated the build path: the generic global-index split deliberately does not propagate deletion files, so DataEvolutionSplitRead indexes the complete physical row-id range; deletion vectors are applied query-side for the pinned snapshot. Consequently the FM writer still receives consecutive relative IDs, including middle/tail deleted rows. Relaxing the writer check or padding deleted rows would weaken this invariant (and padding as null would break IS NULL semantics). Commit 6c9e2b8957 makes this contract explicit in GlobalIndexBuilderUtils and adds a Spark E2E that deletes a middle row and the tail row via DVs before building FM. It verifies the index still covers all 7 physical row IDs and that deleted matches are filtered from results. The focused Spark 3 suite passes 2/2. -- 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]
