wuyunfeng commented on a change in pull request #3454:
URL: https://github.com/apache/incubator-doris/pull/3454#discussion_r439292860
##########
File path:
fe/src/main/java/org/apache/doris/external/elasticsearch/EsRestClient.java
##########
@@ -17,92 +17,183 @@
package org.apache.doris.external.elasticsearch;
+import org.apache.doris.catalog.Column;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.HttpHeaders;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.util.Strings;
import org.codehaus.jackson.JsonParser;
import org.codehaus.jackson.map.DeserializationConfig;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.SerializationConfig;
-
+import org.json.JSONArray;
+import org.json.JSONObject;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
-
import okhttp3.Credentials;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
public class EsRestClient {
+
private static final Logger LOG = LogManager.getLogger(EsRestClient.class);
private ObjectMapper mapper;
-
+
{
mapper = new ObjectMapper();
mapper.configure(DeserializationConfig.Feature.USE_ANNOTATIONS, false);
mapper.configure(SerializationConfig.Feature.USE_ANNOTATIONS, false);
}
-
+
private static OkHttpClient networkClient = new OkHttpClient.Builder()
.readTimeout(10, TimeUnit.SECONDS)
.build();
-
- private String basicAuth;
-
- private int nextClient = 0;
+
+ private Request.Builder builder;
private String[] nodes;
private String currentNode;
-
+ private int currentNodeIndex = 0;
+
public EsRestClient(String[] nodes, String authUser, String authPassword) {
this.nodes = nodes;
+ this.builder = new Request.Builder();
if (!Strings.isEmpty(authUser) && !Strings.isEmpty(authPassword)) {
- basicAuth = Credentials.basic(authUser, authPassword);
+ this.builder.addHeader(HttpHeaders.AUTHORIZATION,
+ Credentials.basic(authUser, authPassword));
}
- selectNextNode();
+ this.currentNode = nodes[currentNodeIndex];
}
-
- private boolean selectNextNode() {
- if (nextClient >= nodes.length) {
- return false;
+
+ private void selectNextNode() {
+ currentNodeIndex++;
+ // reroute, because the previously failed node may have already been
restored
+ if (currentNodeIndex >= nodes.length) {
+ currentNodeIndex = 0;
}
- currentNode = nodes[nextClient++];
- return true;
+ currentNode = nodes[currentNodeIndex];
}
-
+
public Map<String, EsNodeInfo> getHttpNodes() throws Exception {
Map<String, Map<String, Object>> nodesData = get("_nodes/http",
"nodes");
if (nodesData == null) {
return Collections.emptyMap();
}
- Map<String, EsNodeInfo> nodes = new HashMap<>();
+ 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()) {
- nodes.put(node.getId(), node);
+ nodesMap.put(node.getId(), node);
}
}
- return nodes;
+ return nodesMap;
}
-
- public String getIndexMetaData(String indexName) throws Exception {
- String path = "_cluster/state?indices=" + indexName
- + "&metric=routing_table,nodes,metadata&expand_wildcards=open";
- return execute(path);
-
+
+ public EsFieldInfo getFieldInfo(String indexName, String mappingType,
List<Column> colList) throws Exception {
+ String path = indexName + "/_mapping";
+ String indexMapping = execute(path);
+ if (indexMapping == null) {
+ throw new Exception( "index[" + indexName + "] _mapping not found
for the Elasticsearch Cluster");
+ }
+ return getFieldInfo(colList, parseProperties(indexMapping,
mappingType));
}
-
+
+ @VisibleForTesting
+ public EsFieldInfo getFieldInfo(List<Column> colList, JSONObject
properties) {
+ if (properties == null) {
+ return null;
+ }
+ Map<String, String> fetchFieldMap = new HashMap<>();
+ Map<String, String> docValueFieldMap = new HashMap<>();
+ for (Column col : colList) {
+ String colName = col.getName();
+ if (!properties.has(colName)) {
+ continue;
+ }
+ JSONObject fieldObject = properties.optJSONObject(colName);
+ String fetchField = EsUtil.getFetchField(fieldObject, colName);
+ if (StringUtils.isNotEmpty(fetchField)) {
+ fetchFieldMap.put(colName, fetchField);
+ }
+ String docValueField = EsUtil.getDocValueField(fieldObject,
colName);
+ if (StringUtils.isNotEmpty(docValueField)) {
+ docValueFieldMap.put(colName, docValueField);
+ }
+ }
+ return new EsFieldInfo(fetchFieldMap, docValueFieldMap);
+ }
+
+ @VisibleForTesting
+ public JSONObject parseProperties(String indexMapping, String mappingType)
{
+ JSONObject jsonObject = new JSONObject(indexMapping);
+ // the indexName use alias takes the first mapping
+ Iterator<String> keys = jsonObject.keys();
+ String docKey = keys.next();
+ JSONObject docData = jsonObject.optJSONObject(docKey);
+ JSONObject mappings = docData.optJSONObject("mappings");
+ JSONObject rootSchema = mappings.optJSONObject(mappingType);
Review comment:
emm, if user set the wrong type the index does not exists, this would
throw NullPointerException.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
