增加一些注释

Author: dengdejiushini

elasticsearch6/elasticsearch6-side/elasticsearch6-all-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AllReqRow.java

@@ -159,6 +159,7 @@ protected void initCache() throws SQLException {
         Map<String, List<Map<String, Object>>> newCache = Maps.newConcurrentMap();
         cacheRef.set(newCache);
         try {
+            // create search request and build where cause
             searchRequest = Es6Util.setSearchRequest(sideInfo);
             boolQueryBuilder = Es6Util.setPredicateclause(sideInfo);
             loadData(newCache);
@@ -217,6 +218,7 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws IO
         }
     }
 
+    // initialize searchSourceBuilder
     private SearchSourceBuilder initConfiguration(BoolQueryBuilder boolQueryBuilder){
             SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
             if (boolQueryBuilder != null) {
@@ -225,6 +227,8 @@ private SearchSourceBuilder initConfiguration(BoolQueryBuilder boolQueryBuilder)
 
             searchSourceBuilder.size(getFetchSize());
             searchSourceBuilder.sort("_id", SortOrder.DESC);
+
+            // fields included in the source data
             String[] sideFieldNames = StringUtils.split(sideInfo.getSideSelectFields().trim(), ",");
             searchSourceBuilder.fetchSource(sideFieldNames, null);
             return searchSourceBuilder;
@@ -240,6 +244,7 @@ private void searchData(SearchSourceBuilder searchSourceBuilder, Map<String, Lis
         while (true) {
             try {
                 if (searchAfterParameter != null) {
+                    // set search mark
                     searchSourceBuilder.searchAfter(searchAfterParameter);
                 }
                 searchRequest.source(searchSourceBuilder);
@@ -258,6 +263,7 @@ private void searchData(SearchSourceBuilder searchSourceBuilder, Map<String, Lis
         }
     }
 
+    // data load to cache
     private void loadToCache(SearchHit[] searchHits, Map<String, List<Map<String, Object>>> tmpCache) {
         String[] sideFieldNames = StringUtils.split(sideInfo.getSideSelectFields().trim(), ",");
         String[] sideFieldTypes = sideInfo.getSideTableInfo().getFieldTypes();

elasticsearch6/elasticsearch6-side/elasticsearch6-async-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AsyncReqRow.java

@@ -136,10 +136,12 @@ public void onResponse(SearchResponse searchResponse) {
                     try {
                         while (true) {
                             loadDataToCache(searchHits, rowList, cacheContent, copyCrow);
+                            // determine if all results haven been ferched
                             if (searchHits.length < getFetchSize()) {
                                 break;
                             }
                             if (tableInfo == null && tmpRhlClient == null) {
+                                // create new connection to fetch data
                                 tableInfo = (Elasticsearch6SideTableInfo) sideInfo.getSideTableInfo();
                                 tmpRhlClient = Es6Util.getClient(tableInfo.getAddress(), tableInfo.isAuthMesh(), tableInfo.getUserName(), tableInfo.getPassword());
                             }
@@ -187,7 +189,6 @@ private void loadDataToCache(SearchHit[] searchHits, List<CRow> rowList, List<Ob
         rowList.addAll(getRows(copyCrow, cacheContent, results));
     }
 
-
     protected List<CRow> getRows(CRow inputRow, List<Object> cacheContent, List<Object> results) {
         List<CRow> rowList = Lists.newArrayList();
         for (Object line : results) {
@@ -257,7 +258,6 @@ private SearchSourceBuilder initConfiguration() {
         return searchSourceBuilder;
     }
 
-
     private BoolQueryBuilder setInputParams(List<Object> inputParams, BoolQueryBuilder boolQueryBuilder) {
         if (boolQueryBuilder == null) {
             boolQueryBuilder = new BoolQueryBuilder();
@@ -273,7 +273,6 @@ private BoolQueryBuilder setInputParams(List<Object> inputParams, BoolQueryBuild
         return boolQueryBuilder;
     }
 
-
     public int getFetchSize() {
         return 1000;
     }

elasticsearch6/elasticsearch6-side/elasticsearch6-side-core/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/util/Es6Util.java

@@ -51,6 +51,7 @@ public class Es6Util {
     private static final String KEY_WORD_TYPE = ".keyword";
     private static final String APOSTROPHE = "\'";
 
+    // connect to the elasticsearch
     public static RestHighLevelClient getClient(String esAddress, Boolean isAuthMesh, String userName, String password) {
         List<HttpHost> httpHostList = new ArrayList<>();
         String[] address = StringUtils.split(esAddress, ",");
@@ -96,6 +97,7 @@ public static RestHighLevelClient getClient(String esAddress, Boolean isAuthMesh
         return rhlClient;
     }
 
+    // add index and type to search request
     public static SearchRequest setSearchRequest(SideInfo sideInfo) {
         SearchRequest searchRequest = new SearchRequest();
         Elasticsearch6SideTableInfo tableInfo = (Elasticsearch6SideTableInfo) sideInfo.getSideTableInfo();
@@ -126,6 +128,7 @@ public static SearchRequest setSearchRequest(SideInfo sideInfo) {
         return searchRequest;
     }
 
+    // build where cause
     public static BoolQueryBuilder setPredicateclause(SideInfo sideInfo) {
 
         BoolQueryBuilder boolQueryBuilder = null;
@@ -140,6 +143,7 @@ public static BoolQueryBuilder setPredicateclause(SideInfo sideInfo) {
         return boolQueryBuilder;
     }
 
+    // build filter condition
     public static BoolQueryBuilder buildFilterCondition(BoolQueryBuilder boolQueryBuilder, PredicateInfo info, SideInfo sideInfo) {
         switch (info.getOperatorKind()) {
             case "IN":
@@ -184,6 +188,7 @@ public static BoolQueryBuilder buildFilterCondition(BoolQueryBuilder boolQueryBu
 
     }
 
+    // remove extra spaces and apostrophes
     public static String[] removeSpaceAndApostrophe(String str) {
         String[] split = StringUtils.split(str, ",");
         for (int i = 0; i < split.length; i++) {
@@ -196,6 +201,7 @@ public static String[] removeSpaceAndApostrophe(String str) {
         return split;
     }
 
+    // prevent word segmentation
     public static String textConvertToKeyword(String fieldName, SideInfo sideInfo) {
         String[] sideFieldTypes = sideInfo.getSideTableInfo().getFieldTypes();
         int fieldIndex = sideInfo.getSideTableInfo().getFieldList().indexOf(fieldName.trim());