diff --git a/docs/docs/multimodal-table/global-index.mdx b/docs/docs/multimodal-table/global-index.mdx index c97d5328c76d..d7b0041b71fe 100644 --- a/docs/docs/multimodal-table/global-index.mdx +++ b/docs/docs/multimodal-table/global-index.mdx @@ -30,6 +30,7 @@ Global Index is a powerful indexing mechanism for Data Evolution (append) tables without full-table scans. Paimon supports multiple global index types: - **[BTree Index](./global-index/btree)**: A B-tree based index for scalar column lookups. Supports equality, IN, range predicates, and can be combined across multiple columns with AND/OR logic. +- **[Bitmap Index](./global-index/bitmap)**: A bitmap based index for enum-like scalar dimensions and tag columns. Supports equality, IN, prefix match on string columns, complement predicates, and null checks with compressed row-id bitmaps. - **[Vector Index](./global-index/vector)**: An approximate nearest neighbor (ANN) index powered by Paimon's vector index library for vector similarity search. - **[Full-Text Index](./global-index/full-text)**: A full-text search index powered by Tantivy for text retrieval. Supports term matching and relevance scoring. - **[Hybrid Search](./global-index/hybrid-search)**: A multi-route search API that combines results from multiple vector routes, multiple full-text routes, or both before reading table rows. @@ -37,6 +38,7 @@ without full-table scans. Paimon supports multiple global index types: | Index Type | Best For | Notes | |---|---|---| | BTree | Scalar filters on numeric, string, date, and timestamp columns | Best when predicates are selective, such as equality, IN, range, and null checks. | +| Bitmap | Enum-like dimensions and tag columns | Best for equality, IN, string prefix match, complement predicates, and null checks over compressed row-id bitmaps. | | Vector | Top-K similarity search on embeddings | Uses ANN algorithms. Tune build-time and search-time options to balance recall, latency, and index size. | | Full-Text | Keyword search over text columns | Uses Tantivy scoring and tokenizer configuration stored with each index file. | | Hybrid Search | Combining multiple vector routes, multiple full-text routes, or vector and full-text retrieval together | Runs multiple scored routes and merges them with a ranker before reading rows. | @@ -158,6 +160,12 @@ These table options affect global index build and read behavior: Use BTree indexes for scalar column lookups and range predicates. See [BTree Index](./global-index/btree) for build and query examples. +## Bitmap Index + +Use Bitmap indexes for enum-like dimensions and tag columns queried by equality, IN, string prefix +match, complement, or null predicates. See [Bitmap Index](./global-index/bitmap) for build examples, +options, and file format details. + ## Vector Index Use Vector indexes for approximate nearest neighbor (ANN) search. See [Vector Index](./global-index/vector) diff --git a/docs/docs/multimodal-table/global-index/bitmap.mdx b/docs/docs/multimodal-table/global-index/bitmap.mdx new file mode 100644 index 000000000000..b9715687d565 --- /dev/null +++ b/docs/docs/multimodal-table/global-index/bitmap.mdx @@ -0,0 +1,215 @@ +--- +title: "Bitmap Index" +sidebar_position: 2 +--- + + + +# Bitmap Index + +A bitmap index maps each distinct scalar value to a compressed 64-bit row-id bitmap. +Use it for enum-like dimensions and tag columns where queries often use equality, +`IN`, complement, or null predicates, for example `status`, `country`, `tenant_type`, +or business tags. + +Compared with [BTree Index](./btree), bitmap index is optimized for set operations over +exact row-id bitmaps. It is usually a better fit for low-cardinality or medium-cardinality +dimensions, especially when `IN`, `NOT IN`, or `!=` predicates are common. BTree index is +still better for range predicates and high-cardinality columns where each value has only a +few rows and range pruning is important. + +Supported predicate shapes include: + +| Predicate | Example | +|---|---| +| Equality | `tag = 'vip'` | +| IN | `tag IN ('vip', 'trial')` | +| Not equal | `tag != 'blocked'` | +| NOT IN | `tag NOT IN ('blocked', 'test')` | +| Null checks | `tag IS NULL`, `tag IS NOT NULL` | +| String predicates | `tag LIKE 'vip%'`, `tag LIKE '%vip%'`, `tag startsWith 'vip'`, `tag contains 'vip'` | +| Range predicates | `tag >= 'a'`, `tag BETWEEN 'a' AND 'm'` | +| AND / OR combinations | `tag = 'vip' OR tag = 'trial'` | + +Equality, `IN`, null checks, and string prefix predicates use direct dictionary lookup. +Other predicates such as `endsWith`, `contains`, general `LIKE`, and range predicates +fall back to scanning bitmap dictionary entries only when the total size of candidate +bitmap index files is within the configured fallback scan budget. If the budget is +exceeded, Paimon falls back to other matching indexes or regular table scans. + +## Build Bitmap Index + +```sql +-- Create bitmap index on 'tag' column +CALL sys.create_global_index( + table => 'db.my_table', + index_column => 'tag', + index_type => 'bitmap' +); +``` + +You can build only selected partitions: + +```sql +CALL sys.create_global_index( + table => 'db.my_table', + index_column => 'tag', + index_type => 'bitmap', + partitions => 'dt=2026-06-18;dt=2026-06-19' +); +``` + +## Bitmap Options + +| Option | Default | Description | +|---|---|---| +| `bitmap-index.dictionary-block-size` | `16 kb` | Target size of dictionary blocks in bitmap global index files. Smaller blocks reduce dictionary read amplification for high-cardinality columns; larger blocks reduce dictionary block index size. | +| `bitmap-index.compression` | `none` | Compression algorithm for bitmap dictionary blocks and the dictionary block index. Supported values are the same block codecs as BTree index, such as `none`, `lz4`, `lzo`, and `zstd`. | +| `bitmap-index.compression-level` | `1` | Compression level used by codecs that support levels, such as `zstd`. | +| `bitmap-index.fallback-scan-max-size` | `256 mb` | Maximum total size of bitmap global index files in one reader to allow fallback dictionary scans for predicates that cannot use direct bitmap lookup. Set to `0 b` to disable fallback scans. | + +## Query with Bitmap Index + +Once a bitmap index is built, it is automatically used during scan when a filter +predicate matches the indexed column. + +```sql +SELECT * FROM my_table WHERE tag IN ('vip', 'trial'); +``` + +For complement predicates such as `tag != 'blocked'` or +`tag NOT IN ('blocked', 'test')`, bitmap index evaluates the complement against each +index file's own non-null row-id bitmap. This keeps results correct when one logical +query unions multiple bitmap index files. + +## File Format + +A bitmap global index file stores exact row-id bitmaps and a block-indexed dictionary. +The reader opens a file by reading only its fixed-length footer. Null row sets, +non-null row sets, and the dictionary block index are loaded lazily when a matching +predicate needs them. Point lookups then read the dictionary block containing the +target value and the corresponding bitmap block. The format is footer-driven: the +footer stores the version, magic number, and offsets to all metadata blocks. + +```text ++----------------------------------------------+ +| null rows bitmap block | ++----------------------------------------------+ +| non-null rows bitmap block | ++----------------------------------------------+ +| value bitmap and dictionary payload blocks | ++----------------------------------------------+ +| ... | ++----------------------------------------------+ +| dictionary block index | ++----------------------------------------------+ +| footer | ++----------------------------------------------+ +``` + +Each bitmap block is a serialized `RoaringNavigableMap64`. Bitmap blocks are not wrapped +with an additional compression layer because Roaring already stores row ids compactly. +Value bitmap blocks are written for non-null values in serialized-key order. Dictionary +blocks are emitted as groups reach the configured target size, so the physical payload +area can contain both value bitmap blocks and dictionary blocks. Readers use the stored +`offset` and `length` fields and do not require value bitmap blocks or dictionary blocks +to be physically contiguous. + +Each dictionary block stores sorted value entries. Keys are serialized with the same +key serializer as BTree global index keys, and are ordered by serialized bytes: + +```text ++----------------------------------------------+ +| entry count (var-length int) | ++----------------------------------------------+ +| key length (var-length int), key bytes | +| bitmap block offset (var-length long) | +| bitmap block length (var-length int) | ++----------------------------------------------+ +| ... | ++----------------------------------------------+ +``` + +The dictionary block body above is stored as a BTree-style compressed block. The +configured dictionary compression is attempted when writing, and the uncompressed body is +kept if compression does not save enough space. A 5-byte block trailer follows each +dictionary block body and records the actual compression type and CRC. The stored +dictionary block `length` is the block body length and does not include the trailer. + +The dictionary block index stores one entry per dictionary block and is small enough to +load when the index file is opened: + +```text ++----------------------------------------------+ +| block count (var-length int) | ++----------------------------------------------+ +| first key length (var-length int), key bytes | +| dictionary block offset (var-length long) | +| dictionary block length (var-length int) | ++----------------------------------------------+ +| ... | ++----------------------------------------------+ +``` + +The dictionary block index uses the same compressed-block encoding and 5-byte trailer as +dictionary blocks. The footer's dictionary block index `length` also excludes this +trailer. + +The footer has fixed length and points to the main metadata blocks: + +```text ++----------------------------------------------+ +| null rows block offset (long) | +| null rows block length (int) | +| non-null rows block offset (long) | +| non-null rows block length (int) | +| dictionary block index offset (long) | +| dictionary block index length (int) | +| value count (int) | +| version (int) = 1 | +| magic (int) | ++----------------------------------------------+ +``` + +The sorted dictionary block index lets equality and `IN` predicates locate candidate +values by binary search without deserializing every dictionary block or value bitmap in +the file. Null checks can read only the needed null/non-null bitmap. + +Each bitmap index file also stores manifest-level metadata with the logical minimum +non-null key, maximum non-null key, and whether the file contains null values. The scanner +uses this metadata to skip impossible files before opening bitmap index files, similar to +BTree global index file pruning. + +```text ++----------------------------------------------+ +| first key length (int), first key bytes | +| last key length (int), last key bytes | +| has nulls (byte) | +| metadata version (byte) = 1 | +| null key flags (byte) | ++----------------------------------------------+ +``` + +For predicates that cannot be resolved by point or prefix lookup, the reader may scan all +dictionary blocks and read matching bitmap blocks. This fallback is guarded by +`bitmap-index.fallback-scan-max-size`, which compares against the total size of candidate +bitmap index files handled by the reader. This keeps point lookup read amplification +bounded for high-cardinality tag or dimension columns while allowing small bitmap indexes +to answer broader predicates directly. diff --git a/docs/docs/multimodal-table/global-index/btree.mdx b/docs/docs/multimodal-table/global-index/btree.mdx index 52940a223ab5..236447c98781 100644 --- a/docs/docs/multimodal-table/global-index/btree.mdx +++ b/docs/docs/multimodal-table/global-index/btree.mdx @@ -42,6 +42,8 @@ Supported predicate shapes include: | AND / OR combinations | `name = 'a200' OR name = 'a300'` | `LIKE`, `startsWith`, `contains`, and `NOT IN` predicates may still need broader index file reads. +Use `btree-index.fallback-scan-max-size` to cap fallback range and string scans by the total +candidate index file size. For keyword-style text retrieval, use [Full-Text Index](./full-text) instead. ## Build BTree Index @@ -74,6 +76,7 @@ CALL sys.create_global_index( | `btree-index.build.max-parallelism` | `4096` | Maximum Flink or Spark parallelism for building BTree indexes. | | `btree-index.block-size` | `64 kb` | Block size used by BTree index files. | | `btree-index.cache-size` | `128 mb` | Cache size used by BTree index readers. | +| `btree-index.fallback-scan-max-size` | `256 mb` | Maximum total size of candidate BTree global index files to allow fallback index scans. Set to `0 b` to disable fallback scans. | | `btree-index.compression` | `none` | Compression algorithm used by BTree index blocks. | ## Query with BTree Index diff --git a/docs/docs/multimodal-table/global-index/full-text.mdx b/docs/docs/multimodal-table/global-index/full-text.mdx index 1e65d490a660..1eae51c52e0b 100644 --- a/docs/docs/multimodal-table/global-index/full-text.mdx +++ b/docs/docs/multimodal-table/global-index/full-text.mdx @@ -1,6 +1,6 @@ --- title: "Full-Text Index" -sidebar_position: 3 +sidebar_position: 4 --- import Tabs from '@theme/Tabs'; diff --git a/docs/docs/multimodal-table/global-index/hybrid-search.mdx b/docs/docs/multimodal-table/global-index/hybrid-search.mdx index dd8e23cdbdac..bde17d02b605 100644 --- a/docs/docs/multimodal-table/global-index/hybrid-search.mdx +++ b/docs/docs/multimodal-table/global-index/hybrid-search.mdx @@ -1,6 +1,6 @@ --- title: "Hybrid Search" -sidebar_position: 4 +sidebar_position: 5 --- import Tabs from '@theme/Tabs'; diff --git a/docs/docs/multimodal-table/global-index/vector.mdx b/docs/docs/multimodal-table/global-index/vector.mdx index 5f6cb01691e2..0ea650ffbafd 100644 --- a/docs/docs/multimodal-table/global-index/vector.mdx +++ b/docs/docs/multimodal-table/global-index/vector.mdx @@ -1,6 +1,6 @@ --- title: "Vector Index" -sidebar_position: 2 +sidebar_position: 3 --- import Tabs from '@theme/Tabs'; diff --git a/docs/sidebars.js b/docs/sidebars.js index 5fabbcd60fbe..e0558a3a2121 100644 --- a/docs/sidebars.js +++ b/docs/sidebars.js @@ -124,6 +124,7 @@ const sidebars = { }, "items": [ "multimodal-table/global-index/btree", + "multimodal-table/global-index/bitmap", "multimodal-table/global-index/vector", "multimodal-table/global-index/full-text", "multimodal-table/global-index/hybrid-search" diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/KeySerializer.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/KeySerializer.java similarity index 98% rename from paimon-common/src/main/java/org/apache/paimon/globalindex/btree/KeySerializer.java rename to paimon-common/src/main/java/org/apache/paimon/globalindex/KeySerializer.java index a8c848e2b613..a58b573d9de0 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/KeySerializer.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/KeySerializer.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.paimon.globalindex.btree; +package org.apache.paimon.globalindex; import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.Decimal; @@ -45,7 +45,7 @@ import java.util.Comparator; -/** This interface provides core methods to ser/de and compare btree index keys. */ +/** This interface provides core methods to ser/de and compare global index keys. */ @ThreadSafe public interface KeySerializer { @@ -61,7 +61,7 @@ static KeySerializer create(DataType type) { @Override public KeySerializer defaultMethod(DataType dataType) { throw new UnsupportedOperationException( - "DataType: " + dataType + " is not supported by btree index now."); + "DataType: " + dataType + " is not supported by global index now."); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java new file mode 100644 index 000000000000..5d0195b3fce3 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java @@ -0,0 +1,410 @@ +/* + * 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; + +import org.apache.paimon.fs.Path; +import org.apache.paimon.predicate.Contains; +import org.apache.paimon.predicate.EndsWith; +import org.apache.paimon.predicate.Equal; +import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.predicate.LeafBinaryFunction; +import org.apache.paimon.predicate.Like; +import org.apache.paimon.predicate.LikeOptimization; +import org.apache.paimon.predicate.StartsWith; +import org.apache.paimon.types.DataTypeFamily; +import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import java.io.Closeable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.function.Function; +import java.util.function.Supplier; + +/** Base reader for sorted global index files with manifest-level min/max pruning. */ +public abstract class SortedFileGlobalIndexReader + implements GlobalIndexReader { + + private final SortedFileMetaSelector fileSelector; + private final long fallbackScanMaxSize; + private final Map readerCache; + private final ExecutorService executor; + + protected SortedFileGlobalIndexReader( + List files, + KeySerializer keySerializer, + long fallbackScanMaxSize, + ExecutorService executor) { + this.fileSelector = new SortedFileMetaSelector(files, keySerializer); + this.fallbackScanMaxSize = fallbackScanMaxSize; + this.readerCache = new ConcurrentHashMap<>(); + this.executor = executor; + } + + @Override + public CompletableFuture> visitIsNotNull(FieldRef fieldRef) { + return visitParallel(() -> fileSelector.visitIsNotNull(fieldRef), this::visitIsNotNull); + } + + @Override + public CompletableFuture> visitIsNull(FieldRef fieldRef) { + return visitParallel(() -> fileSelector.visitIsNull(fieldRef), this::visitIsNull); + } + + @Override + public CompletableFuture> visitStartsWith( + FieldRef fieldRef, Object literal) { + if (!fieldRef.type().is(DataTypeFamily.CHARACTER_STRING) || literal == null) { + return unsupported(); + } + return visitParallel( + () -> fileSelector.visitStartsWith(fieldRef, literal), + reader -> visitStartsWith(reader, literal)); + } + + @Override + public CompletableFuture> visitEndsWith( + FieldRef fieldRef, Object literal) { + if (!canFallbackStringScan(fieldRef, literal)) { + return unsupported(); + } + return visitFallbackParallel( + () -> fileSelector.visitEndsWith(fieldRef, literal), + reader -> visitEndsWith(reader, literal)); + } + + @Override + public CompletableFuture> visitContains( + FieldRef fieldRef, Object literal) { + if (!canFallbackStringScan(fieldRef, literal)) { + return unsupported(); + } + return visitFallbackParallel( + () -> fileSelector.visitContains(fieldRef, literal), + reader -> visitContains(reader, literal)); + } + + @Override + public CompletableFuture> visitLike( + FieldRef fieldRef, Object literal) { + if (!fieldRef.type().is(DataTypeFamily.CHARACTER_STRING) || literal == null) { + return unsupported(); + } + + Optional> optimized = + LikeOptimization.tryOptimize(literal); + if (!optimized.isPresent()) { + if (!canFallbackStringScan(fieldRef, literal)) { + return unsupported(); + } + return visitFallbackParallel( + () -> fileSelector.visitLike(fieldRef, literal), + reader -> visitLike(reader, fieldRef, literal)); + } + + LeafBinaryFunction function = optimized.get().getKey(); + Object optimizedLiteral = optimized.get().getValue(); + if (function == Equal.INSTANCE) { + return visitEqual(fieldRef, optimizedLiteral); + } + if (function == StartsWith.INSTANCE) { + return visitStartsWith(fieldRef, optimizedLiteral); + } + if (function == EndsWith.INSTANCE) { + return visitEndsWith(fieldRef, optimizedLiteral); + } + if (function == Contains.INSTANCE) { + return visitContains(fieldRef, optimizedLiteral); + } + return unsupported(); + } + + @Override + public CompletableFuture> visitLessThan( + FieldRef fieldRef, Object literal) { + if (!canFallbackScan(literal)) { + return unsupported(); + } + return visitFallbackParallel( + () -> fileSelector.visitLessThan(fieldRef, literal), + reader -> visitLessThan(reader, literal)); + } + + @Override + public CompletableFuture> visitGreaterOrEqual( + FieldRef fieldRef, Object literal) { + if (!canFallbackScan(literal)) { + return unsupported(); + } + return visitFallbackParallel( + () -> fileSelector.visitGreaterOrEqual(fieldRef, literal), + reader -> visitGreaterOrEqual(reader, literal)); + } + + @Override + public CompletableFuture> visitNotEqual( + FieldRef fieldRef, Object literal) { + return visitParallel( + () -> fileSelector.visitNotEqual(fieldRef, literal), + reader -> visitNotEqual(reader, literal)); + } + + @Override + public CompletableFuture> visitLessOrEqual( + FieldRef fieldRef, Object literal) { + if (!canFallbackScan(literal)) { + return unsupported(); + } + return visitFallbackParallel( + () -> fileSelector.visitLessOrEqual(fieldRef, literal), + reader -> visitLessOrEqual(reader, literal)); + } + + @Override + public CompletableFuture> visitEqual( + FieldRef fieldRef, Object literal) { + return visitParallel( + () -> fileSelector.visitEqual(fieldRef, literal), + reader -> visitEqual(reader, literal)); + } + + @Override + public CompletableFuture> visitGreaterThan( + FieldRef fieldRef, Object literal) { + if (!canFallbackScan(literal)) { + return unsupported(); + } + return visitFallbackParallel( + () -> fileSelector.visitGreaterThan(fieldRef, literal), + reader -> visitGreaterThan(reader, literal)); + } + + @Override + public CompletableFuture> visitIn( + FieldRef fieldRef, List literals) { + return visitParallel( + () -> fileSelector.visitIn(fieldRef, literals), + reader -> visitIn(reader, literals)); + } + + @Override + public CompletableFuture> visitNotIn( + FieldRef fieldRef, List literals) { + return visitParallel( + () -> fileSelector.visitNotIn(fieldRef, literals), + reader -> visitNotIn(reader, literals)); + } + + @Override + public CompletableFuture> visitBetween( + FieldRef fieldRef, Object from, Object to) { + if (!canFallbackScan(from) || to == null) { + return unsupported(); + } + return visitFallbackParallel( + () -> fileSelector.visitBetween(fieldRef, from, to), + reader -> visitBetween(reader, from, to)); + } + + @Override + public CompletableFuture> visitNotBetween( + FieldRef fieldRef, Object from, Object to) { + if (!canFallbackScan(from) || to == null) { + return unsupported(); + } + return visitFallbackParallel( + () -> + fileSelector.visitOr( + Arrays.asList( + fileSelector.visitLessThan(fieldRef, from), + fileSelector.visitGreaterThan(fieldRef, to))), + reader -> visitNotBetween(reader, from, to)); + } + + protected abstract Optional visitIsNotNull(R reader); + + protected abstract Optional visitIsNull(R reader); + + protected abstract Optional visitStartsWith(R reader, Object literal); + + protected abstract Optional visitEndsWith(R reader, Object literal); + + protected abstract Optional visitContains(R reader, Object literal); + + protected abstract Optional visitLessThan(R reader, Object literal); + + protected abstract Optional visitGreaterOrEqual(R reader, Object literal); + + protected abstract Optional visitNotEqual(R reader, Object literal); + + protected abstract Optional visitLessOrEqual(R reader, Object literal); + + protected abstract Optional visitEqual(R reader, Object literal); + + protected abstract Optional visitGreaterThan(R reader, Object literal); + + protected abstract Optional visitIn(R reader, List literals); + + protected abstract Optional visitNotIn(R reader, List literals); + + protected abstract Optional visitBetween(R reader, Object from, Object to); + + protected Optional visitLike(R reader, FieldRef fieldRef, Object literal) { + return createResult(like(reader, key -> Like.INSTANCE.test(fieldRef.type(), key, literal))); + } + + protected Optional visitNotBetween(R reader, Object from, Object to) { + RoaringNavigableMap64 result = lessThan(reader, from); + result.or(greaterThan(reader, to)); + return Optional.of(GlobalIndexResult.create(result)); + } + + protected RoaringNavigableMap64 like(R reader, Function keyPredicate) { + throw new UnsupportedOperationException(); + } + + protected abstract RoaringNavigableMap64 lessThan(R reader, Object literal); + + protected abstract RoaringNavigableMap64 greaterThan(R reader, Object literal); + + protected Optional createResult(RoaringNavigableMap64 bitmap) { + return Optional.of(GlobalIndexResult.create(bitmap)); + } + + protected abstract R openReader(GlobalIndexIOMeta meta); + + @Override + public void close() throws IOException { + IOException exception = null; + for (R reader : readerCache.values()) { + try { + reader.close(); + } catch (IOException e) { + if (exception == null) { + exception = e; + } else { + exception.addSuppressed(e); + } + } + } + if (exception != null) { + throw exception; + } + } + + private boolean canFallbackStringScan(FieldRef fieldRef, Object literal) { + return fallbackScanMaxSize > 0 + && fieldRef.type().is(DataTypeFamily.CHARACTER_STRING) + && literal != null; + } + + private boolean canFallbackScan(Object literal) { + return fallbackScanMaxSize > 0 && literal != null; + } + + private static boolean fallbackScanEnabled(List files, long maxSize) { + if (maxSize <= 0) { + return false; + } + long totalSize = 0; + for (GlobalIndexIOMeta file : files) { + if (Long.MAX_VALUE - totalSize < file.fileSize()) { + return false; + } + totalSize += file.fileSize(); + if (totalSize > maxSize) { + return false; + } + } + return true; + } + + private CompletableFuture> unsupported() { + return CompletableFuture.completedFuture(Optional.empty()); + } + + private CompletableFuture> visitParallel( + Supplier>> selector, + Function> visitor) { + return visitSelectedFiles(selector.get(), visitor); + } + + private CompletableFuture> visitFallbackParallel( + Supplier>> selector, + Function> visitor) { + Optional> selectedOpt = selector.get(); + if (!selectedOpt.isPresent()) { + return CompletableFuture.completedFuture(Optional.empty()); + } + List selected = selectedOpt.get(); + if (selected.isEmpty()) { + return CompletableFuture.completedFuture(Optional.of(GlobalIndexResult.createEmpty())); + } + if (!fallbackScanEnabled(selected, fallbackScanMaxSize)) { + return unsupported(); + } + return visitSelectedFiles(selectedOpt, visitor); + } + + private CompletableFuture> visitSelectedFiles( + Optional> selectedOpt, + Function> visitor) { + if (!selectedOpt.isPresent()) { + return unsupported(); + } + List selected = selectedOpt.get(); + if (selected.isEmpty()) { + return CompletableFuture.completedFuture(Optional.of(GlobalIndexResult.createEmpty())); + } + + List>> futures = + new ArrayList<>(selected.size()); + for (GlobalIndexIOMeta meta : selected) { + futures.add( + CompletableFuture.supplyAsync( + () -> visitor.apply(getOrCreateReader(meta)), executor)); + } + return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) + .thenApply(v -> unionResults(futures)); + } + + private R getOrCreateReader(GlobalIndexIOMeta meta) { + return readerCache.computeIfAbsent(meta.filePath(), ignored -> openReader(meta)); + } + + private Optional unionResults( + List>> futures) { + Optional result = Optional.empty(); + for (CompletableFuture> future : futures) { + Optional current = future.join(); + if (!current.isPresent()) { + continue; + } + result = result.isPresent() ? Optional.of(result.get().or(current.get())) : current; + } + return result; + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeFileMetaSelector.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileMetaSelector.java similarity index 52% rename from paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeFileMetaSelector.java rename to paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileMetaSelector.java index a85b501587a9..97d0ada63381 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeFileMetaSelector.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileMetaSelector.java @@ -16,9 +16,8 @@ * limitations under the License. */ -package org.apache.paimon.globalindex.btree; +package org.apache.paimon.globalindex; -import org.apache.paimon.globalindex.GlobalIndexIOMeta; import org.apache.paimon.memory.MemorySlice; import org.apache.paimon.predicate.FieldRef; import org.apache.paimon.predicate.FunctionVisitor; @@ -26,6 +25,7 @@ import org.apache.paimon.utils.Pair; import java.util.ArrayList; +import java.util.Collections; import java.util.Comparator; import java.util.HashSet; import java.util.List; @@ -34,23 +34,29 @@ import java.util.stream.Collectors; /** - * An {@link FunctionVisitor} to select candidate btree index files. All files are expected to - * belong to the same field. The current {@code RowRangeGlobalIndexScanner} can guarantee that. - * Please do not break this premise if you want to implement your own index scanner. + * Selects candidate global index files by per-index-file min/max metadata. + * + *

