[Doris On ES] Split /_cluster/state to [indexName/_mappings, indexName/_search_shards]

docs/en/extending-doris/doris-on-es.md

@@ -130,8 +130,8 @@ The following parameters are accepted by ES table:
 Parameter | Description
 ---|---
 **hosts** | ES Cluster Connection Address, maybe one or more node, load-balance is also accepted
-**index** | the related ES index name
-**type** | the type for this index，If not specified, `_doc` will be used
+**index** | the related ES index name, alias is supported, and if you use doc_value, you need to use the real name
+**type** | the type for this index, If not specified, `_doc` will be used
 **user** | username for ES
 **password** | password for the user
 

docs/zh-CN/extending-doris/doris-on-es.md

@@ -128,7 +128,7 @@ PROPERTIES (
 参数 | 说明
 ---|---
 **hosts** | ES集群地址，可以是一个或多个，也可以是ES前端的负载均衡地址
-**index** | 对应的ES的index名字
+**index** | 对应的ES的index名字，支持alias，如果使用doc_value，需要使用真实的名称
 **type** | index的type，不指定的情况会使用_doc
 **user** | ES集群用户名
 **password** | 对应用户的密码信息

fe/src/main/java/org/apache/doris/catalog/Catalog.java

@@ -127,7 +127,7 @@
 import org.apache.doris.deploy.impl.AmbariDeployManager;
 import org.apache.doris.deploy.impl.K8sDeployManager;
 import org.apache.doris.deploy.impl.LocalFileDeployManager;
-import org.apache.doris.external.elasticsearch.EsStateStore;
+import org.apache.doris.external.elasticsearch.EsRepository;
 import org.apache.doris.ha.BDBHA;
 import org.apache.doris.ha.FrontendNodeType;
 import org.apache.doris.ha.HAProtocol;
@@ -301,7 +301,7 @@ public class Catalog {
     private Daemon replayer;
     private Daemon timePrinter;
     private Daemon listener;
-    private EsStateStore esStateStore;  // it is a daemon, so add it here
+    private EsRepository esRepository;  // it is a daemon, so add it here
 
     private boolean isFirstTimeStartUp = false;
     private boolean isElectable;
@@ -510,7 +510,7 @@ private Catalog() {
         this.auth = new PaloAuth();
         this.domainResolver = new DomainResolver(auth);
 
-        this.esStateStore = new EsStateStore();
+        this.esRepository = new EsRepository();
 
         this.metaContext = new MetaContext();
         this.metaContext.setThreadLocalInfo();
@@ -1290,7 +1290,7 @@ private void startNonMasterDaemonThreads() {
         // load and export job label cleaner thread
         labelCleaner.start();
         // ES state store
-        esStateStore.start();
+        esRepository.start();
         // domain resolver
         domainResolver.start();
     }
@@ -1452,7 +1452,7 @@ public void loadImage(String imageDir) throws IOException, DdlException {
             // ATTN: this should be done after load Db, and before loadAlterJob
             recreateTabletInvertIndex();
             // rebuild es state state
-            esStateStore.loadTableFromCatalog();
+            esRepository.loadTableFromCatalog();
 
             checksum = loadLoadJob(dis, checksum);
             checksum = loadAlterJob(dis, checksum);
@@ -4376,7 +4376,7 @@ public boolean unprotectDropTable(Database db, long tableId) {
         }
 
         if (table.getType() == TableType.ELASTICSEARCH) {
-            esStateStore.deRegisterTable(tableId);
+            esRepository.deRegisterTable(tableId);
         } else if (table.getType() == TableType.OLAP) {
             // drop all temp partitions of this table, so that there is no temp partitions in recycle bin,
             // which make things easier.
@@ -4828,8 +4828,8 @@ public String getMasterIp() {
         return this.masterIp;
     }
 
-    public EsStateStore getEsStateStore() {
-        return this.esStateStore;
+    public EsRepository getEsRepository() {
+        return this.esRepository;
     }
 
     public void setMaster(MasterInfo info) {

fe/src/main/java/org/apache/doris/catalog/Database.java

@@ -304,7 +304,7 @@ public boolean createTableWithLock(Table table, boolean isReplay, boolean setIfN
                     Catalog.getCurrentCatalog().getEditLog().logCreateTable(info);
                 }
                 if (table.getType() == TableType.ELASTICSEARCH) {
-                    Catalog.getCurrentCatalog().getEsStateStore().registerTable((EsTable)table);
+                    Catalog.getCurrentCatalog().getEsRepository().registerTable((EsTable)table);
                 }
             }
             return result;

fe/src/main/java/org/apache/doris/catalog/EsTable.java

@@ -20,8 +20,12 @@
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.FeMetaVersion;
 import org.apache.doris.common.io.Text;
+import org.apache.doris.external.elasticsearch.EsFieldInfos;
 import org.apache.doris.external.elasticsearch.EsMajorVersion;
-import org.apache.doris.external.elasticsearch.EsTableState;
+import org.apache.doris.external.elasticsearch.EsNodeInfo;
+import org.apache.doris.external.elasticsearch.EsRestClient;
+import org.apache.doris.external.elasticsearch.EsShardPartitions;
+import org.apache.doris.external.elasticsearch.EsTablePartitions;
 import org.apache.doris.thrift.TEsTable;
 import org.apache.doris.thrift.TTableDescriptor;
 import org.apache.doris.thrift.TTableType;
@@ -71,7 +75,7 @@ public class EsTable extends Table {
     // only save the partition definition, save the partition key,
     // partition list is got from es cluster dynamically and is saved in esTableState
     private PartitionInfo partitionInfo;
-    private EsTableState esTableState;
+    private EsTablePartitions esTablePartitions;
     private boolean enableDocValueScan = false;
     private boolean enableKeywordSniff = true;
 
@@ -114,18 +118,19 @@ public EsTable(long id, String name, List<Column> schema,
         validate(properties);
     }
 
-    public void addFetchField(String originName, String replaceName) {
-        fieldsContext.put(originName, replaceName);
+    public void addFieldInfos(EsFieldInfos esFieldInfos) {
+        if (enableKeywordSniff && esFieldInfos.getFieldsContext() != null) {
+            fieldsContext = esFieldInfos.getFieldsContext();
+        }
+        if (enableDocValueScan && esFieldInfos.getDocValueContext() != null) {
+            docValueContext = esFieldInfos.getDocValueContext();
+        }
     }
 
     public Map<String, String> fieldsContext() {
         return fieldsContext;
     }
 
-    public void addDocValueField(String name, String fieldsName) {
-        docValueContext.put(name, fieldsName);
-    }
-
     public Map<String, String> docValueContext() {
         return docValueContext;
     }
@@ -386,12 +391,12 @@ public PartitionInfo getPartitionInfo() {
         return partitionInfo;
     }
 
-    public EsTableState getEsTableState() {
-        return esTableState;
+    public EsTablePartitions getEsTablePartitions() {
+        return esTablePartitions;
     }
 
-    public void setEsTableState(EsTableState esTableState) {
-        this.esTableState = esTableState;
+    public void setEsTablePartitions(EsTablePartitions esTablePartitions) {
+        this.esTablePartitions = esTablePartitions;
     }
 
     public Throwable getLastMetaDataSyncException() {
@@ -401,4 +406,29 @@ public Throwable getLastMetaDataSyncException() {
     public void setLastMetaDataSyncException(Throwable lastMetaDataSyncException) {
         this.lastMetaDataSyncException = lastMetaDataSyncException;
     }
+
+    /**
+     * sync es index meta from remote
+     * @param client esRestClient
+     */
+    public void syncESIndexMeta(EsRestClient client) {
+        try {
+            EsFieldInfos fieldInfos = client.getFieldInfos(this.indexName, this.mappingType, this.fullSchema);
+            EsShardPartitions esShardPartitions = client.getShardPartitions(this.indexName);
+            Map<String, EsNodeInfo> nodesInfo = client.getHttpNodes();
+            if (this.enableKeywordSniff || this.enableDocValueScan) {
+                addFieldInfos(fieldInfos);
+            }
+
+            this.esTablePartitions = EsTablePartitions.fromShardPartitions(this, esShardPartitions);
+
+            if (EsTable.TRANSPORT_HTTP.equals(getTransport())) {
+                this.esTablePartitions.addHttpAddress(nodesInfo);
+            }
+        } catch (Throwable e) {
+            LOG.warn("Exception happens when fetch index [{}] meta data from remote es cluster", this.name, e);
+            this.esTablePartitions = null;
+            this.lastMetaDataSyncException = e;
+        }
+    }
 }

fe/src/main/java/org/apache/doris/common/proc/EsPartitionsProcDir.java

@@ -24,7 +24,7 @@
 import org.apache.doris.catalog.RangePartitionInfo;
 import org.apache.doris.catalog.Table.TableType;
 import org.apache.doris.common.AnalysisException;
-import org.apache.doris.external.elasticsearch.EsIndexState;
+import org.apache.doris.external.elasticsearch.EsShardPartitions;
 
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
@@ -66,33 +66,33 @@ public ProcResult fetchResult() throws AnalysisException {
         try {
             RangePartitionInfo rangePartitionInfo = null;
             if (esTable.getPartitionInfo().getType() == PartitionType.RANGE) {
-                rangePartitionInfo = (RangePartitionInfo) esTable.getEsTableState().getPartitionInfo();
+                rangePartitionInfo = (RangePartitionInfo) esTable.getEsTablePartitions().getPartitionInfo();
             }
             Joiner joiner = Joiner.on(", ");
-            Map<String, EsIndexState> unPartitionedIndices = esTable.getEsTableState().getUnPartitionedIndexStates();
-            Map<String, EsIndexState> partitionedIndices = esTable.getEsTableState().getPartitionedIndexStates();
-            for (EsIndexState esIndexState : unPartitionedIndices.values()) {
+            Map<String, EsShardPartitions> unPartitionedIndices = esTable.getEsTablePartitions().getUnPartitionedIndexStates();
+            Map<String, EsShardPartitions> partitionedIndices = esTable.getEsTablePartitions().getPartitionedIndexStates();
+            for (EsShardPartitions esShardPartitions : unPartitionedIndices.values()) {
                 List<Comparable> partitionInfo = new ArrayList<Comparable>();
-                partitionInfo.add(esIndexState.getIndexName());
+                partitionInfo.add(esShardPartitions.getIndexName());
                 partitionInfo.add("-");  // partition key
                 partitionInfo.add("-");  // range
                 partitionInfo.add("-");  // dis
-                partitionInfo.add(esIndexState.getShardRoutings().size());  // shards
+                partitionInfo.add(esShardPartitions.getShardRoutings().size());  // shards
                 partitionInfo.add(1);  //  replica num
                 partitionInfos.add(partitionInfo);
             }
-            for (EsIndexState esIndexState : partitionedIndices.values()) {
+            for (EsShardPartitions esShardPartitions : partitionedIndices.values()) {
                 List<Comparable> partitionInfo = new ArrayList<Comparable>();
-                partitionInfo.add(esIndexState.getIndexName());
+                partitionInfo.add(esShardPartitions.getIndexName());
                 List<Column> partitionColumns = rangePartitionInfo.getPartitionColumns();
                 List<String> colNames = new ArrayList<String>();
                 for (Column column : partitionColumns) {
                     colNames.add(column.getName());
                 }
                 partitionInfo.add(joiner.join(colNames));  // partition key
-                partitionInfo.add(rangePartitionInfo.getRange(esIndexState.getPartitionId()).toString()); // range
+                partitionInfo.add(rangePartitionInfo.getRange(esShardPartitions.getPartitionId()).toString()); // range
                 partitionInfo.add("-");  // dis
-                partitionInfo.add(esIndexState.getShardRoutings().size());  // shards
+                partitionInfo.add(esShardPartitions.getShardRoutings().size());  // shards
                 partitionInfo.add(1);  //  replica num
                 partitionInfos.add(partitionInfo);
             }

fe/src/main/java/org/apache/doris/common/proc/EsShardProcDir.java

@@ -25,7 +25,7 @@
 import org.apache.doris.catalog.EsTable;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.util.ListComparator;
-import org.apache.doris.external.elasticsearch.EsIndexState;
+import org.apache.doris.external.elasticsearch.EsShardPartitions;
 import org.apache.doris.external.elasticsearch.EsShardRouting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
@@ -55,9 +55,9 @@ public ProcResult fetchResult() {
         db.readLock();
         try {
             // get infos
-            EsIndexState esIndexState = esTable.getEsTableState().getIndexState(indexName);
-            for (int shardId : esIndexState.getShardRoutings().keySet()) {
-                List<EsShardRouting> shardRoutings = esIndexState.getShardRoutings().get(shardId);
+            EsShardPartitions esShardPartitions = esTable.getEsTablePartitions().getEsShardPartitions(indexName);
+            for (int shardId : esShardPartitions.getShardRoutings().keySet()) {
+                List<EsShardRouting> shardRoutings = esShardPartitions.getShardRoutings().get(shardId);
                 if (shardRoutings != null && shardRoutings.size() > 0) {
                     for (EsShardRouting esShardRouting : shardRoutings) {
                         List<Comparable> shardInfo = new ArrayList<Comparable>();