Support secondary index on point get operation

client-api/src/main/java/io/streamnative/oxia/client/api/GetOption.java

@@ -18,7 +18,7 @@
 import lombok.NonNull;
 
 public sealed interface GetOption
-        permits OptionComparisonType, OptionIncludeValue, OptionPartitionKey {
+        permits OptionComparisonType, OptionIncludeValue, OptionPartitionKey, OptionSecondaryIndexName {
 
     /** ComparisonEqual sets the Get() operation to compare the stored key for equality. */
     GetOption ComparisonEqual = new OptionComparisonType(OptionComparisonType.ComparisonType.Equal);
@@ -78,4 +78,15 @@ static GetOption PartitionKey(@NonNull String partitionKey) {
     static GetOption IncludeValue(boolean includeValue) {
         return new OptionIncludeValue(includeValue);
     }
+
+    /**
+     * UseIndex let the users specify a different index to follow for the list operation
+     *
+     * <p>Note: if the secondary index is not unique, which primary record is returned is undefined.
+     *
+     * @param secondaryIndexName the name of the secondary index to use for the list operation
+     */
+    static GetOption UseIndex(@NonNull String secondaryIndexName) {
+        return new OptionSecondaryIndexName(secondaryIndexName);
+    }
 }

client-api/src/main/java/io/streamnative/oxia/client/api/OptionSecondaryIndexName.java

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2022-2024 StreamNative Inc.
+ * Copyright © 2022-2025 StreamNative Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,4 +16,4 @@
 package io.streamnative.oxia.client.api;
 
 public record OptionSecondaryIndexName(String secondaryIndexName)
-        implements ListOption, RangeScanOption {}
+        implements ListOption, RangeScanOption, GetOption {}

client-it/src/test/java/io/streamnative/oxia/client/it/OxiaClientIT.java

@@ -705,4 +705,113 @@ void testGetIncludeValue() {
         Assertions.assertEquals(result.getValue().length, 0);
         Assertions.assertEquals(result.getKey(), keys.get(0));
     }
+
+    @Test
+    void testSecondaryIndexGet() throws Exception {
+        @Cleanup
+        SyncOxiaClient client = OxiaClientBuilder.create(oxia.getServiceAddress()).syncClient();
+
+        String prefix = "si-get-" + UUID.randomUUID();
+
+        for (int i = 1; i < 10; i++) {
+            String primKey = String.format("%s-%c", prefix, 'a' + i);
+            String val = String.format("%03d", i);
+
+            client.put(primKey, val.getBytes(), Set.of(PutOption.SecondaryIndex("val-idx", val)));
+        }
+
+        ////////////////////////////////////////////////////////////////////////
+
+        GetResult gr = client.get("000", Set.of(GetOption.UseIndex("val-idx")));
+        assertThat(gr).isNull();
+
+        gr = client.get("001", Set.of(GetOption.UseIndex("val-idx")));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-b");
+        assertThat(gr.getValue()).isEqualTo("001".getBytes());
+
+        gr = client.get("005", Set.of(GetOption.UseIndex("val-idx")));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-f");
+        assertThat(gr.getValue()).isEqualTo("005".getBytes());
+
+        gr = client.get("009", Set.of(GetOption.UseIndex("val-idx")));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-j");
+        assertThat(gr.getValue()).isEqualTo("009".getBytes());
+
+        gr = client.get("999", Set.of(GetOption.UseIndex("val-idx")));
+        assertThat(gr).isNull();
+
+        ////////////////////////////////////////////////////////////////////////
+
+        gr = client.get("000", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonLower));
+        assertThat(gr).isNull();
+
+        gr = client.get("001", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonLower));
+        assertThat(gr).isNull();
+
+        gr = client.get("005", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonLower));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-e");
+        assertThat(gr.getValue()).isEqualTo("004".getBytes());
+
+        gr = client.get("009", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonLower));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-i");
+        assertThat(gr.getValue()).isEqualTo("008".getBytes());
+
+        gr = client.get("999", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonLower));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-j");
+        assertThat(gr.getValue()).isEqualTo("009".getBytes());
+
+        ////////////////////////////////////////////////////////////////////////
+
+        gr = client.get("000", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonHigher));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-b");
+        assertThat(gr.getValue()).isEqualTo("001".getBytes());
+
+        gr = client.get("001", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonHigher));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-c");
+        assertThat(gr.getValue()).isEqualTo("002".getBytes());
+
+        gr = client.get("005", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonHigher));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-g");
+        assertThat(gr.getValue()).isEqualTo("006".getBytes());
+
+        gr = client.get("009", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonHigher));
+        assertThat(gr).isNull();
+
+        gr = client.get("999", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonHigher));
+        assertThat(gr).isNull();
+
+        ////////////////////////////////////////////////////////////////////////
+
+        gr = client.get("000", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonCeiling));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-b");
+        assertThat(gr.getValue()).isEqualTo("001".getBytes());
+
+        gr = client.get("001", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonCeiling));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-b");
+        assertThat(gr.getValue()).isEqualTo("001".getBytes());
+
+        gr = client.get("005", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonCeiling));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-f");
+        assertThat(gr.getValue()).isEqualTo("005".getBytes());
+
+        gr = client.get("009", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonCeiling));
+        assertThat(gr).isNotNull();
+        assertThat(gr.getKey()).isEqualTo(prefix + "-j");
+        assertThat(gr.getValue()).isEqualTo("009".getBytes());
+
+        gr = client.get("999", Set.of(GetOption.UseIndex("val-idx"), GetOption.ComparisonCeiling));
+        assertThat(gr).isNull();
+    }
 }

