适配多biz多partition查询

aios/autil/autil/ThreadPool.cpp

@@ -197,7 +197,7 @@ ThreadPool::ERROR_TYPE ThreadPool::pushWorkItem(WorkItem *item, bool isBlocked)
 
     ScopedLock lock(_cond);
     if (!isBlocked && _queue->size() >= _queueSize) {
-        AUTIL_LOG(INFO,
+        AUTIL_LOG(DEBUG,
                   "thread pool [%s] is full, queueSize [%lu], active thread [%d]",
                   _threadName.c_str(),
                   _queueSize,

aios/ha3/ha3/search/query_executor/IndexPartitionReaderUtil.cpp

@@ -53,7 +53,7 @@ IndexPartitionReaderWrapperPtr IndexPartitionReaderUtil::createIndexPartitionRea
     bool usePartial) {
     auto tabletReader = partitionReaderSnapshot->GetTabletReader(mainTableName);
     if (tabletReader) {
-        AUTIL_LOG(DEBUG,
+        AUTIL_LOG(INFO,
                   "create index partition reader wrapper with tabletReader,"
                   " mainTableName[%s], usePartial[%d]",
                   mainTableName.c_str(),
@@ -64,16 +64,20 @@ IndexPartitionReaderWrapperPtr IndexPartitionReaderUtil::createIndexPartitionRea
             return make_shared<IndexPartitionReaderWrapper>(tabletReader);
         }
     } else {
+        AUTIL_LOG(INFO, "createIndexPartitionReaderWrapper for table 1 %s", mainTableName.c_str());
         const auto &tableMainSubIdxMap = partitionReaderSnapshot->getTableMainSubIdxMap();
         auto iter = tableMainSubIdxMap.find(mainTableName);
         if (iter == tableMainSubIdxMap.end()) {
             IndexPartitionReaderPtr mainIndexPartReader
                 = partitionReaderSnapshot->GetIndexPartitionReader(mainTableName);
             if (mainIndexPartReader == NULL) {
+                AUTIL_LOG(INFO, "createIndexPartitionReaderWrapper for table 2 %s", mainTableName.c_str());
                 return IndexPartitionReaderWrapperPtr();
             }
+            AUTIL_LOG(INFO, "createIndexPartitionReaderWrapper for table 3 %s", mainTableName.c_str());
             return createIndexPartitionReaderWrapper(mainIndexPartReader, usePartial);
         } else {
+            AUTIL_LOG(INFO, "createIndexPartitionReaderWrapper for table 4 %s", mainTableName.c_str());
             return createIndexPartitionReaderWrapper(&(iter->second.indexName2IdMap),
                                                      &(iter->second.attrName2IdMap),
                                                      &(iter->second.indexReaderVec),

aios/navi/engine/ResourceInitContext.cpp

@@ -73,6 +73,7 @@ ResourceInitContext::ResourceInitContext(
     , _namedDataMap(namedDataMap)
     , _requireKernelNode(requireKernelNode)
 {
+    NAVI_KERNEL_LOG(INFO, "ResourceInitContext _partId %d", partId);
 }
 
 ResourceInitContext::~ResourceInitContext() {

aios/sql/iquan/cpp/common/catalog/LocationDef.h

@@ -31,13 +31,15 @@ struct LocationSign {
     uint32_t partitionCnt = 0;
     std::string nodeName;
     std::string nodeType;
+    std::string tableName;
 };
 
 class LocationDef : public autil::legacy::Jsonizable {
 public:
     void Jsonize(autil::legacy::Jsonizable::JsonWrapper &json) override {
         json.Jsonize("partition_cnt", sign.partitionCnt, sign.partitionCnt);
         json.Jsonize("node_name", sign.nodeName, sign.nodeName);
+        json.Jsonize("location_table_name",sign.tableName,sign.tableName);
         json.Jsonize("node_type", sign.nodeType, sign.nodeType);
         json.Jsonize("tables", tableIdentities, tableIdentities);
         json.Jsonize("equivalent_hash_fields", equilvalentHashFields, equilvalentHashFields);

aios/sql/iquan/java/iquan_core/src/main/java/com/taobao/search/iquan/client/common/json/catalog/IquanLocation.java

@@ -16,6 +16,8 @@
 @Getter
 @NoArgsConstructor
 public class IquanLocation {
+    @JsonProperty(value = "location_table_name", required = false)
+    private String locationTableName;
     @JsonProperty(value = "node_name", required = true)
     private String nodeName;
     @JsonProperty(value = "partition_cnt", required = true)

aios/sql/iquan/java/iquan_core/src/main/java/com/taobao/search/iquan/core/api/schema/Location.java

@@ -7,6 +7,7 @@
 public class Location {
     private final String nodeName;
     private final int partitionCnt;
+    private String locationTableName;
 
     public static final Location DEFAULT_QRS = new Location( "qrs", 1);
     public static final Location UNKNOWN = new Location( "unknown", Integer.MAX_VALUE);
@@ -19,6 +20,7 @@ public Location(String nodeName, int partitionCnt) {
     public Location(IquanLocation iquanLocation) {
         this.nodeName = iquanLocation.getNodeName();
         this.partitionCnt = iquanLocation.getPartitionCnt();
+        this.locationTableName=iquanLocation.getLocationTableName();
     }
 
     public boolean isSingle() {

aios/sql/iquan/java/iquan_core/src/main/java/com/taobao/search/iquan/core/rel/ops/physical/IquanExchangeOp.java

@@ -66,6 +66,7 @@ public void explainInternal(final Map<String, Object> map, SqlExplainLevel level
                         ((IquanOptContext) inputRelNode.getCluster().getPlanner().getContext()).getExecutor().getDefaultDbName());
             }
             IquanRelOptUtils.addMapIfNotEmpty(map, ConstantDefine.NODE_NAME, ((IquanRelNode) inputRelNode).getLocation().getNodeName());
+            IquanRelOptUtils.addMapIfNotEmpty(map, "location_table_name", ((IquanRelNode) inputRelNode).getLocation().getLocationTableName());
             IquanRelOptUtils.addMapIfNotEmpty(map, ConstantDefine.TABLE_DISTRIBUTION, RelDistributionUtil.formatDistribution((IquanRelNode) inputRelNode, true));
             map.put(ConstantDefine.OUTPUT_PRUNABLE, outputPrunable);
 

aios/sql/iquan/java/iquan_core/src/main/java/com/taobao/search/iquan/core/rel/ops/physical/IquanRelNode.java

@@ -120,6 +120,7 @@ static void explainIquanRelNode(RelNode relNode, final Map<String, Object> map,
             if (location != null) {
                 Map<String, Object> locationMeta = new TreeMap<>();
                 locationMeta.put(ConstantDefine.NODE_NAME, location.getNodeName());
+                locationMeta.put("location_table_name", location.getLocationTableName());
                 locationMeta.put(ConstantDefine.PARTITION_CNT, location.getPartitionCnt());
                 map.put(ConstantDefine.LOCATION, locationMeta);
             }

aios/sql/iquan/java/iquan_core/src/main/java/com/taobao/search/iquan/core/rel/ops/physical/IquanTableScanBase.java

@@ -412,6 +412,7 @@ public void explainInternal(final Map<String, Object> map, SqlExplainLevel level
         if (isRemoteScan && !Objects.isNull(location)) {
             Map<String, Object> locationMeta = new TreeMap<>();
             locationMeta.put(ConstantDefine.NODE_NAME, location.getNodeName());
+            locationMeta.put("location_table_name", location.getLocationTableName());
             locationMeta.put(ConstantDefine.PARTITION_CNT, location.getPartitionCnt());
             IquanRelOptUtils.addMapIfNotEmpty(map, ConstantDefine.LOCATION, locationMeta);
         }

aios/sql/misc/sql_alog.conf

@@ -45,7 +45,7 @@ alog.appender.KmonAppender.log_keep_count=10
 alog.appender.KmonAppender.layout=PatternLayout
 alog.appender.KmonAppender.layout.LogPattern=[%%d] [%%l] [%%t,%%F -- %%f():%%n] [%%m]
 
-alog.logger.multi_call=INFO, GigAppender
+alog.logger.multi_call=DEBUG, GigAppender
 inherit.multi_call=false
 alog.appender.GigAppender=FileAppender
 alog.appender.GigAppender.fileName=logs/gig/gig.log

aios/sql/ops/delayDp/kernel/DelayDpKernel.cpp

@@ -288,6 +288,10 @@ bool DelayDpKernel::partHasData(int partId) {
         if (info.partRows.empty()) {
             continue;
         }
+        SQL_LOG(WARN, "partHasData size %ld partId %d", info.partRows.size(), partId);
+        if (partId >= info.partRows.size()){
+            continue;
+        }
         const auto &rows = info.partRows[partId];
         if (rows.empty()) {
             return false;

aios/sql/ops/planTransform/GraphTransform.cpp

@@ -336,6 +336,7 @@ const string GraphTransform::EMPTY_STRING = "";
 GraphTransformEnv::GraphTransformEnv() {
     disableWatermark = autil::EnvUtil::getEnv("disableWatermark", disableWatermark);
     useQrsTimestamp = autil::EnvUtil::getEnv("useQrsTimestamp", useQrsTimestamp);
+    searchNewBiz = autil::EnvUtil::getEnv("searchNewBiz", true);
 }
 
 GraphTransformEnv::~GraphTransformEnv() {}
@@ -562,6 +563,10 @@ std::string GraphTransform::getRemoteBizName(iquan::PlanOp &op) {
                 return getQrsBizName();
             }
         }
+        auto tableName = getJsonStringValue(op, "table_name");
+        if (GraphTransformEnv::get().searchNewBiz && tableName.empty() == false) {
+            return name + "." + tableName + ".write";
+        }
         return name + "." + isearch::DEFAULT_SQL_BIZ_NAME;
     } else {
         return _config.searcherBizName;
@@ -804,7 +809,9 @@ plan::PlanNode *GraphTransform::linkExchangeNode(plan::ExchangeNode &node, plan:
         subRoot->setInlineMode(true);
     }
     auto &op = *(node.op);
-    subRoot->setBizName(getRemoteBizName(op));
+    auto remoteBizName=getRemoteBizName(op);
+    SQL_LOG(DEBUG, "getRemoteBizName is %s",remoteBizName.c_str());
+    subRoot->setBizName(remoteBizName);
     subRoot->setCurDist(getJsonSegment(op, "table_distribution"));
     auto remoteDist = getJsonSegment(op, "output_distribution");
     if (remoteDist.empty()) {

aios/sql/ops/planTransform/GraphTransform.h

@@ -211,6 +211,7 @@ struct GraphTransformEnv {
 public:
     bool disableWatermark = false;
     bool useQrsTimestamp = true;
+    bool searchNewBiz = true;
 };
 
 class GraphTransform : public plan::NodeVisitor {

aios/sql/ops/scan/AttributeExpressionCreatorR.cpp

@@ -76,6 +76,10 @@ bool AttributeExpressionCreatorR::initExpressionCreator() {
     _indexPartitionReaderWrapper
         = isearch::search::IndexPartitionReaderUtil::createIndexPartitionReaderWrapper(
             _scanR->partitionReaderSnapshot.get(), _scanInitParamR->tableName);
+    if(!_indexPartitionReaderWrapper){
+        SQL_LOG(ERROR, "initExpressionCreator _indexPartitionReaderWrapper is null table is %s",_scanInitParamR->tableName.c_str() );
+        return false;
+    }
     _indexPartitionReaderWrapper->setSessionPool(_queryMemPoolR->getPool().get());
     if (!createExpressionCreator()) {
         return false;

aios/sql/python/sql_config_loader.py

@@ -89,10 +89,16 @@ def fill_local_config_path(param_map):
 
 def get_log_config():
     return {
-        "file_appenders": [
+        "file_appenders" : [
             {
                 "file_name": "logs/navi.log",
-                "log_level": "info",
+                "log_level": "TRACE3",
+                "bt_filters": [
+                ]
+            },
+            {
+                "file_name": "logs/navi_schduler.log",
+                "log_level": "SCHEDULE3",
                 "bt_filters": [
                 ]
             }