Add ID field to search protobuf

kaldb/src/main/java/com/slack/kaldb/chunk/ReadOnlyChunkImpl.java

@@ -1,5 +1,7 @@
 package com.slack.kaldb.chunk;
 
+import static com.slack.kaldb.server.KaldbConfig.DEFAULT_ZK_TIMEOUT_SECS;
+
 import com.google.common.annotations.VisibleForTesting;
 import com.slack.kaldb.blobfs.BlobFs;
 import com.slack.kaldb.logstore.search.LogIndexSearcher;
@@ -48,10 +50,6 @@
 public class ReadOnlyChunkImpl<T> implements Chunk<T> {
 
   private static final Logger LOG = LoggerFactory.getLogger(ReadOnlyChunkImpl.class);
-
-  @Deprecated // replace with sync methods, which use DEFAULT_ZK_TIMEOUT_SECS where possible
-  private static final int TIMEOUT_MS = 5000;
-
   private ChunkInfo chunkInfo;
   private LogIndexSearcher<T> logSearcher;
   private SearchMetadata searchMetadata;
@@ -174,20 +172,14 @@ public static SearchMetadata registerSearchMetadata(
                 snapshotName, cacheSearchContext.hostname),
             snapshotName,
             cacheSearchContext.toUrl());
-    searchMetadataStore
-        .createAsync(metadata)
-        .toCompletableFuture()
-        .get(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    searchMetadataStore.createSync(metadata);
     return metadata;
   }
 
   private void unregisterSearchMetadata()
       throws ExecutionException, InterruptedException, TimeoutException {
     if (this.searchMetadata != null) {
-      searchMetadataStore
-          .deleteAsync(searchMetadata)
-          .toCompletableFuture()
-          .get(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+      searchMetadataStore.deleteSync(searchMetadata);
     }
   }
 
@@ -265,15 +257,8 @@ private void handleChunkAssignment(CacheSlotMetadata cacheSlotMetadata) {
 
   private SnapshotMetadata getSnapshotMetadata(String replicaId)
       throws ExecutionException, InterruptedException, TimeoutException {
-    ReplicaMetadata replicaMetadata =
-        replicaMetadataStore
-            .findAsync(replicaId)
-            .toCompletableFuture()
-            .get(TIMEOUT_MS, TimeUnit.MILLISECONDS);
-    return snapshotMetadataStore
-        .findAsync(replicaMetadata.snapshotId)
-        .toCompletableFuture()
-        .get(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    ReplicaMetadata replicaMetadata = replicaMetadataStore.findSync(replicaId);
+    return snapshotMetadataStore.findSync(replicaMetadata.snapshotId);
   }
 
   // We lock access when manipulating the chunk, as the close()
@@ -319,7 +304,7 @@ private boolean setChunkMetadataState(
     try {
       cacheSlotMetadataStore
           .updateNonFreeCacheSlotState(cacheSlotMetadata, newState)
-          .get(TIMEOUT_MS, TimeUnit.MILLISECONDS);
+          .get(DEFAULT_ZK_TIMEOUT_SECS, TimeUnit.MILLISECONDS);
       return true;
     } catch (InterruptedException | ExecutionException | TimeoutException e) {
       LOG.error("Error setting chunk metadata state", e);

kaldb/src/main/java/com/slack/kaldb/logstore/search/SearchResultUtils.java

@@ -655,6 +655,8 @@ public static FieldType fromSchemaDefinitionProto(
       return FieldType.BOOLEAN;
     } else if (protoSchemaDefinition.getType().equals(KaldbSearch.FieldType.DOUBLE)) {
       return FieldType.DOUBLE;
+    } else if (protoSchemaDefinition.getType().equals(KaldbSearch.FieldType.ID)) {
+      return FieldType.ID;
     } else {
       throw new IllegalArgumentException(
           String.format("Field type %s is not a supported type", protoSchemaDefinition.getType()));
@@ -678,6 +680,8 @@ public static KaldbSearch.SchemaDefinition toSchemaDefinitionProto(FieldType fie
       schemaBuilder.setType(KaldbSearch.FieldType.BOOLEAN);
     } else if (fieldType.equals(FieldType.DOUBLE)) {
       schemaBuilder.setType(KaldbSearch.FieldType.DOUBLE);
+    } else if (fieldType.equals(FieldType.ID)) {
+      schemaBuilder.setType(KaldbSearch.FieldType.ID);
     } else {
       throw new IllegalArgumentException(
           String.format("Field type %s is not a supported type", fieldType));

kaldb/src/main/proto/kaldb_search.proto

@@ -247,6 +247,7 @@ enum FieldType {
   FLOAT = 4;
   BOOLEAN = 5;
   DOUBLE = 6;
+  ID = 7;
 }
 
 message SchemaDefinition {

kaldb/src/test/java/com/slack/kaldb/logstore/search/SearchResultUtilsTest.java

@@ -404,6 +404,16 @@ public void shouldConvertSchemaDefinitionToFromProto() {
             KaldbSearch.SchemaDefinition.newBuilder()
                 .setType(KaldbSearch.FieldType.INTEGER)
                 .build());
+
+    assertThat(
+            SearchResultUtils.fromSchemaDefinitionProto(
+                KaldbSearch.SchemaDefinition.newBuilder()
+                    .setType(KaldbSearch.FieldType.ID)
+                    .build()))
+        .isEqualTo(FieldType.ID);
+    assertThat(SearchResultUtils.toSchemaDefinitionProto(FieldType.ID))
+        .isEqualTo(
+            KaldbSearch.SchemaDefinition.newBuilder().setType(KaldbSearch.FieldType.ID).build());
   }
 
   @Test