Improve the Explain Plan accuracy

pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java

@@ -26,7 +26,6 @@
 import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -473,6 +472,7 @@ private BrokerResponseNative handleRequest(long requestId, String query, JsonNod
     Map<ServerInstance, List<String>> offlineRoutingTable = null;
     Map<ServerInstance, List<String>> realtimeRoutingTable = null;
     List<String> unavailableSegments = new ArrayList<>();
+    int numPrunedSegmentsTotal = 0;
     if (offlineBrokerRequest != null) {
       // NOTE: Routing table might be null if table is just removed
       RoutingTable routingTable = _routingManager.getRoutingTable(offlineBrokerRequest);
@@ -484,6 +484,7 @@ private BrokerResponseNative handleRequest(long requestId, String query, JsonNod
         } else {
           offlineBrokerRequest = null;
         }
+        numPrunedSegmentsTotal += routingTable.getNumPrunedSegments();
       } else {
         offlineBrokerRequest = null;
       }
@@ -499,6 +500,7 @@ private BrokerResponseNative handleRequest(long requestId, String query, JsonNod
         } else {
           realtimeBrokerRequest = null;
         }
+        numPrunedSegmentsTotal += routingTable.getNumPrunedSegments();
       } else {
         realtimeBrokerRequest = null;
       }