All files are expected to belong to the same field. The current {@code + * RowRangeGlobalIndexScanner} can guarantee that. Please do not break this premise if you want to + * implement your own index scanner. */ -public class BTreeFileMetaSelector implements FunctionVisitor>> { +public class SortedFileMetaSelector implements FunctionVisitor>> { - private final List> files; - private final Comparator comparator; + private final List> files; private final KeySerializer keySerializer; + private final Comparator comparator; - public BTreeFileMetaSelector(List files, KeySerializer keySerializer) { + public SortedFileMetaSelector(List files, KeySerializer keySerializer) { this.files = files.stream() - .map(meta -> Pair.of(meta, BTreeIndexMeta.deserialize(meta.metadata()))) + .map( + meta -> + Pair.of( + meta, + SortedIndexFileMeta.deserialize(meta.metadata()))) .collect(Collectors.toList()); - this.comparator = keySerializer.createComparator(); this.keySerializer = keySerializer; + this.comparator = keySerializer.createComparator(); } @Override @@ -60,96 +66,95 @@ public Optional> visitIsNotNull(FieldRef fieldRef) { @Override public Optional> visitIsNull(FieldRef fieldRef) { - return Optional.of(filter(BTreeIndexMeta::hasNulls)); + return Optional.of(filter(SortedIndexFileMeta::hasNulls)); } @Override public Optional> visitStartsWith(FieldRef fieldRef, Object literal) { - return Optional.of(filter(meta -> true)); + if (literal == null) { + return Optional.of(Collections.emptyList()); + } + + byte[] prefix = serialize(literal); + if (prefix.length == 0) { + return Optional.of(filter(meta -> !meta.onlyNulls())); + } + + Object prefixKey = deserialize(prefix); + byte[] upperBoundBytes = prefixUpperBound(prefix); + Object upperBound = upperBoundBytes == null ? null : deserialize(upperBoundBytes); + return Optional.of( + filter( + meta -> + !meta.onlyNulls() + && compareLastKey(meta, prefixKey) >= 0 + && (upperBound == null + || compareFirstKey(meta, upperBound) < 0))); } @Override public Optional> visitEndsWith(FieldRef fieldRef, Object literal) { - return Optional.of(filter(meta -> true)); + return Optional.of(filter(meta -> literal != null && !meta.onlyNulls())); } @Override public Optional> visitContains(FieldRef fieldRef, Object literal) { - return Optional.of(filter(meta -> true)); + return Optional.of(filter(meta -> literal != null && !meta.onlyNulls())); } @Override public Optional> visitLike(FieldRef fieldRef, Object literal) { - return Optional.of(filter(meta -> true)); + return Optional.of(filter(meta -> literal != null && !meta.onlyNulls())); } @Override public Optional> visitLessThan(FieldRef fieldRef, Object literal) { - // `<` means file.minKey < literal - return Optional.of( - filter( - meta -> - !meta.onlyNulls() - && comparator.compare( - deserialize(meta.getFirstKey()), literal) - < 0)); + if (literal == null) { + return Optional.of(Collections.emptyList()); + } + return Optional.of(filter(meta -> !meta.onlyNulls() && compareFirstKey(meta, literal) < 0)); } @Override public Optional> visitGreaterOrEqual( FieldRef fieldRef, Object literal) { - // `>=` means file.maxKey >= literal - return Optional.of( - filter( - meta -> - !meta.onlyNulls() - && comparator.compare( - deserialize(meta.getLastKey()), literal) - >= 0)); + if (literal == null) { + return Optional.of(Collections.emptyList()); + } + return Optional.of(filter(meta -> !meta.onlyNulls() && compareLastKey(meta, literal) >= 0)); } @Override public Optional> visitNotEqual(FieldRef fieldRef, Object literal) { - return Optional.of(filter(meta -> true)); + if (literal == null) { + return Optional.of(Collections.emptyList()); + } + return Optional.of(filter(meta -> !meta.onlyNulls())); } @Override public Optional> visitLessOrEqual(FieldRef fieldRef, Object literal) { - // `<=` means file.minKey <= literal + if (literal == null) { + return Optional.of(Collections.emptyList()); + } return Optional.of( - filter( - meta -> - !meta.onlyNulls() - && comparator.compare( - deserialize(meta.getFirstKey()), literal) - <= 0)); + filter(meta -> !meta.onlyNulls() && compareFirstKey(meta, literal) <= 0)); } @Override public Optional> visitEqual(FieldRef fieldRef, Object literal) { - return Optional.of( - filter( - meta -> { - if (meta.onlyNulls()) { - return false; - } - Object minKey = deserialize(meta.getFirstKey()); - Object maxKey = deserialize(meta.getLastKey()); - return comparator.compare(literal, minKey) >= 0 - && comparator.compare(literal, maxKey) <= 0; - })); + if (literal == null) { + return Optional.of(Collections.emptyList()); + } + return Optional.of(filter(meta -> !meta.onlyNulls() && overlaps(meta, literal, literal))); } @Override public Optional> visitGreaterThan(FieldRef fieldRef, Object literal) { - // `>` means file.maxKey > literal - return Optional.of( - filter( - meta -> - !meta.onlyNulls() - && comparator.compare( - deserialize(meta.getLastKey()), literal) - > 0)); + if (literal == null) { + return Optional.of(Collections.emptyList()); + } + return Optional.of(filter(meta -> !meta.onlyNulls() && compareLastKey(meta, literal) > 0)); } @Override @@ -160,11 +165,8 @@ public Optional> visitIn(FieldRef fieldRef, List if (meta.onlyNulls()) { return false; } - Object minKey = deserialize(meta.getFirstKey()); - Object maxKey = deserialize(meta.getLastKey()); for (Object literal : literals) { - if (comparator.compare(literal, minKey) >= 0 - && comparator.compare(literal, maxKey) <= 0) { + if (literal != null && overlaps(meta, literal, literal)) { return true; } } @@ -174,24 +176,21 @@ public Optional> visitIn(FieldRef fieldRef, List @Override public Optional> visitNotIn(FieldRef fieldRef, List literals) { - // we can't filter any file meta by NOT IN condition - return Optional.of(filter(meta -> true)); + for (Object literal : literals) { + if (literal == null) { + return Optional.of(Collections.emptyList()); + } + } + return Optional.of(filter(meta -> !meta.onlyNulls())); } @Override public Optional> visitBetween( FieldRef fieldRef, Object from, Object to) { - return Optional.of( - filter( - meta -> { - if (meta.onlyNulls()) { - return false; - } - Object minKey = deserialize(meta.getFirstKey()); - Object maxKey = deserialize(meta.getLastKey()); - return comparator.compare(from, maxKey) <= 0 - && comparator.compare(to, minKey) >= 0; - })); + if (from == null || to == null || compareKeys(from, to) > 0) { + return Optional.of(Collections.emptyList()); + } + return Optional.of(filter(meta -> !meta.onlyNulls() && overlaps(meta, from, to))); } @Override @@ -222,7 +221,7 @@ public Optional> visitOr( if (!child.isPresent()) { return Optional.empty(); } - child.ifPresent(result::addAll); + result.addAll(child.get()); } return Optional.of(new ArrayList<>(result)); } @@ -232,14 +231,48 @@ public Optional> visitNonFieldLeaf(LeafPredicate predica return Optional.empty(); } - private Object deserialize(byte[] valueBytes) { + protected boolean overlaps(SortedIndexFileMeta meta, Object from, Object to) { + return comparator.compare(from, deserialize(meta.lastKey())) <= 0 + && comparator.compare(to, deserialize(meta.firstKey())) >= 0; + } + + protected int compareFirstKey(SortedIndexFileMeta meta, Object literal) { + return comparator.compare(deserialize(meta.firstKey()), literal); + } + + protected int compareLastKey(SortedIndexFileMeta meta, Object literal) { + return comparator.compare(deserialize(meta.lastKey()), literal); + } + + protected int compareKeys(Object left, Object right) { + return comparator.compare(left, right); + } + + protected byte[] serialize(Object key) { + return keySerializer.serialize(key); + } + + protected Object deserialize(byte[] valueBytes) { return keySerializer.deserialize(MemorySlice.wrap(valueBytes)); } - private List filter(Predicate predicate) { + protected List filter(Predicate predicate) { return files.stream() .filter(pair -> predicate.test(pair.getRight())) .map(Pair::getLeft) .collect(Collectors.toList()); } + + protected static byte[] prefixUpperBound(byte[] prefix) { + for (int i = prefix.length - 1; i >= 0; i--) { + int unsignedByte = prefix[i] & 0xFF; + if (unsignedByte != 0xFF) { + byte[] upperBound = new byte[i + 1]; + System.arraycopy(prefix, 0, upperBound, 0, i + 1); + upperBound[i] = (byte) (unsignedByte + 1); + return upperBound; + } + } + return null; + } } diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexMeta.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedIndexFileMeta.java similarity index 87% rename from paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexMeta.java rename to paimon-common/src/main/java/org/apache/paimon/globalindex/SortedIndexFileMeta.java index 96e0d85a7473..cd40dd7a4f47 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexMeta.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedIndexFileMeta.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.paimon.globalindex.btree; +package org.apache.paimon.globalindex; import org.apache.paimon.memory.MemorySlice; import org.apache.paimon.memory.MemorySliceInput; @@ -24,11 +24,8 @@ import javax.annotation.Nullable; -/** - * Index Meta of each BTree index file. The first key and last key of this meta could be null if the - * entire btree index file only contains nulls. - */ -public class BTreeIndexMeta { +/** Manifest-level min/max metadata for one global index file. */ +public class SortedIndexFileMeta { private static final byte FORMAT_VERSION_WITH_NULL_FLAGS = 1; private static final byte FIRST_KEY_IS_NULL = 1; @@ -38,17 +35,28 @@ public class BTreeIndexMeta { @Nullable private final byte[] lastKey; private final boolean hasNulls; - public BTreeIndexMeta(@Nullable byte[] firstKey, @Nullable byte[] lastKey, boolean hasNulls) { + public SortedIndexFileMeta( + @Nullable byte[] firstKey, @Nullable byte[] lastKey, boolean hasNulls) { this.firstKey = firstKey; this.lastKey = lastKey; this.hasNulls = hasNulls; } + @Nullable + public byte[] firstKey() { + return firstKey; + } + @Nullable public byte[] getFirstKey() { return firstKey; } + @Nullable + public byte[] lastKey() { + return lastKey; + } + @Nullable public byte[] getLastKey() { return lastKey; @@ -59,7 +67,7 @@ public boolean hasNulls() { } public boolean onlyNulls() { - return firstKey == null && lastKey == null; + return firstKey() == null && lastKey() == null; } private int memorySize() { @@ -91,7 +99,7 @@ public byte[] serialize() { return sliceOutput.toSlice().getHeapMemory(); } - public static BTreeIndexMeta deserialize(byte[] data) { + public static SortedIndexFileMeta deserialize(byte[] data) { MemorySliceInput sliceInput = MemorySlice.wrap(data).toInput(); int firstKeyLength = sliceInput.readInt(); byte[] firstKey = readKey(sliceInput, firstKeyLength); @@ -114,7 +122,7 @@ public static BTreeIndexMeta deserialize(byte[] data) { firstKey = null; lastKey = null; } - return new BTreeIndexMeta(firstKey, lastKey, hasNulls); + return new SortedIndexFileMeta(firstKey, lastKey, hasNulls); } private static byte[] readKey(MemorySliceInput sliceInput, int keyLength) { diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexFormat.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexFormat.java new file mode 100644 index 000000000000..2e355b8cde5d --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexFormat.java @@ -0,0 +1,561 @@ +/* + * 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.bitmap; + +import org.apache.paimon.compression.BlockCompressionFactory; +import org.apache.paimon.compression.BlockCompressionType; +import org.apache.paimon.compression.BlockCompressor; +import org.apache.paimon.compression.BlockDecompressor; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.globalindex.KeySerializer; +import org.apache.paimon.memory.MemorySegment; +import org.apache.paimon.memory.MemorySlice; +import org.apache.paimon.memory.MemorySliceInput; +import org.apache.paimon.sst.BlockTrailer; +import org.apache.paimon.utils.Preconditions; +import org.apache.paimon.utils.RoaringNavigableMap64; +import org.apache.paimon.utils.VarLengthIntUtils; + +import javax.annotation.Nullable; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import static org.apache.paimon.sst.SstFileUtils.crc32c; + +/** Shared file format helpers for bitmap global index. */ +class BitmapGlobalIndexFormat { + + private static final int MAGIC = 0x42474958; + private static final int VERSION = 1; + private static final int FOOTER_LENGTH = 48; + + private BitmapGlobalIndexFormat() {} + + static void write( + PositionOutputStream outputStream, + RoaringNavigableMap64 nullRows, + RoaringNavigableMap64 nonNullRows, + Map bitmaps, + int dictionaryBlockSize, + @Nullable BlockCompressionFactory compressionFactory) + throws IOException { + Preconditions.checkArgument( + dictionaryBlockSize > 0, "Bitmap dictionary block size must be greater than 0."); + + DataOutputStream out = new DataOutputStream(outputStream); + BlockInfo nullRowsBlock = writeBitmapBlock(outputStream, out, nullRows); + BlockInfo nonNullRowsBlock = writeBitmapBlock(outputStream, out, nonNullRows); + DictionaryBlocks dictionaryBlocks = + writeDictionaryAndBitmapBlocks( + outputStream, out, bitmaps, dictionaryBlockSize, compressionFactory); + BlockInfo indexBlock = + writeIndexBlock(outputStream, out, dictionaryBlocks.blocks, compressionFactory); + + out.writeLong(nullRowsBlock.offset); + out.writeInt(nullRowsBlock.length); + out.writeLong(nonNullRowsBlock.offset); + out.writeInt(nonNullRowsBlock.length); + out.writeLong(indexBlock.offset); + out.writeInt(indexBlock.length); + out.writeInt(dictionaryBlocks.valueCount); + out.writeInt(VERSION); + out.writeInt(MAGIC); + out.flush(); + } + + private static DictionaryBlocks writeDictionaryAndBitmapBlocks( + PositionOutputStream outputStream, + DataOutputStream out, + Map bitmaps, + int dictionaryBlockSize, + @Nullable BlockCompressionFactory compressionFactory) + throws IOException { + List> entries = + new ArrayList<>(bitmaps.entrySet()); + Collections.sort(entries, (o1, o2) -> o1.getKey().compareTo(o2.getKey())); + + List dictionaryBlockMetas = new ArrayList<>(); + DictionaryBlockBuilder current = new DictionaryBlockBuilder(); + int valueCount = 0; + for (Map.Entry entry : entries) { + BlockInfo bitmapBlock = writeBitmapBlock(outputStream, out, entry.getValue()); + DictionaryEntry dictionaryEntry = new DictionaryEntry(entry.getKey(), bitmapBlock); + if (current.hasEntries() + && current.estimatedSizeAfter(dictionaryEntry) > dictionaryBlockSize) { + dictionaryBlockMetas.add( + writeDictionaryBlock(outputStream, out, current, compressionFactory)); + current = new DictionaryBlockBuilder(); + } + current.add(dictionaryEntry); + valueCount++; + } + if (current.hasEntries()) { + dictionaryBlockMetas.add( + writeDictionaryBlock(outputStream, out, current, compressionFactory)); + } + return new DictionaryBlocks(dictionaryBlockMetas, valueCount); + } + + private static BlockInfo writeBitmapBlock( + PositionOutputStream outputStream, DataOutputStream out, RoaringNavigableMap64 bitmap) + throws IOException { + byte[] bytes = bitmap.serialize(); + long offset = outputStream.getPos(); + out.write(bytes); + return new BlockInfo(offset, bytes.length); + } + + private static DictionaryBlockMeta writeDictionaryBlock( + PositionOutputStream outputStream, + DataOutputStream out, + DictionaryBlockBuilder block, + @Nullable BlockCompressionFactory compressionFactory) + throws IOException { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(block.estimatedSize()); + DataOutputStream blockOut = new DataOutputStream(bytes); + writeVarLenInt(blockOut, block.entries.size()); + for (DictionaryEntry entry : block.entries) { + byte[] keyBytes = entry.key.bytes(); + writeVarLenInt(blockOut, keyBytes.length); + blockOut.write(keyBytes); + writeVarLenLong(blockOut, entry.bitmapBlock.offset); + writeVarLenInt(blockOut, entry.bitmapBlock.length); + } + BlockInfo blockInfo = + writeCompressibleBlock(outputStream, out, bytes.toByteArray(), compressionFactory); + return new DictionaryBlockMeta(block.firstKey(), blockInfo.offset, blockInfo.length); + } + + private static BlockInfo writeIndexBlock( + PositionOutputStream outputStream, + DataOutputStream out, + List blocks, + @Nullable BlockCompressionFactory compressionFactory) + throws IOException { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(estimatedIndexBlockSize(blocks)); + DataOutputStream blockOut = new DataOutputStream(bytes); + writeVarLenInt(blockOut, blocks.size()); + for (DictionaryBlockMeta block : blocks) { + byte[] keyBytes = block.firstKey.bytes(); + writeVarLenInt(blockOut, keyBytes.length); + blockOut.write(keyBytes); + writeVarLenLong(blockOut, block.offset); + writeVarLenInt(blockOut, block.length); + } + return writeCompressibleBlock(outputStream, out, bytes.toByteArray(), compressionFactory); + } + + static Footer readFooter(SeekableReader reader, long fileSize) throws IOException { + Preconditions.checkState( + fileSize >= FOOTER_LENGTH, "Invalid bitmap global index file size."); + byte[] bytes = reader.read(fileSize - FOOTER_LENGTH, FOOTER_LENGTH); + DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes)); + BlockInfo nullRowsBlock = new BlockInfo(input.readLong(), input.readInt()); + BlockInfo nonNullRowsBlock = new BlockInfo(input.readLong(), input.readInt()); + BlockInfo indexBlock = new BlockInfo(input.readLong(), input.readInt()); + int valueCount = input.readInt(); + int version = input.readInt(); + int magic = input.readInt(); + Preconditions.checkState( + magic == MAGIC, "File is not a bitmap global index file (bad footer magic)."); + Preconditions.checkState( + version == VERSION, "Unsupported bitmap global index file version: %s", version); + Preconditions.checkState(valueCount >= 0, "Invalid bitmap value count."); + return new Footer(nullRowsBlock, nonNullRowsBlock, indexBlock); + } + + private static List readIndexBlock( + SeekableReader reader, BlockInfo indexBlock) throws IOException { + DataInputStream input = + new DataInputStream( + new ByteArrayInputStream(readCompressibleBlock(reader, indexBlock))); + int blockCount = readVarLenInt(input); + Preconditions.checkState(blockCount >= 0, "Invalid bitmap dictionary block count."); + List blocks = new ArrayList<>(blockCount); + for (int i = 0; i < blockCount; i++) { + int keyLength = readVarLenInt(input); + Preconditions.checkState(keyLength >= 0, "Invalid bitmap key length."); + byte[] keyBytes = new byte[keyLength]; + input.readFully(keyBytes); + long offset = readVarLenLong(input); + int length = readVarLenInt(input); + blocks.add(new DictionaryBlockMeta(new SerializedKey(keyBytes), offset, length)); + } + Collections.sort(blocks, (o1, o2) -> o1.firstKey.compareTo(o2.firstKey)); + return blocks; + } + + static DictionaryBlock readDictionaryBlock(SeekableReader reader, DictionaryBlockMeta block) + throws IOException { + DataInputStream input = + new DataInputStream(new ByteArrayInputStream(readCompressibleBlock(reader, block))); + int entryCount = readVarLenInt(input); + Preconditions.checkState(entryCount >= 0, "Invalid bitmap dictionary entry count."); + List entries = new ArrayList<>(entryCount); + for (int i = 0; i < entryCount; i++) { + int keyLength = readVarLenInt(input); + Preconditions.checkState(keyLength >= 0, "Invalid bitmap key length."); + byte[] keyBytes = new byte[keyLength]; + input.readFully(keyBytes); + long bitmapOffset = readVarLenLong(input); + int bitmapLength = readVarLenInt(input); + entries.add( + new DictionaryEntry( + new SerializedKey(keyBytes), + new BlockInfo(bitmapOffset, bitmapLength))); + } + return new DictionaryBlock(entries); + } + + static RoaringNavigableMap64 readBitmap(SeekableReader reader, BlockInfo block) + throws IOException { + RoaringNavigableMap64 bitmap = new RoaringNavigableMap64(); + bitmap.deserialize(reader.read(block)); + return bitmap; + } + + static RoaringNavigableMap64 readBitmapUnchecked(SeekableReader reader, BlockInfo block) { + try { + return readBitmap(reader, block); + } catch (IOException e) { + throw new RuntimeException("Failed to read bitmap global index block.", e); + } + } + + static List readIndexBlockUnchecked( + SeekableReader reader, BlockInfo block) { + try { + return readIndexBlock(reader, block); + } catch (IOException e) { + throw new RuntimeException("Failed to read bitmap dictionary block index.", e); + } + } + + private static BlockInfo writeCompressibleBlock( + PositionOutputStream outputStream, + DataOutputStream out, + byte[] uncompressed, + @Nullable BlockCompressionFactory compressionFactory) + throws IOException { + BlockEncoding blockEncoding = encodeBlock(uncompressed, compressionFactory); + byte[] blockBytes = Arrays.copyOf(blockEncoding.bytes, blockEncoding.length); + long offset = outputStream.getPos(); + out.write(blockBytes); + MemorySlice trailer = + BlockTrailer.writeBlockTrailer( + new BlockTrailer( + blockEncoding.compressionType, + crc32c( + MemorySlice.wrap(blockBytes), + blockEncoding.compressionType))); + out.write(trailer.getHeapMemory(), trailer.offset(), trailer.length()); + return new BlockInfo(offset, blockBytes.length); + } + + private static BlockEncoding encodeBlock( + byte[] uncompressed, @Nullable BlockCompressionFactory compressionFactory) { + BlockCompressionType compressionType = BlockCompressionType.NONE; + byte[] bytes = uncompressed; + int length = uncompressed.length; + if (compressionFactory != null) { + BlockCompressor compressor = compressionFactory.getCompressor(); + int maxCompressedSize = compressor.getMaxCompressedSize(uncompressed.length); + byte[] compressed = new byte[maxCompressedSize + VarLengthIntUtils.MAX_VAR_INT_SIZE]; + int offset = VarLengthIntUtils.encodeInt(compressed, 0, uncompressed.length); + int compressedSize = + offset + + compressor.compress( + uncompressed, 0, uncompressed.length, compressed, offset); + if (compressedSize < uncompressed.length - (uncompressed.length / 8)) { + bytes = compressed; + length = compressedSize; + compressionType = compressionFactory.getCompressionType(); + } + } + return new BlockEncoding(bytes, length, compressionType); + } + + private static byte[] readCompressibleBlock(SeekableReader reader, BlockInfo block) + throws IOException { + Preconditions.checkState( + block.length <= Integer.MAX_VALUE - BlockTrailer.ENCODED_LENGTH, + "Bitmap block is too large."); + byte[] blockAndTrailer = + reader.read(block.offset, block.length + BlockTrailer.ENCODED_LENGTH); + byte[] blockBytes = Arrays.copyOf(blockAndTrailer, block.length); + byte[] trailerBytes = + Arrays.copyOfRange( + blockAndTrailer, block.length, block.length + BlockTrailer.ENCODED_LENGTH); + BlockTrailer blockTrailer = + BlockTrailer.readBlockTrailer(MemorySlice.wrap(trailerBytes).toInput()); + + MemorySegment blockSegment = MemorySegment.wrap(blockBytes); + int crc32cCode = crc32c(blockSegment, blockTrailer.getCompressionType()); + Preconditions.checkArgument( + blockTrailer.getCrc32c() == crc32cCode, + "Expected CRC32C(%s) but found CRC32C(%s)", + blockTrailer.getCrc32c(), + crc32cCode); + + BlockCompressionFactory compressionFactory = + BlockCompressionFactory.create(blockTrailer.getCompressionType()); + if (compressionFactory == null) { + return blockBytes; + } + + MemorySliceInput compressedInput = MemorySlice.wrap(blockSegment).toInput(); + byte[] uncompressed = new byte[compressedInput.readVarLenInt()]; + BlockDecompressor decompressor = compressionFactory.getDecompressor(); + int uncompressedLength = + decompressor.decompress( + blockSegment.getHeapMemory(), + compressedInput.position(), + compressedInput.available(), + uncompressed, + 0); + Preconditions.checkArgument(uncompressedLength == uncompressed.length); + return uncompressed; + } + + private static void writeVarLenInt(DataOutputStream out, int value) throws IOException { + VarLengthIntUtils.encodeInt((java.io.DataOutput) out, value); + } + + private static int readVarLenInt(DataInputStream input) throws IOException { + return VarLengthIntUtils.decodeInt((java.io.DataInput) input); + } + + private static void writeVarLenLong(DataOutputStream out, long value) throws IOException { + VarLengthIntUtils.encodeLong(out, value); + } + + private static long readVarLenLong(DataInputStream input) throws IOException { + return VarLengthIntUtils.decodeLong(input); + } + + private static int estimatedVarLenIntSize(int value) { + Preconditions.checkArgument(value >= 0, "Invalid negative var length int: %s", value); + int size = 1; + while ((value & ~0x7F) != 0) { + value >>>= 7; + size++; + } + return size; + } + + private static int estimatedVarLenLongSize(long value) { + Preconditions.checkArgument(value >= 0, "Invalid negative var length long: %s", value); + int size = 1; + while ((value & ~0x7FL) != 0) { + value >>>= 7; + size++; + } + return size; + } + + private static int estimatedIndexBlockSize(List blocks) { + int size = estimatedVarLenIntSize(blocks.size()); + for (DictionaryBlockMeta block : blocks) { + size += + estimatedVarLenIntSize(block.firstKey.bytes().length) + + block.firstKey.bytes().length + + estimatedVarLenLongSize(block.offset) + + estimatedVarLenIntSize(block.length); + } + return size; + } + + static class SerializedKey implements Comparable { + + private final byte[] bytes; + + SerializedKey(byte[] bytes) { + this.bytes = bytes; + } + + byte[] bytes() { + return bytes; + } + + static SerializedKey fromObject(KeySerializer serializer, Object key) { + return new SerializedKey(serializer.serialize(key)); + } + + @Override + public int compareTo(SerializedKey other) { + int length = Math.min(bytes.length, other.bytes.length); + for (int i = 0; i < length; i++) { + int diff = (bytes[i] & 0xFF) - (other.bytes[i] & 0xFF); + if (diff != 0) { + return diff; + } + } + return bytes.length - other.bytes.length; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof SerializedKey)) { + return false; + } + SerializedKey that = (SerializedKey) o; + return Arrays.equals(bytes, that.bytes); + } + + @Override + public int hashCode() { + return Arrays.hashCode(bytes); + } + } + + interface SeekableReader { + + byte[] read(long offset, int length) throws IOException; + + default byte[] read(BlockInfo block) throws IOException { + return read(block.offset, block.length); + } + } + + static class BlockInfo { + + final long offset; + final int length; + + BlockInfo(long offset, int length) { + Preconditions.checkState(offset >= 0, "Invalid bitmap block offset."); + Preconditions.checkState(length >= 0, "Invalid bitmap block length."); + this.offset = offset; + this.length = length; + } + } + + static class DictionaryBlockMeta extends BlockInfo { + + final SerializedKey firstKey; + + DictionaryBlockMeta(SerializedKey firstKey, long offset, int length) { + super(offset, length); + this.firstKey = firstKey; + } + } + + static class DictionaryBlock { + + final List entries; + + DictionaryBlock(List entries) { + this.entries = entries; + } + } + + static class DictionaryEntry { + + final SerializedKey key; + final BlockInfo bitmapBlock; + + DictionaryEntry(SerializedKey key, BlockInfo bitmapBlock) { + this.key = key; + this.bitmapBlock = bitmapBlock; + } + + int estimatedSize() { + return estimatedVarLenIntSize(key.bytes().length) + + key.bytes().length + + estimatedVarLenLongSize(bitmapBlock.offset) + + estimatedVarLenIntSize(bitmapBlock.length); + } + } + + private static class DictionaryBlockBuilder { + + private final List entries = new ArrayList<>(); + private int entriesSize; + + boolean hasEntries() { + return !entries.isEmpty(); + } + + int estimatedSize() { + return estimatedVarLenIntSize(entries.size()) + entriesSize; + } + + int estimatedSizeAfter(DictionaryEntry entry) { + return estimatedVarLenIntSize(entries.size() + 1) + entriesSize + entry.estimatedSize(); + } + + void add(DictionaryEntry entry) { + entries.add(entry); + entriesSize += entry.estimatedSize(); + } + + SerializedKey firstKey() { + return entries.get(0).key; + } + } + + private static class DictionaryBlocks { + + private final List blocks; + private final int valueCount; + + private DictionaryBlocks(List blocks, int valueCount) { + this.blocks = blocks; + this.valueCount = valueCount; + } + } + + private static class BlockEncoding { + + private final byte[] bytes; + private final int length; + private final BlockCompressionType compressionType; + + private BlockEncoding(byte[] bytes, int length, BlockCompressionType compressionType) { + this.bytes = bytes; + this.length = length; + this.compressionType = compressionType; + } + } + + static class Footer { + + final BlockInfo nullRowsBlock; + final BlockInfo nonNullRowsBlock; + final BlockInfo indexBlock; + + Footer(BlockInfo nullRowsBlock, BlockInfo nonNullRowsBlock, BlockInfo indexBlock) { + this.nullRowsBlock = nullRowsBlock; + this.nonNullRowsBlock = nonNullRowsBlock; + this.indexBlock = indexBlock; + } + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexOptions.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexOptions.java new file mode 100644 index 000000000000..36c83897c8b5 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexOptions.java @@ -0,0 +1,55 @@ +/* + * 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.bitmap; + +import org.apache.paimon.options.ConfigOption; +import org.apache.paimon.options.ConfigOptions; +import org.apache.paimon.options.MemorySize; + +/** Options for bitmap global index. */ +public class BitmapGlobalIndexOptions { + + public static final ConfigOption BITMAP_INDEX_DICTIONARY_BLOCK_SIZE = + ConfigOptions.key("bitmap-index.dictionary-block-size") + .memoryType() + .defaultValue(MemorySize.ofKibiBytes(16)) + .withDescription("The target dictionary block size for bitmap global index."); + + public static final ConfigOption BITMAP_INDEX_COMPRESSION = + ConfigOptions.key("bitmap-index.compression") + .stringType() + .defaultValue("none") + .withDescription( + "The compression algorithm to use for bitmap dictionary blocks."); + + public static final ConfigOption BITMAP_INDEX_COMPRESSION_LEVEL = + ConfigOptions.key("bitmap-index.compression-level") + .intType() + .defaultValue(1) + .withDescription("The compression level of the bitmap dictionary block codec."); + + public static final ConfigOption BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE = + ConfigOptions.key("bitmap-index.fallback-scan-max-size") + .memoryType() + .defaultValue(MemorySize.ofMebiBytes(256)) + .withDescription( + "The maximum total bitmap global index file size to allow fallback " + + "dictionary scans for predicates that cannot use direct " + + "bitmap lookup."); +} diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexWriter.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexWriter.java new file mode 100644 index 000000000000..25469d962d08 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexWriter.java @@ -0,0 +1,121 @@ +/* + * 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.bitmap; + +import org.apache.paimon.compression.BlockCompressionFactory; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.KeySerializer; +import org.apache.paimon.globalindex.ResultEntry; +import org.apache.paimon.globalindex.SortedIndexFileMeta; +import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** The {@link GlobalIndexSingleColumnWriter} implementation for bitmap index. */ +public class BitmapGlobalIndexWriter implements GlobalIndexSingleColumnWriter { + + private final GlobalIndexFileWriter fileWriter; + private final KeySerializer keySerializer; + private final Comparator comparator; + private final int dictionaryBlockSize; + @Nullable private final BlockCompressionFactory compressionFactory; + private final Map bitmaps; + private final RoaringNavigableMap64 nullRows; + private final RoaringNavigableMap64 nonNullRows; + + private long rowCount; + private Object firstKey; + private Object lastKey; + + BitmapGlobalIndexWriter( + GlobalIndexFileWriter fileWriter, + KeySerializer keySerializer, + int dictionaryBlockSize, + @Nullable BlockCompressionFactory compressionFactory) { + this.fileWriter = fileWriter; + this.keySerializer = keySerializer; + this.comparator = keySerializer.createComparator(); + this.dictionaryBlockSize = dictionaryBlockSize; + this.compressionFactory = compressionFactory; + this.bitmaps = new LinkedHashMap<>(); + this.nullRows = new RoaringNavigableMap64(); + this.nonNullRows = new RoaringNavigableMap64(); + } + + @Override + public void write(@Nullable Object key, long relativeRowId) { + rowCount++; + if (key == null) { + nullRows.add(relativeRowId); + return; + } + + nonNullRows.add(relativeRowId); + updateMinMax(key); + BitmapGlobalIndexFormat.SerializedKey serializedKey = + BitmapGlobalIndexFormat.SerializedKey.fromObject(keySerializer, key); + bitmaps.computeIfAbsent(serializedKey, k -> new RoaringNavigableMap64()).add(relativeRowId); + } + + @Override + public List finish() { + if (rowCount == 0) { + return Collections.emptyList(); + } + + String fileName = fileWriter.newFileName(BitmapGlobalIndexerFactory.IDENTIFIER); + try (PositionOutputStream outputStream = fileWriter.newOutputStream(fileName)) { + BitmapGlobalIndexFormat.write( + outputStream, + nullRows, + nonNullRows, + bitmaps, + dictionaryBlockSize, + compressionFactory); + } catch (IOException e) { + throw new RuntimeException("Error in closing bitmap index writer.", e); + } + + byte[] meta = + new SortedIndexFileMeta( + firstKey == null ? null : keySerializer.serialize(firstKey), + lastKey == null ? null : keySerializer.serialize(lastKey), + !nullRows.isEmpty()) + .serialize(); + return Collections.singletonList(new ResultEntry(fileName, rowCount, meta)); + } + + private void updateMinMax(Object key) { + if (firstKey == null || comparator.compare(key, firstKey) < 0) { + firstKey = key; + } + if (lastKey == null || comparator.compare(key, lastKey) > 0) { + lastKey = key; + } + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java new file mode 100644 index 000000000000..6a97d70bccef --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexer.java @@ -0,0 +1,77 @@ +/* + * 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.bitmap; + +import org.apache.paimon.compression.BlockCompressionFactory; +import org.apache.paimon.compression.CompressOptions; +import org.apache.paimon.globalindex.GlobalIndexIOMeta; +import org.apache.paimon.globalindex.GlobalIndexReader; +import org.apache.paimon.globalindex.GlobalIndexer; +import org.apache.paimon.globalindex.KeySerializer; +import org.apache.paimon.globalindex.io.GlobalIndexFileReader; +import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; +import org.apache.paimon.options.Options; +import org.apache.paimon.types.DataField; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.List; +import java.util.concurrent.ExecutorService; + +/** The {@link GlobalIndexer} for bitmap index. */ +public class BitmapGlobalIndexer implements GlobalIndexer { + + private final KeySerializer keySerializer; + private final int dictionaryBlockSize; + @Nullable private final BlockCompressionFactory compressionFactory; + private final long fallbackScanMaxSize; + + public BitmapGlobalIndexer(DataField dataField, Options options) { + this.keySerializer = KeySerializer.create(dataField.type()); + this.dictionaryBlockSize = + (int) + options.get(BitmapGlobalIndexOptions.BITMAP_INDEX_DICTIONARY_BLOCK_SIZE) + .getBytes(); + CompressOptions compressOptions = + new CompressOptions( + options.get(BitmapGlobalIndexOptions.BITMAP_INDEX_COMPRESSION), + options.get(BitmapGlobalIndexOptions.BITMAP_INDEX_COMPRESSION_LEVEL)); + this.compressionFactory = BlockCompressionFactory.create(compressOptions); + this.fallbackScanMaxSize = + options.get(BitmapGlobalIndexOptions.BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE) + .getBytes(); + } + + @Override + public BitmapGlobalIndexWriter createWriter(GlobalIndexFileWriter fileWriter) + throws IOException { + return new BitmapGlobalIndexWriter( + fileWriter, keySerializer, dictionaryBlockSize, compressionFactory); + } + + @Override + public GlobalIndexReader createReader( + GlobalIndexFileReader fileReader, + List files, + ExecutorService executor) { + return new LazyFilteredBitmapReader( + fileReader, files, keySerializer, fallbackScanMaxSize, executor); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexerFactory.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexerFactory.java new file mode 100644 index 000000000000..4bb128d9ccc2 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapGlobalIndexerFactory.java @@ -0,0 +1,40 @@ +/* + * 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.bitmap; + +import org.apache.paimon.globalindex.GlobalIndexer; +import org.apache.paimon.globalindex.GlobalIndexerFactory; +import org.apache.paimon.options.Options; +import org.apache.paimon.types.DataField; + +/** The {@link GlobalIndexerFactory} for bitmap index. */ +public class BitmapGlobalIndexerFactory implements GlobalIndexerFactory { + + public static final String IDENTIFIER = "bitmap"; + + @Override + public String identifier() { + return IDENTIFIER; + } + + @Override + public GlobalIndexer create(DataField dataField, Options options) { + return new BitmapGlobalIndexer(dataField, options); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapIndexReader.java new file mode 100644 index 000000000000..c5c46394a77d --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/BitmapIndexReader.java @@ -0,0 +1,491 @@ +/* + * 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.bitmap; + +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.globalindex.GlobalIndexIOMeta; +import org.apache.paimon.globalindex.GlobalIndexResult; +import org.apache.paimon.globalindex.KeySerializer; +import org.apache.paimon.globalindex.io.GlobalIndexFileReader; +import org.apache.paimon.memory.MemorySlice; +import org.apache.paimon.utils.IOUtils; +import org.apache.paimon.utils.LazyField; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import java.io.Closeable; +import java.io.IOException; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Predicate; + +/** Reader for one bitmap global index file. */ +class BitmapIndexReader implements BitmapGlobalIndexFormat.SeekableReader, Closeable { + + private final SeekableInputStream input; + private final KeySerializer keySerializer; + private final Comparator comparator; + private final LazyField nullRows; + private final LazyField nonNullRows; + private final LazyField> dictionaryBlocks; + private final Map< + BitmapGlobalIndexFormat.DictionaryBlockMeta, + BitmapGlobalIndexFormat.DictionaryBlock> + dictionaryBlockCache; + + BitmapIndexReader( + KeySerializer keySerializer, GlobalIndexFileReader fileReader, GlobalIndexIOMeta meta) + throws IOException { + this.keySerializer = keySerializer; + this.comparator = keySerializer.createComparator(); + this.dictionaryBlockCache = new ConcurrentHashMap<>(); + this.input = fileReader.getInputStream(meta); + try { + BitmapGlobalIndexFormat.Footer footer = + BitmapGlobalIndexFormat.readFooter(this, meta.fileSize()); + this.nullRows = + new LazyField<>( + () -> + BitmapGlobalIndexFormat.readBitmapUnchecked( + this, footer.nullRowsBlock)); + this.nonNullRows = + new LazyField<>( + () -> + BitmapGlobalIndexFormat.readBitmapUnchecked( + this, footer.nonNullRowsBlock)); + this.dictionaryBlocks = + new LazyField<>( + () -> + BitmapGlobalIndexFormat.readIndexBlockUnchecked( + this, footer.indexBlock)); + } catch (IOException | RuntimeException e) { + IOUtils.closeQuietly(input); + throw e; + } + } + + Optional visitIsNotNull() { + return createResult(isNotNull()); + } + + Optional visitIsNull() { + return createResult(isNull()); + } + + Optional visitStartsWith(Object literal) { + return createResult(startsWith(literal)); + } + + Optional visitEndsWith(Object literal) { + return createResult(endsWith(literal)); + } + + Optional visitContains(Object literal) { + return createResult(contains(literal)); + } + + Optional visitLessThan(Object literal) { + return createResult(lessThan(literal)); + } + + Optional visitGreaterOrEqual(Object literal) { + return createResult(greaterOrEqual(literal)); + } + + Optional visitNotEqual(Object literal) { + return createResult(notEqual(literal)); + } + + Optional visitLessOrEqual(Object literal) { + return createResult(lessOrEqual(literal)); + } + + Optional visitEqual(Object literal) { + return createResult(equal(literal)); + } + + Optional visitGreaterThan(Object literal) { + return createResult(greaterThan(literal)); + } + + Optional visitIn(List literals) { + return createResult(in(literals)); + } + + Optional visitNotIn(List literals) { + return createResult(notIn(literals)); + } + + Optional visitBetween(Object from, Object to) { + return createResult(between(from, to)); + } + + RoaringNavigableMap64 like(Predicate keyPredicate) { + return scanDictionary(keyPredicate); + } + + RoaringNavigableMap64 lessThan(Object literal) { + if (literal == null) { + return new RoaringNavigableMap64(); + } + return scanDictionary(key -> comparator.compare(key, literal) < 0); + } + + RoaringNavigableMap64 greaterThan(Object literal) { + if (literal == null) { + return new RoaringNavigableMap64(); + } + return scanDictionary(key -> comparator.compare(key, literal) > 0); + } + + @Override + public synchronized byte[] read(long offset, int length) throws IOException { + input.seek(offset); + byte[] bytes = new byte[length]; + IOUtils.readFully(input, bytes); + return bytes; + } + + @Override + public void close() throws IOException { + input.close(); + } + + private static Optional createResult(RoaringNavigableMap64 bitmap) { + return Optional.of(GlobalIndexResult.create(bitmap)); + } + + private RoaringNavigableMap64 isNull() { + return copy(nullRows.get()); + } + + private RoaringNavigableMap64 isNotNull() { + return copy(nonNullRows.get()); + } + + private RoaringNavigableMap64 equal(Object literal) { + if (literal == null) { + return new RoaringNavigableMap64(); + } + + BitmapGlobalIndexFormat.BlockInfo bitmapBlock = + findBitmapBlock( + BitmapGlobalIndexFormat.SerializedKey.fromObject(keySerializer, literal)); + if (bitmapBlock == null) { + return new RoaringNavigableMap64(); + } + + return readBitmap(bitmapBlock); + } + + private RoaringNavigableMap64 in(List literals) { + RoaringNavigableMap64 result = new RoaringNavigableMap64(); + Set keys = new HashSet<>(); + for (Object literal : literals) { + if (literal == null) { + continue; + } + + BitmapGlobalIndexFormat.SerializedKey key = + BitmapGlobalIndexFormat.SerializedKey.fromObject(keySerializer, literal); + if (!keys.add(key)) { + continue; + } + + BitmapGlobalIndexFormat.BlockInfo bitmapBlock = findBitmapBlock(key); + if (bitmapBlock != null) { + result.or(readBitmap(bitmapBlock)); + } + } + return result; + } + + private RoaringNavigableMap64 startsWith(Object literal) { + RoaringNavigableMap64 result = new RoaringNavigableMap64(); + BitmapGlobalIndexFormat.SerializedKey prefix = + BitmapGlobalIndexFormat.SerializedKey.fromObject(keySerializer, literal); + if (prefix.bytes().length == 0) { + return isNotNull(); + } + BitmapGlobalIndexFormat.SerializedKey upperBound = prefixUpperBound(prefix); + + List blocks = dictionaryBlocks.get(); + int index = firstPossibleDictionaryBlock(blocks, prefix); + while (index < blocks.size()) { + if (upperBound != null && blocks.get(index).firstKey.compareTo(upperBound) >= 0) { + return result; + } + + BitmapGlobalIndexFormat.DictionaryBlock dictionaryBlock = + dictionaryBlock(blocks.get(index)); + for (BitmapGlobalIndexFormat.DictionaryEntry entry : dictionaryBlock.entries) { + int compare = entry.key.compareTo(prefix); + if (compare < 0) { + continue; + } + if (!startsWith(entry.key, prefix)) { + return result; + } + result.or(readBitmap(entry.bitmapBlock)); + } + index++; + } + return result; + } + + private RoaringNavigableMap64 endsWith(Object literal) { + BitmapGlobalIndexFormat.SerializedKey suffix = + BitmapGlobalIndexFormat.SerializedKey.fromObject(keySerializer, literal); + if (suffix.bytes().length == 0) { + return isNotNull(); + } + return scanSerializedDictionary(key -> endsWith(key, suffix)); + } + + private RoaringNavigableMap64 contains(Object literal) { + BitmapGlobalIndexFormat.SerializedKey infix = + BitmapGlobalIndexFormat.SerializedKey.fromObject(keySerializer, literal); + if (infix.bytes().length == 0) { + return isNotNull(); + } + return scanSerializedDictionary(key -> contains(key, infix)); + } + + private RoaringNavigableMap64 greaterOrEqual(Object literal) { + if (literal == null) { + return new RoaringNavigableMap64(); + } + return scanDictionary(key -> comparator.compare(key, literal) >= 0); + } + + private RoaringNavigableMap64 lessOrEqual(Object literal) { + if (literal == null) { + return new RoaringNavigableMap64(); + } + return scanDictionary(key -> comparator.compare(key, literal) <= 0); + } + + private RoaringNavigableMap64 between(Object from, Object to) { + if (from == null || to == null || comparator.compare(from, to) > 0) { + return new RoaringNavigableMap64(); + } + return scanDictionary( + key -> comparator.compare(key, from) >= 0 && comparator.compare(key, to) <= 0); + } + + private RoaringNavigableMap64 notEqual(Object literal) { + if (literal == null) { + return new RoaringNavigableMap64(); + } + + RoaringNavigableMap64 result = isNotNull(); + result.andNot(equal(literal)); + return result; + } + + private RoaringNavigableMap64 notIn(List literals) { + for (Object literal : literals) { + if (literal == null) { + return new RoaringNavigableMap64(); + } + } + + RoaringNavigableMap64 result = isNotNull(); + result.andNot(in(literals)); + return result; + } + + private RoaringNavigableMap64 scanDictionary(Predicate keyPredicate) { + RoaringNavigableMap64 result = new RoaringNavigableMap64(); + for (BitmapGlobalIndexFormat.DictionaryBlockMeta blockMeta : dictionaryBlocks.get()) { + BitmapGlobalIndexFormat.DictionaryBlock dictionaryBlock = dictionaryBlock(blockMeta); + for (BitmapGlobalIndexFormat.DictionaryEntry entry : dictionaryBlock.entries) { + Object key = keySerializer.deserialize(MemorySlice.wrap(entry.key.bytes())); + if (keyPredicate.test(key)) { + result.or(readBitmap(entry.bitmapBlock)); + } + } + } + return result; + } + + private RoaringNavigableMap64 scanSerializedDictionary( + Predicate keyPredicate) { + RoaringNavigableMap64 result = new RoaringNavigableMap64(); + for (BitmapGlobalIndexFormat.DictionaryBlockMeta blockMeta : dictionaryBlocks.get()) { + BitmapGlobalIndexFormat.DictionaryBlock dictionaryBlock = dictionaryBlock(blockMeta); + for (BitmapGlobalIndexFormat.DictionaryEntry entry : dictionaryBlock.entries) { + if (keyPredicate.test(entry.key)) { + result.or(readBitmap(entry.bitmapBlock)); + } + } + } + return result; + } + + private int firstPossibleDictionaryBlock( + List blocks, + BitmapGlobalIndexFormat.SerializedKey key) { + int index = findDictionaryBlockIndex(blocks, key); + return Math.max(index, 0); + } + + private BitmapGlobalIndexFormat.BlockInfo findBitmapBlock( + BitmapGlobalIndexFormat.SerializedKey key) { + List blocks = dictionaryBlocks.get(); + if (blocks.isEmpty()) { + return null; + } + + int index = findDictionaryBlockIndex(blocks, key); + if (index < 0) { + return null; + } + + BitmapGlobalIndexFormat.DictionaryBlock dictionaryBlock = + dictionaryBlock(blocks.get(index)); + for (BitmapGlobalIndexFormat.DictionaryEntry entry : dictionaryBlock.entries) { + int compare = entry.key.compareTo(key); + if (compare == 0) { + return entry.bitmapBlock; + } else if (compare > 0) { + return null; + } + } + return null; + } + + private RoaringNavigableMap64 readBitmap(BitmapGlobalIndexFormat.BlockInfo bitmapBlock) { + try { + return BitmapGlobalIndexFormat.readBitmap(this, bitmapBlock); + } catch (IOException e) { + throw new RuntimeException("Failed to read bitmap global index block.", e); + } + } + + private int findDictionaryBlockIndex( + List blocks, + BitmapGlobalIndexFormat.SerializedKey key) { + int low = 0; + int high = blocks.size() - 1; + while (low <= high) { + int mid = (low + high) >>> 1; + int compare = blocks.get(mid).firstKey.compareTo(key); + if (compare <= 0) { + low = mid + 1; + } else { + high = mid - 1; + } + } + return high; + } + + private BitmapGlobalIndexFormat.DictionaryBlock dictionaryBlock( + BitmapGlobalIndexFormat.DictionaryBlockMeta blockMeta) { + return dictionaryBlockCache.computeIfAbsent(blockMeta, this::readDictionaryBlock); + } + + private BitmapGlobalIndexFormat.DictionaryBlock readDictionaryBlock( + BitmapGlobalIndexFormat.DictionaryBlockMeta blockMeta) { + try { + return BitmapGlobalIndexFormat.readDictionaryBlock(this, blockMeta); + } catch (IOException e) { + throw new RuntimeException("Failed to read bitmap dictionary block.", e); + } + } + + private static boolean startsWith( + BitmapGlobalIndexFormat.SerializedKey key, + BitmapGlobalIndexFormat.SerializedKey prefix) { + byte[] keyBytes = key.bytes(); + byte[] prefixBytes = prefix.bytes(); + if (keyBytes.length < prefixBytes.length) { + return false; + } + for (int i = 0; i < prefixBytes.length; i++) { + if (keyBytes[i] != prefixBytes[i]) { + return false; + } + } + return true; + } + + private static boolean endsWith( + BitmapGlobalIndexFormat.SerializedKey key, + BitmapGlobalIndexFormat.SerializedKey suffix) { + byte[] keyBytes = key.bytes(); + byte[] suffixBytes = suffix.bytes(); + if (keyBytes.length < suffixBytes.length) { + return false; + } + int keyOffset = keyBytes.length - suffixBytes.length; + for (int i = 0; i < suffixBytes.length; i++) { + if (keyBytes[keyOffset + i] != suffixBytes[i]) { + return false; + } + } + return true; + } + + private static boolean contains( + BitmapGlobalIndexFormat.SerializedKey key, + BitmapGlobalIndexFormat.SerializedKey infix) { + byte[] keyBytes = key.bytes(); + byte[] infixBytes = infix.bytes(); + if (keyBytes.length < infixBytes.length) { + return false; + } + for (int i = 0; i <= keyBytes.length - infixBytes.length; i++) { + boolean found = true; + for (int j = 0; j < infixBytes.length; j++) { + if (keyBytes[i + j] != infixBytes[j]) { + found = false; + break; + } + } + if (found) { + return true; + } + } + return false; + } + + private static BitmapGlobalIndexFormat.SerializedKey prefixUpperBound( + BitmapGlobalIndexFormat.SerializedKey prefix) { + byte[] prefixBytes = prefix.bytes(); + for (int i = prefixBytes.length - 1; i >= 0; i--) { + int unsignedByte = prefixBytes[i] & 0xFF; + if (unsignedByte != 0xFF) { + byte[] upperBound = new byte[i + 1]; + System.arraycopy(prefixBytes, 0, upperBound, 0, i + 1); + upperBound[i] = (byte) (unsignedByte + 1); + return new BitmapGlobalIndexFormat.SerializedKey(upperBound); + } + } + return null; + } + + private static RoaringNavigableMap64 copy(RoaringNavigableMap64 bitmap) { + return RoaringNavigableMap64.or(new RoaringNavigableMap64(), bitmap); + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapReader.java new file mode 100644 index 000000000000..69c083a9fa25 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapReader.java @@ -0,0 +1,152 @@ +/* + * 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.bitmap; + +import org.apache.paimon.globalindex.GlobalIndexIOMeta; +import org.apache.paimon.globalindex.GlobalIndexResult; +import org.apache.paimon.globalindex.KeySerializer; +import org.apache.paimon.globalindex.SortedFileGlobalIndexReader; +import org.apache.paimon.globalindex.io.GlobalIndexFileReader; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import java.io.IOException; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ExecutorService; +import java.util.function.Function; + +/** Reader for bitmap global index files. */ +public class LazyFilteredBitmapReader extends SortedFileGlobalIndexReader { + + private final GlobalIndexFileReader fileReader; + private final KeySerializer keySerializer; + + LazyFilteredBitmapReader( + GlobalIndexFileReader fileReader, + List files, + KeySerializer keySerializer, + long fallbackScanMaxSize, + ExecutorService executor) { + super(files, keySerializer, fallbackScanMaxSize, executor); + this.fileReader = fileReader; + this.keySerializer = keySerializer; + } + + @Override + protected Optional visitIsNotNull(BitmapIndexReader reader) { + return reader.visitIsNotNull(); + } + + @Override + protected Optional visitIsNull(BitmapIndexReader reader) { + return reader.visitIsNull(); + } + + @Override + protected Optional visitStartsWith( + BitmapIndexReader reader, Object literal) { + return reader.visitStartsWith(literal); + } + + @Override + protected Optional visitEndsWith(BitmapIndexReader reader, Object literal) { + return reader.visitEndsWith(literal); + } + + @Override + protected Optional visitContains(BitmapIndexReader reader, Object literal) { + return reader.visitContains(literal); + } + + @Override + protected Optional visitLessThan(BitmapIndexReader reader, Object literal) { + return reader.visitLessThan(literal); + } + + @Override + protected Optional visitGreaterOrEqual( + BitmapIndexReader reader, Object literal) { + return reader.visitGreaterOrEqual(literal); + } + + @Override + protected Optional visitNotEqual(BitmapIndexReader reader, Object literal) { + return reader.visitNotEqual(literal); + } + + @Override + protected Optional visitLessOrEqual( + BitmapIndexReader reader, Object literal) { + return reader.visitLessOrEqual(literal); + } + + @Override + protected Optional visitEqual(BitmapIndexReader reader, Object literal) { + return reader.visitEqual(literal); + } + + @Override + protected Optional visitGreaterThan( + BitmapIndexReader reader, Object literal) { + return reader.visitGreaterThan(literal); + } + + @Override + protected Optional visitIn(BitmapIndexReader reader, List literals) { + return reader.visitIn(literals); + } + + @Override + protected Optional visitNotIn( + BitmapIndexReader reader, List literals) { + return reader.visitNotIn(literals); + } + + @Override + protected Optional visitBetween( + BitmapIndexReader reader, Object from, Object to) { + return reader.visitBetween(from, to); + } + + @Override + protected RoaringNavigableMap64 like( + BitmapIndexReader reader, Function keyPredicate) { + return reader.like(keyPredicate::apply); + } + + @Override + protected RoaringNavigableMap64 lessThan(BitmapIndexReader reader, Object literal) { + return reader.lessThan(literal); + } + + @Override + protected RoaringNavigableMap64 greaterThan(BitmapIndexReader reader, Object literal) { + return reader.greaterThan(literal); + } + + @Override + protected BitmapIndexReader openReader(GlobalIndexIOMeta meta) { + try { + return new BitmapIndexReader(keySerializer, fileReader, meta); + } catch (IOException e) { + throw new RuntimeException( + "Can't create bitmap index reader for " + meta.filePath(), e); + } + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java index 99d72317abaa..24be4096058a 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexer.java @@ -23,6 +23,7 @@ import org.apache.paimon.globalindex.GlobalIndexIOMeta; import org.apache.paimon.globalindex.GlobalIndexReader; import org.apache.paimon.globalindex.GlobalIndexer; +import org.apache.paimon.globalindex.KeySerializer; import org.apache.paimon.globalindex.io.GlobalIndexFileReader; import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; import org.apache.paimon.io.cache.CacheManager; @@ -61,11 +62,14 @@ public class BTreeGlobalIndexer implements GlobalIndexer { private final KeySerializer keySerializer; private final Options options; + private final long fallbackScanMaxSize; private final LazyField cacheManager; public BTreeGlobalIndexer(DataField dataField, Options options) { this.keySerializer = KeySerializer.create(dataField.type()); this.options = options; + this.fallbackScanMaxSize = + options.get(BTreeIndexOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE).getBytes(); this.cacheManager = new LazyField<>( () -> @@ -96,6 +100,11 @@ public GlobalIndexReader createReader( List files, ExecutorService executor) { return new LazyFilteredBTreeReader( - files, keySerializer, fileReader, cacheManager.get(), executor); + files, + keySerializer, + fileReader, + cacheManager.get(), + fallbackScanMaxSize, + executor); } } diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java index ab8636592ac3..d46ef0d92a3d 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexOptions.java @@ -55,6 +55,15 @@ public class BTreeIndexOptions { .defaultValue(0.1) .withDescription("The high priority pool ratio to use for BTreeIndex"); + public static final ConfigOption BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE = + ConfigOptions.key("btree-index.fallback-scan-max-size") + .memoryType() + .defaultValue(MemorySize.ofMebiBytes(256)) + .withDescription( + "The maximum total BTree global index file size to allow fallback " + + "index scans for predicates that cannot use direct lookup. " + + "Set to 0 bytes to disable fallback scans."); + public static final ConfigOption BTREE_INDEX_RECORDS_PER_RANGE = ConfigOptions.key("btree-index.records-per-range") .longType() diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java index 69706c4b6628..92816e909ea1 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java @@ -22,6 +22,8 @@ import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.globalindex.GlobalIndexIOMeta; import org.apache.paimon.globalindex.GlobalIndexResult; +import org.apache.paimon.globalindex.KeySerializer; +import org.apache.paimon.globalindex.SortedIndexFileMeta; import org.apache.paimon.globalindex.io.GlobalIndexFileReader; import org.apache.paimon.io.cache.CacheManager; import org.apache.paimon.memory.MemorySegment; @@ -135,7 +137,8 @@ public BTreeIndexReader( throws IOException { this.keySerializer = keySerializer; this.comparator = keySerializer.createComparator(); - BTreeIndexMeta indexMeta = BTreeIndexMeta.deserialize(globalIndexIOMeta.metadata()); + SortedIndexFileMeta indexMeta = + SortedIndexFileMeta.deserialize(globalIndexIOMeta.metadata()); if (indexMeta.getFirstKey() != null) { this.minKey = keySerializer.deserialize(MemorySlice.wrap(indexMeta.getFirstKey())); this.maxKey = keySerializer.deserialize(MemorySlice.wrap(indexMeta.getLastKey())); diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java index 5053b2dbb18b..ccef1ad471e8 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexWriter.java @@ -21,7 +21,9 @@ import org.apache.paimon.compression.BlockCompressionFactory; import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.KeySerializer; import org.apache.paimon.globalindex.ResultEntry; +import org.apache.paimon.globalindex.SortedIndexFileMeta; import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; import org.apache.paimon.memory.MemorySlice; import org.apache.paimon.memory.MemorySliceOutput; @@ -171,7 +173,7 @@ public List finish() { } byte[] metaBytes = - new BTreeIndexMeta( + new SortedIndexFileMeta( firstKey == null ? null : keySerializer.serialize(firstKey), lastKey == null ? null : keySerializer.serialize(lastKey), nullBitmap.initialized()) diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java index 85eb93500e9b..e7e6217e2720 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java @@ -18,213 +18,137 @@ package org.apache.paimon.globalindex.btree; -import org.apache.paimon.fs.Path; import org.apache.paimon.globalindex.GlobalIndexIOMeta; -import org.apache.paimon.globalindex.GlobalIndexReader; import org.apache.paimon.globalindex.GlobalIndexResult; +import org.apache.paimon.globalindex.KeySerializer; +import org.apache.paimon.globalindex.SortedFileGlobalIndexReader; import org.apache.paimon.globalindex.io.GlobalIndexFileReader; import org.apache.paimon.io.cache.CacheManager; import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.utils.RoaringNavigableMap64; import java.io.IOException; -import java.util.ArrayList; import java.util.List; -import java.util.Map; import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; -import java.util.function.Function; -import java.util.function.Supplier; /** * An Index Reader for BTree which dynamically filters file list by input predicate, then visits * each selected file in parallel via an executor. Each index file is synchronized independently to * allow maximum concurrency. */ -public class LazyFilteredBTreeReader implements GlobalIndexReader { +public class LazyFilteredBTreeReader extends SortedFileGlobalIndexReader { - private final BTreeFileMetaSelector fileSelector; - private final Map readerCache; private final KeySerializer keySerializer; private final CacheManager cacheManager; private final GlobalIndexFileReader fileReader; - private final ExecutorService executor; public LazyFilteredBTreeReader( List files, KeySerializer keySerializer, GlobalIndexFileReader fileReader, CacheManager cacheManager, + long fallbackScanMaxSize, ExecutorService executor) { - this.fileSelector = new BTreeFileMetaSelector(files, keySerializer); - this.readerCache = new ConcurrentHashMap<>(); + super(files, keySerializer, fallbackScanMaxSize, executor); this.cacheManager = cacheManager; this.fileReader = fileReader; this.keySerializer = keySerializer; - this.executor = executor; } @Override - public CompletableFuture> visitIsNotNull(FieldRef fieldRef) { - return visitParallel( - () -> fileSelector.visitIsNotNull(fieldRef), BTreeIndexReader::visitIsNotNull); + protected Optional visitIsNotNull(BTreeIndexReader reader) { + return reader.visitIsNotNull(); } @Override - public CompletableFuture> visitIsNull(FieldRef fieldRef) { - return visitParallel( - () -> fileSelector.visitIsNull(fieldRef), BTreeIndexReader::visitIsNull); + protected Optional visitIsNull(BTreeIndexReader reader) { + return reader.visitIsNull(); } @Override - public CompletableFuture> visitStartsWith( - FieldRef fieldRef, Object literal) { - return visitParallel( - () -> fileSelector.visitStartsWith(fieldRef, literal), - reader -> reader.visitStartsWith(literal)); + protected Optional visitStartsWith(BTreeIndexReader reader, Object literal) { + return reader.visitStartsWith(literal); } @Override - public CompletableFuture> visitEndsWith( - FieldRef fieldRef, Object literal) { - return visitParallel( - () -> fileSelector.visitEndsWith(fieldRef, literal), - reader -> reader.visitEndsWith(literal)); + protected Optional visitEndsWith(BTreeIndexReader reader, Object literal) { + return reader.visitEndsWith(literal); } @Override - public CompletableFuture> visitContains( - FieldRef fieldRef, Object literal) { - return visitParallel( - () -> fileSelector.visitContains(fieldRef, literal), - reader -> reader.visitContains(literal)); + protected Optional visitContains(BTreeIndexReader reader, Object literal) { + return reader.visitContains(literal); } @Override - public CompletableFuture> visitLike( - FieldRef fieldRef, Object literal) { - return visitParallel( - () -> fileSelector.visitLike(fieldRef, literal), - reader -> reader.visitLike(literal)); + protected Optional visitLike( + BTreeIndexReader reader, FieldRef fieldRef, Object literal) { + return reader.visitLike(literal); } @Override - public CompletableFuture> visitLessThan( - FieldRef fieldRef, Object literal) { - return visitParallel( - () -> fileSelector.visitLessThan(fieldRef, literal), - reader -> reader.visitLessThan(literal)); + protected Optional visitLessThan(BTreeIndexReader reader, Object literal) { + return reader.visitLessThan(literal); } @Override - public CompletableFuture> visitGreaterOrEqual( - FieldRef fieldRef, Object literal) { - return visitParallel( - () -> fileSelector.visitGreaterOrEqual(fieldRef, literal), - reader -> reader.visitGreaterOrEqual(literal)); + protected Optional visitGreaterOrEqual( + BTreeIndexReader reader, Object literal) { + return reader.visitGreaterOrEqual(literal); } @Override - public CompletableFuture> visitNotEqual( - FieldRef fieldRef, Object literal) { - return visitParallel( - () -> fileSelector.visitNotEqual(fieldRef, literal), - reader -> reader.visitNotEqual(literal)); + protected Optional visitNotEqual(BTreeIndexReader reader, Object literal) { + return reader.visitNotEqual(literal); } @Override - public CompletableFuture> visitLessOrEqual( - FieldRef fieldRef, Object literal) { - return visitParallel( - () -> fileSelector.visitLessOrEqual(fieldRef, literal), - reader -> reader.visitLessOrEqual(literal)); + protected Optional visitLessOrEqual( + BTreeIndexReader reader, Object literal) { + return reader.visitLessOrEqual(literal); } @Override - public CompletableFuture> visitEqual( - FieldRef fieldRef, Object literal) { - return visitParallel( - () -> fileSelector.visitEqual(fieldRef, literal), - reader -> reader.visitEqual(literal)); + protected Optional visitEqual(BTreeIndexReader reader, Object literal) { + return reader.visitEqual(literal); } @Override - public CompletableFuture> visitGreaterThan( - FieldRef fieldRef, Object literal) { - return visitParallel( - () -> fileSelector.visitGreaterThan(fieldRef, literal), - reader -> reader.visitGreaterThan(literal)); + protected Optional visitGreaterThan( + BTreeIndexReader reader, Object literal) { + return reader.visitGreaterThan(literal); } @Override - public CompletableFuture> visitIn( - FieldRef fieldRef, List literals) { - return visitParallel( - () -> fileSelector.visitIn(fieldRef, literals), reader -> reader.visitIn(literals)); + protected Optional visitIn(BTreeIndexReader reader, List literals) { + return reader.visitIn(literals); } @Override - public CompletableFuture> visitNotIn( - FieldRef fieldRef, List literals) { - return visitParallel( - () -> fileSelector.visitNotIn(fieldRef, literals), - reader -> reader.visitNotIn(literals)); + protected Optional visitNotIn( + BTreeIndexReader reader, List literals) { + return reader.visitNotIn(literals); } @Override - public CompletableFuture> visitBetween( - FieldRef fieldRef, Object from, Object to) { - return visitParallel( - () -> fileSelector.visitBetween(fieldRef, from, to), - reader -> reader.visitBetween(from, to)); + protected Optional visitBetween( + BTreeIndexReader reader, Object from, Object to) { + return reader.visitBetween(from, to); } - private CompletableFuture> visitParallel( - Supplier>> selector, - Function> visitor) { - Optional> selectedOpt = selector.get(); - if (!selectedOpt.isPresent()) { - return CompletableFuture.completedFuture(Optional.empty()); - } - List selected = selectedOpt.get(); - if (selected.isEmpty()) { - return CompletableFuture.completedFuture(Optional.of(GlobalIndexResult.createEmpty())); - } - - List>> futures = - new ArrayList<>(selected.size()); - for (GlobalIndexIOMeta meta : selected) { - futures.add( - CompletableFuture.supplyAsync( - () -> visitor.apply(getOrCreateReader(meta)), executor)); - } - return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) - .thenApply(v -> unionResults(futures)); - } - - private Optional unionResults( - List>> futures) { - Optional result = Optional.empty(); - for (CompletableFuture> future : futures) { - Optional current = future.join(); - if (!current.isPresent()) { - continue; - } - if (!result.isPresent()) { - result = current; - } else { - result = Optional.of(result.get().or(current.get())); - } - } - return result; + @Override + protected RoaringNavigableMap64 lessThan(BTreeIndexReader reader, Object literal) { + return bitmap(reader.visitLessThan(literal)); } - private BTreeIndexReader getOrCreateReader(GlobalIndexIOMeta meta) { - return readerCache.computeIfAbsent(meta.filePath(), k -> createBTreeReader(meta)); + @Override + protected RoaringNavigableMap64 greaterThan(BTreeIndexReader reader, Object literal) { + return bitmap(reader.visitGreaterThan(literal)); } - private BTreeIndexReader createBTreeReader(GlobalIndexIOMeta meta) { + @Override + protected BTreeIndexReader openReader(GlobalIndexIOMeta meta) { try { return new BTreeIndexReader(keySerializer, fileReader, meta, cacheManager); } catch (IOException e) { @@ -232,22 +156,7 @@ private BTreeIndexReader createBTreeReader(GlobalIndexIOMeta meta) { } } - @Override - public void close() throws IOException { - IOException exception = null; - for (Map.Entry entry : this.readerCache.entrySet()) { - try { - entry.getValue().close(); - } catch (IOException ioe) { - if (exception == null) { - exception = ioe; - } else { - exception.addSuppressed(ioe); - } - } - } - if (exception != null) { - throw exception; - } + private RoaringNavigableMap64 bitmap(Optional result) { + return result.get().results(); } } diff --git a/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory b/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory index 1477dcba359e..06722ccd2d78 100644 --- a/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory +++ b/paimon-common/src/main/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory @@ -14,3 +14,4 @@ # limitations under the License. org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory +org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexerFactory diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeFileMetaSelectorTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/SortedFileMetaSelectorTest.java similarity index 50% rename from paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeFileMetaSelectorTest.java rename to paimon-common/src/test/java/org/apache/paimon/globalindex/SortedFileMetaSelectorTest.java index 26a13eeec5a5..59ab26fcde66 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeFileMetaSelectorTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/SortedFileMetaSelectorTest.java @@ -16,42 +16,47 @@ * limitations under the License. */ -package org.apache.paimon.globalindex.btree; +package org.apache.paimon.globalindex; +import org.apache.paimon.data.BinaryString; import org.apache.paimon.fs.Path; -import org.apache.paimon.globalindex.GlobalIndexIOMeta; +import org.apache.paimon.memory.MemorySlice; import org.apache.paimon.memory.MemorySliceOutput; import org.apache.paimon.predicate.FieldRef; import org.apache.paimon.types.DataType; import org.apache.paimon.types.IntType; +import org.apache.paimon.types.VarCharType; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import java.util.Arrays; +import java.util.Comparator; import java.util.List; import java.util.Optional; import java.util.stream.Collectors; -/** Test class for {@link BTreeFileMetaSelector}. */ -public class BTreeFileMetaSelectorTest { +/** Test class for {@link SortedFileMetaSelector}. */ +public class SortedFileMetaSelectorTest { private List files; @BeforeEach public void setUp() { MemorySliceOutput sliceOutput = new MemorySliceOutput(4); - BTreeIndexMeta meta1 = - new BTreeIndexMeta(writeInt(1, sliceOutput), writeInt(10, sliceOutput), true); - BTreeIndexMeta meta2 = - new BTreeIndexMeta(writeInt(15, sliceOutput), writeInt(20, sliceOutput), false); - BTreeIndexMeta meta3 = - new BTreeIndexMeta(writeInt(21, sliceOutput), writeInt(30, sliceOutput), true); - BTreeIndexMeta meta4 = - new BTreeIndexMeta(writeInt(1, sliceOutput), writeInt(5, sliceOutput), false); - BTreeIndexMeta meta5 = - new BTreeIndexMeta(writeInt(19, sliceOutput), writeInt(25, sliceOutput), true); + SortedIndexFileMeta meta1 = + new SortedIndexFileMeta(writeInt(1, sliceOutput), writeInt(10, sliceOutput), true); + SortedIndexFileMeta meta2 = + new SortedIndexFileMeta( + writeInt(15, sliceOutput), writeInt(20, sliceOutput), false); + SortedIndexFileMeta meta3 = + new SortedIndexFileMeta(writeInt(21, sliceOutput), writeInt(30, sliceOutput), true); + SortedIndexFileMeta meta4 = + new SortedIndexFileMeta(writeInt(1, sliceOutput), writeInt(5, sliceOutput), false); + SortedIndexFileMeta meta5 = + new SortedIndexFileMeta(writeInt(19, sliceOutput), writeInt(25, sliceOutput), true); + SortedIndexFileMeta meta6 = new SortedIndexFileMeta(null, null, true); files = Arrays.asList( @@ -59,7 +64,8 @@ public void setUp() { new GlobalIndexIOMeta(new Path("file2"), 1, meta2.serialize()), new GlobalIndexIOMeta(new Path("file3"), 1, meta3.serialize()), new GlobalIndexIOMeta(new Path("file4"), 1, meta4.serialize()), - new GlobalIndexIOMeta(new Path("file5"), 1, meta5.serialize())); + new GlobalIndexIOMeta(new Path("file5"), 1, meta5.serialize()), + new GlobalIndexIOMeta(new Path("file6"), 1, meta6.serialize())); } @Test @@ -67,7 +73,7 @@ public void testMetaSelector() { DataType dataType = new IntType(); FieldRef ref = new FieldRef(1, "testField", dataType); KeySerializer serializer = KeySerializer.create(dataType); - BTreeFileMetaSelector selector = new BTreeFileMetaSelector(files, serializer); + SortedFileMetaSelector selector = new SortedFileMetaSelector(files, serializer); Optional> result; @@ -129,7 +135,7 @@ public void testMetaSelector() { // 2. test isNull & isNotNull result = selector.visitIsNull(ref); Assertions.assertThat(result).isNotEmpty(); - assertFiles(result.get(), Arrays.asList("file1", "file3", "file5")); + assertFiles(result.get(), Arrays.asList("file1", "file3", "file5", "file6")); result = selector.visitIsNotNull(ref); Assertions.assertThat(result).isNotEmpty(); @@ -156,6 +162,86 @@ public void testMetaSelector() { result = selector.visitBetween(ref, 40, 50); Assertions.assertThat(result).isNotEmpty(); Assertions.assertThat(result.get()).isEmpty(); + + // 5. test null literals + result = selector.visitEqual(ref, null); + Assertions.assertThat(result).isNotEmpty(); + Assertions.assertThat(result.get()).isEmpty(); + + result = selector.visitNotEqual(ref, null); + Assertions.assertThat(result).isNotEmpty(); + Assertions.assertThat(result.get()).isEmpty(); + + result = selector.visitIn(ref, Arrays.asList(null, 22)); + Assertions.assertThat(result).isNotEmpty(); + assertFiles(result.get(), Arrays.asList("file3", "file5")); + + result = selector.visitIn(ref, Arrays.asList(null, null)); + Assertions.assertThat(result).isNotEmpty(); + Assertions.assertThat(result.get()).isEmpty(); + + result = selector.visitNotIn(ref, Arrays.asList(1, null)); + Assertions.assertThat(result).isNotEmpty(); + Assertions.assertThat(result.get()).isEmpty(); + + result = selector.visitBetween(ref, 20, 10); + Assertions.assertThat(result).isNotEmpty(); + Assertions.assertThat(result.get()).isEmpty(); + } + + @Test + public void testStringPrefixSelector() { + DataType dataType = new VarCharType(); + FieldRef ref = new FieldRef(1, "testField", dataType); + KeySerializer serializer = KeySerializer.create(dataType); + List stringFiles = + Arrays.asList( + newStringFile("file1", serializer, "alpha", "azalea", true), + newStringFile("file2", serializer, "beta", "delta", false), + newStringFile("file3", serializer, "tag-001", "tag-999", false), + new GlobalIndexIOMeta( + new Path("file4"), + 1, + new SortedIndexFileMeta(null, null, true).serialize())); + SortedFileMetaSelector selector = new SortedFileMetaSelector(stringFiles, serializer); + + Optional> result; + + result = selector.visitStartsWith(ref, str("a")); + Assertions.assertThat(result).isNotEmpty(); + assertFiles(result.get(), Arrays.asList("file1")); + + result = selector.visitStartsWith(ref, str("tag-")); + Assertions.assertThat(result).isNotEmpty(); + assertFiles(result.get(), Arrays.asList("file3")); + + result = selector.visitStartsWith(ref, str("")); + Assertions.assertThat(result).isNotEmpty(); + assertFiles(result.get(), Arrays.asList("file1", "file2", "file3")); + + result = selector.visitStartsWith(ref, null); + Assertions.assertThat(result).isNotEmpty(); + Assertions.assertThat(result.get()).isEmpty(); + + result = selector.visitContains(ref, str("a")); + Assertions.assertThat(result).isNotEmpty(); + assertFiles(result.get(), Arrays.asList("file1", "file2", "file3")); + } + + @Test + public void testStartsWithUsesKeyComparatorForMetaPruning() { + DataType dataType = new VarCharType(); + FieldRef ref = new FieldRef(1, "testField", dataType); + KeySerializer serializer = new LengthPrefixedStringSerializer(); + List stringFiles = + Arrays.asList(newStringFile("file1", serializer, "b", "bb", false)); + SortedFileMetaSelector selector = new SortedFileMetaSelector(stringFiles, serializer); + + // Serialized byte order would overlap ["b", "bb"] with the prefix range ["aa", "ab"). + Optional> result = selector.visitStartsWith(ref, str("aa")); + + Assertions.assertThat(result).isNotEmpty(); + Assertions.assertThat(result.get()).isEmpty(); } private void assertFiles(List files, List expected) { @@ -172,4 +258,45 @@ private byte[] writeInt(int value, MemorySliceOutput sliceOutput) { sliceOutput.writeInt(value); return sliceOutput.toSlice().copyBytes(); } + + private GlobalIndexIOMeta newStringFile( + String fileName, + KeySerializer serializer, + String firstKey, + String lastKey, + boolean hasNulls) { + SortedIndexFileMeta meta = + new SortedIndexFileMeta( + serializer.serialize(str(firstKey)), + serializer.serialize(str(lastKey)), + hasNulls); + return new GlobalIndexIOMeta(new Path(fileName), 1, meta.serialize()); + } + + private BinaryString str(String value) { + return BinaryString.fromString(value); + } + + private static class LengthPrefixedStringSerializer implements KeySerializer { + + @Override + public byte[] serialize(Object key) { + byte[] bytes = ((BinaryString) key).toBytes(); + byte[] result = new byte[bytes.length + 1]; + result[0] = (byte) bytes.length; + System.arraycopy(bytes, 0, result, 1, bytes.length); + return result; + } + + @Override + public Object deserialize(MemorySlice data) { + byte[] bytes = data.copyBytes(); + return BinaryString.fromBytes(Arrays.copyOfRange(bytes, 1, bytes.length)); + } + + @Override + public Comparator createComparator() { + return Comparator.comparing(o -> (BinaryString) o); + } + } } diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexMetaTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/SortedIndexFileMetaTest.java similarity index 82% rename from paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexMetaTest.java rename to paimon-common/src/test/java/org/apache/paimon/globalindex/SortedIndexFileMetaTest.java index 64467cdc6333..537d6724028d 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeIndexMetaTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/SortedIndexFileMetaTest.java @@ -16,15 +16,14 @@ * limitations under the License. */ -package org.apache.paimon.globalindex.btree; +package org.apache.paimon.globalindex; import org.apache.paimon.data.BinaryString; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.fs.local.LocalFileIO; -import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; -import org.apache.paimon.globalindex.ResultEntry; +import org.apache.paimon.globalindex.btree.BTreeGlobalIndexer; import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; import org.apache.paimon.memory.MemorySliceOutput; import org.apache.paimon.options.Options; @@ -39,8 +38,8 @@ import static org.assertj.core.api.Assertions.assertThat; -/** Test for {@link BTreeIndexMeta}. */ -class BTreeIndexMetaTest { +/** Test for {@link SortedIndexFileMeta}. */ +class SortedIndexFileMetaTest { @TempDir private java.nio.file.Path tempPath; @@ -48,7 +47,7 @@ class BTreeIndexMetaTest { void testEmptyFirstKey() { byte[] lastKey = new byte[] {1, 2, 3}; - BTreeIndexMeta meta = roundTrip(new BTreeIndexMeta(new byte[0], lastKey, false)); + SortedIndexFileMeta meta = roundTrip(new SortedIndexFileMeta(new byte[0], lastKey, false)); assertThat(meta.getFirstKey()).isEmpty(); assertThat(meta.getLastKey()).containsExactly(lastKey); @@ -58,7 +57,8 @@ void testEmptyFirstKey() { @Test void testEmptyFirstAndLastKeyWithNulls() { - BTreeIndexMeta meta = roundTrip(new BTreeIndexMeta(new byte[0], new byte[0], true)); + SortedIndexFileMeta meta = + roundTrip(new SortedIndexFileMeta(new byte[0], new byte[0], true)); assertThat(meta.getFirstKey()).isEmpty(); assertThat(meta.getLastKey()).isEmpty(); @@ -68,7 +68,7 @@ void testEmptyFirstAndLastKeyWithNulls() { @Test void testOnlyNulls() { - BTreeIndexMeta meta = roundTrip(new BTreeIndexMeta(null, null, true)); + SortedIndexFileMeta meta = roundTrip(new SortedIndexFileMeta(null, null, true)); assertThat(meta.getFirstKey()).isNull(); assertThat(meta.getLastKey()).isNull(); @@ -82,7 +82,7 @@ void testWriterEmptyStringKeyWithNulls() throws Exception { writeVarCharIndex( null, BinaryString.fromString(""), BinaryString.fromString("abc")); - BTreeIndexMeta meta = BTreeIndexMeta.deserialize(results.get(0).meta()); + SortedIndexFileMeta meta = SortedIndexFileMeta.deserialize(results.get(0).meta()); assertThat(meta.getFirstKey()).isEmpty(); assertThat(meta.getLastKey()).containsExactly(BinaryString.fromString("abc").toBytes()); @@ -92,8 +92,8 @@ void testWriterEmptyStringKeyWithNulls() throws Exception { @Test void testLegacyOnlyNulls() { - BTreeIndexMeta meta = - BTreeIndexMeta.deserialize(legacyMetaBytes(new byte[0], new byte[0], true)); + SortedIndexFileMeta meta = + SortedIndexFileMeta.deserialize(legacyMetaBytes(new byte[0], new byte[0], true)); assertThat(meta.getFirstKey()).isNull(); assertThat(meta.getLastKey()).isNull(); @@ -105,8 +105,8 @@ void testLegacyOnlyNulls() { void testLegacyEmptyFirstKey() { byte[] lastKey = new byte[] {1, 2, 3}; - BTreeIndexMeta meta = - BTreeIndexMeta.deserialize(legacyMetaBytes(new byte[0], lastKey, false)); + SortedIndexFileMeta meta = + SortedIndexFileMeta.deserialize(legacyMetaBytes(new byte[0], lastKey, false)); assertThat(meta.getFirstKey()).isEmpty(); assertThat(meta.getLastKey()).containsExactly(lastKey); @@ -116,8 +116,8 @@ void testLegacyEmptyFirstKey() { @Test void testLegacyEmptyFirstAndLastKeyWithoutNulls() { - BTreeIndexMeta meta = - BTreeIndexMeta.deserialize(legacyMetaBytes(new byte[0], new byte[0], false)); + SortedIndexFileMeta meta = + SortedIndexFileMeta.deserialize(legacyMetaBytes(new byte[0], new byte[0], false)); assertThat(meta.getFirstKey()).isEmpty(); assertThat(meta.getLastKey()).isEmpty(); @@ -125,8 +125,8 @@ void testLegacyEmptyFirstAndLastKeyWithoutNulls() { assertThat(meta.onlyNulls()).isFalse(); } - private BTreeIndexMeta roundTrip(BTreeIndexMeta meta) { - return BTreeIndexMeta.deserialize(meta.serialize()); + private SortedIndexFileMeta roundTrip(SortedIndexFileMeta meta) { + return SortedIndexFileMeta.deserialize(meta.serialize()); } private byte[] legacyMetaBytes(byte[] firstKey, byte[] lastKey, boolean hasNulls) { diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapIndexReaderTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapIndexReaderTest.java new file mode 100644 index 000000000000..0a1439ed86b7 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/bitmap/LazyFilteredBitmapIndexReaderTest.java @@ -0,0 +1,490 @@ +/* + * 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.bitmap; + +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.SeekableInputStreamWrapper; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.globalindex.GlobalIndexIOMeta; +import org.apache.paimon.globalindex.GlobalIndexReader; +import org.apache.paimon.globalindex.GlobalIndexResult; +import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.ResultEntry; +import org.apache.paimon.globalindex.io.GlobalIndexFileReader; +import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; +import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.VarCharType; +import org.apache.paimon.utils.Pair; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService; +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link LazyFilteredBitmapReader}. */ +public class LazyFilteredBitmapIndexReaderTest { + + private final VarCharType dataType = new VarCharType(VarCharType.MAX_LENGTH); + private final DataField dataField = new DataField(1, "tag", dataType); + private final FieldRef fieldRef = new FieldRef(1, "tag", dataType); + + private FileIO fileIO; + private Path basePath; + private GlobalIndexFileWriter fileWriter; + private GlobalIndexFileReader fileReader; + private BitmapGlobalIndexer globalIndexer; + + @TempDir java.nio.file.Path tempPath; + + @BeforeEach + public void setUp() { + fileIO = LocalFileIO.create(); + basePath = new Path(tempPath.toUri()); + fileWriter = + new GlobalIndexFileWriter() { + @Override + public String newFileName(String prefix) { + return prefix + "-" + UUID.randomUUID() + ".index"; + } + + @Override + public PositionOutputStream newOutputStream(String fileName) + throws IOException { + return fileIO.newOutputStream(new Path(basePath, fileName), true); + } + }; + fileReader = meta -> fileIO.newInputStream(meta.filePath()); + globalIndexer = new BitmapGlobalIndexer(dataField, new Options()); + } + + @Test + public void testEqualityFamilyPredicates() throws Exception { + disableFallbackScan(); + GlobalIndexIOMeta meta = + writeData( + Arrays.asList( + Pair.of(str("A"), 0L), + Pair.of(str("B"), 1L), + Pair.of(null, 2L), + Pair.of(str("B"), 3L), + Pair.of(str("C"), 4L), + Pair.of(str("A"), 5L))); + + try (GlobalIndexReader reader = + globalIndexer.createReader( + fileReader, Collections.singletonList(meta), newDirectExecutorService())) { + assertRows(reader.visitEqual(fieldRef, str("A")).join(), 0L, 5L); + assertRows(reader.visitEqual(fieldRef, null).join()); + assertRows(reader.visitEqual(fieldRef, str("missing")).join()); + assertRows( + reader.visitIn(fieldRef, Arrays.asList(str("B"), str("C"))).join(), 1L, 3L, 4L); + assertRows(reader.visitIn(fieldRef, Arrays.asList(str("A"), null)).join(), 0L, 5L); + assertRows(reader.visitStartsWith(fieldRef, str("A")).join(), 0L, 5L); + assertRows(reader.visitStartsWith(fieldRef, str("")).join(), 0L, 1L, 3L, 4L, 5L); + assertRows(reader.visitLike(fieldRef, str("A")).join(), 0L, 5L); + assertRows(reader.visitLike(fieldRef, str("A%")).join(), 0L, 5L); + assertRows(reader.visitNotEqual(fieldRef, str("A")).join(), 1L, 3L, 4L); + assertRows(reader.visitNotEqual(fieldRef, null).join()); + assertRows( + reader.visitNotIn(fieldRef, Arrays.asList(str("B"), str("C"))).join(), 0L, 5L); + assertRows(reader.visitNotIn(fieldRef, Arrays.asList(str("B"), null)).join()); + assertRows(reader.visitIsNull(fieldRef).join(), 2L); + assertRows(reader.visitIsNotNull(fieldRef).join(), 0L, 1L, 3L, 4L, 5L); + + assertThat(reader.visitLessThan(fieldRef, str("B")).join()).isEmpty(); + assertThat(reader.visitBetween(fieldRef, str("A"), str("C")).join()).isEmpty(); + assertThat(reader.visitEndsWith(fieldRef, str("A")).join()).isEmpty(); + assertThat(reader.visitContains(fieldRef, str("A")).join()).isEmpty(); + assertThat(reader.visitLike(fieldRef, str("%A")).join()).isEmpty(); + } + } + + @Test + public void testFallbackScanPredicates() throws Exception { + GlobalIndexIOMeta meta = + writeData( + Arrays.asList( + Pair.of(str("alpha"), 0L), + Pair.of(str("beta"), 1L), + Pair.of(str("alphabet"), 2L), + Pair.of(str("delta"), 3L), + Pair.of(null, 4L))); + + try (GlobalIndexReader reader = + globalIndexer.createReader( + fileReader, Collections.singletonList(meta), newDirectExecutorService())) { + assertRows(reader.visitEndsWith(fieldRef, str("ta")).join(), 1L, 3L); + assertRows(reader.visitContains(fieldRef, str("ph")).join(), 0L, 2L); + assertRows(reader.visitLike(fieldRef, str("%ha%")).join(), 0L, 2L); + assertRows(reader.visitLessThan(fieldRef, str("delta")).join(), 0L, 1L, 2L); + assertRows(reader.visitLessOrEqual(fieldRef, str("beta")).join(), 0L, 1L, 2L); + assertRows(reader.visitGreaterThan(fieldRef, str("beta")).join(), 3L); + assertRows(reader.visitGreaterOrEqual(fieldRef, str("beta")).join(), 1L, 3L); + assertRows(reader.visitBetween(fieldRef, str("beta"), str("delta")).join(), 1L, 3L); + } + } + + @Test + public void testCompressedDictionaryBlocks() throws Exception { + Options compressedOptions = new Options(); + compressedOptions.set(BitmapGlobalIndexOptions.BITMAP_INDEX_COMPRESSION, "lz4"); + compressedOptions.set( + BitmapGlobalIndexOptions.BITMAP_INDEX_DICTIONARY_BLOCK_SIZE, + org.apache.paimon.options.MemorySize.ofKibiBytes(256)); + globalIndexer = new BitmapGlobalIndexer(dataField, compressedOptions); + + List> rows = new ArrayList<>(); + String prefix = + "very-long-common-prefix-for-bitmap-dictionary-compression-" + + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-"; + for (int i = 0; i < 300; i++) { + rows.add(Pair.of(str(String.format("%s%05d", prefix, i)), (long) i)); + } + GlobalIndexIOMeta compressed = writeData(rows); + + Options plainOptions = new Options(); + plainOptions.set( + BitmapGlobalIndexOptions.BITMAP_INDEX_DICTIONARY_BLOCK_SIZE, + org.apache.paimon.options.MemorySize.ofKibiBytes(256)); + globalIndexer = new BitmapGlobalIndexer(dataField, plainOptions); + GlobalIndexIOMeta plain = writeData(rows); + + assertThat(compressed.fileSize()).isLessThan(plain.fileSize()); + + try (GlobalIndexReader reader = + globalIndexer.createReader( + fileReader, + Collections.singletonList(compressed), + newDirectExecutorService())) { + assertRows(reader.visitEqual(fieldRef, str(prefix + "00123")).join(), 123L); + assertRows( + reader.visitStartsWith(fieldRef, str(prefix + "0012")).join(), + 120L, + 121L, + 122L, + 123L, + 124L, + 125L, + 126L, + 127L, + 128L, + 129L); + } + } + + @Test + public void testFallbackScanDisabledByBudget() throws Exception { + Options options = new Options(); + options.set( + BitmapGlobalIndexOptions.BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE, + org.apache.paimon.options.MemorySize.ofBytes(1)); + globalIndexer = new BitmapGlobalIndexer(dataField, options); + + GlobalIndexIOMeta meta = + writeData(Arrays.asList(Pair.of(str("alpha"), 0L), Pair.of(str("beta"), 1L))); + + try (GlobalIndexReader reader = + globalIndexer.createReader( + fileReader, Collections.singletonList(meta), newDirectExecutorService())) { + assertThat(reader.visitEndsWith(fieldRef, str("ta")).join()).isEmpty(); + assertThat(reader.visitContains(fieldRef, str("ph")).join()).isEmpty(); + assertThat(reader.visitLike(fieldRef, str("%ha%")).join()).isEmpty(); + assertThat(reader.visitLessThan(fieldRef, str("beta")).join()).isEmpty(); + assertRows(reader.visitEqual(fieldRef, str("beta")).join(), 1L); + } + } + + @Test + public void testMultiFileComplementsUsePerFileOwnedRows() throws Exception { + GlobalIndexIOMeta first = + writeData( + Arrays.asList( + Pair.of(str("A"), 0L), Pair.of(str("B"), 1L), Pair.of(null, 2L))); + GlobalIndexIOMeta second = + writeData( + Arrays.asList( + Pair.of(str("B"), 3L), + Pair.of(str("C"), 4L), + Pair.of(str("A"), 5L))); + + try (GlobalIndexReader reader = + globalIndexer.createReader( + fileReader, Arrays.asList(first, second), newDirectExecutorService())) { + assertRows(reader.visitEqual(fieldRef, str("B")).join(), 1L, 3L); + assertRows(reader.visitNotEqual(fieldRef, str("A")).join(), 1L, 3L, 4L); + assertRows( + reader.visitNotIn(fieldRef, Collections.singletonList(str("B"))).join(), + 0L, + 4L, + 5L); + assertRows(reader.visitIsNotNull(fieldRef).join(), 0L, 1L, 3L, 4L, 5L); + } + } + + @Test + public void testManifestMetaPrunesFilesBeforeLookup() throws Exception { + GlobalIndexIOMeta first = + writeData(Arrays.asList(Pair.of(str("A"), 0L), Pair.of(str("B"), 1L))); + GlobalIndexIOMeta second = + writeData(Arrays.asList(Pair.of(str("Y"), 2L), Pair.of(str("Z"), 3L))); + + CountingGlobalIndexFileReader countingFileReader = new CountingGlobalIndexFileReader(); + try (GlobalIndexReader reader = + globalIndexer.createReader( + countingFileReader, + Arrays.asList(first, second), + newDirectExecutorService())) { + assertRows(reader.visitEqual(fieldRef, str("Z")).join(), 3L); + + assertThat(countingFileReader.openCount()) + .as("manifest min/max should skip the first index file") + .isEqualTo(1); + + assertRows(reader.visitStartsWith(fieldRef, str("Z")).join(), 3L); + assertThat(countingFileReader.openCount()) + .as("manifest prefix range should also skip the first index file") + .isEqualTo(1); + + assertRows(reader.visitNotEqual(fieldRef, null).join()); + assertThat(countingFileReader.openCount()) + .as("null complement should not open extra index files") + .isEqualTo(1); + } + } + + @Test + public void testFallbackBudgetUsesSelectedFiles() throws Exception { + GlobalIndexIOMeta first = + writeData(Arrays.asList(Pair.of(str("A"), 0L), Pair.of(str("B"), 1L))); + GlobalIndexIOMeta second = + writeData(Arrays.asList(Pair.of(str("Y"), 2L), Pair.of(str("Z"), 3L))); + + Options options = new Options(); + options.set( + BitmapGlobalIndexOptions.BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE, + org.apache.paimon.options.MemorySize.ofBytes(second.fileSize())); + globalIndexer = new BitmapGlobalIndexer(dataField, options); + + try (GlobalIndexReader reader = + globalIndexer.createReader( + fileReader, Arrays.asList(first, second), newDirectExecutorService())) { + assertRows(reader.visitGreaterOrEqual(fieldRef, str("Y")).join(), 2L, 3L); + assertThat(reader.visitContains(fieldRef, str("Z")).join()).isEmpty(); + } + } + + @Test + public void testStartsWithAcrossDictionaryBlockBoundary() throws Exception { + Options options = new Options(); + options.set( + BitmapGlobalIndexOptions.BITMAP_INDEX_DICTIONARY_BLOCK_SIZE, + org.apache.paimon.options.MemorySize.ofBytes(32)); + globalIndexer = new BitmapGlobalIndexer(dataField, options); + + GlobalIndexIOMeta meta = + writeData( + Arrays.asList( + Pair.of(str("tag"), 1L), + Pair.of(str("tag-000"), 2L), + Pair.of(str("tag-001"), 3L), + Pair.of(str("tag."), 4L), + Pair.of(str("zzz"), 5L))); + + try (GlobalIndexReader reader = + globalIndexer.createReader( + fileReader, Collections.singletonList(meta), newDirectExecutorService())) { + assertRows(reader.visitStartsWith(fieldRef, str("tag-")).join(), 2L, 3L); + } + } + + @Test + public void testStartsWithReadsOnlyMatchingDictionaryBlocks() throws Exception { + Options options = new Options(); + options.set( + BitmapGlobalIndexOptions.BITMAP_INDEX_DICTIONARY_BLOCK_SIZE, + org.apache.paimon.options.MemorySize.ofBytes(32)); + globalIndexer = new BitmapGlobalIndexer(dataField, options); + + List> rows = new ArrayList<>(); + for (int i = 0; i < 50; i++) { + rows.add(Pair.of(str(String.format("other-%03d", i)), (long) i)); + } + for (int i = 0; i < 3; i++) { + rows.add(Pair.of(str(String.format("tag-match-%03d", i)), 100L + i)); + } + for (int i = 0; i < 50; i++) { + rows.add(Pair.of(str(String.format("zzz-%03d", i)), 200L + i)); + } + GlobalIndexIOMeta meta = writeData(rows); + + CountingGlobalIndexFileReader countingFileReader = new CountingGlobalIndexFileReader(); + try (GlobalIndexReader reader = + globalIndexer.createReader( + countingFileReader, + Collections.singletonList(meta), + newDirectExecutorService())) { + assertRows(reader.visitStartsWith(fieldRef, str("tag-match")).join(), 100L, 101L, 102L); + + assertThat(countingFileReader.seekCount()) + .as("footer, index, matching dictionary blocks and bitmaps") + .isLessThanOrEqualTo(9); + } + } + + @Test + public void testEqualDoesNotReadAllDictionaryBlocks() throws Exception { + Options options = new Options(); + options.set( + BitmapGlobalIndexOptions.BITMAP_INDEX_DICTIONARY_BLOCK_SIZE, + org.apache.paimon.options.MemorySize.ofBytes(32)); + globalIndexer = new BitmapGlobalIndexer(dataField, options); + + List> rows = new ArrayList<>(); + for (int i = 0; i < 100; i++) { + rows.add(Pair.of(str(String.format("tag-%03d", i)), (long) i)); + } + GlobalIndexIOMeta meta = writeData(rows); + + CountingGlobalIndexFileReader countingFileReader = new CountingGlobalIndexFileReader(); + try (GlobalIndexReader reader = + globalIndexer.createReader( + countingFileReader, + Collections.singletonList(meta), + newDirectExecutorService())) { + assertRows(reader.visitEqual(fieldRef, str("tag-050")).join(), 50L); + + assertThat(countingFileReader.seekCount()) + .as("footer, index, one dictionary block and one bitmap") + .isLessThanOrEqualTo(4); + } + } + + @Test + public void testNullChecksDoNotReadDictionary() throws Exception { + GlobalIndexIOMeta meta = + writeData( + Arrays.asList( + Pair.of(str("A"), 0L), Pair.of(str("B"), 1L), Pair.of(null, 2L))); + + CountingGlobalIndexFileReader countingFileReader = new CountingGlobalIndexFileReader(); + try (GlobalIndexReader reader = + globalIndexer.createReader( + countingFileReader, + Collections.singletonList(meta), + newDirectExecutorService())) { + assertRows(reader.visitIsNull(fieldRef).join(), 2L); + + assertThat(countingFileReader.seekCount()) + .as("footer and null bitmap") + .isLessThanOrEqualTo(2); + } + } + + private GlobalIndexIOMeta writeData(List> data) throws IOException { + GlobalIndexSingleColumnWriter writer = globalIndexer.createWriter(fileWriter); + for (Pair pair : data) { + writer.write(pair.getKey(), pair.getValue()); + } + + List results = writer.finish(); + assertThat(results).hasSize(1); + ResultEntry resultEntry = results.get(0); + Path filePath = new Path(basePath, resultEntry.fileName()); + return new GlobalIndexIOMeta(filePath, fileIO.getFileSize(filePath), resultEntry.meta()); + } + + private static BinaryString str(String value) { + return BinaryString.fromString(value); + } + + private void disableFallbackScan() { + Options options = new Options(); + options.set( + BitmapGlobalIndexOptions.BITMAP_INDEX_FALLBACK_SCAN_MAX_SIZE, + org.apache.paimon.options.MemorySize.ofBytes(0)); + globalIndexer = new BitmapGlobalIndexer(dataField, options); + } + + private static void assertRows(Optional result, Long... expected) { + assertThat(result).isPresent(); + + Iterator iterator = result.get().results().iterator(); + List actual = new ArrayList<>(); + while (iterator.hasNext()) { + actual.add(iterator.next()); + } + assertThat(actual).containsExactlyInAnyOrder(expected); + } + + private class CountingGlobalIndexFileReader implements GlobalIndexFileReader { + + private final AtomicInteger seekCount = new AtomicInteger(); + private final AtomicInteger openCount = new AtomicInteger(); + + @Override + public SeekableInputStream getInputStream(GlobalIndexIOMeta meta) throws IOException { + openCount.incrementAndGet(); + return new CountingSeekableInputStream(fileReader.getInputStream(meta), seekCount); + } + + int seekCount() { + return seekCount.get(); + } + + int openCount() { + return openCount.get(); + } + } + + private static class CountingSeekableInputStream extends SeekableInputStreamWrapper { + + private final AtomicInteger seekCount; + + private CountingSeekableInputStream(SeekableInputStream wrapped, AtomicInteger seekCount) { + super(wrapped); + this.seekCount = seekCount; + } + + @Override + public void seek(long desired) throws IOException { + if (desired != getPos()) { + seekCount.incrementAndGet(); + } + super.seek(desired); + } + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java index b31cc248e06a..b2d0fa2d7f3a 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java @@ -29,6 +29,7 @@ import org.apache.paimon.globalindex.GlobalIndexReader; import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.KeySerializer; import org.apache.paimon.globalindex.ResultEntry; import org.apache.paimon.globalindex.io.GlobalIndexFileReader; import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java index 55d08432a6f4..b0ad16e98d08 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/BTreeThreadSafetyTest.java @@ -26,6 +26,7 @@ import org.apache.paimon.globalindex.GlobalIndexReader; import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.KeySerializer; import org.apache.paimon.globalindex.ResultEntry; import org.apache.paimon.globalindex.io.GlobalIndexFileReader; import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java index cad9861f4060..760c5daa9f18 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java @@ -84,6 +84,65 @@ private List writeData() throws Exception { return written; } + private int firstStrictlyGreaterKeyIndex() { + for (int i = 1; i < dataNum; i++) { + if (comparator.compare(data.get(i - 1).getKey(), data.get(i).getKey()) < 0) { + return i; + } + } + return -1; + } + + @TestTemplate + public void testFallbackScanDisabledByBudget() throws Exception { + options.set(BTreeIndexOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE, MemorySize.ofBytes(1)); + globalIndexer = new BTreeGlobalIndexer(new DataField(1, "testField", dataType), options); + + List written = writeData(); + FieldRef ref = new FieldRef(1, "testField", dataType); + Object literal = data.get(dataNum / 2).getKey(); + Object min = data.get(0).getKey(); + Object max = data.get(dataNum - 1).getKey(); + + try (GlobalIndexReader reader = + globalIndexer.createReader(fileReader, written, newDirectExecutorService())) { + assertThat(reader.visitBetween(ref, min, max).join()).isEmpty(); + + GlobalIndexResult result = reader.visitEqual(ref, literal).join().get(); + assertResult(result, filter(obj -> comparator.compare(obj, literal) == 0)); + } + } + + @TestTemplate + public void testFallbackBudgetUsesSelectedFiles() throws Exception { + int split = firstStrictlyGreaterKeyIndex(); + if (split <= 0 || split >= dataNum) { + return; + } + + List written = new ArrayList<>(2); + written.add(writeData(data.subList(0, split))); + written.add(writeData(data.subList(split, dataNum))); + + options.set( + BTreeIndexOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE, + MemorySize.ofBytes(written.get(1).fileSize())); + globalIndexer = new BTreeGlobalIndexer(new DataField(1, "testField", dataType), options); + + FieldRef ref = new FieldRef(1, "testField", dataType); + Object min = data.get(0).getKey(); + Object max = data.get(dataNum - 1).getKey(); + Object secondFileMin = data.get(split).getKey(); + + try (GlobalIndexReader reader = + globalIndexer.createReader(fileReader, written, newDirectExecutorService())) { + assertThat(reader.visitBetween(ref, min, max).join()).isEmpty(); + + GlobalIndexResult result = reader.visitGreaterOrEqual(ref, secondFileMin).join().get(); + assertResult(result, filter(obj -> comparator.compare(obj, secondFileMin) >= 0)); + } + } + @TestTemplate public void testUnorderedIterator() throws Exception { // Set some null values diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilderTest.java index 0c55221a50f7..5e8c5cb82e65 100644 --- a/paimon-core/src/test/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilderTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexBuilderTest.java @@ -24,6 +24,7 @@ import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.BlobData; import org.apache.paimon.data.GenericRow; +import org.apache.paimon.globalindex.KeySerializer; import org.apache.paimon.index.GlobalIndexMeta; import org.apache.paimon.index.IndexFileHandler; import org.apache.paimon.index.IndexFileMeta; @@ -445,15 +446,17 @@ private static class FileStats { this.lastKey = lastKey; } - static FileStats fromIndexFileMeta(IndexFileMeta meta) { + static FileStats fromIndexFileMeta(org.apache.paimon.index.IndexFileMeta meta) { Assertions.assertNotNull(meta.globalIndexMeta()); GlobalIndexMeta globalIndexMeta = meta.globalIndexMeta(); - BTreeIndexMeta btreeMeta = BTreeIndexMeta.deserialize(globalIndexMeta.indexMeta()); + org.apache.paimon.globalindex.SortedIndexFileMeta indexMeta = + org.apache.paimon.globalindex.SortedIndexFileMeta.deserialize( + globalIndexMeta.indexMeta()); return new FileStats( meta.rowCount(), - deserialize(btreeMeta.getFirstKey()), - deserialize(btreeMeta.getLastKey())); + deserialize(indexMeta.getFirstKey()), + deserialize(indexMeta.getLastKey())); } } } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/BitmapGlobalIndexTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/BitmapGlobalIndexTableTest.java new file mode 100644 index 000000000000..d7318e0eb5f7 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/BitmapGlobalIndexTableTest.java @@ -0,0 +1,284 @@ +/* + * 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.table; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.globalindex.GlobalIndexBuilderUtils; +import org.apache.paimon.globalindex.GlobalIndexScanner; +import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.GlobalIndexWriter; +import org.apache.paimon.globalindex.IndexedSplit; +import org.apache.paimon.globalindex.ResultEntry; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.CloseableIterator; +import org.apache.paimon.utils.Range; +import org.apache.paimon.utils.RoaringNavigableMap64; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Test for bitmap indexed batch scan. */ +public class BitmapGlobalIndexTableTest extends DataEvolutionTestBase { + + private static final String INDEX_TYPE = "bitmap"; + + @Test + public void testBitmapGlobalIndexWithCoreScanAcrossRanges() throws Exception { + write(1000L); + createIndex("f1", Collections.singletonList(new Range(0L, 499L))); + createIndex("f1", Collections.singletonList(new Range(500L, 999L))); + + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + Predicate predicate = + new PredicateBuilder(table.rowType()) + .in( + 1, + Arrays.asList( + BinaryString.fromString("a200"), + BinaryString.fromString("a700"))); + + RoaringNavigableMap64 rowIds = globalIndexScan(table, predicate); + assertThat(rowIds.toRangeList()) + .containsExactlyInAnyOrder(new Range(200L, 200L), new Range(700L, 700L)); + + ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate); + List readF1 = new ArrayList<>(); + readBuilder + .newRead() + .createReader(readBuilder.newScan().plan()) + .forEachRemaining(row -> readF1.add(row.getString(1).toString())); + + assertThat(readF1).containsExactly("a200", "a700"); + } + + @Test + public void testBitmapGlobalIndexComplementsAndNulls() throws Exception { + long oldRowCount = 10L; + write(oldRowCount); + + catalog.alterTable(identifier(), SchemaChange.addColumn("f3", DataTypes.STRING()), false); + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite()) { + write.write( + GenericRow.of( + 100, + BinaryString.fromString("a-new"), + BinaryString.fromString("b-new"), + BinaryString.fromString("not-null"))); + try (BatchTableCommit commit = writeBuilder.newCommit()) { + commit.commit(write.prepareCommit()); + } + } + + createIndex("f3", null); + + table = (FileStoreTable) catalog.getTable(identifier()); + Predicate isNull = new PredicateBuilder(table.rowType()).isNull(3); + RoaringNavigableMap64 rowIds = globalIndexScan(table, isNull); + assertThat(rowIds.getLongCardinality()).isEqualTo(oldRowCount); + assertThat(rowIds.toRangeList()).containsExactly(new Range(0L, oldRowCount - 1)); + + Predicate isNotNull = new PredicateBuilder(table.rowType()).isNotNull(3); + rowIds = globalIndexScan(table, isNotNull); + assertThat(rowIds.toRangeList()).containsExactly(new Range(oldRowCount, oldRowCount)); + + Predicate notEqual = + new PredicateBuilder(table.rowType()) + .notEqual(3, BinaryString.fromString("not-null")); + rowIds = globalIndexScan(table, notEqual); + assertThat(rowIds.isEmpty()).isTrue(); + } + + @Test + public void testBitmapGlobalIndexStartsWith() throws Exception { + write(1000L); + createIndex("f1", null); + + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + Predicate predicate = + new PredicateBuilder(table.rowType()).startsWith(1, BinaryString.fromString("a20")); + + RoaringNavigableMap64 rowIds = globalIndexScan(table, predicate); + assertThat(rowIds.toRangeList()) + .containsExactlyInAnyOrder(new Range(20L, 20L), new Range(200L, 209L)); + } + + @Test + public void testBitmapGlobalIndexFallbackScan() throws Exception { + write(1000L); + createIndex("f1", null); + + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + Predicate contains = + new PredicateBuilder(table.rowType()).contains(1, BinaryString.fromString("20")); + + RoaringNavigableMap64 rowIds = globalIndexScan(table, contains); + assertThat(rowIds.toRangeList()) + .containsExactlyInAnyOrder( + new Range(20L, 20L), + new Range(120L, 120L), + new Range(200L, 209L), + new Range(220L, 220L), + new Range(320L, 320L), + new Range(420L, 420L), + new Range(520L, 520L), + new Range(620L, 620L), + new Range(720L, 720L), + new Range(820L, 820L), + new Range(920L, 920L)); + + Predicate range = + new PredicateBuilder(table.rowType()) + .between( + 1, + BinaryString.fromString("a200"), + BinaryString.fromString("a209")); + rowIds = globalIndexScan(table, range); + assertThat(rowIds.toRangeList()).containsExactly(new Range(200L, 209L)); + } + + private void createIndex(String fieldName, List rowRanges) throws Exception { + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + DataField indexField = table.rowType().getField(fieldName); + RowType readRowType = + SpecialFields.rowTypeWithRowId( + table.rowType().project(Collections.singletonList(fieldName))); + ReadBuilder readBuilder = table.newReadBuilder().withReadType(readRowType); + + List splits = + rowRanges == null + ? readBuilder.newScan().plan().splits() + : readBuilder.withRowRanges(rowRanges).newScan().plan().splits(); + + List commitMessages = new ArrayList<>(); + for (Split split : splits) { + commitMessages.add(buildIndex(table, indexField, readRowType, split)); + } + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit(commitMessages); + } + } + + private CommitMessage buildIndex( + FileStoreTable table, DataField indexField, RowType readRowType, Split split) + throws Exception { + Range rowRange = rowRange(split); + GlobalIndexWriter indexWriter = + GlobalIndexBuilderUtils.createIndexWriter( + table, INDEX_TYPE, indexField, table.coreOptions().toConfiguration()); + GlobalIndexSingleColumnWriter writer = (GlobalIndexSingleColumnWriter) indexWriter; + InternalRow.FieldGetter fieldGetter = + InternalRow.createFieldGetter( + indexField.type(), readRowType.getFieldIndex(indexField.name())); + int rowIdIndex = readRowType.getFieldIndex(SpecialFields.ROW_ID.name()); + + try (RecordReader reader = + table.newReadBuilder() + .withReadType(readRowType) + .newRead() + .createReader(Collections.singletonList(split)); + CloseableIterator iterator = reader.toCloseableIterator()) { + while (iterator.hasNext()) { + InternalRow row = iterator.next(); + long rowId = row.getLong(rowIdIndex); + if (rowId >= rowRange.from && rowId <= rowRange.to) { + writer.write(fieldGetter.getFieldOrNull(row), rowId - rowRange.from); + } + } + } + + List resultEntries = writer.finish(); + List indexFileMetas = + GlobalIndexBuilderUtils.toIndexFileMetas( + table.fileIO(), + table.store().pathFactory().globalIndexFileFactory(), + table.coreOptions(), + rowRange, + indexField.id(), + INDEX_TYPE, + resultEntries); + DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFileMetas); + return new CommitMessageImpl( + partition(split), 0, null, dataIncrement, CompactIncrement.emptyIncrement()); + } + + private BinaryRow partition(Split split) { + return dataSplit(split).partition(); + } + + private Range rowRange(Split split) { + if (split instanceof IndexedSplit) { + List rowRanges = ((IndexedSplit) split).rowRanges(); + assertThat(rowRanges).hasSize(1); + return rowRanges.get(0); + } + + List ranges = + dataSplit(split).dataFiles().stream() + .map(DataFileMeta::nonNullRowIdRange) + .sorted(Comparator.comparingLong(range -> range.from)) + .collect(Collectors.toList()); + return new Range(ranges.get(0).from, ranges.get(ranges.size() - 1).to); + } + + private DataSplit dataSplit(Split split) { + return split instanceof IndexedSplit + ? ((IndexedSplit) split).dataSplit() + : (DataSplit) split; + } + + private RoaringNavigableMap64 globalIndexScan(FileStoreTable table, Predicate predicate) + throws Exception { + try (GlobalIndexScanner scanner = + GlobalIndexScanner.create(table, PartitionPredicate.ALWAYS_TRUE, predicate).get()) { + return scanner.scan(predicate).get().results(); + } + } +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala index cba3c3cd63f3..9a28029f1c01 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CreateGlobalIndexProcedureTest.scala @@ -18,7 +18,7 @@ package org.apache.paimon.spark.procedure -import org.apache.paimon.globalindex.btree.{BTreeIndexMeta, KeySerializer} +import org.apache.paimon.globalindex.{KeySerializer, SortedIndexFileMeta} import org.apache.paimon.memory.MemorySlice import org.apache.paimon.spark.PaimonSparkTestBase import org.apache.paimon.types.VarCharType @@ -80,7 +80,7 @@ class CreateGlobalIndexProcedureTest extends PaimonSparkTestBase with StreamTest btreeEntries.foreach(e => assert(e.globalIndexMeta() != null)) // 3. assert btree index file range non-overlapping - case class MetaWithKey(meta: BTreeIndexMeta, first: Object, last: Object) + case class MetaWithKey(meta: SortedIndexFileMeta, first: Object, last: Object) val keySerializer = KeySerializer.create(new VarCharType()) val comparator = keySerializer.createComparator() @@ -90,7 +90,7 @@ class CreateGlobalIndexProcedureTest extends PaimonSparkTestBase with StreamTest val btreeMetas = btreeEntries .map(_.globalIndexMeta().indexMeta()) - .map(meta => BTreeIndexMeta.deserialize(meta)) + .map(meta => SortedIndexFileMeta.deserialize(meta)) .map( m => { assert(m.getFirstKey != null) @@ -216,7 +216,7 @@ class CreateGlobalIndexProcedureTest extends PaimonSparkTestBase with StreamTest val entriesByPart = btreeEntries.groupBy(_.partition()) assert(entriesByPart.size == partCount) - case class MetaWithKey(meta: BTreeIndexMeta, first: Object, last: Object) + case class MetaWithKey(meta: SortedIndexFileMeta, first: Object, last: Object) val keySerializer = KeySerializer.create(new VarCharType()) val comparator = keySerializer.createComparator() @@ -227,7 +227,7 @@ class CreateGlobalIndexProcedureTest extends PaimonSparkTestBase with StreamTest for ((k, v) <- entriesByPart) { val metas = v .map(_.indexFile().globalIndexMeta().indexMeta()) - .map(bytes => BTreeIndexMeta.deserialize(bytes)) + .map(bytes => SortedIndexFileMeta.deserialize(bytes)) .map( m => { assert(m.getFirstKey != null)