Repository navigation
[kv] Support global secondary indexes - #4820
platinumhamburg wants to merge 2 commits into
Conversation
1ad09b1 to
89b5574
Compare
|
@platinumhamburg can you rebase and fix the CI? I you want i can also help do that. I want to go through the code and also build and test this work in a test environment, verify everything looks good and measure the impact of indexes on disk 👍 If you have some numbers yourself it would be great if you can share 🙏 |
89b5574 to
795f152
Compare
Hi @polyzos , Rebase is done. Welcome to test it and review the code together! I’ve done some internal canary validation of the end-to-end secondary index functionality, but don’t have large-scale benchmark results to share yet. Architecturally, an index table is a regular primary key table, so it uses the same storage and replication mechanisms. We expect similar performance characteristics, with additional index-table writes for maintenance and an extra lookup step for reads. Index lookups rely heavily on |
98b2816 to
2960f49
Compare
| snapshotSize = snapshotResult.getSnapshotSize(); | ||
| // update LogTablet to notify the lowest offset that should be retained | ||
| updateMinRetainOffset.accept(flushedLogOffset); | ||
| updateMinRetainOffset.accept(completedSnapshot.getMinRetainLogOffset()); |
There was a problem hiding this comment.
The leader retains WAL from getMinRetainLogOffset() here, but followers don't. CoordinatorEventProcessor still notifies followers with the data offset:
coordinatorEventManager.put(
new NotifyKvSnapshotOffsetEvent(
tb, completedSnapshot.getLogOffset()));So followers can drop [indexPushedOffset, logOffset) once it's in remote. Since LogTablet#updateMinRetainOffset is a monotonic max, a follower promoted to leader can't lower it back either. This should be completedSnapshot.getMinRetainLogOffset().
| } | ||
|
|
||
| logTablet.updateMinRetainOffset(restoreStartOffset); | ||
| logTablet.updateMinRetainOffset( |
There was a problem hiding this comment.
Retention is gated by the index offset locally, but remote expiry isn't. RemoteLogTablet#expiredRemoteLogSegments only checks TTL and lake offset:
if (currentTimeMs - ts > ttlMs) {
...
expiredSegments.add(segment);Scenario: an async index lags past table.log.ttl (7d default), then failover. The new leader has no local WAL (see the comment on KvTabletSnapshotTarget) and no remote WAL, so it hits a WAL gap and goes FAILED permanently, with no rebuild path. Remote expiry should be bounded by the snapshot's indexPushedOffset too.
| if (indexReplicator != failed || state.get() != State.RUNNING) { | ||
| return; | ||
| } | ||
| state.set(State.FAILED); |
There was a problem hiding this comment.
This is the whole fail-stop: set a flag and log. The replica stays leader and keeps accepting writes. Async indexes go silently stale, SYNC writes time out forever, and allIndexPushedOffset pins the WAL indefinitely. After a restart, replication resumes from the same snapshot offset and fails again.
What's the intended terminal behaviour (resign leadership / reject writes / ...)? Can you implement it and assert the end state in a test. IndexReplicationWalGapFailStopTest only checks that polling stops.
| } | ||
|
|
||
| /** Marks only the expected failed replica offline without taking the lifecycle lock. */ | ||
| public void markReplicaOffline(TableBucket tableBucket, Replica expectedReplica) { |
There was a problem hiding this comment.
No callers in main or test code. Was this meant to be wired into the FAILED path in IndexReplicationSupervisor#onIndexReplicatorFailed? If not, please remove it.
|
|
||
| @Override | ||
| public CompletableFuture<PutIndexResponse> putIndex(PutIndexRequest request) { | ||
| authorizeTable(WRITE, request.getTableId()); |
There was a problem hiding this comment.
PRIVATE on ApiKeys.PUT_INDEX is only a label; nothing enforces it. Any external client with WRITE on the index table can call this (a DB-level grant is enough), and anyone can when no authorizer is configured. A PROGRESS row with sourceEndOffset = Long.MAX_VALUE makes every future window a "duplicate", so the index silently freezes forever. Please reject unless currentSession().isInternal().
|
|
||
| ## Operations and Limits | ||
|
|
||
| Index maintenance consumes committed main-table WAL. Snapshot metadata preserves index replay progress, and WAL retention covers the data required to resume replication after recovery. A lagging asynchronous index can therefore retain main-table WAL even though it does not delay write acknowledgement. |
There was a problem hiding this comment.
This isn't true yet: follower retention and remote TTL don't honour index progress (see the comments on KvTabletSnapshotTarget / Replica#initKvTablet). Please update once that's fixed.
| repeated PbTableMetadata table_metadata = 3; | ||
| repeated PbPartitionMetadata partition_metadata = 4; | ||
| optional int32 coordinator_epoch = 5; | ||
| // Per-main-table partition tombstones propagated from the Coordinator (P3T6). |
There was a problem hiding this comment.
nit: please drop the internal planning tag (P3T6).
| class IndexReplicatorTest { | ||
|
|
||
| @Test | ||
| void sharesSourceDecodeAcrossIndexesWithoutSplittingSourceBatch() throws Exception { |
There was a problem hiding this comment.
nit: the project guide says no new Mockito usage. The same applies to IndexReplicationWalGapFailStopTest, CompactedKeyEncoderTest and TabletServerMetadataCacheTest. Please use real MemoryLogRecords / LogRecordBatch or a small Testing* fake.
| } | ||
| validateBatch(sourceBucket, sourceEndOffset, progressKey, records); | ||
|
|
||
| long durableProgress = |
There was a problem hiding this comment.
Nice: deciding duplicates from committed RocksDB state, and failing retriably when progress is visible but not yet flushed, means a retry can never ack an uncommitted window. Could you add a short comment here explaining why both checks are needed?
| LookupResult hop1Result, Object[] expectedIdxValues) { | ||
| List<InternalRow> candidateIndexRows = hop1Result.getRowList(); | ||
| if (candidateIndexRows.size() >= LOW_SELECTIVITY_WARN_CANDIDATE_THRESHOLD) { | ||
| LOG.warn( |
There was a problem hiding this comment.
This only warns. Hop 1 returns all candidates in one response and hop 2 fans out one point lookup per candidate. Can we add a configurable hard limit that fails fast? Low-selectivity indexes, or accumulated stale entries (see the TTL comment), will otherwise impact both the server and the client.
There was a problem hiding this comment.
A limit would really help here. In my test one lookup over 300k candidates takes about a second and around 300 MB of client heap.
In a delta join, renaming a customer with 1M orders took about 19 s and pushed the TaskManager past its memory limit. A configurable cap that fails fast would help, plus paging and a bound on in-flight primary-key lookups, so large keys stream instead of loading everything at once.
|
@platinumhamburg Thanks for all the great work here 🙏, the overall design looks solid. A few blockers before this can go in:
The rest are details in the inline comments. After these are addressed I will build and test internally 🚀 Let me know if you want any help |
fresh-borzoni
left a comment
fresh-borzoni
left a comment
There was a problem hiding this comment.
@platinumhamburg Nice work overall, left a couple of comments, PTAL
| LookupResult hop1Result, Object[] expectedIdxValues) { | ||
| List<InternalRow> candidateIndexRows = hop1Result.getRowList(); | ||
| if (candidateIndexRows.size() >= LOW_SELECTIVITY_WARN_CANDIDATE_THRESHOLD) { | ||
| LOG.warn( |
There was a problem hiding this comment.
A limit would really help here. In my test one lookup over 300k candidates takes about a second and around 300 MB of client heap.
In a delta join, renaming a customer with 1M orders took about 19 s and pushed the TaskManager past its memory limit. A configurable cap that fails fast would help, plus paging and a bound on in-flight primary-key lookups, so large keys stream instead of loading everything at once.
| oldWatermark.get(), | ||
| newHighWatermark, | ||
| tableBucket); | ||
| indexManager.onHighWatermarkAdvanced(); |
There was a problem hiding this comment.
The index is only updated after the high watermark moves, but readers can already see the change at that point. So a delta join can read a new order, look it up through the index a moment later, and not find it. Under load (5–10k rows/s, 3 servers, RF 3), 51–76% of rows were readable before they were indexed. A forced race in a delta join left 88 of 600 joined rows stale.
I'd suggest letting readers of tables with sync indexes go only as far as min(HW, syncIndexPushedOffset), and waking waiting fetches when that offset moves. It does mean readers stop when the index stops, so the fail-stop behaviour Giannis asked about matters even more.
WDYT?
| long requiredIndexOffset = delayedBucketStatus.getRequiredIndexOffset(); | ||
| boolean indexOk = | ||
| requiredIndexOffset == DelayedBucketStatus.NO_INDEX_OFFSET_REQUIRED | ||
| || replica.getSyncIndexPushedOffset() >= requiredIndexOffset; |
There was a problem hiding this comment.
sync holds back the writer's ack here, but not what readers can see. I'd suggest applying the same limit on the read side (see Replica comment), so sync means "visible only once indexed" for everyone.
| this.kvSnapshotHandle = kvSnapshotHandle; | ||
| this.logOffset = logOffset; | ||
| this.rowCount = rowCount; | ||
| this.indexPushedOffset = indexPushedOffset; |
There was a problem hiding this comment.
A snapshot can contain rows the index doesn't have yet, so a job that starts from it runs into the same race. In my tests, a delta join started from a snapshot left 25 of 300 rows stale, while one reading the log had none. I'd suggest not handing out a snapshot until the sync index has reached its offset. Alternatively, return indexPushedOffset with the snapshot so readers can wait.
| } | ||
|
|
||
| for (org.apache.fluss.metadata.Schema.Index index : tableInfo.getSchema().getIndexes()) { | ||
| indexes.add(index.getColumnNames()); |
There was a problem hiding this comment.
An index that matches the primary key, the bucket key or another index(async/sync?) makes every Flink query on the table fail with "duplicated index". Deduplicating the list before passing it to Flink would fix that.
| public CompletableFuture<LookupResult> lookup(InternalRow lookupKey) { | ||
| Object[] expectedIdxValues = snapshotLookupKey(lookupKey); | ||
| return indexTablePrefixLookuper | ||
| .lookup(indexLookupKeyEncoder.apply(lookupKey)) |
There was a problem hiding this comment.
Each of the two steps waits for the 100 ms lookup batch timeout, so an index lookup costs about 200 ms when traffic is low. The index itself was only about 5 ms behind the log in my tests. Mb worth sending the second step right away instead of queuing it for the next batch. WDYT?
| lookupKey -> | ||
| GenericRow.of( | ||
| IndexTableUtils.DATA_RECORD_KIND, | ||
| logicalIndexKeyEncoder.encodeKey(lookupKey)); |
There was a problem hiding this comment.
Looking up null throws an exception. Null values are never indexed, so I'd suggest returning an empty result here instead.
| } | ||
|
|
||
| // try secondary index lookup (supports remaining filters when lookup keys are superset) | ||
| int[] matchedIndex = |
There was a problem hiding this comment.
A secondary index is chosen ahead of the prefix lookup. For example, a lookup on (customer_id, status) with an index on status scans every row with that status instead of doing the cheap customer_id prefix lookup. I'd suggest trying the primary key first, then the prefix, and only then indexes. A small unit test for that order would help.
| if (!tableDescriptor.getSchema().getIndexes().isEmpty()) { | ||
| Map<String, String> properties = new HashMap<>(tableDescriptor.getProperties()); | ||
| properties.put( | ||
| ConfigOptions.TABLE_CHANGELOG_IMAGE.key(), ChangelogImage.FULL.toString()); |
There was a problem hiding this comment.
If a user sets table.changelog.image to WAL on an indexed table, it's silently changed to FULL.
I'd suggest rejecting an explicit non-FULL value with a clear error, and only defaulting to FULL when the user didn't set it.
| } | ||
| } | ||
|
|
||
| private void restoreIndexProgress(Long indexPushedOffset) { |
There was a problem hiding this comment.
Index progress is only saved with KV snapshots, so after a restart everything since the last snapshot is replayed, and sync writes wait meanwhile. Mb restoring from the progress rows the index tables already store, so a restart replays only what's actually missing.
cac151b to
19a9d98
Compare
Bound client log fetches and latest offsets by synchronous index progress, and wake delayed fetches when that progress advances. Apply the same bound to remote log files while keeping follower replication and index WAL consumption independent.
19a9d98 to
cf3d4ec
Compare
Purpose
Linked issue: close #4813
Implement global secondary indexes for primary-key tables as described in FIP-38, allowing Java clients and Flink lookup joins to find current rows by non-primary-key equality predicates.
Brief change log
PutIndexRPC.Tests
Passed local checks:
Clean build of all 48 selected Maven modules, including test compilation, Checkstyle, and Spotless:
mvn clean install -DskipTests -pl '!fluss-lake/fluss-lake-lance,!fluss-dist'.47 common metadata/schema/encoding tests and 5 RPC API-key tests.
205 server tests covering index replication, ordering, source/target failover, remote WAL recovery, WAL gaps, snapshot serialization, KV lookups, metadata, and validation.
92 client tests covering secondary-index table lifecycle, partition deletion/recreation, lookup, authorization, and writer accumulation.
11 Flink conversion tests and 15 local secondary-index lookup-join cases across Flink 1.18, 1.19, 1.20, 2.2, and 2.3, including synchronous/asynchronous lookup.
The full unit-test run encountered five
DiskWriteLockedExceptionerrors inReplicaFetcherThreadTest: this host reports 95.55% disk usage against the fixture's default 85% write limit. A diagnostic rerun with the fixture's write-limit ratio set to 1.0 passed all 8 tests. The submitted fixture retains its original configuration.The feature and E2E results listed above were obtained locally. No remote E2E pipeline was triggered.
API and Format
Schema, secondary-index lookup to the Java table API, andsecondary-index.<name>.*Flink table options.PutIndexv0 (API key 1075), optional index replay offsets in KV snapshot metadata, and partition tombstone metadata.PutIndexAPI key 1068 are not wire-compatible with this implementation.Documentation
Add a secondary-index user guide covering table creation, Java lookups, Flink lookup joins, visibility guarantees, lifecycle, recovery, and limitations. Link it from the primary-key table and Java API guides, add a Flink join example, and document all index replication options and the client continuation thread setting. The design follows FIP-38.