com.hazelcast.map.impl.query

Class MapQueryEngineImpl

java.lang.Object

com.hazelcast.map.impl.query.MapQueryEngineImpl

All Implemented Interfaces:

MapQueryEngine

public class MapQueryEngineImpl
extends Object
implements MapQueryEngine

The MapQueryEngine implementation.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected class	MapQueryEngineImpl.QueryPartitionCallable

Field Summary

Fields

Modifier and Type	Field and Description
protected ClusterService	clusterService
protected ManagedExecutorService	executor
protected LocalMapStatsProvider	localMapStatsProvider
protected ILogger	logger
protected MapServiceContext	mapServiceContext
protected NodeEngine	nodeEngine
protected OperationService	operationService
protected boolean	parallelEvaluation
protected IPartitionService	partitionService
protected static long	QUERY_EXECUTION_TIMEOUT_MINUTES
protected QueryOptimizer	queryOptimizer
protected QueryResultSizeLimiter	queryResultSizeLimiter
protected InternalSerializationService	serializationService

Constructor Summary

Constructors

Constructor and Description
MapQueryEngineImpl(MapServiceContext mapServiceContext, QueryOptimizer optimizer)

Method Summary

Methods

Modifier and Type	Method and Description
protected void	addResultsOfPagingPredicate(List<Future<QueryResult>> futures, Collection result, Collection<Integer> partitionIds) Adds results of paging predicates to result set and removes queried partition ids.
protected void	addResultsOfPredicate(List<Future<QueryResult>> futures, QueryResult result, Collection<Integer> partitionIds) Adds results of non-paging predicates to result set and removes queried partition ids.
protected void	checkNotPagingPredicate(Predicate predicate)
protected Set<Integer>	createSetWithPopulatedPartitionIds(int partitionCount)
protected Set<Integer>	getAllPartitionIds()
protected List<Integer>	getLocalPartitionIds()
protected long	getNow()
protected static Collection<Collection<QueryableEntry>>	getResult(List<Future<Collection<QueryableEntry>>> lsFutures)
protected boolean	hasOwnerMigrationsInFlight() Check whether migrations of owner partition are currently executed.
protected boolean	hasPartitionStateVersionChanged(int initialPartitionStateVersion) Check whether partition state version has changed since initialPartitionStateVersion.
protected boolean	hasPartitionVersion(int expectedVersion, Predicate predicate)
QueryResult	invokeQueryAllPartitions(String mapName, Predicate predicate, IterationType iterationType) Queries all partitions.
QueryResult	invokeQueryLocalPartitions(String mapName, Predicate predicate, IterationType iterationType) Query all local partitions.
protected boolean	isResultSafe(int initialPartitionStateVersion) Check whether results obtained since partition state version was at initialPartitionStateVersion are safe to be returned to the caller.
protected QueryResult	newQueryResult(int numberOfPartitions, IterationType iterationType) Creates a QueryResult with configured result limit (according to the number of partitions) if feature is enabled.
Set	queryAllPartitionsWithPagingPredicate(String mapName, PagingPredicate predicate, IterationType iterationType) Queries all partitions with a paging predicate.
QueryResult	queryLocalPartition(String mapName, Predicate predicate, int partitionId, IterationType iterationType) Executes a query on a specific local partition.
QueryResult	queryLocalPartitions(String mapName, Predicate predicate, IterationType iterationType) Executes a query on all the local partitions.
Set	queryLocalPartitionsWithPagingPredicate(String mapName, PagingPredicate predicate, IterationType iterationType) Query all local partitions with a paging predicate.
protected Future<QueryResult>	queryOnLocalMember(String mapName, Predicate predicate, IterationType iterationType)
protected List<Future<QueryResult>>	queryOnMembers(String mapName, Predicate predicate, IterationType iterationType)
protected QueryResult	queryParallel(String name, Predicate predicate, Collection<Integer> partitions, IterationType iterationType)
protected QueryResult	queryParallelForPaging(String name, PagingPredicate predicate, Collection<Integer> partitions, IterationType iterationType)
protected List<Future<QueryResult>>	queryPartitions(String mapName, Predicate predicate, Collection<Integer> partitionIds, IterationType iterationType)
protected QueryResult	querySafelyUsingFullTableScan(String name, Predicate predicate, Collection<Integer> partitions, IterationType iterationType, int initialPartitionStateVersion) Wraps queryUsingFullTableScan(String, Predicate, Collection, IterationType) to avoid returning potentially flawed results.
protected QueryResult	querySequential(String name, Predicate predicate, Collection<Integer> partitions, IterationType iterationType)
protected Collection<QueryableEntry>	queryTheLocalPartition(String mapName, Predicate predicate, int partitionId)
protected QueryResult	queryUsingFullTableScan(String name, Predicate predicate, Collection<Integer> partitions, IterationType iterationType)
protected Object	toObject(Object obj)
protected QueryResult	tryQueryUsingIndexes(Predicate predicate, Collection<Integer> partitions, MapContainer mapContainer, IterationType iterationType, int initialPartitionStateVersion)
protected void	updateStatistics(MapContainer mapContainer)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

