timoninmaxim commented on code in PR #11317: URL: https://github.com/apache/ignite/pull/11317#discussion_r1622336748
########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java: ########## @@ -1499,6 +1502,79 @@ protected GridCloseableIterator scanQueryLocal(final GridCacheQueryAdapter qry, } } + /** + * Process local index query. + * + * @param qry Query. + * @return GridCloseableIterator. + */ + @SuppressWarnings({"unchecked"}) + public GridCloseableIterator indexQueryLocal(final GridCacheQueryAdapter qry) throws IgniteCheckedException { + if (!enterBusy()) + throw new IllegalStateException("Failed to process query request (grid is stopping)."); + + try { + assert qry.type() == INDEX : "Wrong processing of query: " + qry.type(); + + cctx.checkSecurity(SecurityPermission.CACHE_READ); + + GridDhtCacheAdapter<?, ?> cacheAdapter = cctx.isNear() ? cctx.near().dht() : cctx.dht(); + + Set<Integer> lostParts = cacheAdapter.topology().lostPartitions(); + + Integer part = qry.partition(); + + if (!lostParts.isEmpty()) { + if (part == null || lostParts.contains(part)) { + throw new CacheException(new CacheInvalidStateException("Failed to execute query because cache partition " + + "has been lost [cacheName=" + cctx.name() + + ", part=" + (part == null ? lostParts.iterator().next() : part) + ']')); + } + } + + if (qry.nodes().isEmpty()) + throw new IgniteException(new ClusterGroupEmptyException()); + + if (cctx.deploymentEnabled()) + cctx.deploy().registerClasses(qry.scanFilter()); Review Comment: This code doesn't covered by tests. For what purpose do you need this code? ########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java: ########## @@ -1499,6 +1502,79 @@ protected GridCloseableIterator scanQueryLocal(final GridCacheQueryAdapter qry, } } + /** + * Process local index query. + * + * @param qry Query. + * @return GridCloseableIterator. + */ + @SuppressWarnings({"unchecked"}) + public GridCloseableIterator indexQueryLocal(final GridCacheQueryAdapter qry) throws IgniteCheckedException { + if (!enterBusy()) + throw new IllegalStateException("Failed to process query request (grid is stopping)."); + + try { + assert qry.type() == INDEX : "Wrong processing of query: " + qry.type(); + + cctx.checkSecurity(SecurityPermission.CACHE_READ); + + GridDhtCacheAdapter<?, ?> cacheAdapter = cctx.isNear() ? cctx.near().dht() : cctx.dht(); + + Set<Integer> lostParts = cacheAdapter.topology().lostPartitions(); + + Integer part = qry.partition(); + + if (!lostParts.isEmpty()) { + if (part == null || lostParts.contains(part)) { + throw new CacheException(new CacheInvalidStateException("Failed to execute query because cache partition " + + "has been lost [cacheName=" + cctx.name() + + ", part=" + (part == null ? lostParts.iterator().next() : part) + ']')); + } + } + + if (qry.nodes().isEmpty()) Review Comment: This code is already starts on local node, why do we need this check? ########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java: ########## @@ -1499,6 +1502,79 @@ protected GridCloseableIterator scanQueryLocal(final GridCacheQueryAdapter qry, } } + /** + * Process local index query. + * + * @param qry Query. + * @return GridCloseableIterator. + */ + @SuppressWarnings({"unchecked"}) + public GridCloseableIterator indexQueryLocal(final GridCacheQueryAdapter qry) throws IgniteCheckedException { + if (!enterBusy()) + throw new IllegalStateException("Failed to process query request (grid is stopping)."); + + try { + assert qry.type() == INDEX : "Wrong processing of query: " + qry.type(); + + cctx.checkSecurity(SecurityPermission.CACHE_READ); + + GridDhtCacheAdapter<?, ?> cacheAdapter = cctx.isNear() ? cctx.near().dht() : cctx.dht(); + + Set<Integer> lostParts = cacheAdapter.topology().lostPartitions(); + + Integer part = qry.partition(); + + if (!lostParts.isEmpty()) { + if (part == null || lostParts.contains(part)) { + throw new CacheException(new CacheInvalidStateException("Failed to execute query because cache partition " + + "has been lost [cacheName=" + cctx.name() + + ", part=" + (part == null ? lostParts.iterator().next() : part) + ']')); + } + } + + if (qry.nodes().isEmpty()) + throw new IgniteException(new ClusterGroupEmptyException()); + + if (cctx.deploymentEnabled()) + cctx.deploy().registerClasses(qry.scanFilter()); + + if (log.isDebugEnabled()) + log.debug("Running local index query: " + qry); + + if (cctx.events().isRecordable(EVT_CACHE_QUERY_EXECUTED)) { + cctx.gridEvents().record(new CacheQueryExecutedEvent<>( + cctx.localNode(), + "Index query executed.", + EVT_CACHE_QUERY_EXECUTED, + CacheQueryType.INDEX.name(), + cctx.name(), + qry.queryClassName(), + null, + qry.scanFilter(), + null, + null, + securitySubjectId(cctx), + cctx.kernalContext().task().resolveTaskName(qry.taskHash()))); + } + + int[] parts = null; + + if (part != null) + parts = new int[] {part}; + + IndexQueryResult<K, V> idxQryRes = qryProc.queryIndex(cacheName, qry.queryClassName(), qry.idxQryDesc(), + qry.scanFilter(), filter(qry, parts, parts != null), qry.keepBinary(), qry.taskHash()); + + return new IndexQueryIterator(idxQryRes.iter()); + } + catch (Exception e) { + throw e; Review Comment: Useless catch ########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxyImpl.java: ########## @@ -557,6 +558,17 @@ else if (query instanceof IndexQuery) { if (q.getLimit() > 0) qry.limit(q.getLimit()); + if (query.isLocal()) { + final GridCloseableIterator iter = ctx.kernalContext().query().executeQuery(GridCacheQueryType.INDEX, + cacheName, ctx, new IgniteOutClosureX<GridCloseableIterator>() { + @Override public GridCloseableIterator applyx() throws IgniteCheckedException { + return ctx.queries().indexQueryLocal((GridCacheQueryAdapter)qry); + } + }, true); + + return new QueryCursorImpl<>(iter); Review Comment: `QueryCursorImpl<>` -> `QueryCursorImpl` there is no any class to inject to the generic. ########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java: ########## @@ -1499,6 +1502,79 @@ protected GridCloseableIterator scanQueryLocal(final GridCacheQueryAdapter qry, } } + /** + * Process local index query. + * + * @param qry Query. + * @return GridCloseableIterator. + */ + @SuppressWarnings({"unchecked"}) + public GridCloseableIterator indexQueryLocal(final GridCacheQueryAdapter qry) throws IgniteCheckedException { + if (!enterBusy()) + throw new IllegalStateException("Failed to process query request (grid is stopping)."); + + try { + assert qry.type() == INDEX : "Wrong processing of query: " + qry.type(); + + cctx.checkSecurity(SecurityPermission.CACHE_READ); + + GridDhtCacheAdapter<?, ?> cacheAdapter = cctx.isNear() ? cctx.near().dht() : cctx.dht(); + + Set<Integer> lostParts = cacheAdapter.topology().lostPartitions(); + + Integer part = qry.partition(); + + if (!lostParts.isEmpty()) { Review Comment: This code isn't covered with tests. Let's remove it and add test and code in a separate ticket ########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java: ########## @@ -3427,4 +3503,56 @@ IgniteBiPredicate<K, V> scanFilter() { return scanFilter; } } + + /** */ + public static final class IndexQueryIterator<K, V> extends GridCloseableIteratorAdapter<Object> { Review Comment: private -- 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: notifications-unsubscr...@ignite.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org