From a5a3209d18b4dc63f5a6671c10d385e2affc759d Mon Sep 17 00:00:00 2001 From: Bryan Burkholder Date: Fri, 9 Feb 2024 13:24:43 -0700 Subject: [PATCH 1/2] Add ID field to search protobuf --- .../com/slack/kaldb/logstore/search/SearchResultUtils.java | 4 ++++ kaldb/src/main/proto/kaldb_search.proto | 1 + 2 files changed, 5 insertions(+) diff --git a/kaldb/src/main/java/com/slack/kaldb/logstore/search/SearchResultUtils.java b/kaldb/src/main/java/com/slack/kaldb/logstore/search/SearchResultUtils.java index ffa2cf9f85..d2e62b4f89 100644 --- a/kaldb/src/main/java/com/slack/kaldb/logstore/search/SearchResultUtils.java +++ b/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)); diff --git a/kaldb/src/main/proto/kaldb_search.proto b/kaldb/src/main/proto/kaldb_search.proto index 079ddd1e5c..3161df1312 100644 --- a/kaldb/src/main/proto/kaldb_search.proto +++ b/kaldb/src/main/proto/kaldb_search.proto @@ -247,6 +247,7 @@ enum FieldType { FLOAT = 4; BOOLEAN = 5; DOUBLE = 6; + ID = 7; } message SchemaDefinition { From 961df72d62bae9d481ff485221e212f547f0de3c Mon Sep 17 00:00:00 2001 From: Bryan Burkholder Date: Mon, 12 Feb 2024 15:53:33 -0700 Subject: [PATCH 2/2] Refactor read only chunk timeout --- .../slack/kaldb/chunk/ReadOnlyChunkImpl.java | 29 +++++-------------- .../search/SearchResultUtilsTest.java | 10 +++++++ 2 files changed, 17 insertions(+), 22 deletions(-) diff --git a/kaldb/src/main/java/com/slack/kaldb/chunk/ReadOnlyChunkImpl.java b/kaldb/src/main/java/com/slack/kaldb/chunk/ReadOnlyChunkImpl.java index ed3ad5c54d..b220c2304b 100644 --- a/kaldb/src/main/java/com/slack/kaldb/chunk/ReadOnlyChunkImpl.java +++ b/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 implements Chunk { 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 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); diff --git a/kaldb/src/test/java/com/slack/kaldb/logstore/search/SearchResultUtilsTest.java b/kaldb/src/test/java/com/slack/kaldb/logstore/search/SearchResultUtilsTest.java index 90126ded85..f72a52233d 100644 --- a/kaldb/src/test/java/com/slack/kaldb/logstore/search/SearchResultUtilsTest.java +++ b/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