This is an automated email from the ASF dual-hosted git repository.
yashmayya pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 08c64f3d722 Fix multi-stage physical optimizer failing to query empty
tables (#18925)
08c64f3d722 is described below
commit 08c64f3d722a5ab877098f4d09ae5ee918402d6e
Author: Yash Mayya <[email protected]>
AuthorDate: Wed Jul 8 14:11:56 2026 -0700
Fix multi-stage physical optimizer failing to query empty tables (#18925)
---
.../core/routing/MockRoutingManagerFactory.java | 9 +
.../tests/MultiStageEngineIntegrationTest.java | 53 +++++-
.../planner/physical/PinotDispatchPlanner.java | 47 +++++-
.../v2/PlanFragmentAndMailboxAssignment.java | 9 +
.../opt/rules/LeafStageWorkerAssignmentRule.java | 11 +-
.../v2/opt/rules/WorkerExchangeAssignmentRule.java | 6 +-
.../planner/physical/v2/EmptyTablePlanTest.java | 186 +++++++++++++++++++++
.../rules/LeafStageWorkerAssignmentRuleTest.java | 12 ++
8 files changed, 328 insertions(+), 5 deletions(-)
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/routing/MockRoutingManagerFactory.java
b/pinot-core/src/test/java/org/apache/pinot/core/routing/MockRoutingManagerFactory.java
index 6a1e44f30ae..8127629ca45 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/routing/MockRoutingManagerFactory.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/routing/MockRoutingManagerFactory.java
@@ -93,6 +93,15 @@ public class MockRoutingManagerFactory {
.add(serverInstance);
}
+ /**
+ * Registers a routing entry for a table that has no routable segments (e.g.
an empty table, or one whose segments
+ * were all pruned). This mirrors {@code BaseBrokerRoutingManager}, which
returns a non-null {@link RoutingTable}
+ * with an empty server-to-segments map for such a table (as opposed to
{@code null} when no routing entry exists).
+ */
+ public void registerEmptyTable(String tableNameWithType) {
+ _tableSegmentServersMap.computeIfAbsent(tableNameWithType, k -> new
HashMap<>());
+ }
+
public void disableTable(String tableNameWithType) {
_disabledTables.add(tableNameWithType);
}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
index 479c537d25d..a1d32da14dd 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
@@ -272,6 +272,51 @@ public class MultiStageEngineIntegrationTest extends
BaseClusterIntegrationTestS
List.of(), "LONG");
}
+ @Test
+ public void testAllLeafStagesEmptyBrokerResponsesWithPhysicalOptimizer()
+ throws Exception {
+ // Same short-circuit behavior must hold under the multi-stage physical
optimizer, which previously failed with
+ // "No routing entry for offline or realtime type" when a leaf stage had
no routable segments.
+ String table = "mytable";
+ String prefix = "SET useBrokerPruning = 'true'; SET usePhysicalOptimizer =
'true'; ";
+ assertAllLeafStagesEmptyRows(prefix, "SELECT AirlineID, Carrier FROM " +
table + " WHERE DaysSinceEpoch < 0",
+ List.of(), "LONG", "STRING");
+ assertAllLeafStagesEmptyRows(prefix, "SELECT COUNT(*) FROM " + table + "
WHERE DaysSinceEpoch < 0",
+ List.of(List.<Object>of(0)), "LONG");
+ assertAllLeafStagesEmptyRows(prefix, "SELECT SUM(ActualElapsedTime) FROM "
+ table + " WHERE DaysSinceEpoch < 0",
+ List.of(Arrays.asList((Object) null)), "LONG");
+ assertAllLeafStagesEmptyRows(prefix, "SELECT COUNT(*) + 1 FROM " + table +
" WHERE DaysSinceEpoch < 0",
+ List.of(List.<Object>of(1)), "LONG");
+ assertAllLeafStagesEmptyRows(prefix,
+ "SELECT AirlineID, COUNT(*) FROM " + table + " WHERE DaysSinceEpoch <
0 GROUP BY AirlineID",
+ List.of(), "LONG", "LONG");
+ }
+
+ @Test
+ public void testPartiallyEmptyWithPhysicalOptimizerFailsFast()
+ throws Exception {
+ // Combining an empty/fully-pruned table with a non-empty table is not yet
supported by the physical optimizer.
+ // It must fail fast with a clear, actionable error rather than silently
drop rows.
+ String table = "mytable";
+ String expectedError = "combine an empty or fully-pruned table with a
non-empty table";
+ // The empty branch (all segments pruned) contributes zero rows, so the
union counts only the non-empty branch.
+ String unionQuery = "SELECT COUNT(*) FROM (SELECT AirlineID FROM " + table
+ " WHERE DaysSinceEpoch < 0 "
+ + "UNION ALL SELECT AirlineID FROM " + table + ") u";
+
+ JsonNode overUnion = postQuery("SET useBrokerPruning = 'true'; SET
usePhysicalOptimizer = 'true'; " + unionQuery);
+ assertFalse(overUnion.get("exceptions").isEmpty(), "Expected a
partially-empty error: " + overUnion);
+ assertTrue(overUnion.get("exceptions").toString().contains(expectedError),
+ "Expected a clear partially-empty error, got: " +
overUnion.get("exceptions"));
+
+ // The suggested fallback (legacy engine) must answer the same query
correctly: the empty branch adds nothing, so
+ // the union count equals the total row count of the non-empty table.
+ long total = postQuery("SELECT COUNT(*) FROM " +
table).get("resultTable").get("rows").get(0).get(0).asLong();
+ assertTrue(total > 0, "Sanity: mytable should be non-empty");
+ JsonNode fallback = postQuery("SET useBrokerPruning = 'true'; SET
usePhysicalOptimizer = 'false'; " + unionQuery);
+ assertTrue(fallback.get("exceptions").isEmpty(), "Fallback should not
error: " + fallback);
+
assertEquals(fallback.get("resultTable").get("rows").get(0).get(0).asLong(),
total);
+ }
+
@Test
public void testReplicatedLeavesThatCanProduceRowsDoNotShortCircuit()
throws Exception {
@@ -296,7 +341,13 @@ public class MultiStageEngineIntegrationTest extends
BaseClusterIntegrationTestS
private JsonNode assertAllLeafStagesEmptyRows(String query,
List<List<Object>> expectedRows, String... expectedTypes)
throws Exception {
- JsonNode response = postQuery("SET useBrokerPruning = 'true'; " + query);
+ return assertAllLeafStagesEmptyRows("SET useBrokerPruning = 'true'; ",
query, expectedRows, expectedTypes);
+ }
+
+ private JsonNode assertAllLeafStagesEmptyRows(String setPrefix, String
query, List<List<Object>> expectedRows,
+ String... expectedTypes)
+ throws Exception {
+ JsonNode response = postQuery(setPrefix + query);
assertTrue(response.get("exceptions").isEmpty(), "Unexpected exceptions
for query: " + query);
assertEquals(response.get("numServersQueried").asInt(), 0, "Query should
not dispatch to servers: " + query);
assertEquals(response.get("numServersResponded").asInt(), 0, "Query should
not dispatch to servers: " + query);
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/PinotDispatchPlanner.java
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/PinotDispatchPlanner.java
index 5e21eea6988..b32c4164693 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/PinotDispatchPlanner.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/PinotDispatchPlanner.java
@@ -109,10 +109,53 @@ public class PinotDispatchPlanner {
for (var entry : result._planFragmentMap.entrySet()) {
context.getDispatchablePlanStageRootMap().put(entry.getKey(),
entry.getValue().getFragmentRoot());
}
+ trackEmptyLeafStages(context);
runValidations(rootFragment, context);
return finalizeDispatchableSubPlan(rootFragment, context);
}
+ /**
+ * Records empty leaf stages so that {@link
DispatchablePlanContext#isAllNonReplicatedLeafStagesEmpty()} works for the
+ * physical optimizer path, mirroring what {@code WorkerManager} does for
the legacy path. A leaf stage is a fragment
+ * that scans one or more tables; it is empty when it was assigned zero
workers (no routable segments, e.g. an empty
+ * table or all segments pruned by the broker). When every leaf stage is
empty, this drives the empty-leaf
+ * short-circuit in {@link #finalizeDispatchableSubPlan}.
+ * <p>
+ * When only <i>some</i> leaf stages are empty (an empty or fully-pruned
table combined with a non-empty table in the
+ * same query), the physical optimizer cannot yet produce a correct plan: an
empty leaf is assigned zero workers, and
+ * a downstream join/set-op/aggregate derives its worker set from its
inputs, so the empty branch can zero out a whole
+ * stage and silently drop rows. Rather than return wrong results, fail fast
with an actionable error suggesting the
+ * legacy (non-physical) engine, which handles this case.
+ * <p>
+ * Unlike the legacy path (which excludes broadcast-replicated dim leaves
from this tracking via
+ * {@code WorkerManager}'s early return), every table-scanning fragment is
counted here. The physical optimizer does
+ * not honor the {@code IS_REPLICATED} broadcast hint (a dim table is routed
like any other table), and it never
+ * populates {@code replicatedSegments}, so {@link
#hasNonEmptyReplicatedLeaf} cannot rescue an incorrectly
+ * short-circuited replicated join. Counting every leaf keeps the fail-fast
conservative: a query mixing an empty
+ * table with a non-empty (dim or fact) table is rejected rather than
risking a wrong result.
+ */
+ private static void trackEmptyLeafStages(DispatchablePlanContext context) {
+ int leafStages = 0;
+ int emptyLeafStages = 0;
+ for (DispatchablePlanMetadata metadata :
context.getDispatchablePlanMetadataMap().values()) {
+ if (metadata.getScannedTables().isEmpty()) {
+ // Not a leaf stage (no table scan).
+ continue;
+ }
+ leafStages++;
+ context.recordLeafStageAssigned();
+ if (metadata.getWorkerIdToServerInstanceMap().isEmpty()) {
+ emptyLeafStages++;
+ context.recordLeafStageEmpty();
+ }
+ }
+ if (emptyLeafStages > 0 && emptyLeafStages < leafStages) {
+ throw new UnsupportedOperationException("The multi-stage physical
optimizer does not yet support queries that "
+ + "combine an empty or fully-pruned table with a non-empty table
(e.g. a join or union where one side has "
+ + "no routable segments). Retry with the query option
'usePhysicalOptimizer=false'.");
+ }
+ }
+
/**
* Run validations on the plan. Since there is only one validator right now,
don't try to over-engineer it.
*/
@@ -128,8 +171,8 @@ public class PinotDispatchPlanner {
private static DispatchableSubPlan finalizeDispatchableSubPlan(PlanFragment
subPlanRoot,
DispatchablePlanContext dispatchablePlanContext) {
- // TODO: Physical Optimizer path does not track empty leaf stages. To
support short-circuit,
- // check if all leaf TableScanMetadata have empty workerIdToSegmentsMap.
+ // Empty leaf stages are tracked for both engines: the legacy path in
WorkerManager, and the physical optimizer
+ // path in #trackEmptyLeafStages.
boolean allLeafStagesEmpty =
dispatchablePlanContext.isAllNonReplicatedLeafStagesEmpty();
if (allLeafStagesEmpty &&
hasNonEmptyReplicatedLeaf(dispatchablePlanContext.getDispatchablePlanMetadataMap()))
{
allLeafStagesEmpty = false;
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/PlanFragmentAndMailboxAssignment.java
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/PlanFragmentAndMailboxAssignment.java
index 29ef661150e..8e622b36171 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/PlanFragmentAndMailboxAssignment.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/PlanFragmentAndMailboxAssignment.java
@@ -274,6 +274,15 @@ public class PlanFragmentAndMailboxAssignment {
private void computeMailboxInfos(int senderStageId, int receiverStageId,
Map<Integer, QueryServerInstance> senderWorkers, Map<Integer,
QueryServerInstance> receiverWorkers,
ExchangeStrategy exchangeDesc, Context context) {
+ if (senderWorkers.isEmpty() && receiverWorkers.isEmpty()) {
+ // Both stages have no workers, which happens for an all-empty subtree:
a leaf stage with no routable segments
+ // (e.g. an empty table or all segments pruned by the broker) whose
emptiness propagates up. There is no data to
+ // route and no receiver to inform, and these stages are discarded by
the empty-leaf short-circuit in
+ // PinotDispatchPlanner#finalizeDispatchableSubPlan. Skipping avoids the
single-instance-receiver check below for
+ // a degenerate SINGLETON gather. When only the sender is empty (e.g.
the empty side of a union or join), the
+ // exchange-specific logic below still wires the receiver with an empty
sender list so it knows to expect no rows.
+ return;
+ }
DispatchablePlanMetadata senderMetadata =
context._fragmentMetadataMap.get(senderStageId);
DispatchablePlanMetadata receiverMetadata =
context._fragmentMetadataMap.get(receiverStageId);
Map<Integer, Map<Integer, MailboxInfos>> senderMailboxMap =
senderMetadata.getWorkerIdToMailboxesMap();
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LeafStageWorkerAssignmentRule.java
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LeafStageWorkerAssignmentRule.java
index d0ae144ae5e..c087d251980 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LeafStageWorkerAssignmentRule.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LeafStageWorkerAssignmentRule.java
@@ -218,8 +218,17 @@ public class LeafStageWorkerAssignmentRule extends
PRelOptRule {
InstanceIdToSegments instanceIdToSegments, Map<String,
TablePartitionInfo> tpiMap,
boolean inferInvalidPartitionSegment, boolean
inferRealtimeSegmentPartition) {
Set<String> tableTypes = instanceIdToSegments.getActiveTableTypes();
+ if (tableTypes.isEmpty()) {
+ // The table has a routing entry but no routable segments for either
table type (e.g. an empty table, or all
+ // segments were pruned by the broker). Assign zero workers instead of
throwing, mirroring the legacy
+ // WorkerManager. The resulting empty leaf stage is detected and
short-circuited to an empty result on the
+ // broker (see PinotDispatchPlanner#finalizeDispatchableSubPlan).
+ List<String> noWorkers = List.of();
+ PinotDataDistribution emptyDistribution = new
PinotDataDistribution(RelDistribution.Type.RANDOM_DISTRIBUTED,
+ noWorkers, noWorkers.hashCode(), null, null);
+ return new TableScanWorkerAssignmentResult(emptyDistribution, new
HashMap<>());
+ }
Set<String> partitionedTableTypes =
tableTypes.stream().filter(tpiMap::containsKey).collect(Collectors.toSet());
- Preconditions.checkState(!tableTypes.isEmpty(), "No routing entry for
offline or realtime type");
if (tableTypes.equals(partitionedTableTypes)) {
// TODO(mse-physical): Support auto-partitioning inference for Hybrid
tables.
if (partitionedTableTypes.size() == 1) {
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/WorkerExchangeAssignmentRule.java
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/WorkerExchangeAssignmentRule.java
index 11fdc0ce94b..f03e15caf25 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/WorkerExchangeAssignmentRule.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/physical/v2/opt/rules/WorkerExchangeAssignmentRule.java
@@ -322,7 +322,11 @@ public class WorkerExchangeAssignmentRule implements
PRelNodeTransformer {
_physicalPlannerContext.getDefaultHashFunction());
}
if (distributionConstraint.getType() == RelDistribution.Type.SINGLETON) {
- List<String> newWorkers =
currentNodeDistribution.getWorkers().subList(0, 1);
+ // Pick a single worker to gather the data on. When the input has no
workers (e.g. an empty leaf stage whose
+ // table has no routable segments), keep the singleton empty; the empty
leaf is short-circuited to an empty
+ // result on the broker later (see
PinotDispatchPlanner#finalizeDispatchableSubPlan).
+ List<String> currentWorkers = currentNodeDistribution.getWorkers();
+ List<String> newWorkers = currentWorkers.isEmpty() ? currentWorkers :
currentWorkers.subList(0, 1);
PinotDataDistribution pinotDataDistribution = new
PinotDataDistribution(RelDistribution.Type.SINGLETON,
newWorkers, newWorkers.hashCode(), null, null);
return new PhysicalExchange(nodeId(), currentNode,
pinotDataDistribution, List.of(),
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/query/planner/physical/v2/EmptyTablePlanTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/query/planner/physical/v2/EmptyTablePlanTest.java
new file mode 100644
index 00000000000..fabcb5fedf3
--- /dev/null
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/query/planner/physical/v2/EmptyTablePlanTest.java
@@ -0,0 +1,186 @@
+/**
+ * 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.query.planner.physical.v2;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.common.config.provider.TableCache;
+import org.apache.pinot.core.routing.MockRoutingManagerFactory;
+import org.apache.pinot.core.routing.RoutingManager;
+import org.apache.pinot.query.QueryEnvironment;
+import org.apache.pinot.query.QueryEnvironmentTestBase;
+import org.apache.pinot.query.planner.physical.DispatchablePlanFragment;
+import org.apache.pinot.query.planner.physical.DispatchableSubPlan;
+import org.apache.pinot.query.planner.plannode.PlanNode;
+import org.apache.pinot.query.planner.plannode.TableScanNode;
+import org.apache.pinot.query.routing.WorkerManager;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.*;
+
+
+/**
+ * Verifies that querying an empty table (a table that exists and has a
routing entry, but no routable segments) under
+ * the multi-stage physical optimizer does not fail. Such a query should be
short-circuited to an empty result on the
+ * broker, mirroring the single-stage engine and the legacy multi-stage path
(see PinotDispatchPlanner and #18538).
+ */
+public class EmptyTablePlanTest {
+ private static final String EMPTY_TABLE = "emptytable_OFFLINE";
+ private static final String EMPTY_TABLE_2 = "emptytable2_OFFLINE";
+
+ private QueryEnvironment _queryEnvironment;
+
+ @BeforeClass
+ public void setUp() {
+ MockRoutingManagerFactory factory = new MockRoutingManagerFactory(1, 2);
+ for (Map.Entry<String, Schema> entry :
QueryEnvironmentTestBase.TABLE_SCHEMAS.entrySet()) {
+ factory.registerTable(entry.getValue(), entry.getKey());
+ }
+ for (Map.Entry<String, List<String>> entry :
QueryEnvironmentTestBase.SERVER1_SEGMENTS.entrySet()) {
+ for (String segment : entry.getValue()) {
+ factory.registerSegment(1, entry.getKey(), segment);
+ }
+ }
+ for (Map.Entry<String, List<String>> entry :
QueryEnvironmentTestBase.SERVER2_SEGMENTS.entrySet()) {
+ for (String segment : entry.getValue()) {
+ factory.registerSegment(2, entry.getKey(), segment);
+ }
+ }
+ // Register offline tables that have a routing entry but no routable
segments (empty tables).
+
factory.registerTable(QueryEnvironmentTestBase.TABLE_SCHEMAS.get("c_OFFLINE"),
EMPTY_TABLE);
+ factory.registerEmptyTable(EMPTY_TABLE);
+
factory.registerTable(QueryEnvironmentTestBase.TABLE_SCHEMAS.get("c_OFFLINE"),
EMPTY_TABLE_2);
+ factory.registerEmptyTable(EMPTY_TABLE_2);
+
+ RoutingManager routingManager = factory.buildRoutingManager(null);
+ TableCache tableCache = factory.buildTableCache();
+ WorkerManager workerManager = new WorkerManager("Broker_localhost",
"localhost", 3, routingManager);
+
+ _queryEnvironment = new QueryEnvironment(
+ QueryEnvironment.configBuilder()
+ .requestId(-1L)
+ .database(CommonConstants.DEFAULT_DATABASE)
+ .tableCache(tableCache)
+ .workerManager(workerManager)
+ .defaultUsePhysicalOptimizer(true)
+ .defaultInferPartitionHint(true)
+ .build());
+ }
+
+ @Test
+ public void testSelectStarFromEmptyTable() {
+ DispatchableSubPlan plan = _queryEnvironment.planQuery("SELECT * FROM
emptytable");
+ assertNotNull(plan);
+ assertTrue(plan.isAllLeafStagesEmpty(), "All leaf stages should be
detected as empty");
+ assertShortCircuited(plan);
+ }
+
+ @Test
+ public void testCountStarFromEmptyTable() {
+ DispatchableSubPlan plan = _queryEnvironment.planQuery("SELECT COUNT(*)
FROM emptytable");
+ assertNotNull(plan);
+ assertTrue(plan.isAllLeafStagesEmpty(), "All leaf stages should be
detected as empty");
+ assertShortCircuited(plan);
+ }
+
+ @Test
+ public void testFilterFromEmptyTable() {
+ DispatchableSubPlan plan = _queryEnvironment.planQuery("SELECT col1, col3
FROM emptytable WHERE col3 > 5");
+ assertNotNull(plan);
+ assertTrue(plan.isAllLeafStagesEmpty(), "All leaf stages should be
detected as empty");
+ assertShortCircuited(plan);
+ }
+
+ @Test
+ public void testGroupByFromEmptyTable() {
+ DispatchableSubPlan plan =
+ _queryEnvironment.planQuery("SELECT col1, SUM(col3) FROM emptytable
GROUP BY col1");
+ assertNotNull(plan);
+ assertTrue(plan.isAllLeafStagesEmpty(), "All leaf stages should be
detected as empty");
+ assertShortCircuited(plan);
+ }
+
+ @Test
+ public void testNonExistentTableStillErrors() {
+ // A table that does not exist must still fail (caught during validation /
table-cache lookup), not be treated as
+ // an empty table and short-circuited to an empty result.
+ RuntimeException e = expectThrows(RuntimeException.class,
+ () -> _queryEnvironment.planQuery("SELECT * FROM
this_table_does_not_exist"));
+ assertTrue(e.getMessage().toLowerCase().contains("not found")
+ || e.getMessage().toLowerCase().contains("object
'this_table_does_not_exist'")
+ ||
e.getMessage().toLowerCase().contains("this_table_does_not_exist"),
+ "Expected a table-not-found error, got: " + e.getMessage());
+ }
+
+ @Test
+ public void testUnionOfTwoEmptyTablesShortCircuits() {
+ // Multiple leaf stages, all empty: still short-circuited (exercises the
all-empty boundary with >1 leaf).
+ DispatchableSubPlan plan = _queryEnvironment.planQuery(
+ "SELECT col1, col2 FROM emptytable UNION ALL SELECT col1, col2 FROM
emptytable2");
+ assertNotNull(plan);
+ assertTrue(plan.isAllLeafStagesEmpty(), "All leaf stages should be
detected as empty");
+ assertShortCircuited(plan);
+ }
+
+ @Test
+ public void testUnionWithNonEmptyTableFailsFast() {
+ // Combining an empty table with a non-empty table is not yet supported by
the physical optimizer. It must fail
+ // fast with a clear error rather than silently drop rows (see
PinotDispatchPlanner#trackEmptyLeafStages).
+ assertPartiallyEmptyError("SELECT col1, col2 FROM emptytable UNION ALL
SELECT col1, col2 FROM c");
+ }
+
+ @Test
+ public void testJoinWithNonEmptyTableFailsFast() {
+ assertPartiallyEmptyError("SELECT t.col1 FROM emptytable t JOIN c ON
t.col1 = c.col1");
+ }
+
+ private void assertPartiallyEmptyError(String query) {
+ RuntimeException e = expectThrows(RuntimeException.class, () ->
_queryEnvironment.planQuery(query));
+ assertTrue(e.getMessage().contains("combine an empty or fully-pruned table
with a non-empty table"),
+ "Expected a clear partially-empty error, got: " + e.getMessage());
+ }
+
+ /**
+ * When all leaf stages are empty, the plan is rewritten so that only the
broker reduce stage (stage 0) remains,
+ * and its plan tree no longer contains any table scan (the leaves are
inlined to empty literals).
+ */
+ private void assertShortCircuited(DispatchableSubPlan plan) {
+ assertEquals(plan.getQueryStageMap().keySet(), Set.of(0),
+ "Only the reduce stage should remain after short-circuiting empty
leaves");
+ DispatchablePlanFragment reduceStage = plan.getQueryStageMap().get(0);
+
assertFalse(containsTableScan(reduceStage.getPlanFragment().getFragmentRoot()),
+ "Reduce stage should not contain a TableScanNode after inlining empty
leaves");
+ }
+
+ private static boolean containsTableScan(PlanNode node) {
+ if (node instanceof TableScanNode) {
+ return true;
+ }
+ for (PlanNode input : node.getInputs()) {
+ if (containsTableScan(input)) {
+ return true;
+ }
+ }
+ return false;
+ }
+}
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LeafStageWorkerAssignmentRuleTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LeafStageWorkerAssignmentRuleTest.java
index 870399cfc6b..14b4d9b5992 100644
---
a/pinot-query-planner/src/test/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LeafStageWorkerAssignmentRuleTest.java
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/query/planner/physical/v2/opt/rules/LeafStageWorkerAssignmentRuleTest.java
@@ -115,6 +115,18 @@ public class LeafStageWorkerAssignmentRuleTest {
validateTableScanAssignment(result,
HYBRID_INSTANCE_ID_TO_SEGMENTS._realtimeTableSegmentsMap, "REALTIME");
}
+ @Test
+ public void testAssignTableScanWithEmptyTable() {
+ // An empty table (or one whose segments were all pruned) has a routing
entry but no routable segments, so
+ // neither the offline nor the realtime segments map is populated. This
should yield a zero-worker assignment
+ // (which is later short-circuited to an empty result on the broker), NOT
throw. See LeafStageWorkerAssignmentRule.
+ InstanceIdToSegments emptyInstanceIdToSegments = new
InstanceIdToSegments();
+ TableScanWorkerAssignmentResult result =
LeafStageWorkerAssignmentRule.assignTableScan(TABLE_NAME, FIELDS_IN_SCAN,
+ emptyInstanceIdToSegments, Map.of(), false, false);
+ assertTrue(result._pinotDataDistribution.getWorkers().isEmpty(), "Empty
table should assign zero workers");
+ assertTrue(result._workerIdToSegmentsMap.isEmpty(), "Empty table should
have no worker-to-segments entries");
+ }
+
@Test
public void testAssignTableScanPartitionedOfflineTable() {
TableScanWorkerAssignmentResult result =
LeafStageWorkerAssignmentRule.assignTableScan(TABLE_NAME, FIELDS_IN_SCAN,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
