leaves12138 commented on code in PR #8613:
URL: https://github.com/apache/paimon/pull/8613#discussion_r3576760875
##########
paimon-api/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -4339,7 +4333,12 @@ public Options primaryKeyBitmapIndexOptions(String
column) {
private Options primaryKeySortedIndexOptions(
String column, String optionFamily, String algorithmPrefix) {
- Options resolved = new Options(toConfiguration().toMap());
+ Options resolved = new Options();
Review Comment:
Filtering the resolved options down to only the algorithm prefix also drops
the generic sort settings consumed by `PkSortedIndexBuilder`, including
`write-buffer-size`, `page-size`, `local-sort.max-num-file-handles`, spill
compression, and `write-buffer-spill.max-disk-size`. User-provided resource
limits will therefore silently fall back to defaults. Could we preserve the
general build options and remove only the range-sizing option, or pass build
and algorithm options separately?
##########
paimon-core/src/main/java/org/apache/paimon/index/pksorted/PkSortedIndexFile.java:
##########
@@ -40,94 +38,86 @@
import javax.annotation.Nullable;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import static org.apache.paimon.utils.Preconditions.checkArgument;
-/** Builds source-backed BTree or Bitmap payloads for one physical data file.
*/
+/** Builds source-backed BTree or Bitmap payloads for ordered physical data
files. */
public class PkSortedIndexFile extends IndexFile {
public PkSortedIndexFile(FileIO fileIO, IndexPathFactory pathFactory) {
super(fileIO, pathFactory);
}
- public List<IndexFileMeta> build(
- PrimaryKeyIndexSourceFile sourceFile,
+ public IndexFileMeta build(
+ List<PrimaryKeyIndexSourceFile> sourceFiles,
DataField indexField,
String indexType,
Options indexOptions,
Iterator<Entry> sortedEntries)
throws IOException {
+ long sourceRowCount = 0;
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
+ sourceRowCount = Math.addExact(sourceRowCount,
sourceFile.rowCount());
+ }
checkArgument(
- sourceFile.rowCount() > 0,
- "A sorted index group must reference at least one source
row.");
+ sourceRowCount > 0, "A sorted index group must reference at
least one source row.");
TrackingFileWriter fileWriter = new TrackingFileWriter();
boolean success = false;
try {
- long recordsPerRange =
-
indexOptions.get(SortedIndexOptions.SORTED_INDEX_RECORDS_PER_RANGE);
- SortedSingleColumnIndexWriter writer =
- new SortedSingleColumnIndexWriter(
- recordsPerRange,
- () -> createWriter(indexType, indexField,
indexOptions, fileWriter));
+ GlobalIndexSingleColumnWriter writer =
+ createWriter(indexType, indexField, indexOptions,
fileWriter);
long writtenRows = 0;
while (sortedEntries.hasNext()) {
Entry entry = sortedEntries.next();
checkArgument(
- entry.localRowId >= 0 && entry.localRowId <
sourceFile.rowCount(),
- "Local row id %s is outside source file %s row range
[0, %s).",
- entry.localRowId,
- sourceFile.fileName(),
- sourceFile.rowCount());
- writer.write(entry.value, entry.localRowId);
+ entry.rowId >= 0 && entry.rowId < sourceRowCount,
+ "Row id %s is outside sorted index group row range [0,
%s).",
+ entry.rowId,
+ sourceRowCount);
+ writer.write(entry.value, entry.rowId);
writtenRows++;
}
checkArgument(
- writtenRows == sourceFile.rowCount(),
- "Sorted index input row count %s does not match source
file %s row count %s.",
+ writtenRows == sourceRowCount,
+ "Sorted index input row count %s does not match source row
count %s.",
writtenRows,
- sourceFile.fileName(),
- sourceFile.rowCount());
-
- List<List<ResultEntry>> resultGroups = writer.finish();
- List<IndexFileMeta> payloads = new ArrayList<>();
- long payloadRows = 0;
- byte[] sourceMeta = new
PrimaryKeyIndexSourceMeta(sourceFile).serialize();
- for (List<ResultEntry> resultGroup : resultGroups) {
- for (ResultEntry result : resultGroup) {
- payloadRows = Math.addExact(payloadRows,
result.rowCount());
- Path payloadPath = fileWriter.path(result.fileName());
- payloads.add(
- new IndexFileMeta(
- indexType,
- result.fileName(),
- fileIO.getFileSize(payloadPath),
- result.rowCount(),
- new GlobalIndexMeta(
- 0,
- sourceFile.rowCount() - 1,
- indexField.id(),
- null,
- result.meta(),
- sourceMeta),
- pathFactory.isExternalPath() ?
payloadPath.toString() : null));
- }
- }
+ sourceRowCount);
+
+ List<ResultEntry> results = writer.finish();
+ checkArgument(
+ results.size() == 1,
Review Comment:
Requiring exactly one physical payload removes the size bound for sorted
primary-key index builds. In particular, `BitmapGlobalIndexWriter` materializes
all key bitmaps for the whole merged LSM group in memory, while groups can
continue growing across levels. This can cause OOMs or very large index files.
Could one logical group still contain multiple physical payloads using
group-global row IDs, or could we enforce an explicit row/file-size bound?
##########
paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySortedIndexScan.java:
##########
@@ -224,47 +242,459 @@ static EvaluatedPlan evaluate(
}
}
+ Map<PkSortedIndexGroup, SharedGlobalIndexReader> sharedReaders = new
IdentityHashMap<>();
List<EvaluatedFile> files = new ArrayList<>();
- for (FilePlan file : plan.files()) {
- GlobalIndexEvaluator evaluator =
- new GlobalIndexEvaluator(
- rowType,
- fieldId -> {
- PrimaryKeyIndexDefinition definition =
- definitionsByField.get(fieldId);
- Optional<PkSortedIndexGroup> group =
file.group(fieldId);
- if (definition == null || !group.isPresent()) {
- return Collections.emptyList();
- }
- return Collections.singletonList(
- readerFactory.create(
- file, definition,
group.get().payloads()));
- });
- Optional<GlobalIndexResult> result;
- try {
- result = evaluator.evaluate(predicate);
- } catch (RuntimeException e) {
- rethrowIfInterrupted(e);
- LOG.warn(
- "Failed to evaluate primary-key sorted index for data
file {}; "
- + "falling back to a raw scan for this file.",
- file.dataFile().fileName(),
- e);
- result = Optional.empty();
- } finally {
- evaluator.close();
+ try {
+ for (FilePlan file : plan.files()) {
+ GlobalIndexEvaluator evaluator =
+ new GlobalIndexEvaluator(
+ rowType,
+ fieldId -> {
+ PrimaryKeyIndexDefinition definition =
+ definitionsByField.get(fieldId);
+ Optional<PkSortedIndexGroup> group =
file.group(fieldId);
+ if (definition == null ||
!group.isPresent()) {
+ return Collections.emptyList();
+ }
+ SharedGlobalIndexReader reader =
sharedReaders.get(group.get());
+ if (reader == null) {
+ reader =
+ new SharedGlobalIndexReader(
+ () ->
+
readerFactory.create(
+ file,
+
definition,
+
group.get().payloads()));
+ sharedReaders.put(group.get(), reader);
+ }
+ return Collections.singletonList(
+ fileLocalReader(file, group.get(),
reader));
+ });
+ Optional<GlobalIndexResult> result;
+ try {
+ result = evaluator.evaluate(predicate);
+ } catch (RuntimeException e) {
+ rethrowIfInterrupted(e);
+ LOG.warn(
+ "Failed to evaluate primary-key sorted index for
data file {}; "
+ + "falling back to a raw scan for this
file.",
+ file.dataFile().fileName(),
+ e);
+ result = Optional.empty();
+ } finally {
+ evaluator.close();
+ }
+ files.add(new EvaluatedFile(file, result));
}
- files.add(new EvaluatedFile(file, result));
+ } finally {
+ IOUtils.closeAllQuietly(sharedReaders.values());
}
return new EvaluatedPlan(plan.snapshotId(), files);
}
+ private static GlobalIndexReader fileLocalReader(
+ FilePlan file, PkSortedIndexGroup group, GlobalIndexReader reader)
{
+ List<PrimaryKeyIndexSourceFile> sourceFiles = group.sourceFiles();
+ PrimaryKeyIndexSourceFile target =
+ new PrimaryKeyIndexSourceFile(
+ file.dataFile().fileName(),
file.dataFile().rowCount());
+ long totalRowCount = 0;
+ long from = -1;
+ long to = -1;
+ for (PrimaryKeyIndexSourceFile sourceFile : sourceFiles) {
+ long next = Math.addExact(totalRowCount, sourceFile.rowCount());
+ if (sourceFile.equals(target)) {
+ from = totalRowCount;
+ to = next;
+ }
+ totalRowCount = next;
+ }
+ checkArgument(
+ from >= 0,
+ "Data file %s is not covered by its sorted-index source
group.",
+ file.dataFile().fileName());
+ return new FileLocalGlobalIndexReader(reader, from, to, totalRowCount);
+ }
+
private static void rethrowIfInterrupted(RuntimeException exception) {
if (Thread.currentThread().isInterrupted()) {
throw exception;
}
}
+ /** Shares one source-group reader and its group-global query results
across source files. */
+ private static final class SharedGlobalIndexReader implements
GlobalIndexReader {
+
+ private final Supplier<GlobalIndexReader> readerFactory;
+ private final Map<QueryKey,
CompletableFuture<Optional<GlobalIndexResult>>> results;
+
+ private GlobalIndexReader reader;
+ private RuntimeException readerFailure;
+
+ private SharedGlobalIndexReader(Supplier<GlobalIndexReader>
readerFactory) {
+ this.readerFactory = readerFactory;
+ this.results = new ConcurrentHashMap<>();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNotNull(FieldRef fieldRef) {
+ return query(
+ QueryKey.of(QueryOperation.IS_NOT_NULL, fieldRef),
+ () -> reader().visitIsNotNull(fieldRef));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNull(FieldRef fieldRef) {
+ return query(
+ QueryKey.of(QueryOperation.IS_NULL, fieldRef),
+ () -> reader().visitIsNull(fieldRef));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitStartsWith(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.STARTS_WITH, fieldRef, literal),
+ () -> reader().visitStartsWith(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEndsWith(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.ENDS_WITH, fieldRef, literal),
+ () -> reader().visitEndsWith(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitContains(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.CONTAINS, fieldRef, literal),
+ () -> reader().visitContains(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.LIKE, fieldRef, literal),
+ () -> reader().visitLike(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessThan(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.LESS_THAN, fieldRef, literal),
+ () -> reader().visitLessThan(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitGreaterOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.GREATER_OR_EQUAL, fieldRef,
literal),
+ () -> reader().visitGreaterOrEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotEqual(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.NOT_EQUAL, fieldRef, literal),
+ () -> reader().visitNotEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.LESS_OR_EQUAL, fieldRef,
literal),
+ () -> reader().visitLessOrEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEqual(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.EQUAL, fieldRef, literal),
+ () -> reader().visitEqual(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterThan(
+ FieldRef fieldRef, Object literal) {
+ return query(
+ QueryKey.of(QueryOperation.GREATER_THAN, fieldRef,
literal),
+ () -> reader().visitGreaterThan(fieldRef, literal));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return query(
+ QueryKey.ofLiterals(QueryOperation.IN, fieldRef, literals),
+ () -> reader().visitIn(fieldRef, literals));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return query(
+ QueryKey.ofLiterals(QueryOperation.NOT_IN, fieldRef,
literals),
+ () -> reader().visitNotIn(fieldRef, literals));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitBetween(
+ FieldRef fieldRef, Object from, Object to) {
+ return query(
+ QueryKey.of(QueryOperation.BETWEEN, fieldRef, from, to),
+ () -> reader().visitBetween(fieldRef, from, to));
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotBetween(
+ FieldRef fieldRef, Object from, Object to) {
+ return query(
+ QueryKey.of(QueryOperation.NOT_BETWEEN, fieldRef, from,
to),
+ () -> reader().visitNotBetween(fieldRef, from, to));
+ }
+
+ private CompletableFuture<Optional<GlobalIndexResult>> query(
+ QueryKey key,
+ Supplier<CompletableFuture<Optional<GlobalIndexResult>>>
querySupplier) {
+ return results.computeIfAbsent(
+ key,
+ ignored -> {
+ try {
+ return checkNotNull(querySupplier.get());
+ } catch (RuntimeException e) {
+ CompletableFuture<Optional<GlobalIndexResult>>
failed =
+ new CompletableFuture<>();
+ failed.completeExceptionally(e);
+ return failed;
+ }
+ });
+ }
+
+ private synchronized GlobalIndexReader reader() {
+ if (reader != null) {
+ return reader;
+ }
+ if (readerFailure != null) {
+ throw readerFailure;
+ }
+ try {
+ reader = checkNotNull(readerFactory.get());
+ return reader;
+ } catch (RuntimeException e) {
+ readerFailure = e;
+ throw e;
+ }
+ }
+
+ @Override
+ public synchronized void close() throws IOException {
+ if (reader != null) {
+ GlobalIndexReader readerToClose = reader;
+ reader = null;
+ readerToClose.close();
+ }
+ }
+ }
+
+ private enum QueryOperation {
+ IS_NOT_NULL,
+ IS_NULL,
+ STARTS_WITH,
+ ENDS_WITH,
+ CONTAINS,
+ LIKE,
+ LESS_THAN,
+ GREATER_OR_EQUAL,
+ NOT_EQUAL,
+ LESS_OR_EQUAL,
+ EQUAL,
+ GREATER_THAN,
+ IN,
+ NOT_IN,
+ BETWEEN,
+ NOT_BETWEEN
+ }
+
+ private static final class QueryKey {
+
+ private final QueryOperation operation;
+ private final FieldRef fieldRef;
+ private final List<Object> literals;
+
+ private QueryKey(QueryOperation operation, FieldRef fieldRef,
List<Object> literals) {
+ this.operation = operation;
+ this.fieldRef = fieldRef;
+ this.literals = Collections.unmodifiableList(new
ArrayList<>(literals));
+ }
+
+ private static QueryKey of(
+ QueryOperation operation, FieldRef fieldRef, Object...
literals) {
+ return new QueryKey(operation, fieldRef, Arrays.asList(literals));
+ }
+
+ private static QueryKey ofLiterals(
+ QueryOperation operation, FieldRef fieldRef, List<Object>
literals) {
+ return new QueryKey(operation, fieldRef, literals);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof QueryKey)) {
+ return false;
+ }
+ QueryKey queryKey = (QueryKey) o;
+ return operation == queryKey.operation
+ && Objects.equals(fieldRef, queryKey.fieldRef)
+ && Objects.equals(literals, queryKey.literals);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(operation, fieldRef, literals);
+ }
+ }
+
+ /** Restricts merged source-group ordinals to one source file's local row
positions. */
+ private static final class FileLocalGlobalIndexReader implements
GlobalIndexReader {
+
+ private final GlobalIndexReader wrapped;
+ private final long from;
+ private final long to;
+ private final long totalRowCount;
+
+ private FileLocalGlobalIndexReader(
+ GlobalIndexReader wrapped, long from, long to, long
totalRowCount) {
+ this.wrapped = wrapped;
+ this.from = from;
+ this.to = to;
+ this.totalRowCount = totalRowCount;
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNotNull(FieldRef fieldRef) {
+ return
wrapped.visitIsNotNull(fieldRef).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNull(FieldRef fieldRef) {
+ return wrapped.visitIsNull(fieldRef).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitStartsWith(
+ FieldRef fieldRef, Object literal) {
+ return wrapped.visitStartsWith(fieldRef,
literal).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEndsWith(
+ FieldRef fieldRef, Object literal) {
+ return wrapped.visitEndsWith(fieldRef,
literal).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitContains(
+ FieldRef fieldRef, Object literal) {
+ return wrapped.visitContains(fieldRef,
literal).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
+ FieldRef fieldRef, Object literal) {
+ return wrapped.visitLike(fieldRef,
literal).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessThan(
+ FieldRef fieldRef, Object literal) {
+ return wrapped.visitLessThan(fieldRef,
literal).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitGreaterOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return wrapped.visitGreaterOrEqual(fieldRef,
literal).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotEqual(
+ FieldRef fieldRef, Object literal) {
+ return wrapped.visitNotEqual(fieldRef,
literal).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return wrapped.visitLessOrEqual(fieldRef,
literal).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEqual(
+ FieldRef fieldRef, Object literal) {
+ return wrapped.visitEqual(fieldRef,
literal).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterThan(
+ FieldRef fieldRef, Object literal) {
+ return wrapped.visitGreaterThan(fieldRef,
literal).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return wrapped.visitIn(fieldRef,
literals).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return wrapped.visitNotIn(fieldRef,
literals).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitBetween(
+ FieldRef fieldRef, Object from, Object to) {
+ return wrapped.visitBetween(fieldRef, from,
to).thenApply(this::toFileLocal);
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotBetween(
+ FieldRef fieldRef, Object from, Object to) {
+ return wrapped.visitNotBetween(fieldRef, from,
to).thenApply(this::toFileLocal);
+ }
+
+ private Optional<GlobalIndexResult>
toFileLocal(Optional<GlobalIndexResult> result) {
+ if (!result.isPresent() || (from == 0 && to == totalRowCount)) {
+ return result;
+ }
+ RoaringNavigableMap64 positions = new RoaringNavigableMap64();
+ for (long position : result.get().results()) {
Review Comment:
Sharing the underlying query fixes the repeated index reads, but every
source file still traverses the complete group-global result here. For a group
with `N` source files and a broad predicate returning `M` positions, evaluation
remains `O(N * M)` and allocates `N` file-local bitmaps. Could the cached
result be partitioned across all source ranges once per query, or use bitmap
range slicing to avoid repeatedly scanning the full result?
--
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]