@@ -548,22 +550,26 @@ private BrokerResponseNative handleRequest(long requestId, String query, JsonNod
 
     // Execute the query
     ServerStats serverStats = new ServerStats();
-    if (serverPinotQuery.isExplain()) {
-      // Update routing tables to only send request to 1 server (& generate the plan for 1 segment).
+    // TODO: Handle broker specific operations for explain plan queries such as:
+    //       - Alias handling
+    //       - Compile time function invocation
+    //       - Literal only queries
+    //       - Any rewrites
+    if (pinotQuery.isExplain()) {
+      // Update routing tables to only send request to offline servers for OFFLINE and HYBRID tables.
+      // TODO: Assess if the Explain Plan Query should also be routed to REALTIME servers for HYBRID tables
       if (offlineRoutingTable != null) {
-        setRoutingToOneSegment(offlineRoutingTable);
         // For OFFLINE and HYBRID tables, don't send EXPLAIN query to realtime servers.
         realtimeBrokerRequest = null;
         realtimeRoutingTable = null;
-      } else {
-        setRoutingToOneSegment(realtimeRoutingTable);
       }
     }
     // TODO: Modify processBrokerRequest() to directly take PinotQuery
     BrokerResponseNative brokerResponse =
         processBrokerRequest(requestId, brokerRequest, serverBrokerRequest, offlineBrokerRequest, offlineRoutingTable,
             realtimeBrokerRequest, realtimeRoutingTable, remainingTimeMs, serverStats, requestContext);
     brokerResponse.setExceptions(exceptions);
+    brokerResponse.setNumSegmentsPrunedByBroker(numPrunedSegmentsTotal);
     long executionEndTimeNs = System.nanoTime();
     _brokerMetrics.addPhaseTiming(rawTableName, BrokerQueryPhase.QUERY_EXECUTION,
         executionEndTimeNs - routingEndTimeNs);
@@ -640,15 +646,6 @@ private void setTimestampIndexExpressionOverrideHints(@Nullable Expression expre
         .forEach(operand -> setTimestampIndexExpressionOverrideHints(operand, timestampIndexColumns, pinotQuery));
   }
 
-  /** Set EXPLAIN PLAN query to route to only one segment on one server. */
-  private void setRoutingToOneSegment(Map<ServerInstance, List<String>> routingTable) {
-    Set<Map.Entry<ServerInstance, List<String>>> servers = routingTable.entrySet();
-    // only send request to 1 server
-    Map.Entry<ServerInstance, List<String>> server = servers.iterator().next();
-    routingTable.clear();
-    routingTable.put(server.getKey(), Collections.singletonList(server.getValue().get(0)));
-  }
-
   /** Given a {@link PinotQuery}, check if the WHERE clause will always evaluate to false. */
   private boolean isFilterAlwaysFalse(PinotQuery pinotQuery) {
     return FALSE.equals(pinotQuery.getFilterExpression());

pinot-broker/src/main/java/org/apache/pinot/broker/routing/BrokerRoutingManager.java

@@ -573,7 +573,8 @@ public RoutingTable getRoutingTable(BrokerRequest brokerRequest) {
         _brokerMetrics.addMeteredTableValue(tableNameWithType, BrokerMeter.SERVER_MISSING_FOR_ROUTING, 1L);
       }
     }
-    return new RoutingTable(serverInstanceToSegmentsMap, selectionResult.getUnavailableSegments());
+    return new RoutingTable(serverInstanceToSegmentsMap, selectionResult.getUnavailableSegments(),
+        selectionResult.getNumPrunedSegments());
   }
 
   @Override
@@ -705,15 +706,20 @@ void refreshSegment(String segment) {
 
     InstanceSelector.SelectionResult calculateRouting(BrokerRequest brokerRequest) {
       Set<String> selectedSegments = _segmentSelector.select(brokerRequest);
+      int numTotalSelectedSegments = selectedSegments.size();
       if (!selectedSegments.isEmpty()) {
         for (SegmentPruner segmentPruner : _segmentPruners) {
           selectedSegments = segmentPruner.prune(brokerRequest, selectedSegments);
         }
       }
+      int numPrunedSegments = numTotalSelectedSegments - selectedSegments.size();
       if (!selectedSegments.isEmpty()) {
-        return _instanceSelector.select(brokerRequest, new ArrayList<>(selectedSegments));
+        InstanceSelector.SelectionResult selectionResult = _instanceSelector.select(brokerRequest,
+            new ArrayList<>(selectedSegments));
+        selectionResult.setNumPrunedSegments(numPrunedSegments);
+        return selectionResult;
       } else {
-        return new InstanceSelector.SelectionResult(Collections.emptyMap(), Collections.emptyList());
+        return new InstanceSelector.SelectionResult(Collections.emptyMap(), Collections.emptyList(), numPrunedSegments);
       }
     }
   }

pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelector.java

@@ -61,10 +61,17 @@ public interface InstanceSelector {
   class SelectionResult {
     private final Map<String, String> _segmentToInstanceMap;
     private final List<String> _unavailableSegments;
+    private int _numPrunedSegments;
 
     public SelectionResult(Map<String, String> segmentToInstanceMap, List<String> unavailableSegments) {
+      this(segmentToInstanceMap, unavailableSegments, 0);
+    }
+
+    public SelectionResult(Map<String, String> segmentToInstanceMap, List<String> unavailableSegments,
+        int numPrunedSegments) {
       _segmentToInstanceMap = segmentToInstanceMap;
       _unavailableSegments = unavailableSegments;
+      _numPrunedSegments = numPrunedSegments;
     }
 
     /**
@@ -80,5 +87,19 @@ public Map<String, String> getSegmentToInstanceMap() {
     public List<String> getUnavailableSegments() {
       return _unavailableSegments;
     }
+
+    /**
+     * Returns the number of segments pruned by the broker
+     */
+    public int getNumPrunedSegments() {
+      return _numPrunedSegments;
+    }
+
+    /**
+     * Sets the number of segments pruned by the broker
+     */
+    public void setNumPrunedSegments(int numPrunedSegments) {
+      _numPrunedSegments = numPrunedSegments;
+    }
   }
 }

pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java

@@ -220,4 +220,44 @@ String toJsonString()
    * against realtime table in request handling, into the broker response.
    */
   void setRealtimeTotalCpuTimeNs(long realtimeTotalCpuTimeNs);
+
+  /**
+   * Get the total number of segments pruned on the Broker side
+   */
+  long getNumSegmentsPrunedByBroker();
+
+  /**
+   * Set the total number of segments pruned on the Broker side
+   */
+  void setNumSegmentsPrunedByBroker(long numSegmentsPrunedByBroker);
+
+  /**
+   * Get the total number of segments pruned on the Server side
+   */
+  long getNumSegmentsPrunedByServer();
+
+  /**
+   * Set the total number of segments pruned on the Server side
+   */
+  void setNumSegmentsPrunedByServer(long numSegmentsPrunedByServer);
+
+  /**
+   * Get the total number of segments with an EmptyFilterOperator when Explain Plan is called
+   */
+  long getExplainPlanNumEmptyFilterSegments();
+
+  /**
+   * Set the total number of segments with an EmptyFilterOperator when Explain Plan is called
+   */
+  void setExplainPlanNumEmptyFilterSegments(long explainPlanNumEmptyFilterSegments);
+
+  /**
+   * Get the total number of segments with a MatchAllFilterOperator when Explain Plan is called
+   */
+  long getExplainPlanNumMatchAllFilterSegments();
+
+  /**
+   * Set the total number of segments with a MatchAllFilterOperator when Explain Plan is called
+   */
+  void setExplainPlanNumMatchAllFilterSegments(long explainPlanNumMatchAllFilterSegments);
 }

pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java

