zstan commented on code in PR #2848:
URL: https://github.com/apache/ignite-3/pull/2848#discussion_r1399267102
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java:
##########
@@ -317,6 +334,60 @@ 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 partitionId = 0; partitionId < partitions; ++partitionId) {
+ ReplicationGroupId partGroupId = new TablePartitionId(tableId,
partitionId);
+
+ CompletableFuture<ReplicaMeta> f =
placementDriver.awaitPrimaryReplica(
+ partGroupId,
+ clockNow,
+ AWAIT_PRIMARY_REPLICA_TIMEOUT,
+ SECONDS
+ );
+
+ result.add(f.thenApply(primaryReplica -> {
+ String holder = primaryReplica.getLeaseholder();
+
+ if (holder == null) {
+ // additional recovery logic is need to be present around
here.
+ throw new IgniteInternalException(Sql.MAPPING_ERR, "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());
+ }));
+ }
+
+ CompletableFuture<Void> all =
CompletableFuture.allOf(result.toArray(new CompletableFuture[0]));
Review Comment:
> why don't you just await primary replica here
cause **assignments** are also used for colocation detection logic (i.e. sql
through tables with different affinity distributions) thus it just reused here.
--
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]
