Optimized text embedding processor for single document update #1191

will-hwang · 2025-02-19T06:30:54Z

Description

This PR implements optimized text embedding processor for single document update scenario.

Proposed State [Single Document Update]

Steps:

Process Map is generated based on IngestDocument and Field Map defined.
If ignore_existing feature is set to true, filter the process map.
1. Existing document is fetched via OpenSearch client’s Get Action to compare the existing inference text against IngestDocument
  1. if document does not exist, or any exception is thrown, fallback to calling model inference
2. Locate the embedding fields in the existing document
  1. Recursively traverse the process map to find embedding fields. Keeping track of the traversal path is required for look up in existing document.
  2. Once embedding fields are found, attempt to copy embeddings from existing document to ingest document.
3. If eligible, copy over the vector embeddings from existing document to ingest document
  1. It is eligible for copy if inference text in ingest document and existing document is the same, and embeddings for the inference text exist in the existing document.
  2. Note, if in case of values in list, the fields in the same index are compared to determine text equality
4. If eligible fields have been copied, remove the entry in process map
Inference List is generated based on entries in Filtered Process Map.
ML Commons InferenceSentence API is invoked with filtered inference list.
Embeddings for filtered inference list are generated in ML Commons.
Embeddings for filtered inference list are mapped to target fields via entries defined in process map.
Embeddings for filtered inference list are populated to target fields in IngestDocument

User Scenarios

Scenario 1: Document Update with no change to all fields

Update Document
ID: 1
"category": [
    {
        "name": {
            "en": "this is 1st name"
        }
    },
    {
        "name": {
            "en": "this is 2nd name"
        }
    }
]

Existing Document
ID: 1
"category": [
    {
        "name": {
            "category_name_vector": [
                -0.1,
                -0.1,
                -0.1,
                ...
            ],
            "en": "this is 1st name"
        }
        
    },
    {
        "name": {
            "category_name_vector": [
                -0.2,
               -0.2,
               -0.2,
                ...
            ],
            "en": "this is 2nd name"
        }
    }
]

Filtered Process Map 
{
  "category":` `
        {
            "name": {
                "category_name_vector": [~~"this is 1st name", "this is 2nd name"~~]
            }
        }
}

Filtered Inference List
[]

Ingested Document
"category": [
    {
        "name": {
            "category_name_vector": [ // copied from existing document
                -0.1,
                -0.1,
                -0.1,
                ...
            ],
            "en": "this is 1st name"
        }
        
    },
    {
        "name": {
            "category_name_vector": [ // copied from existing document
                -0.2,
                -0.2,
                -0.2,
                ...
            ],
            "en": "this is 2nd name"
        }
    }
]

Explanation: Both inference texts “this is 1st name”, “this is 2nd name” mapped with embedding have remained unchanged, so the corresponding embedding values have been copied over, instead of making inference calls

Scenario 2: Document Update with partial change to fields

Update Document
ID: 1
"category": [
   {
        "name": {
            "en": **"this is 2nd name"**
        }
   },
    {
        "name": {
            "en": "this is 2nd name"
        }
    }
]

Existing Document
ID: 1
"category": [
    {
        "name": {
            "category_name_vector": [
                -0.1,
                -0.1,
                -0.1,
                ...
            ],
            "en": "this is 1st name"
        }
        
    },
    {
        "name": {
            "category_name_vector": [
                -0.2,
                -0.2,
                -0.2,
                ...
            ],
            "en": "this is 2nd name"
        }
    }
]

Filtered Process Map 
{
  "category":` `
        {
            "name": {
                "category_name_vector": ["this is 2nd name", "this is 2nd name" ]
            }
        }
}

Filtered Inference List
["this is 2nd name"]

Ingested Document
ID: 1
"category": [
    {
        "name": {
            "category_name_vector" [ // generated via model inference
                -0.2,
                -0.2,
                -0.2,
                ...
            ],
            "en":"this is 2nd name"**
        }
    },
    {
        "name": {
            "category_name_vector": [ // copied from existing document
                -0.2,
                -0.2,
                -0.2,
                ...
            ],
            "en": "this is 2nd name"
        }
    }
]

Explanation: The inference text at index 0 changed from “this is 1st name” to “this is 2nd name”, while the text at index 1 (“this is 2nd name”) remained unchanged. An inference call is made only for the modified text at index 0. Even if the existing document already contains embeddings for “this is 2nd name”, the inference call is still triggered because text changes are evaluated based on their respective indices, not on content duplication.

