Allow to clear the fielddata cache per field

With this commit we clear the fielddata cache per field as it is
supposed to be. Previously we retrieved the proper field from the cache
but then cleared the entire cache anyway.

Closes #33798
Relates #33807

Author: danielmitterdorfer

server/src/main/java/org/elasticsearch/index/fielddata/IndexFieldDataService.java

@@ -97,7 +97,7 @@ public synchronized void clearField(final String fieldName) {
         final IndexFieldDataCache cache = fieldDataCaches.remove(fieldName);
         if (cache != null) {
             try {
-                cache.clear();
+                cache.clear(fieldName);
             } catch (Exception e) {
                 exceptions.add(e);
             }

server/src/test/java/org/elasticsearch/index/fielddata/IndexFieldDataServiceTests.java

@@ -100,6 +100,66 @@ public void testGetForFieldDefaults() {
         assertTrue(fd instanceof SortedNumericDVIndexFieldData);
     }
 
+    public void testClearField() throws Exception {
+        final IndexService indexService = createIndex("test");
+        final IndicesService indicesService = getInstanceFromNode(IndicesService.class);
+        // copy the ifdService since we can set the listener only once.
+        final IndexFieldDataService ifdService = new IndexFieldDataService(indexService.getIndexSettings(),
+            indicesService.getIndicesFieldDataCache(), indicesService.getCircuitBreakerService(), indexService.mapperService());
+
+        final BuilderContext ctx = new BuilderContext(indexService.getIndexSettings().getSettings(), new ContentPath(1));
+        final MappedFieldType mapper1 = new TextFieldMapper.Builder("field_1").fielddata(true).build(ctx).fieldType();
+        final MappedFieldType mapper2 = new TextFieldMapper.Builder("field_2").fielddata(true).build(ctx).fieldType();
+        final IndexWriter writer = new IndexWriter(new RAMDirectory(), new IndexWriterConfig(new KeywordAnalyzer()));
+        Document doc = new Document();
+        doc.add(new StringField("field_1", "thisisastring", Store.NO));
+        doc.add(new StringField("field_2", "thisisanotherstring", Store.NO));
+        writer.addDocument(doc);
+        final IndexReader reader = DirectoryReader.open(writer);
+        final AtomicInteger onCacheCalled = new AtomicInteger();
+        final AtomicInteger onRemovalCalled = new AtomicInteger();
+        ifdService.setListener(new IndexFieldDataCache.Listener() {
+            @Override
+            public void onCache(ShardId shardId, String fieldName, Accountable ramUsage) {
+                onCacheCalled.incrementAndGet();
+            }
+
+            @Override
+            public void onRemoval(ShardId shardId, String fieldName, boolean wasEvicted, long sizeInBytes) {
+                onRemovalCalled.incrementAndGet();
+            }
+        });
+        IndexFieldData<?> ifd1 = ifdService.getForField(mapper1);
+        IndexFieldData<?> ifd2 = ifdService.getForField(mapper2);
+        LeafReaderContext leafReaderContext = reader.getContext().leaves().get(0);
+        AtomicFieldData loadField1 = ifd1.load(leafReaderContext);
+        AtomicFieldData loadField2 = ifd2.load(leafReaderContext);
+
+        assertEquals(2, onCacheCalled.get());
+        assertEquals(0, onRemovalCalled.get());
+
+        ifdService.clearField("field_1");
+
+        assertEquals(2, onCacheCalled.get());
+        assertEquals(1, onRemovalCalled.get());
+
+        ifdService.clearField("field_1");
+
+        assertEquals(2, onCacheCalled.get());
+        assertEquals(1, onRemovalCalled.get());
+
+        ifdService.clearField("field_2");
+
+        assertEquals(2, onCacheCalled.get());
+        assertEquals(2, onRemovalCalled.get());
+
+        reader.close();
+        loadField1.close();
+        loadField2.close();
+        writer.close();
+        ifdService.clear();
+    }
+
     public void testFieldDataCacheListener() throws Exception {
         final IndexService indexService = createIndex("test");
         final IndicesService indicesService = getInstanceFromNode(IndicesService.class);