Fix trim stats propagation and address review findings

1. Propagate _numResizes/_resizeTimeNs in mergePartitionTable() so
   isTrimmed() reflects trimming from absorbed tables. Also add
   absorbResizeStats() for the merge() path used in publishLocalPartitions().
   Without this, unsafe-trimmed results could be reported as exact.

2. Add regression test testMergePartitionTablePropagatesResizeStats()
   that verifies groupsTrimmed, numResizes, and resizeTimeMs survive
   absorbed-table merges via both mergePartitionTable() and
   merge()+absorbResizeStats().

3. Fix Javadoc on mergeResults(): "selection result blocks" -> "group-by
   result blocks" with accurate description of partition tree-reduction.

4. Fix timeout/error strings: "group-by order-by" -> "group-by", and
   clamp timeout to non-negative in error message.

5. Add Javadoc to mergePartitionTable() documenting disjoint-keys
   precondition.

6. Fix integration test: remove stale groupByAlgorithm query option,
   use only numGroupByPartitions to control partitioned combine.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: xiangfu0

pinot-core/src/main/java/org/apache/pinot/core/data/table/IndexedTable.java

@@ -287,10 +287,20 @@ public Iterator<Record> iterator() {
   }
 
 
+  /**
+   * Merges records from a disjoint partition table into this table using {@code putAll}.
+   * <p><b>Precondition:</b> the source table must have no overlapping keys with this table
+   * (i.e. they come from disjoint hash partitions). If keys overlap, {@code putAll} will
+   * silently overwrite existing entries instead of aggregating them. Use {@link #merge(Table)}
+   * for tables that may share keys.
+   */
   public void mergePartitionTable(Table table) {
     _topRecords = null;
     if (table instanceof IndexedTable) {
-      _lookupMap.putAll(((IndexedTable) table)._lookupMap);
+      IndexedTable other = (IndexedTable) table;
+      _lookupMap.putAll(other._lookupMap);
+      _numResizes += other._numResizes;
+      _resizeTimeNs += other._resizeTimeNs;
     } else {
       Iterator<Record> iterator = table.iterator();
       while (iterator.hasNext()) {
@@ -304,6 +314,15 @@ public void mergePartitionTable(Table table) {
     }
   }
 
+  /**
+   * Absorbs trim/resize statistics from another table that was merged into this one.
+   * Call after {@link #merge(Table)} to preserve the absorbed table's trim history.
+   */
+  public void absorbResizeStats(IndexedTable other) {
+    _numResizes += other._numResizes;
+    _resizeTimeNs += other._resizeTimeNs;
+  }
+
   public int getNumResizes() {
     return _numResizes;
   }

pinot-core/src/main/java/org/apache/pinot/core/operator/combine/GroupByCombineOperator.java

@@ -219,14 +219,14 @@ public BaseResultsBlock mergeResults()
   }
 
   protected ExceptionResultsBlock getTimeoutResultsBlock(long timeoutMs) {
-    String userError = "Timed out while combining group-by order-by results after " + timeoutMs + "ms";
+    String userError = "Timed out while combining group-by results after " + Math.max(0, timeoutMs) + "ms";
     String logMsg = userError + ", queryContext = " + _queryContext;
     LOGGER.error(logMsg);
     return new ExceptionResultsBlock(new QueryErrorMessage(QueryErrorCode.EXECUTION_TIMEOUT, userError, logMsg));
   }
 
   protected ExceptionResultsBlock getExceptionResultsBlock(Throwable ex) {
-    String userError = "Caught exception while processing group-by order-by query";
+    String userError = "Caught exception while processing group-by query";
     String devError = userError + ": " + ex.getMessage();
     QueryErrorMessage errMsg;
     if (ex instanceof QueryException) {

pinot-core/src/main/java/org/apache/pinot/core/operator/combine/PartitionedGroupByCombineOperator.java

@@ -110,14 +110,10 @@ protected void processSegments() {
   /**
    * {@inheritDoc}
    *
-   * <p>Combines intermediate selection result blocks from underlying operators and returns a merged one.
+   * <p>Combines intermediate group-by result blocks from underlying operators and returns a merged one.
    * <ul>
-   *   <li>
-   *     Merges multiple intermediate selection result blocks as a merged one.
-   *   </li>
-   *   <li>
-   *     Set all exceptions encountered during execution into the merged result block
-   *   </li>
+   *   <li>Merges partition tables into a single table via parallel tree-reduction.</li>
+   *   <li>Sets all exceptions encountered during execution into the merged result block.</li>
    * </ul>
    */
   @Override
@@ -215,9 +211,11 @@ private void publishLocalPartitions(IndexedTable[] localIndexedTables) {
           _indexedTables[partitionId] = localIndexedTable;
         } else if (localIndexedTable.size() > indexedTable.size()) {
           localIndexedTable.merge(indexedTable);
+          localIndexedTable.absorbResizeStats(indexedTable);
           _indexedTables[partitionId] = localIndexedTable;
         } else {
           indexedTable.merge(localIndexedTable);
+          indexedTable.absorbResizeStats(localIndexedTable);
         }
       }
     }

pinot-core/src/test/java/org/apache/pinot/core/data/table/IndexedTableTest.java

@@ -337,4 +337,56 @@ public void testMergePartitionTableInvalidatesFinishedTopRecords() {
       executorService.shutdownNow();
     }
   }
+
+  @Test
+  public void testMergePartitionTablePropagatesResizeStats() {
+    QueryContext queryContext = QueryContextConverterUtils.getQueryContext(
+        "SELECT SUM(m1) FROM testTable GROUP BY d1 ORDER BY SUM(m1) DESC");
+    DataSchema dataSchema = new DataSchema(new String[]{"d1", "sum(m1)"},
+        new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.DOUBLE});
+
+    ExecutorService executorService = Executors.newCachedThreadPool();
+    try {
+      // Create a table that will be trimmed (trimThreshold=5, trimSize=3)
+      IndexedTable trimmedTable =
+          new SimpleIndexedTable(dataSchema, false, queryContext, 3, 3, 5, INITIAL_CAPACITY, executorService);
+      for (int i = 0; i < 6; i++) {
+        trimmedTable.upsert(new Key(new Object[]{"k" + i}), getRecord(new Object[]{"k" + i, (double) i}));
+      }
+      Assert.assertTrue(trimmedTable.isTrimmed(), "trimmedTable should be trimmed after exceeding threshold");
+      int trimmedResizes = trimmedTable.getNumResizes();
+      Assert.assertTrue(trimmedResizes > 0);
+
+      // Create a non-trimmed table
+      IndexedTable survivingTable =
+          new SimpleIndexedTable(dataSchema, false, queryContext, 100, 100, Integer.MAX_VALUE, INITIAL_CAPACITY,
+              executorService);
+      survivingTable.upsert(new Key(new Object[]{"x"}), getRecord(new Object[]{"x", 99d}));
+      Assert.assertFalse(survivingTable.isTrimmed());
+      Assert.assertEquals(survivingTable.getNumResizes(), 0);
+
+      // mergePartitionTable should propagate resize stats
+      survivingTable.mergePartitionTable(trimmedTable);
+      Assert.assertEquals(survivingTable.getNumResizes(), trimmedResizes);
+      Assert.assertTrue(survivingTable.isTrimmed(),
+          "Surviving table should report trimmed after absorbing a trimmed table");
+
+      // absorbResizeStats should also work for merge() path
+      IndexedTable tableA =
+          new SimpleIndexedTable(dataSchema, false, queryContext, 100, 100, Integer.MAX_VALUE, INITIAL_CAPACITY,
+              executorService);
+      IndexedTable tableB =
+          new SimpleIndexedTable(dataSchema, false, queryContext, 3, 3, 5, INITIAL_CAPACITY, executorService);
+      for (int i = 0; i < 6; i++) {
+        tableB.upsert(new Key(new Object[]{"b" + i}), getRecord(new Object[]{"b" + i, (double) i}));
+      }
+      Assert.assertTrue(tableB.isTrimmed());
+      tableA.merge(tableB);
+      tableA.absorbResizeStats(tableB);
+      Assert.assertTrue(tableA.isTrimmed(),
+          "tableA should report trimmed after merge + absorbResizeStats from trimmed tableB");
+    } finally {
+      executorService.shutdownNow();
+    }
+  }
 }

pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentPartitionedGroupBySingleValueQueriesTest.java

@@ -30,8 +30,7 @@
 public class InterSegmentPartitionedGroupBySingleValueQueriesTest
     extends InterSegmentGroupBySingleValueQueriesTest {
   private static final InstancePlanMakerImplV2 TRIM_ENABLED_PLAN_MAKER = new InstancePlanMakerImplV2();
-  private static final Map<String, String> QUERY_OPTIONS =
-      Map.of("groupByAlgorithm", "partitioned", "numGroupByPartitions", "8");
+  private static final Map<String, String> QUERY_OPTIONS = Map.of("numGroupByPartitions", "8");
 
   static {
     TRIM_ENABLED_PLAN_MAKER.setMinSegmentGroupTrimSize(1);