korlov42 commented on code in PR #2848:
URL: https://github.com/apache/ignite-3/pull/2848#discussion_r1402984280
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java:
##########
@@ -317,6 +341,69 @@ public CompletableFuture<ExecutionTarget>
forSystemView(ExecutionTargetFactory f
services.forEach(LifecycleAware::start);
}
+ /** Get primary replicas. */
+ private CompletableFuture<List<PrimaryReplica>> primaryReplicas(int
tableId) {
+ int catalogVersion = catalogManager.latestCatalogVersion();
+
+ Catalog catalog = catalogManager.catalog(catalogVersion);
+
+ CatalogTableDescriptor tblDesc =
Objects.requireNonNull(catalog.table(tableId), "table");
+
+ CatalogZoneDescriptor zoneDesc =
Objects.requireNonNull(catalog.zone(tblDesc.zoneId()), "zone");
+
+ int partitions = zoneDesc.partitions();
+
+ List<CompletableFuture<PrimaryReplica>> result = new
ArrayList<>(partitions);
+
+ HybridTimestamp clockNow = clock.now();
+
+ // no need to wait all partitions after pruning was implemented.
+ for (int partId = 0; partId < partitions; ++partId) {
+ int partitionId = partId;
+ ReplicationGroupId partGroupId = new TablePartitionId(tableId,
partitionId);
+
+ CompletableFuture<ReplicaMeta> f =
placementDriver.awaitPrimaryReplica(
+ partGroupId,
+ clockNow,
+ AWAIT_PRIMARY_REPLICA_TIMEOUT,
+ SECONDS
+ );
+
+ result.add(f.handle((primaryReplica, e) -> {
+ if (e != null) {
+ LOG.error("Failed to retrieve primary replica for
partition {}", e, partitionId);
+
+ throw withCause(IgniteInternalException::new,
REPLICA_UNAVAILABLE_ERR, "Failed to get the primary replica"
+ + " [tablePartitionId=" + partGroupId + ']', e);
+ }
+
+ return primaryReplica;
+ })
+ .thenApply(primaryReplica -> {
+ String holder = primaryReplica.getLeaseholder();
+
+ assert holder != null : "Unable to map query, nothing
holds the lease";
+
+ ClusterNode node =
clusterSrvc.topologyService().getByConsistentId(holder);
+
+ if (node == null) {
+ // additional recovery logic is need to be present
around here.
+ throw new IgniteInternalException(Sql.MAPPING_ERR,
"Unable to map query, node is lost or offline");
+ }
+
+ return new PrimaryReplica(node,
primaryReplica.getStartTime().longValue());
Review Comment:
I mean why do we need to return PrimaryReplica on this step? On the very
next one we will convert it NodeWithTerm, so why don't return NodeWithTerm here
and reduce pipeline by one step?
--
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]