@@ -81,6 +81,10 @@ public class BrokerResponseNative implements BrokerResponse {
   private long _realtimeResponseSerializationCpuTimeNs = 0L;
   private long _offlineTotalCpuTimeNs = 0L;
   private long _realtimeTotalCpuTimeNs = 0L;
+  private long _numSegmentsPrunedByBroker = 0L;
+  private long _numSegmentsPrunedByServer = 0L;
+  private long _explainPlanNumEmptyFilterSegments = 0L;
+  private long _explainPlanNumMatchAllFilterSegments = 0L;
   private int _numRowsResultSet = 0;
   private ResultTable _resultTable;
   private Map<String, String> _traceInfo = new HashMap<>();
@@ -217,6 +221,54 @@ public void setRealtimeTotalCpuTimeNs(long realtimeTotalCpuTimeNs) {
     _realtimeTotalCpuTimeNs = realtimeTotalCpuTimeNs;
   }
 
+  @JsonProperty("numSegmentsPrunedByBroker")
+  @Override
+  public long getNumSegmentsPrunedByBroker() {
+    return _numSegmentsPrunedByBroker;
+  }
+
+  @JsonProperty("numSegmentsPrunedByBroker")
+  @Override
+  public void setNumSegmentsPrunedByBroker(long numSegmentsPrunedByBroker) {
+    _numSegmentsPrunedByBroker = numSegmentsPrunedByBroker;
+  }
+
+  @JsonProperty("numSegmentsPrunedByServer")
+  @Override
+  public long getNumSegmentsPrunedByServer() {
+    return _numSegmentsPrunedByServer;
+  }
+
+  @JsonProperty("numSegmentsPrunedByServer")
+  @Override
+  public void setNumSegmentsPrunedByServer(long numSegmentsPrunedByServer) {
+    _numSegmentsPrunedByServer = numSegmentsPrunedByServer;
+  }
+
+  @JsonProperty("explainPlanNumEmptyFilterSegments")
+  @Override
+  public long getExplainPlanNumEmptyFilterSegments() {
+    return _explainPlanNumEmptyFilterSegments;
+  }
+
+  @JsonProperty("explainPlanNumEmptyFilterSegments")
+  @Override
+  public void setExplainPlanNumEmptyFilterSegments(long explainPlanNumEmptyFilterSegments) {
+    _explainPlanNumEmptyFilterSegments = explainPlanNumEmptyFilterSegments;
+  }
+
+  @JsonProperty("explainPlanNumMatchAllFilterSegments")
+  @Override
+  public long getExplainPlanNumMatchAllFilterSegments() {
+    return _explainPlanNumMatchAllFilterSegments;
+  }
+
+  @JsonProperty("explainPlanNumMatchAllFilterSegments")
+  @Override
+  public void setExplainPlanNumMatchAllFilterSegments(long explainPlanNumMatchAllFilterSegments) {
+    _explainPlanNumMatchAllFilterSegments = explainPlanNumMatchAllFilterSegments;
+  }
+
   @JsonProperty("resultTable")
   @JsonInclude(JsonInclude.Include.NON_NULL)
   public ResultTable getResultTable() {

pinot-common/src/main/java/org/apache/pinot/common/utils/DataTable.java

@@ -111,7 +111,10 @@ enum MetadataKey {
     RESIZE_TIME_MS("resizeTimeMs", MetadataValueType.LONG),
     THREAD_CPU_TIME_NS("threadCpuTimeNs", MetadataValueType.LONG),
     SYSTEM_ACTIVITIES_CPU_TIME_NS("systemActivitiesCpuTimeNs", MetadataValueType.LONG),
-    RESPONSE_SER_CPU_TIME_NS("responseSerializationCpuTimeNs", MetadataValueType.LONG);
+    RESPONSE_SER_CPU_TIME_NS("responseSerializationCpuTimeNs", MetadataValueType.LONG),
+    NUM_SEGMENTS_PRUNED_BY_SERVER("numSegmentsPrunedByServer", MetadataValueType.INT),
+    EXPLAIN_PLAN_NUM_EMPTY_FILTER_SEGMENTS("explainPlanNumEmptyFilterSegments", MetadataValueType.INT),
+    EXPLAIN_PLAN_NUM_MATCH_ALL_FILTER_SEGMENTS("explainPlanNumMatchAllFilterSegments", MetadataValueType.INT);
 
     private static final MetadataKey[] VALUES;
     private static final Map<String, MetadataKey> NAME_TO_ENUM_KEY_MAP = new HashMap<>();

pinot-core/src/main/java/org/apache/pinot/core/common/ExplainPlanRowData.java

@@ -0,0 +1,66 @@
+/**
+ * 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.pinot.core.common;
+
+import java.util.Objects;
+
+/**
+ * Class to hold the data for a single Explain plan row
+ */
+public class ExplainPlanRowData {
+  private final String _explainPlanString;
+  private final int _operatorId;
+  private final int _parentId;
+
+  public ExplainPlanRowData(String explainPlanString, int operatorId, int parentId) {
+    _explainPlanString = explainPlanString;
+    _operatorId = operatorId;
+    _parentId = parentId;
+  }
+
+  public String getExplainPlanString() {
+    return _explainPlanString;
+  }
+
+  public int getOperatorId() {
+    return _operatorId;
+  }
+
+  public int getParentId() {
+    return _parentId;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    ExplainPlanRowData that = (ExplainPlanRowData) o;
+    return _operatorId == that._operatorId && _parentId == that._parentId
+        && _explainPlanString.equals(that._explainPlanString);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(_explainPlanString, _operatorId, _parentId);
+  }
+}