morrySnow commented on code in PR #63366:
URL: https://github.com/apache/doris/pull/63366#discussion_r3372699227


##########
fe/fe-core/src/main/java/org/apache/doris/planner/HashJoinNode.java:
##########
@@ -285,4 +302,71 @@ public List<Expr> getOtherJoinConjuncts() {
     public List<Expr> getMarkJoinConjuncts() {
         return markJoinConjuncts;
     }
+
+    @Override
+    public Pair<PlanNode, LocalExchangeType> enforceAndDeriveLocalExchange(
+            PlanTranslatorContext translatorContext, PlanNode parent, 
LocalExchangeTypeRequire parentRequire) {
+
+        LocalExchangeTypeRequire probeSideRequire;
+        LocalExchangeTypeRequire buildSideRequire;
+        LocalExchangeType outputType = null;
+
+        if (joinOp == JoinOperator.NULL_AWARE_LEFT_ANTI_JOIN) {
+            buildSideRequire = probeSideRequire = 
LocalExchangeTypeRequire.noRequire();
+            outputType = LocalExchangeType.NOOP;
+        } else if (distrMode == DistributionMode.BROADCAST) {
+            // BE: _child->is_serial_operator() ? PASSTHROUGH/PASS_TO_ONE : 
NOOP
+            boolean probeChildSerial = children.get(0).isSerialOperatorOnBe(
+                    translatorContext.getConnectContext());
+            boolean buildChildSerial = children.get(1).isSerialOperatorOnBe(
+                    translatorContext.getConnectContext());
+            probeSideRequire = probeChildSerial

Review Comment:
   [P2] Preserve `enable_broadcast_join_force_passthrough`
   
   The BE planner returns `PASSTHROUGH` unconditionally for a broadcast-join 
probe when `state->enable_broadcast_join_force_passthrough()` is true 
(`HashJoinProbeOperatorX::required_data_distribution`). This FE branch only 
checks whether the probe child is serial. Once `enable_local_shuffle_planner` 
is enabled, BE local-exchange planning is skipped, so a non-serial broadcast 
join silently ignores this session variable and no balancing exchange is 
inserted. Please include the session flag in `probeSideRequire` and add a 
plan-shape assertion for the true/false cases.



##########
fe/fe-core/src/main/java/org/apache/doris/planner/AddLocalExchange.java:
##########
@@ -0,0 +1,182 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.planner;
+
+import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.glue.translator.PlanTranslatorContext;
+import org.apache.doris.nereids.trees.plans.distribute.DistributedPlan;
+import org.apache.doris.nereids.trees.plans.distribute.FragmentIdMapping;
+import org.apache.doris.nereids.trees.plans.distribute.PipelineDistributedPlan;
+import org.apache.doris.planner.LocalExchangeNode.LocalExchangeType;
+import org.apache.doris.planner.LocalExchangeNode.LocalExchangeTypeRequire;
+import org.apache.doris.planner.LocalExchangeNode.RequireHash;
+
+/**
+ * FE-side local exchange planner — inserts {@link LocalExchangeNode} into 
each fragment's
+ * plan tree so that within-fragment data redistribution is decided at 
planning time
+ * instead of at BE pipeline-build time.
+ *
+ * <h3>When this runs</h3>
+ * Invoked from {@code NereidsPlanner.addLocalExchangeAfterDistribute()} right 
after
+ * {@code DistributePlanner} has assigned instances to fragments and before 
the plan is
+ * serialized to BE.  Gated by session variable {@code 
enable_local_shuffle_planner}
+ * (default true) and {@code enable_local_shuffle}; when either is off this 
pass is
+ * skipped entirely and BE falls back to its own {@code _plan_local_exchange}. 
 The two
+ * paths are mutually exclusive: BE consults {@code 
runtime_state.h::plan_local_shuffle()}
+ * to know whether it should plan LE itself.
+ *
+ * <h3>What it changes</h3>
+ * <ul>
+ *   <li>For each fragment with {@code maxPerBeInstances > 1}, walks the plan 
tree
+ *       bottom-up via {@link PlanNode#enforceAndDeriveLocalExchange} and 
inserts
+ *       LocalExchangeNodes where children's output distribution doesn't 
satisfy the
+ *       parent's requirement.</li>
+ *   <li>May wrap the fragment root with an extra PASSTHROUGH LE so the data 
sink
+ *       (DataStreamSink / OlapTableSink) runs with the full instance count 
even when
+ *       the root operator is serial — see {@link 
#addLocalExchangeForFragment}.</li>
+ *   <li>Does NOT modify the fragment sink itself, fragment boundaries, or 
instance
+ *       assignment.</li>
+ * </ul>
+ *
+ * <h3>Per-BE instance semantics</h3>
+ * Skips fragments where every BE has at most 1 instance.  Using a global 
instance count
+ * would insert LE for "2 BEs × 1 instance" cases, which BE's own
+ * {@code _plan_local_exchange} would not — leading to pipeline task-count 
mismatch and
+ * deadlock.  See {@link #addLocalExchange}.
+ *
+ * <h3>Reading order</h3>
+ * Start with {@link PlanNode#enforceRequire} (the recursion engine), then 
individual
+ * {@code enforceAndDeriveLocalExchange} overrides on PlanNode subclasses.
+ */
+public class AddLocalExchange {
+    /** addLocalExchange with distributed plans, skipping single-instance 
fragments.
+     *  BE's _plan_local_exchange checks _num_instances which is the per-BE 
instance count.
+     *  With _num_instances<=1 all pipelines on that BE have 1 task so local 
exchange is a no-op.
+     *  We must use the same per-BE semantics: skip when every BE has at most 
1 instance.
+     *  Using global instanceCount would insert LE for fragments where 2 BEs 
each have 1 instance
+     *  (global=2, per-BE=1), causing pipeline task mismatch and deadlock. */
+    public void addLocalExchange(FragmentIdMapping<DistributedPlan> 
distributedPlans,
+            PlanTranslatorContext context) {
+        for (DistributedPlan plan : distributedPlans.values()) {
+            PipelineDistributedPlan pipePlan = (PipelineDistributedPlan) plan;
+            long maxPerBeInstances = pipePlan.getInstanceJobs().stream()
+                    .collect(java.util.stream.Collectors.groupingBy(
+                            j -> j.getAssignedWorker().id(), 
java.util.stream.Collectors.counting()))
+                    
.values().stream().mapToLong(Long::longValue).max().orElse(0);
+            if (maxPerBeInstances <= 1) {
+                continue;
+            }
+            PlanFragment fragment = pipePlan.getFragmentJob().getFragment();
+            addLocalExchangeForFragment(fragment, context);
+        }
+    }
+
+    private void addLocalExchangeForFragment(PlanFragment fragment, 
PlanTranslatorContext context) {
+        DataSink sink = fragment.getSink();
+        LocalExchangeTypeRequire require = sink == null
+                ? LocalExchangeTypeRequire.noRequire() : 
sink.getLocalExchangeTypeRequire();
+        PlanNode root = fragment.getPlanRoot();
+        context.setHasSerialAncestorInPipeline(root, false);
+        Pair<PlanNode, LocalExchangeType> output = root
+                .enforceAndDeriveLocalExchange(context, null, require);
+        PlanNode newRoot = output.first;
+        // The fragment data sink (DataStreamSink, OlapTableSink) runs in the 
same pipeline
+        // as the root. If the root will be serial on BE, the sink pipeline 
has 1 task —
+        // only instance 0 sends data, others hang or miss writes.
+        // Insert PASSTHROUGH fan-out so sink runs with _num_instances tasks.
+        // This matches BE-native's default required_data_distribution():
+        //   _child->is_serial_operator() ? PASSTHROUGH : NOOP
+        if (newRoot.isSerialOperatorOnBe(context.getConnectContext())) {
+            newRoot = new LocalExchangeNode(context.nextPlanNodeId(), newRoot,
+                    LocalExchangeType.PASSTHROUGH, null);
+        }
+        if (newRoot != root) {
+            fragment.setPlanRoot(newRoot);
+        }
+        validateNoSerialWithoutLocalExchange(fragment.getPlanRoot(), 
context.getConnectContext());
+    }
+
+    /**
+     * In a local-shuffle fragment, the root check above guarantees the root 
pipeline
+     * has N tasks. Any serial operator reduces its pipeline to 1 task. If 
this serial
+     * operator feeds into a non-serial parent without LocalExchangeNode in 
between,
+     * some pipelines have 1 task while others have N → shared_state mismatch, 
data loss.
+     *
+     * Serial→serial chains are fine (all at 1 task, consistent). Only the 
transition
+     * from serial to non-serial needs LE to restore parallelism.
+     */
+    private void validateNoSerialWithoutLocalExchange(PlanNode node,
+            org.apache.doris.qe.ConnectContext context) {
+        for (PlanNode child : node.getChildren()) {
+            validateNoSerialWithoutLocalExchange(child, context);
+            if (child.isSerialOperatorOnBe(context)
+                    && !(child instanceof LocalExchangeNode)
+                    && !(node instanceof LocalExchangeNode)
+                    && !(node instanceof ExchangeNode)
+                    && !node.isSerialOperatorOnBe(context)) {
+                
org.apache.logging.log4j.LogManager.getLogger(AddLocalExchange.class).warn(

Review Comment:
   [P1] Reject invalid serial-to-parallel plans instead of logging
   
   This predicate identifies a definite task-count mismatch: the child pipeline 
runs one task while the parent runs `_num_instances`. The comments and 
regression cases in this PR document that executing such a shape can hang, lose 
rows, or initialize shared state inconsistently. Logging a warning and 
continuing still sends the known-invalid plan to BE, potentially returning 
incorrect results. Please throw a planning error/assert here (or repair the 
plan) rather than continue.



##########
regression-test/suites/nereids_p0/local_shuffle/test_local_shuffle_rqg_bugs.groovy:
##########
@@ -0,0 +1,1557 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+/**
+ * Regression tests for bugs discovered by RQG testing on the local-exchange2 
branch.
+ *
+ * These queries triggered "must set shared state" errors or incorrect results
+ * in RQG build 183992.  Common conditions:
+ *   - use_serial_exchange=true  (makes ALL Exchanges serial, not just 
UNPARTITIONED)
+ *   - enable_local_shuffle_planner=true (FE-planned local exchange)
+ *   - parallel_pipeline_task_num > 1
+ *
+ * Error types reproduced:
+ *   1. must set shared state, in AGGREGATION_OPERATOR
+ *   2. must set shared state, in SORT_OPERATOR
+ *   3. incorrect results with GROUPING SETS + scalar subquery + window 
function
+ */
+suite("test_local_shuffle_rqg_bugs") {
+
+    // ============================================================
+    //  Table setup — mirrors RQG table structure
+    //  10 buckets to match RQG (replication_num=1 for single-BE testing)
+    // ============================================================
+    sql "DROP TABLE IF EXISTS rqg_t1"
+    sql "DROP TABLE IF EXISTS rqg_t2"
+    sql "DROP TABLE IF EXISTS rqg_t3"
+    sql "DROP TABLE IF EXISTS rqg_t4"
+
+    sql """
+        CREATE TABLE rqg_t1 (
+            pk INT NOT NULL,
+            col_int_undef_signed INT,
+            col_int_undef_signed2 INT,
+            col_int_undef_signed_not_null INT NOT NULL
+        ) ENGINE=OLAP
+        DUPLICATE KEY(pk)
+        DISTRIBUTED BY HASH(pk) BUCKETS 10
+        PROPERTIES ("replication_num" = "1")
+    """
+
+    sql """
+        CREATE TABLE rqg_t2 (
+            pk INT NOT NULL,
+            col_int_undef_signed INT,
+            col_int_undef_signed2 INT,
+            col_bigint_undef_signed_not_null BIGINT NOT NULL,
+            col_decimal_38_10__undef_signed_not_null DECIMAL(38,10) NOT NULL
+        ) ENGINE=OLAP
+        DUPLICATE KEY(pk)
+        DISTRIBUTED BY HASH(pk) BUCKETS 10
+        PROPERTIES ("replication_num" = "1")
+    """
+
+    // Table for build 184181 GLOBAL_HASH_SHUFFLE bugs — needs varchar + 
bigint columns
+    sql """
+        CREATE TABLE rqg_t3 (
+            pk INT NOT NULL,
+            col_bigint_undef_signed BIGINT,
+            col_varchar_10__undef_signed VARCHAR(10),
+            col_varchar_64__undef_signed VARCHAR(64)
+        ) ENGINE=OLAP
+        DUPLICATE KEY(pk)
+        DISTRIBUTED BY HASH(pk) BUCKETS 10
+        PROPERTIES ("replication_num" = "1")
+    """
+
+    // Second table for FULL OUTER JOIN case (col_bigint_undef_signed_not_null)
+    sql """
+        CREATE TABLE rqg_t4 (
+            pk INT NOT NULL,
+            col_bigint_undef_signed_not_null BIGINT NOT NULL
+        ) ENGINE=OLAP
+        DUPLICATE KEY(pk)
+        DISTRIBUTED BY HASH(pk) BUCKETS 10
+        PROPERTIES ("replication_num" = "1")
+    """
+
+    sql """
+        INSERT INTO rqg_t3 VALUES
+            (0, -94, 'Abc', 'hello world'),
+            (1, 672609, 'Xyz', null),
+            (2, -3766684, 'Pqr', 'test string'),
+            (3, 5070261, 'abc', 'another row'),
+            (4, null, 'def', 'value four'),
+            (5, -86, 'XgpxlHBLEM', null),
+            (6, 21910, 'abc', 'they'),
+            (7, -63, 'zzzz', 'some text'),
+            (8, -8276281, 'AHlvNtoGLO', 'longer string here'),
+            (9, -101, 'mid', 'final row')
+    """
+
+    sql """
+        INSERT INTO rqg_t4 VALUES
+            (0, 0), (1, 1), (2, 2), (3, 3), (4, 4),
+            (5, 5), (6, 6), (7, 7), (8, 8), (9, 9),
+            (10, 2), (11, 2), (12, 2), (13, 3), (14, 4),
+            (15, 5), (16, 2), (17, 2), (18, 2), (19, 9)
+    """
+
+    // Insert enough rows to exercise multiple pipeline tasks
+    sql """
+        INSERT INTO rqg_t1 VALUES
+            (0, 0, 10, 0), (1, 1, 11, 1), (2, 2, 12, 2), (3, 3, 13, 3),
+            (4, 4, 14, 4), (5, 5, 15, 5), (6, 6, 16, 6), (7, 7, 17, 7),
+            (8, 8, 18, 8), (9, 9, 19, 9), (10, 0, 20, 10), (11, 1, 21, 11),
+            (12, 2, 22, 12), (13, 3, 23, 13), (14, 4, 24, 14), (15, 5, 25, 15),
+            (16, 6, 26, 16), (17, 7, 27, 17), (18, 8, 28, 18), (19, 9, 29, 19)
+    """
+
+    sql """
+        INSERT INTO rqg_t2 VALUES
+            (0, 0, 10, 100, 1.5), (1, 1, 11, 101, 2.5), (2, 2, 12, 102, 3.5),
+            (3, 3, 13, 103, 4.5), (4, 4, 14, 104, 5.5), (5, 5, 15, 105, 6.5),
+            (6, 6, 16, 106, 7.5), (7, 7, 17, 107, 8.5), (8, 8, 18, 108, 9.5),
+            (9, 9, 19, 109, 10.5), (10, 0, 20, 110, 11.5), (11, 1, 21, 111, 
12.5),
+            (12, 2, 22, 112, 13.5), (13, 3, 23, 113, 14.5), (14, 4, 24, 114, 
15.5),
+            (15, 5, 25, 115, 16.5), (16, 6, 26, 116, 17.5), (17, 7, 27, 117, 
18.5),
+            (18, 8, 28, 118, 19.5), (19, 9, 29, 119, 20.5)
+    """
+
+    // Wait for data to be visible
+    Thread.sleep(5000)
+
+    // ============================================================
+    //  Common settings
+    // ============================================================
+    sql "SET enable_nereids_planner=true"
+    sql "SET enable_fallback_to_original_planner=false"
+    sql "SET runtime_filter_mode=off"
+    sql "SET enable_profile=true"
+    sql "SET enable_sql_cache=false"
+    sql "SET enable_local_shuffle=true"
+
+    // ============================================================
+    //  Bug 1: must set shared state, in AGGREGATION_OPERATOR
+    //  RQG case: eliminate_group_by_uniform.case_id_11007680713
+    //  Key conditions: use_serial_exchange=true, parallel_pipeline_task_num=3
+    //  SQL: EXCEPT with count(*) GROUP BY on both sides
+    // ============================================================
+
+    // Test with FE planner (the buggy path)
+    logger.info("=== Bug 1a: AGG shared state - EXCEPT with serial exchange 
(FE planner) ===")
+    try {
+        sql """
+            SELECT 
/*+SET_VAR(use_serial_exchange=true,parallel_pipeline_task_num=3,
+                              enable_local_shuffle_planner=true,
+                              ignore_storage_data_distribution=true,
+                              enable_common_expr_pushdown=false,
+                              disable_streaming_preaggregations=true)*/
+                col_int_undef_signed_not_null as col1,
+                col_int_undef_signed_not_null as col2,
+                0 as col3, count(1)
+            FROM rqg_t1
+            GROUP BY col1, col2, col3
+            EXCEPT
+            SELECT col_bigint_undef_signed_not_null as col1,
+                   col_decimal_38_10__undef_signed_not_null as col2,
+                   5 as col3, count(1)
+            FROM rqg_t2
+            GROUP BY col1, col2, col3
+        """
+        logger.info("Bug 1a: PASSED (no crash)")
+    } catch (Throwable t) {
+        logger.error("Bug 1a FAILED: ${t.message}")
+        assertTrue(false, "Bug 1a: must set shared state in 
AGGREGATION_OPERATOR: ${t.message}")
+    }
+
+    // Compare with BE native planner
+    logger.info("=== Bug 1b: AGG shared state - EXCEPT with serial exchange 
(BE native) ===")
+    try {
+        sql """
+            SELECT 
/*+SET_VAR(use_serial_exchange=true,parallel_pipeline_task_num=3,
+                              enable_local_shuffle_planner=false,
+                              ignore_storage_data_distribution=true,
+                              enable_common_expr_pushdown=false,
+                              disable_streaming_preaggregations=true)*/
+                col_int_undef_signed_not_null as col1,
+                col_int_undef_signed_not_null as col2,
+                0 as col3, count(1)
+            FROM rqg_t1
+            GROUP BY col1, col2, col3
+            EXCEPT
+            SELECT col_bigint_undef_signed_not_null as col1,
+                   col_decimal_38_10__undef_signed_not_null as col2,
+                   5 as col3, count(1)
+            FROM rqg_t2
+            GROUP BY col1, col2, col3
+        """
+        logger.info("Bug 1b: PASSED (no crash)")
+    } catch (Throwable t) {
+        logger.error("Bug 1b FAILED: ${t.message}")
+        assertTrue(false, "Bug 1b: BE native also fails: ${t.message}")
+    }
+
+    // ============================================================
+    //  Bug 2: must set shared state, in SORT_OPERATOR
+    //  RQG case: grouping_set.case_id_5308471751
+    //  Key conditions: use_serial_exchange=true, parallel_pipeline_task_num=5
+    //  SQL: GROUPING SETS + window function (PERCENT_RANK)
+    // ============================================================
+
+    logger.info("=== Bug 2a: SORT shared state - GROUPING SETS + window (FE 
planner) ===")
+    try {
+        sql """
+            SELECT 
/*+SET_VAR(use_serial_exchange=true,parallel_pipeline_task_num=5,
+                              enable_local_shuffle_planner=true,
+                              ignore_storage_data_distribution=true,
+                              enable_share_hash_table_for_broadcast_join=false,
+                              disable_streaming_preaggregations=true)*/
+                SUM(PERCENT_RANK() OVER (PARTITION BY col_int_undef_signed2 
ORDER BY col_int_undef_signed2))
+            FROM rqg_t1
+            GROUP BY GROUPING SETS ((col_int_undef_signed2),(pk, 
pk),(col_int_undef_signed))
+        """
+        logger.info("Bug 2a: PASSED (no crash)")
+    } catch (Throwable t) {
+        logger.error("Bug 2a FAILED: ${t.message}")
+        assertTrue(false, "Bug 2a: must set shared state in SORT_OPERATOR: 
${t.message}")
+    }
+
+    logger.info("=== Bug 2b: SORT shared state - GROUPING SETS + window (BE 
native) ===")
+    try {
+        sql """
+            SELECT 
/*+SET_VAR(use_serial_exchange=true,parallel_pipeline_task_num=5,
+                              enable_local_shuffle_planner=false,
+                              ignore_storage_data_distribution=true,
+                              enable_share_hash_table_for_broadcast_join=false,
+                              disable_streaming_preaggregations=true)*/
+                SUM(PERCENT_RANK() OVER (PARTITION BY col_int_undef_signed2 
ORDER BY col_int_undef_signed2))
+            FROM rqg_t1
+            GROUP BY GROUPING SETS ((col_int_undef_signed2),(pk, 
pk),(col_int_undef_signed))
+        """
+        logger.info("Bug 2b: PASSED (no crash)")
+    } catch (Throwable t) {
+        logger.error("Bug 2b FAILED: ${t.message}")
+        assertTrue(false, "Bug 2b: BE native also fails: ${t.message}")
+    }
+
+    // ============================================================
+    //  Bug 3: incorrect results with GROUPING SETS + scalar subquery + window
+    //  RQG case: grouping_set.case_id_5694495756
+    //  Key conditions: parallel_pipeline_task_num=2, 
disable_streaming_preaggregations=true
+    //  Expected: all rows same value; Actual: values split proportionally 
(1/3, 2/3)
+    // ============================================================
+
+    logger.info("=== Bug 3: incorrect results - GROUPING SETS + subquery + 
window ===")
+    // FE planner
+    def result_fe = sql """
+        SELECT /*+SET_VAR(parallel_pipeline_task_num=2,
+                          enable_local_shuffle_planner=true,
+                          disable_streaming_preaggregations=true,
+                          enable_share_hash_table_for_broadcast_join=true)*/
+            SUM((SELECT MAX(col_int_undef_signed2) FROM rqg_t1))
+                OVER (PARTITION BY pk ORDER BY pk)
+        FROM rqg_t1
+        GROUP BY GROUPING SETS ((col_int_undef_signed2, pk),(pk), (pk))
+    """
+    // BE native
+    def result_be = sql """
+        SELECT /*+SET_VAR(parallel_pipeline_task_num=2,
+                          enable_local_shuffle_planner=false,
+                          disable_streaming_preaggregations=true,
+                          enable_share_hash_table_for_broadcast_join=true)*/
+            SUM((SELECT MAX(col_int_undef_signed2) FROM rqg_t1))
+                OVER (PARTITION BY pk ORDER BY pk)
+        FROM rqg_t1
+        GROUP BY GROUPING SETS ((col_int_undef_signed2, pk),(pk), (pk))
+    """
+    logger.info("Bug 3 FE result rows: ${result_fe.size()}, first few: 
${result_fe.take(5)}")
+    logger.info("Bug 3 BE result rows: ${result_be.size()}, first few: 
${result_be.take(5)}")
+
+    // All values in both should be the same
+    if (result_fe.size() != result_be.size()) {
+        logger.warn("Bug 3: row count mismatch FE=${result_fe.size()} vs 
BE=${result_be.size()}")

Review Comment:
   [P2] Make this regression fail the suite
   
   When FE and BE produce different row counts this only logs a warning, so the 
regression still passes. The same issue appears in the Bug 4/5 catches below, 
which log query failures without rethrowing or asserting. These cases therefore 
provide no protection against the crashes and result mismatches they describe. 
Please use a result assertion/`qt_*`, and let unexpected SQL exceptions 
propagate, so a recurrence fails the suite.



-- 
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.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to