Add changes to build star tree in off heap (opensearch-project#14817)

---------

Signed-off-by: Bharathwaj G <[email protected]>

server/src/internalClusterTest/java/org/opensearch/index/mapper/StarTreeMapperIT.java

@@ -275,7 +275,7 @@ public void testValidCompositeIndex() {
                     assertEquals(expectedMetrics, starTreeFieldType.getMetrics().get(0).getMetrics());
                     assertEquals(10000, starTreeFieldType.getStarTreeConfig().maxLeafDocs());
                     assertEquals(
-                        StarTreeFieldConfiguration.StarTreeBuildMode.ON_HEAP,
+                        StarTreeFieldConfiguration.StarTreeBuildMode.OFF_HEAP,
                         starTreeFieldType.getStarTreeConfig().getBuildMode()
                     );
                     assertEquals(Collections.emptySet(), starTreeFieldType.getStarTreeConfig().getSkipStarNodeCreationInDims());
@@ -359,7 +359,7 @@ public void testUpdateIndexWhenMappingIsSame() {
                     assertEquals(expectedMetrics, starTreeFieldType.getMetrics().get(0).getMetrics());
                     assertEquals(10000, starTreeFieldType.getStarTreeConfig().maxLeafDocs());
                     assertEquals(
-                        StarTreeFieldConfiguration.StarTreeBuildMode.ON_HEAP,
+                        StarTreeFieldConfiguration.StarTreeBuildMode.OFF_HEAP,
                         starTreeFieldType.getStarTreeConfig().getBuildMode()
                     );
                     assertEquals(Collections.emptySet(), starTreeFieldType.getStarTreeConfig().getSkipStarNodeCreationInDims());

server/src/main/java/org/opensearch/common/util/ByteArrayBackedBitset.java

@@ -0,0 +1,86 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.common.util;
+
+import org.apache.lucene.store.IndexInput;
+import org.apache.lucene.store.IndexOutput;
+import org.apache.lucene.store.RandomAccessInput;
+
+import java.io.IOException;
+
+/**
+ * A bitset backed by a byte array. This will initialize and set bits in the byte array based on the index.
+ */
+public class ByteArrayBackedBitset {
+    private final byte[] byteArray;
+
+    /**
+     * Constructor which uses an on heap list. This should be using during construction of the bitset.
+     */
+    public ByteArrayBackedBitset(int capacity) {
+        byteArray = new byte[capacity];
+    }
+
+    /**
+     * Constructor which set the Lucene's RandomAccessInput to read the bitset into a read-only buffer.
+     */
+    public ByteArrayBackedBitset(RandomAccessInput in, long offset, int length) throws IOException {
+        byteArray = new byte[length];
+        int i = 0;
+        while (i < length) {
+            byteArray[i] = in.readByte(offset + i);
+            i++;
+        }
+    }
+
+    /**
+     * Constructor which set the Lucene's IndexInput to read the bitset into a read-only buffer.
+     */
+    public ByteArrayBackedBitset(IndexInput in, int length) throws IOException {
+        byteArray = new byte[length];
+        int i = 0;
+        while (i < length) {
+            byteArray[i] = in.readByte();
+            i++;
+        }
+    }
+
+    /**
+     * Sets the bit at the given index to 1.
+     * Each byte can indicate 8 bits, so the index is divided by 8 to get the byte array index.
+     * @param index the index to set the bit
+     */
+    public void set(int index) {
+        int byteArrIndex = index >> 3;
+        byteArray[byteArrIndex] |= (byte) (1 << (index & 7));
+    }
+
+    public int write(IndexOutput output) throws IOException {
+        int numBytes = 0;
+        for (Byte bitSet : byteArray) {
+            output.writeByte(bitSet);
+            numBytes += Byte.BYTES;
+        }
+        return numBytes;
+    }
+
+    /**
+     * Retrieves whether the bit is set or not at the given index.
+     * @param index the index to look up for the bit
+     * @return true if bit is set, false otherwise
+     */
+    public boolean get(int index) throws IOException {
+        int byteArrIndex = index >> 3;
+        return (byteArray[byteArrIndex] & (1 << (index & 7))) != 0;
+    }
+
+    public int getCurrBytesRead() {
+        return byteArray.length;
+    }
+}

server/src/main/java/org/opensearch/index/codec/composite/Composite99DocValuesWriter.java

@@ -8,8 +8,6 @@
 
 package org.opensearch.index.codec.composite;
 
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import org.apache.lucene.codecs.DocValuesConsumer;
 import org.apache.lucene.codecs.DocValuesProducer;
 import org.apache.lucene.index.DocValues;
@@ -50,9 +48,9 @@ public class Composite99DocValuesWriter extends DocValuesConsumer {
     private final Set<CompositeMappedFieldType> compositeMappedFieldTypes;
     private final Set<String> compositeFieldSet;
     private final Set<String> segmentFieldSet;
+    private final boolean segmentHasCompositeFields;
 
     private final Map<String, DocValuesProducer> fieldProducerMap = new HashMap<>();
-    private static final Logger logger = LogManager.getLogger(Composite99DocValuesWriter.class);
 
     public Composite99DocValuesWriter(DocValuesConsumer delegate, SegmentWriteState segmentWriteState, MapperService mapperService) {
 
@@ -70,6 +68,8 @@ public Composite99DocValuesWriter(DocValuesConsumer delegate, SegmentWriteState
         for (CompositeMappedFieldType type : compositeMappedFieldTypes) {
             compositeFieldSet.addAll(type.fields());
         }
+        // check if there are any composite fields which are part of the segment
+        segmentHasCompositeFields = Collections.disjoint(segmentFieldSet, compositeFieldSet) == false;
     }
 
     @Override
@@ -91,7 +91,7 @@ public void addSortedField(FieldInfo field, DocValuesProducer valuesProducer) th
     public void addSortedNumericField(FieldInfo field, DocValuesProducer valuesProducer) throws IOException {
         delegate.addSortedNumericField(field, valuesProducer);
         // Perform this only during flush flow
-        if (mergeState.get() == null) {
+        if (mergeState.get() == null && segmentHasCompositeFields) {
             createCompositeIndicesIfPossible(valuesProducer, field);
         }
     }

server/src/main/java/org/opensearch/index/compositeindex/datacube/startree/aggregators/CountValueAggregator.java

@@ -68,4 +68,9 @@ public Long toLongValue(Long value) {
     public Long toStarTreeNumericTypeValue(Long value) {
         return value;
     }
+
+    @Override
+    public Long getIdentityMetricValue() {
+        return 0L;
+    }
 }

server/src/main/java/org/opensearch/index/compositeindex/datacube/startree/aggregators/SumValueAggregator.java

@@ -103,4 +103,9 @@ public Double toStarTreeNumericTypeValue(Long value) {
             throw new IllegalStateException("Cannot convert " + value + " to sortable aggregation type", e);
         }
     }
+
+    @Override
+    public Double getIdentityMetricValue() {
+        return 0D;
+    }
 }

server/src/main/java/org/opensearch/index/compositeindex/datacube/startree/aggregators/ValueAggregator.java

@@ -61,4 +61,9 @@ public interface ValueAggregator<A> {
      * Converts an aggregated value from a Long type.
      */
     A toStarTreeNumericTypeValue(Long rawValue);
+
+    /**
+     * Fetches a value that does not alter the result of aggregations
+     */
+    A getIdentityMetricValue();
 }