branch-3.1: [improve](planner) select * unique_table limit n should use one instance apache#53948 (apache#54186)

Cherry-picked from apache#53948

Co-authored-by: zhangstar333 <[email protected]>

Author: github-actions[bot]

be/src/pipeline/exec/scan_operator.cpp

@@ -1213,7 +1213,9 @@ Status ScanOperatorX<LocalStateType>::init(const TPlanNode& tnode, RuntimeState*
         // is checked in previous branch.
         if (query_options.enable_adaptive_pipeline_task_serial_read_on_limit) {
             DCHECK(query_options.__isset.adaptive_pipeline_task_serial_read_on_limit);
-            if (!tnode.__isset.conjuncts || tnode.conjuncts.empty()) {
+            if (!tnode.__isset.conjuncts || tnode.conjuncts.empty() ||
+                (tnode.conjuncts.size() == 1 && tnode.__isset.olap_scan_node &&
+                 tnode.olap_scan_node.keyType == TKeysType::UNIQUE_KEYS)) {
                 if (tnode.limit > 0 &&
                     tnode.limit <= query_options.adaptive_pipeline_task_serial_read_on_limit) {
                     _should_run_serial = true;

fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java

@@ -38,6 +38,7 @@
 import org.apache.doris.analysis.TupleDescriptor;
 import org.apache.doris.analysis.TupleId;
 import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.KeysType;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.PartitionInfo;
 import org.apache.doris.catalog.PrimitiveType;
@@ -765,8 +766,30 @@ public boolean shouldUseOneInstance(ConnectContext ctx) {
             // No connection context, typically for broker load.
         }
 
-        // For UniqueKey table, we will use multiple instance.
-        return hasLimit() && getLimit() <= adaptivePipelineTaskSerialReadOnLimit && conjuncts.isEmpty();
+        if (hasLimit() && getLimit() <= adaptivePipelineTaskSerialReadOnLimit) {
+            if (conjuncts.isEmpty()) {
+                return true;
+            } else {
+                if (this instanceof OlapScanNode) {
+                    OlapScanNode olapScanNode = (OlapScanNode) this;
+                    if (olapScanNode.getOlapTable() != null
+                            && olapScanNode.getOlapTable().getKeysType() == KeysType.UNIQUE_KEYS) {
+                        // If the table is unique keys, we can check if the conjuncts only contains
+                        // delete sign
+                        if (conjuncts.size() == 1 && conjuncts.get(0) instanceof BinaryPredicate) {
+                            BinaryPredicate binaryPredicate = (BinaryPredicate) conjuncts.get(0);
+                            if (binaryPredicate.getOp() == BinaryPredicate.Operator.EQ
+                                    && binaryPredicate.getChild(0) instanceof SlotRef
+                                    && ((SlotRef) binaryPredicate.getChild(0)).getDesc().getColumn().getName()
+                                            .equals(Column.DELETE_SIGN)) {
+                                return true;
+                            }
+                        }
+                    }
+                }
+            }
+        }
+        return false;
     }
 
     // In cloud mode, meta read lock is not enough to keep a snapshot of the partition versions.