stalary commented on a change in pull request #3454:
URL: https://github.com/apache/incubator-doris/pull/3454#discussion_r430430020
##########
File path: fe/src/main/java/org/apache/doris/external/EsStateStore.java
##########
@@ -17,94 +17,95 @@
package org.apache.doris.external;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Range;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
import org.apache.doris.catalog.Catalog;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.EsTable;
-import org.apache.doris.catalog.PartitionInfo;
import org.apache.doris.catalog.PartitionKey;
import org.apache.doris.catalog.RangePartitionInfo;
import org.apache.doris.catalog.SinglePartitionInfo;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.Table.TableType;
-import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.util.MasterDaemon;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Range;
-
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.json.JSONObject;
-import java.io.IOException;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
-
-import okhttp3.Authenticator;
-import okhttp3.Call;
-import okhttp3.Credentials;
-import okhttp3.OkHttpClient;
-import okhttp3.Request;
-import okhttp3.Response;
-import okhttp3.Route;
-
/**
* it is used to call es api to get shard allocation state
*/
public class EsStateStore extends MasterDaemon {
+
private static final Logger LOG = LogManager.getLogger(EsStateStore.class);
private Map<Long, EsTable> esTables;
+ private Map<Long, EsRestClient> esClients;
+
public EsStateStore() {
super("es state store", Config.es_state_sync_interval_second * 1000);
esTables = Maps.newConcurrentMap();
+ esClients = Maps.newConcurrentMap();
}
public void registerTable(EsTable esTable) {
if (Catalog.isCheckpointThread()) {
return;
}
esTables.put(esTable.getId(), esTable);
- LOG.info("register a new table [{}] to sync list", esTable.toString());
+ esClients.put(esTable.getId(),
+ new EsRestClient(esTable.getSeeds(), esTable.getUserName(),
esTable.getPasswd()));
+ LOG.info("register a new table [{}] to sync list", esTable);
}
public void deRegisterTable(long tableId) {
esTables.remove(tableId);
+ esClients.remove(tableId);
LOG.info("deregister table [{}] from sync list", tableId);
}
-
+
@Override
protected void runAfterCatalogReady() {
for (EsTable esTable : esTables.values()) {
try {
- EsRestClient client = new EsRestClient(esTable.getSeeds(),
- esTable.getUserName(), esTable.getPasswd());
- // if user not specify the es version, try to get the remote
cluster versoin
- // in the future, we maybe need this version
- String indexMetaData =
client.getIndexMetaData(esTable.getIndexName());
- if (indexMetaData == null) {
+ EsRestClient client = esClients.get(esTable.getId());
+
+ if (esTable.isKeywordSniffEnable() ||
esTable.isDocValueScanEnable()) {
+ JSONObject properties =
client.getIndexProperties(esTable.getIndexName(), esTable.getMappingType());
Review comment:
We don't know the full structure of properties, and not using JSONObject
feels like no other good way to do it
##########
File path: fe/src/main/java/org/apache/doris/external/EsStateStore.java
##########
@@ -17,94 +17,95 @@
package org.apache.doris.external;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Range;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
import org.apache.doris.catalog.Catalog;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.EsTable;
-import org.apache.doris.catalog.PartitionInfo;
import org.apache.doris.catalog.PartitionKey;
import org.apache.doris.catalog.RangePartitionInfo;
import org.apache.doris.catalog.SinglePartitionInfo;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.Table.TableType;
-import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.util.MasterDaemon;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Range;
-
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.json.JSONObject;
-import java.io.IOException;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
-
-import okhttp3.Authenticator;
-import okhttp3.Call;
-import okhttp3.Credentials;
-import okhttp3.OkHttpClient;
-import okhttp3.Request;
-import okhttp3.Response;
-import okhttp3.Route;
-
/**
* it is used to call es api to get shard allocation state
*/
public class EsStateStore extends MasterDaemon {
+
private static final Logger LOG = LogManager.getLogger(EsStateStore.class);
private Map<Long, EsTable> esTables;
+ private Map<Long, EsRestClient> esClients;
+
public EsStateStore() {
super("es state store", Config.es_state_sync_interval_second * 1000);
esTables = Maps.newConcurrentMap();
+ esClients = Maps.newConcurrentMap();
}
public void registerTable(EsTable esTable) {
if (Catalog.isCheckpointThread()) {
return;
}
esTables.put(esTable.getId(), esTable);
- LOG.info("register a new table [{}] to sync list", esTable.toString());
+ esClients.put(esTable.getId(),
+ new EsRestClient(esTable.getSeeds(), esTable.getUserName(),
esTable.getPasswd()));
+ LOG.info("register a new table [{}] to sync list", esTable);
}
public void deRegisterTable(long tableId) {
esTables.remove(tableId);
+ esClients.remove(tableId);
LOG.info("deregister table [{}] from sync list", tableId);
}
-
+
@Override
protected void runAfterCatalogReady() {
for (EsTable esTable : esTables.values()) {
try {
- EsRestClient client = new EsRestClient(esTable.getSeeds(),
- esTable.getUserName(), esTable.getPasswd());
- // if user not specify the es version, try to get the remote
cluster versoin
- // in the future, we maybe need this version
- String indexMetaData =
client.getIndexMetaData(esTable.getIndexName());
- if (indexMetaData == null) {
+ EsRestClient client = esClients.get(esTable.getId());
+
+ if (esTable.isKeywordSniffEnable() ||
esTable.isDocValueScanEnable()) {
+ JSONObject properties =
client.getIndexProperties(esTable.getIndexName(), esTable.getMappingType());
+ if (properties == null) {
+ continue;
+ }
+ setEsTableContext(properties, esTable);
+ }
+
+ EsIndexState esIndexState =
client.getIndexState(esTable.getIndexName());
+ if (esIndexState == null) {
continue;
}
- EsTableState esTableState = parseClusterState55(indexMetaData,
esTable);
+
+ EsTableState esTableState =
setTableStatePartitionInfo(esTable, esIndexState);
Review comment:
This is an old piece of code that I'm not familiar with
----------------------------------------------------------------
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:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
