fix(search): prevent ES8 reindex loops and normalize legacy range queries (#17641)

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: david-leifker

docs/how/updating-datahub.md

@@ -170,9 +170,11 @@ Requirements:
 
 ### Known Issues
 
+- **(Operations / Elasticsearch 8) Spurious mappings reindex on every Helm upgrade.** Deployments backed by **Elasticsearch 8.x** may trigger a full index mappings reindex on **every** system-update run (including each Helm upgrade), even when no authored mapping change was shipped. ES8 persists mappings differently than DataHub's authored definitions (for example injecting `type: object` on nested object fields), so the index builder detects spurious drift and schedules reindex work repeatedly. This is most likely when **`ELASTICSEARCH_INDEX_BUILDER_MAPPINGS_REINDEX=true`** (including Docker quickstart, which enables it for initial index builds). **Workaround:** Let the **first** system-update after install or upgrade finish so indices are created, then set **`ELASTICSEARCH_INDEX_BUILDER_MAPPINGS_REINDEX=false`** on **`datahub-gms`** and **`datahub-upgrade`** before later Helm upgrades until a fixed release is available.
+
 ### Potential Downtime
 
-- **System-update / Elasticsearch:** Upgrades that trigger reindexing, incremental index migration, or optional Elasticsearch ZDU (#16887) can run for an extended period depending on catalog size. Reindex duration was improved (#16949) and a perpetual ES8 reindex loop fix was included (#17402). Helm upgrades without `--atomic` may report failure while background reindex jobs continue — allow system-update to finish before retrying. See **Helm Notes** under the [0.10.0](#0100) section below for timeout guidance patterns.
+- **System-update / Elasticsearch:** Upgrades that trigger reindexing, incremental index migration, or optional Elasticsearch ZDU (#16887) can run for an extended period depending on catalog size. Reindex duration was improved (#16949). Elasticsearch 8 deployments with mappings reindex enabled may repeat unnecessary reindex work on every Helm upgrade — see **Known Issues** above. Helm upgrades without `--atomic` may report failure while background reindex jobs continue — allow system-update to finish before retrying. See **Helm Notes** under the [0.10.0](#0100) section below for timeout guidance patterns.
 - **First deploy after bootstrap moves:** Entity Types ingestion moved to system-update (see Breaking Changes). Expect additional system-update work on the first upgrade after this release.
 - **Aspect schema version sweep (#16930):** Large deployments may observe background migration activity during upgrade; plan maintenance windows accordingly.
 

metadata-io/src/main/java/com/linkedin/metadata/search/elasticsearch/client/shim/impl/Es8SearchClientShim.java

@@ -90,6 +90,7 @@
 import com.linkedin.metadata.search.elasticsearch.client.shim.builder.es8.Es8SemanticIndexSettingsBuilder;
 import com.linkedin.metadata.search.elasticsearch.client.shim.impl.v8.CustomQuery;
 import com.linkedin.metadata.search.elasticsearch.client.shim.impl.v8.Es8BulkListener;
+import com.linkedin.metadata.search.elasticsearch.client.shim.impl.v8.LegacyRangeQueryNormalizer;
 import com.linkedin.metadata.utils.elasticsearch.responses.GetIndexResponse;
 import com.linkedin.metadata.utils.elasticsearch.responses.RawResponse;
 import com.linkedin.metadata.utils.elasticsearch.shim.EmbeddingBatch;
@@ -105,6 +106,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -242,6 +244,22 @@
 public class Es8SearchClientShim extends AbstractBulkProcessorShim<BulkIngester<?>>
     implements ElasticSearchClientShim<ElasticsearchClient> {
 
+  /**
+   * ES8+ silently strips {@code doc_values: false} from {@code search_as_you_type} fields on
+   * round-trip. Including it in the authored mapping creates a permanent diff against what the
+   * cluster returns and triggers a reindex on every system update cycle, so we omit it here.
+   */
+  public static final Map<String, String> PARTIAL_NGRAM_CONFIG =
+      ImmutableMap.of(
+          "type", "search_as_you_type",
+          "max_shingle_size", "4");
+
+  /**
+   * ES8 injects {@code type: custom} on custom analyzers when settings are persisted, but authored
+   * V2 index settings omit {@code type} on analyzer definitions.
+   */
+  public static final String INJECTED_CUSTOM_ANALYZER_TYPE = "custom";
+
   @Getter private final ShimConfiguration shimConfiguration;
   private final SearchEngineType engineType;
   private final ElasticsearchClient client;
@@ -1424,22 +1442,54 @@ public SearchEngineType getEngineType() {
     return engineType;
   }
 
-  /**
-   * ES8+ silently strips {@code doc_values: false} from {@code search_as_you_type} fields on
-   * round-trip. Including it in the authored mapping creates a permanent diff against what the
-   * cluster returns and triggers a reindex on every system update cycle, so we omit it here.
-   */
-  public static final Map<String, String> PARTIAL_NGRAM_CONFIG =
-      ImmutableMap.of(
-          "type", "search_as_you_type",
-          "max_shingle_size", "4");
+  /** ES8-specific index analysis settings comparison rules for reindex detection. */
+  public static final class IndexSettingsComparison {
+    private IndexSettingsComparison() {}
+
+    @Nonnull
+    public static Set<String> storedNamesForComparison(
+        @Nonnull Map<String, Object> targetSettings, @Nonnull Settings storedSettings) {
+      Set<String> names = new HashSet<>(storedSettings.names());
+      if (!targetSettings.containsKey("type") && names.remove("type")) {
+        String typeValue = storedSettings.get("type");
+        if (typeValue == null || !INJECTED_CUSTOM_ANALYZER_TYPE.equalsIgnoreCase(typeValue)) {
+          names.add("type");
+        }
+      }
+      return names;
+    }
+
+    public static boolean valuesEqual(@Nullable Object targetValue, @Nullable String storedValue) {
+      if (com.linkedin.metadata.utils.elasticsearch.IndexSettingsComparison.Strict.INSTANCE
+          .indexSettingValuesEqual(targetValue, storedValue)) {
+        return true;
+      }
+      if (targetValue == null || storedValue == null) {
+        return false;
+      }
+      return targetValue.toString().equalsIgnoreCase(storedValue);
+    }
+  }
 
   @Nonnull
   @Override
   public Map<String, String> partialNgramConfig() {
     return PARTIAL_NGRAM_CONFIG;
   }
 
+  @Nonnull
+  @Override
+  public Set<String> indexSettingNamesForComparison(
+      @Nonnull Map<String, Object> targetSettings, @Nonnull Settings storedSettings) {
+    return IndexSettingsComparison.storedNamesForComparison(targetSettings, storedSettings);
+  }
+
+  @Override
+  public boolean indexSettingValuesEqual(
+      @Nullable Object targetValue, @Nullable String storedValue) {
+    return IndexSettingsComparison.valuesEqual(targetValue, storedValue);
+  }
+
   @Nonnull
   @Override
   public String getEngineVersion() throws IOException {
@@ -1808,7 +1858,7 @@ private Query convertQuery(org.opensearch.index.query.QueryBuilder osQuery) {
     if (osQuery == null) {
       return null;
     }
-    String jsonString = osQuery.toString();
+    String jsonString = normalizeQueryJson(osQuery.toString());
     return Query.of(
         q ->
             q.withJson(
@@ -1817,14 +1867,22 @@ private Query convertQuery(org.opensearch.index.query.QueryBuilder osQuery) {
   }
 
   private Rescore convertRescore(RescorerBuilder<?> rescorerBuilder) {
-    String jsonString = rescorerBuilder.toString();
+    String jsonString = normalizeQueryJson(rescorerBuilder.toString());
     return Rescore.of(
         q ->
             q.withJson(
                 jacksonJsonpMapper.jsonProvider().createParser(new StringReader(jsonString)),
                 jacksonJsonpMapper));
   }
 
+  private String normalizeQueryJson(String jsonString) {
+    try {
+      return LegacyRangeQueryNormalizer.normalize(jsonString, objectMapper);
+    } catch (JsonProcessingException e) {
+      return jsonString;
+    }
+  }
+
   private FieldSuggester convertSuggestion(SuggestionBuilder<?> suggestionBuilder) {
     String jsonString = Strings.toString(MediaTypeRegistry.JSON, suggestionBuilder, true, true);
     return FieldSuggester.of(

metadata-io/src/main/java/com/linkedin/metadata/search/elasticsearch/client/shim/impl/v8/LegacyRangeQueryNormalizer.java

@@ -0,0 +1,72 @@
+package com.linkedin.metadata.search.elasticsearch.client.shim.impl.v8;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import javax.annotation.Nonnull;
+
+/**
+ * Rewrites legacy OpenSearch {@code RangeQueryBuilder} JSON ({@code from}/{@code to} with {@code
+ * include_lower}/{@code include_upper}) into ES 8-compatible bounds ({@code gte}/{@code gt}/{@code
+ * lte}/{@code lt}). OpenSearch HLRC {@code QueryBuilder#toString()} still emits the legacy shape,
+ * which Elasticsearch 8.18+ deprecates.
+ */
+public final class LegacyRangeQueryNormalizer {
+
+  private LegacyRangeQueryNormalizer() {}
+
+  @Nonnull
+  public static String normalize(@Nonnull String queryJson, @Nonnull ObjectMapper objectMapper)
+      throws JsonProcessingException {
+    JsonNode root = objectMapper.readTree(queryJson);
+    normalizeRangeNodes(root);
+    return objectMapper.writeValueAsString(root);
+  }
+
+  private static void normalizeRangeNodes(JsonNode node) {
+    if (node == null) {
+      return;
+    }
+    if (node.isObject()) {
+      ObjectNode objectNode = (ObjectNode) node;
+      if (objectNode.has("range") && objectNode.get("range").isObject()) {
+        ObjectNode rangeNode = (ObjectNode) objectNode.get("range");
+        rangeNode
+            .properties()
+            .forEach(
+                entry -> {
+                  if (entry.getValue().isObject()) {
+                    normalizeRangeSpec((ObjectNode) entry.getValue());
+                  }
+                });
+      }
+      objectNode.properties().forEach(entry -> normalizeRangeNodes(entry.getValue()));
+    } else if (node.isArray()) {
+      node.forEach(LegacyRangeQueryNormalizer::normalizeRangeNodes);
+    }
+  }
+
+  private static void normalizeRangeSpec(ObjectNode spec) {
+    if (!spec.has("from") && !spec.has("to")) {
+      return;
+    }
+
+    JsonNode from = spec.get("from");
+    JsonNode to = spec.get("to");
+    boolean includeLower = !spec.has("include_lower") || spec.get("include_lower").asBoolean(true);
+    boolean includeUpper = !spec.has("include_upper") || spec.get("include_upper").asBoolean(true);
+
+    if (from != null && !from.isNull()) {
+      spec.set(includeLower ? "gte" : "gt", from);
+    }
+    if (to != null && !to.isNull()) {
+      spec.set(includeUpper ? "lte" : "lt", to);
+    }
+
+    spec.remove("from");
+    spec.remove("to");
+    spec.remove("include_lower");
+    spec.remove("include_upper");
+  }
+}

metadata-io/src/main/java/com/linkedin/metadata/search/elasticsearch/indexbuilder/ESIndexBuilder.java

@@ -381,7 +381,8 @@ public ReindexConfig buildReindexState(
                 structPropConfig.isEnabled()
                     && structPropConfig.isSystemUpdateEnabled()
                     && !copyStructuredPropertyMappings)
-            .version(gitVersion.getVersion());
+            .version(gitVersion.getVersion())
+            .settingsComparisonShim(searchClient);
 
     Map<String, Object> baseSettings = new HashMap<>(settings);
     baseSettings.put(NUMBER_OF_SHARDS, indexConfig.getNumShards());

metadata-io/src/main/java/com/linkedin/metadata/search/elasticsearch/indexbuilder/ReindexConfig.java

@@ -10,10 +10,13 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.MapDifference;
 import com.google.common.collect.Maps;
+import com.linkedin.metadata.utils.elasticsearch.IndexSettingsComparison;
+import com.linkedin.metadata.utils.elasticsearch.SearchClientShim;
 import com.linkedin.util.Pair;
 import java.util.*;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import javax.annotation.Nonnull;
 import lombok.Builder;
 import lombok.Getter;
 import lombok.experimental.Accessors;
@@ -143,6 +146,15 @@ private ReindexConfigBuilder hasRemovedStructuredProperty(boolean ignored) {
       return this;
     }
 
+    @Nonnull
+    private IndexSettingsComparison settingsComparison = IndexSettingsComparison.Strict.INSTANCE;
+
+    public ReindexConfigBuilder settingsComparisonShim(
+        @Nonnull SearchClientShim<?> settingsComparisonShim) {
+      this.settingsComparison = settingsComparisonShim;
+      return this;
+    }
+
     // ensure sorted
     public ReindexConfigBuilder currentMappings(Map<String, Object> currentMappings) {
       this.currentMappings = sortMap(currentMappings);
@@ -446,7 +458,7 @@ private boolean isAnalysisEqual() {
       // Compare analysis section
       Map<String, Object> newAnalysis = (Map<String, Object>) indexSettings.get("analysis");
       Settings oldAnalysis = super.currentSettings.getByPrefix("index.analysis.");
-      return equalsGroup(newAnalysis, oldAnalysis);
+      return equalsGroup(newAnalysis, oldAnalysis, super.settingsComparison);
     }
 
     private boolean isSettingsEqual() {
@@ -504,7 +516,8 @@ private boolean isSettingsReindexRequired() {
       return (indexSettings.containsKey("analysis")
           && !equalsGroup(
               (Map<String, Object>) indexSettings.get("analysis"),
-              super.currentSettings.getByPrefix("index.analysis.")));
+              super.currentSettings.getByPrefix("index.analysis."),
+              super.settingsComparison));
     }
 
     /**
@@ -690,8 +703,13 @@ private static Map<String, Object> removeKeys(
         .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
   }
 
-  private static boolean equalsGroup(Map<String, Object> newSettings, Settings oldSettings) {
-    if (!newSettings.keySet().equals(oldSettings.names())) {
+  private static boolean equalsGroup(
+      Map<String, Object> newSettings,
+      Settings oldSettings,
+      IndexSettingsComparison settingsComparison) {
+    Set<String> oldNames =
+        settingsComparison.indexSettingNamesForComparison(newSettings, oldSettings);
+    if (!newSettings.keySet().equals(oldNames)) {
       return false;
     }
 
@@ -703,7 +721,9 @@ private static boolean equalsGroup(Map<String, Object> newSettings, Settings old
       }
       if (newSettings.get(key) instanceof Map) {
         if (!equalsGroup(
-            (Map<String, Object>) newSettings.get(key), oldSettings.getByPrefix(key + "."))) {
+            (Map<String, Object>) newSettings.get(key),
+            oldSettings.getByPrefix(key + "."),
+            settingsComparison)) {
           return false;
         }
       } else if (newSettings.get(key) instanceof List) {
@@ -713,14 +733,7 @@ private static boolean equalsGroup(Map<String, Object> newSettings, Settings old
       } else {
         String oldValue = oldSettings.get(key);
         Object newValue = newSettings.get(key);
-        // Handle null values properly
-        if (newValue == null && oldValue == null) {
-          continue;
-        }
-        if (newValue == null || oldValue == null) {
-          return false;
-        }
-        if (!newValue.toString().equals(oldValue)) {
+        if (!settingsComparison.indexSettingValuesEqual(newValue, oldValue)) {
           return false;
         }
       }

metadata-io/src/test/java/com/linkedin/metadata/search/elasticsearch/client/shim/Es8SearchClientShimConversionTest.java

@@ -5,10 +5,12 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotNull;
 import static org.testng.Assert.assertTrue;
 
 import co.elastic.clients.elasticsearch._helpers.bulk.BulkIngester;
+import co.elastic.clients.elasticsearch._types.query_dsl.Query;
 import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
 import co.elastic.clients.elasticsearch.core.search.FieldSuggester;
 import co.elastic.clients.elasticsearch.indices.update_aliases.Action;
@@ -25,6 +27,7 @@
 import org.opensearch.common.xcontent.LoggingDeprecationHandler;
 import org.opensearch.common.xcontent.XContentType;
 import org.opensearch.core.xcontent.XContentParser;
+import org.opensearch.index.query.QueryBuilder;
 import org.opensearch.index.query.QueryBuilders;
 import org.opensearch.search.builder.SearchSourceBuilder;
 import org.opensearch.search.suggest.Suggest;
@@ -353,6 +356,34 @@ private SearchResponse parseSearchResponse(String json) throws Exception {
     }
   }
 
+  /** convertQuery rewrites legacy range bounds before sending to Elasticsearch 8. */
+  @Test
+  public void testConvertQueryNormalizesLegacyRangeBounds() throws Exception {
+    Method normalizeQueryJsonMethod =
+        Es8SearchClientShim.class.getDeclaredMethod("normalizeQueryJson", String.class);
+    normalizeQueryJsonMethod.setAccessible(true);
+
+    String legacy = QueryBuilders.rangeQuery("timestamp").gte(100L).lt(200L).toString();
+    String normalized = (String) normalizeQueryJsonMethod.invoke(shim, legacy);
+
+    assertFalse(normalized.contains("\"from\""));
+    assertFalse(normalized.contains("\"to\""));
+    assertTrue(normalized.contains("\"gte\":100"));
+    assertTrue(normalized.contains("\"lt\":200"));
+
+    Query result = invokeConvertQuery(QueryBuilders.rangeQuery("timestamp").gte(100L).lt(200L));
+    assertNotNull(result);
+  }
+
+  /** Helper method to invoke the private convertQuery method via reflection. */
+  private Query invokeConvertQuery(QueryBuilder queryBuilder) throws Exception {
+    Method convertQueryMethod =
+        Es8SearchClientShim.class.getDeclaredMethod(
+            "convertQuery", org.opensearch.index.query.QueryBuilder.class);
+    convertQueryMethod.setAccessible(true);
+    return (Query) convertQueryMethod.invoke(shim, queryBuilder);
+  }
+
   /** Helper method to invoke the private convertSuggestion method via reflection. */
   private FieldSuggester invokeConvertSuggestion(SuggestionBuilder<?> suggestionBuilder)
       throws Exception {