client/src/main/java/io/streamnative/oxia/client/AsyncOxiaClientImpl.java

@@ -493,17 +493,19 @@ private CompletableFuture<PutResult> internalPut(
     }
 
     private void internalGet(String key, GetOptions options, CompletableFuture<GetResult> result) {
-        if (options.comparisonType() == KeyComparisonType.EQUAL || options.partitionKey() != null) {
+        if (options.partitionKey() == null
+                && (options.comparisonType() != KeyComparisonType.EQUAL
+                        || options.secondaryIndexName() != null)) {
+            internalGetMultiShards(key, options, result);
+        } else {
             // Single shard get operation
             long shardId =
                     shardManager.getShardForKey(Optional.ofNullable(options.partitionKey()).orElse(key));
             readBatchManager.getBatcher(shardId).add(new GetOperation(result, key, options));
-        } else {
-            internalGetFloorCeiling(key, options, result);
         }
     }
 
-    private void internalGetFloorCeiling(
+    private void internalGetMultiShards(
             String key, GetOptions options, CompletableFuture<GetResult> result) {
         // We need check on all the shards for a floor/ceiling query
         List<CompletableFuture<GetResult>> futures = new ArrayList<>();
@@ -536,11 +538,9 @@ private void internalGetFloorCeiling(
 
                                 GetResult gr =
                                         switch (options.comparisonType()) {
-                                            case EQUAL,
-                                                    UNRECOGNIZED -> null; // This would be handled withing context of single
-                                                // shard
+                                            case EQUAL, CEILING, HIGHER -> results.get(0);
                                             case FLOOR, LOWER -> results.get(results.size() - 1);
-                                            case CEILING, HIGHER -> results.get(0);
+                                            case UNRECOGNIZED -> null;
                                         };
 
                                 result.complete(gr);

client/src/main/java/io/streamnative/oxia/client/batch/Operation.java

@@ -64,11 +64,15 @@ record GetOperation(
                 @NonNull GetOptions options)
                 implements ReadOperation<GetResult> {
             GetRequest toProto() {
-                return GetRequest.newBuilder()
-                        .setKey(key)
-                        .setComparisonType(options.comparisonType())
-                        .setIncludeValue(options.includeValue())
-                        .build();
+                var builder =
+                        GetRequest.newBuilder()
+                                .setKey(key)
+                                .setComparisonType(options.comparisonType())
+                                .setIncludeValue(options.includeValue());
+                if (options.secondaryIndexName() != null) {
+                    builder.setSecondaryIndexName(options.secondaryIndexName());
+                }
+                return builder.build();
             }
 
             void complete(@NonNull GetResponse response) {

client/src/main/java/io/streamnative/oxia/client/options/GetOptions.java

@@ -19,16 +19,21 @@
 import io.streamnative.oxia.client.api.OptionComparisonType;
 import io.streamnative.oxia.client.api.OptionIncludeValue;
 import io.streamnative.oxia.client.api.OptionPartitionKey;
+import io.streamnative.oxia.client.api.OptionSecondaryIndexName;
 import io.streamnative.oxia.proto.KeyComparisonType;
 import java.util.Set;
 
 public record GetOptions(
-        String partitionKey, boolean includeValue, KeyComparisonType comparisonType) {
+        String partitionKey,
+        boolean includeValue,
+        KeyComparisonType comparisonType,
+        String secondaryIndexName) {
 
     public static GetOptions parseFrom(Set<GetOption> options) {
         boolean includeValue = true;
         KeyComparisonType comparisonType = KeyComparisonType.EQUAL;
         String partitionKey = null;
+        String secondaryIndexName = null;
         for (GetOption option : options) {
             if (option instanceof OptionIncludeValue) {
                 includeValue = ((OptionIncludeValue) option).includeValue();
@@ -49,7 +54,11 @@ public static GetOptions parseFrom(Set<GetOption> options) {
                 partitionKey = ((OptionPartitionKey) option).partitionKey();
                 continue;
             }
+            if (option instanceof OptionSecondaryIndexName) {
+                secondaryIndexName = ((OptionSecondaryIndexName) option).secondaryIndexName();
+                continue;
+            }
         }
-        return new GetOptions(partitionKey, includeValue, comparisonType);
+        return new GetOptions(partitionKey, includeValue, comparisonType, secondaryIndexName);
     }
 }

client/src/main/proto/io/streamnative/oxia/client.proto

@@ -324,6 +324,7 @@ message GetRequest {
   bool include_value = 2;
 
   KeyComparisonType comparison_type = 3;
+  optional string secondary_index_name = 4;
 }
 
 /**
@@ -339,6 +340,7 @@ message GetResponse {
   // In case of non-exact queries (eg. floor, ceiling) the found key will be
   // returned in the GetResponse.
   optional string key = 4;
+  optional string secondary_index_key = 5;
 }
 
 /**

client/src/test/java/io/streamnative/oxia/client/batch/BatchTest.java

@@ -395,7 +395,8 @@ class ReadBatchTests {
         ReadBatch batch;
         CompletableFuture<GetResult> getCallable = new CompletableFuture<>();
         GetOperation get =
-                new GetOperation(getCallable, "", new GetOptions(null, true, KeyComparisonType.EQUAL));
+                new GetOperation(
+                        getCallable, "", new GetOptions(null, true, KeyComparisonType.EQUAL, null));
 
         @BeforeEach
         void setup() {

client/src/test/java/io/streamnative/oxia/client/batch/BatcherTest.java

@@ -89,7 +89,8 @@ void teardown() {
     void createBatchAndAdd() throws Exception {
         var callback = new CompletableFuture<GetResult>();
         Operation<?> op =
-                new GetOperation(callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL));
+                new GetOperation(
+                        callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL, null));
         when(batchFactory.getBatch(shardId)).thenReturn(batch);
         when(batch.size()).thenReturn(1);
         when(batch.canAdd(any())).thenReturn(true);
@@ -102,7 +103,8 @@ void createBatchAndAdd() throws Exception {
     void sendBatchOnFull() throws Exception {
         var callback = new CompletableFuture<GetResult>();
         Operation<?> op =
-                new GetOperation(callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL));
+                new GetOperation(
+                        callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL, null));
         when(batchFactory.getBatch(shardId)).thenReturn(batch);
         when(batch.size()).thenReturn(config.maxRequestsPerBatch());
         when(batch.canAdd(any())).thenReturn(true);
@@ -140,7 +142,8 @@ void addWhenNextDoesNotFit() {
     void sendBatchOnFullThenNewBatch() throws Exception {
         var callback = new CompletableFuture<GetResult>();
         Operation<?> op =
-                new GetOperation(callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL));
+                new GetOperation(
+                        callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL, null));
         when(batchFactory.getBatch(shardId)).thenReturn(batch);
         when(batch.size()).thenReturn(config.maxRequestsPerBatch(), 1);
         when(batch.canAdd(any())).thenReturn(true);
@@ -159,7 +162,8 @@ void sendBatchOnFullThenNewBatch() throws Exception {
     void sendBatchOnLingerExpiration() throws Exception {
         var callback = new CompletableFuture<GetResult>();
         Operation<?> op =
-                new GetOperation(callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL));
+                new GetOperation(
+                        callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL, null));
         when(batchFactory.getBatch(shardId)).thenReturn(batch);
         when(batch.size()).thenReturn(1);
         when(batch.canAdd(any())).thenReturn(true);
@@ -173,7 +177,8 @@ void sendBatchOnLingerExpiration() throws Exception {
     void sendBatchOnLingerExpirationMulti() throws Exception {
         var callback = new CompletableFuture<GetResult>();
         Operation<?> op =
-                new GetOperation(callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL));
+                new GetOperation(
+                        callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL, null));
         when(batchFactory.getBatch(shardId)).thenReturn(batch);
         when(batch.size()).thenReturn(1);
         when(batch.canAdd(any())).thenReturn(true);
@@ -190,7 +195,8 @@ void sendBatchOnLingerExpirationMulti() throws Exception {
     void unboundedTakeAtStart() throws Exception {
         var callback = new CompletableFuture<GetResult>();
         Operation<?> op =
-                new GetOperation(callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL));
+                new GetOperation(
+                        callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL, null));
 
         when(batchFactory.getBatch(shardId)).thenReturn(batch);
         when(batch.size()).thenReturn(1);

client/src/test/java/io/streamnative/oxia/client/batch/OperationTest.java

@@ -63,7 +63,8 @@ class GetOperationTests {
 
         CompletableFuture<GetResult> callback = new CompletableFuture<>();
         GetOperation op =
-                new GetOperation(callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL));
+                new GetOperation(
+                        callback, "key", new GetOptions(null, true, KeyComparisonType.EQUAL, null));
 
         @Test
         void toProto() {