[fix][meta] Find the records written with a partition key in Oxia secondary-index scans

[merlimat]

Motivation

With an Oxia metadata store whose namespace has more than one shard, scanByIndex drops the records written with Option.PartitionKey. Every PIP-473 transaction record has a partition key (its transaction or its segment), and TxnMetadataStore scans them through secondary indexes without one. So on a sharded Oxia, these all work from incomplete lists:

• transaction buffer and pending-ack recovery;
• the coordinator's end-transaction fan-out, and its timeout and GC sweeps;
• the segment and subscription cleanups.

The cause is in OxiaMetadataStore.storeScanByIndex:

• It lists the matching keys with client.list(from, to, UseIndex(...)), which reads every shard.
• It then reads each record with storeGet(key, opts). Without a partition key in opts, the Oxia client sends that get to the shard of the record's key. A record written with a partition key lives on the shard of its partition key.
• So each record whose key and partition key hash to different shards is listed, not found, and skipped.

MetadataStoreSecondaryIndexTest runs on a single-shard Oxia without partition keys, and OxiaPartitionKeyTest (3 shards) only reads records back with the partition key they were written with, so neither caught this.

Example

On a 3-shard Oxia (oxia/oxia:nightly-beta), on master:

for (int i = 0; i < 30; i++) {
    store.put(prefix + "/r-" + i, value, Optional.of(-1L),
            Set.of(new Option.PartitionKey("group-" + i), new Option.SecondaryIndex("by-tag", "tag"))).get();
}
store.scanByIndex(prefix, "by-tag", "tag", "tag", r -> true, ScanConsumer.collectInto(results)).get();
// results: 8 to 13 of the 30 records, depending on the run

client.list(scopedFrom, scopedTo, listOpts)                       // every shard
        .thenCompose(primaryKeys -> {
            ...
            chain = chain.thenCompose(__ -> storeGet(key, opts))  // the shard of `key`
                    .thenAccept(opt -> opt.ifPresent(consumer::onNext));

With the Oxia CLI on the same data, the index list returns all 30 keys. A get without the partition key finds 8 records; with it, all 30.

Through TxnMetadataStore, with 30 transactions that each wrote and acked on one segment (15 open, 15 committed):

Query	Found
listWritesBySegment	12 of 30
listAcksBySegmentSubscription	8 of 30
listOpenByDeadlineRange	3 of 15
listFinalizedByStateAndTimeRange	4 of 15

scanAbortedTxns found 9 of 30, and in one run listOpsByTxn returned none of the transaction's ops.

Modifications

• OxiaMetadataStore.storeScanByIndex reads the records in the index scan with client.rangeScan(from, to, consumer, {UseIndex(indexName), [PartitionKey]}). Each shard returns the matching records it holds, with their key, value and version. The bounds are unchanged. The server has returned the primary key with index range-scan results since secondary indexes were added (Secondary indexes implementation oxia-db/oxia#544).
• It skips the entries of records deleted after the scan read their index entry:
  • The server walks the index from a snapshot but reads each record from the latest state, so it returns such a record as not found, with no version or value.
  • The Java client turns that into a GetResult with an empty value and a zero Version. In a test that deleted indexed records while scanning them, 1,529 of the 16,753 entries the client returned were like this.
  • Before this change, the per-key get skipped them by returning nothing.
• A consumer that throws stops the scan (onNext returns false, and the client then calls onCompleted), and the scan fails with that exception, as before.
• TxnMetadataStore.listOpsByTxn and scanAbortedTxns pass the partition key their records share (the transaction, the segment), so these scans read one shard instead of all of them.
• Across shards, records now come in shard order instead of sorted by key. scanByIndex doesn't promise an order, and no caller relies on one.

Verifying this change

• OxiaPartitionKeyTest (3 shards): a scan across 30 partition keys returns every record (8 to 13 of 30 before the change), and a scan scoped to one partition key returns that partition's records.
• OxiaMetadataStoreScanByIndexTest (mocked Oxia client): the scan passes the index and the partition key to rangeScan, skips a not-found entry, and fails with a consumer's exception.
• TxnMetadataStoreShardedOxiaTest (3 shards): each of the six index queries returns the records of all 30 transactions. Without the store change, the four cross-partition queries fail and the two partition-keyed ones pass. Without either change, all six fail.
• MetadataStoreSecondaryIndexTest (60 tests, every provider), MetadataCacheSecondaryIndexTest, TxnMetadataStoreTest, MetadataTransactionBufferTest, MetadataPendingAckStoreTest, TransactionCoordinatorV5Test, ScalableTopicPropertyIndexTest and V5Transaction* pass. quickCheck passes.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

[codelipenghui]

The native indexed range scan looks like the right fix for missing partition-keyed records: returning values from the shards holding them avoids re-routing each primary key without its original partition key.

One compatibility question about the ordering statement in the description, based on head 1cc87bcfcb9d65746890caf8046a1ac1db81b7e2:

• ScanConsumer's Javadoc explicitly covers both scanChildren and scanByIndex and says: “The store invokes … onNext for each record (in key order).”
• In Oxia client 0.9.5, list without a partition key uses internalListMultiShards, which combines the returned primary keys and calls list.sort(CompareWithSlash.INSTANCE). The previous Pulsar implementation chained its reads/callbacks sequentially, preserving that order among the records it actually returned.
• The native range scan's multi-shard consumer forwards serialized arrivals without globally sorting them, so this changes observable ordering on that path.

This is narrower than saying every old scan guaranteed primary-key order: an explicit partition key takes the single-shard list path without that client-side sort, and the Javadoc does not distinguish primary-key order from secondary-index order.

Could we clarify the intended ordering contract and reconcile the Javadoc with the implementation? If unordered indexed scans are intended, please make that explicit and address compatibility with existing consumers; if ordering is required, it needs to be preserved. I am not suggesting unconditional buffering/sorting of all returned values without considering the memory and streaming-latency cost.

This is based on source inspection only; no tests were run, and no consumer regression has been reproduced.

[codelipenghui]

Reviewed head 358e23227196f5edebf798f200bf26fab71c231f.

The native indexed scan fixes the partition-key routing problem, and the updated ordering documentation plus caller-side topic-list sorting addresses my earlier concern. I am requesting changes for one introduced timeout/liveness regression, detailed inline.

Validation: all 158 focused tests passed (83 metadata and 75 broker tests). An isolated A/B probe using the actual pre-PR adapter reproduced the original omission (8/30 records before, 30/30 after). However, after stopping the initialized three-shard Oxia server with a configured one-second request timeout, the pre-PR adapter failed after 1,007 ms with one error callback, while this head remained pending after 5,006 ms with no terminal callback.

Please preserve a bounded retry deadline for index scans and add regression coverage for persistent retryable shard failures. No additional actionable findings were identified.

[codelipenghui]

Reviewed the follow-up at b1ee6cbba0dd333fbea9df175df6afb7915f8f10. The previous timeout finding is resolved: index scans now use the configured deadline, select exactly one terminal callback, and suppress late callbacks.

All 160 focused tests passed (85 metadata and 75 broker tests). An independent probe through the production factory/config-file path verified that a one-second timeout fails the outage scan after 1,002 ms with exactly one error callback; the healthy three-shard scan still returns all 30 records.

No actionable findings identified within the reviewed scope. Approving the code changes; required CI checks should pass before merging.