QUERY_EXECUTION_TIMEOUT_MINUTES

protected static final long QUERY_EXECUTION_TIMEOUT_MINUTES

See Also:

Constant Field Values

mapServiceContext

protected final MapServiceContext mapServiceContext

nodeEngine

protected final NodeEngine nodeEngine

logger

protected final ILogger logger

queryResultSizeLimiter

protected final QueryResultSizeLimiter queryResultSizeLimiter

serializationService

protected final InternalSerializationService serializationService

partitionService

protected final IPartitionService partitionService

queryOptimizer

protected final QueryOptimizer queryOptimizer

operationService

protected final OperationService operationService

clusterService

protected final ClusterService clusterService

localMapStatsProvider

protected final LocalMapStatsProvider localMapStatsProvider

parallelEvaluation

protected final boolean parallelEvaluation

executor

protected final ManagedExecutorService executor

Constructor Detail

MapQueryEngineImpl

public MapQueryEngineImpl(MapServiceContext mapServiceContext,
                  QueryOptimizer optimizer)

Method Detail

queryLocalPartitions

public QueryResult queryLocalPartitions(String mapName,
                               Predicate predicate,
                               IterationType iterationType)
                                 throws ExecutionException,
                                        InterruptedException

Description copied from interface: MapQueryEngine

Executes a query on all the local partitions.

- Uses Indexes - Accepts PagingPredicate - Query executed in the calling thread - predicate evaluation will be parallelized if QUERY_PREDICATE_PARALLEL_EVALUATION is enabled or a PagingPredicate is used. - may return empty QueryResult (with a properly initialized result limit size) if query is executed during migrations and migration conditions do not allow for safe retrieval of results (ie. there are migrations in flight or partition state version is different at the end of query vs the version observed at the beginning of query). In this case, partitionIds are not set on the QueryResult, so callers will ignore these results and fallback to QueryPartitionOperations.

Specified by:

queryLocalPartitions in interface MapQueryEngine

Parameters:

mapName - the name of the map

predicate - the predicate

Returns:

the QueryResult

Throws:

ExecutionException

InterruptedException

tryQueryUsingIndexes

protected QueryResult tryQueryUsingIndexes(Predicate predicate,
                               Collection<Integer> partitions,
                               MapContainer mapContainer,
                               IterationType iterationType,
                               int initialPartitionStateVersion)

updateStatistics

protected void updateStatistics(MapContainer mapContainer)

querySafelyUsingFullTableScan

protected QueryResult querySafelyUsingFullTableScan(String name,
                                        Predicate predicate,
                                        Collection<Integer> partitions,
                                        IterationType iterationType,
                                        int initialPartitionStateVersion)
                                             throws InterruptedException,
                                                    ExecutionException

Wraps queryUsingFullTableScan(String, Predicate, Collection, IterationType) to avoid returning potentially flawed results.

• if owner partition migrations are executing before running the query, then return immediately null, as results cannot be reliably obtained
• execute the query using full table scan
• if after the query has been executed migrations are in-flight or partition state version has changed, then results are not considered reliable

Parameters:

name -

predicate -

partitions -

iterationType -

initialPartitionStateVersion -

Returns:

query results or null if results are considered potentially flawed.

Throws:

InterruptedException

