You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/14 00:58:21 UTC
(pulsar) branch master updated: [fix][meta] Fix the upper bound of the Oxia secondary-index scans (#26890)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 1b990e2983a4 [fix][meta] Fix the upper bound of the Oxia secondary-index scans (#26890)
1b990e2983a4 is described below
commit 1b990e2983a4c5f2ed32cad3c6c03bba089ee0d9
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Tue Oct 13 17:58:12 2026 -0700
[fix][meta] Fix the upper bound of the Oxia secondary-index scans (#26890)
---
.../metadata/impl/oxia/OxiaMetadataStore.java | 20 +++---
.../metadata/MetadataStoreSecondaryIndexTest.java | 84 ++++++++++++++++++++++
2 files changed, 96 insertions(+), 8 deletions(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java
index d84c72407aba..0a719e8bd9ae 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/oxia/OxiaMetadataStore.java
@@ -256,18 +256,22 @@ public class OxiaMetadataStore extends AbstractMetadataStore {
ScanConsumer consumer, Set<Option> opts) {
// Index entries are stored at "<parentPath>/<secondaryKey>" by doStorePut, where
// parentPath == scanPathPrefix for records directly under the prefix. The API uses
- // inclusive-on-both-sides bounds; Oxia's list(start, end, ...) is half-open, so we
- // append a sentinel character to the upper bound to widen the half-open range to
- // include `toKeyInclusive`. We use '~' (0x7E) — it lex-sorts after every printable
- // ASCII byte, which is a safe choice for the secondary keys callers actually use today
- // (numeric-encoded timestamps, fixed-tag enums, etc.). Callers that need full byte-range
- // coverage can pass `toKeyInclusive=null` or use a different scheme.
+ // inclusive-on-both-sides bounds; Oxia's list(start, end, ...) is half-open, and it compares
+ // the bounds with whole index entries, which Oxia stores as the secondary key, a 0x01
+ // separator and the escaped primary key. So the end that follows the entries of
+ // `toKeyInclusive`, and precedes those of the keys that extend it (e.g. "k-12" for "k-1"),
+ // is `toKeyInclusive` followed by 0x02. Without `toKeyInclusive`, the range ends at the end
+ // of the level of the records, as the range [path + "/", path + "//") of
+ // getChildrenFromStore does with Oxia's hierarchical key sorting.
+ // The ranges follow that sorting, which orders the secondary keys by their number of '/'
+ // first: a range without `toKeyInclusive` leaves out the keys with a '/' (e.g.
+ // "tenant/ns1"), and the range ["b", "tenant/ns1"] also returns "zz".
String scopedFrom = fromKeyInclusive == null
? scanPathPrefix + "/"
: scanPathPrefix + "/" + fromKeyInclusive;
String scopedTo = toKeyInclusive == null
- ? scanPathPrefix + "0" // '0' (0x30) is the lex successor of '/' (0x2F)
- : scanPathPrefix + "/" + toKeyInclusive + "~";
+ ? scanPathPrefix + "//"
+ : scanPathPrefix + "/" + toKeyInclusive + "\u0002";
Set<ListOption> listOpts = new HashSet<>(listOptions(opts));
listOpts.add(ListOption.UseIndex(indexName));
CompletableFuture<Void> done = new CompletableFuture<>();
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreSecondaryIndexTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreSecondaryIndexTest.java
index b842b5c45d28..bd9e21161cb2 100644
--- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreSecondaryIndexTest.java
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreSecondaryIndexTest.java
@@ -27,6 +27,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import lombok.Cleanup;
@@ -244,4 +245,87 @@ public class MetadataStoreSecondaryIndexTest extends BaseMetadataStoreTest {
.collect(Collectors.toSet());
assertEquals(values, Set.of("v-100", "v-200"));
}
+
+ @Test(dataProvider = "impl")
+ public void scanByIndexUnboundedToKey(String provider, Supplier<String> urlSupplier) throws Exception {
+ @Cleanup
+ MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
+ MetadataStoreConfig.builder().build());
+
+ String basePath = newKey();
+ for (long t : new long[]{100L, 200L, 300L}) {
+ String key = String.format("%020d", t);
+ store.put(basePath + "/r-" + t, ("v-" + t).getBytes(StandardCharsets.UTF_8),
+ Optional.of(-1L), EnumSet.noneOf(CreateOption.class),
+ Map.of("by-time", key)).join();
+ }
+
+ Predicate<GetResult> from200 = r -> {
+ String v = new String(r.getValue(), StandardCharsets.UTF_8);
+ return Long.parseLong(v.substring(2)) >= 200L;
+ };
+
+ List<GetResult> results = new ArrayList<>();
+ store.scanByIndex(basePath, "by-time", String.format("%020d", 200L), null,
+ from200, ScanConsumer.collectInto(results)).join();
+ assertEquals(values(results), Set.of("v-200", "v-300"));
+
+ results.clear();
+ store.scanByIndex(basePath, "by-time", null, null, r -> true, ScanConsumer.collectInto(results)).join();
+ assertEquals(values(results), Set.of("v-100", "v-200", "v-300"));
+ }
+
+ @Test(dataProvider = "impl")
+ public void scanByIndexPointLookupSkipsKeysExtendingTheKey(String provider, Supplier<String> urlSupplier)
+ throws Exception {
+ @Cleanup
+ MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
+ MetadataStoreConfig.builder().build());
+
+ String basePath = newKey();
+ // "broker-1" is a prefix of the other owners.
+ List<String> owners = List.of("broker-1", "broker-12", "broker-1:a", "broker-1_b", "broker-1~c");
+ for (int i = 0; i < owners.size(); i++) {
+ store.put(basePath + "/r-" + i, owners.get(i).getBytes(StandardCharsets.UTF_8),
+ Optional.of(-1L), EnumSet.noneOf(CreateOption.class),
+ Map.of("by-owner", owners.get(i))).join();
+ }
+
+ List<GetResult> results = findByIndex(store, basePath, "by-owner", "broker-1",
+ r -> new String(r.getValue(), StandardCharsets.UTF_8).equals("broker-1"));
+
+ assertEquals(values(results), Set.of("broker-1"));
+ }
+
+ @Test(dataProvider = "impl")
+ public void scanByIndexInclusiveRangeSkipsKeysExtendingTheUpperBound(String provider,
+ Supplier<String> urlSupplier)
+ throws Exception {
+ @Cleanup
+ MetadataStoreExtended store = MetadataStoreExtended.create(urlSupplier.get(),
+ MetadataStoreConfig.builder().build());
+
+ String basePath = newKey();
+ for (String key : List.of("k-0", "k-1", "k-12", "k-2")) {
+ store.put(basePath + "/r-" + key, key.getBytes(StandardCharsets.UTF_8),
+ Optional.of(-1L), EnumSet.noneOf(CreateOption.class),
+ Map.of("by-key", key)).join();
+ }
+
+ Predicate<GetResult> inRange = r -> {
+ String v = new String(r.getValue(), StandardCharsets.UTF_8);
+ return v.compareTo("k-0") >= 0 && v.compareTo("k-1") <= 0;
+ };
+
+ List<GetResult> results = new ArrayList<>();
+ store.scanByIndex(basePath, "by-key", "k-0", "k-1", inRange, ScanConsumer.collectInto(results)).join();
+
+ assertEquals(values(results), Set.of("k-0", "k-1"));
+ }
+
+ private static Set<String> values(List<GetResult> results) {
+ return results.stream()
+ .map(r -> new String(r.getValue(), StandardCharsets.UTF_8))
+ .collect(Collectors.toSet());
+ }
}