Skip to content

Commit 841156b

Browse files
peteralfonsiPeter Alfonsi
authored and
jainankitk
committed
Remove unnecessary looping in field data cache clear (opensearch-project#19116)
--------- Signed-off-by: Peter Alfonsi <[email protected]> Signed-off-by: Peter Alfonsi <[email protected]> Co-authored-by: Peter Alfonsi <[email protected]> Signed-off-by: Ankit Jain <[email protected]>
1 parent 9e768ab commit 841156b

File tree

12 files changed

+465
-100
lines changed

12 files changed

+465
-100
lines changed

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
5656
- Fix the `scaled_float` precision issue ([#19188](https://github.com/opensearch-project/OpenSearch/pull/19188))
5757
- Fix Using an excessively large reindex slice can lead to a JVM OutOfMemoryError on coordinator.([#18964](https://github.com/opensearch-project/OpenSearch/pull/18964))
5858
- [Flaky Test] Fix flaky test in SecureReactorNetty4HttpServerTransportTests with reproducible seed ([#19327](https://github.com/opensearch-project/OpenSearch/pull/19327))
59+
- Remove unnecessary looping in field data cache clear ([#19116](https://github.com/opensearch-project/OpenSearch/pull/19116))
5960

6061

6162
### Dependencies

‎server/src/internalClusterTest/java/org/opensearch/index/fielddata/FieldDataLoadingIT.java‎

Lines changed: 120 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,14 +33,22 @@
3333
package org.opensearch.index.fielddata;
3434

3535
import org.opensearch.action.admin.cluster.stats.ClusterStatsResponse;
36+
import org.opensearch.action.admin.indices.cache.clear.ClearIndicesCacheRequest;
37+
import org.opensearch.core.xcontent.XContentBuilder;
38+
import org.opensearch.index.query.MatchAllQueryBuilder;
39+
import org.opensearch.search.sort.SortOrder;
3640
import org.opensearch.test.OpenSearchIntegTestCase;
3741

42+
import java.util.HashMap;
43+
import java.util.Map;
44+
import java.util.concurrent.CountDownLatch;
45+
import java.util.concurrent.Phaser;
46+
3847
import static org.opensearch.common.xcontent.XContentFactory.jsonBuilder;
3948
import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
4049
import static org.hamcrest.Matchers.greaterThan;
4150

4251
public class FieldDataLoadingIT extends OpenSearchIntegTestCase {
43-
4452
public void testEagerGlobalOrdinalsFieldDataLoading() throws Exception {
4553
assertAcked(
4654
prepareCreate("test").setMapping(
@@ -62,6 +70,117 @@ public void testEagerGlobalOrdinalsFieldDataLoading() throws Exception {
6270

6371
ClusterStatsResponse response = client().admin().cluster().prepareClusterStats().get();
6472
assertThat(response.getIndicesStats().getFieldData().getMemorySizeInBytes(), greaterThan(0L));
73+
74+
// Ensure cache cleared before other tests in the suite begin
75+
client().admin().indices().clearCache(new ClearIndicesCacheRequest().fieldDataCache(true)).actionGet();
76+
assertBusy(() -> {
77+
ClusterStatsResponse clearedResponse = client().admin().cluster().prepareClusterStats().get();
78+
assertEquals(0, clearedResponse.getIndicesStats().getFieldData().getMemorySizeInBytes());
79+
});
80+
}
81+
82+
public void testFieldDataCacheClearConcurrentIndices() throws Exception {
83+
// Check concurrently clearing multiple indices from the FD cache correctly removes all expected keys.
84+
int numIndices = 10;
85+
String indexPrefix = "test";
86+
createAndSearchIndices(numIndices, 1, indexPrefix, "field");
87+
// TODO: Should be 1 entry per field per index in cache, but cannot check this directly until we add the items count stat in a
88+
// future PR
89+
90+
// Concurrently clear multiple indices from FD cache
91+
Thread[] threads = new Thread[numIndices];
92+
Phaser phaser = new Phaser(numIndices + 1);
93+
CountDownLatch countDownLatch = new CountDownLatch(numIndices);
94+
95+
for (int i = 0; i < numIndices; i++) {
96+
int finalI = i;
97+
threads[i] = new Thread(() -> {
98+
try {
99+
ClearIndicesCacheRequest clearCacheRequest = new ClearIndicesCacheRequest().fieldDataCache(true)
100+
.indices(indexPrefix + finalI);
101+
client().admin().indices().clearCache(clearCacheRequest).actionGet();
102+
phaser.arriveAndAwaitAdvance();
103+
} catch (Exception e) {
104+
throw new RuntimeException(e);
105+
}
106+
countDownLatch.countDown();
107+
});
108+
threads[i].start();
109+
}
110+
phaser.arriveAndAwaitAdvance();
111+
countDownLatch.await();
112+
113+
// Cache size should be 0
114+
assertBusy(() -> {
115+
ClusterStatsResponse response = client().admin().cluster().prepareClusterStats().get();
116+
assertEquals(0, response.getIndicesStats().getFieldData().getMemorySizeInBytes());
117+
});
65118
}
66119

120+
public void testFieldDataCacheClearConcurrentFields() throws Exception {
121+
// Check concurrently clearing multiple indices + fields from the FD cache correctly removes all expected keys.
122+
int numIndices = 10;
123+
int numFieldsPerIndex = 5;
124+
String indexPrefix = "test";
125+
String fieldPrefix = "field";
126+
createAndSearchIndices(numIndices, numFieldsPerIndex, indexPrefix, fieldPrefix);
127+
128+
// Concurrently clear multiple indices+fields from FD cache
129+
Thread[] threads = new Thread[numIndices * numFieldsPerIndex];
130+
Phaser phaser = new Phaser(numIndices * numFieldsPerIndex + 1);
131+
CountDownLatch countDownLatch = new CountDownLatch(numIndices * numFieldsPerIndex);
132+
133+
for (int i = 0; i < numIndices; i++) {
134+
int finalI = i;
135+
for (int j = 0; j < numFieldsPerIndex; j++) {
136+
int finalJ = j;
137+
threads[i * numFieldsPerIndex + j] = new Thread(() -> {
138+
try {
139+
ClearIndicesCacheRequest clearCacheRequest = new ClearIndicesCacheRequest().fieldDataCache(true)
140+
.indices(indexPrefix + finalI)
141+
.fields(fieldPrefix + finalJ);
142+
client().admin().indices().clearCache(clearCacheRequest).actionGet();
143+
phaser.arriveAndAwaitAdvance();
144+
} catch (Exception e) {
145+
throw new RuntimeException(e);
146+
}
147+
countDownLatch.countDown();
148+
});
149+
threads[i * numFieldsPerIndex + j].start();
150+
}
151+
}
152+
phaser.arriveAndAwaitAdvance();
153+
countDownLatch.await();
154+
155+
// Cache size should be 0
156+
assertBusy(() -> {
157+
ClusterStatsResponse response = client().admin().cluster().prepareClusterStats().get();
158+
assertEquals(0, response.getIndicesStats().getFieldData().getMemorySizeInBytes());
159+
});
160+
}
161+
162+
private void createAndSearchIndices(int numIndices, int numFieldsPerIndex, String indexPrefix, String fieldPrefix) throws Exception {
163+
for (int i = 0; i < numIndices; i++) {
164+
String index = indexPrefix + i;
165+
XContentBuilder req = jsonBuilder().startObject().startObject("properties");
166+
for (int j = 0; j < numFieldsPerIndex; j++) {
167+
req.startObject(fieldPrefix + j).field("type", "text").field("fielddata", true).endObject();
168+
}
169+
req.endObject().endObject();
170+
assertAcked(prepareCreate(index).setMapping(req));
171+
Map<String, String> source = new HashMap<>();
172+
for (int j = 0; j < numFieldsPerIndex; j++) {
173+
source.put(fieldPrefix + j, "value");
174+
}
175+
client().prepareIndex(index).setId("1").setSource(source).get();
176+
client().admin().indices().prepareRefresh(index).get();
177+
// Search on each index to fill the cache
178+
for (int j = 0; j < numFieldsPerIndex; j++) {
179+
client().prepareSearch(index).setQuery(new MatchAllQueryBuilder()).addSort(fieldPrefix + j, SortOrder.ASC).get();
180+
}
181+
}
182+
ensureGreen();
183+
ClusterStatsResponse response = client().admin().cluster().prepareClusterStats().get();
184+
assertTrue(response.getIndicesStats().getFieldData().getMemorySizeInBytes() > 0L);
185+
}
67186
}

‎server/src/internalClusterTest/java/org/opensearch/indices/stats/IndexStatsIT.java‎

Lines changed: 49 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -165,7 +165,7 @@ private Settings.Builder settingsBuilder() {
165165
return Settings.builder().put(indexSettings());
166166
}
167167

168-
public void testFieldDataStats() throws InterruptedException {
168+
public void testFieldDataStats() throws Exception {
169169
assertAcked(
170170
client().admin()
171171
.indices()
@@ -274,18 +274,30 @@ public void testFieldDataStats() throws InterruptedException {
274274
);
275275

276276
client().admin().indices().prepareClearCache().setFieldDataCache(true).execute().actionGet();
277-
nodesStats = client().admin().cluster().prepareNodesStats("data:true").setIndices(true).execute().actionGet();
278-
assertThat(
279-
nodesStats.getNodes().get(0).getIndices().getFieldData().getMemorySizeInBytes() + nodesStats.getNodes()
280-
.get(1)
281-
.getIndices()
282-
.getFieldData()
283-
.getMemorySizeInBytes(),
284-
equalTo(0L)
285-
);
286-
indicesStats = client().admin().indices().prepareStats("test").clear().setFieldData(true).execute().actionGet();
287-
assertThat(indicesStats.getTotal().getFieldData().getMemorySizeInBytes(), equalTo(0L));
288-
277+
assertBusy(() -> {
278+
NodesStatsResponse postClearNodesStats = client().admin()
279+
.cluster()
280+
.prepareNodesStats("data:true")
281+
.setIndices(true)
282+
.execute()
283+
.actionGet();
284+
assertThat(
285+
postClearNodesStats.getNodes().get(0).getIndices().getFieldData().getMemorySizeInBytes() + postClearNodesStats.getNodes()
286+
.get(1)
287+
.getIndices()
288+
.getFieldData()
289+
.getMemorySizeInBytes(),
290+
equalTo(0L)
291+
);
292+
IndicesStatsResponse postClearIndicesStats = client().admin()
293+
.indices()
294+
.prepareStats("test")
295+
.clear()
296+
.setFieldData(true)
297+
.execute()
298+
.actionGet();
299+
assertThat(postClearIndicesStats.getTotal().getFieldData().getMemorySizeInBytes(), equalTo(0L));
300+
});
289301
}
290302

291303
public void testClearAllCaches() throws Exception {
@@ -369,24 +381,30 @@ public void testClearAllCaches() throws Exception {
369381

370382
client().admin().indices().prepareClearCache().execute().actionGet();
371383
Thread.sleep(100); // Make sure the filter cache entries have been removed...
372-
nodesStats = client().admin().cluster().prepareNodesStats("data:true").setIndices(true).execute().actionGet();
373-
assertThat(
374-
nodesStats.getNodes().get(0).getIndices().getFieldData().getMemorySizeInBytes() + nodesStats.getNodes()
375-
.get(1)
376-
.getIndices()
377-
.getFieldData()
378-
.getMemorySizeInBytes(),
379-
equalTo(0L)
380-
);
381-
assertThat(
382-
nodesStats.getNodes().get(0).getIndices().getQueryCache().getMemorySizeInBytes() + nodesStats.getNodes()
383-
.get(1)
384-
.getIndices()
385-
.getQueryCache()
386-
.getMemorySizeInBytes(),
387-
equalTo(0L)
388-
);
389-
384+
assertBusy(() -> {
385+
NodesStatsResponse postClearNodesStats = client().admin()
386+
.cluster()
387+
.prepareNodesStats("data:true")
388+
.setIndices(true)
389+
.execute()
390+
.actionGet();
391+
assertThat(
392+
postClearNodesStats.getNodes().get(0).getIndices().getFieldData().getMemorySizeInBytes() + postClearNodesStats.getNodes()
393+
.get(1)
394+
.getIndices()
395+
.getFieldData()
396+
.getMemorySizeInBytes(),
397+
equalTo(0L)
398+
);
399+
assertThat(
400+
postClearNodesStats.getNodes().get(0).getIndices().getQueryCache().getMemorySizeInBytes() + postClearNodesStats.getNodes()
401+
.get(1)
402+
.getIndices()
403+
.getQueryCache()
404+
.getMemorySizeInBytes(),
405+
equalTo(0L)
406+
);
407+
});
390408
indicesStats = client().admin().indices().prepareStats("test").clear().setFieldData(true).setQueryCache(true).execute().actionGet();
391409
assertThat(indicesStats.getTotal().getFieldData().getMemorySizeInBytes(), equalTo(0L));
392410
assertThat(indicesStats.getTotal().getQueryCache().getMemorySizeInBytes(), equalTo(0L));

‎server/src/main/java/org/opensearch/index/fielddata/IndexFieldDataService.java‎

Lines changed: 6 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@
3333
package org.opensearch.index.fielddata;
3434

3535
import org.apache.lucene.util.Accountable;
36-
import org.opensearch.ExceptionsHelper;
3736
import org.opensearch.common.settings.Setting;
3837
import org.opensearch.common.settings.Setting.Property;
3938
import org.opensearch.core.index.shard.ShardId;
@@ -48,10 +47,7 @@
4847

4948
import java.io.Closeable;
5049
import java.io.IOException;
51-
import java.util.ArrayList;
52-
import java.util.Collection;
5350
import java.util.HashMap;
54-
import java.util.List;
5551
import java.util.Map;
5652
import java.util.function.Supplier;
5753

@@ -110,30 +106,15 @@ public IndexFieldDataService(
110106
}
111107

112108
public synchronized void clear() {
113-
List<Exception> exceptions = new ArrayList<>(0);
114-
final Collection<IndexFieldDataCache> fieldDataCacheValues = fieldDataCaches.values();
115-
for (IndexFieldDataCache cache : fieldDataCacheValues) {
116-
try {
117-
cache.clear();
118-
} catch (Exception e) {
119-
exceptions.add(e);
120-
}
121-
}
122-
fieldDataCacheValues.clear();
123-
ExceptionsHelper.maybeThrowRuntimeAndSuppress(exceptions);
109+
// Since IndexFieldDataCache implementation is now tied to a single node-level IndicesFieldDataCache, it's safe to clear using that
110+
// IndicesFieldDataCache.
111+
indicesFieldDataCache.clear(index());
112+
fieldDataCaches.clear();
124113
}
125114

126115
public synchronized void clearField(final String fieldName) {
127-
List<Exception> exceptions = new ArrayList<>(0);
128-
final IndexFieldDataCache cache = fieldDataCaches.remove(fieldName);
129-
if (cache != null) {
130-
try {
131-
cache.clear(fieldName);
132-
} catch (Exception e) {
133-
exceptions.add(e);
134-
}
135-
}
136-
ExceptionsHelper.maybeThrowRuntimeAndSuppress(exceptions);
116+
indicesFieldDataCache.clear(index(), fieldName);
117+
fieldDataCaches.remove(fieldName);
137118
}
138119

139120
/**

‎server/src/main/java/org/opensearch/indices/IndicesService.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -383,7 +383,7 @@ public class IndicesService extends AbstractLifecycleComponent
383383
private final MapperRegistry mapperRegistry;
384384
private final NamedWriteableRegistry namedWriteableRegistry;
385385
private final IndexingMemoryController indexingMemoryController;
386-
private final TimeValue cleanInterval;
386+
private final TimeValue cleanInterval; // clean interval for the field data cache
387387
final IndicesRequestCache indicesRequestCache; // pkg-private for testing
388388
private final IndicesQueryCache indicesQueryCache;
389389
private final MetaStateService metaStateService;
@@ -1877,7 +1877,7 @@ public void run() {
18771877
logger.trace("running periodic field data cache cleanup");
18781878
}
18791879
try {
1880-
this.cache.getCache().refresh();
1880+
this.cache.clear();
18811881
} catch (Exception e) {
18821882
logger.warn("Exception during periodic field data cache cleanup:", e);
18831883
}

0 commit comments

Comments
 (0)