Support merging object-type fields when fetching the schema from the index

[xinyual]

Description

This PR supports merging object-type fields when fetching the schema from the several. For example, we have

PUT demo1
{
  "mappings": {
    "properties": {
      "machine": {
        "properties": {
          "os1": {
            "type": "text"
          },
          "ram1": {
            "type": "long"
          }
        }
      }
    }
  }
}

And

PUT demo2
{
  "mappings": {
    "properties": {
      "machine": {
        "properties": {
          "os2": {
            "type": "text"
          },
          "ram2": {
            "type": "long"
          }
        }
      }
    }
  }
}

Now we support source=demo1, demo2 | fields machine.os1, machine.os2

Also, did some benchmark test with different indices number and depth, reporting the average time of merging operation

	indices=120	indices=1000
depth=15	0.103ms	0.951ms
depth=5	0.041ms	0.329ms

You can also do benchmark using MergeArrayAndObjectMapBenchmark with different arguments.

Related Issues

Resolves #[Issue number to be closed when this PR is merged]
#3625

Check List

• New functionality includes testing.
• New functionality has been documented.
• New functionality has javadoc added.
• New functionality has a user manual doc added.
• API changes companion pull request created.
• Commits are signed per the DCO using --signoff.
• 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.

[penghuo]

consider yamlRestIT, it does not required prepare mapping and data file, https://github.com/opensearch-project/sql/tree/main/integ-test/src/yamlRestTest/resources/rest-api-spec/test/issues.

[penghuo]

add an MergeRule abstraction.

• For basic data type, before PR, the rule is Latest, after PR, the rule is noChange
• For advance type, the rule is DeepMerge.

[xinyual]

I add a mergeRule utils but not sure whether it meets your requirement. For basic data type, I still keep the latest datatype according the order of indices.

[penghuo]

your code looks good.

We may support other type rules, for instance WideningMergeRule, so my previous thoughts is each Rule define it is match and mergeInto method

  public static void merge(Map<String, OpenSearchDataType> target, Map<String, OpenSearchDataType> source) {
    for (Map.Entry<String, OpenSearchDataType> entry : source.entrySet()) {
      String key = entry.getKey();
      OpenSearchDataType sourceValue = entry.getValue();
      OpenSearchDataType targetValue = target.get(key);
      RuleSelectorChain.selectRule(sourceValue, targetValue).mergeInto(key, sourceValue, target);
    }
  }
  
public class RuleSelectorChain {

  private static final List<MergeRuleSelector> RULE_SELECTORS = List.of(
      new DeepMergeSelector(),
      new LatestSelector()
  );

  public static MergeRule selectRule(OpenSearchDataType source, OpenSearchDataType target) {
    if (target == null) {
      return new LatestWinsRule();
    }
  
    return RULE_SELECTORS.stream()
        .map(selector -> selector.select(source, target))
        .filter(Optional::isPresent)
        .map(Optional::get)
        .findFirst()
        .orElse(new LatestSelector());  // this is default
  }
}

public class DeepMergeSelector implements MergeRuleSelector {
  @Override
  public Optional<MergeRule> select(OpenSearchDataType source, OpenSearchDataType target) {
     // return Optional.of(new DeepMergeRule()) if condition meet
  }
}

public class DeepMergeRule implements MergeRule {
  @Override
  public void mergeInto(String key, OpenSearchDataType source, Map<String, OpenSearchDataType> target) {
    OpenSearchDataType existing = target.get(key);
    merge(existing.getProperties(), source.getProperties());
    target.put(key, existing);
  }
}

[xinyual]

Cool. Will refactor code like your suggestion.

[xinyual]

I already refactor code and add interface with two implementations. Please check it.

[penghuo]

1. Add a performance test for deep nesting, e.g., 10+ levels and 100/1000 indices. you can leverage benchmark in repo.
2. based on test result,
   a. consider depth limit settings
   b. document merging limitations

[xinyual]

Already add a benchmark for it. I tried 15 depth with 120 indices. The result is
Benchmark Mode Cnt Score Error Units
testMerge thrpt 25 9619.794 ± 393.331 ops/s
What is the expectation minimum ops of this merging action?

[penghuo]

Could u publish test result in PR description?
How long it will take to merge 15 depth with 120 indices? what if 1000 indices?

[xinyual]

Sure. Already add result to the description. Let me know if you want more data.

[penghuo]

Run a test load test, with 1000 indices, results shows when concurrent requred increased to 64, the latency increase to 12s.

next step.

1. Could u double confirm load test results and update PR descritions?
2. Profile OpenSearch, if the major latency contributor is getIndexMapping API, open issue in core repo.
3. Update PPL Inconsistent Field Types across indices section with test results.

[penghuo]

Could u publish test result in PR description?
How long it will take to merge 15 depth with 120 indices? what if 1000 indices?

[penghuo]

your code looks good.

We may support other type rules, for instance WideningMergeRule, so my previous thoughts is each Rule define it is match and mergeInto method

  public static void merge(Map<String, OpenSearchDataType> target, Map<String, OpenSearchDataType> source) {
    for (Map.Entry<String, OpenSearchDataType> entry : source.entrySet()) {
      String key = entry.getKey();
      OpenSearchDataType sourceValue = entry.getValue();
      OpenSearchDataType targetValue = target.get(key);
      RuleSelectorChain.selectRule(sourceValue, targetValue).mergeInto(key, sourceValue, target);
    }
  }
  
public class RuleSelectorChain {

  private static final List<MergeRuleSelector> RULE_SELECTORS = List.of(
      new DeepMergeSelector(),
      new LatestSelector()
  );

  public static MergeRule selectRule(OpenSearchDataType source, OpenSearchDataType target) {
    if (target == null) {
      return new LatestWinsRule();
    }
  
    return RULE_SELECTORS.stream()
        .map(selector -> selector.select(source, target))
        .filter(Optional::isPresent)
        .map(Optional::get)
        .findFirst()
        .orElse(new LatestSelector());  // this is default
  }
}

public class DeepMergeSelector implements MergeRuleSelector {
  @Override
  public Optional<MergeRule> select(OpenSearchDataType source, OpenSearchDataType target) {
     // return Optional.of(new DeepMergeRule()) if condition meet
  }
}

public class DeepMergeRule implements MergeRule {
  @Override
  public void mergeInto(String key, OpenSearchDataType source, Map<String, OpenSearchDataType> target) {
    OpenSearchDataType existing = target.get(key);
    merge(existing.getProperties(), source.getProperties());
    target.put(key, existing);
  }
}

[penghuo]

mergeObjectAndArrayInsideMap not exist

[penghuo]

@xinyual Please create a issue to track pressure test and support merge limitation.

[LantaoJin]

@xinyual could you resolve conflicts?

[anasalkouz]

Is this backported to 3.0 or 3.1?