Skip to content

Commit 7b49072

Browse files
RavirajDataEngineeringRavirajDataEngineering
committed
docvalue update API changes to the v0.x branch
1 parent a96a1a8 commit 7b49072

File tree

7 files changed

+450
-2
lines changed

7 files changed

+450
-2
lines changed

‎clientlib/src/main/proto/yelp/nrtsearch/luceneserver.proto

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -684,6 +684,8 @@ message AddDocumentRequest {
684684
repeated FacetHierarchyPath faceHierarchyPaths = 2;
685685
}
686686
map<string, MultiValuedField> fields = 3; //map of field name to a list of string values.
687+
688+
IndexingRequestType requestType = 4;
687689
}
688690

689691
message FacetHierarchyPath {
@@ -1156,3 +1158,9 @@ message CustomRequest {
11561158
message CustomResponse {
11571159
map<string, string> response = 1; // Custom response sent by the plugin
11581160
}
1161+
enum IndexingRequestType {
1162+
// Request to add a document
1163+
ADD_DOCUMENT = 0;
1164+
// Request to update a document
1165+
UPDATE_DOCUMENT = 2;
1166+
}

‎src/main/java/com/yelp/nrtsearch/server/grpc/LuceneServer.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@
5858
import com.yelp.nrtsearch.server.luceneserver.similarity.SimilarityCreator;
5959
import com.yelp.nrtsearch.server.luceneserver.warming.Warmer;
6060
import com.yelp.nrtsearch.server.monitoring.*;
61+
import com.yelp.nrtsearch.server.monitoring.IndexingMetrics;
6162
import com.yelp.nrtsearch.server.monitoring.ThreadPoolCollector.RejectionCounterWrapper;
6263
import com.yelp.nrtsearch.server.plugins.Plugin;
6364
import com.yelp.nrtsearch.server.plugins.PluginsService;
@@ -248,6 +249,7 @@ private void registerMetrics(GlobalState globalState) {
248249
new ProcStatCollector().register(collectorRegistry);
249250
new MergeSchedulerCollector(globalState).register(collectorRegistry);
250251
new SearchResponseCollector(globalState).register(collectorRegistry);
252+
IndexingMetrics.register(collectorRegistry);
251253
}
252254

253255
/** Main launches the server from the command line. */

‎src/main/java/com/yelp/nrtsearch/server/luceneserver/AddDocumentHandler.java

Lines changed: 72 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,15 @@
1717

1818
import com.google.protobuf.ProtocolStringList;
1919
import com.yelp.nrtsearch.server.grpc.AddDocumentRequest;
20+
import com.yelp.nrtsearch.server.grpc.AddDocumentRequest.MultiValuedField;
2021
import com.yelp.nrtsearch.server.grpc.DeadlineUtils;
2122
import com.yelp.nrtsearch.server.grpc.FacetHierarchyPath;
23+
import com.yelp.nrtsearch.server.grpc.IndexingRequestType;
2224
import com.yelp.nrtsearch.server.luceneserver.field.FieldDef;
2325
import com.yelp.nrtsearch.server.luceneserver.field.IdFieldDef;
2426
import com.yelp.nrtsearch.server.luceneserver.field.IndexableFieldDef;
27+
import com.yelp.nrtsearch.server.luceneserver.field.properties.DocValueUpdatable;
28+
import com.yelp.nrtsearch.server.monitoring.IndexingMetrics;
2529
import java.io.IOException;
2630
import java.util.ArrayList;
2731
import java.util.HashMap;
@@ -33,7 +37,9 @@
3337
import java.util.concurrent.LinkedBlockingDeque;
3438
import java.util.stream.Collectors;
3539
import org.apache.lucene.document.Document;
40+
import org.apache.lucene.document.Field;
3641
import org.apache.lucene.index.IndexableField;
42+
import org.apache.lucene.index.Term;
3743
import org.slf4j.Logger;
3844
import org.slf4j.LoggerFactory;
3945

@@ -77,6 +83,8 @@ public static DocumentsContext getDocumentsContext(
7783
AddDocumentRequest addDocumentRequest, IndexState indexState)
7884
throws AddDocumentHandlerException {
7985
DocumentsContext documentsContext = new DocumentsContext();
86+
if (addDocumentRequest.getRequestType().equals(IndexingRequestType.UPDATE_DOCUMENT))
87+
return documentsContext;
8088
Map<String, AddDocumentRequest.MultiValuedField> fields = addDocumentRequest.getFieldsMap();
8189
for (Map.Entry<String, AddDocumentRequest.MultiValuedField> entry : fields.entrySet()) {
8290
parseOneField(entry.getKey(), entry.getValue(), documentsContext, indexState);
@@ -222,7 +230,12 @@ public long runIndexingJob() throws Exception {
222230
throw new IOException(e);
223231
}
224232
} else {
225-
documents.add(documentsContext.getRootDocument());
233+
if (addDocumentRequest.getRequestType().equals(IndexingRequestType.UPDATE_DOCUMENT)) {
234+
executeDocValueUpdateRequest(
235+
documentsContext, indexState, shardState, addDocumentRequest);
236+
} else {
237+
documents.add(documentsContext.getRootDocument());
238+
}
226239
}
227240
}
228241
} catch (Exception e) {
@@ -252,6 +265,54 @@ public long runIndexingJob() throws Exception {
252265
return shardState.writer.getMaxCompletedSequenceNumber();
253266
}
254267

268+
private void executeDocValueUpdateRequest(
269+
DocumentsContext documentsContext,
270+
IndexState indexState,
271+
ShardState shardState,
272+
AddDocumentRequest addDocumentRequest) {
273+
try {
274+
IndexingMetrics.updateDocValuesRequestsReceived.labels(indexName).inc();
275+
List<Field> updatableDocValueFields = new ArrayList<>();
276+
Term term = null;
277+
for (Map.Entry<String, MultiValuedField> entry :
278+
addDocumentRequest.getFieldsMap().entrySet()) {
279+
FieldDef field = indexState.getField(entry.getKey());
280+
if (field.getName().equals(indexState.getIdFieldDef().get().getName())) {
281+
282+
String idFieldName = indexState.getIdFieldDef().get().getName();
283+
String idFieldValue = addDocumentRequest.getFieldsMap().get(idFieldName).getValue(0);
284+
285+
if (idFieldValue == null || idFieldValue.isEmpty()) {
286+
throw new IllegalArgumentException(
287+
String.format("the _ID should have a value set to execute update DocValue"));
288+
}
289+
term = new Term(idFieldName, idFieldValue);
290+
continue;
291+
}
292+
if (!(field instanceof DocValueUpdatable updatable) || !(updatable.isUpdatable())) {
293+
throw new IllegalArgumentException(
294+
String.format("Field: %s is not updatable", field.getName()));
295+
}
296+
updatableDocValueFields.add(
297+
((DocValueUpdatable) field).getUpdatableDocValueField(entry.getValue().getValue(0)));
298+
}
299+
if (term == null) {
300+
throw new RuntimeException("_ ID field should be present for the update request");
301+
}
302+
long nanoTime = System.nanoTime();
303+
shardState.writer.updateDocValues(term, updatableDocValueFields.toArray(new Field[0]));
304+
IndexingMetrics.updateDocValuesLatency
305+
.labels(indexName)
306+
.set((System.nanoTime() - nanoTime));
307+
} catch (Throwable t) {
308+
logger.warn(
309+
String.format(
310+
"ThreadId: %s, IndexWriter.updateDocValues failed",
311+
Thread.currentThread().getName() + Thread.currentThread().threadId()));
312+
throw new RuntimeException("Error occurred when updating docValues ", t);
313+
}
314+
}
315+
255316
/**
256317
* update documents with nested objects
257318
*
@@ -315,7 +376,10 @@ private void updateDocuments(
315376
throws IOException {
316377
for (Document nextDoc : documents) {
317378
nextDoc = handleFacets(indexState, shardState, nextDoc);
379+
IndexingMetrics.addDocumentRequestsReceived.labels(indexName).inc();
380+
long nanoTime = System.nanoTime();
318381
shardState.writer.updateDocument(idFieldDef.getTerm(nextDoc), nextDoc);
382+
IndexingMetrics.addDocumentLatency.labels(indexName).set((System.nanoTime() - nanoTime));
319383
}
320384
}
321385

@@ -326,6 +390,8 @@ private void addDocuments(
326390
throw new IllegalStateException(
327391
"Adding documents to an index on a replica node is not supported");
328392
}
393+
IndexingMetrics.addDocumentRequestsReceived.labels(indexName).inc(documents.size());
394+
long nanoTime = System.nanoTime();
329395
shardState.writer.addDocuments(
330396
(Iterable<Document>)
331397
() ->
@@ -349,6 +415,11 @@ public Document next() {
349415
return nextDoc;
350416
}
351417
});
418+
if (documents.size() >= 1) {
419+
IndexingMetrics.addDocumentLatency
420+
.labels(indexName)
421+
.set((System.nanoTime() - nanoTime) / documents.size());
422+
}
352423
}
353424

354425
private Document handleFacets(IndexState indexState, ShardState shardState, Document nextDoc) {

‎src/main/java/com/yelp/nrtsearch/server/luceneserver/field/NumberFieldDef.java

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import com.yelp.nrtsearch.server.luceneserver.field.BindingValuesSources.SortedNumericLengthValuesSource;
2929
import com.yelp.nrtsearch.server.luceneserver.field.BindingValuesSources.SortedNumericMinValuesSource;
3030
import com.yelp.nrtsearch.server.luceneserver.field.properties.Bindable;
31+
import com.yelp.nrtsearch.server.luceneserver.field.properties.DocValueUpdatable;
3132
import com.yelp.nrtsearch.server.luceneserver.field.properties.RangeQueryable;
3233
import com.yelp.nrtsearch.server.luceneserver.field.properties.Sortable;
3334
import com.yelp.nrtsearch.server.luceneserver.field.properties.TermQueryable;
@@ -53,7 +54,7 @@
5354
* all of number fields and provides abstract functions for type specific operations.
5455
*/
5556
public abstract class NumberFieldDef extends IndexableFieldDef
56-
implements Bindable, Sortable, RangeQueryable, TermQueryable {
57+
implements Bindable, Sortable, RangeQueryable, TermQueryable, DocValueUpdatable {
5758
public static final Function<String, Number> INT_PARSER = Integer::valueOf;
5859
public static final Function<String, Number> LONG_PARSER = Long::valueOf;
5960
public static final Function<String, Number> FLOAT_PARSER = Float::valueOf;
@@ -285,4 +286,17 @@ public SortField getSortField(SortType type) {
285286
sortField.setMissingValue(getSortMissingValue(missingLast));
286287
return sortField;
287288
}
289+
290+
@Override
291+
public boolean isUpdatable() {
292+
if (isSearchable() || isMultiValue() || !hasDocValues()) {
293+
return false;
294+
}
295+
return true;
296+
}
297+
298+
@Override
299+
public org.apache.lucene.document.Field getUpdatableDocValueField(Object val) {
300+
return getDocValueField(parseNumberString((String) val));
301+
}
288302
}

‎src/main/java/com/yelp/nrtsearch/server/luceneserver/field/properties/DocValueUpdatable.java

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
/*
2+
* Copyright 2025 Yelp Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.yelp.nrtsearch.server.luceneserver.field.properties;
17+
18+
public interface DocValueUpdatable<T> {
19+
20+
/**
21+
* determine if this {@link com.yelp.nrtsearch.server.luceneserver.field.FieldDef} can be updated,
22+
* this will depend on other factors such as, if the field is searchable or is multivalued etc.
23+
*
24+
* @return if the field can be updated.
25+
*/
26+
boolean isUpdatable();
27+
28+
/**
29+
* @param value value to be updated
30+
* @return get the docValue for this {@link com.yelp.nrtsearch.server.luceneserver.field.FieldDef}
31+
* to update.
32+
*/
33+
org.apache.lucene.document.Field getUpdatableDocValueField(T value);
34+
}

‎src/main/java/com/yelp/nrtsearch/server/monitoring/IndexingMetrics.java

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
/*
2+
* Copyright 2025 Yelp Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.yelp.nrtsearch.server.monitoring;
17+
18+
import io.prometheus.client.CollectorRegistry;
19+
import io.prometheus.client.Counter;
20+
import io.prometheus.client.Gauge;
21+
22+
public class IndexingMetrics {
23+
24+
public static final String UPDATE_DOC_VALUES_REQUESTS_RECEIVED =
25+
"update_doc_values_requests_received";
26+
public static final String ADD_DOCUMENT_REQUESTS_RECEIVED = "add_document_requests_received";
27+
public static final String UPDATE_DOC_VALUES_LATENCY = "update_doc_values_latency";
28+
public static final String ADD_DOCUMENT_LATENCY = "add_document_latency";
29+
30+
public static final Counter updateDocValuesRequestsReceived =
31+
Counter.build()
32+
.name(UPDATE_DOC_VALUES_REQUESTS_RECEIVED)
33+
.help("Number of requests received for the update doc values API ")
34+
.labelNames("index")
35+
.create();
36+
37+
// counter for addDocument requests received for the index with the index name as the label value
38+
public static final Counter addDocumentRequestsReceived =
39+
Counter.build()
40+
.name(ADD_DOCUMENT_REQUESTS_RECEIVED)
41+
.help("Number of requests received for the add document API ")
42+
.labelNames("index")
43+
.create();
44+
45+
public static final Gauge updateDocValuesLatency =
46+
Gauge.build()
47+
.name(UPDATE_DOC_VALUES_LATENCY)
48+
.help("Latency of the update doc values API")
49+
.labelNames("index")
50+
.create();
51+
52+
// gauge for the latency of the addDocument API with the index name as the label value
53+
public static final Gauge addDocumentLatency =
54+
Gauge.build()
55+
.name(ADD_DOCUMENT_LATENCY)
56+
.help("Latency of the add document API")
57+
.labelNames("index")
58+
.create();
59+
60+
public static void register(CollectorRegistry registry) {
61+
registry.register(updateDocValuesRequestsReceived);
62+
registry.register(addDocumentRequestsReceived);
63+
registry.register(updateDocValuesLatency);
64+
registry.register(addDocumentLatency);
65+
}
66+
}

0 commit comments

Comments
 (0)