[Doris On ES] [UT] Add UT test for all search phase

fe/src/test/java/org/apache/doris/catalog/CatalogTestUtil.java

@@ -256,7 +256,7 @@ public static void createEsTable(Database db) throws DdlException {
         RangePartitionInfo partitionInfo = new RangePartitionInfo(partitionColumns);
         Map<String, String> properties = Maps.newHashMap();
         properties.put(EsTable.HOSTS, "xxx");
-        properties.put(EsTable.INDEX, "indexa");
+        properties.put(EsTable.INDEX, "doe");
         properties.put(EsTable.TYPE, "doc");
         properties.put(EsTable.PASSWORD, "");
         properties.put(EsTable.USER, "root");

fe/src/test/java/org/apache/doris/external/elasticsearch/EsShardPartitionsTest.java

@@ -0,0 +1,43 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.external.elasticsearch;
+
+import org.apache.doris.catalog.Catalog;
+import org.apache.doris.catalog.CatalogTestUtil;
+import org.apache.doris.catalog.EsTable;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+public class EsShardPartitionsTest extends EsTestCase {
+
+    @Test
+    public void testPartition() throws Exception {
+        EsTable esTable = (EsTable) Catalog.getCurrentCatalog()
+                .getDb(CatalogTestUtil.testDb1)
+                .getTable(CatalogTestUtil.testEsTableId1);
+        EsShardPartitions esShardPartitions = EsShardPartitions.findShardPartitions("doe",
+                loadJsonFromFile("data/es/test_search_shards.json"));
+        EsTablePartitions esTablePartitions = EsTablePartitions.fromShardPartitions(esTable, esShardPartitions);
+        assertNotNull(esTablePartitions);
+        assertEquals(1, esTablePartitions.getUnPartitionedIndexStates().size());
+        assertEquals(5, esTablePartitions.getEsShardPartitions("doe").getShardRoutings().size());
+    }
+}

fe/src/test/java/org/apache/doris/external/elasticsearch/EsTestCase.java

@@ -0,0 +1,87 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.external.elasticsearch;
+
+import org.apache.doris.catalog.Catalog;
+import org.apache.doris.catalog.CatalogTestUtil;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.EsTable;
+import org.apache.doris.catalog.FakeCatalog;
+import org.apache.doris.catalog.FakeEditLog;
+import org.apache.doris.common.DdlException;
+import org.apache.doris.common.FeMetaVersion;
+import org.apache.doris.meta.MetaContext;
+
+import org.junit.Before;
+import org.junit.BeforeClass;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.URISyntaxException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+public class EsTestCase {
+
+    protected static FakeEditLog fakeEditLog;
+    protected static FakeCatalog fakeCatalog;
+    protected static Catalog masterCatalog;
+    protected static String mappingsStr = "";
+
+    @BeforeClass
+    public static void init() throws Exception {
+        fakeEditLog = new FakeEditLog();
+        fakeCatalog = new FakeCatalog();
+        masterCatalog = CatalogTestUtil.createTestCatalog();
+        MetaContext metaContext = new MetaContext();
+        metaContext.setMetaVersion(FeMetaVersion.VERSION_40);
+        metaContext.setThreadLocalInfo();
+        // masterCatalog.setJournalVersion(FeMetaVersion.VERSION_40);
+        FakeCatalog.setCatalog(masterCatalog);
+    }
+
+    protected String loadJsonFromFile(String fileName) throws IOException, URISyntaxException {
+        File file = new File(MappingPhaseTest.class.getClassLoader().getResource(fileName).toURI());
+        InputStream is = new FileInputStream(file);
+        BufferedReader br = new BufferedReader(new InputStreamReader(is));
+        StringBuilder jsonStr = new StringBuilder();
+        String line = "";
+        while ((line = br.readLine()) != null) {
+            jsonStr.append(line);
+        }
+        br.close();
+        is.close();
+        return jsonStr.toString();
+    }
+
+    public EsTable fakeEsTable(String table, String index, String type, List<Column> columns) throws DdlException {
+        Map<String, String> props = new HashMap<>();
+        props.put(EsTable.HOSTS, "127.0.0.1:8200");
+        props.put(EsTable.INDEX, index);
+        props.put(EsTable.TYPE, type);
+        props.put(EsTable.VERSION, "6.5.3");
+        return new EsTable(new Random().nextLong(), table, columns, props, null);
+
+    }
+}

fe/src/test/java/org/apache/doris/external/elasticsearch/MappingPhaseTest.java

@@ -0,0 +1,105 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.external.elasticsearch;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.EsTable;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.common.ExceptionChecker;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import mockit.Expectations;
+import mockit.Injectable;
+
+import static org.junit.Assert.assertEquals;
+
+public class MappingPhaseTest extends EsTestCase {
+
+
+    List<Column> columns = new ArrayList<>();
+
+    @Before
+    public void setUp() {
+        Column k1 = new Column("k1", PrimitiveType.BIGINT);
+        Column k2 = new Column("k2", PrimitiveType.VARCHAR);
+        Column k3 = new Column("k3", PrimitiveType.VARCHAR);
+        columns.add(k1);
+        columns.add(k2);
+        columns.add(k3);
+    }
+
+    @Test
+    public void testExtractFieldsNormal() throws Exception {
+
+        MappingPhase mappingPhase = new MappingPhase(null);
+        // ES version < 7.0
+        EsTable esTableBefore7X = fakeEsTable("fake", "test", "doc", columns);
+        SearchContext searchContext = new SearchContext(esTableBefore7X);
+        mappingPhase.resolveFields(searchContext, loadJsonFromFile("data/es/test_index_mapping.json"));
+        assertEquals("k3.keyword", searchContext.fetchFieldsContext().get("k3"));
+        assertEquals("k3.keyword", searchContext.docValueFieldsContext().get("k3"));
+        assertEquals("k1", searchContext.docValueFieldsContext().get("k1"));
+        assertEquals("k2", searchContext.docValueFieldsContext().get("k2"));
+
+        // ES version >= 7.0
+        EsTable esTableAfter7X = fakeEsTable("fake", "test", "_doc", columns);
+        SearchContext searchContext1 = new SearchContext(esTableAfter7X);
+        mappingPhase.resolveFields(searchContext1, loadJsonFromFile("data/es/test_index_mapping_after_7x.json"));
+        assertEquals("k3.keyword", searchContext1.fetchFieldsContext().get("k3"));
+        assertEquals("k3.keyword", searchContext1.docValueFieldsContext().get("k3"));
+        assertEquals("k1", searchContext1.docValueFieldsContext().get("k1"));
+        assertEquals("k2", searchContext1.docValueFieldsContext().get("k2"));
+    }
+
+    @Test
+    public void testTypeNotExist() throws Exception {
+        MappingPhase mappingPhase = new MappingPhase(null);
+        EsTable table = fakeEsTable("fake", "test", "not_exists", columns);
+        SearchContext searchContext = new SearchContext(table);
+        // type not exists
+        ExceptionChecker.expectThrows(DorisEsException.class,
+                () -> mappingPhase.resolveFields(searchContext, loadJsonFromFile("data/es/test_index_mapping.json")));
+    }
+
+    @Test
+    public void testWorkFlow(@Injectable EsRestClient client) throws Exception{
+        EsTable table = fakeEsTable("fake", "test", "doc", columns);
+        SearchContext searchContext1 = new SearchContext(table);
+        String jsonMapping = loadJsonFromFile("data/es/test_index_mapping.json");
+        new Expectations(client) {
+            {
+                client.getMapping(anyString, anyBoolean);
+                minTimes = 0;
+                result = jsonMapping;
+            }
+        };
+        MappingPhase mappingPhase = new MappingPhase(client);
+        ExceptionChecker.expectThrowsNoException(() -> mappingPhase.execute(searchContext1));
+        ExceptionChecker.expectThrowsNoException(() -> mappingPhase.postProcess(searchContext1));
+        assertEquals("k3.keyword", searchContext1.fetchFieldsContext().get("k3"));
+        assertEquals("k3.keyword", searchContext1.docValueFieldsContext().get("k3"));
+        assertEquals("k1", searchContext1.docValueFieldsContext().get("k1"));
+        assertEquals("k2", searchContext1.docValueFieldsContext().get("k2"));
+
+    }
+}

fe/src/test/java/org/apache/doris/external/elasticsearch/PartitionPhaseTest.java

@@ -0,0 +1,80 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.external.elasticsearch;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.EsTable;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.common.ExceptionChecker;
+
+import org.codehaus.jackson.JsonParser;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import mockit.Expectations;
+import mockit.Injectable;
+
+import static org.junit.Assert.assertNotNull;
+
+public class PartitionPhaseTest extends EsTestCase {
+
+    @Test
+    public void testWorkFlow(@Injectable EsRestClient client) throws Exception {
+        final EsShardPartitions[] esShardPartitions = {null};
+        ExceptionChecker.expectThrowsNoException(() ->
+                esShardPartitions[0] = EsShardPartitions.findShardPartitions("doe",
+                        loadJsonFromFile("data/es/test_search_shards.json")));
+        assertNotNull(esShardPartitions[0]);
+        ObjectMapper mapper = new ObjectMapper();
+        JsonParser jsonParser = mapper.getJsonFactory().createJsonParser(loadJsonFromFile("data/es/test_nodes_http.json"));
+        Map<String, Map<String, Object>> nodesData = (Map<String, Map<String, Object>>) mapper.readValue(jsonParser, Map.class).get("nodes");
+        Map<String, EsNodeInfo> nodesMap = new HashMap<>();
+        for (Map.Entry<String, Map<String, Object>> entry : nodesData.entrySet()) {
+            EsNodeInfo node = new EsNodeInfo(entry.getKey(), entry.getValue());
+            if (node.hasHttp()) {
+                nodesMap.put(node.getId(), node);
+            }
+        }
+
+        new Expectations(client) {
+            {
+                client.getHttpNodes();
+                minTimes = 0;
+                result = nodesMap;
+
+                client.searchShards("doe");
+                minTimes = 0;
+                result = esShardPartitions[0];
+            }
+        };
+        List<Column> columns = new ArrayList<>();
+        Column k1 = new Column("k1", PrimitiveType.BIGINT);
+        columns.add(k1);
+        EsTable esTableBefore7X = fakeEsTable("doe", "doe", "doc", columns);
+        SearchContext context = new SearchContext(esTableBefore7X);
+        PartitionPhase partitionPhase = new PartitionPhase(client);
+        ExceptionChecker.expectThrowsNoException(() -> partitionPhase.execute(context));
+        ExceptionChecker.expectThrowsNoException(() -> partitionPhase.postProcess(context));
+        assertNotNull(context.tablePartitions());
+    }
+}

fe/src/test/java/org/apache/doris/external/elasticsearch/VersionPhaseTest.java

@@ -0,0 +1,58 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.external.elasticsearch;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.EsTable;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.common.ExceptionChecker;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import mockit.Expectations;
+import mockit.Injectable;
+
+import static org.junit.Assert.assertTrue;
+
+public class VersionPhaseTest extends EsTestCase {
+
+    @Test
+    public void testWorkFlow(@Injectable EsRestClient client) throws Exception {
+        List<Column> columns = new ArrayList<>();
+        Column k1 = new Column("k1", PrimitiveType.BIGINT);
+        columns.add(k1);
+        EsTable esTableBefore7X = fakeEsTable("fake", "test", "doc", columns);
+        SearchContext context = new SearchContext(esTableBefore7X);
+
+        new Expectations(client) {
+            {
+                client.version();
+                minTimes = 0;
+                result = EsMajorVersion.V_6_X;
+            }
+        };
+        VersionPhase versionPhase = new VersionPhase(client);
+        ExceptionChecker.expectThrowsNoException(() -> versionPhase.preProcess(context));
+        ExceptionChecker.expectThrowsNoException(() -> versionPhase.execute(context));
+        assertTrue(context.version().on(EsMajorVersion.V_6_X));
+    }
+
+}