Scenario 3: Document Update with change to only irrelevant fields

Update Document
ID: 1
"category": [
    {
        "name": {
            "en": "this is 1st name"
        }
    },
    {
        "name": {
            "en": "this is 2nd name"
        }
    }
]
"irrelevant_field": "hello world"

Existing Document
ID: 1
"category": [
    {
        "name": {
            "category_name_vector": [
                -0.1,
                -0.1,
                -0.1,
                ...
            ],
            "en": "this is 1st name"
        }
        
    },
    {
        "name": {
            "category_name_vector": [
                -0.2,
                -0.2,
                -0.2,
                ...
            ],
            "en": "this is 2nd name"
        }
    }
]

Filtered Process Map 
{
  "category":` `
        {
            "name": {
                "category_name_vector": ["this is 1st name", "this is 2nd name"]
            }
        }
}

Filtered Inference List
[]

Ingested Document
"category": [
    {
        "name": {
            "category_name_vector": [ // copied from existing document
                -0.1,
                -0.1,
                -0.1,
                ...
            ],
            "en": "this is 1st name"
        }
        
    },
    {
        "name": {
            "category_name_vector": [ // copied from existing document
                -0.2,
                -0.2,
                -0.2,
                ...
            ],
            "en": "this is 2nd name"
        }
    }
]
**"irrelevant_field": "hello world"**

Explanation: “irrelevant_field” is not mapped with vector embedding, so inference call is not made regardless of the feature. Meanwhile, both inference texts “this is 1st name”, “this is 2nd name” mapped with embedding have remained unchanged, so the corresponding embedding values have been copied over instead of making inference calls.

Scenario 4: Document Update with changes in list values

Update Document
ID: 1
"category": ["category3", "category2"]

Existing Document
ID: 1
"category": ["category1", "category2"]
"category_knn": [
    {
        "knn": [
             -0.1,
             -0.1,
             -0.1,
        ]
    },
    {
        "knn": [
             -0.2,
             -0.2,
             -0.2,
        ]
    }
]

Filtered Process Map 
{
  "category_knn":` ["category3", "category2"]`
}

Filtered Inference List
["category3", "category2"]

Ingested Document
ID: 1
"category": ["category3", "category2"]
"category_knn": [
    {
        **"knn": [ // generated via inference call 
             -0.3,
             -0.3,
             -0.3,
        ]**
    },
    {
        "knn": [ // generated via inference call
             -0.2,
             -0.2,
             -0.2,
        ]
    }
]

Explanation: The optimized text embedding processor compares lists as a whole. If all texts in the list match, the existing embeddings are copied over. If any of the texts doesn't match, inference call is made for all values in the list regardless of matching texts

Testing:
Benchmark Test in progress. Will discuss the results before merge

Related Issues

Resolves #[Issue number to be closed when this PR is merged]

HLD: #1138

Check List

New functionality includes testing.
New functionality has been documented.
API changes companion pull request created.
Commits are signed per the DCO using --signoff.
Public documentation issue/PR created.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

codecov · 2025-02-19T06:41:59Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 81.81%. Comparing base (628cb64) to head (aaf44d3).
Report is 1 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #1191      +/-   ##
============================================
+ Coverage     81.74%   81.81%   +0.06%     
+ Complexity     2511     1300    -1211     
============================================
  Files           190       97      -93     
  Lines          8564     4421    -4143     
  Branches       1436      749     -687     
============================================
- Hits           7001     3617    -3384     
+ Misses         1006      513     -493     
+ Partials        557      291     -266

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

…cessor Signed-off-by: will-hwang <[email protected]>

Signed-off-by: will-hwang <[email protected]>

junqiu-lei · 2025-02-19T23:18:04Z

...ain/java/org/opensearch/neuralsearch/processor/optimization/OptimizedInferenceProcessor.java

