leaves12138 commented on code in PR #9442: URL: https://github.com/apache/paimon/pull/9442#discussion_r3878324597
########## 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: The generic builders only call for rows returned by the table reader. On a data-evolution table with deletion vectors, deleted rows are filtered out, so the relative values can contain gaps (for example 0, 2 after deleting row 1). This check makes fail on such a table. More importantly, the file currently derives result IDs from , so simply relaxing the check would remap rows after the first gap, and a deleted tail cannot be represented because the generic Spark/Flink builders call rather than .\n\nCould we preserve the full source row-id range (for example by padding missing IDs, including trailing gaps via , or by storing an explicit row-id mapping) and add an end-to-end test that builds the FM index after a deletion? -- 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]