ExecutionException

queryUsingFullTableScan

protected QueryResult queryUsingFullTableScan(String name,
                                  Predicate predicate,
                                  Collection<Integer> partitions,
                                  IterationType iterationType)
                                       throws InterruptedException,
                                              ExecutionException

Throws:

InterruptedException

ExecutionException

querySequential

protected QueryResult querySequential(String name,
                          Predicate predicate,
                          Collection<Integer> partitions,
                          IterationType iterationType)

queryParallel

protected QueryResult queryParallel(String name,
                        Predicate predicate,
                        Collection<Integer> partitions,
                        IterationType iterationType)
                             throws InterruptedException,
                                    ExecutionException

Throws:

InterruptedException

ExecutionException

queryParallelForPaging

protected QueryResult queryParallelForPaging(String name,
                                 PagingPredicate predicate,
                                 Collection<Integer> partitions,
                                 IterationType iterationType)
                                      throws InterruptedException,
                                             ExecutionException

Throws:

InterruptedException

ExecutionException

getResult

protected static Collection<Collection<QueryableEntry>> getResult(List<Future<Collection<QueryableEntry>>> lsFutures)

hasPartitionVersion

protected boolean hasPartitionVersion(int expectedVersion,
                          Predicate predicate)

queryTheLocalPartition

protected Collection<QueryableEntry> queryTheLocalPartition(String mapName,
                                                Predicate predicate,
                                                int partitionId)

queryLocalPartition

public QueryResult queryLocalPartition(String mapName,
                              Predicate predicate,
                              int partitionId,
                              IterationType iterationType)

Description copied from interface: MapQueryEngine

Executes a query on a specific local partition.

- Does NOT use Indexes - Accepts PagingPredicate - Sequential full table scan - Query executed in the calling thread

todo: what happens when the partition is not local?

Specified by:

queryLocalPartition in interface MapQueryEngine

Parameters:

mapName - map name.

predicate - any predicate.

partitionId - partition id.

Returns:

result of query

invokeQueryLocalPartitions

public QueryResult invokeQueryLocalPartitions(String mapName,
                                     Predicate predicate,
                                     IterationType iterationType)

Description copied from interface: MapQueryEngine

Query all local partitions.

- Does NOT accept PagingPredicate - Query executed in an Operation on this member (NOT in the calling thread) - Calls MapQueryEngine.queryLocalPartitions(String, Predicate, IterationType) in an operation

Specified by:

invokeQueryLocalPartitions in interface MapQueryEngine

Parameters:

mapName - map name.

predicate - except paging predicate.

iterationType - the IterationType

queryLocalPartitionsWithPagingPredicate

public Set queryLocalPartitionsWithPagingPredicate(String mapName,
                                          PagingPredicate predicate,
                                          IterationType iterationType)

Description copied from interface: MapQueryEngine

Query all local partitions with a paging predicate.

- Query executed in an Operation on this member (or other members if some partitions are not local)

TODO: it would be better to have a single queryLocal... method and let the implementation figure out how to deal with a regular predicate and a paging predicate. No need to have that in the interface. The problem is that currently the signatures don't match up. This implementation detail should not be exposed through the interface.

Specified by:

queryLocalPartitionsWithPagingPredicate in interface MapQueryEngine

Parameters:

mapName - map name.

predicate - to queryOnMembers.

iterationType - type of IterationType

Returns:

SortedQueryResultSet

queryAllPartitionsWithPagingPredicate

public Set queryAllPartitionsWithPagingPredicate(String mapName,
                                        PagingPredicate predicate,
                                        IterationType iterationType)

Description copied from interface: MapQueryEngine

Queries all partitions with a paging predicate.

- Query executed in an Operation on each member (NOT in the calling thread)

TODO: it would be better to have single queryAll method and let the implementation figure out how to deal with a paging predicate. See comment in MapQueryEngine.queryLocalPartitionsWithPagingPredicate(java.lang.String, com.hazelcast.query.PagingPredicate, com.hazelcast.util.IterationType)

Specified by:

queryAllPartitionsWithPagingPredicate in interface MapQueryEngine

Parameters:

mapName - map name.

predicate - to queryOnMembers.

