You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "shauryachats (via GitHub)" <gi...@apache.org> on 2025/11/05 03:15:49 UTC
[PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
shauryachats opened a new pull request, #17145:
URL: https://github.com/apache/pinot/pull/17145
(no comment)
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2535076356
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/TimeSeriesRequestHandler.java:
##########
@@ -52,6 +52,7 @@
import org.apache.pinot.common.utils.HumanReadableDuration;
import org.apache.pinot.core.auth.Actions;
import org.apache.pinot.core.auth.TargetType;
+import org.apache.pinot.core.routing.FederationProvider;
Review Comment:
I have not tested it out yet, but I will extend support for federation subsequently for timeseries. Timeseries engine still needs support for query options since federation is gated by the `enableFederation` query option.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2578502782
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
In that case do we need this field in the `PhysicalTableConfig`?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats closed pull request #17145: [DRAFT] Implementing federation for logical tables in Pinot
URL: https://github.com/apache/pinot/pull/17145
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2578620562
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
The `PhysicalTableConfig` resides within the `LogicalTableConfig` only: https://github.com/apache/pinot/blob/master/pinot-spi/src/main/java/org/apache/pinot/spi/data/LogicalTableConfig.java#L70
This `PhysicalTableConfig` is not used anywhere else. No changes are being made to any other table config apart from logical tables.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2557854873
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
In the current implementation, a physical table CAN be federated across multiple clusters simply by passing the `SET enableFederation=true; ` flag only. We can decide if we want to expose federation for physical tables or only keep it limited to logical tables.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2566182735
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/FederatedRoutingManager.java:
##########
@@ -0,0 +1,343 @@
+/**
+ * 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.routing;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.request.BrokerRequest;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils;
+import org.apache.pinot.core.routing.RoutingManager;
+import org.apache.pinot.core.routing.RoutingTable;
+import org.apache.pinot.core.routing.SegmentsToQuery;
+import org.apache.pinot.core.routing.TablePartitionInfo;
+import org.apache.pinot.core.routing.TablePartitionReplicatedServersInfo;
+import org.apache.pinot.core.routing.timeboundary.TimeBoundaryInfo;
+import org.apache.pinot.core.transport.ServerInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * FederatedRoutingManager manages routing across multiple ZooKeeper clusters.
+ * It maintains a primary BrokerRoutingManager and multiple secondary routing managers,
+ * each connected to their own ZooKeeper cluster.
+ */
+public class FederatedRoutingManager implements RoutingManager {
+ private static final Logger LOGGER = LoggerFactory.getLogger(FederatedRoutingManager.class);
+
+ // Primary routing manager (connected to primary ZooKeeper cluster)
+ private final BrokerRoutingManager _primaryRoutingManager;
+
+ // Secondary routing managers (connected to secondary ZooKeeper clusters)
+ private final List<BrokerRoutingManager> _secondaryRoutingManagers;
+
+ // Combined routing information
+ private final Map<String, ServerInstance> _combinedServerInstanceMap = new ConcurrentHashMap<>();
+ private final Map<String, RoutingTable> _combinedRoutingTableCache = new ConcurrentHashMap<>();
+
+ public FederatedRoutingManager(BrokerRoutingManager primaryRoutingManager,
+ List<BrokerRoutingManager> secondaryRoutingManagers) {
+ _primaryRoutingManager = primaryRoutingManager;
+ _secondaryRoutingManagers = secondaryRoutingManagers;
+ }
+
+ private void updateCombinedServerInstances() {
+ _combinedServerInstanceMap.clear();
+
+ // Add server instances from primary routing manager
+ Map<String, ServerInstance> primaryServerInstances = _primaryRoutingManager.getEnabledServerInstanceMap();
+ _combinedServerInstanceMap.putAll(primaryServerInstances);
+
+ // Add server instances from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ Map<String, ServerInstance> secondaryServerInstances = secondaryRoutingManager.getEnabledServerInstanceMap();
+ _combinedServerInstanceMap.putAll(secondaryServerInstances);
+ }
+
+ LOGGER.info("Updated combined server instances. Total: {}", _combinedServerInstanceMap.size());
+ }
+
+ @Override
+ public boolean routingExists(String tableNameWithType) {
+ // Check if routing exists in primary routing manager
+ if (_primaryRoutingManager.routingExists(tableNameWithType)) {
+ return true;
+ }
+
+ // Check if routing exists in any secondary routing manager
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ if (secondaryRoutingManager.routingExists(tableNameWithType)) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ @Override
+ public boolean isTableDisabled(String tableNameWithType) {
+ // Check if table is disabled in primary routing manager
+ if (_primaryRoutingManager.isTableDisabled(tableNameWithType)) {
+ return true;
+ }
+
+ // Check if table is disabled in any secondary routing manager
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ if (secondaryRoutingManager.isTableDisabled(tableNameWithType)) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public RoutingTable getRoutingTable(BrokerRequest brokerRequest, long requestId) {
+ String tableNameWithType = brokerRequest.getQuerySource().getTableName();
+ return getRoutingTable(brokerRequest, tableNameWithType, requestId);
+ }
+
+ @Nullable
+ @Override
+ public RoutingTable getRoutingTable(BrokerRequest brokerRequest, String tableNameWithType, long requestId) {
+ // Check cache first
+ String cacheKey = tableNameWithType + "_" + requestId;
+ RoutingTable cachedRoutingTable = _combinedRoutingTableCache.get(cacheKey);
+ if (cachedRoutingTable != null) {
+ return cachedRoutingTable;
+ }
+
+ // Get routing table from primary routing manager
+ RoutingTable primaryRoutingTable = _primaryRoutingManager.getRoutingTable(brokerRequest,
+ tableNameWithType, requestId);
+
+ // Combine with routing tables from secondary routing managers
+ RoutingTable combinedRoutingTable = combineRoutingTables(primaryRoutingTable, tableNameWithType,
+ brokerRequest, requestId);
+
+ // Cache the result
+ if (combinedRoutingTable != null) {
+ _combinedRoutingTableCache.put(cacheKey, combinedRoutingTable);
+ }
+
+ return combinedRoutingTable;
+ }
+
+ private RoutingTable combineRoutingTables(RoutingTable primaryRoutingTable, String tableNameWithType,
+ BrokerRequest brokerRequest, long requestId) {
+ if (primaryRoutingTable == null) {
+ // If primary routing table is null, try to get routing table from secondary managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ RoutingTable secondaryRoutingTable = secondaryRoutingManager.getRoutingTable(brokerRequest,
+ tableNameWithType, requestId);
+ if (secondaryRoutingTable != null) {
+ return secondaryRoutingTable;
+ }
+ }
+ return null;
+ }
+
+ // Start with primary routing table
+ Map<ServerInstance, SegmentsToQuery> combinedServerInstanceToSegmentsMap =
+ new HashMap<>(primaryRoutingTable.getServerInstanceToSegmentsMap());
+ List<String> combinedUnavailableSegments = new ArrayList<>(primaryRoutingTable.getUnavailableSegments());
+ int combinedNumPrunedSegments = primaryRoutingTable.getNumPrunedSegments();
+
+ // Add routing information from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ RoutingTable secondaryRoutingTable = secondaryRoutingManager.getRoutingTable(brokerRequest,
+ tableNameWithType, requestId);
+ if (secondaryRoutingTable != null) {
+ // Combine server instance to segments map
+ for (Map.Entry<ServerInstance, SegmentsToQuery> entry
+ : secondaryRoutingTable.getServerInstanceToSegmentsMap().entrySet()) {
+ ServerInstance serverInstance = entry.getKey();
+ SegmentsToQuery secondaryRouteInfo = entry.getValue();
+
+ SegmentsToQuery existingRouteInfo = combinedServerInstanceToSegmentsMap.get(serverInstance);
+ if (existingRouteInfo != null) {
+ // Merge segments
+ existingRouteInfo.getSegments().addAll(secondaryRouteInfo.getSegments());
+ existingRouteInfo.getOptionalSegments().addAll(secondaryRouteInfo.getOptionalSegments());
+ } else {
+ // Add new server instance
+ combinedServerInstanceToSegmentsMap.put(serverInstance, secondaryRouteInfo);
+ }
+ }
+
+ // Combine unavailable segments
+ combinedUnavailableSegments.addAll(secondaryRoutingTable.getUnavailableSegments());
+
+ // Add pruned segments count
+ combinedNumPrunedSegments += secondaryRoutingTable.getNumPrunedSegments();
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error combining routing table from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return new RoutingTable(combinedServerInstanceToSegmentsMap,
+ combinedUnavailableSegments, combinedNumPrunedSegments);
+ }
+
+ @Nullable
+ @Override
+ public TimeBoundaryInfo getTimeBoundaryInfo(String tableNameWithType) {
+ // Try to get time boundary info from primary routing manager first
+ TimeBoundaryInfo primaryTimeBoundaryInfo = _primaryRoutingManager.getTimeBoundaryInfo(tableNameWithType);
+ if (primaryTimeBoundaryInfo != null) {
+ return primaryTimeBoundaryInfo;
+ }
+
+ // Try to get time boundary info from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ TimeBoundaryInfo secondaryTimeBoundaryInfo = secondaryRoutingManager.getTimeBoundaryInfo(tableNameWithType);
+ if (secondaryTimeBoundaryInfo != null) {
+ return secondaryTimeBoundaryInfo;
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error getting time boundary info from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return null;
+ }
+
+ @Override
+ public Map<String, ServerInstance> getEnabledServerInstanceMap() {
+ updateCombinedServerInstances();
+ return _combinedServerInstanceMap;
+ }
+
+ @Override
+ public TablePartitionInfo getTablePartitionInfo(String tableNameWithType) {
+ // Try to get table partition info from primary routing manager first
+ TablePartitionInfo primaryTablePartitionInfo = _primaryRoutingManager.getTablePartitionInfo(tableNameWithType);
+ if (primaryTablePartitionInfo != null) {
+ return primaryTablePartitionInfo;
+ }
+
+ // Try to get table partition info from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ TablePartitionInfo secondaryTablePartitionInfo =
+ secondaryRoutingManager.getTablePartitionInfo(tableNameWithType);
+ if (secondaryTablePartitionInfo != null) {
+ return secondaryTablePartitionInfo;
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error getting table partition info from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return null;
+ }
+
+ @Override
+ public Set<String> getServingInstances(String tableNameWithType) {
+ // Get serving instances from primary routing manager
+ Set<String> primaryServingInstances = _primaryRoutingManager.getServingInstances(tableNameWithType);
+ if (primaryServingInstances == null) {
+ primaryServingInstances = Collections.emptySet();
+ }
+ // Combine with serving instances from secondary routing managers
+ Set<String> combinedServingInstances = new HashSet<>(primaryServingInstances);
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ Set<String> secondaryServingInstances = secondaryRoutingManager.getServingInstances(tableNameWithType);
+ if (secondaryServingInstances != null) {
+ combinedServingInstances.addAll(secondaryServingInstances);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error getting serving instances from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return combinedServingInstances.isEmpty() ? null : combinedServingInstances;
+ }
+
+ @Override
+ public List<String> getSegments(BrokerRequest brokerRequest) {
+ // Get segments from primary routing manager
+ List<String> primarySegments = _primaryRoutingManager.getSegments(brokerRequest);
+
+ // Combine with segments from secondary routing managers
+ List<String> combinedSegments = new ArrayList<>(primarySegments);
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ List<String> secondarySegments = secondaryRoutingManager.getSegments(brokerRequest);
+ combinedSegments.addAll(secondarySegments);
+ } catch (Exception e) {
+ LOGGER.error("Error getting segments from secondary routing manager", e);
+ }
+ }
+
+ return combinedSegments;
+ }
+
+ @Override
+ public TablePartitionReplicatedServersInfo getTablePartitionReplicatedServersInfo(String tableNameWithType) {
+ // Try to get table partition replicated servers info from primary routing manager first
+ TablePartitionReplicatedServersInfo primaryInfo =
+ _primaryRoutingManager.getTablePartitionReplicatedServersInfo(tableNameWithType);
+ if (primaryInfo != null) {
+ return primaryInfo;
+ }
+
+ // Try to get table partition replicated servers info from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ TablePartitionReplicatedServersInfo secondaryInfo =
+ secondaryRoutingManager.getTablePartitionReplicatedServersInfo(tableNameWithType);
+ if (secondaryInfo != null) {
+ return secondaryInfo;
+ }
+ } catch (Exception e) {
+ LOGGER.error(
+ "Error getting table partition replicated servers info from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return null;
+ }
+
+ public RoutingManager getPrimaryRoutingManager(Map<String, String> queryOptions) {
Review Comment:
Delegated getting the correct RoutingManager to `CrossClusterFederationProvider`.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2566737844
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
Yes. We will add logic to leverage the `FederatedRoutingManager` only when the table in question is a logical table. Physical table will never be federated, even with a `SET enableFederation=true;`.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2578620562
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
The `PhysicalTableConfig` resides within the `LogicalTableConfig` only: https://github.com/apache/pinot/blob/master/pinot-spi/src/main/java/org/apache/pinot/spi/data/LogicalTableConfig.java#L70
This `PhysicalTableConfig` is not used anywhere else. We
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
The `PhysicalTableConfig` resides within the `LogicalTableConfig` only: https://github.com/apache/pinot/blob/master/pinot-spi/src/main/java/org/apache/pinot/spi/data/LogicalTableConfig.java#L70
This `PhysicalTableConfig` is not used anywhere else.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #17145:
URL: https://github.com/apache/pinot/pull/17145#issuecomment-3514671960
### :x: 1 Tests Failed:
| Tests completed | Failed | Passed | Skipped |
|---|---|---|---|
| 5830 | 1 | 5829 | 0 |
<details><summary>View the top 1 failed test(s) by shortest run time</summary>
> <pre><code class="language-python">org.apache.pinot.broker.requesthandler.BaseSingleStageBrokerRequestHandlerTest::testCancelQuery</code></pre>
> <details><summary>Stack Traces | 6.08s run time</summary>
>
> > <pre><code class="language-python">Failed to meet condition in 5000ms, error message: Failed to submit query</code></pre>
>
> </details>
</details>
To view more test analytics, go to the [Test Analytics Dashboard](https://app.codecov.io/gh/apache/pinot/tests/shauryachats%3Afederated_logical_rebased)
<sub>📋 Got 3 mins? [Take this short survey](https://forms.gle/22i53Qa1CySZjA6c7) to help us improve Test Analytics.</sub>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2535073883
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java:
##########
@@ -426,7 +430,8 @@ private ImmutableQueryEnvironment.Config getQueryEnvConf(HttpHeaders httpHeaders
.requestId(requestId)
.database(database)
.tableCache(_tableCache)
- .workerManager(_workerManager)
+ .workerManager(new WorkerManager(_brokerId, _hostname, _port,
Review Comment:
If we want to retain the _workerManager, we would have to pass in `queryOptions` for each call for its member methods, to determine when to use the federatedRoutingManager or the brokerRoutingManager.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "ankitsultana (via GitHub)" <gi...@apache.org>.
ankitsultana commented on PR #17145:
URL: https://github.com/apache/pinot/pull/17145#issuecomment-3549418948
@Jackie-Jiang : could we get your review on this? It's a P0 from our end so Shaurya can address feedback asap
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2566184749
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java:
##########
@@ -426,7 +430,8 @@ private ImmutableQueryEnvironment.Config getQueryEnvConf(HttpHeaders httpHeaders
.requestId(requestId)
.database(database)
.tableCache(_tableCache)
- .workerManager(_workerManager)
+ .workerManager(new WorkerManager(_brokerId, _hostname, _port,
Review Comment:
Delegated getting the correct RoutingManager to `CrossClusterFederationProvider` and instead of initing a new workerManager, created two workerManagers: regular and federated and selecting the correct one based on query options.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2557850029
##########
pinot-core/src/main/java/org/apache/pinot/core/routing/FederationProvider.java:
##########
@@ -0,0 +1,43 @@
+/**
+ * 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.routing;
+
+import java.util.Map;
+import org.apache.pinot.common.config.provider.TableCache;
+
+
+/**
+ * A generic class which provides the dependencies for federation routing.
+ */
+public class FederationProvider {
Review Comment:
Makes sense, will rethink a better name.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2557653120
##########
pinot-core/src/main/java/org/apache/pinot/core/routing/FederationProvider.java:
##########
@@ -0,0 +1,43 @@
+/**
+ * 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.routing;
+
+import java.util.Map;
+import org.apache.pinot.common.config.provider.TableCache;
+
+
+/**
+ * A generic class which provides the dependencies for federation routing.
+ */
+public class FederationProvider {
+ // Maps clusterName to TableCache.
+ Map<String, TableCache> _tableCacheMap;
Review Comment:
(minor) `private final`
##########
pinot-core/src/main/java/org/apache/pinot/core/routing/FederationProvider.java:
##########
@@ -0,0 +1,43 @@
+/**
+ * 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.routing;
+
+import java.util.Map;
+import org.apache.pinot.common.config.provider.TableCache;
+
+
+/**
+ * A generic class which provides the dependencies for federation routing.
+ */
+public class FederationProvider {
Review Comment:
This name is too general. Rename it to something more specific, e.g. `CrossClusterRoutingInfoProvider`
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
Is there a concept of federated physical table?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2578924280
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
I see. Then that makes sense
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2566185903
##########
pinot-core/src/main/java/org/apache/pinot/core/routing/FederationProvider.java:
##########
@@ -0,0 +1,43 @@
+/**
+ * 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.routing;
+
+import java.util.Map;
+import org.apache.pinot.common.config.provider.TableCache;
+
+
+/**
+ * A generic class which provides the dependencies for federation routing.
+ */
+public class FederationProvider {
Review Comment:
Renamed to `CrossClusterFederationProvider`.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2578601039
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
If the federation concept is within logical table only, the config should reside in the `LogicalTableConfig`. `isFederated` property should not be associated with a physical table. E.g. what if a physical table is federated in one logical table, but not the other one?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "ankitsultana (via GitHub)" <gi...@apache.org>.
ankitsultana commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2529193161
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/TimeSeriesRequestHandler.java:
##########
@@ -52,6 +52,7 @@
import org.apache.pinot.common.utils.HumanReadableDuration;
import org.apache.pinot.core.auth.Actions;
import org.apache.pinot.core.auth.TargetType;
+import org.apache.pinot.core.routing.FederationProvider;
Review Comment:
Do time series engine queries also work in our test cluster setup at Uber?
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java:
##########
@@ -426,7 +430,8 @@ private ImmutableQueryEnvironment.Config getQueryEnvConf(HttpHeaders httpHeaders
.requestId(requestId)
.database(database)
.tableCache(_tableCache)
- .workerManager(_workerManager)
+ .workerManager(new WorkerManager(_brokerId, _hostname, _port,
Review Comment:
let's avoid this change and re-use `_workerManager`
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/FederatedRoutingManager.java:
##########
@@ -0,0 +1,343 @@
+/**
+ * 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.routing;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.request.BrokerRequest;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils;
+import org.apache.pinot.core.routing.RoutingManager;
+import org.apache.pinot.core.routing.RoutingTable;
+import org.apache.pinot.core.routing.SegmentsToQuery;
+import org.apache.pinot.core.routing.TablePartitionInfo;
+import org.apache.pinot.core.routing.TablePartitionReplicatedServersInfo;
+import org.apache.pinot.core.routing.timeboundary.TimeBoundaryInfo;
+import org.apache.pinot.core.transport.ServerInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * FederatedRoutingManager manages routing across multiple ZooKeeper clusters.
+ * It maintains a primary BrokerRoutingManager and multiple secondary routing managers,
+ * each connected to their own ZooKeeper cluster.
+ */
+public class FederatedRoutingManager implements RoutingManager {
+ private static final Logger LOGGER = LoggerFactory.getLogger(FederatedRoutingManager.class);
+
+ // Primary routing manager (connected to primary ZooKeeper cluster)
+ private final BrokerRoutingManager _primaryRoutingManager;
+
+ // Secondary routing managers (connected to secondary ZooKeeper clusters)
+ private final List<BrokerRoutingManager> _secondaryRoutingManagers;
+
+ // Combined routing information
+ private final Map<String, ServerInstance> _combinedServerInstanceMap = new ConcurrentHashMap<>();
+ private final Map<String, RoutingTable> _combinedRoutingTableCache = new ConcurrentHashMap<>();
+
+ public FederatedRoutingManager(BrokerRoutingManager primaryRoutingManager,
+ List<BrokerRoutingManager> secondaryRoutingManagers) {
+ _primaryRoutingManager = primaryRoutingManager;
+ _secondaryRoutingManagers = secondaryRoutingManagers;
+ }
+
+ private void updateCombinedServerInstances() {
+ _combinedServerInstanceMap.clear();
+
+ // Add server instances from primary routing manager
+ Map<String, ServerInstance> primaryServerInstances = _primaryRoutingManager.getEnabledServerInstanceMap();
+ _combinedServerInstanceMap.putAll(primaryServerInstances);
+
+ // Add server instances from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ Map<String, ServerInstance> secondaryServerInstances = secondaryRoutingManager.getEnabledServerInstanceMap();
+ _combinedServerInstanceMap.putAll(secondaryServerInstances);
+ }
+
+ LOGGER.info("Updated combined server instances. Total: {}", _combinedServerInstanceMap.size());
+ }
+
+ @Override
+ public boolean routingExists(String tableNameWithType) {
+ // Check if routing exists in primary routing manager
+ if (_primaryRoutingManager.routingExists(tableNameWithType)) {
+ return true;
+ }
+
+ // Check if routing exists in any secondary routing manager
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ if (secondaryRoutingManager.routingExists(tableNameWithType)) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ @Override
+ public boolean isTableDisabled(String tableNameWithType) {
+ // Check if table is disabled in primary routing manager
+ if (_primaryRoutingManager.isTableDisabled(tableNameWithType)) {
+ return true;
+ }
+
+ // Check if table is disabled in any secondary routing manager
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ if (secondaryRoutingManager.isTableDisabled(tableNameWithType)) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public RoutingTable getRoutingTable(BrokerRequest brokerRequest, long requestId) {
+ String tableNameWithType = brokerRequest.getQuerySource().getTableName();
+ return getRoutingTable(brokerRequest, tableNameWithType, requestId);
+ }
+
+ @Nullable
+ @Override
+ public RoutingTable getRoutingTable(BrokerRequest brokerRequest, String tableNameWithType, long requestId) {
+ // Check cache first
+ String cacheKey = tableNameWithType + "_" + requestId;
+ RoutingTable cachedRoutingTable = _combinedRoutingTableCache.get(cacheKey);
+ if (cachedRoutingTable != null) {
+ return cachedRoutingTable;
+ }
+
+ // Get routing table from primary routing manager
+ RoutingTable primaryRoutingTable = _primaryRoutingManager.getRoutingTable(brokerRequest,
+ tableNameWithType, requestId);
+
+ // Combine with routing tables from secondary routing managers
+ RoutingTable combinedRoutingTable = combineRoutingTables(primaryRoutingTable, tableNameWithType,
+ brokerRequest, requestId);
+
+ // Cache the result
+ if (combinedRoutingTable != null) {
+ _combinedRoutingTableCache.put(cacheKey, combinedRoutingTable);
+ }
+
+ return combinedRoutingTable;
+ }
+
+ private RoutingTable combineRoutingTables(RoutingTable primaryRoutingTable, String tableNameWithType,
+ BrokerRequest brokerRequest, long requestId) {
+ if (primaryRoutingTable == null) {
+ // If primary routing table is null, try to get routing table from secondary managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ RoutingTable secondaryRoutingTable = secondaryRoutingManager.getRoutingTable(brokerRequest,
+ tableNameWithType, requestId);
+ if (secondaryRoutingTable != null) {
+ return secondaryRoutingTable;
+ }
+ }
+ return null;
+ }
+
+ // Start with primary routing table
+ Map<ServerInstance, SegmentsToQuery> combinedServerInstanceToSegmentsMap =
+ new HashMap<>(primaryRoutingTable.getServerInstanceToSegmentsMap());
+ List<String> combinedUnavailableSegments = new ArrayList<>(primaryRoutingTable.getUnavailableSegments());
+ int combinedNumPrunedSegments = primaryRoutingTable.getNumPrunedSegments();
+
+ // Add routing information from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ RoutingTable secondaryRoutingTable = secondaryRoutingManager.getRoutingTable(brokerRequest,
+ tableNameWithType, requestId);
+ if (secondaryRoutingTable != null) {
+ // Combine server instance to segments map
+ for (Map.Entry<ServerInstance, SegmentsToQuery> entry
+ : secondaryRoutingTable.getServerInstanceToSegmentsMap().entrySet()) {
+ ServerInstance serverInstance = entry.getKey();
+ SegmentsToQuery secondaryRouteInfo = entry.getValue();
+
+ SegmentsToQuery existingRouteInfo = combinedServerInstanceToSegmentsMap.get(serverInstance);
+ if (existingRouteInfo != null) {
+ // Merge segments
+ existingRouteInfo.getSegments().addAll(secondaryRouteInfo.getSegments());
+ existingRouteInfo.getOptionalSegments().addAll(secondaryRouteInfo.getOptionalSegments());
+ } else {
+ // Add new server instance
+ combinedServerInstanceToSegmentsMap.put(serverInstance, secondaryRouteInfo);
+ }
+ }
+
+ // Combine unavailable segments
+ combinedUnavailableSegments.addAll(secondaryRoutingTable.getUnavailableSegments());
+
+ // Add pruned segments count
+ combinedNumPrunedSegments += secondaryRoutingTable.getNumPrunedSegments();
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error combining routing table from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return new RoutingTable(combinedServerInstanceToSegmentsMap,
+ combinedUnavailableSegments, combinedNumPrunedSegments);
+ }
+
+ @Nullable
+ @Override
+ public TimeBoundaryInfo getTimeBoundaryInfo(String tableNameWithType) {
+ // Try to get time boundary info from primary routing manager first
+ TimeBoundaryInfo primaryTimeBoundaryInfo = _primaryRoutingManager.getTimeBoundaryInfo(tableNameWithType);
+ if (primaryTimeBoundaryInfo != null) {
+ return primaryTimeBoundaryInfo;
+ }
+
+ // Try to get time boundary info from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ TimeBoundaryInfo secondaryTimeBoundaryInfo = secondaryRoutingManager.getTimeBoundaryInfo(tableNameWithType);
+ if (secondaryTimeBoundaryInfo != null) {
+ return secondaryTimeBoundaryInfo;
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error getting time boundary info from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return null;
+ }
+
+ @Override
+ public Map<String, ServerInstance> getEnabledServerInstanceMap() {
+ updateCombinedServerInstances();
+ return _combinedServerInstanceMap;
+ }
+
+ @Override
+ public TablePartitionInfo getTablePartitionInfo(String tableNameWithType) {
+ // Try to get table partition info from primary routing manager first
+ TablePartitionInfo primaryTablePartitionInfo = _primaryRoutingManager.getTablePartitionInfo(tableNameWithType);
+ if (primaryTablePartitionInfo != null) {
+ return primaryTablePartitionInfo;
+ }
+
+ // Try to get table partition info from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ TablePartitionInfo secondaryTablePartitionInfo =
+ secondaryRoutingManager.getTablePartitionInfo(tableNameWithType);
+ if (secondaryTablePartitionInfo != null) {
+ return secondaryTablePartitionInfo;
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error getting table partition info from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return null;
+ }
+
+ @Override
+ public Set<String> getServingInstances(String tableNameWithType) {
+ // Get serving instances from primary routing manager
+ Set<String> primaryServingInstances = _primaryRoutingManager.getServingInstances(tableNameWithType);
+ if (primaryServingInstances == null) {
+ primaryServingInstances = Collections.emptySet();
+ }
+ // Combine with serving instances from secondary routing managers
+ Set<String> combinedServingInstances = new HashSet<>(primaryServingInstances);
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ Set<String> secondaryServingInstances = secondaryRoutingManager.getServingInstances(tableNameWithType);
+ if (secondaryServingInstances != null) {
+ combinedServingInstances.addAll(secondaryServingInstances);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error getting serving instances from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return combinedServingInstances.isEmpty() ? null : combinedServingInstances;
+ }
+
+ @Override
+ public List<String> getSegments(BrokerRequest brokerRequest) {
+ // Get segments from primary routing manager
+ List<String> primarySegments = _primaryRoutingManager.getSegments(brokerRequest);
+
+ // Combine with segments from secondary routing managers
+ List<String> combinedSegments = new ArrayList<>(primarySegments);
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ List<String> secondarySegments = secondaryRoutingManager.getSegments(brokerRequest);
+ combinedSegments.addAll(secondarySegments);
+ } catch (Exception e) {
+ LOGGER.error("Error getting segments from secondary routing manager", e);
+ }
+ }
+
+ return combinedSegments;
+ }
+
+ @Override
+ public TablePartitionReplicatedServersInfo getTablePartitionReplicatedServersInfo(String tableNameWithType) {
+ // Try to get table partition replicated servers info from primary routing manager first
+ TablePartitionReplicatedServersInfo primaryInfo =
+ _primaryRoutingManager.getTablePartitionReplicatedServersInfo(tableNameWithType);
+ if (primaryInfo != null) {
+ return primaryInfo;
+ }
+
+ // Try to get table partition replicated servers info from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ TablePartitionReplicatedServersInfo secondaryInfo =
+ secondaryRoutingManager.getTablePartitionReplicatedServersInfo(tableNameWithType);
+ if (secondaryInfo != null) {
+ return secondaryInfo;
+ }
+ } catch (Exception e) {
+ LOGGER.error(
+ "Error getting table partition replicated servers info from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return null;
+ }
+
+ public RoutingManager getPrimaryRoutingManager(Map<String, String> queryOptions) {
Review Comment:
this is confusing for me. Here RoutingManager is returning another routing manager, and I think this can be quite error prone long term.
Maybe the right abstraction here is a "RoutingManagerProvider" concept?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on PR #17145:
URL: https://github.com/apache/pinot/pull/17145#issuecomment-3994789091
Closing since it is already implemented.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2566706077
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
I thought we wanted to model table federation as logical table. If that is the case, physical table should never be federated itself
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2566737844
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
Yes. We will add logic to leverage the `FederatedRoutingManager` only when the table in question is a logical table, and ensure physical tables are never federated.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2535083033
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/FederatedRoutingManager.java:
##########
@@ -0,0 +1,343 @@
+/**
+ * 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.routing;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.request.BrokerRequest;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils;
+import org.apache.pinot.core.routing.RoutingManager;
+import org.apache.pinot.core.routing.RoutingTable;
+import org.apache.pinot.core.routing.SegmentsToQuery;
+import org.apache.pinot.core.routing.TablePartitionInfo;
+import org.apache.pinot.core.routing.TablePartitionReplicatedServersInfo;
+import org.apache.pinot.core.routing.timeboundary.TimeBoundaryInfo;
+import org.apache.pinot.core.transport.ServerInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * FederatedRoutingManager manages routing across multiple ZooKeeper clusters.
+ * It maintains a primary BrokerRoutingManager and multiple secondary routing managers,
+ * each connected to their own ZooKeeper cluster.
+ */
+public class FederatedRoutingManager implements RoutingManager {
+ private static final Logger LOGGER = LoggerFactory.getLogger(FederatedRoutingManager.class);
+
+ // Primary routing manager (connected to primary ZooKeeper cluster)
+ private final BrokerRoutingManager _primaryRoutingManager;
+
+ // Secondary routing managers (connected to secondary ZooKeeper clusters)
+ private final List<BrokerRoutingManager> _secondaryRoutingManagers;
+
+ // Combined routing information
+ private final Map<String, ServerInstance> _combinedServerInstanceMap = new ConcurrentHashMap<>();
+ private final Map<String, RoutingTable> _combinedRoutingTableCache = new ConcurrentHashMap<>();
+
+ public FederatedRoutingManager(BrokerRoutingManager primaryRoutingManager,
+ List<BrokerRoutingManager> secondaryRoutingManagers) {
+ _primaryRoutingManager = primaryRoutingManager;
+ _secondaryRoutingManagers = secondaryRoutingManagers;
+ }
+
+ private void updateCombinedServerInstances() {
+ _combinedServerInstanceMap.clear();
+
+ // Add server instances from primary routing manager
+ Map<String, ServerInstance> primaryServerInstances = _primaryRoutingManager.getEnabledServerInstanceMap();
+ _combinedServerInstanceMap.putAll(primaryServerInstances);
+
+ // Add server instances from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ Map<String, ServerInstance> secondaryServerInstances = secondaryRoutingManager.getEnabledServerInstanceMap();
+ _combinedServerInstanceMap.putAll(secondaryServerInstances);
+ }
+
+ LOGGER.info("Updated combined server instances. Total: {}", _combinedServerInstanceMap.size());
+ }
+
+ @Override
+ public boolean routingExists(String tableNameWithType) {
+ // Check if routing exists in primary routing manager
+ if (_primaryRoutingManager.routingExists(tableNameWithType)) {
+ return true;
+ }
+
+ // Check if routing exists in any secondary routing manager
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ if (secondaryRoutingManager.routingExists(tableNameWithType)) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ @Override
+ public boolean isTableDisabled(String tableNameWithType) {
+ // Check if table is disabled in primary routing manager
+ if (_primaryRoutingManager.isTableDisabled(tableNameWithType)) {
+ return true;
+ }
+
+ // Check if table is disabled in any secondary routing manager
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ if (secondaryRoutingManager.isTableDisabled(tableNameWithType)) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public RoutingTable getRoutingTable(BrokerRequest brokerRequest, long requestId) {
+ String tableNameWithType = brokerRequest.getQuerySource().getTableName();
+ return getRoutingTable(brokerRequest, tableNameWithType, requestId);
+ }
+
+ @Nullable
+ @Override
+ public RoutingTable getRoutingTable(BrokerRequest brokerRequest, String tableNameWithType, long requestId) {
+ // Check cache first
+ String cacheKey = tableNameWithType + "_" + requestId;
+ RoutingTable cachedRoutingTable = _combinedRoutingTableCache.get(cacheKey);
+ if (cachedRoutingTable != null) {
+ return cachedRoutingTable;
+ }
+
+ // Get routing table from primary routing manager
+ RoutingTable primaryRoutingTable = _primaryRoutingManager.getRoutingTable(brokerRequest,
+ tableNameWithType, requestId);
+
+ // Combine with routing tables from secondary routing managers
+ RoutingTable combinedRoutingTable = combineRoutingTables(primaryRoutingTable, tableNameWithType,
+ brokerRequest, requestId);
+
+ // Cache the result
+ if (combinedRoutingTable != null) {
+ _combinedRoutingTableCache.put(cacheKey, combinedRoutingTable);
+ }
+
+ return combinedRoutingTable;
+ }
+
+ private RoutingTable combineRoutingTables(RoutingTable primaryRoutingTable, String tableNameWithType,
+ BrokerRequest brokerRequest, long requestId) {
+ if (primaryRoutingTable == null) {
+ // If primary routing table is null, try to get routing table from secondary managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ RoutingTable secondaryRoutingTable = secondaryRoutingManager.getRoutingTable(brokerRequest,
+ tableNameWithType, requestId);
+ if (secondaryRoutingTable != null) {
+ return secondaryRoutingTable;
+ }
+ }
+ return null;
+ }
+
+ // Start with primary routing table
+ Map<ServerInstance, SegmentsToQuery> combinedServerInstanceToSegmentsMap =
+ new HashMap<>(primaryRoutingTable.getServerInstanceToSegmentsMap());
+ List<String> combinedUnavailableSegments = new ArrayList<>(primaryRoutingTable.getUnavailableSegments());
+ int combinedNumPrunedSegments = primaryRoutingTable.getNumPrunedSegments();
+
+ // Add routing information from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ RoutingTable secondaryRoutingTable = secondaryRoutingManager.getRoutingTable(brokerRequest,
+ tableNameWithType, requestId);
+ if (secondaryRoutingTable != null) {
+ // Combine server instance to segments map
+ for (Map.Entry<ServerInstance, SegmentsToQuery> entry
+ : secondaryRoutingTable.getServerInstanceToSegmentsMap().entrySet()) {
+ ServerInstance serverInstance = entry.getKey();
+ SegmentsToQuery secondaryRouteInfo = entry.getValue();
+
+ SegmentsToQuery existingRouteInfo = combinedServerInstanceToSegmentsMap.get(serverInstance);
+ if (existingRouteInfo != null) {
+ // Merge segments
+ existingRouteInfo.getSegments().addAll(secondaryRouteInfo.getSegments());
+ existingRouteInfo.getOptionalSegments().addAll(secondaryRouteInfo.getOptionalSegments());
+ } else {
+ // Add new server instance
+ combinedServerInstanceToSegmentsMap.put(serverInstance, secondaryRouteInfo);
+ }
+ }
+
+ // Combine unavailable segments
+ combinedUnavailableSegments.addAll(secondaryRoutingTable.getUnavailableSegments());
+
+ // Add pruned segments count
+ combinedNumPrunedSegments += secondaryRoutingTable.getNumPrunedSegments();
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error combining routing table from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return new RoutingTable(combinedServerInstanceToSegmentsMap,
+ combinedUnavailableSegments, combinedNumPrunedSegments);
+ }
+
+ @Nullable
+ @Override
+ public TimeBoundaryInfo getTimeBoundaryInfo(String tableNameWithType) {
+ // Try to get time boundary info from primary routing manager first
+ TimeBoundaryInfo primaryTimeBoundaryInfo = _primaryRoutingManager.getTimeBoundaryInfo(tableNameWithType);
+ if (primaryTimeBoundaryInfo != null) {
+ return primaryTimeBoundaryInfo;
+ }
+
+ // Try to get time boundary info from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ TimeBoundaryInfo secondaryTimeBoundaryInfo = secondaryRoutingManager.getTimeBoundaryInfo(tableNameWithType);
+ if (secondaryTimeBoundaryInfo != null) {
+ return secondaryTimeBoundaryInfo;
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error getting time boundary info from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return null;
+ }
+
+ @Override
+ public Map<String, ServerInstance> getEnabledServerInstanceMap() {
+ updateCombinedServerInstances();
+ return _combinedServerInstanceMap;
+ }
+
+ @Override
+ public TablePartitionInfo getTablePartitionInfo(String tableNameWithType) {
+ // Try to get table partition info from primary routing manager first
+ TablePartitionInfo primaryTablePartitionInfo = _primaryRoutingManager.getTablePartitionInfo(tableNameWithType);
+ if (primaryTablePartitionInfo != null) {
+ return primaryTablePartitionInfo;
+ }
+
+ // Try to get table partition info from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ TablePartitionInfo secondaryTablePartitionInfo =
+ secondaryRoutingManager.getTablePartitionInfo(tableNameWithType);
+ if (secondaryTablePartitionInfo != null) {
+ return secondaryTablePartitionInfo;
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error getting table partition info from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return null;
+ }
+
+ @Override
+ public Set<String> getServingInstances(String tableNameWithType) {
+ // Get serving instances from primary routing manager
+ Set<String> primaryServingInstances = _primaryRoutingManager.getServingInstances(tableNameWithType);
+ if (primaryServingInstances == null) {
+ primaryServingInstances = Collections.emptySet();
+ }
+ // Combine with serving instances from secondary routing managers
+ Set<String> combinedServingInstances = new HashSet<>(primaryServingInstances);
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ Set<String> secondaryServingInstances = secondaryRoutingManager.getServingInstances(tableNameWithType);
+ if (secondaryServingInstances != null) {
+ combinedServingInstances.addAll(secondaryServingInstances);
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error getting serving instances from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return combinedServingInstances.isEmpty() ? null : combinedServingInstances;
+ }
+
+ @Override
+ public List<String> getSegments(BrokerRequest brokerRequest) {
+ // Get segments from primary routing manager
+ List<String> primarySegments = _primaryRoutingManager.getSegments(brokerRequest);
+
+ // Combine with segments from secondary routing managers
+ List<String> combinedSegments = new ArrayList<>(primarySegments);
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ List<String> secondarySegments = secondaryRoutingManager.getSegments(brokerRequest);
+ combinedSegments.addAll(secondarySegments);
+ } catch (Exception e) {
+ LOGGER.error("Error getting segments from secondary routing manager", e);
+ }
+ }
+
+ return combinedSegments;
+ }
+
+ @Override
+ public TablePartitionReplicatedServersInfo getTablePartitionReplicatedServersInfo(String tableNameWithType) {
+ // Try to get table partition replicated servers info from primary routing manager first
+ TablePartitionReplicatedServersInfo primaryInfo =
+ _primaryRoutingManager.getTablePartitionReplicatedServersInfo(tableNameWithType);
+ if (primaryInfo != null) {
+ return primaryInfo;
+ }
+
+ // Try to get table partition replicated servers info from secondary routing managers
+ for (BrokerRoutingManager secondaryRoutingManager : _secondaryRoutingManagers) {
+ try {
+ TablePartitionReplicatedServersInfo secondaryInfo =
+ secondaryRoutingManager.getTablePartitionReplicatedServersInfo(tableNameWithType);
+ if (secondaryInfo != null) {
+ return secondaryInfo;
+ }
+ } catch (Exception e) {
+ LOGGER.error(
+ "Error getting table partition replicated servers info from secondary routing manager for table {}",
+ tableNameWithType, e);
+ }
+ }
+
+ return null;
+ }
+
+ public RoutingManager getPrimaryRoutingManager(Map<String, String> queryOptions) {
Review Comment:
Good point. Instead of passing a RoutingManager for the RequestHandler methods, I could instead pass a RoutingManagerProvider and determine which routingManager to use at runtime.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] [DRAFT] Implementing federation for logical tables in Pinot [pinot]
Posted by "shauryachats (via GitHub)" <gi...@apache.org>.
shauryachats commented on code in PR #17145:
URL: https://github.com/apache/pinot/pull/17145#discussion_r2578574040
##########
pinot-spi/src/main/java/org/apache/pinot/spi/data/PhysicalTableConfig.java:
##########
@@ -26,4 +27,40 @@
* This is empty by design and more docs would be added as features are added.
*/
public class PhysicalTableConfig extends BaseJsonConfig {
+ @JsonProperty("isFederated")
+ boolean _isFederated;
Review Comment:
Yes we do, because since the controller does not know which tables are federated or not, the controller can skip physical table validations only for physical tables which are federated, when the logical table is created or updated.
Furthermore, there might be a case where we want to selectively federate only some physical tables in a logical table and for other physical tables which are explicitly set to non-federated (by `isFederated = false`), we will not consider that table for federation even if the same physical table name is also present cross-cluster.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org