yashmayya opened a new pull request, #19174: URL: https://github.com/apache/pinot/pull/19174
Follow-up to #19166, which reduced a colocated join to the partition classes that hold data. This does the same for the classes that a filter empties, so a filtered colocated join runs on fewer workers and reaches fewer servers. Only the default (logical) MSE planner is affected. The V2 physical optimizer already derives its partitions from a pruned routing table. ## What this enables A colocated join over two partitioned tables, with a filter on the partition key: ```sql SELECT /*+ joinOptions(is_colocated_by_join_keys='true') */ l.key, r.value FROM left /*+ tableOptions(partition_key='key', partition_size='8', partition_function='Modulo') */ l JOIN right /*+ tableOptions(partition_key='key', partition_size='8', partition_function='Modulo') */ r ON l.key = r.key WHERE l.key IN (1, 2) AND r.key IN (1, 2) ``` Both sides prune every partition class except 1 and 2. The group drops the rest. The result: - **Fewer workers.** Each leaf runs 2 workers instead of 8. - **Fewer servers.** The broker dispatches only to the servers that hold classes 1 and 2. It also waits on only those servers, so the query loses the tail latency of the servers it skips. - **Fewer segments.** Each worker scans only its own class. - **A cheaper cancel.** Cancellation reaches the same reduced server set. These cases get the reduction: - A filter on the partition key, written on both sides. - A filter on the partition key on one side, when Calcite carries it to the other side through the join equality. - A partition class that neither table populates. #19166 already dropped these. The 1-to-1 exchange survives the reduction. Every leaf of the group keeps one worker per surviving class, in the same order, so worker `k` means the same class on every side. ## What this does not help The common case is in this list, so it is worth stating plainly. - **A time filter.** The time pruner removes segments, not partitions. A table partitioned by key and segmented by time holds every time range in every partition. This is the most common filter in Pinot, and it buys nothing here. - **A filter on one side only.** A class survives while any member of the group still holds a matching row. An unfiltered side keeps every class it populates. See the trade-off below. - **A table without `segmentPrunerTypes: ["partition"]`.** Pinot builds the partition pruner only when the routing config asks for it. Without it, nothing is pruned. - **A filter that matches nothing.** The group keeps every populated class. A group with zero workers has nothing to wire a 1-to-1 exchange to, and the servers return the same empty result anyway. - **The probe leaf of a colocated semi-join.** That leaf holds the join, so the routing-query builder cannot fold it. - **A non-colocated stage above the join, with the default config.** `getCandidateServersPerTables` ignores the query and expands the dispatched set again. With `useLeafServerForIntermediateStage=true`, the reduction carries upward. ## Trade-offs **A class drops only when every member of the group prunes it.** The alternative is to drop a class as soon as one side prunes it. That is wrong for a RIGHT join, a FULL join, a union, and an anti-join. In each of those, the other side still produces rows. The union rule needs no knowledge of the operator above. A new colocated operator therefore cannot break it in silence. The cost is the one-sided filter case above. **A surviving class dispatches all of its segments on every member.** A member whose own filter excludes them still scans them. The server-side filter removes them again. This keeps the class on the same server it uses without pruning, so the exchange stays in process. A finer verdict inside a surviving class is possible later. **Planning costs one routing call per member.** Only a leaf that carries a filter pays it. The result is cached per fragment, so the leaf assignment reuses it. ## A sound proof of emptiness The old verdict read "this partition is pruned" from the absence of its segments in the routing table. Absence has innocent causes. Instance selection can class a segment as optional. The server that holds a segment can leave the enabled server map. A segment can enter the partition metadata before it becomes selectable. Each of these loses matching rows. This adds a planner-only entry point that reports which segments the pruners **pruned**: ```java // RoutingManager, added as a default that proves nothing, so no implementation breaks @Nullable Set<String> getPrunedSegments(BrokerRequest brokerRequest); ``` Only presence in that set is a proof. Instance selection takes no part and there is no request id, so two leaves over one table and one filter cannot disagree. `MultiClusterRoutingManager` combines the set by intersection, because a proof holds only when every cluster that can route the segment pruned it. The ordinary partitioned leaf now uses the same proof. That path is strictly more conservative than before. ## How to turn it off No new flag. The query option `useBrokerPruning` and the broker config `pinot.broker.multistage.logical.planner.use.broker.pruning` already do it. With pruning off, every populated class keeps its worker, which is the behavior of #19166. ## Tests - `WorkerManagerTest` covers both sides pruning and one side pruning. It also covers the all-pruned fallback, an unavailable segment, a routing failure and the kill switch. Two more cases cover padding together with reduction, and a self-join whose sides prune different classes. - `BrokerRoutingManagerTest` and `MultiClusterRoutingManagerTest` cover the pruned set and the multi-cluster intersection. - `ColocatedJoinEmptyPartitionTest` runs the filtered join end to end. It proves the 1-to-1 wiring at the reduced width. It also matches the rows against the same query without the colocation hint. A filter that matches nothing returns an empty result. The integration fixture had no `RoutingConfig`, so it built no partition pruner. This adds one. Without it a filtered assertion proves nothing. ## Notes - Depends on #19166. - Includes the pre-partitioned leaf guard from #19173. Review that one first. - The dispatched server count does not move in the integration cluster. It runs 2 servers with `numReplicas 2`, so both servers hold every segment. The end-to-end tests assert worker count, segments queried, fan-out and receiver ids. `WorkerManagerTest` asserts the dispatched server set. - A worker with no segments on an upsert table can still acquire newly-added segments, so it does not always scan nothing. This predates the change and is filed separately. -- 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]