iterationType - type of IterationType

Returns:

SortedQueryResultSet

invokeQueryAllPartitions

public QueryResult invokeQueryAllPartitions(String mapName,
                                   Predicate predicate,
                                   IterationType iterationType)

Description copied from interface: MapQueryEngine

Queries all partitions. Paging predicates are not allowed. - Does NOT accept PagingPredicate - Query executed in an Operation on each member (NOT in the calling thread) - Calls MapQueryEngine.queryLocalPartitions(String, Predicate, IterationType) in an operation

Specified by:

invokeQueryAllPartitions in interface MapQueryEngine

Parameters:

mapName - map name.

predicate - except paging predicate.

iterationType - the IterationType

newQueryResult

protected QueryResult newQueryResult(int numberOfPartitions,
                         IterationType iterationType)

Creates a QueryResult with configured result limit (according to the number of partitions) if feature is enabled.

Parameters:

numberOfPartitions - number of partitions to calculate result limit

Returns:

QueryResult

checkNotPagingPredicate

protected void checkNotPagingPredicate(Predicate predicate)

queryOnLocalMember

protected Future<QueryResult> queryOnLocalMember(String mapName,
                                     Predicate predicate,
                                     IterationType iterationType)

queryOnMembers

protected List<Future<QueryResult>> queryOnMembers(String mapName,
                                       Predicate predicate,
                                       IterationType iterationType)

queryPartitions

protected List<Future<QueryResult>> queryPartitions(String mapName,
                                        Predicate predicate,
                                        Collection<Integer> partitionIds,
                                        IterationType iterationType)

addResultsOfPagingPredicate

protected void addResultsOfPagingPredicate(List<Future<QueryResult>> futures,
                               Collection result,
                               Collection<Integer> partitionIds)
                                    throws ExecutionException,
                                           InterruptedException

Adds results of paging predicates to result set and removes queried partition ids.

Throws:

ExecutionException

InterruptedException

addResultsOfPredicate

protected void addResultsOfPredicate(List<Future<QueryResult>> futures,
                         QueryResult result,
                         Collection<Integer> partitionIds)
                              throws ExecutionException,
                                     InterruptedException

Adds results of non-paging predicates to result set and removes queried partition ids.

Throws:

ExecutionException

InterruptedException

toObject

protected Object toObject(Object obj)

getLocalPartitionIds

protected List<Integer> getLocalPartitionIds()

getAllPartitionIds

protected Set<Integer> getAllPartitionIds()

createSetWithPopulatedPartitionIds

protected Set<Integer> createSetWithPopulatedPartitionIds(int partitionCount)

getNow

protected long getNow()

hasOwnerMigrationsInFlight

protected boolean hasOwnerMigrationsInFlight()

Check whether migrations of owner partition are currently executed. If a migration is in progress, do not attempt to use an index as they may have not been created yet. MapService.getMigrationsInFlight() returns the number of currently executing migrations (for which beforeMigration has been executed but commit/rollback is not yet executed). This check is a temporary fix for 3.7, the actual issue will be addressed with an additional migration hook in 3.8. see https://github.com/hazelcast/hazelcast/issues/6471 & https://github.com/hazelcast/hazelcast/issues/8046

Returns:

true if owner partition migrations are currently being executed, otherwise false.

See Also:

DelegatingMigrationAwareService

hasPartitionStateVersionChanged

protected boolean hasPartitionStateVersionChanged(int initialPartitionStateVersion)

Check whether partition state version has changed since initialPartitionStateVersion.

Parameters:

initialPartitionStateVersion - the initial partition state version to compare against

Returns:

true if current partition state version is not equal to initialPartitionStateVersion

isResultSafe

protected boolean isResultSafe(int initialPartitionStateVersion)

Check whether results obtained since partition state version was at initialPartitionStateVersion are safe to be returned to the caller. Effectively this method checks:

• whether owner migrations are currently in flight; if there are any, then results are considered flawed
• whether current partition state version has changed, implying that some migrations were executed in the meanwhile, so results are again considered flawed

Parameters:

initialPartitionStateVersion -

Returns:

true if no owner migrations are currently executing and initialPartitionStateVersion is the same as the current partition state version, otherwise false.