Fix flaky test MoreExpressionIT.testSpecialValueVariable in concurrent search path

[sohami]

Description

fix the flaky test org.opensearch.script.expression.MoreExpressionIT.testSpecialValueVariable {p0={"search.concurrent_segment_search.enabled":"true"}}

Related Issues

Resolves #10079

Check List

• New functionality includes testing.
  • All tests pass
    [ ] New functionality has been documented.
  • New functionality has javadoc added
• Failing checks are inspected and point to the corresponding known issue(s) (See: Troubleshooting Failing Builds)
• Commits are signed per the DCO using --signoff
• Commit changes are listed out in CHANGELOG.md file (See: Changelog)
• 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.

[github-actions]

Gradle Check (Jenkins) Run Completed with:

• RESULT: FAILURE ❌
• URL: https://build.ci.opensearch.org/job/gradle-check/29533/
• CommitID: 3a43c84
  Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green.
  Is the failure a flaky test unrelated to your change?

[github-actions]

Compatibility status:

Checks if related components are compatible with change f1b39cd

Incompatible components

Incompatible components: [https://github.com/opensearch-project/performance-analyzer.git]

Skipped components

Compatible components

Compatible components: [https://github.com/opensearch-project/security-analytics.git, https://github.com/opensearch-project/security.git, https://github.com/opensearch-project/custom-codecs.git, https://github.com/opensearch-project/opensearch-oci-object-storage.git, https://github.com/opensearch-project/index-management.git, https://github.com/opensearch-project/geospatial.git, https://github.com/opensearch-project/sql.git, https://github.com/opensearch-project/notifications.git, https://github.com/opensearch-project/job-scheduler.git, https://github.com/opensearch-project/observability.git, https://github.com/opensearch-project/neural-search.git, https://github.com/opensearch-project/k-nn.git, https://github.com/opensearch-project/cross-cluster-replication.git, https://github.com/opensearch-project/alerting.git, https://github.com/opensearch-project/anomaly-detection.git, https://github.com/opensearch-project/asynchronous-search.git, https://github.com/opensearch-project/performance-analyzer-rca.git, https://github.com/opensearch-project/ml-commons.git, https://github.com/opensearch-project/common-utils.git, https://github.com/opensearch-project/reporting.git]

[github-actions]

Gradle Check (Jenkins) Run Completed with:

• RESULT: FAILURE ❌
• URL: https://build.ci.opensearch.org/job/gradle-check/29540/
• CommitID: 3c9e07b
  Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green.
  Is the failure a flaky test unrelated to your change?

[github-actions]

Gradle Check (Jenkins) Run Completed with:

• RESULT: UNSTABLE ❕
• TEST FAILURES:

      1 org.opensearch.remotestore.SegmentReplicationUsingRemoteStoreIT.testRestartPrimary_NoReplicas

• URL: https://build.ci.opensearch.org/job/gradle-check/29543/
• CommitID: 3c9e07b
  Please review all flaky tests that succeeded after retry and create an issue if one does not already exist to track the flaky failure.

[codecov]

Codecov Report

Attention: 16 lines in your changes are missing coverage. Please review.

Comparison is base (3ff7e97) 71.15% compared to head (f1b39cd) 71.22%.
Report is 2 commits behind head on main.

Files	Patch %	Lines
...on/PerThreadReplaceableConstDoubleValueSource.java	0.00%	11 Missing ⚠️
...arch/script/expression/ExpressionScriptEngine.java	0.00%	4 Missing ⚠️
...script/expression/ExpressionAggregationScript.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main   #11088      +/-   ##
============================================
+ Coverage     71.15%   71.22%   +0.06%     
- Complexity    58783    58829      +46     
============================================
  Files          4885     4885              
  Lines        277199   277205       +6     
  Branches      40285    40286       +1     
============================================
+ Hits         197247   197440     +193     
+ Misses        63448    63258     -190     
- Partials      16504    16507       +3     

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

[reta]

My understanding of the concurrent hash map is that all read operations are non-blocking

I think the non-blocking is not really what I meant (and it does not apply here to be fair), the get operation uses volatile read, but since we not using just simple get but computeIfAbsent, it could also use lock for segment.

2. In ExpressionAggregationScript::newInstance which is called as a part of Aggregator::getLeafCollector, so it is called on each slice.

The usage of ConcurrentHashMap to me looks like patching - we are trying to fix the data structures that were not designed to be shareable to be as such. Changing the way data structures are used is 100% more robust and maintainable option, I would definitely investigate this route, thanks @jed326

[sohami]

My understanding of the concurrent hash map is that all read operations are non-blocking

I think the non-blocking is not really what I meant (and it does not apply here to be fair), the get operation uses volatile read, but since we not using just simple get but computeIfAbsent, it could also use lock for segment.

2. In ExpressionAggregationScript::newInstance which is called as a part of Aggregator::getLeafCollector, so it is called on each slice.

The usage of ConcurrentHashMap to me looks like patching - we are trying to fix the data structures that were not designed to be shareable to be as such. Changing the way data structures are used is 100% more robust and maintainable option, I would definitely investigate this route, thanks @jed326

@reta @jed326 Thanks for the discussion here. Sorry for my delayed response (recovering from cold). So the other option which you are mentioning was also explored but I didn't found a good way to handle it. The crux of the problem is this special ValueSource used in this test for _value field is shared between the ExpressionAggregationScript and ExpressionValueSource which works independent of each other. The ExpressionAggregationScript uses it to cache the next document value for the leaf via setNextAggregationValue whereas ExpressionValueSource (a lucene class) which expects the passed in value source to handle per leaf level values. So this special ValueSource ReplaceableConstDoubleValueSource needs to be leaf aware so it can correctly cache the values for each leaf (which is what the current PR is trying to achieve). Let me give some more insights here:

1. When the AggregationFactories are setup as part of query parsing, then it constructs the ValueSourceConfig (defines where Aggs will gets the value to operate on). During the ValueSourceConfig construction if there is a usage of script then that is also evaluated and the fields used in it is also associated with its ValueSource Ref here. I think of ValueSource as a producer of the values needed by aggs. Looking into most of the ValueSource, it handles providing the values at each leaf level by constructing different values object. There can be different use cases such as i) reading a field data or ii) computing something via scripts on a field value, etc. and depending on use case it will backed by a FieldDataCache or primitive ones such as double, bytes, numeric, etc with or without scripts form of ValueSource implementation. So ideally, the ValueSource implementation should be able to handle the per leaf processing in thread safe manner. For context, FieldDataValueSource relies on the global field data cache to load values of each segment.
2. The ValueSource created for this special parameter field in script is however not creating separate Values object across leaves (that is what this PR is trying to solve). The above createScript method in ValueSourceConfig will eventually call ExpressionScriptEngine::newAggregationScript where for a special _value type field it creates this ReplaceableConstDoubleValueSource. This value source object is passed into the SimpleBindings and to the ExpressionAggregationScript object (which doesn't have notion of the segments yet). Ref here
3. The ExpressionAggregationScript creates this source member using bindings (which has reference to same ReplaceableConstDoubleValueSource as passed in the constructor). Ref here. Now for each leaf this source is called to create a leaf specific DoubleValues here. And during collection it again use the same shared ValueSource directly via its member specialValue to set a value here.

Regarding the inefficiency of the ConcurrentHashMap, the computeIfAbsent will add lock in cases when there is no value associated with a key and uses CAS semantics for it. In other cases, it fetches the value via lock free mechanism. For this use-case given we are creating 1 DoubleValue object per slice (not per segment), it will only result in contention for the very first access by all the slice threads (that will also be quick as it will just instantiate the value object and return). Once the instance is created all the access will be lock free.

[jed326]

@sohami thanks for the explanation! Originally I was thinking since ExpressionAggregationScript::newInstance gets called on each leaf it could be possible to initialize the ReplaceableConstDoubleValueSource in that method call instead but it sounds like that’s tricky to do because it’s also created using the bindings -- I did not see ExpressionValueSource component of this previously so I see how that can make this tricky.

Taking a step back though I do see the bigger picture in this

I think of ValueSource as a producer of the values needed by aggs. Looking into most of the ValueSource, it handles providing the values at each leaf level by constructing different values object.

and I agree with the conclusion that this specific type of ValueSource is not following the same pattern.

[github-actions]

❕ Gradle check result for f1b39cd: UNSTABLE

• TEST FAILURES:

      1 org.opensearch.repositories.azure.AzureBlobContainerRetriesTests.testWriteBlobWithRetries

Please review all flaky tests that succeeded after retry and create an issue if one does not already exist to track the flaky failure.

[reta]

Thanks a lot for exploring different routes, @sohami

So this special ValueSource ReplaceableConstDoubleValueSource needs to be leaf aware so it can correctly cache the values for each leaf (which is what the current PR is trying to achieve).

Aha, I would argue this is not what this change is doing: why don't we make ReplaceableConstDoubleValueSource indeed leaf aware (using LeafReaderContext) instead of being thread aware?

[sohami]

Thanks a lot for exploring different routes, @sohami

So this special ValueSource ReplaceableConstDoubleValueSource needs to be leaf aware so it can correctly cache the values for each leaf (which is what the current PR is trying to achieve).

Aha, I would argue this is not what this change is doing: why don't we make ReplaceableConstDoubleValueSource indeed leaf aware (using LeafReaderContext) instead of being thread aware?

@reta Thread aware is better than leaf aware since it has to initialize and create ReplaceableConstDoubleValues only once per thread compared to once per leaf. Also for non-concurrent path, it is keeping the existing behavior of 1 ReplaceableConstDoubleValues instance as compared to 1 per leaf.

[sohami]

@reta Checking back to see if we can get this merged in ?

[reta]

@reta Thread aware is better than leaf aware since it has to initialize and create ReplaceableConstDoubleValues only once per thread compared to once per leaf. Also for non-concurrent path, it is keeping the existing behavior of 1 ReplaceableConstDoubleValues instance as compared to 1 per leaf.

@sohami thread is a low level implementation detail here where leaf is not - it has very precise meaning and place in the algorithm (concurrent search).

[reta]

@reta Checking back to see if we can get this merged in ?

Your call folks, I have not changed my opinion about this change (== patching), but if you feel it is way to go, I won't object

[sohami]

@reta Thread aware is better than leaf aware since it has to initialize and create ReplaceableConstDoubleValues only once per thread compared to once per leaf. Also for non-concurrent path, it is keeping the existing behavior of 1 ReplaceableConstDoubleValues instance as compared to 1 per leaf.

@sohami thread is a low level implementation detail here where leaf is not - it has very precise meaning and place in the algorithm (concurrent search).

@reta I understand about the low level implementation detail but that is something we have considered in other places too for concurrent search flow that a slice is mapped to a thread. Using that if there is scope of improvement which in this case is by re-using the Value object for a slice thread then I feel it should still be fine. The class name also reflects the intent of per thread for this specific use case.

[reta]

@reta I understand about the low level implementation detail but that is something we have considered in other places too for concurrent search flow that a slice is mapped to a thread

@sohami I don't like it either, I think we took shortcuts and now taking even more shortcuts using previous ones as justification.

[sohami]

@reta I understand about the low level implementation detail but that is something we have considered in other places too for concurrent search flow that a slice is mapped to a thread

@sohami I don't like it either, I think we took shortcuts and now taking even more shortcuts using previous ones as justification.

@reta I am not sure what you are referring to as shortcut. It is an invariant which we are using for concurrent flow. We tried to use the approach which we can come up with and think is the best known at any given time and considering performance impact in mind as well. If you feel there are better ways to implement, please let us know, we can definitely evaluate and take it up.

For this specific change, I have cited 2 reasons on why I am pushing for threadId based implementation compared to leaf based ValueSource which IMO is fair. Would like to hear thoughts from others as well @andrross @jed326 so that we can find the best path forward here.

[andrross]

thread is a low level implementation detail here where leaf is not - it has very precise meaning and place in the algorithm (concurrent search).

This is definitely the crux of the issue. The concurrent map + caching by thread ID is tying the data structure implementation to the low level details of how search is being executed by the platform. I agree with @reta that changing the data structures to be thread safe using well defined domain constructs (e.g. leaf) is more maintainable and extensible. @sohami makes good points about the performance considerations (caching per-thread creates fewer instances and doesn't change the way non-concurrent search works), though having concrete benchmarks would help show how significant those points are in practice.

I feel the same way as @reta here...this solution isn't perfect but we generally want progress not perfection so I won't block this. The bigger concern is that if the "one slice is always executed on one thread" invariant continues to be baked into many places throughout the code then we may end up in a bad spot down the road (like, for example, we find a way to use virtual threads that breaks that invariant).

[jed326]

I feel the same way as @reta here...this solution isn't perfect but we generally want progress not perfection so I won't block this. The bigger concern is that if the "one slice is always executed on one thread" invariant continues to be baked into many places throughout the code then we may end up in a bad spot down the road (like, for example, we find a way to use virtual threads that breaks that invariant).

Today this "one slice is always executed on one thread" invariant is essentially enforced by the Executor we provide to Lucene and at least for now I don't see a way that the Executor interface can break this invariant (either multiple slices in the same thread at the same time or a single slice on multiple threads at the same time).

Maybe in general it's better to not to rely on the thread id itself, especially in cases where there are data structures that shouldn't be re-used in the same thread across thread executions, but in the specific concurrent search case because these data structures are specific to a single request/query and we have the slice -> thread invariant I don't think this usage poses a problem. We have done similar in other cases for concurrent search (for example #10352) but in this specific case it seems like using the thread id is actually an optimization over the per-leaf data structure and perhaps more importantly doesn't change how the existing non-concurrent search path functions. Given that, in my opinion the solution provided in this PR looks like the best way forward for now.

[sohami]

@andrross Thanks for your inputs.

This is definitely the crux of the issue. The concurrent map + caching by thread ID is tying the data structure implementation to the low level details of how search is being executed by the platform

Caching by leaf is also relying on how the search is executed such that 1 leaf is processed by only single thread at a time. If that changes the leaf based mechanism will break in that case. My point is how search is executed is baked in the implementation in one form or other.

. @sohami makes good points about the performance considerations (caching per-thread creates fewer instances and doesn't change the way non-concurrent search works), though having concrete benchmarks would help show how significant those points are in practice.

I don't think changes like this will show much difference while measuring the performance via benchmarks. It is about avoiding creation of unnecessary objects. In theory, comparing 2 approaches, I was seeing if we can optimize then why not. For keeping code maintainable, we are adding the reasons in java doc as to why threadId is chosen.

The bigger concern is that if the "one slice is always executed on one thread" invariant continues to be baked into many places throughout the code then we may end up in a bad spot down the road (like, for example, we find a way to use virtual threads that breaks that invariant).

I don't see that we can break this invariant, as it is fundamental to all the collectors implementation. At a slice level there are different collector instances which are created and all the data structures being used inside the collector implementation are not thread safe.

I will keep it as is for now as there are no strong objections. Thanks a lot for all the inputs and discussion.