xiangfu0 commented on code in PR #19396: URL: https://github.com/apache/pinot/pull/19396#discussion_r4213852932
########## pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java: ########## @@ -0,0 +1,99 @@ +/** + * 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.broker.requesthandler; + +import java.util.List; +import java.util.function.BooleanSupplier; +import org.apache.helix.HelixAdmin; +import org.apache.helix.HelixManager; +import org.apache.helix.NotificationContext; +import org.apache.helix.api.listeners.BatchMode; +import org.apache.helix.api.listeners.InstanceConfigChangeListener; +import org.apache.helix.api.listeners.PreFetch; +import org.apache.helix.model.InstanceConfig; +import org.apache.pinot.common.version.PinotVersion; +import org.apache.pinot.spi.config.instance.InstanceType; +import org.apache.pinot.spi.utils.CommonConstants; +import org.apache.pinot.spi.utils.InstanceTypeUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot. +/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures. +/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations. +/// Helix drives updates serially; query threads read the volatile result without locking. +@BatchMode(enabled = false) +@PreFetch(enabled = false) +public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier { + private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class); + private final HelixManager _manager; + private final String _version; + private volatile boolean _supported; + + public KWayMergeSupportPredicate(HelixManager manager) { + this(manager, PinotVersion.VERSION); + } + + KWayMergeSupportPredicate(HelixManager manager, String version) { + _manager = manager; + _version = version; + } + + @Override + public boolean getAsBoolean() { + return _supported && _manager.isConnected(); + } + + @Override + public synchronized void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext context) { + // Clear first: no partial snapshot or failed refresh may leave the new wire node enabled. + _supported = false; + if (context.getType() != NotificationContext.Type.INIT + && context.getType() != NotificationContext.Type.CALLBACK) { + return; + } + if (_version == null || _version.equals(PinotVersion.UNKNOWN) || _version.contains("SNAPSHOT")) { Review Comment: Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current implementation. The current WindowKWayMergeIntegrationTest forces the test-only supplier true inside try/finally and resets it to false in finally; it does not restore a saved production supplier. No production query option bypasses the release-version gate. The fixture contains 8 offline + 8 realtime rows and compares windowKWayMerge=true/false for ROWS/FOLLOWING and RANGE/peer frames with DESC NULLS FIRST, null and duplicate keys. It asserts complete 16-row results and a physical merge plan through the real cluster planner/proto/server/opchain path; it does not assert local/gRPC mailbox statistics. The earlier local pass is historical validation, not a claim about fresh rebased-head CI. ########## pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/KWayMergeSupportPredicate.java: ########## @@ -0,0 +1,99 @@ +/** + * 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.broker.requesthandler; + +import java.util.List; +import java.util.function.BooleanSupplier; +import org.apache.helix.HelixAdmin; +import org.apache.helix.HelixManager; +import org.apache.helix.NotificationContext; +import org.apache.helix.api.listeners.BatchMode; +import org.apache.helix.api.listeners.InstanceConfigChangeListener; +import org.apache.helix.api.listeners.PreFetch; +import org.apache.helix.model.InstanceConfig; +import org.apache.pinot.common.version.PinotVersion; +import org.apache.pinot.spi.config.instance.InstanceType; +import org.apache.pinot.spi.utils.CommonConstants; +import org.apache.pinot.spi.utils.InstanceTypeUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/// Enables new merge plan nodes only after a complete homogeneous broker/server release-version snapshot. +/// Uses the SAFE stats policy's version comparison, but fails closed on initialization and read failures. +/// Snapshot builds are excluded because equal snapshot versions can contain different protocol implementations. +/// Helix drives updates serially; query threads read the volatile result without locking. +@BatchMode(enabled = false) +@PreFetch(enabled = false) +public class KWayMergeSupportPredicate implements InstanceConfigChangeListener, BooleanSupplier { + private static final Logger LOGGER = LoggerFactory.getLogger(KWayMergeSupportPredicate.class); + private final HelixManager _manager; + private final String _version; + private volatile boolean _supported; + + public KWayMergeSupportPredicate(HelixManager manager) { + this(manager, PinotVersion.VERSION); + } + + KWayMergeSupportPredicate(HelixManager manager, String version) { + _manager = manager; + _version = version; + } + + @Override + public boolean getAsBoolean() { + return _supported && _manager.isConnected(); + } + + @Override + public synchronized void onInstanceConfigChange(List<InstanceConfig> configs, NotificationContext context) { + // Clear first: no partial snapshot or failed refresh may leave the new wire node enabled. + _supported = false; + if (context.getType() != NotificationContext.Type.INIT + && context.getType() != NotificationContext.Type.CALLBACK) { + return; + } + if (_version == null || _version.equals(PinotVersion.UNKNOWN) || _version.contains("SNAPSHOT")) { + return; + } + try { + HelixAdmin admin = _manager.getClusterManagmentTool(); + String cluster = _manager.getClusterName(); + List<String> instances = admin.getInstancesInCluster(cluster); Review Comment: Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current implementation. The current shared ClusterVersionTracker is reused by the merge gate and SendStatsPredicate.Safe. Config data changes read only a changed tracked live broker/server config. Topology callbacks read live child names, reuse cached configs for still-present names and read newly live participants; INIT or recovery after a failed read refreshes every live participant. Non-live configs do not block the gate. This implementation does not track live-session IDs or separately detect same-name restarts; config notifications provide version updates. Tests cover incremental read counts, stale non-live configs and failure recovery, rather than session-based restart detection. ########## pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java: ########## @@ -583,6 +584,13 @@ public void start() _accessControlFactory, _queryQuotaManager, _tableCache, _multiStageQueryThrottler, _failureDetector, _threadAccountant, multiClusterRoutingContext, workerManager, multiClusterWorkerManager, _serverRoutingStatsManager); + KWayMergeSupportPredicate mergeSupport = new KWayMergeSupportPredicate(_spectatorHelixManager); + try { + _spectatorHelixManager.addInstanceConfigChangeListener(mergeSupport); Review Comment: Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current implementation. The current implementation registers both incremental listeners even for SNAPSHOT/unknown brokers, but the tracker disables unsupported local versions and returns before callback reads. Release brokers keep those listeners because an explicit windowKWayMerge=true query can override a false broker default. Option-off queries do not consult the supplier. Config data events read only the changed tracked live participant; topology updates reuse still-present configs. If registration fails, both listeners are removed and the handler retains its default disabled supplier; there is no permanent-stop method on the tracker. ########## pinot-query-planner/src/main/java/org/apache/pinot/query/planner/plannode/MailboxMergeReceiveNode.java: ########## @@ -0,0 +1,87 @@ +/** + * 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.plannode; + +import java.util.List; +import java.util.Objects; +import javax.annotation.Nullable; +import org.apache.calcite.rel.RelDistribution; +import org.apache.calcite.rel.RelFieldCollation; +import org.apache.pinot.calcite.rel.logical.PinotRelExchangeType; +import org.apache.pinot.common.utils.DataSchema; + + +/// Receives streams ordered by the logical plan and preserves that order with a k-way merge. +/// This distinct wire node is emitted only when the broker verifies homogeneous cluster versions. +/// Instances belong to a single query plan and are not thread-safe. +public class MailboxMergeReceiveNode extends MailboxReceiveNode { + private final int _fetch; + private final int _offset; + + public MailboxMergeReceiveNode(int stageId, DataSchema schema, int senderStageId, + PinotRelExchangeType exchangeType, RelDistribution.Type distribution, @Nullable List<Integer> keys, + List<RelFieldCollation> collations, @Nullable MailboxSendNode sender) { + this(stageId, schema, senderStageId, exchangeType, distribution, keys, collations, -1, -1, sender); + } + + public MailboxMergeReceiveNode(int stageId, DataSchema schema, int senderStageId, + PinotRelExchangeType exchangeType, RelDistribution.Type distribution, @Nullable List<Integer> keys, + List<RelFieldCollation> collations, int fetch, int offset, @Nullable MailboxSendNode sender) { + super(stageId, schema, senderStageId, exchangeType, distribution, keys, collations, false, false, sender); Review Comment: Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current implementation. MailboxMergeReceiveNode and ordinary MailboxReceiveNode are now sibling classes under BaseMailboxReceiveNode, with a dedicated visitMailboxMergeReceive method. The merge receive exposes collations without the ordinary receiver sorting flags. The legacy sort/sortedOnSender Java accessors and proto fields, including sender field 7, are deprecated but retained for decoding old plans. Earlier serde/planner/opchain/runtime validation is historical evidence; fresh rebased-head CI remains separate. ########## pinot-query-planner/src/main/java/org/apache/pinot/query/planner/plannode/KWayMergeExchangeNode.java: ########## @@ -0,0 +1,44 @@ +/** + * 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.plannode; + +import java.util.List; +import javax.annotation.Nullable; +import org.apache.calcite.rel.RelDistribution; +import org.apache.calcite.rel.RelFieldCollation; +import org.apache.pinot.calcite.rel.logical.PinotRelExchangeType; +import org.apache.pinot.common.utils.DataSchema; + + +/// Planning-only exchange requiring ordered inputs and producing a k-way merge receive and a plain send. +/// Instances are local to a query's planning thread and are never serialized. +public class KWayMergeExchangeNode extends ExchangeNode { + public KWayMergeExchangeNode(int stageId, DataSchema schema, List<PlanNode> inputs, + RelDistribution.Type distribution, @Nullable List<Integer> keys, boolean prePartitioned, + List<RelFieldCollation> collations, String hashFunction) { + super(stageId, schema, inputs, PinotRelExchangeType.STREAMING, distribution, keys, prePartitioned, collations, Review Comment: Updated at `85be3b0e80e4265ceeb6258e7db4582c4415511c` to reflect the current implementation. The current implementation uses the single-exchange alternative you suggested: ExchangeNode.kWayMerge(...) records an explicit isKWayMerge mode, rather than a separate KWayMergeExchangeNode subtype. It retains the existing visitExchange method; there is no visitKWayMergeExchange method. Fragmentation creates a plain mailbox sender plus the sibling MailboxMergeReceiveNode, whose dedicated receive visitor carries the ordered-stream contract. -- 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]