+ */
+
+@Log4j2
+public abstract class OptimizedInferenceProcessor extends InferenceProcessor {


I feel the term "optimization" is too generic to use

thanks for pointing this out. Will wait to see if others have better opinions on the naming

I think I agree with @junqiu-lei here. There is a already a InferenceProcessor and we are trying to Optimize what i do already. So better if we name it around what optimization it does upon the existing one.
But if we are not able to come up with something more appropriate then I am good with this one. No strong concerns.

will name as ConditionalTextEmbeddingProcessor to convey that it selectively generates embeddings when conditions are met. Let me know if there are better suggestions

will name it SelectiveTextEmbeddingProcessor for now

junqiu-lei · 2025-02-19T23:24:20Z

src/main/java/org/opensearch/neuralsearch/processor/TextEmbeddingProcessor.java

@@ -26,6 +26,8 @@ public final class TextEmbeddingProcessor extends InferenceProcessor {

    public static final String TYPE = "text_embedding";
    public static final String LIST_TYPE_NESTED_MAP_KEY = "knn";
+    public static final String IGNORE_EXISTING = "ignore_existing";


From user's point, they might not clear what we're ignoring. maybe rename the flag to something like forceRegeneration or alwaysGenerateEmbedding?

ignore prefix was what we preferred based on discussions I had with ML-commons team because it is consistent with other flags that are in use. But open to more suggestions

I like "ignore_existing_embedding" since it's more descriptive. If the default to maintain backwards compatibility is to have optimization disabled, "ignore" makes it clearer that it's a feature they have to turn on to get the benefit (default false). like I wonder if users would be confused if a setting labelled "force" was default true?

Just saw the above comment about the name of the processor too, maybe ideally the processor name and the option name are similar since the processor exists to support the option?

I think we need to make the naming more clear. ignore_existing is a little bit confusing that we actually want to do the optimization when it is true. People can view it as ignore_existing_embedding which means we want to always do the inference work.

I think even we can document how the parameter works it should be self explain and give people a gut feel what does it control. It should convey the info that we will do some optimization when the doc has existing embedding. It's actually kind weird we have to start with ignore. And if starting with ignore can make it confusing we should propose to not do that. I feel even we simply call it as "optimize_embedding_generation" is more descriptive.

what do we think about skipExisting?

That is a good point. How about skip_existing?

junqiu-lei · 2025-02-19T23:29:00Z

src/testFixtures/java/org/opensearch/neuralsearch/BaseNeuralSearchIT.java

Do we have any integration test covered for the new processors?

will include IT tests in this PR

junqiu-lei · 2025-02-19T23:37:06Z

We might need BWC tests to verify mixed-version cluster scenarios where:

Some nodes are running the old version (without optimized processor)
Some nodes are running the new version

q-andy

If I'm understanding correctly, when a user adds a text embedding processor to the pipeline with the config option as true, then we create a different processor object than if the config was false.

Are there any scenarios where this is confusing for users that the processor object we create isn't the exact same one that they define? Or is this all abstracted away?

q-andy · 2025-02-20T18:34:40Z

...java/org/opensearch/neuralsearch/processor/optimization/OptimizedTextEmbeddingProcessor.java

+            int min = Math.min(((List) sourceList.get()).size(), ((List) existingList.get()).size());
+            for (int j = 0; j < min; j++) {
+                if (((List) sourceList.get()).get(j).equals(((List) existingList.get()).get(j))) {
+                    updatedEmbeddings.add(((List) embeddingList.get()).get(j));


Would having a helper method to cast lists would make this more readable? Like private List asList(Object object). Since raw casting requires double parentheses making it hard to parse inline, compare ((List) object) vs asList(object).

will make it more readable. thanks!

You don't need an extra method here. You just need to declare another variable to avoid repetitive casting.

List sList = (List)sourceList.get(); List eList = (List)existsingList.get(); if (sList.get(j).equals(eList.get(j)) { ... }

will-hwang · 2025-02-20T21:39:15Z

If I'm understanding correctly, when a user adds a text embedding processor to the pipeline with the config option as true, then we create a different processor object than if the config was false.

Are there any scenarios where this is confusing for users that the processor object we create isn't the exact same one that they define? Or is this all abstracted away?

the implementation detail should be abstracted away from user, and all user needs to know is the behavior of the flag.

bzhangam · 2025-02-21T18:51:20Z

...java/org/opensearch/neuralsearch/processor/optimization/OptimizedTextEmbeddingProcessor.java

+        Map<String, Object> existingSourceAndMetadataMap,
+        int index
+    ) {
+        String textKey = ProcessorUtils.findKeyFromFromValue(ProcessorDocumentUtils.unflattenJson(fieldMap), currentPath, level);


I notice we do ProcessorDocumentUtils.unflattenJson(fieldMap) multiple times. Can we only do it once?

fixed. unflattened in constructor

vibrantvarun

Completed the 1st round of review. Will wait for comments to be addressed before doing 2nd round of review.

Thanks

CHANGELOG.md

vibrantvarun · 2025-02-21T18:29:05Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

-            return v1;
-        } else {
-            return v2;
+        if (v1 instanceof List<?> && v2 instanceof List<?>) {


Can you Add a comment on what does this peice of code do?

This is a leftover from previous code that lacks comments explaining its functionality. It would be great if you could add a comment to clarify it while making your modifications, helping to address this technical debt.

added comments

vibrantvarun · 2025-02-21T18:29:15Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

        }
+
+        if (v1 instanceof Map<?, ?> && v2 instanceof Map<?, ?>) {


Can you Add a comment on what does this peice of code do?

yes, will add comments for this method

reverted back to previous after updated implementation

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

vibrantvarun · 2025-02-21T18:30:33Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

+        IngestDocument ingestDocument,
+        Map<String, Object> processorMap,
+        List<?> results,
+        boolean update


Suggested change

boolean update

boolean update

Can you come up with a little usecase specific variable name?

Or adding a javadoc explaining each parameter for this method might be good as well.

added javadoc for this

vibrantvarun · 2025-02-21T18:41:27Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

+        }
+    }
+
+    protected void setVectorFieldsToDocument(IngestDocument ingestDocument, Map<String, Object> processorMap, List<?> results) {


When is this method called? I see there are 2 methods with the same name. If we need this method then please change the name of this method to usecase specific.

Or better is to make the earlier method except this flag and put if else

Agree. We can remove this method.

refactored to remove this method

vibrantvarun · 2025-02-21T18:47:12Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

@@ -512,17 +560,18 @@ private void processMapEntryValue(
    ) {
        // build nlp output for object in sourceValue which is map type
        Iterator<Map<String, Object>> iterator = sourceAndMetadataMapValueInList.iterator();
-        IntStream.range(0, sourceAndMetadataMapValueInList.size()).forEach(index -> {
+        IndexWrapper listIndexWrapper = new IndexWrapper(0);


We don't needs this listIndexWrapper object. The argument indexWrapper is same object what you are trying to create here. Please traceback the indexWrapper object and you will find that it is called from buildNLPResult method in this class https://github.com/opensearch-project/neural-search/pull/1191/files#diff-d2b79f65d193c79dd65558833fcf583eb2c29301325e1eb4eb83a963c737f2f8R468. Which has the same declaration. They what is the need of making this change?

indexWrapper is used to iterate the embedding list for update. listIndexWrapper is used to replace the existingIntStream.range(0, sourceAndMetadataMapValueInList.size()). Currently, every update is just another ingest so we overwrite the values and its embeddings every time. With the new way, some values and its embeddings will have been copied over, so we need a way to skip overwriting those values

If I am understanding correctly then we need only listIndexMapper not the indexMapper. As the listIndexMapper have the details of what needs to be updated right?

this change was done to increment the nestedIndex based on the existence of embedding in document, instead of sequentially incrementing like IntStream.range(0, sourceAndMetadataMapValueInList.size()). This change was done to handle recently found edge cases such as #1024. Since this change is only needed only for specific cases and seems like a distraction to overall objective of the PR, I will raise a separate PR for this change.

The IndexWrapper is necessary. We want to increase counter only when we consumed value from inputNestedMapEntryValue so we need a way to keep the counter value. I see the problem is in the class name which is confusing because index in OpenSearch has its own meaning.

updated implementation logic to remove the need for this change.

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

vibrantvarun · 2025-02-21T19:00:45Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

@@ -543,7 +591,7 @@ private void putNLPResultToSingleSourceMapInList(
        List<?> results,
        IndexWrapper indexWrapper,
        Map<String, Object> sourceAndMetadataMap,
-        int nestedElementIndex
+        IndexWrapper listIndexWrapper


indexWrapper and listIndexWrapper are same. Please reevaluate the need of this object here.

vibrantvarun · 2025-02-21T19:02:28Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

@@ -54,15 +55,32 @@ public abstract class InferenceProcessor extends AbstractBatchingProcessor {
    public static final String MODEL_ID_FIELD = "model_id";
    public static final String FIELD_MAP_FIELD = "field_map";
    private static final BiFunction<Object, Object, Object> REMAPPING_FUNCTION = (v1, v2) -> {


Can we make this REMAPPING _Function a method? Loading this method in memory by default is not ideal.

Not sure if making it as method is necessary. In terms of loading into memory, there is no difference between this and method.

bzhangam · 2025-02-21T19:48:42Z

src/main/java/org/opensearch/neuralsearch/processor/factory/TextEmbeddingProcessorFactory.java

-        return new TextEmbeddingProcessor(tag, description, batchSize, modelId, filedMap, clientAccessor, environment, clusterService);
+        Map<String, Object> fieldMap = readMap(TYPE, tag, config, FIELD_MAP_FIELD);
+        boolean ignoreExisting = readBooleanProperty(TYPE, tag, config, IGNORE_EXISTING, DEFAULT_IGNORE_EXISTING);
+        if (ignoreExisting == true) {


When we say ignore_existing does that means we want to always do inference work? Which is our existing behavior so we should use the existing processor rather than the optimized one?

open to suggestions, we can make the feature name more descriptive

bzhangam · 2025-02-21T19:53:22Z

...java/org/opensearch/neuralsearch/processor/optimization/OptimizedTextEmbeddingProcessor.java

+    ) {
+        String index = ingestDocument.getSourceAndMetadata().get(INDEX_FIELD).toString();
+        String id = ingestDocument.getSourceAndMetadata().get(ID_FIELD).toString();
+        openSearchClient.execute(GetAction.INSTANCE, new GetRequest(index, id), ActionListener.wrap(response -> {


In this processor we always do this get call which kind downgrade the performance of the operations we know for sure there is no existing embedding can be reused. We can control this through the ignore_existing flag but just thinking would it better in this optimized processor we also consider if the we are processing a create doc API? I think we can tell this info based on the which API is invoked and attach it to the IngestDocument.

that would require model change in ingestDocument to attach API performed on the ingestDocument. This change should be a separate feature in my opinion, and should be implemented separately if needed

There is no difference between create doc and update doc API.

q-andy · 2025-02-21T19:55:04Z

src/main/java/org/opensearch/neuralsearch/processor/factory/TextEmbeddingProcessorFactory.java

-        return new TextEmbeddingProcessor(tag, description, batchSize, modelId, filedMap, clientAccessor, environment, clusterService);
+        Map<String, Object> fieldMap = readMap(TYPE, tag, config, FIELD_MAP_FIELD);
+        boolean ignoreExisting = readBooleanProperty(TYPE, tag, config, IGNORE_EXISTING, DEFAULT_IGNORE_EXISTING);
+        if (ignoreExisting == true) {


nit: Since ignoreExisting is already a boolean you don't need to check equality to true here, you can use the boolean as the condition

following the developer guide: https://github.com/opensearch-project/OpenSearch/blob/main/DEVELOPER_GUIDE.md#java-language-formatting-guidelines

explicit true/false check is preferred

It is only applicable for false conditions. If(ignoreExisting) by all means mean a true condition.

Therefore, it is recommended to use if(boolvariable) rather than if(boolvariable==true). The end goal is to improve readability. For false conditions it makes sense to write that explicitly.

okay, will change

This is standard format for OpenSearch core where we explicitly compare it to either true or false. Even if we don't force it in neural, would be nice to follow it.

heemin32 · 2025-02-24T00:31:16Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

-        } else {
-            return v2;
+        if (v1 instanceof List<?> && v2 instanceof List<?>) {
+            List<Object> v1List = new ArrayList<>((List<?>) v1);


Can we avoid creating new array here?

heemin32 · 2025-02-24T00:37:14Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

+            List<?> v2List = (List<?>) v2;
+
+            Iterator<?> iterator = v2List.iterator();
+            for (int i = 0; i < v1List.size() && iterator.hasNext(); i++) {


In previous code, we added all item from v2 to v1. Now, we are setting null value in v1 using v2 value. Could you tell how this new logic can cover the previous use case as well?

The assumption here is that, the number of null value in v1List is same as the size of v2. Can we make it obvious from the code? For example we can do like

for (int i = 0; i , v1List.size(); i++) { if (v1List.get(i) == null) { assert iterator.hasNext() == true; v1List.set(i, iterator.next()); } } assert iterator.hasNext() == false;

refactored with comments

heemin32 · 2025-02-24T00:37:52Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

        }
+
+        if (v1 instanceof Map<?, ?> && v2 instanceof Map<?, ?>) {
+            Map<String, Object> v1Map = new LinkedHashMap<>((Map<String, Object>) v1);


Can we avoid creating new map? This might be costly.

We checked if v1 is instance of Map<?, ?> but not we are casting it to Map<String, Object>. This is not safe.

heemin32 · 2025-02-24T00:39:04Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

+            }
+            return v1Map;
+        }
+        return v2 != null ? v2 : v1;


Do we ever get to this case? What is the logic behind this returning v2 and ignoring v1?

heemin32 · 2025-02-24T00:44:32Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

+            Map<?, ?> v2Map = (Map<?, ?>) v2;
+
+            for (Map.Entry<?, ?> entry : v2Map.entrySet()) {
+                if (entry.getKey() instanceof String && !v1Map.containsKey(entry.getKey())) {


Is there a case where v2Map has different types of key? Like one key is string type and other int type?

src/main/java/org/opensearch/neuralsearch/processor/util/ProcessorUtils.java

heemin32 · 2025-02-24T02:00:21Z

src/main/java/org/opensearch/neuralsearch/processor/util/ProcessorUtils.java

+        Map<String, Object> currentMap = sourceAsMap;
+        String targetValue = keys[keys.length - 1];
+        for (String key : keys) {
+            if (key.equals(targetValue)) {


what if path is "level1.level2.level1" and level is 3?
The code will break here and lastFoundKey will be null.

* map: * { * "level1": { * "level2" : { * "first_text": "level1" * } * } * }

fixed, and added a test case

heemin32 · 2025-02-24T02:01:30Z

src/main/java/org/opensearch/neuralsearch/processor/util/ProcessorUtils.java

+            if (currentMap.containsKey(key)) {
+                Object value = currentMap.get(key);
+                if (value instanceof Map) {
+                    currentMap = (Map<String, Object>) value;


This is unsafe cast

src/main/java/org/opensearch/neuralsearch/processor/util/ProcessorUtils.java

Signed-off-by: Will Hwang <[email protected]>

heemin32 · 2025-02-24T16:51:22Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

-            ((Collection) v1).addAll((Collection) v2);
+            Iterator<?> iterator = ((Collection) v2).iterator();
+            for (int i = 0; i < ((Collection) v1).size(); i++) {
+                if (((List) v1).get(i) == null) {


You should not cast Collection to List without type checking.

This get(i) is not efficient for some of the list type.

ListIterator<String> v1Iterator = v1.listIterator(); while (v1Iterator.hasNext()) { if (v1Iterator.next() == null) { v1Iterator.set(v2Iterator.next()); } }

One question is, how is this compatible with previous code logic? The previous and current logic is totally different and I wonder how this new logic can support the previous use case.

@vibrantvarun can confirm, but previously, maps would merge without remapping function, because the key would be inserted for the first time, not requiring remapping.

I mean about List case not about Map case.

.merge still requires a function to be passed in, regardless of whether it's used or not.

Google's answer on .merge():

The merge() method, available in Java's Map interface, operates as a bifunction, taking a key, a value, and a BiFunction as arguments. It either adds a new key-value pair to the map or updates an existing entry based on the provided BiFunction.
If the specified key is not already present in the map, the merge() method simply inserts the key with the given value. However, if the key exists, the BiFunction is applied to the current value associated with the key and the provided value. The result of this function determines the new value for the key. If the BiFunction returns null, the entry for the key is removed from the map. Otherwise, the key is updated with the new value returned by the BiFunction

You can use .put() instead of .merge() if there is no key conflict.

@martin-gaievski @vibrantvarun can you verify my assumption is correct?

actually, this would be invoked when user directly ingests with vector field.

leaving the inferenceProcessor the same

heemin32 · 2025-02-24T17:01:52Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

-        nlpResult.forEach(ingestDocument::setFieldValue);
+        // if partialUpdate is set to false, full update is required where each vector embedding in nlpResult
+        // can directly be populated to ingestDocument
+        if (partialUpdate == false) {


Let's return early to eliminate the else block and reduce nesting?

Suggested change

if (partialUpdate == false) {

if (partialUpdate == false) {

// ..

return;

}

// ...

heemin32 · 2025-02-24T17:06:53Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

@@ -534,7 +590,6 @@ private void processMapEntryValue(
     * @param results
     * @param indexWrapper
     * @param sourceAndMetadataMap
-     * @param nestedElementIndex index of the element in the list field of source document


This should not be removed.

heemin32 · 2025-02-24T17:07:20Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

                );
            }
        } else {
-            if (sourceValue instanceof List && ((List<Object>) sourceValue).get(nestedElementIndex) != null) {
-                sourceAndMetadataMap.merge(processorKey, results.get(indexWrapper.index++), REMAPPING_FUNCTION);
+            if (sourceValue instanceof List) {


If there is no change, can we revert it back to original code?

heemin32 · 2025-02-24T17:15:46Z

src/main/java/org/opensearch/neuralsearch/processor/InferenceProcessor.java

-        IntStream.range(0, sourceValue.size())
-            .forEachOrdered(x -> keyToResult.add(ImmutableMap.of(listTypeNestedMapKey, results.get(indexWrapper.index++))));
+        IntStream.range(0, sourceValue.size()).forEachOrdered(x -> {
+            if (sourceValue.get(x) != null) { // only add to keyToResult when sourceValue.get(x) exists,


I see we use get(i) in the original code which is not safe.

FYI, we should either use iterator or force it to be array list. I think this require changes in the original code as well so I won't push for it in this PR.

simply changing the type fails the integ tests. This will require additional changes

heemin32 · 2025-02-24T18:16:46Z

src/main/java/org/opensearch/neuralsearch/processor/util/ProcessorUtils.java

+
+        if (existingValue instanceof List) {
+            if (index >= 0 && index < ((List) existingValue).size()) {
+                ((List) existingValue).set(index, targetValue);


This is not efficient when existingValue is not array list.

src/main/java/org/opensearch/neuralsearch/processor/util/ProcessorUtils.java

Signed-off-by: Will Hwang <[email protected]>

vibrantvarun

Did 2nd round of review. Will wait for comments to address before doing 3rd round.

vibrantvarun · 2025-02-26T19:26:24Z

src/main/java/org/opensearch/neuralsearch/processor/factory/TextEmbeddingProcessorFactory.java

-        return new TextEmbeddingProcessor(tag, description, batchSize, modelId, filedMap, clientAccessor, environment, clusterService);
+        Map<String, Object> fieldMap = readMap(TYPE, tag, config, FIELD_MAP_FIELD);
+        boolean skipExisting = readBooleanProperty(TYPE, tag, config, SKIP_EXISTING, DEFAULT_SKIP_EXISTING);
+        if (skipExisting == true) {


if (true == true) is not an appropriate way of checking boolean value. Even if OpenSearch core follows it, it only makes sense if we do not have a primitive data type Boolean.

if(skipExisting) is the right way.

This is valid comment. OpenSearch core only does this for false case to avoid !

will address

vibrantvarun · 2025-02-26T19:26:47Z

...ain/java/org/opensearch/neuralsearch/processor/optimization/SelectiveInferenceProcessor.java

+        this.reversedFieldMap = ProcessorDocumentUtils.flattenAndFlip(fieldMap);
+    }
+
+    public abstract Object processValue(


Add javadoc to all public methods

vibrantvarun · 2025-02-26T19:27:44Z

...ain/java/org/opensearch/neuralsearch/processor/optimization/SelectiveInferenceProcessor.java

+        Map<String, Object> existingSourceAndMetadataMap,
+        Map<String, Object> sourceAndMetadataMap,
+        Map<String, Object> processMap,
+        String prevPath


Can you think of a better name here?

vibrantvarun · 2025-02-26T19:27:53Z

...ain/java/org/opensearch/neuralsearch/processor/optimization/SelectiveInferenceProcessor.java

+                Map<String, Object> filteredInnerMap = filterProcessMap(
+                    existingSourceAndMetadataMap,
+                    sourceAndMetadataMap,
+                    (Map<String, Object>) value,


vibrantvarun · 2025-02-26T19:29:41Z

...ain/java/org/opensearch/neuralsearch/processor/optimization/SelectiveInferenceProcessor.java

+     * @param existingSourceAndMetadataMap SourceAndMetadataMap of existing Document
+     * @param sourceAndMetadataMap SourceAndMetadataMap of ingestDocument Document
+     * @param processMap The current processMap
+     * @param prevPath The dot-notation path of the parent elements


Add a small example how does that dot-notation path look like

vibrantvarun · 2025-02-26T19:33:58Z

...ain/java/org/opensearch/neuralsearch/processor/optimization/SelectiveInferenceProcessor.java

+        Map<String, Object> existingSourceAndMetadataMap
+    ) {
+        String textKey = reversedFieldMap.get(embeddingKey);
+        if (textKey == null) {


nit: use Objects.isNull

vibrantvarun · 2025-02-26T19:34:23Z

...java/org/opensearch/neuralsearch/processor/optimization/SelectiveTextEmbeddingProcessor.java

+    }
+
+    @Override
+    public void doExecute(


Add javadoc to all public methods

vibrantvarun · 2025-02-26T19:35:44Z

src/main/java/org/opensearch/neuralsearch/processor/util/ProcessorUtils.java

+     *                    -1 is passed in
+     */
+
+    public static void setValueToSource(Map<String, Object> sourceAsMap, String targetKey, Object targetValue, int index) {


There is a lot happening in this method. Please break it down. Validations, function.

vibrantvarun · 2025-02-26T19:37:46Z

src/main/java/org/opensearch/neuralsearch/util/ProcessorDocumentUtils.java

+     */
+    public static Map<String, String> flattenAndFlip(Map<String, Object> map) {
+        Map<String, String> flippedMap = new HashMap<>();
+        flattenAndFlip("", map, flippedMap);


Can you make return type of flattenAndFlip method as Map<String, String> ? This way you need not have to declare it in this method. Instead add it in flattenAndFlip method and it will return it.

i'm constructing a new string in flattenAndFlip, so it's convenient for its helper method to be void.

vibrantvarun · 2025-02-26T19:39:04Z

src/main/java/org/opensearch/neuralsearch/util/ProcessorDocumentUtils.java

+            Object value = entry.getValue();
+            String newKey = prefix.isEmpty() ? key : prefix + "." + key;
+            if (value instanceof Map) {
+                flattenAndFlip(newKey, (Map<String, Object>) value, flippedMap);


You have used recursion here. Can you evaluate on improving the worst case Time Complexity?

time complexity would be O(number of entries in map) regardless of recursion or not. We would need to traverse every entry once. This is common all across other methods, since we handle complex nested structures.

will-hwang and others added 3 commits February 18, 2025 23:18

implement single doc update scenario for optimized text embedding pro…

39d7600

…cessor Signed-off-by: will-hwang <[email protected]>

fix indexing in inference processor

9f80f9a

Signed-off-by: will-hwang <[email protected]>

add change log

d68dca2

Signed-off-by: will-hwang <[email protected]>

will-hwang force-pushed the optimized_text_embedding_processor_single_doc_update branch from 9264760 to d68dca2 Compare February 19, 2025 07:18

will-hwang marked this pull request as ready for review February 19, 2025 07:21

will-hwang requested review from heemin32, navneet1v, VijayanB, vamshin, jmazanec15, naveentatikonda, junqiu-lei, martin-gaievski, sean-zheng-amazon, model-collapse, zane-neo, vibrantvarun, zhichao-aws, yuye-aws and minalsha as code owners February 19, 2025 07:21

junqiu-lei reviewed Feb 19, 2025

View reviewed changes

q-andy reviewed Feb 20, 2025

View reviewed changes

bzhangam reviewed Feb 21, 2025

View reviewed changes

vibrantvarun reviewed Feb 21, 2025

View reviewed changes

bzhangam reviewed Feb 21, 2025

View reviewed changes

q-andy reviewed Feb 21, 2025

View reviewed changes

heemin32 reviewed Feb 24, 2025

View reviewed changes

refactor with integration tests

3b7769e

Signed-off-by: Will Hwang <[email protected]>

will-hwang force-pushed the optimized_text_embedding_processor_single_doc_update branch from 52b6dbc to 3b7769e Compare February 24, 2025 17:11

heemin32 reviewed Feb 24, 2025

View reviewed changes

refactor code to handle embedding population more effectively

bf7e35f

Signed-off-by: Will Hwang <[email protected]>

will-hwang force-pushed the optimized_text_embedding_processor_single_doc_update branch from c1f9d45 to bf7e35f Compare February 25, 2025 04:57

will-hwang added 2 commits February 24, 2025 21:28

revert baack to List from ArrayList

5df31d3

Signed-off-by: Will Hwang <[email protected]>

change list comparison logic

aaf44d3

Signed-off-by: Will Hwang <[email protected]>

vibrantvarun reviewed Feb 26, 2025

View reviewed changes

Optimized text embedding processor for single document update #1191

Are you sure you want to change the base?

Optimized text embedding processor for single document update #1191

Conversation

will-hwang commented Feb 19, 2025 • edited Loading

Description

Proposed State [Single Document Update]

User Scenarios

Related Issues

Check List

codecov bot commented Feb 19, 2025 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

junqiu-lei commented Feb 19, 2025

q-andy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

will-hwang commented Feb 20, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vibrantvarun left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

heemin32 Feb 21, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

will-hwang commented Feb 19, 2025 •

edited

Loading

codecov bot commented Feb 19, 2025 •

edited

Loading

heemin32 Feb 21, 2025 •

edited

Loading