You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by gv...@apache.org on 2020/04/06 18:51:39 UTC
[ignite] branch ignite-12248 updated (4654a3e -> 79d31de)
This is an automated email from the ASF dual-hosted git repository.
gvvinblade pushed a change to branch ignite-12248
in repository https://gitbox.apache.org/repos/asf/ignite.git.
from 4654a3e IGNITE-12820 Calcite integration. Do not use AbstarctConverter while planning. This closes #7553
add 57cd12c IGNITE-12763: OpenCensus example. (#7506)
add 683f22e IGNITE-12756 TcpCommunication SPI metrics improvement
add 3094ade IGNITE-12753 Cache SSLContext in SslContextFactory - Fixes #7497.
add 009b0ee IGNITE-12740: Supports feature flags on index meta page (#7493)
add 9b63bc7 IGNITE-12771: Fix javadoc. (#7512)
add 4921fcf IGNITE-12701 : Disallow silent deactivation in CLI and REST. (#7471)
add e389bb8 IGNITE-12728 Collect time statistics on cache#putAllAsync - Fixes #7483.
add 13637a2 IGNITE-12745: Compute job view implemented. (#7495)
add ef4f67e IGNITE-12743 Java thin client: Fixed thread shutdown on client close when partition awareness is enabled - Fixes #7522.
add dc547e2 IGNITE-12746 Regression in GridCacheColocatedDebugTest: putAll of sorted keys causes deadlock - Fixes #7507.
add 19ffc88 IGNITE-12752 Pass client certificates to security - Fixes #7508.
add 577be1d IGNITE-12503 Fix httpCfg assignment in Jetty REST configurations - Fixes #7525.
add 336a7ed IGNITE-12752 Pass client certificates to security - fix resource.
add bcaae8d IGNITE-12734 Fixed scan query over evicted partition - Fixes #7494.
add a855014 IGNITE-12768: MetricRegistryMBean doesn't show histogram values in case when histogram name contains underscore character (#7542)
add 3daa183 IGNITE-12787 Remove obsolete Selector.open workaround code from GridNioServer (#7535)
add 2261487 IGNITE-7792 Generalize logic for checkpointing machinery - Fixes #7513.
add 3928bb3 IGNITE-12799 Fixed wrong SpEL expression.
add 795617f IGNITE-12774 Handle "Too many open files" exception - Fixes #7516.
add 4a94ac2 IGNITE-12761 Added ability to disable check crc sums of stored pages due to invalidate_indexes.
add 6a4a9c8 IGNITE-12773 : Reduce number of cluster deactivation methods in internal API. (#7514)
add 89e7112 IGNITE-9017 .NET: Add ICache.ClearStatistics, IClusterGroup.ClearStatistics
add 0b185b1 IGNITE-12800 SQL: local queries cursors must be closed or full read to unlock the GridH2Table. (#7551)
add 1c50667 IGNITE-8152 Forbid empty sql schema name (#7521)
add 7aa5cfd IGNITE-12709 Server latch initialized after client latch in Zookeeper discovery - Fixes #7459.
add 5bc0900 IGNITE-12810: Cancel compute task command. (#7555)
add fd3065b IGNITE-12789 Delta record for tracking page repairing.
add 6f35156 IGNITE-12769: histogramNames cache in MetricRegistryMBean removed. (#7549)
add d0bae90 IGNITE-12757 Visor CMD: Fixed invalid properties - Fixes #7505.
add 9f87186 IGNITE-12811: Cancel service command (#7565)
add 458b452 IGNITE-12829: Custom GROUP_CONCAT separator is ignored. (#7574)
add 3f0f818 IGNITE-12812: Cancel transaction command. (#7568)
add 50547b9 IGNITE-12826: JdbcDialect#setFetchSize added. (#7577)
add 87d5526 IGNITE-12788 Adds cluster fully rebalanced state metric. (#7561)
add aefad94 IGNITE-12815: SQL query cancel command. (#7580)
add c9309dc IGNITE-12400: Remove the stopProcess method from the DiscoveryCustomMessage interface (#7268)
add 068b1dc IGNITE-12842 Cache IGNITE_DAEMON system property lookup - Fixes #7579.
add 169099f IGNITE-12783 Fixed erroneously triggered partitions validation for cache group. Fixes #7564
add 61a1aaa IGNITE-12220 Allow to use cache-related permissions both at system and per-cache levels (#6904)
add bf125403 IGNITE-12813: Cancel scan query command. (#7588)
add 7f018a8 IGNITE-12710 Introduce extended log mode for index creation\rebuilt. (#7461)
add 7d55653 IGNITE-12818 SoLinger should be used for all sockets in discovery - Fixes #7557.
add a8b4efc IGNITE-12691 .NET: Add native Near Cache
add 41da31e IGNITE-12836 Writing methods to avoid limit of 65535 bytes for java.io.DataOutput#writeUTF - Fixes #7571.
add 5175780 IGNITE-12804 SQL: ConnectionManager refactoring (#7569)
add da48436 IGNITE-12344 Remote listener of IgniteMessaging has to run with appropriate SecurityContext - Fixes #7338.
add 02ecb25 IGNITE-12860 .NET: Fix NRE in BinaryStructure.GetFieldId
add ae95eef IGNITE-12797 Add command line option to CommandHandler to be able to see full stack trace and cause exception in log in case of error. - Fixes #7544.
add ccea4f7 IGNITE-12814: Cancel continuous query command. (#7608)
add 2b1d2b4 IGNITE-12859: Support of .Net service call with the Timestamp and Guid params. (#7618)
add ee07f34 IGNITE-12839 WAL serialization test fixed. - Fixes #7573.
add bc810eb IGNITE-12766 Added GridCacheProcessor.LocalAffinityFunction for backward compatilility.
add d580b84 IGNITE-12802 Move checkpoint state fields to CheckpointProgress - Fixes #7589.
new 10b2768 Merge branch 'ignite-master' into ignite-12248
new 79d31de IGNITE-12700: Calcite integration. Aggregates support. #7598
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
examples/pom.xml | 18 +
.../OpenCensusMetricsExporterExample.java | 105 ++
.../OpenCensusMetricsExporterExampleSelfTest.java | 18 +-
.../testsuites/IgniteExamplesSelfTestSuite.java | 4 +
.../exec/{exp/Scalar.java => ArrayRowHandler.java} | 27 +-
.../query/calcite/exec/LogicalRelImplementor.java | 49 +
.../query/calcite/exec/PhysicalRelImplementor.java | 35 +
.../exec/{exp/Scalar.java => RowHandler.java} | 15 +-
.../processors/query/calcite/exec/exp/Call.java | 3 +-
.../query/calcite/exec/exp/ExpToRexTranslator.java | 19 +
.../query/calcite/exec/exp/ExpressionFactory.java | 67 +-
.../query/calcite/exec/exp/RexToExpTranslator.java | 21 +
.../processors/query/calcite/exec/exp/Scalar.java | 2 +-
.../query/calcite/exec/exp/agg/Accumulator.java} | 25 +-
.../{Scalar.java => agg/AccumulatorWrapper.java} | 18 +-
.../query/calcite/exec/exp/agg/Accumulators.java | 674 ++++++++++
.../query/calcite/exec/exp/agg/AggCallExp.java | 178 +++
.../query/calcite/exec/exp/agg/GroupKey.java | 91 ++
.../calcite/exec/exp/agg/WrappersFactoryImpl.java | 406 ++++++
.../query/calcite/exec/rel/AggregateNode.java | 360 +++++
.../metadata/IgniteMdDerivedDistribution.java | 31 +-
.../calcite/metadata/IgniteMdDistribution.java | 69 +-
.../calcite/metadata/IgniteMdFragmentInfo.java | 22 +-
.../processors/query/calcite/prepare/Cloner.java | 29 +-
.../query/calcite/prepare/FragmentSplitter.java | 95 +-
.../query/calcite/prepare/IgniteSqlValidator.java | 28 +
.../query/calcite/prepare/PlannerPhase.java | 4 +
.../processors/query/calcite/prepare/Splitter.java | 18 +
.../query/calcite/rel/IgniteAggregate.java | 73 ++
...gniteTableScan.java => IgniteMapAggregate.java} | 31 +-
.../query/calcite/rel/IgniteReduceAggregate.java | 105 ++
.../query/calcite/rel/IgniteRelVisitor.java | 15 +
.../query/calcite/rel/IgniteTableScan.java | 2 +
.../processors/query/calcite/rel/IgniteValues.java | 6 +-
...verterRule.java => AggregateConverterRule.java} | 28 +-
.../rule/AggregateTraitsPropagationRule.java | 94 ++
.../calcite/rule/FilterTraitsPropagationRule.java | 5 +-
.../calcite/rule/JoinTraitsPropagationRule.java | 20 +-
.../calcite/rule/ProjectTraitsPropagationRule.java | 7 +-
.../processors/query/calcite/rule/RuleUtils.java | 11 +
.../query/calcite/rule/ValuesConverterRule.java | 4 +-
.../calcite/serialize/AggregatePhysicalRel.java | 108 ++
.../calcite/serialize/PhysicalRelVisitor.java | 5 +
.../calcite/serialize/RelToPhysicalConverter.java | 40 +
.../query/calcite/trait/IgniteDistributions.java | 318 ++++-
.../processors/query/calcite/util/Commons.java | 25 +
.../query/calcite/util/IgniteResource.java | 4 +
.../query/calcite/CalciteQueryProcessorTest.java | 46 +
.../processors/query/calcite/PlannerTest.java | 92 ++
.../query/calcite/exec/rel/ExecutionTest.java | 554 ++++++++
.../internal/jdbc2/JdbcMetadataSelfTest.java | 1 +
.../rest/ChangeStateCommandHandlerTest.java | 13 +-
.../rest/JettyRestProcessorAbstractSelfTest.java | 67 +-
.../rest/JettyRestProcessorCommonSelfTest.java | 25 +-
.../rest/protocols/tcp/TcpRestParserSelfTest.java | 1 +
.../JdbcThinConnectionAdditionalSecurityTest.java | 2 +-
.../jdbc/thin/JdbcThinConnectionSSLTest.java | 2 +-
.../jdbc/thin/JdbcThinLocalQueriesSelfTest.java | 45 +-
.../ignite/jdbc/thin/JdbcThinMetadataSelfTest.java | 20 +-
...nsactionFinishOnDeactivatedClusterSelfTest.java | 4 +-
modules/clients/src/test/keystore/README.txt | 11 +
modules/clients/src/test/keystore/ca/node01.jks | Bin 3719 -> 0 bytes
modules/clients/src/test/keystore/ca/node02.jks | Bin 4599 -> 0 bytes
modules/clients/src/test/keystore/ca/node02old.jks | Bin 4598 -> 0 bytes
modules/clients/src/test/keystore/ca/node03.jks | Bin 3753 -> 0 bytes
.../clients/src/test/keystore/ca/oneca-index.txt | 3 +
modules/clients/src/test/keystore/ca/oneca-serial | 1 +
modules/clients/src/test/keystore/ca/oneca.cnf | 13 +-
modules/clients/src/test/keystore/ca/oneca.key | 52 +-
modules/clients/src/test/keystore/ca/oneca.pem | 21 +
modules/clients/src/test/keystore/ca/oneindex.txt | 1 -
.../clients/src/test/keystore/ca/oneindex.txt.attr | 1 -
modules/clients/src/test/keystore/ca/oneserial | 1 -
.../clients/src/test/keystore/ca/threeca-index.txt | 2 +
.../test/keystore/ca/{twoserial => threeca-serial} | 0
.../test/keystore/ca/{oneca.cnf => threeca.cnf} | 17 +-
modules/clients/src/test/keystore/ca/threeca.key | 28 +
modules/clients/src/test/keystore/ca/threeca.pem | 21 +
.../clients/src/test/keystore/ca/trust-both.jks | Bin 1718 -> 0 bytes
modules/clients/src/test/keystore/ca/trust-one.jks | Bin 877 -> 0 bytes
modules/clients/src/test/keystore/ca/trust-two.jks | Bin 891 -> 0 bytes
.../clients/src/test/keystore/ca/twoca-index.txt | 5 +
modules/clients/src/test/keystore/ca/twoca-serial | 1 +
modules/clients/src/test/keystore/ca/twoca.cnf | 15 +-
modules/clients/src/test/keystore/ca/twoca.key | 52 +-
modules/clients/src/test/keystore/ca/twoca.pem | 21 +
modules/clients/src/test/keystore/ca/twoindex.txt | 2 -
.../clients/src/test/keystore/ca/twoindex.txt.attr | 1 -
modules/clients/src/test/keystore/client.jks | Bin 3232 -> 2324 bytes
modules/clients/src/test/keystore/client.pem | 69 -
modules/clients/src/test/keystore/client.pfx | Bin 3148 -> 0 bytes
.../clients/src/test/keystore/connectorClient.jks | Bin 0 -> 2353 bytes
.../clients/src/test/keystore/connectorServer.jks | Bin 0 -> 2355 bytes
modules/clients/src/test/keystore/generate-ca.sh | 97 ++
modules/clients/src/test/keystore/generate-keys.sh | 92 ++
modules/clients/src/test/keystore/generate.sh | 99 --
modules/clients/src/test/keystore/node01.jks | Bin 0 -> 2325 bytes
modules/clients/src/test/keystore/node0102.jks | Bin 0 -> 4618 bytes
modules/clients/src/test/keystore/node02.jks | Bin 0 -> 2325 bytes
modules/clients/src/test/keystore/node02old.jks | Bin 0 -> 2331 bytes
modules/clients/src/test/keystore/node03.jks | Bin 0 -> 2327 bytes
modules/clients/src/test/keystore/server.jks | Bin 3230 -> 2324 bytes
modules/clients/src/test/keystore/server.pem | 69 -
modules/clients/src/test/keystore/server.pfx | Bin 3148 -> 0 bytes
modules/clients/src/test/keystore/thinClient.jks | Bin 0 -> 2332 bytes
modules/clients/src/test/keystore/thinServer.jks | Bin 0 -> 2332 bytes
modules/clients/src/test/keystore/trust-both.jks | Bin 0 -> 1888 bytes
modules/clients/src/test/keystore/trust-one.jks | Bin 0 -> 960 bytes
modules/clients/src/test/keystore/trust-three.jks | Bin 0 -> 970 bytes
modules/clients/src/test/keystore/trust-two.jks | Bin 0 -> 960 bytes
modules/clients/src/test/keystore/trust.jks | Bin 2432 -> 0 bytes
.../jetty/rest-jetty-ssl-client-auth.xml} | 15 +-
.../src/test/resources/jetty/rest-jetty-ssl.xml | 2 +-
.../src/test/resources/jetty/rest-jetty.xml | 2 +-
.../src/test/resources/jetty/router-jetty-ssl.xml | 2 +-
.../src/test/resources/jetty/router-jetty.xml | 2 +-
.../src/test/resources/spring-router-ssl.xml | 2 +-
.../src/test/resources/spring-server-ssl-node.xml | 2 +-
.../SystemViewRowAttributeWalkerGenerator.java | 2 +
.../ignite/compatibility/cache/LocalCacheTest.java | 136 ++
.../ignite/compatibility/cache/package-info.java} | 11 +-
.../IgniteCompatibilityBasicTestSuite.java | 4 +-
modules/core/pom.xml | 2 +-
.../src/main/java/org/apache/ignite/Ignite.java | 3 +
.../main/java/org/apache/ignite/IgniteCluster.java | 6 +
.../org/apache/ignite/IgniteSystemProperties.java | 8 +
.../cache/store/jdbc/CacheAbstractJdbcStore.java | 8 +-
.../cache/store/jdbc/dialect/BasicJdbcDialect.java | 14 +-
.../cache/store/jdbc/dialect/MySQLDialect.java | 14 +-
.../org/apache/ignite/cluster/ClusterState.java | 7 +-
.../ignite/configuration/CacheConfiguration.java | 34 +
.../PlatformNearCacheConfiguration.java | 113 ++
.../apache/ignite/internal/ComputeMXBeanImpl.java | 92 ++
.../ignite/internal/GridKernalContextImpl.java | 6 +-
.../org/apache/ignite/internal/IgniteFeatures.java | 8 +
.../ignite/internal/IgniteMessagingImpl.java | 20 +-
.../internal/IgniteTooManyOpenFilesException.java | 66 +
.../apache/ignite/internal/QueryMXBeanImpl.java | 210 +++
...reFilter.java => SecurityAwareBiPredicate.java} | 25 +-
.../apache/ignite/internal/ServiceMXBeanImpl.java | 57 +
.../ignite/internal/TransactionsMXBeanImpl.java | 18 +-
.../internal/client/GridClientClusterState.java | 18 +-
.../client/impl/GridClientClusterStateImpl.java | 33 +-
.../impl/connection/GridClientConnection.java | 22 +-
.../connection/GridClientNioTcpConnection.java | 31 +-
.../internal/client/thin/ReliableChannel.java | 28 +-
.../internal/client/util/GridClientUtils.java | 41 +
.../ignite/internal/cluster/IgniteClusterImpl.java | 10 +-
.../internal/commandline/ActivateCommand.java | 3 +-
.../commandline/ClusterStateChangeCommand.java | 25 +-
.../internal/commandline/CommandHandler.java | 37 +-
.../ignite/internal/commandline/CommandList.java | 6 +-
.../internal/commandline/CommonArgParser.java | 13 +-
.../commandline/ConnectionAndSslParameters.java | 16 +-
.../internal/commandline/DeactivateCommand.java | 28 +-
.../ignite/internal/commandline/TxCommands.java | 32 +-
.../commandline/cache/CacheValidateIndexes.java | 37 +-
.../cache/argument/ValidateIndexesCommandArg.java | 3 +
.../internal/commandline/query/KillCommand.java | 214 +++
.../query/KillSubcommand.java} | 65 +-
.../internal/dto/IgniteDataTransferObject.java | 3 +
.../internal/managers/GridManagerAdapter.java | 18 +
.../internal/managers/IgniteMBeansManager.java | 18 +
.../managers/communication/GridIoManager.java | 94 +-
.../communication/IgniteMessageFactoryImpl.java | 37 +-
.../managers/discovery/CustomMessageWrapper.java | 3 +-
.../managers/discovery/DiscoveryCustomMessage.java | 8 -
.../systemview/walker/ComputeJobViewWalker.java | 77 ++
.../systemview/walker/ComputeTaskViewWalker.java | 50 +-
.../internal/pagemem/wal/record/WALRecord.java | 7 +-
...ageInitRootInlineFlagsCreatedVersionRecord.java | 98 ++
.../wal/record/delta/TrackingPageDeltaRecord.java | 30 +-
.../delta/TrackingPageRepairDeltaRecord.java} | 34 +-
.../authentication/UserAcceptedMessage.java | 5 -
.../authentication/UserProposedMessage.java | 5 -
.../cache/CacheAffinityChangeMessage.java | 5 -
.../cache/CacheStatisticsClearMessage.java | 5 -
.../cache/CacheStatisticsModeChangeMessage.java | 5 -
.../cache/ClientCacheChangeDiscoveryMessage.java | 5 -
.../ClientCacheChangeDummyDiscoveryMessage.java | 5 -
.../processors/cache/DynamicCacheChangeBatch.java | 5 -
.../cache/DynamicCacheChangeFailureMessage.java | 5 -
.../processors/cache/GridCacheAdapter.java | 11 +-
.../processors/cache/GridCacheContext.java | 5 +-
.../processors/cache/GridCacheEntryEx.java | 7 +-
.../processors/cache/GridCacheMapEntry.java | 70 +-
.../cache/GridCachePartitionExchangeManager.java | 17 +
.../processors/cache/GridCacheProcessor.java | 59 +-
.../cache/IgniteCacheOffheapManager.java | 5 +
.../cache/IgniteCacheOffheapManagerImpl.java | 61 +-
.../internal/processors/cache/QueryCursorImpl.java | 40 +-
...TimeoutOnPartitionMapExchangeChangeMessage.java | 5 -
.../processors/cache/WalStateFinishMessage.java | 8 +-
.../internal/processors/cache/WalStateManager.java | 2 +-
.../processors/cache/WalStateProposeMessage.java | 10 +-
.../binary/MetadataUpdateAcceptedMessage.java | 5 -
.../binary/MetadataUpdateProposedMessage.java | 5 -
.../distributed/GridDistributedCacheEntry.java | 7 +-
.../cache/distributed/dht/GridDhtCacheEntry.java | 2 +
.../preloader/GridDhtPartitionsExchangeFuture.java | 16 +-
.../dht/preloader/latch/ExchangeLatchManager.java | 13 +-
.../cache/distributed/near/GridNearCacheEntry.java | 10 +
.../cache/local/GridLocalCacheEntry.java | 8 +-
.../CheckpointWriteProgressSupplier.java | 45 -
.../GridCacheDatabaseSharedManager.java | 786 ++++-------
.../cache/persistence/GridCacheOffheapManager.java | 10 +
.../IgniteCacheDatabaseSharedManager.java | 1 +
.../persistence/checkpoint/CheckpointProgress.java | 106 ++
.../checkpoint/CheckpointProgressImpl.java | 271 ++++
.../checkpoint/PartitionDestroyQueue.java | 77 ++
.../checkpoint/PartitionDestroyRequest.java | 127 ++
.../cache/persistence/metastorage/MetaStorage.java | 2 +-
.../cache/persistence/pagemem/PageMemoryImpl.java | 19 +-
.../pagemem/PagesWriteSpeedBasedThrottle.java | 19 +-
.../persistence/pagemem/PagesWriteThrottle.java | 15 +-
.../cache/persistence/tree/BPlusTree.java | 12 +-
.../cache/persistence/tree/io/BPlusMetaIO.java | 175 ++-
.../cache/persistence/tree/io/TrackingPageIO.java | 18 +-
.../wal/serializer/RecordDataV1Serializer.java | 60 +-
.../wal/serializer/RecordDataV2Serializer.java | 18 +
.../cache/query/GridCacheQueryManager.java | 8 +-
.../query/continuous/SecurityAwareFilter.java | 1 +
.../continuous/SecurityAwareFilterFactory.java | 1 +
.../SecurityAwareTransformerFactory.java | 1 +
.../cache/transactions/IgniteTxStateImpl.java | 2 +-
.../processors/cache/tree/CacheDataTree.java | 2 +
.../processors/cache/verify/IdleVerifyUtility.java | 2 +-
.../cluster/ChangeGlobalStateFinishMessage.java | 5 -
.../cluster/ChangeGlobalStateMessage.java | 19 +-
.../cluster/GridClusterStateProcessor.java | 119 +-
.../cluster/IGridClusterStateProcessor.java | 17 +-
.../continuous/AbstractContinuousMessage.java | 5 -
.../continuous/StopRoutineAckDiscoveryMessage.java | 5 -
.../internal/processors/job/GridJobProcessor.java | 35 +-
.../internal/processors/job/GridJobWorker.java | 28 +-
.../marshaller/MappingAcceptedMessage.java | 5 -
.../marshaller/MappingProposedMessage.java | 5 -
.../DistributedMetaStorageUpdateAckMessage.java | 5 -
.../DistributedMetaStorageUpdateMessage.java | 5 -
.../processors/metric/GridMetricManager.java | 6 +
.../processors/metric/impl/MetricUtils.java | 24 +-
.../ClientListenerAbstractConnectionContext.java | 8 +-
.../odbc/ClientListenerConnectionContext.java | 5 +-
.../processors/odbc/ClientListenerNioListener.java | 10 +-
.../odbc/jdbc/JdbcConnectionContext.java | 14 +-
.../odbc/odbc/OdbcConnectionContext.java | 12 +-
.../processors/platform/PlatformContext.java | 29 +
.../processors/platform/PlatformContextImpl.java | 78 +-
.../processors/platform/PlatformNoopProcessor.java | 16 +
.../processors/platform/PlatformProcessor.java | 22 +
.../processors/platform/PlatformProcessorImpl.java | 63 +-
.../processors/platform/cache/PlatformCache.java | 22 +
.../platform/cache/PlatformCacheEntryFilter.java | 9 +
.../cache/PlatformCacheEntryFilterImpl.java | 67 +-
.../platform/cache/PlatformCacheManager.java | 86 ++
.../platform/cache/affinity/PlatformAffinity.java | 30 +
.../platform/callback/PlatformCallbackGateway.java | 74 ++
.../platform/callback/PlatformCallbackOp.java | 12 +
.../platform/client/ClientConnectionContext.java | 6 +-
.../platform/cluster/PlatformClusterGroup.java | 45 +-
.../platform/utils/PlatformConfigurationUtils.java | 28 +
.../processors/platform/utils/PlatformUtils.java | 4 +-
.../internal/processors/query/GridQueryCancel.java | 7 +-
.../processors/query/GridQueryIndexing.java | 11 +
.../processors/query/GridQueryProcessor.java | 9 +-
.../schema/SchemaIndexCachePartitionWorker.java | 72 +-
.../query/schema/SchemaIndexCacheStat.java | 88 ++
.../query/schema/SchemaIndexCacheVisitorImpl.java | 98 +-
.../message/SchemaFinishDiscoveryMessage.java | 5 -
.../message/SchemaProposeDiscoveryMessage.java | 5 -
.../processors/resource/GridResourceIoc.java | 19 +-
.../processors/resource/GridResourceProcessor.java | 9 +-
.../processors/rest/GridRestProcessor.java | 1 +
.../message/GridClientClusterStateRequest.java | 5 +-
...t.java => GridClientClusterStateRequestV2.java} | 52 +-
.../GridChangeClusterStateCommandHandler.java | 3 +-
.../cluster/GridChangeStateCommandHandler.java | 5 +-
.../rest/protocols/tcp/GridTcpRestNioListener.java | 7 +
.../rest/request/GridRestChangeStateRequest.java | 20 +
.../rest/request/GridRestClusterStateRequest.java | 21 +
.../processors/rest/request/GridRestRequest.java | 18 +
.../AbstractSecurityAwareExternalizable.java | 2 +-
.../service/ServiceChangeBatchRequest.java | 5 -
.../ServiceClusterDeploymentResultBatch.java | 5 -
.../processors/task/GridTaskProcessor.java | 4 +-
.../internal/resources/MetricManagerResource.java | 32 -
.../org/apache/ignite/internal/sql/SqlKeyword.java | 12 +
.../org/apache/ignite/internal/sql/SqlParser.java | 24 +
.../sql/command/SqlKillComputeTaskCommand.java | 66 +
.../sql/command/SqlKillContinuousQueryCommand.java | 81 ++
.../internal/sql/command/SqlKillQueryCommand.java | 51 +-
.../sql/command/SqlKillScanQueryCommand.java | 97 ++
.../sql/command/SqlKillServiceCommand.java | 63 +
.../sql/command/SqlKillTransactionCommand.java | 63 +
.../util/GridConcurrentMultiPairQueue.java | 202 +++
.../internal/util/GridReadOnlyArrayView.java | 80 --
.../apache/ignite/internal/util/IgniteUtils.java | 198 +++
.../internal/util/distributed/FullMessage.java | 5 -
.../internal/util/distributed/InitMessage.java | 5 -
.../ignite/internal/util/nio/GridNioServer.java | 14 -
.../ignite/internal/util/nio/GridNioSession.java | 6 +
.../internal/util/nio/GridNioSessionImpl.java | 20 +
.../internal/util/nio/ssl/GridNioSslFilter.java | 2 +
...ask.java => VisorComputeCancelSessionTask.java} | 37 +-
.../compute/VisorComputeCancelSessionTaskArg.java | 73 ++
.../compute/VisorComputeCancelSessionsTask.java | 4 +-
.../VisorContinuousQueryCancelTask.java} | 38 +-
.../query/VisorContinuousQueryCancelTaskArg.java | 89 ++
.../VisorQueryCancelOnInitiatorTask.java} | 43 +-
.../query/VisorQueryCancelOnInitiatorTaskArg.java | 81 ++
.../VisorScanQueryCancelTask.java} | 35 +-
.../visor/query/VisorScanQueryCancelTaskArg.java | 101 ++
.../visor/service/VisorCancelServiceTask.java | 6 +-
.../visor/verify/VisorValidateIndexesTaskArg.java | 34 +-
.../apache/ignite/lang/IgniteProductVersion.java | 14 +-
.../AbstractNodeNameAwareMarshaller.java | 8 +-
.../org/apache/ignite/marshaller/Marshaller.java | 10 +-
.../ComputeMXBean.java} | 20 +-
.../org/apache/ignite/mxbean/IgniteMXBean.java | 9 +-
.../java/org/apache/ignite/mxbean/QueryMXBean.java | 67 +
.../ServiceMXBean.java} | 16 +-
.../apache/ignite/mxbean/TransactionsMXBean.java | 12 +
.../plugin/security/AuthenticationContext.java | 21 +
.../security/SecurityPermissionSetBuilder.java | 7 +-
.../ignite/plugin/security/SecuritySubject.java | 10 +
.../org/apache/ignite/spi/IgniteSpiAdapter.java | 33 +-
.../org/apache/ignite/spi/IgniteSpiContext.java | 36 +
.../tcp/TcpCommunicationMetricsListener.java | 156 ++-
.../spi/communication/tcp/TcpCommunicationSpi.java | 79 +-
.../spi/discovery/DiscoverySpiCustomMessage.java | 4 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 10 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 3 +-
.../ignite/spi/discovery/tcp/TcpDiscoverySpi.java | 18 +-
.../ignite/spi/metric/jmx/MetricRegistryMBean.java | 55 +-
.../ignite/spi/systemview/view/ComputeJobView.java | 197 +++
.../spi/systemview/view/ComputeTaskView.java | 60 +-
.../org/apache/ignite/ssl/SslContextFactory.java | 29 +-
.../main/resources/META-INF/classnames.properties | 2 +
modules/core/src/test/config/tests.properties | 26 +-
.../jdbc/CacheJdbcPojoStoreAbstractSelfTest.java | 42 +-
.../failure/FailureHandlerTriggeredTest.java | 4 -
.../DiagnosticLogForPartitionStatesTest.java | 103 +-
.../ThinClientAbstractPartitionAwarenessTest.java | 22 +
...lientPartitionAwarenessResourceReleaseTest.java | 74 ++
.../commandline/CommandHandlerParsingTest.java | 138 +-
.../IgniteMessageFactoryImplTest.java | 47 +-
.../IncompleteDeserializationExceptionTest.java | 5 -
.../ignite/internal/metric/JmxExporterSpiTest.java | 51 +-
.../ignite/internal/metric/MetricsSelfTest.java | 8 +-
.../internal/metric/SystemViewComputeJobTest.java | 517 ++++++++
.../cache/ClusterStateThinClientAbstractTest.java | 2 +-
.../cache/GridCacheAbstractMetricsSelfTest.java | 27 +-
.../processors/cache/GridCacheTestEntryEx.java | 4 +-
.../dht/GridCacheColocatedDebugTest.java | 109 ++
.../preloader/latch/ExchangeLatchManagerTest.java | 75 ++
.../persistence/IgnitePdsCorruptedStoreTest.java | 3 +-
.../IgniteCheckpointDirtyPagesForLowLoadTest.java | 2 +-
.../db/checkpoint/ProgressWatchdog.java | 6 +-
.../pagemem/BPlusTreePageMemoryImplTest.java | 12 +-
.../BPlusTreeReuseListPageMemoryImplTest.java | 12 +-
.../IgnitePageMemReplaceDelayedWriteUnitTest.java | 12 +-
.../pagemem/IgniteThrottlingUnitTest.java | 14 +-
.../pagemem/IndexStoragePageMemoryImplTest.java | 12 +-
.../pagemem/PageMemoryImplNoLoadTest.java | 12 +-
.../persistence/pagemem/PageMemoryImplTest.java | 17 +-
.../pagemem/PagesWriteThrottleSandboxTest.java | 9 +-
.../cache/query/CacheScanQueryFailoverTest.java | 131 ++
.../CacheContinuousQueryLongP2PTest.java | 6 +-
.../continuous/GridEventConsumeSelfTest.java | 91 +-
.../processors/database/BPlusTreeSelfTest.java | 2 +-
.../GridCommandHandlerSslWithSecurityTest.java | 1 +
...eOperationPermissionCreateDestroyCheckTest.java | 164 +++
.../client/AdditionalSecurityCheckTest.java | 3 +-
.../client/ThinClientSslPermissionCheckTest.java | 237 ++++
.../TestCertificateSecurityPluginProvider.java} | 36 +-
....java => TestCertificateSecurityProcessor.java} | 91 +-
.../security/impl/TestSecurityContext.java | 7 +-
.../security/impl/TestSecurityProcessor.java | 1 +
.../security/impl/TestSecuritySubject.java | 19 +
.../MessagingRemoteSecurityContextCheckTest.java | 148 +++
.../util/GridConcurrentMultiPairQueueTest.java | 155 +++
.../ignite/internal/util/IgniteUtilsSelfTest.java | 137 +-
.../util/nio/impl/GridNioFilterChainSelfTest.java | 139 --
.../internal/util/nio/impl}/MockNioSession.java | 8 +-
.../loadtests/hashmap/GridCacheTestContext.java | 3 +-
.../ignite/platform/PlatformDeployServiceTask.java | 48 +-
.../ignite/platform/PlatformThreadUtils.java | 51 +
.../security/SecurityPermissionSetBuilderTest.java | 22 +-
.../TooManyOpenFilesTcpCommunicationSpiTest.java | 130 ++
.../spi/discovery/ClusterRebalancedMetricTest.java | 187 +++
.../FilterDataForClientNodeDiscoveryTest.java | 5 -
.../discovery/tcp/DiscoveryClientSocketTest.java | 184 +++
.../TcpDiscoveryPendingMessageDeliveryTest.java | 5 -
.../tcp/TcpDiscoverySslTrustedUntrustedTest.java | 16 +
.../apache/ignite/ssl/MultipleSSLContextsTest.java | 263 ++++
.../ignite/testframework/GridSpiTestContext.java | 30 +-
.../testframework/junits/IgniteTestResources.java | 16 +-
.../testframework/wal/record/RecordUtils.java | 8 +
.../testsuites/IgniteCacheMvccTestSuite6.java | 2 +
.../ignite/testsuites/IgniteCacheTestSuite6.java | 2 +
.../ignite/testsuites/IgniteCacheTestSuite9.java | 2 +
.../IgniteSpiCommunicationSelfTestSuite.java | 3 +
.../IgniteSpiDiscoverySelfTestSuite.java | 5 +-
.../ignite/testsuites/IgniteUtilSelfTestSuite.java | 4 +
.../ignite/testsuites/SecurityTestSuite.java | 12 +-
.../util/GridCommandHandlerClusterByClassTest.java | 104 +-
.../apache/ignite/util/GridCommandHandlerTest.java | 58 +
...andHandlerClusterByClassTest_cache_help.output} | 3 +-
...idCommandHandlerClusterByClassTest_help.output} | 45 +-
...lerClusterByClassWithSSLTest_cache_help.output} | 3 +-
...ndHandlerClusterByClassWithSSLTest_help.output} | 45 +-
.../resources/org.apache.ignite.util/bigUtf.txt | Bin 0 -> 430138 bytes
.../resources/org.apache.ignite.util/limitUtf.txt | 10 +
.../query/h2/opt/GridH2SpatialIndex.java | 8 +-
modules/hibernate-4.2/pom.xml | 2 +-
modules/hibernate-5.1/pom.xml | 2 +-
modules/hibernate-5.3/pom.xml | 2 +-
.../cache/query/RegisteredQueryCursor.java | 62 +-
.../processors/query/h2/CommandProcessor.java | 73 +-
.../processors/query/h2/ConcurrentStripedPool.java | 177 +++
.../processors/query/h2/ConnectionManager.java | 426 ++----
.../processors/query/h2/H2CachedStatementKey.java | 27 +-
.../internal/processors/query/h2/H2Connection.java | 207 +++
.../processors/query/h2/H2ConnectionWrapper.java | 133 --
.../processors/query/h2/H2FieldsIterator.java | 20 +-
.../processors/query/h2/H2PooledConnection.java | 129 ++
.../processors/query/h2/H2ResultSetIterator.java | 263 +++-
.../processors/query/h2/H2StatementCache.java | 25 +-
.../internal/processors/query/h2/H2Utils.java | 231 +++-
.../processors/query/h2/IgniteH2Indexing.java | 110 +-
.../internal/processors/query/h2/QueryParser.java | 448 +++----
.../processors/query/h2/SchemaManager.java | 54 +-
.../query/h2/database/H2PkHashIndex.java | 3 +-
.../processors/query/h2/database/H2Tree.java | 150 ++-
.../processors/query/h2/database/H2TreeIndex.java | 60 +-
.../query/h2/database/InlineIndexHelper.java | 2 +-
.../processors/query/h2/dml/UpdatePlan.java | 48 +-
.../processors/query/h2/dml/UpdatePlanBuilder.java | 13 +-
.../processors/query/h2/opt/GridH2IndexBase.java | 5 +-
.../processors/query/h2/opt/GridH2Table.java | 22 +-
.../processors/query/h2/opt/QueryContext.java | 46 +
.../query/h2/opt/QueryContextRegistry.java | 34 +-
.../query/h2/opt/join/DistributedLookupBatch.java | 29 +-
.../query/h2/sql/GridSqlQuerySplitter.java | 30 +-
.../query/h2/twostep/GridMapQueryExecutor.java | 64 +-
.../query/h2/twostep/GridReduceQueryExecutor.java | 260 ++--
.../query/h2/twostep/MapQueryResult.java | 23 +-
.../query/h2/twostep/MapQueryResults.java | 9 -
.../query/h2/twostep/ReduceQueryRun.java | 15 -
.../visor/verify/ValidateIndexesClosure.java | 37 +-
.../visor/verify/VisorValidateIndexesTask.java | 7 +-
.../org/apache/ignite/client/ClientTestSuite.java | 4 +-
.../cache/IgniteCacheQueryH2IndexingLeakTest.java | 14 +-
.../cache/index/H2ConnectionLeaksSelfTest.java | 54 +-
.../cache/metric/SqlViewExporterSpiTest.java | 1 +
...xingMultithreadedLoadContinuousRestartTest.java | 2 +-
.../db/LongDestroyDurableBackgroundTaskTest.java | 2 +-
.../processors/database/RebuildIndexTest.java | 232 ++++
.../RebuildIndexWithHistoricalRebalanceTest.java | 2 +-
...actQueryTableLockAndConnectionPoolSelfTest.java | 123 +-
.../query/IgniteSqlGroupConcatCollocatedTest.java | 26 +
.../IgniteSqlGroupConcatNotCollocatedTest.java | 25 +
.../IgniteSqlSkipReducerOnUpdateDmlSelfTest.java | 2 +-
.../query/IgniteSqlSplitterSelfTest.java | 3 +-
.../internal/processors/query/KillQueryTest.java | 56 +-
.../processors/query/LocalQueryLazyTest.java | 128 +-
.../query/h2/GridIndexFullRebuildTest.java | 2 +-
.../processors/query/h2/QueryDataPageScanTest.java | 7 +-
.../query/h2/sql/GridQueryParsingTest.java | 60 +-
.../org/apache/ignite/sqltests/BaseSqlTest.java | 39 +
.../IgniteBinaryCacheQueryTestSuite.java | 6 +
.../testsuites/IgnitePdsWithIndexingTestSuite.java | 4 +-
...idCommandHandlerIndexingClusterByClassTest.java | 12 +
.../util/GridCommandHandlerIndexingTest.java | 55 +-
.../ignite/util/KillCommandsCommandShTest.java | 176 +++
.../apache/ignite/util/KillCommandsMXBeanTest.java | 181 +++
.../apache/ignite/util/KillCommandsSQLTest.java | 196 +++
.../org/apache/ignite/util/KillCommandsTests.java | 396 ++++++
.../opencensus/OpenCensusMetricExporterSpi.java | 25 +-
modules/platforms/.gitignore | 7 +-
.../ExpiryCacheHolderTest.cs | 5 +
.../Apache.Ignite.Benchmarks.csproj | 3 +
.../Apache.Ignite.Benchmarks/BenchmarkRunner.cs | 17 +-
.../Apache.Ignite.Benchmarks/Config/benchmark.xml | 23 +
.../Interop/GetNearBenchmark.cs | 64 +
.../Interop/PutNearBenchmark.cs | 59 +
.../Interop/ScanQueryBenchmark.cs | 86 ++
.../Apache.Ignite.Core.Tests.DotNetCore.csproj | 59 +
.../Apache.Ignite.Core.Tests.csproj | 10 +-
.../Binary/BinaryStructureTest.cs | 130 ++
.../Binary/TypeNameParserTest.cs | 22 +-
.../Cache/CacheMetricsTest.cs | 82 ++
.../Cache/CacheNearTest.cs | 200 ---
.../Cache/CacheTestAsyncWrapper.cs | 8 +-
.../Cache/Near/CacheNearCreateDestroy.cs | 348 +++++
.../Cache/Near/CacheNearTest.cs | 1371 ++++++++++++++++++++
.../Cache/Near/CacheNearTopologyChangeTest.cs | 614 +++++++++
.../Cache/Near/FailingCacheStore.cs | 72 +
.../Apache.Ignite.Core.Tests/Cache/Near/Foo.cs | 66 +
.../Cache/Near/ScanQueryNearCacheFilter.cs} | 33 +-
.../Cache/Near/ScanQueryNoNearCacheFilter.cs} | 36 +-
.../Cache/Near/StoreNoNearCacheFilter.cs} | 32 +-
.../Cache/Query/Linq/CacheLinqTest.Misc.cs | 7 +-
.../Cache/Store/CacheStoreTest.cs | 3 +-
.../Client/Cache/ListLogger.cs | 19 +-
.../Config/full-config.xml | 1 +
.../IgniteConfigurationSerializerTest.cs | 11 +
.../Apache.Ignite.Core.Tests/IgniteUtilsTest.cs | 71 +
.../Services/ServicesTest.cs | 47 +
.../Apache.Ignite.Core.Tests/TestUtils.Common.cs | 49 +-
.../Apache.Ignite.Core/Apache.Ignite.Core.csproj | 5 +
.../Apache.Ignite.Core/Cache/CachePeekMode.cs | 9 +-
.../Cache/Configuration/CacheConfiguration.cs | 31 +
.../PlatformNearCacheConfiguration.cs | 72 +
.../Cache/Eviction/FifoEvictionPolicy.cs | 2 +-
.../dotnet/Apache.Ignite.Core/Cache/ICache.cs | 5 +
.../Cache/Store/CacheStoreAdapter.cs | 2 +-
.../Apache.Ignite.Core/Cluster/IClusterGroup.cs | 7 +-
.../Apache.Ignite.Core/Events/DiscoveryEvent.cs | 2 +-
.../platforms/dotnet/Apache.Ignite.Core/IIgnite.cs | 53 +
.../IgniteConfigurationSection.xsd | 22 +
.../Impl/Binary/Structure/BinaryStructure.cs | 135 +-
.../Impl/Binary/Structure/BinaryStructureEntry.cs | 2 +-
.../Impl/Cache/CacheAffinityImpl.cs | 18 +-
.../Impl/Cache/CacheEntryFilterHolder.cs | 40 +-
.../Apache.Ignite.Core/Impl/Cache/CacheImpl.cs | 480 ++++++-
.../Apache.Ignite.Core/Impl/Cache/CacheOp.cs | 4 +-
.../Impl/Cache/Near/INearCache.cs | 82 ++
.../Impl/Cache/Near/NearCache.cs | 275 ++++
.../Impl/Cache/Near/NearCacheEntry.cs | 85 ++
.../Impl/Cache/Near/NearCacheManager.cs | 198 +++
.../Impl/Cache/Query/QueryCursor.cs | 2 +-
.../Impl/Cache/Query/QueryCursorBase.cs | 2 +-
.../Impl/Client/Cache/ClientCachePartitionMap.cs | 14 +-
.../Impl/Client/ClientFailoverSocket.cs | 8 +-
.../Impl/Client/Cluster/ClientClusterGroup.cs | 3 +-
.../Apache.Ignite.Core/Impl/Client/IgniteClient.cs | 26 +
.../Impl/Cluster/ClusterGroupImpl.cs | 66 +-
.../Impl/Common/CopyOnWriteConcurrentDictionary.cs | 24 +
.../Apache.Ignite.Core/Impl/Common/TaskRunner.cs | 22 +-
.../Apache.Ignite.Core/Impl/IIgniteInternal.cs | 30 +-
.../Impl/IPlatformTargetInternal.cs | 2 +-
.../dotnet/Apache.Ignite.Core/Impl/Ignite.cs | 118 +-
.../dotnet/Apache.Ignite.Core/Impl/IgniteUtils.cs | 12 +-
.../Apache.Ignite.Core/Impl/PlatformJniTarget.cs | 10 +-
.../Impl/PlatformTargetAdapter.cs | 53 +-
.../Impl/Services/ServiceProxySerializer.cs | 13 +
.../Impl/Unmanaged/UnmanagedCallbackOp.cs | 6 +-
.../Impl/Unmanaged/UnmanagedCallbacks.cs | 61 +-
.../Apache.Ignite.DotNetCore.sln.DotSettings | 1 +
modules/platforms/dotnet/Apache.Ignite.sln | 1 -
.../platforms/dotnet/Apache.Ignite.sln.DotSettings | 1 +
.../Datagrid/NearCacheExample.cs | 10 +-
.../protocols/http/jetty/GridJettyRestHandler.java | 28 +-
.../IgniteSpringDataCrudSelfExpressionTest.java | 21 +
.../misc/PersonExpressionRepository.java | 2 +-
.../IgniteSpringDataCrudSelfExpressionTest.java | 21 +
.../misc/PersonExpressionRepository.java | 2 +-
modules/spring/pom.xml | 7 +
.../visor/commands/open/VisorOpenCommand.scala | 13 +-
.../src/test/resources/jetty-with-ciphers-0.xml | 2 +-
.../src/test/resources/jetty-with-ciphers-1.xml | 2 +-
.../src/test/resources/jetty-with-ciphers-2.xml | 2 +-
.../src/test/resources/jetty-with-ssl.xml | 2 +-
.../ZkCommunicationErrorResolveFinishMessage.java | 1 +
.../ZkCommunicationErrorResolveStartMessage.java | 1 +
.../zk/internal/ZkForceNodeFailMessage.java | 1 +
.../discovery/zk/internal/ZkNoServersMessage.java | 1 +
.../zk/internal/ZookeeperDiscoveryImpl.java | 33 +-
.../ZookeeperDiscoveryCustomEventsTest.java | 339 -----
parent/pom.xml | 4 +-
571 files changed, 23051 insertions(+), 5046 deletions(-)
create mode 100644 examples/src/main/java/org/apache/ignite/examples/opencensus/OpenCensusMetricsExporterExample.java
copy modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheEntryFilter.java => examples/src/test/java/org/apache/ignite/examples/OpenCensusMetricsExporterExampleSelfTest.java (68%)
copy modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/{exp/Scalar.java => ArrayRowHandler.java} (61%)
copy modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/{exp/Scalar.java => RowHandler.java} (82%)
rename modules/{core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CheckpointProgress.java => calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulator.java} (60%)
copy modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/{Scalar.java => agg/AccumulatorWrapper.java} (80%)
create mode 100644 modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulators.java
create mode 100644 modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/AggCallExp.java
create mode 100644 modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/GroupKey.java
create mode 100644 modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/WrappersFactoryImpl.java
create mode 100644 modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AggregateNode.java
create mode 100644 modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteAggregate.java
copy modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/{IgniteTableScan.java => IgniteMapAggregate.java} (54%)
create mode 100644 modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReduceAggregate.java
copy modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/{ValuesConverterRule.java => AggregateConverterRule.java} (67%)
create mode 100644 modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AggregateTraitsPropagationRule.java
create mode 100644 modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/AggregatePhysicalRel.java
create mode 100644 modules/clients/src/test/keystore/README.txt
delete mode 100644 modules/clients/src/test/keystore/ca/node01.jks
delete mode 100644 modules/clients/src/test/keystore/ca/node02.jks
delete mode 100644 modules/clients/src/test/keystore/ca/node02old.jks
delete mode 100644 modules/clients/src/test/keystore/ca/node03.jks
create mode 100644 modules/clients/src/test/keystore/ca/oneca-index.txt
create mode 100644 modules/clients/src/test/keystore/ca/oneca-serial
create mode 100644 modules/clients/src/test/keystore/ca/oneca.pem
delete mode 100644 modules/clients/src/test/keystore/ca/oneindex.txt
delete mode 100644 modules/clients/src/test/keystore/ca/oneindex.txt.attr
delete mode 100644 modules/clients/src/test/keystore/ca/oneserial
create mode 100644 modules/clients/src/test/keystore/ca/threeca-index.txt
rename modules/clients/src/test/keystore/ca/{twoserial => threeca-serial} (100%)
copy modules/clients/src/test/keystore/ca/{oneca.cnf => threeca.cnf} (82%)
create mode 100644 modules/clients/src/test/keystore/ca/threeca.key
create mode 100644 modules/clients/src/test/keystore/ca/threeca.pem
delete mode 100644 modules/clients/src/test/keystore/ca/trust-both.jks
delete mode 100644 modules/clients/src/test/keystore/ca/trust-one.jks
delete mode 100644 modules/clients/src/test/keystore/ca/trust-two.jks
create mode 100644 modules/clients/src/test/keystore/ca/twoca-index.txt
create mode 100644 modules/clients/src/test/keystore/ca/twoca-serial
create mode 100644 modules/clients/src/test/keystore/ca/twoca.pem
delete mode 100644 modules/clients/src/test/keystore/ca/twoindex.txt
delete mode 100644 modules/clients/src/test/keystore/ca/twoindex.txt.attr
delete mode 100644 modules/clients/src/test/keystore/client.pem
delete mode 100644 modules/clients/src/test/keystore/client.pfx
create mode 100644 modules/clients/src/test/keystore/connectorClient.jks
create mode 100644 modules/clients/src/test/keystore/connectorServer.jks
create mode 100755 modules/clients/src/test/keystore/generate-ca.sh
create mode 100755 modules/clients/src/test/keystore/generate-keys.sh
delete mode 100644 modules/clients/src/test/keystore/generate.sh
create mode 100644 modules/clients/src/test/keystore/node01.jks
create mode 100644 modules/clients/src/test/keystore/node0102.jks
create mode 100644 modules/clients/src/test/keystore/node02.jks
create mode 100644 modules/clients/src/test/keystore/node02old.jks
create mode 100644 modules/clients/src/test/keystore/node03.jks
delete mode 100644 modules/clients/src/test/keystore/server.pem
delete mode 100644 modules/clients/src/test/keystore/server.pfx
create mode 100644 modules/clients/src/test/keystore/thinClient.jks
create mode 100644 modules/clients/src/test/keystore/thinServer.jks
create mode 100644 modules/clients/src/test/keystore/trust-both.jks
create mode 100644 modules/clients/src/test/keystore/trust-one.jks
create mode 100644 modules/clients/src/test/keystore/trust-three.jks
create mode 100644 modules/clients/src/test/keystore/trust-two.jks
delete mode 100644 modules/clients/src/test/keystore/trust.jks
copy modules/{web-console/web-agent/src/test/resources/jetty-with-ssl.xml => clients/src/test/resources/jetty/rest-jetty-ssl-client-auth.xml} (86%)
create mode 100644 modules/compatibility/src/test/java/org/apache/ignite/compatibility/cache/LocalCacheTest.java
copy modules/{calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java => compatibility/src/test/java/org/apache/ignite/compatibility/cache/package-info.java} (76%)
create mode 100644 modules/core/src/main/java/org/apache/ignite/configuration/PlatformNearCacheConfiguration.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/ComputeMXBeanImpl.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/IgniteTooManyOpenFilesException.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/QueryMXBeanImpl.java
copy modules/core/src/main/java/org/apache/ignite/internal/{processors/cache/query/continuous/SecurityAwareFilter.java => SecurityAwareBiPredicate.java} (62%)
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/ServiceMXBeanImpl.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/commandline/query/KillCommand.java
copy modules/core/src/main/java/org/apache/ignite/internal/{processors/rest/request/GridRestChangeStateRequest.java => commandline/query/KillSubcommand.java} (55%)
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/managers/systemview/walker/ComputeJobViewWalker.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/delta/MetaPageInitRootInlineFlagsCreatedVersionRecord.java
copy modules/core/src/main/java/org/apache/ignite/internal/{processors/continuous/StopRoutineAckDiscoveryMessage.java => pagemem/wal/record/delta/TrackingPageRepairDeltaRecord.java} (51%)
delete mode 100644 modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CheckpointWriteProgressSupplier.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointProgress.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/CheckpointProgressImpl.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/PartitionDestroyQueue.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/checkpoint/PartitionDestroyRequest.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/processors/platform/cache/PlatformCacheManager.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCacheStat.java
copy modules/core/src/main/java/org/apache/ignite/internal/processors/rest/client/message/{GridClientClusterStateRequest.java => GridClientClusterStateRequestV2.java} (58%)
rename modules/core/src/main/java/org/apache/ignite/internal/processors/{cache/query/continuous => security}/AbstractSecurityAwareExternalizable.java (96%)
delete mode 100644 modules/core/src/main/java/org/apache/ignite/internal/resources/MetricManagerResource.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/sql/command/SqlKillComputeTaskCommand.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/sql/command/SqlKillContinuousQueryCommand.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/sql/command/SqlKillScanQueryCommand.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/sql/command/SqlKillServiceCommand.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/sql/command/SqlKillTransactionCommand.java
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/util/GridConcurrentMultiPairQueue.java
delete mode 100644 modules/core/src/main/java/org/apache/ignite/internal/util/GridReadOnlyArrayView.java
copy modules/core/src/main/java/org/apache/ignite/internal/visor/compute/{VisorComputeCancelSessionsTask.java => VisorComputeCancelSessionTask.java} (59%)
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/visor/compute/VisorComputeCancelSessionTaskArg.java
copy modules/core/src/main/java/org/apache/ignite/internal/visor/{service/VisorCancelServiceTask.java => query/VisorContinuousQueryCancelTask.java} (53%)
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorContinuousQueryCancelTaskArg.java
copy modules/core/src/main/java/org/apache/ignite/internal/visor/{service/VisorCancelServiceTask.java => query/VisorQueryCancelOnInitiatorTask.java} (53%)
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorQueryCancelOnInitiatorTaskArg.java
copy modules/core/src/main/java/org/apache/ignite/internal/visor/{service/VisorCancelServiceTask.java => query/VisorScanQueryCancelTask.java} (59%)
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/visor/query/VisorScanQueryCancelTaskArg.java
copy modules/core/src/main/java/org/apache/ignite/{internal/processors/platform/cache/PlatformCacheEntryFilter.java => mxbean/ComputeMXBean.java} (58%)
create mode 100644 modules/core/src/main/java/org/apache/ignite/mxbean/QueryMXBean.java
copy modules/core/src/main/java/org/apache/ignite/{internal/processors/platform/cache/PlatformCacheEntryFilter.java => mxbean/ServiceMXBean.java} (68%)
create mode 100644 modules/core/src/main/java/org/apache/ignite/spi/systemview/view/ComputeJobView.java
create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/client/thin/ThinClientPartitionAwarenessResourceReleaseTest.java
create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/metric/SystemViewComputeJobTest.java
create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/latch/ExchangeLatchManagerTest.java
create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/processors/security/cache/CacheOperationPermissionCreateDestroyCheckTest.java
create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/processors/security/client/ThinClientSslPermissionCheckTest.java
rename modules/{indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2KeyValueIterator.java => core/src/test/java/org/apache/ignite/internal/processors/security/impl/TestCertificateSecurityPluginProvider.java} (51%)
copy modules/core/src/test/java/org/apache/ignite/internal/processors/security/impl/{TestSecurityProcessor.java => TestCertificateSecurityProcessor.java} (64%)
create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/processors/security/messaging/MessagingRemoteSecurityContextCheckTest.java
create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/util/GridConcurrentMultiPairQueueTest.java
rename modules/{clients/src/test/java/org/apache/ignite/internal/processors/rest/protocols/tcp => core/src/test/java/org/apache/ignite/internal/util/nio/impl}/MockNioSession.java (95%)
create mode 100644 modules/core/src/test/java/org/apache/ignite/platform/PlatformThreadUtils.java
create mode 100644 modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/TooManyOpenFilesTcpCommunicationSpiTest.java
create mode 100644 modules/core/src/test/java/org/apache/ignite/spi/discovery/ClusterRebalancedMetricTest.java
create mode 100644 modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/DiscoveryClientSocketTest.java
create mode 100644 modules/core/src/test/java/org/apache/ignite/ssl/MultipleSSLContextsTest.java
copy modules/core/src/test/resources/org.apache.ignite.util/{control.sh_cache_help.output => GridCommandHandlerClusterByClassTest_cache_help.output} (87%)
copy modules/core/src/test/resources/org.apache.ignite.util/{control.sh_help.output => GridCommandHandlerClusterByClassTest_help.output} (69%)
rename modules/core/src/test/resources/org.apache.ignite.util/{control.sh_cache_help.output => GridCommandHandlerClusterByClassWithSSLTest_cache_help.output} (87%)
rename modules/core/src/test/resources/org.apache.ignite.util/{control.sh_help.output => GridCommandHandlerClusterByClassWithSSLTest_help.output} (69%)
create mode 100644 modules/core/src/test/resources/org.apache.ignite.util/bigUtf.txt
create mode 100644 modules/core/src/test/resources/org.apache.ignite.util/limitUtf.txt
create mode 100644 modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/ConcurrentStripedPool.java
create mode 100644 modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2Connection.java
delete mode 100644 modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2ConnectionWrapper.java
create mode 100644 modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2PooledConnection.java
create mode 100644 modules/indexing/src/test/java/org/apache/ignite/internal/processors/database/RebuildIndexTest.java
create mode 100644 modules/indexing/src/test/java/org/apache/ignite/util/KillCommandsCommandShTest.java
create mode 100644 modules/indexing/src/test/java/org/apache/ignite/util/KillCommandsMXBeanTest.java
create mode 100644 modules/indexing/src/test/java/org/apache/ignite/util/KillCommandsSQLTest.java
create mode 100644 modules/indexing/src/test/java/org/apache/ignite/util/KillCommandsTests.java
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Benchmarks/Interop/GetNearBenchmark.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Benchmarks/Interop/PutNearBenchmark.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Benchmarks/Interop/ScanQueryBenchmark.cs
delete mode 100644 modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/CacheNearTest.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Near/CacheNearCreateDestroy.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Near/CacheNearTest.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Near/CacheNearTopologyChangeTest.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Near/FailingCacheStore.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Core.Tests/Cache/Near/Foo.cs
copy modules/platforms/dotnet/{Apache.Ignite.Core/Impl/Cache/Query/QueryCursor.cs => Apache.Ignite.Core.Tests/Cache/Near/ScanQueryNearCacheFilter.cs} (57%)
copy modules/platforms/dotnet/{Apache.Ignite.Core/Impl/Cache/Query/QueryCursor.cs => Apache.Ignite.Core.Tests/Cache/Near/ScanQueryNoNearCacheFilter.cs} (54%)
copy modules/platforms/dotnet/{Apache.Ignite.Core/Impl/Cache/Query/QueryCursor.cs => Apache.Ignite.Core.Tests/Cache/Near/StoreNoNearCacheFilter.cs} (52%)
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Core.Tests/IgniteUtilsTest.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Core/Cache/Configuration/PlatformNearCacheConfiguration.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Near/INearCache.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Near/NearCache.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Near/NearCacheEntry.cs
create mode 100644 modules/platforms/dotnet/Apache.Ignite.Core/Impl/Cache/Near/NearCacheManager.cs
[ignite] 02/02: IGNITE-12700: Calcite integration. Aggregates
support. #7598
Posted by gv...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gvvinblade pushed a commit to branch ignite-12248
in repository https://gitbox.apache.org/repos/asf/ignite.git
commit 79d31de9f23af35e1ecd789cdb5e6a0f96b8691c
Author: Igor Seliverstov <gv...@gmail.com>
AuthorDate: Mon Apr 6 17:55:35 2020 +0300
IGNITE-12700: Calcite integration. Aggregates support. #7598
---
.../exec/{exp/Scalar.java => ArrayRowHandler.java} | 27 +-
.../query/calcite/exec/LogicalRelImplementor.java | 49 ++
.../query/calcite/exec/PhysicalRelImplementor.java | 35 ++
.../exec/{exp/Scalar.java => RowHandler.java} | 15 +-
.../processors/query/calcite/exec/exp/Call.java | 3 +-
.../query/calcite/exec/exp/ExpToRexTranslator.java | 19 +
.../query/calcite/exec/exp/ExpressionFactory.java | 67 +-
.../query/calcite/exec/exp/RexToExpTranslator.java | 21 +
.../processors/query/calcite/exec/exp/Scalar.java | 2 +-
.../exec/exp/{Scalar.java => agg/Accumulator.java} | 24 +-
.../{Scalar.java => agg/AccumulatorWrapper.java} | 18 +-
.../query/calcite/exec/exp/agg/Accumulators.java | 674 +++++++++++++++++++++
.../query/calcite/exec/exp/agg/AggCallExp.java | 178 ++++++
.../query/calcite/exec/exp/agg/GroupKey.java | 91 +++
.../calcite/exec/exp/agg/WrappersFactoryImpl.java | 406 +++++++++++++
.../query/calcite/exec/rel/AggregateNode.java | 360 +++++++++++
.../metadata/IgniteMdDerivedDistribution.java | 31 +-
.../calcite/metadata/IgniteMdDistribution.java | 69 +--
.../calcite/metadata/IgniteMdFragmentInfo.java | 22 +-
.../processors/query/calcite/prepare/Cloner.java | 29 +-
.../query/calcite/prepare/FragmentSplitter.java | 95 ++-
.../query/calcite/prepare/IgniteSqlValidator.java | 28 +
.../query/calcite/prepare/PlannerPhase.java | 4 +
.../processors/query/calcite/prepare/Splitter.java | 18 +
.../query/calcite/rel/IgniteAggregate.java | 73 +++
...gniteTableScan.java => IgniteMapAggregate.java} | 31 +-
.../query/calcite/rel/IgniteReduceAggregate.java | 105 ++++
.../query/calcite/rel/IgniteRelVisitor.java | 15 +
.../query/calcite/rel/IgniteTableScan.java | 2 +
.../processors/query/calcite/rel/IgniteValues.java | 6 +-
...verterRule.java => AggregateConverterRule.java} | 28 +-
.../rule/AggregateTraitsPropagationRule.java | 94 +++
.../calcite/rule/FilterTraitsPropagationRule.java | 5 +-
.../calcite/rule/JoinTraitsPropagationRule.java | 20 +-
.../calcite/rule/ProjectTraitsPropagationRule.java | 7 +-
.../processors/query/calcite/rule/RuleUtils.java | 11 +
.../query/calcite/rule/ValuesConverterRule.java | 4 +-
.../calcite/serialize/AggregatePhysicalRel.java | 108 ++++
.../calcite/serialize/PhysicalRelVisitor.java | 5 +
.../calcite/serialize/RelToPhysicalConverter.java | 40 ++
.../query/calcite/trait/IgniteDistributions.java | 318 ++++++++--
.../processors/query/calcite/util/Commons.java | 25 +
.../query/calcite/util/IgniteResource.java | 4 +
.../query/calcite/CalciteQueryProcessorTest.java | 46 ++
.../processors/query/calcite/PlannerTest.java | 92 +++
.../query/calcite/exec/rel/ExecutionTest.java | 554 +++++++++++++++++
46 files changed, 3669 insertions(+), 209 deletions(-)
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ArrayRowHandler.java
similarity index 61%
copy from modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java
copy to modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ArrayRowHandler.java
index 02ec14e..db8f4a4 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ArrayRowHandler.java
@@ -15,13 +15,30 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.query.calcite.exec.exp;
-
-import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+package org.apache.ignite.internal.processors.query.calcite.exec;
/**
*
*/
-public interface Scalar {
- void execute(ExecutionContext ctx, Object[] in, Object[] out);
+public class ArrayRowHandler implements RowHandler<Object[]> {
+ /** */
+ public static final RowHandler<Object[]> INSTANCE = new ArrayRowHandler();
+
+ /** */
+ private ArrayRowHandler() {}
+
+ /** {@inheritDoc} */
+ @Override public Object[] create(Object... fields) {
+ return fields;
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> T get(int field, Object[] row) {
+ return (T) row[field];
+ }
+
+ /** {@inheritDoc} */
+ @Override public void set(int field, Object[] row, Object value) {
+ row[field] = value;
+ }
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
index b6e9b3b..68c55fc 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java
@@ -17,8 +17,10 @@
package org.apache.ignite.internal.processors.query.calcite.exec;
+import java.util.List;
import java.util.function.Function;
import java.util.function.Predicate;
+import java.util.function.Supplier;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.schema.ScannableTable;
@@ -26,6 +28,8 @@ import org.apache.calcite.sql.SqlOperatorTable;
import org.apache.calcite.sql.validate.SqlConformance;
import org.apache.ignite.internal.processors.failure.FailureProcessor;
import org.apache.ignite.internal.processors.query.calcite.exec.exp.ExpressionFactory;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AccumulatorWrapper;
+import org.apache.ignite.internal.processors.query.calcite.exec.rel.AggregateNode;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.FilterNode;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.Inbox;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.JoinNode;
@@ -37,11 +41,14 @@ import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanNode;
import org.apache.ignite.internal.processors.query.calcite.metadata.PartitionService;
import org.apache.ignite.internal.processors.query.calcite.prepare.Fragment;
import org.apache.ignite.internal.processors.query.calcite.prepare.RelTarget;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteExchange;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteFilter;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteJoin;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteMapAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteProject;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReceiver;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReduceAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRelVisitor;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
@@ -179,6 +186,48 @@ public class LogicalRelImplementor implements IgniteRelVisitor<Node<Object[]>> {
}
/** {@inheritDoc} */
+ @Override public Node<Object[]> visit(IgniteAggregate rel) {
+ AggregateNode.AggregateType type = AggregateNode.AggregateType.SINGLE;
+ RowHandler<Object[]> rowHandler = ArrayRowHandler.INSTANCE;
+
+ Supplier<List<AccumulatorWrapper>> factory = expressionFactory.wrappersFactory(ctx,
+ rowHandler, type, rel.getAggCallList(), rel.getInput().getRowType());
+
+ AggregateNode<Object[]> node = new AggregateNode<>(ctx, type, rel.getGroupSets(), factory, rowHandler);
+ node.register(visit(rel.getInput()));
+
+ return node;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Node<Object[]> visit(IgniteMapAggregate rel) {
+ AggregateNode.AggregateType type = AggregateNode.AggregateType.MAP;
+ RowHandler<Object[]> rowHandler = ArrayRowHandler.INSTANCE;
+
+ Supplier<List<AccumulatorWrapper>> factory = expressionFactory.wrappersFactory(ctx,
+ rowHandler, type, rel.getAggCallList(), rel.getInput().getRowType());
+
+ AggregateNode<Object[]> node = new AggregateNode<>(ctx, type, rel.getGroupSets(), factory, rowHandler);
+ node.register(visit(rel.getInput()));
+
+ return node;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Node<Object[]> visit(IgniteReduceAggregate rel) {
+ AggregateNode.AggregateType type = AggregateNode.AggregateType.REDUCE;
+ RowHandler<Object[]> rowHandler = ArrayRowHandler.INSTANCE;
+
+ Supplier<List<AccumulatorWrapper>> factory = expressionFactory.wrappersFactory(ctx,
+ rowHandler, type, rel.aggregateCalls(), null);
+
+ AggregateNode<Object[]> node = new AggregateNode<>(ctx, type, rel.groupSets(), factory, rowHandler);
+ node.register(visit(rel.getInput()));
+
+ return node;
+ }
+
+ /** {@inheritDoc} */
@Override public Node<Object[]> visit(IgniteRel rel) {
return rel.accept(this);
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PhysicalRelImplementor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PhysicalRelImplementor.java
index fbcac21..1760465 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PhysicalRelImplementor.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/PhysicalRelImplementor.java
@@ -17,11 +17,15 @@
package org.apache.ignite.internal.processors.query.calcite.exec;
+import java.util.List;
+import java.util.function.Supplier;
import org.apache.calcite.schema.ScannableTable;
import org.apache.calcite.sql.SqlOperatorTable;
import org.apache.calcite.sql.validate.SqlConformance;
import org.apache.ignite.internal.processors.failure.FailureProcessor;
import org.apache.ignite.internal.processors.query.calcite.exec.exp.ExpressionFactory;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AccumulatorWrapper;
+import org.apache.ignite.internal.processors.query.calcite.exec.rel.AggregateNode;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.FilterNode;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.Inbox;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.JoinNode;
@@ -32,6 +36,7 @@ import org.apache.ignite.internal.processors.query.calcite.exec.rel.ProjectNode;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanNode;
import org.apache.ignite.internal.processors.query.calcite.metadata.PartitionService;
import org.apache.ignite.internal.processors.query.calcite.schema.TableDescriptor;
+import org.apache.ignite.internal.processors.query.calcite.serialize.AggregatePhysicalRel;
import org.apache.ignite.internal.processors.query.calcite.serialize.FilterPhysicalRel;
import org.apache.ignite.internal.processors.query.calcite.serialize.JoinPhysicalRel;
import org.apache.ignite.internal.processors.query.calcite.serialize.PhysicalRel;
@@ -164,6 +169,36 @@ public class PhysicalRelImplementor implements PhysicalRelVisitor<Node<Object[]>
}
/** {@inheritDoc} */
+ @Override public Node<Object[]> visit(AggregatePhysicalRel rel) {
+ AggregateNode.AggregateType type;
+
+ switch (rel.type()) {
+ case AggregatePhysicalRel.MAP:
+ type = AggregateNode.AggregateType.MAP;
+ break;
+ case AggregatePhysicalRel.REDUCE:
+ type = AggregateNode.AggregateType.REDUCE;
+ break;
+ case AggregatePhysicalRel.SINGLE:
+ type = AggregateNode.AggregateType.SINGLE;
+ break;
+ default:
+ throw new AssertionError();
+ }
+
+ RowHandler<Object[]> rowHandler = ArrayRowHandler.INSTANCE;
+
+ Supplier<List<AccumulatorWrapper>> factory = expressionFactory.wrappersFactory(ctx,
+ rowHandler, type, rel.calls(), rel.inputRowType());
+
+ AggregateNode<Object[]> node = new AggregateNode<>(ctx, type, rel.groupSets(), factory, rowHandler);
+
+ node.register(visit(rel.input()));
+
+ return node;
+ }
+
+ /** {@inheritDoc} */
@Override public Node<Object[]> visit(PhysicalRel rel) {
return rel.accept(this);
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/RowHandler.java
similarity index 82%
copy from modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java
copy to modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/RowHandler.java
index 02ec14e..b9579c8 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/RowHandler.java
@@ -15,13 +15,18 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.query.calcite.exec.exp;
-
-import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+package org.apache.ignite.internal.processors.query.calcite.exec;
/**
*
*/
-public interface Scalar {
- void execute(ExecutionContext ctx, Object[] in, Object[] out);
+public interface RowHandler<R> {
+ /** */
+ R create(Object... fields);
+
+ /** */
+ <T> T get(int field, R row);
+
+ /** */
+ void set(int field, R row, Object value);
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Call.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Call.java
index d128d2d..91573d3 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Call.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Call.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.query.calcite.exec.exp;
import java.util.List;
+import java.util.Objects;
import org.apache.calcite.sql.SqlIdentifier;
import org.apache.calcite.sql.SqlOperator;
import org.apache.calcite.sql.SqlOperatorTable;
@@ -77,7 +78,7 @@ public class Call implements Expression {
new SqlIdentifier(opName, SqlParserPos.ZERO), null,
syntax, bag, SqlNameMatchers.withCaseSensitive(true));
- return F.first(bag);
+ return Objects.requireNonNull(F.first(bag));
}
/** */
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpToRexTranslator.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpToRexTranslator.java
index 8e9a9e8..d02c449 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpToRexTranslator.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpToRexTranslator.java
@@ -18,9 +18,13 @@
package org.apache.ignite.internal.processors.query.calcite.exec.exp;
import java.util.List;
+import org.apache.calcite.rel.RelCollations;
+import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rex.RexBuilder;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.SqlOperatorTable;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AggCallExp;
import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
@@ -70,6 +74,21 @@ public class ExpToRexTranslator implements ExpressionVisitor<RexNode> {
return exp.accept(this);
}
+ /** */
+ public AggregateCall translate(AggCallExp desc) {
+ return AggregateCall.create(
+ desc.sqlOperator(opTable),
+ desc.distinct(),
+ desc.approximate(),
+ desc.ignoreNulls(),
+ ImmutableIntList.of(desc.args()),
+ desc.filterArg(),
+ RelCollations.EMPTY,
+ desc.resultType().logicalType(typeFactory),
+ desc.name()
+ );
+ }
+
/** {@inheritDoc} */
@Override public RexNode visit(InputRef exp) {
return builder.makeInputRef(exp.logicalType(typeFactory), exp.index());
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactory.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactory.java
index 0c026b6..d3e30ff 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactory.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactory.java
@@ -28,6 +28,7 @@ import java.util.Map;
import java.util.NoSuchElementException;
import java.util.function.Function;
import java.util.function.Predicate;
+import java.util.function.Supplier;
import org.apache.calcite.adapter.enumerable.JavaRowFormat;
import org.apache.calcite.adapter.enumerable.PhysTypeImpl;
import org.apache.calcite.adapter.enumerable.RexImpTable;
@@ -38,6 +39,7 @@ import org.apache.calcite.linq4j.tree.Expressions;
import org.apache.calcite.linq4j.tree.MethodDeclaration;
import org.apache.calcite.linq4j.tree.ParameterExpression;
import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rex.RexBuilder;
@@ -52,8 +54,14 @@ import org.apache.calcite.util.Pair;
import org.apache.calcite.util.Util;
import org.apache.ignite.IgniteException;
import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AccumulatorWrapper;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AggCallExp;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.WrappersFactoryImpl;
import org.apache.ignite.internal.processors.query.calcite.exec.exp.type.DataType;
+import org.apache.ignite.internal.processors.query.calcite.exec.rel.AggregateNode;
import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.processors.query.calcite.util.IgniteMethod;
import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashMap;
import org.apache.ignite.internal.util.typedef.F;
@@ -65,12 +73,13 @@ import org.codehaus.commons.compiler.ICompilerFactory;
* Implements rex expression into a function object. Uses JaninoRexCompiler under the hood.
* Each expression compiles into a class and a wrapper over it is returned.
*/
+@SuppressWarnings({"rawtypes"})
public class ExpressionFactory {
/** */
- private static final Map<String, Scalar> CACHE_1 = new GridBoundedConcurrentLinkedHashMap<>(1024);
+ private static final Map<String, Scalar> SCALAR_CACHE = new GridBoundedConcurrentLinkedHashMap<>(1024);
/** */
- private static final Map<String, CallOperation> CACHE_2 = new GridBoundedConcurrentLinkedHashMap<>(1024);
+ private static final Map<String, CallOperation> CALL_OP_CACHE = new GridBoundedConcurrentLinkedHashMap<>(1024);
/** */
private final IgniteTypeFactory typeFactory;
@@ -105,6 +114,39 @@ public class ExpressionFactory {
emptyType = new RelDataTypeFactory.Builder(typeFactory).build();
}
+ /** */
+ public IgniteTypeFactory typeFactory() {
+ return typeFactory;
+ }
+
+ /** */
+ public RexBuilder rexBuilder() {
+ return rexBuilder;
+ }
+
+ /** */
+ public ExpToRexTranslator expToRexTranslator() {
+ return expToRexTranslator;
+ }
+
+ /** */
+ public RexToExpTranslator rexToExpTranslator() {
+ return rexToExpTranslator;
+ }
+
+ /** */
+ public Supplier<List<AccumulatorWrapper>> wrappersFactory(ExecutionContext root, RowHandler handler, AggregateNode.AggregateType type, List<AggregateCall> calls, RelDataType rowType) {
+ return new WrappersFactoryImpl(root, type, handler, calls, rowType);
+ }
+
+ /** */
+ public Supplier<List<AccumulatorWrapper>> wrappersFactory(ExecutionContext root, RowHandler handler, AggregateNode.AggregateType type, List<AggCallExp> calls, DataType rowType) {
+ List<AggregateCall> calls0 = Commons.transform(calls, expToRexTranslator::translate);
+ RelDataType rowType0 = rowType == null ? null : rowType.logicalType(typeFactory);
+
+ return wrappersFactory(root, handler, type, calls0, rowType0);
+ }
+
/**
* Creates a comparator for given data type and collations. Mainly used for sorted exchange.
*
@@ -215,17 +257,17 @@ public class ExpressionFactory {
/** */
public CallOperation implement(Call call) {
- return CACHE_2.computeIfAbsent(digest(call), k -> compile(call));
+ return CALL_OP_CACHE.computeIfAbsent(digest(call), k -> compile(call));
}
/** */
- private Scalar scalar(RexNode node, RelDataType type) {
+ public Scalar scalar(RexNode node, RelDataType type) {
return scalar(ImmutableList.of(node), type);
}
/** */
- private Scalar scalar(List<RexNode> nodes, RelDataType type) {
- return CACHE_1.computeIfAbsent(digest(nodes, type), k -> compile(nodes, type));
+ public Scalar scalar(List<RexNode> nodes, RelDataType type) {
+ return SCALAR_CACHE.computeIfAbsent(digest(nodes, type), k -> compile(nodes, type));
}
/** */
@@ -270,8 +312,10 @@ public class ExpressionFactory {
list.get(i))));
}
+ builder.add(outputValues_); // return out
+
MethodDeclaration declaration = Expressions.methodDecl(
- Modifier.PUBLIC, void.class, IgniteMethod.SCALAR_EXECUTE.method().getName(),
+ Modifier.PUBLIC, Object[].class, IgniteMethod.SCALAR_EXECUTE.method().getName(),
ImmutableList.of(context_, inputValues_, outputValues_), builder.toBlock());
return compile(Scalar.class, Expressions.toString(F.asList(declaration), "\n", false));
@@ -428,9 +472,7 @@ public class ExpressionFactory {
/** {@inheritDoc} */
@Override public boolean test(T r) {
- scalar.execute(ctx, (Object[]) r, out);
-
- return (Boolean) out[0];
+ return (Boolean) scalar.execute(ctx, (Object[]) r, out)[0];
}
}
@@ -458,10 +500,7 @@ public class ExpressionFactory {
/** {@inheritDoc} */
@Override public T apply(T r) {
- Object[] out = new Object[count];
- scalar.execute(ctx, (Object[]) r, out);
-
- return (T) out;
+ return (T) scalar.execute(ctx, (Object[]) r, new Object[count]);
}
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/RexToExpTranslator.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/RexToExpTranslator.java
index 1642be3..104bb50 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/RexToExpTranslator.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/RexToExpTranslator.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.query.calcite.exec.exp;
import java.util.List;
+import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexCorrelVariable;
import org.apache.calcite.rex.RexDynamicParam;
@@ -32,6 +33,8 @@ import org.apache.calcite.rex.RexRangeRef;
import org.apache.calcite.rex.RexSubQuery;
import org.apache.calcite.rex.RexTableInputRef;
import org.apache.calcite.rex.RexVisitor;
+import org.apache.calcite.sql.SqlSyntax;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AggCallExp;
import org.apache.ignite.internal.processors.query.calcite.exec.exp.type.DataType;
import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
@@ -67,6 +70,24 @@ public class RexToExpTranslator implements RexVisitor<Expression> {
return rex.accept(this);
}
+ /** */
+ public AggCallExp translate(AggregateCall call) {
+ DataType type = DataType.fromType(call.type);
+ SqlSyntax syntax = call.getAggregation().getSyntax();
+ String function = call.getAggregation().getName();
+ String name = call.name;
+ boolean distinct = call.isDistinct();
+ boolean approximate = call.isApproximate();
+ boolean ignoreNulls = call.ignoreNulls();
+ int[] args = call.getArgList().stream()
+ .mapToInt(Integer::intValue)
+ .toArray();
+ int filterArg = call.filterArg;
+
+ return new AggCallExp(type, syntax, function,
+ name, distinct, approximate, ignoreNulls, args, filterArg);
+ }
+
/** {@inheritDoc} */
@Override public Expression visitInputRef(RexInputRef rex) {
return new InputRef(DataType.fromType(rex.getType()), rex.getIndex());
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java
index 02ec14e..e96a9a7 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java
@@ -23,5 +23,5 @@ import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext
*
*/
public interface Scalar {
- void execute(ExecutionContext ctx, Object[] in, Object[] out);
+ Object[] execute(ExecutionContext ctx, Object[] in, Object[] out);
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulator.java
similarity index 63%
copy from modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java
copy to modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulator.java
index 02ec14e..6ccd661 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulator.java
@@ -15,13 +15,29 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.query.calcite.exec.exp;
+package org.apache.ignite.internal.processors.query.calcite.exec.exp.agg;
-import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import java.io.Serializable;
+import java.util.List;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
/**
*
*/
-public interface Scalar {
- void execute(ExecutionContext ctx, Object[] in, Object[] out);
+public interface Accumulator extends Serializable {
+ /** */
+ void add(Object... args);
+
+ /** */
+ void apply(Accumulator other);
+
+ /** */
+ Object end();
+
+ /** */
+ List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory);
+
+ /** */
+ RelDataType returnType(IgniteTypeFactory typeFactory);
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/AccumulatorWrapper.java
similarity index 80%
copy from modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java
copy to modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/AccumulatorWrapper.java
index 02ec14e..29ce10f5 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/Scalar.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/AccumulatorWrapper.java
@@ -15,13 +15,21 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.processors.query.calcite.exec.exp;
-
-import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+package org.apache.ignite.internal.processors.query.calcite.exec.exp.agg;
/**
*
*/
-public interface Scalar {
- void execute(ExecutionContext ctx, Object[] in, Object[] out);
+public interface AccumulatorWrapper {
+ /** */
+ void add(Object row);
+
+ /** */
+ Object end();
+
+ /** */
+ void apply(Accumulator accumulator);
+
+ /** */
+ Accumulator accumulator();
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulators.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulators.java
new file mode 100644
index 0000000..1e80ab2
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/Accumulators.java
@@ -0,0 +1,674 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.exec.exp.agg;
+
+import java.math.BigDecimal;
+import java.math.MathContext;
+import java.util.List;
+import java.util.function.Supplier;
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
+import org.apache.ignite.internal.util.typedef.F;
+
+import static org.apache.calcite.sql.type.SqlTypeName.BIGINT;
+import static org.apache.calcite.sql.type.SqlTypeName.DECIMAL;
+import static org.apache.calcite.sql.type.SqlTypeName.DOUBLE;
+import static org.apache.calcite.sql.type.SqlTypeName.INTEGER;
+
+/**
+ *
+ */
+public class Accumulators {
+ /** */
+ public static Supplier<Accumulator> accumulatorFactory(AggregateCall call) {
+ switch (call.getAggregation().getName()) {
+ case "COUNT":
+ return LongCount.FACTORY;
+ case "AVG":
+ return avgFactory(call);
+ case "SUM":
+ return sumFactory(call);
+ case "MIN":
+ return minFactory(call);
+ case "MAX":
+ return maxFactory(call);
+ default:
+ throw new AssertionError();
+ }
+ }
+
+ /** */
+ private static Supplier<Accumulator> avgFactory(AggregateCall call) {
+ switch (call.type.getSqlTypeName()) {
+ case BIGINT:
+ case DECIMAL:
+ return DecimalAvg.FACTORY;
+ case DOUBLE:
+ case REAL:
+ case FLOAT:
+ case INTEGER:
+ default:
+ return DoubleAvg.FACTORY;
+ }
+ }
+
+ /** */
+ private static Supplier<Accumulator> sumFactory(AggregateCall call) {
+ switch (call.type.getSqlTypeName()) {
+ case DOUBLE:
+ case REAL:
+ case FLOAT:
+ return DoubleSum.FACTORY;
+ case DECIMAL:
+ return DecimalSum.FACTORY;
+ case INTEGER:
+ return IntSum.FACTORY;
+ case BIGINT:
+ default:
+ return LongSum.FACTORY;
+ }
+ }
+
+ /** */
+ private static Supplier<Accumulator> minFactory(AggregateCall call) {
+ switch (call.type.getSqlTypeName()) {
+ case DOUBLE:
+ case REAL:
+ case FLOAT:
+ return DoubleMinMax.MIN_FACTORY;
+ case DECIMAL:
+ return DecimalMinMax.MIN_FACTORY;
+ case INTEGER:
+ return IntMinMax.MIN_FACTORY;
+ case BIGINT:
+ default:
+ return LongMinMax.MIN_FACTORY;
+ }
+ }
+
+ /** */
+ private static Supplier<Accumulator> maxFactory(AggregateCall call) {
+ switch (call.type.getSqlTypeName()) {
+ case DOUBLE:
+ case REAL:
+ case FLOAT:
+ return DoubleMinMax.MAX_FACTORY;
+ case DECIMAL:
+ return DecimalMinMax.MAX_FACTORY;
+ case INTEGER:
+ return IntMinMax.MAX_FACTORY;
+ case BIGINT:
+ default:
+ return LongMinMax.MAX_FACTORY;
+ }
+ }
+
+ /** */
+ public static class DecimalAvg implements Accumulator {
+ /** */
+ public static final Supplier<Accumulator> FACTORY = DecimalAvg::new;
+
+ /** */
+ private BigDecimal sum = BigDecimal.ZERO;
+
+ /** */
+ private BigDecimal cnt = BigDecimal.ZERO;
+
+ /** {@inheritDoc} */
+ @Override public void add(Object... args) {
+ BigDecimal in = (BigDecimal) args[0];
+
+ if (in == null)
+ return;
+
+ sum = sum.add(in);
+ cnt = cnt.add(BigDecimal.ONE);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator other) {
+ DecimalAvg other0 = (DecimalAvg) other;
+
+ sum = sum.add(other0.sum);
+ cnt = cnt.add(other0.cnt);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return cnt.compareTo(BigDecimal.ZERO) == 0 ? null : sum.divide(cnt, MathContext.DECIMAL64);
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
+ return F.asList(typeFactory.createTypeWithNullability(typeFactory.createSqlType(DECIMAL), true));
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelDataType returnType(IgniteTypeFactory typeFactory) {
+ return typeFactory.createTypeWithNullability(typeFactory.createSqlType(DECIMAL), true);
+ }
+ }
+
+ /** */
+ public static class DoubleAvg implements Accumulator {
+ /** */
+ public static final Supplier<Accumulator> FACTORY = DoubleAvg::new;
+
+ /** */
+ private double sum;
+
+ /** */
+ private long cnt;
+
+ /** {@inheritDoc} */
+ @Override public void add(Object... args) {
+ Double in = (Double) args[0];
+
+ if (in == null)
+ return;
+
+ sum += in;
+ cnt++;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator other) {
+ DoubleAvg other0 = (DoubleAvg) other;
+
+ sum += other0.sum;
+ cnt += other0.cnt;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return cnt > 0 ? sum / cnt : null;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
+ return F.asList(typeFactory.createTypeWithNullability(typeFactory.createSqlType(DOUBLE), true));
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelDataType returnType(IgniteTypeFactory typeFactory) {
+ return typeFactory.createTypeWithNullability(typeFactory.createSqlType(DOUBLE), true);
+ }
+ }
+
+ /** */
+ private static class LongCount implements Accumulator {
+ /** */
+ public static final Supplier<Accumulator> FACTORY = LongCount::new;
+
+ /** */
+ private long cnt;
+
+ /** {@inheritDoc} */
+ @Override public void add(Object... args) {
+ cnt++;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator other) {
+ LongCount other0 = (LongCount) other;
+ cnt += other0.cnt;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return cnt;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
+ return F.asList();
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelDataType returnType(IgniteTypeFactory typeFactory) {
+ return typeFactory.createSqlType(BIGINT);
+ }
+ }
+
+ /** */
+ private static class DoubleSum implements Accumulator {
+ /** */
+ public static final Supplier<Accumulator> FACTORY = DoubleSum::new;
+
+ /** */
+ private double sum;
+
+ /** */
+ private boolean empty = true;
+
+ /** {@inheritDoc} */
+ @Override public void add(Object... args) {
+ Double in = (Double) args[0];
+
+ if (in == null)
+ return;
+
+ empty = false;
+ sum += in;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator other) {
+ DoubleSum other0 = (DoubleSum) other;
+
+ if (other0.empty)
+ return;
+
+ empty = false;
+ sum += other0.sum;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return empty ? null : sum;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
+ return F.asList(typeFactory.createTypeWithNullability(typeFactory.createSqlType(DOUBLE), true));
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelDataType returnType(IgniteTypeFactory typeFactory) {
+ return typeFactory.createTypeWithNullability(typeFactory.createSqlType(DOUBLE), true);
+ }
+ }
+
+ /** */
+ private static class IntSum implements Accumulator {
+ /** */
+ public static final Supplier<Accumulator> FACTORY = IntSum::new;
+
+ /** */
+ private int sum;
+
+ /** */
+ private boolean empty = true;
+
+ /** {@inheritDoc} */
+ @Override public void add(Object... args) {
+ Integer in = (Integer) args[0];
+
+ if (in == null)
+ return;
+
+ empty = false;
+ sum += in;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator other) {
+ IntSum other0 = (IntSum) other;
+
+ if (other0.empty)
+ return;
+
+ empty = false;
+ sum += other0.sum;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return empty ? null : sum;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
+ return F.asList(typeFactory.createTypeWithNullability(typeFactory.createSqlType(INTEGER), true));
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelDataType returnType(IgniteTypeFactory typeFactory) {
+ return typeFactory.createTypeWithNullability(typeFactory.createSqlType(INTEGER), true);
+ }
+ }
+
+ /** */
+ private static class LongSum implements Accumulator {
+ /** */
+ public static final Supplier<Accumulator> FACTORY = LongSum::new;
+
+ /** */
+ private long sum;
+
+ /** */
+ private boolean empty = true;
+
+ /** {@inheritDoc} */
+ @Override public void add(Object... args) {
+ Long in = (Long) args[0];
+
+ if (in == null)
+ return;
+
+ empty = false;
+ sum += in;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator other) {
+ LongSum other0 = (LongSum) other;
+
+ if (other0.empty)
+ return;
+
+ empty = false;
+ sum += other0.sum;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return empty ? null : sum;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
+ return F.asList(typeFactory.createTypeWithNullability(typeFactory.createSqlType(BIGINT), true));
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelDataType returnType(IgniteTypeFactory typeFactory) {
+ return typeFactory.createTypeWithNullability(typeFactory.createSqlType(BIGINT), true);
+ }
+ }
+
+ /** */
+ private static class DecimalSum implements Accumulator {
+ /** */
+ public static final Supplier<Accumulator> FACTORY = DecimalSum::new;
+
+ /** */
+ private BigDecimal sum;
+
+ /** {@inheritDoc} */
+ @Override public void add(Object... args) {
+ BigDecimal in = (BigDecimal) args[0];
+
+ if (in == null)
+ return;
+
+ sum = sum == null ? in : sum.add(in);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator other) {
+ DecimalSum other0 = (DecimalSum) other;
+
+ if (other0.sum == null)
+ return;
+
+ sum = sum == null ? other0.sum : sum.add(other0.sum);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return sum;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
+ return F.asList(typeFactory.createTypeWithNullability(typeFactory.createSqlType(DECIMAL), true));
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelDataType returnType(IgniteTypeFactory typeFactory) {
+ return typeFactory.createTypeWithNullability(typeFactory.createSqlType(DECIMAL), true);
+ }
+ }
+
+ /** */
+ private static class DoubleMinMax implements Accumulator {
+ /** */
+ public static final Supplier<Accumulator> MIN_FACTORY = () -> new DoubleMinMax(true);
+
+ /** */
+ public static final Supplier<Accumulator> MAX_FACTORY = () -> new DoubleMinMax(false);
+
+ /** */
+ private final boolean min;
+
+ /** */
+ private double val;
+
+ /** */
+ private boolean empty = true;
+
+ /** */
+ private DoubleMinMax(boolean min) {
+ this.min = min;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void add(Object... args) {
+ Double in = (Double) args[0];
+
+ if (in == null)
+ return;
+
+ val = empty ? in : min ? Math.min(val, in) : Math.max(val, in);
+ empty = false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator other) {
+ DoubleMinMax other0 = (DoubleMinMax) other;
+
+ if (other0.empty)
+ return;
+
+ val = empty ? other0.val : min ? Math.min(val, other0.val) : Math.max(val, other0.val);
+ empty = false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return empty ? null : val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
+ return F.asList(typeFactory.createTypeWithNullability(typeFactory.createSqlType(DOUBLE), true));
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelDataType returnType(IgniteTypeFactory typeFactory) {
+ return typeFactory.createTypeWithNullability(typeFactory.createSqlType(DOUBLE), true);
+ }
+ }
+
+ /** */
+ private static class IntMinMax implements Accumulator {
+ /** */
+ public static final Supplier<Accumulator> MIN_FACTORY = () -> new IntMinMax(true);
+
+ /** */
+ public static final Supplier<Accumulator> MAX_FACTORY = () -> new IntMinMax(false);
+
+ /** */
+ private final boolean min;
+
+ /** */
+ private int val;
+
+ /** */
+ private boolean empty = true;
+
+ /** */
+ private IntMinMax(boolean min) {
+ this.min = min;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void add(Object... args) {
+ Integer in = (Integer) args[0];
+
+ if (in == null)
+ return;
+
+ val = empty ? in : min ? Math.min(val, in) : Math.max(val, in);
+ empty = false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator other) {
+ IntMinMax other0 = (IntMinMax) other;
+
+ if (other0.empty)
+ return;
+
+ val = empty ? other0.val : min ? Math.min(val, other0.val) : Math.max(val, other0.val);
+ empty = false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return empty ? null : val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
+ return F.asList(typeFactory.createTypeWithNullability(typeFactory.createSqlType(INTEGER), true));
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelDataType returnType(IgniteTypeFactory typeFactory) {
+ return typeFactory.createTypeWithNullability(typeFactory.createSqlType(INTEGER), true);
+ }
+ }
+
+ /** */
+ private static class LongMinMax implements Accumulator {
+ /** */
+ public static final Supplier<Accumulator> MIN_FACTORY = () -> new LongMinMax(true);
+
+ /** */
+ public static final Supplier<Accumulator> MAX_FACTORY = () -> new LongMinMax(false);
+
+ /** */
+ private final boolean min;
+
+ /** */
+ private long val;
+
+ /** */
+ private boolean empty = true;
+
+ /** */
+ private LongMinMax(boolean min) {
+ this.min = min;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void add(Object... args) {
+ Long in = (Long) args[0];
+
+ if (in == null)
+ return;
+
+ val = empty ? in : min ? Math.min(val, in) : Math.max(val, in);
+ empty = false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator other) {
+ LongMinMax other0 = (LongMinMax) other;
+
+ if (other0.empty)
+ return;
+
+ val = empty ? other0.val : min ? Math.min(val, other0.val) : Math.max(val, other0.val);
+ empty = false;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return empty ? null : val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
+ return F.asList(typeFactory.createTypeWithNullability(typeFactory.createSqlType(BIGINT), true));
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelDataType returnType(IgniteTypeFactory typeFactory) {
+ return typeFactory.createTypeWithNullability(typeFactory.createSqlType(BIGINT), true);
+ }
+ }
+
+ /** */
+ private static class DecimalMinMax implements Accumulator {
+ /** */
+ public static final Supplier<Accumulator> MIN_FACTORY = () -> new DecimalMinMax(true);
+
+ /** */
+ public static final Supplier<Accumulator> MAX_FACTORY = () -> new DecimalMinMax(false);
+
+ /** */
+ private final boolean min;
+
+ /** */
+ private BigDecimal val;
+
+ /** */
+ private DecimalMinMax(boolean min) {
+ this.min = min;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void add(Object... args) {
+ BigDecimal in = (BigDecimal) args[0];
+
+ if (in == null)
+ return;
+
+ val = val == null ? in : min ? val.min(in) : val.max(in);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator other) {
+ DecimalMinMax other0 = (DecimalMinMax) other;
+
+ if (other0.val == null)
+ return;
+
+ val = val == null ? other0.val : min ? val.min(other0.val) : val.max(other0.val);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<RelDataType> argumentTypes(IgniteTypeFactory typeFactory) {
+ return F.asList(typeFactory.createTypeWithNullability(typeFactory.createSqlType(DECIMAL), true));
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelDataType returnType(IgniteTypeFactory typeFactory) {
+ return typeFactory.createTypeWithNullability(typeFactory.createSqlType(DECIMAL), true);
+ }
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/AggCallExp.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/AggCallExp.java
new file mode 100644
index 0000000..c9227bc
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/AggCallExp.java
@@ -0,0 +1,178 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.exec.exp.agg;
+
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.List;
+import java.util.Objects;
+import org.apache.calcite.sql.SqlAggFunction;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlOperatorTable;
+import org.apache.calcite.sql.SqlSyntax;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.validate.SqlNameMatchers;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.type.DataType;
+import org.apache.ignite.internal.util.MutableSingletonList;
+import org.apache.ignite.internal.util.typedef.F;
+
+/**
+ *
+ */
+public class AggCallExp implements Externalizable {
+ /** */
+ private static final byte DISTINCT = 1;
+
+ /** */
+ private static final byte APPROXIMATE = 1 << 1;
+
+ /** */
+ private static final byte IGNORE_NULLS = 1 << 2;
+
+ /** */
+ private static final byte HAS_NAME = 1 << 3;
+
+ /** */
+ private DataType type;
+
+ /** */
+ private SqlSyntax syntax;
+
+ /** */
+ private String function;
+
+ /** */
+ private String name;
+
+ /** */
+ private int[] args;
+
+ /** */
+ private int filterArg;
+
+ /** */
+ private byte flags;
+
+ /** */
+ public AggCallExp(DataType type, SqlSyntax syntax, String function, String name,
+ boolean distinct, boolean approximate, boolean ignoreNulls, int[] args, int filterArg) {
+ this.type = type;
+ this.syntax = syntax;
+ this.function = function;
+ this.name = name;
+ this.args = args;
+ this.filterArg = filterArg;
+
+ byte flags = 0;
+
+ if (distinct)
+ flags |= DISTINCT;
+ if (approximate)
+ flags |= APPROXIMATE;
+ if (ignoreNulls)
+ flags |= IGNORE_NULLS;
+
+ this.flags = flags;
+ }
+
+ public AggCallExp() {
+ }
+
+ /** */
+ public DataType resultType() {
+ return type;
+ }
+
+ /** */
+ public SqlSyntax syntax() {
+ return syntax;
+ }
+
+ /** */
+ public String function() {
+ return function;
+ }
+
+ /** */
+ public SqlAggFunction sqlOperator(SqlOperatorTable opTable) {
+ List<SqlOperator> bag = new MutableSingletonList<>();
+
+ opTable.lookupOperatorOverloads(
+ new SqlIdentifier(function, SqlParserPos.ZERO), null,
+ syntax, bag, SqlNameMatchers.withCaseSensitive(true));
+
+ return (SqlAggFunction) Objects.requireNonNull(F.first(bag));
+ }
+
+ /** */
+ public String name() {
+ return name;
+ }
+
+ /** */
+ public boolean distinct() {
+ return (flags & DISTINCT) == DISTINCT;
+ }
+
+ /** */
+ public boolean approximate() {
+ return (flags & APPROXIMATE) == APPROXIMATE;
+ }
+
+ /** */
+ public boolean ignoreNulls() {
+ return (flags & IGNORE_NULLS) == IGNORE_NULLS;
+ }
+
+ /** */
+ public int[] args() {
+ return args;
+ }
+
+ /** */
+ public int filterArg() {
+ return filterArg;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeByte(flags);
+ out.writeByte(syntax.ordinal());
+ out.writeUTF(function);
+ if ((flags & HAS_NAME) == HAS_NAME)
+ out.writeUTF(name);
+ out.writeObject(type);
+ out.writeObject(args);
+ out.writeInt(filterArg);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ flags = in.readByte();
+ syntax = SqlSyntax.values()[in.readByte()];
+ function = in.readUTF();
+ if ((flags & HAS_NAME) == HAS_NAME)
+ name = in.readUTF();
+ type = (DataType) in.readObject();
+ args = (int[]) in.readObject();
+ filterArg = in.readInt();
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/GroupKey.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/GroupKey.java
new file mode 100644
index 0000000..b57e549
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/GroupKey.java
@@ -0,0 +1,91 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.exec.exp.agg;
+
+import java.util.Arrays;
+
+/**
+ *
+ */
+public class GroupKey {
+ /** */
+ private final Object[] fields;
+
+ /** */
+ public GroupKey(Object[] fields) {
+ this.fields = fields;
+ }
+
+ /** */
+ public Object field(int idx) {
+ return fields[idx];
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ GroupKey groupKey = (GroupKey) o;
+
+ return Arrays.equals(fields, groupKey.fields);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return Arrays.hashCode(fields);
+ }
+
+ /** */
+ public static Builder builder(int rowLen) {
+ return new Builder(rowLen);
+ }
+
+ /** */
+ public static class Builder {
+ /** */
+ private final Object[] fields;
+
+ /** */
+ private int idx;
+
+ /** */
+ private Builder(int rowLen) {
+ fields = new Object[rowLen];
+ }
+
+ /** */
+ public Builder add(Object val) {
+ if (idx == fields.length)
+ throw new IndexOutOfBoundsException();
+
+ fields[idx++] = val;
+
+ return this;
+ }
+
+ /** */
+ public GroupKey build() {
+ assert idx == fields.length;
+
+ return new GroupKey(fields);
+ }
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/WrappersFactoryImpl.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/WrappersFactoryImpl.java
new file mode 100644
index 0000000..c6e7687
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/WrappersFactoryImpl.java
@@ -0,0 +1,406 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.exec.exp.agg;
+
+import com.google.common.primitives.Primitives;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlOperatorTable;
+import org.apache.calcite.sql.validate.SqlConformance;
+import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.ExpressionFactory;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.Scalar;
+import org.apache.ignite.internal.processors.query.calcite.exec.rel.AggregateNode.AggregateType;
+import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+import org.apache.ignite.internal.util.typedef.F;
+
+/** */
+@SuppressWarnings({"rawtypes", "unchecked"})
+public class WrappersFactoryImpl implements Supplier<List<AccumulatorWrapper>> {
+ /** */
+ private static final Object[] EMPTY = new Object[0];
+
+ /** */
+ private final ExecutionContext ctx;
+
+ /** */
+ private final AggregateType type;
+
+ /** */
+ private final RowHandler handler;
+
+ /** */
+ private final RelDataType inputRowType;
+
+ /** */
+ private final List<WrapperPrototype> prototypes;
+
+ /** */
+ public WrappersFactoryImpl(ExecutionContext ctx, AggregateType type, RowHandler handler,
+ List<AggregateCall> aggCalls, RelDataType inputRowType) {
+ this.ctx = ctx;
+ this.type = type;
+ this.handler = handler;
+ this.inputRowType = inputRowType;
+
+ ArrayList<WrapperPrototype> prototypes = new ArrayList<>(aggCalls.size());
+
+ for (AggregateCall aggCall : aggCalls)
+ prototypes.add(new WrapperPrototype(aggCall));
+
+ this.prototypes = prototypes;
+ }
+
+ /** {@inheritDoc} */
+ @Override public List<AccumulatorWrapper> get() {
+ return Commons.transform(prototypes, WrapperPrototype::get);
+ }
+
+ /** */
+ private abstract class AbstractWrapper implements AccumulatorWrapper {
+ /** */
+ protected final Accumulator accumulator;
+
+ /** */
+ protected final Function<Object, Object> outAdapter;
+
+ /** */
+ private AbstractWrapper(Accumulator accumulator, Function<Object, Object> outAdapter) {
+ this.accumulator = accumulator;
+ this.outAdapter = outAdapter;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ assert type != AggregateType.MAP;
+
+ return outAdapter.apply(accumulator.end());
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator accumulator) {
+ assert type == AggregateType.REDUCE;
+
+ this.accumulator.apply(accumulator);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Accumulator accumulator() {
+ return accumulator;
+ }
+ }
+
+ /** */
+ private final class NoArgsWrapper extends AbstractWrapper {
+ /** */
+ private NoArgsWrapper(Accumulator accumulator, Function<Object, Object> outAdapter) {
+ super(accumulator, outAdapter);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void add(Object row) {
+ assert type != AggregateType.REDUCE;
+
+ accumulator.add(EMPTY);
+ }
+ }
+
+ /** */
+ private final class Wrapper extends AbstractWrapper {
+ /** */
+ private final List<Integer> args;
+
+ /** */
+ private final boolean ignoreNulls;
+
+ /** */
+ private final Function<Object[], Object[]> inAdapter;
+
+ /** */
+ private Wrapper(Accumulator accumulator, List<Integer> args, boolean ignoreNulls,
+ Function<Object[], Object[]> inAdapter, Function<Object, Object> outAdapter) {
+ super(accumulator, outAdapter);
+
+ this.args = args;
+ this.ignoreNulls = ignoreNulls;
+ this.inAdapter = inAdapter;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void add(Object row) {
+ Object[] args0 = new Object[args.size()];
+
+ for (int i = 0; i < args.size(); i++) {
+ Object val = handler.get(args.get(i), row);
+
+ if (ignoreNulls && val == null)
+ return;
+
+ args0[i] = val;
+ }
+
+ accumulator.add(inAdapter.apply(args0));
+ }
+ }
+
+ /** */
+ private final class FilteringWrapper implements AccumulatorWrapper {
+ /** */
+ private final int filterArg;
+
+ /** */
+ private final AccumulatorWrapper delegate;
+
+ /** */
+ private FilteringWrapper(int filterArg, AccumulatorWrapper delegate) {
+ this.filterArg = filterArg;
+ this.delegate = delegate;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void add(Object row) {
+ if (handler.get(filterArg, row) == Boolean.TRUE)
+ delegate.add(row);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object end() {
+ return delegate.end();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void apply(Accumulator accumulator) {
+ delegate.apply(accumulator);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Accumulator accumulator() {
+ return delegate.accumulator();
+ }
+ }
+
+ /** */
+ private static class InTypeAdapter implements Function<Object[], Object[]> {
+ /** */
+ private final ExecutionContext ctx;
+
+ /** */
+ private final Scalar scalar;
+
+ /** */
+ private InTypeAdapter(ExecutionContext ctx, Scalar scalar) {
+ this.ctx = ctx;
+ this.scalar = scalar;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object[] apply(Object[] in) {
+ return scalar.execute(ctx, in, new Object[in.length]);
+ }
+ }
+
+ /** */
+ private static class OutTypeAdapter implements Function<Object, Object> {
+ /** */
+ private final ExecutionContext ctx;
+
+ /** */
+ private final Scalar scalar;
+
+ /** */
+ private OutTypeAdapter(ExecutionContext ctx, Scalar scalar) {
+ this.ctx = ctx;
+ this.scalar = scalar;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object apply(Object val) {
+ return scalar.execute(ctx, new Object[] {val}, new Object[1])[0];
+ }
+ }
+
+ /** */
+ private final class WrapperPrototype implements Supplier<AccumulatorWrapper> {
+ /** */
+ private final AggregateCall call;
+
+ /** */
+ private Function<Object[], Object[]> inAdapter;
+
+ /** */
+ private Function<Object, Object> outAdapter;
+
+ /** */
+ private Supplier<Accumulator> accFactory;
+
+ /** */
+ private WrapperPrototype(AggregateCall call) {
+ this.call = call;
+ }
+
+ /** {@inheritDoc} */
+ @Override public AccumulatorWrapper get() {
+ Accumulator accumulator = accumulator();
+
+ if (type == AggregateType.REDUCE)
+ return new NoArgsWrapper(accumulator, outAdapter);
+
+ AccumulatorWrapper wrapper;
+
+ if (F.isEmpty(call.getArgList()))
+ wrapper = new NoArgsWrapper(accumulator, outAdapter);
+ else
+ wrapper = new Wrapper(accumulator, call.getArgList(), call.ignoreNulls(), inAdapter, outAdapter);
+
+ if (call.filterArg >= 0)
+ wrapper = new FilteringWrapper(call.filterArg, wrapper);
+
+ return wrapper;
+ }
+
+ /** */
+ public Accumulator accumulator() {
+ if (accFactory != null)
+ return accFactory.get();
+
+ // init factory and adapters
+ accFactory = Accumulators.accumulatorFactory(call);
+ Accumulator accumulator = accFactory.get();
+
+ IgniteTypeFactory typeFactory = ctx.parent().typeFactory();
+ SqlConformance conformance = ctx.parent().conformance();
+ SqlOperatorTable opTable = ctx.parent().opTable();
+
+ ExpressionFactory factory = new ExpressionFactory(typeFactory, conformance, opTable);
+
+ if (type != AggregateType.REDUCE && !F.isEmpty(call.getArgList()))
+ inAdapter = createInAdapter(accumulator, factory);
+
+ if (type != AggregateType.MAP)
+ outAdapter = createOutAdapter(accumulator, factory);
+
+ return accumulator;
+ }
+
+ /** */
+ public Function<Object[], Object[]> createInAdapter(Accumulator accumulator, ExpressionFactory factory) {
+ IgniteTypeFactory typeFactory = factory.typeFactory();
+
+ List<Integer> argList = call.getArgList();
+ List<RelDataType> argTypes = accumulator.argumentTypes(typeFactory);
+
+ assert !F.isEmpty(argTypes) && argTypes.size() == argList.size();
+
+ List<RelDataType> rowTypes = new ArrayList<>(argList.size());
+ List<RelDataTypeField> rowFields = inputRowType.getFieldList();
+
+ for (Integer arg : argList)
+ rowTypes.add(rowFields.get(arg).getType());
+
+ RexBuilder rexBuilder = factory.rexBuilder();
+ List<RexNode> rexNodes = new ArrayList<>(argList.size());
+
+ boolean shouldCast = false;
+
+ for (int i = 0; i < argList.size(); i++) {
+ RelDataType rowType = rowTypes.get(i);
+ RelDataType argType = argTypes.get(i);
+
+ RexNode rexNode = rexBuilder.makeInputRef(rowType, i);
+
+ if (shouldCast(typeFactory, rowType, argType)) {
+ rexNode = rexBuilder.makeCast(argType, rexNode);
+
+ shouldCast = true;
+ }
+
+ rexNodes.add(rexNode);
+ }
+
+ if (shouldCast) {
+ RelDataTypeFactory.Builder builder = new RelDataTypeFactory.Builder(typeFactory);
+
+ for (int i = 0; i < rowTypes.size(); i++)
+ builder.add("$EXP" + i, rowTypes.get(i));
+
+ return new InTypeAdapter(ctx, factory.scalar(rexNodes, builder.build()));
+ }
+
+ return Function.identity();
+ }
+
+ /** */
+ public Function<Object, Object> createOutAdapter(Accumulator accumulator, ExpressionFactory factory) {
+ IgniteTypeFactory typeFactory = factory.typeFactory();
+
+ RelDataType inType = accumulator.returnType(typeFactory);
+ RelDataType outType = call.getType();
+
+ assert !inType.isStruct();
+ assert !outType.isStruct();
+
+ if (shouldCast(typeFactory, inType, outType)) {
+ RexBuilder rexBuilder = factory.rexBuilder();
+
+ RelDataType rowType = new RelDataTypeFactory.Builder(typeFactory)
+ .add("$EXP0", inType).build();
+
+ Scalar scalar = factory.scalar(
+ rexBuilder.makeCast(outType,
+ rexBuilder.makeInputRef(inType, 0)), rowType);
+
+ return new OutTypeAdapter(ctx, scalar);
+ }
+
+ return Function.identity();
+ }
+
+ /** */
+ private boolean shouldCast(IgniteTypeFactory typeFactory, RelDataType inDataType, RelDataType outDataType) {
+ Type inType = typeFactory.getJavaClass(inDataType);
+
+ if (!(inType instanceof Class))
+ return false; // null or synthetic type, impossible to cast
+
+ Type outType = typeFactory.getJavaClass(outDataType);
+
+ if (!(outType instanceof Class))
+ return false; // null or synthetic type, impossible to cast
+
+ // since we can't use primitive types in rows
+ // we have to box both types before comparison
+ Class inBoxed = Primitives.wrap((Class) inType);
+ Class outBoxed = Primitives.wrap((Class) outType);
+
+ // the classes are expected to be basic types like
+ // Integer or Long, so, it's OK to compare by links
+ return inBoxed != outBoxed;
+ }
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AggregateNode.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AggregateNode.java
new file mode 100644
index 0000000..b65f7ca
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/AggregateNode.java
@@ -0,0 +1,360 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.exec.rel;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.Accumulator;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AccumulatorWrapper;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.GroupKey;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+import org.apache.ignite.internal.util.typedef.F;
+
+/**
+ *
+ */
+public class AggregateNode<T> extends AbstractNode<T> implements SingleNode<T>, Downstream<T> {
+ /** */
+ private final AggregateType type;
+
+ /** */
+ private final Supplier<List<AccumulatorWrapper>> wrappersFactory;
+
+ /** */
+ private final RowHandler<T> handler;
+
+ /** */
+ private final ImmutableBitSet groupSet;
+
+ /** */
+ private final List<Grouping> groupings;
+
+ /** */
+ private int requested;
+
+ /** */
+ private int waiting;
+
+ /** */
+ private boolean inLoop;
+
+ /**
+ * @param ctx Execution context.
+ */
+ public AggregateNode(ExecutionContext ctx, AggregateType type, List<ImmutableBitSet> groupSets,
+ Supplier<List<AccumulatorWrapper>> wrappersFactory, RowHandler<T> handler) {
+ super(ctx);
+
+ this.type = type;
+ this.wrappersFactory = wrappersFactory;
+ this.handler = handler;
+
+ ImmutableBitSet.Builder b = ImmutableBitSet.builder();
+
+ if (groupSets.size() > Byte.MAX_VALUE)
+ throw new IgniteException("Too many groups");
+
+ groupings = new ArrayList<>(groupSets.size());
+
+ for (byte i = 0; i < groupSets.size(); i++) {
+ ImmutableBitSet groupFields = groupSets.get(i);
+ groupings.add(new Grouping(i, groupFields));
+
+ b.addAll(groupFields);
+ }
+
+ groupSet = b.build();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void request(int rowsCount) {
+ checkThread();
+
+ assert !F.isEmpty(sources) && sources.size() == 1;
+ assert rowsCount > 0 && requested == 0;
+
+ requested = rowsCount;
+
+ if (waiting == -1 && !inLoop)
+ context().execute(this::flushFromBuffer);
+ else if (waiting == 0)
+ F.first(sources).request(waiting = IN_BUFFER_SIZE);
+ else
+ throw new AssertionError();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void push(T row) {
+ checkThread();
+
+ assert downstream != null;
+ assert waiting > 0;
+
+ waiting--;
+
+ try {
+ for (Grouping grouping : groupings)
+ grouping.add(row);
+
+ if (waiting == 0)
+ F.first(sources).request(waiting = IN_BUFFER_SIZE);
+ }
+ catch (Exception e) {
+ downstream.onError(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void end() {
+ checkThread();
+
+ assert downstream != null;
+ assert waiting > 0;
+
+ waiting = -1;
+
+ try {
+ flushFromBuffer();
+ }
+ catch (Exception e) {
+ downstream.onError(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onError(Throwable e) {
+ checkThread();
+
+ assert downstream != null;
+
+ downstream.onError(e);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected Downstream<T> requestDownstream(int idx) {
+ if (idx != 0)
+ throw new IndexOutOfBoundsException();
+
+ return this;
+ }
+
+ /** */
+ public void flushFromBuffer() {
+ assert waiting == -1;
+
+ inLoop = true;
+ try {
+ int processed = 0;
+
+ ArrayDeque<Grouping> groupingsQueue = groupingsQueue();
+
+ while (requested > 0 && !groupingsQueue.isEmpty()) {
+ Grouping grouping = groupingsQueue.peek();
+
+ int toSend = Math.min(requested, IN_BUFFER_SIZE - processed);
+
+ for (T row : grouping.getRows(toSend)) {
+ requested--;
+ downstream.push(row);
+
+ processed++;
+ }
+
+ if (processed >= IN_BUFFER_SIZE && requested > 0) {
+ // allow others to do their job
+ context().execute(this::flushFromBuffer);
+
+ return;
+ }
+
+ if (grouping.isEmpty())
+ groupingsQueue.remove();
+ }
+
+ if (requested > 0) {
+ downstream.end();
+ requested = 0;
+ }
+ }
+ finally {
+ inLoop = false;
+ }
+ }
+
+ /** */
+ private ArrayDeque<Grouping> groupingsQueue() {
+ ArrayDeque<Grouping> res = new ArrayDeque<>(groupings.size());
+
+ for (Grouping grouping : groupings) {
+ if (!grouping.isEmpty())
+ res.add(grouping);
+ }
+
+ return res;
+ }
+
+ /** */
+ public enum AggregateType {
+ MAP, REDUCE, SINGLE
+ }
+
+ /** */
+ private class Grouping {
+ /** */
+ private final byte groupId;
+
+ /** */
+ private final ImmutableBitSet groupFields;
+
+ /** */
+ private final Map<GroupKey, List<AccumulatorWrapper>> groups = new HashMap<>();
+
+ /** */
+ private Grouping(byte groupId, ImmutableBitSet groupFields) {
+ this.groupId = groupId;
+ this.groupFields = groupFields;
+ }
+
+ /** */
+ private void add(T row) {
+ if (type == AggregateType.REDUCE)
+ addOnReducer(row);
+ else
+ addOnMapper(row);
+ }
+
+ /**
+ * @param cnt Number of rows.
+ *
+ * @return Actually sent rows number.
+ */
+ private List<T> getRows(int cnt) {
+ if (F.isEmpty(groups))
+ return Collections.emptyList();
+ else if (type == AggregateType.MAP)
+ return getOnMapper(cnt);
+ else
+ return getOnReducer(cnt);
+ }
+
+ /** */
+ private void addOnMapper(T row) {
+ GroupKey.Builder b = GroupKey.builder(groupFields.cardinality());
+
+ for (Integer field : groupFields)
+ b.add(handler.get(field, row));
+
+ GroupKey groupKey = b.build();
+
+ List<AccumulatorWrapper> wrappers = groups.computeIfAbsent(groupKey, this::create);
+
+ for (AccumulatorWrapper wrapper : wrappers)
+ wrapper.add(row);
+ }
+
+ /** */
+ private void addOnReducer(T row) {
+ byte targetGroupId = handler.get(0, row);
+
+ if (targetGroupId != groupId)
+ return;
+
+ GroupKey groupKey = handler.get(1, row);
+
+ List<AccumulatorWrapper> wrappers = groups.computeIfAbsent(groupKey, this::create);
+ List<Accumulator> accums = handler.get(2, row);
+
+ for (int i = 0; i < wrappers.size(); i++) {
+ AccumulatorWrapper wrapper = wrappers.get(i);
+ Accumulator accum = accums.get(i);
+
+ wrapper.apply(accum);
+ }
+ }
+
+ /** */
+ private List<T> getOnMapper(int cnt) {
+ Iterator<Map.Entry<GroupKey, List<AccumulatorWrapper>>> it = groups.entrySet().iterator();
+
+ int amount = Math.min(cnt, groups.size());
+ List<T> res = new ArrayList<>(amount);
+
+ for (int i = 0; i < amount; i++) {
+ Map.Entry<GroupKey, List<AccumulatorWrapper>> entry = it.next();
+
+ GroupKey groupKey = entry.getKey();
+ List<Accumulator> accums = Commons.transform(entry.getValue(), AccumulatorWrapper::accumulator);
+
+ res.add(handler.create(groupId, groupKey, accums));
+ it.remove();
+ }
+
+ return res;
+ }
+
+ /** */
+ private List<T> getOnReducer(int cnt) {
+ Iterator<Map.Entry<GroupKey, List<AccumulatorWrapper>>> it = groups.entrySet().iterator();
+
+ int amount = Math.min(cnt, groups.size());
+ List<T> res = new ArrayList<>(amount);
+
+ for (int i = 0; i < amount; i++) {
+ Map.Entry<GroupKey, List<AccumulatorWrapper>> entry = it.next();
+
+ GroupKey groupKey = entry.getKey();
+ List<AccumulatorWrapper> wrappers = entry.getValue();
+
+ Object[] fields = new Object[groupSet.cardinality() + wrappers.size()];
+
+ int j = 0, k = 0;
+
+ for (Integer field : groupSet)
+ fields[j++] = groupFields.get(field) ? groupKey.field(k++) : null;
+
+ for (AccumulatorWrapper wrapper : wrappers)
+ fields[j++] = wrapper.end();
+
+ res.add(handler.create(fields));
+ it.remove();
+ }
+
+ return res;
+ }
+
+ /** */
+ private List<AccumulatorWrapper> create(GroupKey key) {
+ return wrappersFactory.get();
+ }
+
+ /** */
+ private boolean isEmpty() {
+ return groups.isEmpty();
+ }
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDerivedDistribution.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDerivedDistribution.java
index 69e7c90..82ca615 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDerivedDistribution.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDerivedDistribution.java
@@ -27,8 +27,8 @@ import org.apache.calcite.plan.volcano.AbstractConverter;
import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.plan.volcano.VolcanoPlanner;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.SingleRel;
import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.logical.LogicalAggregate;
import org.apache.calcite.rel.logical.LogicalFilter;
import org.apache.calcite.rel.logical.LogicalJoin;
import org.apache.calcite.rel.logical.LogicalProject;
@@ -107,20 +107,15 @@ public class IgniteMdDerivedDistribution implements MetadataHandler<DerivedDistr
* See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
*/
public List<IgniteDistribution> deriveDistributions(LogicalProject rel, RelMetadataQuery mq) {
+ HashSet<IgniteDistribution> res = new HashSet<>();
+
Mappings.TargetMapping mapping =
Project.getPartialMapping(rel.getInput().getRowType().getFieldCount(), rel.getProjects());
- return Commons.transform(_deriveDistributions(rel.getInput(), mq), i -> i.apply(mapping));
- }
-
- /**
- * See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
- */
- public List<IgniteDistribution> deriveDistributions(SingleRel rel, RelMetadataQuery mq) {
- if (rel instanceof IgniteRel)
- return deriveDistributions((IgniteRel)rel, mq);
+ for (IgniteDistribution inDistr : _deriveDistributions(rel.getInput(), mq))
+ res.add(inDistr.apply(mapping));
- return _deriveDistributions(rel.getInput(), mq);
+ return new ArrayList<>(res);
}
/**
@@ -148,7 +143,7 @@ public class IgniteMdDerivedDistribution implements MetadataHandler<DerivedDistr
if (!F.isEmpty(res))
return new ArrayList<>(res);
- return Collections.emptyList();
+ return F.asList(IgniteMdDistribution._distribution(rel, mq));
}
/**
@@ -168,9 +163,19 @@ public class IgniteMdDerivedDistribution implements MetadataHandler<DerivedDistr
/**
* See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
*/
+ public List<IgniteDistribution> deriveDistributions(LogicalAggregate rel, RelMetadataQuery mq) {
+ List<IgniteDistributions.Suggestion> suggestions = IgniteDistributions.suggestAggregate(
+ mq, rel.getInput(), rel.getGroupSet(), rel.getGroupSets());
+
+ return Commons.transform(suggestions, IgniteDistributions.Suggestion::out);
+ }
+
+ /**
+ * See {@link IgniteMdDerivedDistribution#deriveDistributions(RelNode, RelMetadataQuery)}
+ */
public List<IgniteDistribution> deriveDistributions(LogicalJoin rel, RelMetadataQuery mq) {
List<IgniteDistributions.BiSuggestion> suggestions = IgniteDistributions.suggestJoin(
- rel.getLeft(), rel.getRight(), rel.analyzeCondition(), rel.getJoinType());
+ mq, rel.getLeft(), rel.getRight(), rel.analyzeCondition(), rel.getJoinType());
return Commons.transform(suggestions, IgniteDistributions.BiSuggestion::out);
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDistribution.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDistribution.java
index e94a31e..5e7f7f2 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDistribution.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdDistribution.java
@@ -17,14 +17,11 @@
package org.apache.ignite.internal.processors.query.calcite.metadata;
-import com.google.common.collect.ImmutableList;
import java.util.List;
import org.apache.calcite.plan.hep.HepRelVertex;
import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.core.JoinInfo;
-import org.apache.calcite.rel.core.JoinRelType;
-import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.logical.LogicalAggregate;
import org.apache.calcite.rel.logical.LogicalExchange;
import org.apache.calcite.rel.logical.LogicalFilter;
import org.apache.calcite.rel.logical.LogicalJoin;
@@ -38,9 +35,6 @@ import org.apache.calcite.rel.metadata.MetadataHandler;
import org.apache.calcite.rel.metadata.ReflectiveRelMetadataProvider;
import org.apache.calcite.rel.metadata.RelMetadataProvider;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
-import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rex.RexLiteral;
-import org.apache.calcite.rex.RexNode;
import org.apache.calcite.util.BuiltInMethod;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import org.apache.ignite.internal.processors.query.calcite.schema.DistributedTable;
@@ -93,21 +87,39 @@ public class IgniteMdDistribution implements MetadataHandler<BuiltInMetadata.Dis
* See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
*/
public IgniteDistribution distribution(LogicalFilter rel, RelMetadataQuery mq) {
- return filter(mq, rel.getInput(), rel.getCondition());
+ return _distribution(rel.getInput(), mq);
}
/**
* See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
*/
public IgniteDistribution distribution(LogicalProject rel, RelMetadataQuery mq) {
- return project(mq, rel.getInput(), rel.getProjects());
+ return IgniteDistributions.project(mq, rel.getInput(), rel.getProjects());
}
/**
* See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
*/
public IgniteDistribution distribution(LogicalJoin rel, RelMetadataQuery mq) {
- return join(mq, rel.getLeft(), rel.getRight(), rel.analyzeCondition(), rel.getJoinType());
+ IgniteDistribution leftIn = _distribution(rel.getLeft(), mq);
+ IgniteDistribution rightIn = _distribution(rel.getRight(), mq);
+
+ List<IgniteDistributions.BiSuggestion> suggestions = IgniteDistributions.suggestJoin(
+ leftIn, rightIn, rel.analyzeCondition(), rel.getJoinType());
+
+ return F.first(suggestions).out();
+ }
+
+ /**
+ * See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
+ */
+ public IgniteDistribution distribution(LogicalAggregate rel, RelMetadataQuery mq) {
+ IgniteDistribution inDistr = _distribution(rel.getInput(), mq);
+
+ List<IgniteDistributions.Suggestion> suggestions =
+ IgniteDistributions.suggestAggregate(inDistr, rel.getGroupSet(), rel.getGroupSets());
+
+ return F.first(suggestions).out();
}
/**
@@ -128,7 +140,7 @@ public class IgniteMdDistribution implements MetadataHandler<BuiltInMetadata.Dis
* See {@link IgniteMdDistribution#distribution(RelNode, RelMetadataQuery)}
*/
public IgniteDistribution distribution(LogicalValues rel, RelMetadataQuery mq) {
- return values(rel.getRowType(), rel.getTuples());
+ return IgniteDistributions.broadcast();
}
/**
@@ -146,41 +158,6 @@ public class IgniteMdDistribution implements MetadataHandler<BuiltInMetadata.Dis
}
/**
- * @return Values relational node distribution.
- */
- public static IgniteDistribution values(RelDataType rowType, ImmutableList<ImmutableList<RexLiteral>> tuples) {
- return IgniteDistributions.broadcast();
- }
-
- /**
- * @return Project relational node distribution calculated on the basis of its input and projections.
- */
- public static IgniteDistribution project(RelMetadataQuery mq, RelNode input, List<? extends RexNode> projects) {
- return project(input.getRowType(), _distribution(input, mq), projects);
- }
-
- /**
- * @return Project relational node distribution calculated on the basis of its input distribution and projections.
- */
- public static IgniteDistribution project(RelDataType inType, IgniteDistribution inDistr, List<? extends RexNode> projects) {
- return inDistr.apply(Project.getPartialMapping(inType.getFieldCount(), projects));
- }
-
- /**
- * @return Filter relational node distribution calculated on the basis of its input.
- */
- public static IgniteDistribution filter(RelMetadataQuery mq, RelNode input, RexNode condition) {
- return _distribution(input, mq);
- }
-
- /**
- * @return Join relational node distribution calculated on the basis of its inputs and join information.
- */
- public static IgniteDistribution join(RelMetadataQuery mq, RelNode left, RelNode right, JoinInfo joinInfo, JoinRelType joinType) {
- return F.first(IgniteDistributions.suggestJoin(_distribution(left, mq), _distribution(right, mq), joinInfo, joinType)).out();
- }
-
- /**
* Distribution request entry point.
* @param rel Relational node.
* @param mq Metadata query instance.
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentInfo.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentInfo.java
index b672ed0..9488866 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentInfo.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/metadata/IgniteMdFragmentInfo.java
@@ -20,9 +20,9 @@ package org.apache.ignite.internal.processors.query.calcite.metadata;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptCost;
import org.apache.calcite.plan.volcano.RelSubset;
+import org.apache.calcite.rel.BiRel;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.SingleRel;
-import org.apache.calcite.rel.core.Join;
import org.apache.calcite.rel.metadata.MetadataDef;
import org.apache.calcite.rel.metadata.MetadataHandler;
import org.apache.calcite.rel.metadata.ReflectiveRelMetadataProvider;
@@ -74,15 +74,6 @@ public class IgniteMdFragmentInfo implements MetadataHandler<FragmentMetadata> {
/**
* See {@link IgniteMdFragmentInfo#fragmentInfo(RelNode, RelMetadataQuery)}
- *
- * Prunes involved partitions (hence nodes, involved in query execution) if possible.
- */
- public FragmentInfo fragmentInfo(IgniteFilter rel, RelMetadataQuery mq) {
- return _fragmentInfo(rel.getInput(), mq).prune(rel);
- }
-
- /**
- * See {@link IgniteMdFragmentInfo#fragmentInfo(RelNode, RelMetadataQuery)}
*/
public FragmentInfo fragmentInfo(SingleRel rel, RelMetadataQuery mq) {
return _fragmentInfo(rel.getInput(), mq);
@@ -97,7 +88,7 @@ public class IgniteMdFragmentInfo implements MetadataHandler<FragmentMetadata> {
* exchange. After the exchange is put into the fragment and the fragment is split into two ones, fragment meta
* information will be recalculated for all fragments.
*/
- public FragmentInfo fragmentInfo(Join rel, RelMetadataQuery mq) {
+ public FragmentInfo fragmentInfo(BiRel rel, RelMetadataQuery mq) {
FragmentInfo left = _fragmentInfo(rel.getLeft(), mq);
FragmentInfo right = _fragmentInfo(rel.getRight(), mq);
@@ -126,6 +117,15 @@ public class IgniteMdFragmentInfo implements MetadataHandler<FragmentMetadata> {
/**
* See {@link IgniteMdFragmentInfo#fragmentInfo(RelNode, RelMetadataQuery)}
+ *
+ * Prunes involved partitions (hence nodes, involved in query execution) if possible.
+ */
+ public FragmentInfo fragmentInfo(IgniteFilter rel, RelMetadataQuery mq) {
+ return _fragmentInfo(rel.getInput(), mq).prune(rel);
+ }
+
+ /**
+ * See {@link IgniteMdFragmentInfo#fragmentInfo(RelNode, RelMetadataQuery)}
*/
public FragmentInfo fragmentInfo(IgniteReceiver rel, RelMetadataQuery mq) {
return new FragmentInfo(rel.source());
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java
index 3e4de38..1edb424 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Cloner.java
@@ -23,11 +23,14 @@ import java.util.List;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.prepare.Prepare;
import org.apache.calcite.rel.RelNode;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteExchange;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteFilter;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteJoin;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteMapAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteProject;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReceiver;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReduceAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRelVisitor;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
@@ -126,7 +129,7 @@ class Cloner implements IgniteRelVisitor<IgniteRel> {
/** {@inheritDoc} */
@Override public IgniteRel visit(IgniteReceiver rel) {
- Fragment fragment = (Fragment) rel.source();
+ Fragment fragment = rel.source();
fragments.add(fragment = new Fragment(fragment.fragmentId(), visit(fragment.root())));
return new IgniteReceiver(cluster, rel.getTraitSet(), rel.getRowType(), fragment);
@@ -140,6 +143,30 @@ class Cloner implements IgniteRelVisitor<IgniteRel> {
}
/** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteAggregate rel) {
+ RelNode input = visit((IgniteRel) rel.getInput());
+
+ return new IgniteAggregate(cluster, rel.getTraitSet(), input,
+ rel.getGroupSet(), rel.getGroupSets(), rel.getAggCallList());
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteMapAggregate rel) {
+ RelNode input = visit((IgniteRel) rel.getInput());
+
+ return new IgniteMapAggregate(cluster, rel.getTraitSet(), input,
+ rel.getGroupSet(), rel.getGroupSets(), rel.getAggCallList());
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteReduceAggregate rel) {
+ RelNode input = visit((IgniteRel) rel.getInput());
+
+ return new IgniteReduceAggregate(cluster, rel.getTraitSet(), input,
+ rel.groupSet(), rel.groupSets(), rel.aggregateCalls(), rel.getRowType());
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteRel visit(IgniteRel rel) {
return rel.accept(this);
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/FragmentSplitter.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/FragmentSplitter.java
index a46242c2..aae90d9 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/FragmentSplitter.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/FragmentSplitter.java
@@ -21,22 +21,29 @@ import com.google.common.collect.ImmutableList;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Objects;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteExchange;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteFilter;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteJoin;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteMapAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteProject;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReceiver;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReduceAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRelVisitor;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableModify;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteTableScan;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteValues;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
+import org.apache.ignite.internal.util.typedef.internal.U;
/**
*
@@ -88,15 +95,76 @@ public class FragmentSplitter implements IgniteRelVisitor<IgniteRel> {
}
/** {@inheritDoc} */
- @Override public IgniteRel visit(IgniteFilter rel) {
+ @Override public IgniteRel visit(IgniteAggregate rel) {
boolean split = rel == cutPoint;
RelNode input = rel.getInput();
RelNode newInput = visit((IgniteRel) input);
if (input != newInput) {
+ IgniteDistribution newDistr = IgniteDistributions.aggregate(mq,
+ newInput, rel.getGroupSet(), rel.getGroupSets(), rel.getAggCallList());
+
+ assert newDistr != null;
+
RelTraitSet traits = rel.getTraitSet()
- .replace(IgniteMdDistribution.filter(mq, newInput, rel.getCondition()));
+ .replace(newDistr);
+
+ rel = (IgniteAggregate) rel.copy(traits, ImmutableList.of(newInput));
+ }
+
+ return split ? split(rel, rel.getTraitSet()) : rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteMapAggregate rel) {
+ // a split may happen on BiRel inputs merge. An map aggregate node cannot be a BiRel input.
+ assert rel != cutPoint;
+
+ RelNode input = rel.getInput();
+ RelNode newInput = visit((IgniteRel) input);
+
+ if (input == newInput)
+ return rel;
+
+ IgniteDistribution newDistr = IgniteDistributions.mapAggregate(mq,
+ newInput, rel.getGroupSet(), rel.getGroupSets(), rel.getAggCallList());
+
+ assert newDistr != null;
+
+ RelTraitSet traits = rel.getTraitSet()
+ .replace(newDistr);
+
+ return (IgniteRel) rel.copy(traits, ImmutableList.of(newInput));
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteReduceAggregate rel) {
+ // a split may happen on BiRel inputs merge. An reduce aggregate node doesn't have a
+ // physical mapping (because it always goes after receiver node), so, its merge with
+ // any input cannot cause the split.
+ assert rel != cutPoint;
+
+ if (U.assertionsEnabled()) {
+ RelNode input = rel.getInput();
+ RelNode newInput = visit((IgniteRel) input);
+
+ assert input == newInput;
+ }
+
+ return rel;
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteFilter rel) {
+ boolean split = rel == cutPoint;
+
+ RelNode input = rel.getInput();
+ IgniteRel newInput = visit((IgniteRel) input);
+
+ if (input != newInput) {
+ RelTraitSet traits = rel.getTraitSet()
+ .replace(newInput.distribution());
rel = (IgniteFilter) rel.copy(traits, ImmutableList.of(newInput));
}
@@ -113,7 +181,7 @@ public class FragmentSplitter implements IgniteRelVisitor<IgniteRel> {
if (input != newInput) {
RelTraitSet traits = rel.getTraitSet()
- .replace(IgniteMdDistribution.project(mq, newInput, rel.getProjects()));
+ .replace(IgniteDistributions.project(mq, newInput, rel.getProjects()));
rel = (IgniteProject) rel.copy(traits, ImmutableList.of(newInput));
}
@@ -131,10 +199,14 @@ public class FragmentSplitter implements IgniteRelVisitor<IgniteRel> {
RelNode newLeft = visit((IgniteRel) left);
RelNode newRight = visit((IgniteRel) right);
- // Join requires input distribution and produces its own one.
- // It cannot change the distribution on a child node change.
- if (left != newLeft || right != newRight)
+ if (left != newLeft || right != newRight) {
+ // Join requires input distribution and produces its own one.
+ // It cannot change the distribution on a child node change.
+ checkDistributionEqual(left, newLeft);
+ checkDistributionEqual(right, newRight);
+
rel = (IgniteJoin) rel.copy(rel.getTraitSet(), ImmutableList.of(newLeft, newRight));
+ }
return split ? split(rel, rel.getTraitSet()) : rel;
}
@@ -195,4 +267,15 @@ public class FragmentSplitter implements IgniteRelVisitor<IgniteRel> {
return new IgniteReceiver(cluster, traits, input.getRowType(), fragment);
}
+
+ /** */
+ private void checkDistributionEqual(RelNode rel1, RelNode rel2) {
+ if (!U.assertionsEnabled())
+ return;
+
+ IgniteDistribution distr1 = IgniteMdDistribution._distribution(rel1, mq);
+ IgniteDistribution distr2 = IgniteMdDistribution._distribution(rel2, mq);
+
+ assert Objects.equals(distr1, distr2) : "distr1=" + distr1 + "; distr2=" + distr2;
+ }
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java
index 558e6a9..29e8afe 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/IgniteSqlValidator.java
@@ -25,6 +25,7 @@ import org.apache.calcite.prepare.CalciteCatalogReader;
import org.apache.calcite.prepare.Prepare;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.sql.SqlAggFunction;
import org.apache.calcite.sql.SqlCall;
import org.apache.calcite.sql.SqlDelete;
import org.apache.calcite.sql.SqlIdentifier;
@@ -136,6 +137,13 @@ public class IgniteSqlValidator extends SqlValidatorImpl {
}
/** {@inheritDoc} */
+ @Override public void validateAggregateParams(SqlCall aggCall, SqlNode filter, SqlNodeList orderList, SqlValidatorScope scope) {
+ validateAggregateFunction(aggCall, (SqlAggFunction) aggCall.getOperator());
+
+ super.validateAggregateParams(aggCall, filter, orderList, scope);
+ }
+
+ /** {@inheritDoc} */
@Override protected void addToSelectList(List<SqlNode> list, Set<String> aliases,
List<Map.Entry<String, RelDataType>> fieldList, SqlNode exp, SelectScope scope, boolean includeSystemVars) {
if (includeSystemVars || exp.getKind() != SqlKind.IDENTIFIER || !isSystemFieldName(deriveAlias(exp, 0)))
@@ -148,6 +156,26 @@ public class IgniteSqlValidator extends SqlValidatorImpl {
}
/** */
+ private void validateAggregateFunction(SqlCall call, SqlAggFunction aggFunction) {
+ if (!SqlKind.AGGREGATE.contains(aggFunction.kind))
+ throw newValidationError(call,
+ IgniteResource.INSTANCE.unsupportedAggregationFunction(aggFunction.getName()));
+
+ switch (aggFunction.kind) {
+ case COUNT:
+ case SUM:
+ case AVG:
+ case MIN:
+ case MAX:
+
+ return;
+ default:
+ throw newValidationError(call,
+ IgniteResource.INSTANCE.unsupportedAggregationFunction(aggFunction.getName()));
+ }
+ }
+
+ /** */
private SqlNodeList inferColumnList(SqlInsert call) {
final SqlValidatorTable table = table(validatedNamespace(call, unknownType));
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java
index 9192bdc..551aa99 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/PlannerPhase.java
@@ -21,6 +21,8 @@ import org.apache.calcite.rel.rules.SubQueryRemoveRule;
import org.apache.calcite.tools.Program;
import org.apache.calcite.tools.RuleSet;
import org.apache.calcite.tools.RuleSets;
+import org.apache.ignite.internal.processors.query.calcite.rule.AggregateConverterRule;
+import org.apache.ignite.internal.processors.query.calcite.rule.AggregateTraitsPropagationRule;
import org.apache.ignite.internal.processors.query.calcite.rule.FilterConverterRule;
import org.apache.ignite.internal.processors.query.calcite.rule.FilterTraitsPropagationRule;
import org.apache.ignite.internal.processors.query.calcite.rule.JoinConverterRule;
@@ -61,6 +63,8 @@ public enum PlannerPhase {
/** {@inheritDoc} */
@Override public RuleSet getRules(PlanningContext ctx) {
return RuleSets.ofList(
+ AggregateConverterRule.INSTANCE,
+ AggregateTraitsPropagationRule.INSTANCE,
JoinConverterRule.INSTANCE,
JoinTraitsPropagationRule.INSTANCE,
ProjectConverterRule.INSTANCE,
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Splitter.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Splitter.java
index 00cac1a..cacaed6 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Splitter.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/prepare/Splitter.java
@@ -22,11 +22,14 @@ import java.util.Collections;
import java.util.List;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.rel.RelNode;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteExchange;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteFilter;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteJoin;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteMapAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteProject;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReceiver;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReduceAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRelVisitor;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
@@ -78,6 +81,21 @@ public class Splitter implements IgniteRelVisitor<IgniteRel> {
}
/** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteAggregate rel) {
+ return visitChildren(rel);
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteMapAggregate rel) {
+ return visitChildren(rel);
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteRel visit(IgniteReduceAggregate rel) {
+ return visitChildren(rel);
+ }
+
+ /** {@inheritDoc} */
@Override public IgniteRel visit(IgniteTableScan rel) {
return rel;
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteAggregate.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteAggregate.java
new file mode 100644
index 0000000..c4b507b
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteAggregate.java
@@ -0,0 +1,73 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.rel;
+
+import java.util.List;
+import org.apache.calcite.linq4j.Ord;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptCost;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Aggregate;
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.mapping.MappingType;
+import org.apache.calcite.util.mapping.Mappings;
+
+/**
+ *
+ */
+public class IgniteAggregate extends Aggregate implements IgniteRel {
+ /** {@inheritDoc} */
+ public IgniteAggregate(RelOptCluster cluster, RelTraitSet traitSet, RelNode input, ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
+ super(cluster, traitSet, input, groupSet, groupSets, aggCalls);
+ }
+
+ /** {@inheritDoc} */
+ @Override public Aggregate copy(RelTraitSet traitSet, RelNode input, ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
+ return new IgniteAggregate(getCluster(), traitSet, input, groupSet, groupSets, aggCalls);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> T accept(IgniteRelVisitor<T> visitor) {
+ return visitor.visit(this);
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
+ if (distribution().getType() == RelDistribution.Type.ANY)
+ return planner.getCostFactory().makeInfiniteCost();
+
+ return super.computeSelfCost(planner, mq);
+ }
+
+ /** */
+ public static Mappings.TargetMapping partialMapping(int inputFieldCount, int outputFieldCount, ImmutableBitSet groupSet) {
+ Mappings.TargetMapping mapping =
+ Mappings.create(MappingType.INVERSE_FUNCTION,
+ inputFieldCount, outputFieldCount);
+
+ for (Ord<Integer> group : Ord.zip(groupSet))
+ mapping.set(group.e, group.i);
+
+ return mapping;
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableScan.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteMapAggregate.java
similarity index 54%
copy from modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableScan.java
copy to modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteMapAggregate.java
index 37090db..39439c2 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableScan.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteMapAggregate.java
@@ -19,33 +19,34 @@ package org.apache.ignite.internal.processors.query.calcite.rel;
import java.util.List;
import org.apache.calcite.plan.RelOptCluster;
-import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.core.Aggregate;
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
/**
- * Relational operator that returns the contents of a table.
+ *
*/
-public class IgniteTableScan extends TableScan implements IgniteRel {
- /**
- * Creates a TableScan.
- *
- * @param cluster Cluster that this relational expression belongs to
- * @param traits Traits of this relational expression
- * @param table Table definition.
- */
- public IgniteTableScan(RelOptCluster cluster, RelTraitSet traits, RelOptTable table) {
- super(cluster, traits, table);
+public class IgniteMapAggregate extends Aggregate implements IgniteRel {
+ public IgniteMapAggregate(RelOptCluster cluster, RelTraitSet traitSet, RelNode input, ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
+ super(cluster, traitSet, input, groupSet, groupSets, aggCalls);
}
/** {@inheritDoc} */
- @Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
- return this;
+ @Override public Aggregate copy(RelTraitSet traitSet, RelNode input, ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
+ return new IgniteMapAggregate(getCluster(), traitSet, input, groupSet, groupSets, aggCalls);
}
/** {@inheritDoc} */
@Override public <T> T accept(IgniteRelVisitor<T> visitor) {
return visitor.visit(this);
}
+
+ /** {@inheritDoc} */
+ @Override protected RelDataType deriveRowType() {
+ return Commons.aggregationDataRowType(getCluster().getTypeFactory());
+ }
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReduceAggregate.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReduceAggregate.java
new file mode 100644
index 0000000..3595a9b
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteReduceAggregate.java
@@ -0,0 +1,105 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.rel;
+
+import java.util.List;
+import org.apache.calcite.linq4j.Ord;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelWriter;
+import org.apache.calcite.rel.SingleRel;
+import org.apache.calcite.rel.core.Aggregate.Group;
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.Util;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+
+/**
+ *
+ */
+public class IgniteReduceAggregate extends SingleRel implements IgniteRel {
+ /** */
+ private final ImmutableBitSet groupSet;
+
+ /** */
+ private final List<ImmutableBitSet> groupSets;
+
+ /** */
+ private final List<AggregateCall> aggCalls;
+
+ /***/
+ public IgniteReduceAggregate(RelOptCluster cluster, RelTraitSet traits, RelNode input, ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls, RelDataType rowType) {
+ super(cluster, traits, input);
+
+ assert rowType != null;
+ assert RelOptUtil.areRowTypesEqual(input.getRowType(),
+ Commons.aggregationDataRowType(getCluster().getTypeFactory()), true);
+ assert getInput().getTraitSet().contains(IgniteDistributions.single());
+
+ this.groupSet = groupSet;
+ this.groupSets = groupSets;
+ this.aggCalls = aggCalls;
+ this.rowType = rowType;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected RelDataType deriveRowType() {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+ return new IgniteReduceAggregate(getCluster(), traitSet, sole(inputs), groupSet, groupSets, aggCalls, rowType);
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> T accept(IgniteRelVisitor<T> visitor) {
+ return visitor.visit(this);
+ }
+
+ /** {@inheritDoc} */
+ @Override public RelWriter explainTerms(RelWriter pw) {
+ super.explainTerms(pw)
+ .item("group", groupSet)
+ .itemIf("groups", groupSets, Group.induce(groupSet, groupSets) != Group.SIMPLE)
+ .itemIf("aggs", aggCalls, pw.nest());
+
+ if (!pw.nest()) {
+ for (Ord<AggregateCall> ord : Ord.zip(aggCalls))
+ pw.item(Util.first(ord.e.name, "agg#" + ord.i), ord.e);
+ }
+
+ return pw;
+ }
+
+ public ImmutableBitSet groupSet() {
+ return groupSet;
+ }
+
+ public List<ImmutableBitSet> groupSets() {
+ return groupSets;
+ }
+
+ public List<AggregateCall> aggregateCalls() {
+ return aggCalls;
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRelVisitor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRelVisitor.java
index fc24278..94b2a34 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRelVisitor.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteRelVisitor.java
@@ -59,6 +59,21 @@ public interface IgniteRelVisitor<T> {
/**
* See {@link IgniteRelVisitor#visit(IgniteRel)}
*/
+ T visit(IgniteAggregate rel);
+
+ /**
+ * See {@link IgniteRelVisitor#visit(IgniteRel)}
+ */
+ T visit(IgniteMapAggregate rel);
+
+ /**
+ * See {@link IgniteRelVisitor#visit(IgniteRel)}
+ */
+ T visit(IgniteReduceAggregate rel);
+
+ /**
+ * See {@link IgniteRelVisitor#visit(IgniteRel)}
+ */
T visit(IgniteTableModify rel);
/**
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableScan.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableScan.java
index 37090db..3ba3d35 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableScan.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteTableScan.java
@@ -41,6 +41,8 @@ public class IgniteTableScan extends TableScan implements IgniteRel {
/** {@inheritDoc} */
@Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+ assert inputs.isEmpty();
+
return this;
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteValues.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteValues.java
index 4597193..f631f68 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteValues.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rel/IgniteValues.java
@@ -25,7 +25,6 @@ import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.Values;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexLiteral;
-import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
/**
*
@@ -52,10 +51,7 @@ public class IgniteValues extends Values implements IgniteRel {
@Override public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
assert inputs.isEmpty();
- RelTraitSet traits = getCluster().traitSetOf(IgniteConvention.INSTANCE)
- .replace(IgniteMdDistribution.values(rowType, tuples));
-
- return new IgniteValues(getCluster(), rowType, tuples, traits);
+ return this;
}
/** {@inheritDoc} */
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ValuesConverterRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AggregateConverterRule.java
similarity index 67%
copy from modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ValuesConverterRule.java
copy to modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AggregateConverterRule.java
index 5d59a58..6a7d2a9 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ValuesConverterRule.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AggregateConverterRule.java
@@ -21,35 +21,33 @@ import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.logical.LogicalValues;
-import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.logical.LogicalAggregate;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteConvention;
-import org.apache.ignite.internal.processors.query.calcite.rel.IgniteValues;
/**
*
*/
-public class ValuesConverterRule extends RelOptRule {
+public class AggregateConverterRule extends RelOptRule {
/** */
- public static final RelOptRule INSTANCE = new ValuesConverterRule();
+ public static final RelOptRule INSTANCE = new AggregateConverterRule();
- /**
- * Creates a ConverterRule.
- */
- protected ValuesConverterRule() {
- super(operand(LogicalValues.class, none()));
+ /** */
+ public AggregateConverterRule() {
+ super(operand(LogicalAggregate.class, any()));
}
+ /** {@inheritDoc} */
@Override public void onMatch(RelOptRuleCall call) {
- LogicalValues rel = call.rel(0);
+ LogicalAggregate rel = call.rel(0);
RelOptCluster cluster = rel.getCluster();
-
+ RelNode input = convert(rel.getInput(), IgniteConvention.INSTANCE);
RelTraitSet traits = rel.getTraitSet()
- .replace(IgniteConvention.INSTANCE)
- .replace(IgniteMdDistribution.values(rel.getRowType(), rel.getTuples()));
+ .replace(IgniteConvention.INSTANCE);
RuleUtils.transformTo(call,
- new IgniteValues(cluster, rel.getRowType(), rel.getTuples(), traits));
+ new IgniteAggregate(cluster, traits, input, rel.getGroupSet(), rel.getGroupSets(), rel.getAggCallList()));
}
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AggregateTraitsPropagationRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AggregateTraitsPropagationRule.java
new file mode 100644
index 0000000..8bd7a6c
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/AggregateTraitsPropagationRule.java
@@ -0,0 +1,94 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.rule;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteAggregate;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteMapAggregate;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReduceAggregate;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistribution;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
+
+/**
+ *
+ */
+public class AggregateTraitsPropagationRule extends RelOptRule {
+ /** */
+ public static final RelOptRule INSTANCE = new AggregateTraitsPropagationRule();
+
+ /** */
+ public AggregateTraitsPropagationRule() {
+ super(RuleUtils.traitPropagationOperand(IgniteAggregate.class));
+ }
+
+ /** */
+ @Override public void onMatch(RelOptRuleCall call) {
+ IgniteAggregate rel = call.rel(0);
+ RelNode input = call.rel(1);
+
+ RelOptCluster cluster = rel.getCluster();
+ RelMetadataQuery mq = cluster.getMetadataQuery();
+
+ ImmutableBitSet groupSet = rel.getGroupSet();
+ List<ImmutableBitSet> groupSets = rel.getGroupSets();
+ List<AggregateCall> aggCalls = rel.getAggCallList();
+
+ IgniteDistribution inDistr = IgniteMdDistribution._distribution(input, mq);
+
+ List<IgniteDistributions.Suggestion> suggestions =
+ IgniteDistributions.suggestAggregate(inDistr, groupSet, groupSets);
+
+ List<RelNode> newRels = new ArrayList<>(suggestions.size());
+
+ for (IgniteDistributions.Suggestion suggestion : suggestions) {
+ RelTraitSet traits = rel.getTraitSet().replace(suggestion.out());
+ RelNode input0 = RuleUtils.changeTraits(input, suggestion.in());
+
+ if (isMapReduce(suggestion)) {
+ RelTraitSet mapTraits = input0.getTraitSet()
+ .replace(IgniteDistributions.mapAggregate(mq, input0, groupSet, groupSets, aggCalls));
+
+ input0 = new IgniteMapAggregate(cluster, mapTraits, input0, groupSet, groupSets, aggCalls);
+ input0 = RuleUtils.changeTraits(input0, suggestion.out());
+
+ newRels.add(new IgniteReduceAggregate(cluster, traits, input0, groupSet, groupSets, aggCalls, rel.getRowType()));
+ }
+ else
+ newRels.add(new IgniteAggregate(cluster, traits, input0, groupSet, groupSets, aggCalls));
+ }
+
+ RuleUtils.transformTo(call, newRels);
+ }
+
+ /** */
+ private boolean isMapReduce(IgniteDistributions.Suggestion suggestion) {
+ return Objects.equals(suggestion.out(), IgniteDistributions.single())
+ && suggestion.in().satisfies(IgniteDistributions.random());
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/FilterTraitsPropagationRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/FilterTraitsPropagationRule.java
index 14d5a34..a15fbd4 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/FilterTraitsPropagationRule.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/FilterTraitsPropagationRule.java
@@ -21,7 +21,6 @@ import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
@@ -35,7 +34,7 @@ public class FilterTraitsPropagationRule extends RelOptRule {
public static final RelOptRule INSTANCE = new FilterTraitsPropagationRule();
public FilterTraitsPropagationRule() {
- super(operand(IgniteFilter.class, operand(RelSubset.class, any())));
+ super(RuleUtils.traitPropagationOperand(IgniteFilter.class));
}
/** {@inheritDoc} */
@@ -47,7 +46,7 @@ public class FilterTraitsPropagationRule extends RelOptRule {
RelMetadataQuery mq = cluster.getMetadataQuery();
RelTraitSet traits = rel.getTraitSet()
- .replace(IgniteMdDistribution.filter(mq, input, rel.getCondition()));
+ .replace(IgniteMdDistribution._distribution(input, mq));
RuleUtils.transformTo(call,
new IgniteFilter(cluster, traits, input, rel.getCondition()));
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/JoinTraitsPropagationRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/JoinTraitsPropagationRule.java
index e6476d5..359b078 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/JoinTraitsPropagationRule.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/JoinTraitsPropagationRule.java
@@ -24,10 +24,10 @@ import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.CorrelationId;
import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rex.RexNode;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteJoin;
import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
@@ -40,7 +40,7 @@ public class JoinTraitsPropagationRule extends RelOptRule {
public static final RelOptRule INSTANCE = new JoinTraitsPropagationRule();
public JoinTraitsPropagationRule() {
- super(operand(IgniteJoin.class, operand(RelSubset.class, any())));
+ super(RuleUtils.traitPropagationOperand(IgniteJoin.class));
}
/** {@inheritDoc} */
@@ -51,20 +51,22 @@ public class JoinTraitsPropagationRule extends RelOptRule {
RelNode right = rel.getRight();
RelOptCluster cluster = rel.getCluster();
+ RelMetadataQuery mq = call.getMetadataQuery();
+
RexNode condition = rel.getCondition();
Set<CorrelationId> variablesSet = rel.getVariablesSet();
JoinRelType joinType = rel.getJoinType();
- List<IgniteDistributions.BiSuggestion> suggests = IgniteDistributions.suggestJoin(
- left, right, rel.analyzeCondition(), joinType);
+ List<IgniteDistributions.BiSuggestion> suggestions = IgniteDistributions.suggestJoin(
+ mq, left, right, rel.analyzeCondition(), joinType);
- List<RelNode> newRels = new ArrayList<>(suggests.size());
+ List<RelNode> newRels = new ArrayList<>(suggestions.size());
- for (IgniteDistributions.BiSuggestion suggest : suggests) {
- RelTraitSet traits = rel.getTraitSet().replace(suggest.out());
+ for (IgniteDistributions.BiSuggestion suggestion : suggestions) {
+ RelTraitSet traits = rel.getTraitSet().replace(suggestion.out());
- RelNode left0 = RuleUtils.changeTraits(left, suggest.left());
- RelNode right0 = RuleUtils.changeTraits(right, suggest.right());
+ RelNode left0 = RuleUtils.changeTraits(left, suggestion.left());
+ RelNode right0 = RuleUtils.changeTraits(right, suggestion.right());
newRels.add(new IgniteJoin(cluster, traits, left0, right0,
condition, variablesSet, joinType));
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ProjectTraitsPropagationRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ProjectTraitsPropagationRule.java
index 2a2b491..da307af 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ProjectTraitsPropagationRule.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ProjectTraitsPropagationRule.java
@@ -21,11 +21,10 @@ import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
-import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteProject;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
/**
* Ignite Project converter.
@@ -38,7 +37,7 @@ public class ProjectTraitsPropagationRule extends RelOptRule {
* Creates a converter.
*/
public ProjectTraitsPropagationRule() {
- super(operand(IgniteProject.class, operand(RelSubset.class, any())));
+ super(RuleUtils.traitPropagationOperand(IgniteProject.class));
}
@Override public void onMatch(RelOptRuleCall call) {
@@ -49,7 +48,7 @@ public class ProjectTraitsPropagationRule extends RelOptRule {
RelMetadataQuery mq = cluster.getMetadataQuery();
RelTraitSet traits = rel.getTraitSet()
- .replace(IgniteMdDistribution.project(mq, input, rel.getProjects()));
+ .replace(IgniteDistributions.project(mq, input, rel.getProjects()));
RuleUtils.transformTo(call,
new IgniteProject(cluster, traits, input, rel.getProjects(), rel.getRowType()));
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RuleUtils.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RuleUtils.java
index 9bba103..1131ac2 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RuleUtils.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/RuleUtils.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Map;
import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptRuleOperand;
import org.apache.calcite.plan.RelTrait;
import org.apache.calcite.plan.RelTraitDef;
import org.apache.calcite.plan.RelTraitSet;
@@ -30,15 +31,25 @@ import org.apache.calcite.plan.hep.HepRelVertex;
import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.plan.volcano.VolcanoPlanner;
import org.apache.calcite.rel.RelNode;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
import org.apache.ignite.internal.util.typedef.F;
import org.jetbrains.annotations.NotNull;
+import static org.apache.calcite.plan.RelOptRule.any;
+import static org.apache.calcite.plan.RelOptRule.operand;
+import static org.apache.calcite.plan.RelOptRule.some;
+
/**
*
*/
public class RuleUtils {
/** */
+ public static RelOptRuleOperand traitPropagationOperand(Class<? extends RelNode> clazz) {
+ return operand(clazz, IgniteDistributions.any(), some(operand(RelSubset.class, any())));
+ }
+
+ /** */
public static RelNode convert(RelNode rel, @NotNull RelTrait toTrait) {
RelTraitSet toTraits = rel.getTraitSet().replace(toTrait);
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ValuesConverterRule.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ValuesConverterRule.java
index 5d59a58..f3726df 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ValuesConverterRule.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/rule/ValuesConverterRule.java
@@ -22,9 +22,9 @@ import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.logical.LogicalValues;
-import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteConvention;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteValues;
+import org.apache.ignite.internal.processors.query.calcite.trait.IgniteDistributions;
/**
*
@@ -47,7 +47,7 @@ public class ValuesConverterRule extends RelOptRule {
RelTraitSet traits = rel.getTraitSet()
.replace(IgniteConvention.INSTANCE)
- .replace(IgniteMdDistribution.values(rel.getRowType(), rel.getTuples()));
+ .replace(IgniteDistributions.values(rel.getRowType(), rel.getTuples()));
RuleUtils.transformTo(call,
new IgniteValues(cluster, rel.getRowType(), rel.getTuples(), traits));
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/AggregatePhysicalRel.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/AggregatePhysicalRel.java
new file mode 100644
index 0000000..ef0e9ed
--- /dev/null
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/AggregatePhysicalRel.java
@@ -0,0 +1,108 @@
+/*
+ * 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.ignite.internal.processors.query.calcite.serialize;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.List;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AggCallExp;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.type.DataType;
+
+/**
+ *
+ */
+public class AggregatePhysicalRel implements PhysicalRel {
+ /** */
+ public static final byte SINGLE = 0;
+
+ /** */
+ public static final byte MAP = 1;
+
+ /** */
+ public static final byte REDUCE = 2;
+
+ /** */
+ private byte type;
+
+ /** */
+ private List<ImmutableBitSet> groupSets;
+
+ /** */
+ private List<AggCallExp> calls;
+
+ /** */
+ private PhysicalRel input;
+
+ /** */
+ private DataType inputRowType;
+
+ public AggregatePhysicalRel(byte type, List<ImmutableBitSet> groupSets, List<AggCallExp> calls, PhysicalRel input, DataType inputRowType) {
+ this.type = type;
+ this.groupSets = groupSets;
+ this.calls = calls;
+ this.input = input;
+ this.inputRowType = inputRowType;
+ }
+
+ public AggregatePhysicalRel() {
+ }
+
+ public byte type() {
+ return type;
+ }
+
+ public List<ImmutableBitSet> groupSets() {
+ return groupSets;
+ }
+
+ public List<AggCallExp> calls() {
+ return calls;
+ }
+
+ public PhysicalRel input() {
+ return input;
+ }
+
+ public DataType inputRowType() {
+ return inputRowType;
+ }
+
+ @Override public <T> T accept(PhysicalRelVisitor<T> visitor) {
+ return visitor.visit(this);
+ }
+
+ @Override public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeByte(type);
+ out.writeObject(groupSets);
+ out.writeObject(calls);
+ out.writeObject(input);
+ if (type != REDUCE)
+ out.writeObject(inputRowType);
+ }
+
+ @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ type = in.readByte();
+ groupSets = (List<ImmutableBitSet>) in.readObject();
+ calls = (List<AggCallExp>) in.readObject();
+ input = (PhysicalRel) in.readObject();
+ if (type != REDUCE)
+ inputRowType = (DataType) in.readObject();
+ }
+}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/PhysicalRelVisitor.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/PhysicalRelVisitor.java
index ca12805..5bbbd0e 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/PhysicalRelVisitor.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/PhysicalRelVisitor.java
@@ -59,6 +59,11 @@ public interface PhysicalRelVisitor<T> {
/**
* See {@link PhysicalRelVisitor#visit(PhysicalRel)}
*/
+ T visit(AggregatePhysicalRel rel);
+
+ /**
+ * See {@link PhysicalRelVisitor#visit(PhysicalRel)}
+ */
T visit(TableModifyPhysicalRel rel);
/**
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/RelToPhysicalConverter.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/RelToPhysicalConverter.java
index a5effbf..d73a744 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/RelToPhysicalConverter.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/serialize/RelToPhysicalConverter.java
@@ -20,16 +20,21 @@ package org.apache.ignite.internal.processors.query.calcite.serialize;
import com.google.common.collect.ImmutableList;
import java.util.List;
import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.ImmutableIntList;
import org.apache.ignite.internal.processors.query.calcite.exec.exp.Expression;
import org.apache.ignite.internal.processors.query.calcite.exec.exp.RexToExpTranslator;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AggCallExp;
import org.apache.ignite.internal.processors.query.calcite.exec.exp.type.DataType;
import org.apache.ignite.internal.processors.query.calcite.metadata.NodesMapping;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteExchange;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteFilter;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteJoin;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteMapAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteProject;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReceiver;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteReduceAggregate;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRel;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteRelVisitor;
import org.apache.ignite.internal.processors.query.calcite.rel.IgniteSender;
@@ -94,6 +99,41 @@ public class RelToPhysicalConverter implements IgniteRelVisitor<PhysicalRel> {
}
/** {@inheritDoc} */
+ @Override public PhysicalRel visit(IgniteAggregate rel) {
+ byte type = AggregatePhysicalRel.SINGLE;
+ List<ImmutableBitSet> groupSets = rel.getGroupSets();
+ List<AggCallExp> calls = Commons.transform(rel.getAggCallList(), rexTranslator::translate);
+ PhysicalRel input = visit((IgniteRel) rel.getInput());
+ DataType inputRowType = DataType.fromType(rel.getInput().getRowType());
+
+ return new AggregatePhysicalRel(type, groupSets, calls, input, inputRowType);
+ }
+
+ /** {@inheritDoc} */
+ @Override public PhysicalRel visit(IgniteMapAggregate rel) {
+ byte type = AggregatePhysicalRel.MAP;
+ List<ImmutableBitSet> groupSets = rel.getGroupSets();
+ List<AggCallExp> calls = Commons.transform(
+ rel.getAggCallList(), rexTranslator::translate);
+ PhysicalRel input = visit((IgniteRel) rel.getInput());
+ DataType inputRowType = DataType.fromType(rel.getInput().getRowType());
+
+ return new AggregatePhysicalRel(type, groupSets, calls, input, inputRowType);
+ }
+
+ /** {@inheritDoc} */
+ @Override public PhysicalRel visit(IgniteReduceAggregate rel) {
+ byte type = AggregatePhysicalRel.REDUCE;
+ List<ImmutableBitSet> groupSets = rel.groupSets();
+ List<AggCallExp> calls = Commons.transform(
+ rel.aggregateCalls(), rexTranslator::translate);
+ PhysicalRel input = visit((IgniteRel) rel.getInput());
+ DataType inputRowType = null;
+
+ return new AggregatePhysicalRel(type, groupSets, calls, input, inputRowType);
+ }
+
+ /** {@inheritDoc} */
@Override public PhysicalRel visit(IgniteTableScan rel) {
return new TableScanPhysicalRel(rel.getTable().getQualifiedName());
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributions.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributions.java
index 2385a19..4c0e492 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributions.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/trait/IgniteDistributions.java
@@ -17,22 +17,32 @@
package org.apache.ignite.internal.processors.query.calcite.trait;
+import com.google.common.collect.ImmutableList;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
-import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import org.apache.calcite.plan.RelTraitDef;
import org.apache.calcite.rel.RelDistribution;
import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Aggregate;
+import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.core.JoinInfo;
import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.ImmutableIntList;
import org.apache.calcite.util.mapping.Mappings;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDerivedDistribution;
+import org.apache.ignite.internal.processors.query.calcite.metadata.IgniteMdDistribution;
+import org.apache.ignite.internal.processors.query.calcite.rel.IgniteAggregate;
import org.apache.ignite.internal.processors.query.calcite.trait.DistributionFunction.AffinityDistribution;
import org.apache.ignite.internal.processors.query.calcite.trait.DistributionFunction.AnyDistribution;
import org.apache.ignite.internal.processors.query.calcite.trait.DistributionFunction.BroadcastDistribution;
@@ -137,21 +147,86 @@ public class IgniteDistributions {
}
/**
+ * Suggests possible aggregate distributions.
+ *
+ * @param inDistr Input node distribution.
+ * @param groupSet Aggregate group set.
+ * @param groupSets Aggregate group sets.
+ * @return Array of possible distributions, sorted by their efficiency (cheaper first).
+ */
+ public static List<Suggestion> suggestAggregate(IgniteDistribution inDistr, ImmutableBitSet groupSet,
+ List<ImmutableBitSet> groupSets) {
+ return sorted(suggestAggregate0(new HashMap<>(), inDistr, groupSet, groupSets));
+ }
+
+ /**
+ * Suggests possible aggregate distributions.
+ *
+ * @param mq Metadata query.
+ * @param input Input node.
+ * @param groupSet Aggregate group set.
+ * @param groupSets Aggregate group sets.
+ * @return Array of possible distributions, sorted by their efficiency (cheaper first).
+ */
+ public static List<Suggestion> suggestAggregate(RelMetadataQuery mq, RelNode input, ImmutableBitSet groupSet,
+ List<ImmutableBitSet> groupSets) {
+ List<IgniteDistribution> inDistrs = IgniteMdDerivedDistribution._deriveDistributions(input, mq);
+
+ Map<Suggestion, Integer> suggestions = new HashMap<>();
+
+ for (IgniteDistribution inDistr : inDistrs)
+ suggestions = suggestAggregate0(suggestions, inDistr, groupSet, groupSets);
+
+ return sorted(suggestions);
+ }
+
+ /**
+ * Suggests possible aggregate distributions based on next principles:
+ * <li>Simple aggregate may be done in a distributed form in case its input is hash distributed by aggregate group keys</li>
+ * <li>Any aggregate may be done without map-reduce processing in case its input is singleton or broadcast distributed</li>
+ * <li>Any aggregate may be done in map-reduce style in case its input is hash or random distributed</li>
+ */
+ private static Map<Suggestion,Integer> suggestAggregate0(Map<Suggestion, Integer> dst, IgniteDistribution inDistr,
+ ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets) {
+
+ IgniteDistribution out, in;
+
+ if (simpleAggregate(groupSet, groupSets)) {
+ // re-hash by group keys
+ in = hash(groupSet.asList());
+ out = hash(ImmutableIntList.range(0, groupSet.cardinality()));
+ dst = add(dst, out, inDistr, in);
+ }
+
+ if (inDistr.satisfies(random()))
+ // map-reduce case
+ dst = add(dst, new Suggestion(single(), inDistr), BEST_CNT > 0 ? INTS[1] : INTS[0]);
+
+ out = broadcast(); in = broadcast();
+ dst = add(dst, out, inDistr, in);
+
+ out = single(); in = single();
+ dst = add(dst, out, inDistr, in);
+
+ return dst;
+ }
+
+ /**
* Suggests possible join distributions.
*
+ * @param mq Metadata query.
* @param left Left node.
* @param right Right node.
* @param joinInfo Join info.
* @param joinType Join type.
* @return Array of possible distributions, sorted by their efficiency (cheaper first).
*/
- public static List<BiSuggestion> suggestJoin(RelNode left, RelNode right, JoinInfo joinInfo, JoinRelType joinType) {
- RelMetadataQuery mq = left.getCluster().getMetadataQuery();
-
+ public static List<BiSuggestion> suggestJoin(RelMetadataQuery mq, RelNode left, RelNode right, JoinInfo joinInfo,
+ JoinRelType joinType) {
List<IgniteDistribution> leftIn = IgniteMdDerivedDistribution._deriveDistributions(left, mq);
List<IgniteDistribution> rightIn = IgniteMdDerivedDistribution._deriveDistributions(right, mq);
- Map<BiSuggestion, Integer> suggestions = new LinkedHashMap<>();
+ Map<BiSuggestion, Integer> suggestions = new HashMap<>();
for (IgniteDistribution leftIn0 : leftIn)
for (IgniteDistribution rightIn0 : rightIn)
@@ -171,45 +246,37 @@ public class IgniteDistributions {
*/
public static List<BiSuggestion> suggestJoin(IgniteDistribution leftIn, IgniteDistribution rightIn,
JoinInfo joinInfo, JoinRelType joinType) {
- return sorted(suggestJoin0(new LinkedHashMap<>(), leftIn, rightIn, joinInfo, joinType));
+ return sorted(suggestJoin0(new HashMap<>(), leftIn, rightIn, joinInfo, joinType));
}
/**
* Suggests possible join distributions using inputs distributions and join info
* based on next distributions table:
- *
- * ===============INNER JOIN==============
- * hash + hash = hash
- * broadcast + hash = hash
- * hash + broadcast = hash
- * broadcast + broadcast = broadcast
- * single + single = single
- *
- * ===============LEFT JOIN===============
- * hash + hash = hash
- * hash + broadcast = hash
- * broadcast + broadcast = broadcast
- * single + single = single
- *
- * ===============RIGHT JOIN==============
- * hash + hash = hash
- * broadcast + hash = hash
- * broadcast + broadcast = broadcast
- * single + single = single
- *
- * ===========FULL JOIN/CROSS JOIN========
- * broadcast + broadcast = broadcast
- * single + single = single
- *
- * others require redistribution.
- *
- * @param leftIn Left distribution.
- * @param rightIn Right distribution.
- * @param joinInfo Join info.
- * @param joinType Join type.
+ * <p/><table>
+ * <tr><th>===============INNER JOIN==============</th></tr>
+ * <tr><td>hash + hash = hash</td></tr>
+ * <tr><td>broadcast + hash = hash</td></tr>
+ * <tr><td>hash + broadcast = hash</td></tr>
+ * <tr><td>broadcast + broadcast = broadcast</td></tr>
+ * <tr><td>single + single = single</td></tr>
+ * <tr><th>===============LEFT JOIN===============</th></tr>
+ * <tr><td>hash + hash = hash</td></tr>
+ * <tr><td>hash + broadcast = hash</td></tr>
+ * <tr><td>broadcast + broadcast = broadcast</td></tr>
+ * <tr><td>single + single = single</td></tr>
+ * <tr><th>===============RIGHT JOIN==============</th></tr>
+ * <tr><td>hash + hash = hash</td></tr>
+ * <tr><td>broadcast + hash = hash</td></tr>
+ * <tr><td>broadcast + broadcast = broadcast</td></tr>
+ * <tr><td>single + single = single</td></tr>
+ * <tr><th>===========FULL JOIN/CROSS JOIN========</th></tr>
+ * <tr><td>broadcast + broadcast = broadcast</td></tr>
+ * <tr><td>single + single = single</td></tr>
+ * </table>
+ * <p/>others require redistribution.
*/
- private static Map<BiSuggestion, Integer> suggestJoin0(Map<BiSuggestion, Integer> dst, IgniteDistribution leftIn, IgniteDistribution rightIn,
- JoinInfo joinInfo, JoinRelType joinType) {
+ private static Map<BiSuggestion, Integer> suggestJoin0(Map<BiSuggestion, Integer> dst, IgniteDistribution leftIn,
+ IgniteDistribution rightIn, JoinInfo joinInfo, JoinRelType joinType) {
IgniteDistribution out, left, right;
if (joinType == LEFT || joinType == RIGHT || (joinType == INNER && !F.isEmpty(joinInfo.pairs()))) {
@@ -253,8 +320,23 @@ public class IgniteDistributions {
}
/** */
- private static Map<BiSuggestion, Integer> add(Map<BiSuggestion, Integer> dst, IgniteDistribution out, IgniteDistribution left, IgniteDistribution right,
- IgniteDistribution newLeft, IgniteDistribution newRight) {
+ private static Map<Suggestion, Integer> add(Map<Suggestion, Integer> dst, IgniteDistribution out,
+ IgniteDistribution in, IgniteDistribution newIn) {
+ if (BEST_CNT > 0) {
+ int exch = 0;
+
+ if (!in.satisfies(newIn))
+ exch++;
+
+ return add(dst, new Suggestion(out, newIn), INTS[exch]);
+ }
+ else
+ return add(dst, new Suggestion(out, newIn), INTS[0]);
+ }
+
+ /** */
+ private static Map<BiSuggestion, Integer> add(Map<BiSuggestion, Integer> dst, IgniteDistribution out,
+ IgniteDistribution left, IgniteDistribution right, IgniteDistribution newLeft, IgniteDistribution newRight) {
if (BEST_CNT > 0) {
int exch = 0;
@@ -271,9 +353,9 @@ public class IgniteDistributions {
}
/** */
- private static Map<BiSuggestion, Integer> add(Map<BiSuggestion, Integer> dst, BiSuggestion suggest, Integer exchCnt) {
+ private static <T> Map<T, Integer> add(Map<T, Integer> dst, T suggest, Integer exchCnt) {
if (dst == null)
- dst = new LinkedHashMap<>();
+ dst = new HashMap<>();
dst.merge(suggest, exchCnt, IgniteDistributions::min);
@@ -281,9 +363,9 @@ public class IgniteDistributions {
}
/** */
- private static List<BiSuggestion> sorted(Map<BiSuggestion, Integer> src) {
+ private static <T> List<T> sorted(Map<T, Integer> src) {
if (BEST_CNT > 0) {
- List<Map.Entry<BiSuggestion, Integer>> entries = new ArrayList<>(src.entrySet());
+ List<Map.Entry<T, Integer>> entries = new ArrayList<>(src.entrySet());
entries.sort(Map.Entry.comparingByValue());
@@ -337,12 +419,160 @@ public class IgniteDistributions {
}
/**
+ * @return Values relational node distribution.
+ */
+ public static IgniteDistribution values(RelDataType rowType, ImmutableList<ImmutableList<RexLiteral>> tuples) {
+ return broadcast();
+ }
+
+ /**
+ * @return Project relational node distribution calculated on the basis of its input and projections.
+ */
+ public static IgniteDistribution project(RelMetadataQuery mq, RelNode input, List<? extends RexNode> projects) {
+ IgniteDistribution inDistr = IgniteMdDistribution._distribution(input, mq);
+ Mappings.TargetMapping mapping = Project.getPartialMapping(input.getRowType().getFieldCount(), projects);
+
+ return inDistr.apply(mapping);
+ }
+
+ /**
+ * @return Single aggregate relational node distribution calculated on the basis of its input and groupingSets.
+ * <b>Note</b> that the method returns {@code null} in case the given input cannot be processed by a single aggregate.
+ */
+ public static IgniteDistribution aggregate(RelMetadataQuery mq, RelNode input, ImmutableBitSet groupSet,
+ List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
+ RelDataType inType = input.getRowType();
+ IgniteDistribution inDistr = IgniteMdDistribution._distribution(input, mq);
+
+ return aggregate(inType, inDistr, groupSet, groupSets, aggCalls);
+ }
+
+ /**
+ * @return Single aggregate relational node distribution calculated on the basis of its input and groupingSets.
+ * <b>Note</b> that the method returns {@code null} in case the given input cannot be processed by a single aggregate.
+ */
+ public static IgniteDistribution aggregate(RelDataType inType, IgniteDistribution inDistr, ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
+ switch (inDistr.getType()) {
+ case SINGLETON:
+ case BROADCAST_DISTRIBUTED:
+ return inDistr;
+ case HASH_DISTRIBUTED:
+ if (!simpleAggregate(groupSet, groupSets))
+ return null; // ROLLUP and CUBE require map-reduce processing
+
+ int inFields = inType.getFieldCount();
+ int outFields = groupSet.cardinality() + aggCalls.size();
+
+ Mappings.TargetMapping mapping = IgniteAggregate.partialMapping(inFields, outFields, groupSet);
+
+ return inDistr.apply(mapping);
+ default:
+ // Other group types require map-reduce aggregate processing;
+ return null;
+ }
+ }
+
+ /**
+ * @return Map aggregate relational node distribution calculated on the basis of its input and groupingSets.
+ * <b>Note</b> that the method returns {@code null} in case the given input cannot be processed in map-reduce
+ * style by an aggregate.
+ */
+ public static IgniteDistribution mapAggregate(RelMetadataQuery mq, RelNode input, ImmutableBitSet groupSet,
+ List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
+ IgniteDistribution inputDistr = IgniteMdDistribution._distribution(input, mq);
+
+ switch (inputDistr.getType()) {
+ case SINGLETON:
+ case BROADCAST_DISTRIBUTED:
+ return inputDistr;
+
+ case RANDOM_DISTRIBUTED:
+ case HASH_DISTRIBUTED:
+ return random(); // its OK to just erase distribution here
+
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * @return Reduce aggregate relational node distribution calculated on the basis of its input and groupingSets.
+ * <b>Note</b> that the method returns {@code null} in case the given input cannot be reduced.
+ */
+ public static IgniteDistribution reduceAggregate(RelMetadataQuery mq, RelNode input, ImmutableBitSet groupSet,
+ List<ImmutableBitSet> groupSets, List<AggregateCall> aggCalls) {
+ IgniteDistribution inputDistr = IgniteMdDistribution._distribution(input, mq);
+
+ // requires singleton distribution
+ if (inputDistr.getType() == RelDistribution.Type.SINGLETON)
+ return inputDistr;
+
+ return null;
+ }
+
+ /**
* See {@link RelTraitDef#canonize(org.apache.calcite.plan.RelTrait)}.
*/
private static IgniteDistribution canonize(IgniteDistribution distr) {
return DistributionTraitDef.INSTANCE.canonize(distr);
}
+ /** */
+ private static boolean simpleAggregate(ImmutableBitSet groupSet, List<ImmutableBitSet> groupSets) {
+ return Aggregate.Group.induce(groupSet, groupSets) == Aggregate.Group.SIMPLE;
+ }
+
+ /**
+ * Distribution suggestion for SingleRel.
+ */
+ public static class Suggestion {
+ /** */
+ private final IgniteDistribution out;
+
+ /** */
+ private final IgniteDistribution in;
+
+ /**
+ * @param out Result distribution.
+ * @param in Required in distribution.
+ */
+ public Suggestion(IgniteDistribution out, IgniteDistribution in) {
+ this.out = out;
+ this.in = in;
+ }
+
+ /**
+ * @return Result distribution.
+ */
+ public IgniteDistribution out() {
+ return out;
+ }
+
+ /**
+ * @return Required in distribution.
+ */
+ public IgniteDistribution in() {
+ return in;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ Suggestion that = (Suggestion) o;
+
+ return out == that.out && in == that.in;
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ int result = out.hashCode();
+ result = 31 * result + in.hashCode();
+ return result;
+ }
+ }
+
/**
* Distribution suggestion for BiRel.
*/
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java
index 01c63b6..f66fb54 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/Commons.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.processors.query.calcite.util;
+import java.io.PrintWriter;
+import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -31,6 +33,7 @@ import org.apache.calcite.linq4j.tree.Primitive;
import org.apache.calcite.plan.Context;
import org.apache.calcite.plan.Contexts;
import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.externalize.RelWriterImpl;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelDataTypeField;
@@ -38,6 +41,8 @@ import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.GridComponent;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.processors.query.QueryContext;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.Accumulator;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.GroupKey;
import org.apache.ignite.internal.processors.query.calcite.prepare.PlanningContext;
import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
import org.apache.ignite.internal.util.typedef.F;
@@ -245,4 +250,24 @@ public final class Commons {
public static <T> List<T> flat(List<List<? extends T>> src) {
return src.stream().flatMap(List::stream).collect(Collectors.toList());
}
+
+ /** */
+ public static RelDataType aggregationDataRowType(RelDataTypeFactory typeFactory) {
+ assert typeFactory instanceof IgniteTypeFactory;
+
+ RelDataTypeFactory.Builder builder = new RelDataTypeFactory.Builder(typeFactory);
+
+ builder.add("GROUP_ID", typeFactory.createJavaType(byte.class));
+ builder.add("GROUP_KEY", typeFactory.createJavaType(GroupKey.class));
+ builder.add("AGG_DATA", typeFactory.createArrayType(typeFactory.createJavaType(Accumulator.class), -1));
+
+ return builder.build();
+ }
+
+ /** */
+ public static String explain(RelNode rel) {
+ StringWriter writer = new StringWriter();
+ rel.explain(new RelWriterImpl(new PrintWriter(writer)));
+ return writer.toString();
+ }
}
diff --git a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/IgniteResource.java b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/IgniteResource.java
index e8a5ade..8a5f02b 100644
--- a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/IgniteResource.java
+++ b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/util/IgniteResource.java
@@ -34,4 +34,8 @@ public interface IgniteResource {
/** */
@Resources.BaseMessage("Cannot update field \"{0}\". You cannot update key, key fields or val field in case the val is a complex type.")
Resources.ExInst<SqlValidatorException> cannotUpdateField(String field);
+
+ /** */
+ @Resources.BaseMessage("Illegal aggregate function. {0} is unsupported at the moment.")
+ Resources.ExInst<SqlValidatorException> unsupportedAggregationFunction(String a0);
}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java
index 5af47fb..aef2231 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/CalciteQueryProcessorTest.java
@@ -56,6 +56,35 @@ public class CalciteQueryProcessorTest extends GridCommonAbstractTest {
}
@Test
+ public void aggregate() throws Exception {
+ IgniteCache<Integer, Employer> employer = ignite.getOrCreateCache(new CacheConfiguration<Integer, Employer>()
+ .setName("employer")
+ .setSqlSchema("PUBLIC")
+ .setIndexedTypes(Integer.class, Employer.class)
+ .setBackups(2)
+ );
+
+ waitForReadyTopology(internalCache(employer).context().topology(), new AffinityTopologyVersion(5, 2));
+
+ employer.putAll(ImmutableMap.of(
+ 0, new Employer("Igor", 10d),
+ 1, new Employer("Roman", 15d),
+ 2, new Employer("Nikolay", 20d)
+ ));
+
+ QueryEngine engine = Commons.lookupComponent(grid(1).context(), QueryEngine.class);
+
+ List<FieldsQueryCursor<List<?>>> query = engine.query(null, "PUBLIC",
+ "SELECT * FROM employer WHERE employer.salary = (SELECT AVG(employer.salary) FROM employer)");
+
+ assertEquals(1, query.size());
+
+ List<List<?>> rows = query.get(0).getAll();
+ assertEquals(1, rows.size());
+ assertEquals(Arrays.asList("Roman", 15d), F.first(rows));
+ }
+
+ @Test
public void query() throws Exception {
IgniteCache<Integer, Developer> developer = ignite.getOrCreateCache(new CacheConfiguration<Integer, Developer>()
.setName("developer")
@@ -298,6 +327,23 @@ public class CalciteQueryProcessorTest extends GridCommonAbstractTest {
}
/** */
+ public static class Employer {
+ /** */
+ @QuerySqlField
+ public String name;
+
+ /** */
+ @QuerySqlField
+ public Double salary;
+
+ /** */
+ public Employer(String name, Double salary) {
+ this.name = name;
+ this.salary = salary;
+ }
+ }
+
+ /** */
public static class Developer {
/** */
@QuerySqlField
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/PlannerTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/PlannerTest.java
index 25bbe86..d47b4f0 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/PlannerTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/PlannerTest.java
@@ -110,8 +110,10 @@ public class PlannerTest extends GridCommonAbstractTest {
/** */
private List<UUID> nodes;
+ /** */
private List<QueryTaskExecutorImpl> executors;
+ /** */
private volatile Throwable lastException;
/** */
@@ -366,6 +368,96 @@ public class PlannerTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
@Test
+ public void testAggregate() throws Exception {
+ IgniteTypeFactory f = new IgniteTypeFactory(IgniteTypeSystem.INSTANCE);
+
+ TestTable employer = new TestTable(
+ new RelDataTypeFactory.Builder(f)
+ .add("ID", f.createJavaType(Integer.class))
+ .add("NAME", f.createJavaType(String.class))
+ .add("SALARY", f.createJavaType(Double.class))
+ .build()) {
+
+ @Override public NodesMapping mapping(PlanningContext ctx) {
+ return new NodesMapping(null, Arrays.asList(
+ select(nodes, 0, 1),
+ select(nodes, 1, 2),
+ select(nodes, 2, 0),
+ select(nodes, 0, 1),
+ select(nodes, 1, 2)
+ ), NodesMapping.HAS_PARTITIONED_CACHES);
+ }
+
+ @Override public IgniteDistribution distribution() {
+ return IgniteDistributions.affinity(0, "Employers", "hash");
+ }
+ };
+
+ IgniteSchema publicSchema = new IgniteSchema("PUBLIC");
+
+ publicSchema.addTable("EMPS", employer);
+
+ SchemaPlus schema = createRootSchema(false)
+ .add("PUBLIC", publicSchema);
+
+ String sql = "SELECT * FROM emps WHERE emps.salary = (SELECT AVG(emps.salary) FROM emps)";
+
+ RelTraitDef<?>[] traitDefs = {
+ DistributionTraitDef.INSTANCE,
+ ConventionTraitDef.INSTANCE
+ };
+
+ PlanningContext ctx = PlanningContext.builder()
+ .localNodeId(F.first(nodes))
+ .originatingNodeId(F.first(nodes))
+ .parentContext(Contexts.empty())
+ .frameworkConfig(newConfigBuilder(FRAMEWORK_CONFIG)
+ .defaultSchema(schema)
+ .traitDefs(traitDefs)
+ .build())
+ .logger(log)
+ .query(sql)
+ .parameters(new Object[]{2})
+ .topologyVersion(AffinityTopologyVersion.NONE)
+ .build();
+
+ RelNode root;
+
+ try (IgnitePlanner planner = ctx.planner()){
+ assertNotNull(planner);
+
+ String query = ctx.query();
+
+ assertNotNull(query);
+
+ // Parse
+ SqlNode sqlNode = planner.parse(query);
+
+ // Validate
+ sqlNode = planner.validate(sqlNode);
+
+ // Convert to Relational operators graph
+ root = planner.convert(sqlNode);
+
+ // Transformation chain
+ root = planner.transform(PlannerPhase.HEURISTIC_OPTIMIZATION, root.getTraitSet(), root);
+
+ // Transformation chain
+ RelTraitSet desired = root.getCluster().traitSet()
+ .replace(IgniteConvention.INSTANCE)
+ .replace(IgniteDistributions.single())
+ .simplify();
+
+ root = planner.transform(PlannerPhase.OPTIMIZATION, desired, root);
+ }
+
+ assertNotNull(root);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
public void testHepPlaner() throws Exception {
IgniteTypeFactory f = new IgniteTypeFactory(IgniteTypeSystem.INSTANCE);
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ExecutionTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ExecutionTest.java
index 1d93e03..6b98fda 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ExecutionTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ExecutionTest.java
@@ -17,15 +17,32 @@
package org.apache.ignite.internal.processors.query.calcite.exec.rel;
+import com.google.common.collect.ImmutableList;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.UUID;
+import org.apache.calcite.rel.RelCollations;
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.calcite.util.Pair;
+import org.apache.ignite.internal.processors.query.calcite.exec.ArrayRowHandler;
import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
+import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler;
+import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.WrappersFactoryImpl;
+import org.apache.ignite.internal.processors.query.calcite.type.IgniteTypeFactory;
import org.apache.ignite.internal.util.typedef.F;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import static org.apache.ignite.internal.processors.query.calcite.exec.rel.AggregateNode.AggregateType.MAP;
+import static org.apache.ignite.internal.processors.query.calcite.exec.rel.AggregateNode.AggregateType.REDUCE;
+import static org.apache.ignite.internal.processors.query.calcite.exec.rel.AggregateNode.AggregateType.SINGLE;
+
/**
*
*/
@@ -87,4 +104,541 @@ public class ExecutionTest extends AbstractExecutionTest {
Assert.assertArrayEquals(new Object[]{2, "Ivan", "Calcite"}, rows.get(0));
Assert.assertArrayEquals(new Object[]{2, "Ivan", "Ignite"}, rows.get(1));
}
+
+ @Test
+ public void testAggregateMapReduceAvg() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ IgniteTypeFactory typeFactory = ctx.getTypeFactory();
+
+ RelDataType rowType = typeFactory.createStructType(F.asList(
+ Pair.of("NAME", typeFactory.createSqlType(SqlTypeName.VARCHAR)),
+ Pair.of("SALARY", typeFactory.createSqlType(SqlTypeName.INTEGER))));
+
+ ImmutableList<ImmutableBitSet> groupSets = ImmutableList.of(ImmutableBitSet.of());
+
+ AggregateCall call = AggregateCall.create(
+ SqlStdOperatorTable.AVG,
+ false,
+ false,
+ false,
+ ImmutableIntList.of(1),
+ -1,
+ RelCollations.EMPTY,
+ typeFactory.createSqlType(SqlTypeName.DOUBLE),
+ null);
+
+ WrappersFactoryImpl factory;
+
+ factory = new WrappersFactoryImpl(ctx, MAP, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> map = new AggregateNode<>(ctx, MAP, groupSets, factory, rowHnd);
+ map.register(scan);
+
+ factory = new WrappersFactoryImpl(ctx, REDUCE, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> reduce = new AggregateNode<>(ctx, REDUCE, groupSets, factory, rowHnd);
+ reduce.register(map);
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(reduce);
+
+ assertTrue(root.hasNext());
+ assertEquals(725d, rowHnd.get(0, root.next()));
+ assertFalse(root.hasNext());
+ }
+
+ @Test
+ public void testAggregateMapReduceSum() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ IgniteTypeFactory typeFactory = ctx.getTypeFactory();
+
+ RelDataType rowType = typeFactory.createStructType(F.asList(
+ Pair.of("NAME", typeFactory.createSqlType(SqlTypeName.VARCHAR)),
+ Pair.of("SALARY", typeFactory.createSqlType(SqlTypeName.INTEGER))));
+
+ // empty groups means SELECT SUM(field) FROM table
+ ImmutableList<ImmutableBitSet> groupSets = ImmutableList.of(ImmutableBitSet.of());
+
+ // AVG on second field
+ AggregateCall call = AggregateCall.create(
+ SqlStdOperatorTable.SUM,
+ false,
+ false,
+ false,
+ ImmutableIntList.of(1),
+ -1,
+ RelCollations.EMPTY,
+ typeFactory.createSqlType(SqlTypeName.INTEGER),
+ null);
+
+ WrappersFactoryImpl factory;
+
+ factory = new WrappersFactoryImpl(ctx, MAP, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> map = new AggregateNode<>(ctx, MAP, groupSets, factory, rowHnd);
+ map.register(scan);
+
+ factory = new WrappersFactoryImpl(ctx, REDUCE, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> reduce = new AggregateNode<>(ctx, REDUCE, groupSets, factory, rowHnd);
+ reduce.register(map);
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(reduce);
+
+ assertTrue(root.hasNext());
+ assertEquals((Integer)2900, rowHnd.get(0, root.next()));
+ assertFalse(root.hasNext());
+ }
+
+ @Test
+ public void testAggregateMapReduceMin() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ IgniteTypeFactory typeFactory = ctx.getTypeFactory();
+
+ RelDataType rowType = typeFactory.createStructType(F.asList(
+ Pair.of("NAME", typeFactory.createSqlType(SqlTypeName.VARCHAR)),
+ Pair.of("SALARY", typeFactory.createSqlType(SqlTypeName.INTEGER))));
+
+ ImmutableList<ImmutableBitSet> groupSets = ImmutableList.of(ImmutableBitSet.of());
+
+ AggregateCall call = AggregateCall.create(
+ SqlStdOperatorTable.MIN,
+ false,
+ false,
+ false,
+ ImmutableIntList.of(1),
+ -1,
+ RelCollations.EMPTY,
+ typeFactory.createSqlType(SqlTypeName.INTEGER),
+ null);
+
+ WrappersFactoryImpl factory;
+
+ factory = new WrappersFactoryImpl(ctx, MAP, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> map = new AggregateNode<>(ctx, MAP, groupSets, factory, rowHnd);
+ map.register(scan);
+
+ factory = new WrappersFactoryImpl(ctx, REDUCE, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> reduce = new AggregateNode<>(ctx, REDUCE, groupSets, factory, rowHnd);
+ reduce.register(map);
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(reduce);
+
+ assertTrue(root.hasNext());
+ assertEquals((Integer)200, rowHnd.get(0, root.next()));
+ assertFalse(root.hasNext());
+ }
+
+ @Test
+ public void testAggregateMapReduceMax() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ IgniteTypeFactory typeFactory = ctx.getTypeFactory();
+
+ RelDataType rowType = typeFactory.createStructType(F.asList(
+ Pair.of("NAME", typeFactory.createSqlType(SqlTypeName.VARCHAR)),
+ Pair.of("SALARY", typeFactory.createSqlType(SqlTypeName.INTEGER))));
+
+ ImmutableList<ImmutableBitSet> groupSets = ImmutableList.of(ImmutableBitSet.of());
+
+ AggregateCall call = AggregateCall.create(
+ SqlStdOperatorTable.MAX,
+ false,
+ false,
+ false,
+ ImmutableIntList.of(1),
+ -1,
+ RelCollations.EMPTY,
+ typeFactory.createSqlType(SqlTypeName.INTEGER),
+ null);
+
+ WrappersFactoryImpl factory;
+
+ factory = new WrappersFactoryImpl(ctx, MAP, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> map = new AggregateNode<>(ctx, MAP, groupSets, factory, rowHnd);
+ map.register(scan);
+
+ factory = new WrappersFactoryImpl(ctx, REDUCE, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> reduce = new AggregateNode<>(ctx, REDUCE, groupSets, factory, rowHnd);
+ reduce.register(map);
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(reduce);
+
+ assertTrue(root.hasNext());
+ assertEquals((Integer)1400, rowHnd.get(0, root.next()));
+ assertFalse(root.hasNext());
+ }
+
+ @Test
+ public void testAggregateMapReduceCount() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ IgniteTypeFactory typeFactory = ctx.getTypeFactory();
+
+ RelDataType rowType = typeFactory.createStructType(F.asList(
+ Pair.of("NAME", typeFactory.createSqlType(SqlTypeName.VARCHAR)),
+ Pair.of("SALARY", typeFactory.createSqlType(SqlTypeName.INTEGER))));
+
+ ImmutableList<ImmutableBitSet> groupSets = ImmutableList.of(ImmutableBitSet.of());
+
+ AggregateCall call = AggregateCall.create(
+ SqlStdOperatorTable.COUNT,
+ false,
+ false,
+ false,
+ ImmutableIntList.of(),
+ -1,
+ RelCollations.EMPTY,
+ typeFactory.createSqlType(SqlTypeName.INTEGER),
+ null);
+
+ WrappersFactoryImpl factory;
+
+ factory = new WrappersFactoryImpl(ctx, MAP, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> map = new AggregateNode<>(ctx, MAP, groupSets, factory, rowHnd);
+ map.register(scan);
+
+ factory = new WrappersFactoryImpl(ctx, REDUCE, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> reduce = new AggregateNode<>(ctx, REDUCE, groupSets, factory, rowHnd);
+ reduce.register(map);
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(reduce);
+
+ assertTrue(root.hasNext());
+ assertEquals((Integer)4, rowHnd.get(0, root.next()));
+ assertFalse(root.hasNext());
+ }
+
+ @Test
+ public void testAggregateAvg() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ IgniteTypeFactory typeFactory = ctx.getTypeFactory();
+
+ RelDataType rowType = typeFactory.createStructType(F.asList(
+ Pair.of("NAME", typeFactory.createSqlType(SqlTypeName.VARCHAR)),
+ Pair.of("SALARY", typeFactory.createSqlType(SqlTypeName.INTEGER))));
+
+ ImmutableList<ImmutableBitSet> groupSets = ImmutableList.of(ImmutableBitSet.of());
+
+ AggregateCall call = AggregateCall.create(
+ SqlStdOperatorTable.AVG,
+ false,
+ false,
+ false,
+ ImmutableIntList.of(1),
+ -1,
+ RelCollations.EMPTY,
+ typeFactory.createSqlType(SqlTypeName.DOUBLE),
+ null);
+
+ WrappersFactoryImpl factory = new WrappersFactoryImpl(ctx, SINGLE, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> agg = new AggregateNode<>(ctx, SINGLE, groupSets, factory, rowHnd);
+ agg.register(scan);
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(agg);
+
+ assertTrue(root.hasNext());
+ assertEquals(725d, rowHnd.get(0, root.next()));
+ assertFalse(root.hasNext());
+ }
+
+ @Test
+ public void testAggregateSum() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ IgniteTypeFactory typeFactory = ctx.getTypeFactory();
+
+ RelDataType rowType = typeFactory.createStructType(F.asList(
+ Pair.of("NAME", typeFactory.createSqlType(SqlTypeName.VARCHAR)),
+ Pair.of("SALARY", typeFactory.createSqlType(SqlTypeName.INTEGER))));
+
+ // empty groups means SELECT SUM(field) FROM table
+ ImmutableList<ImmutableBitSet> groupSets = ImmutableList.of(ImmutableBitSet.of());
+
+ // AVG on second field
+ AggregateCall call = AggregateCall.create(
+ SqlStdOperatorTable.SUM,
+ false,
+ false,
+ false,
+ ImmutableIntList.of(1),
+ -1,
+ RelCollations.EMPTY,
+ typeFactory.createSqlType(SqlTypeName.INTEGER),
+ null);
+
+ WrappersFactoryImpl factory = new WrappersFactoryImpl(ctx, SINGLE, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> agg = new AggregateNode<>(ctx, SINGLE, groupSets, factory, rowHnd);
+ agg.register(scan);
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(agg);
+
+ assertTrue(root.hasNext());
+ assertEquals((Integer)2900, rowHnd.get(0, root.next()));
+ assertFalse(root.hasNext());
+ }
+
+ @Test
+ public void testAggregateMin() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ IgniteTypeFactory typeFactory = ctx.getTypeFactory();
+
+ RelDataType rowType = typeFactory.createStructType(F.asList(
+ Pair.of("NAME", typeFactory.createSqlType(SqlTypeName.VARCHAR)),
+ Pair.of("SALARY", typeFactory.createSqlType(SqlTypeName.INTEGER))));
+
+ ImmutableList<ImmutableBitSet> groupSets = ImmutableList.of(ImmutableBitSet.of());
+
+ AggregateCall call = AggregateCall.create(
+ SqlStdOperatorTable.MIN,
+ false,
+ false,
+ false,
+ ImmutableIntList.of(1),
+ -1,
+ RelCollations.EMPTY,
+ typeFactory.createSqlType(SqlTypeName.INTEGER),
+ null);
+
+ WrappersFactoryImpl factory = new WrappersFactoryImpl(ctx, SINGLE, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> agg = new AggregateNode<>(ctx, SINGLE, groupSets, factory, rowHnd);
+ agg.register(scan);
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(agg);
+
+ assertTrue(root.hasNext());
+ assertEquals((Integer)200, rowHnd.get(0, root.next()));
+ assertFalse(root.hasNext());
+ }
+
+ @Test
+ public void testAggregateMax() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ IgniteTypeFactory typeFactory = ctx.getTypeFactory();
+
+ RelDataType rowType = typeFactory.createStructType(F.asList(
+ Pair.of("NAME", typeFactory.createSqlType(SqlTypeName.VARCHAR)),
+ Pair.of("SALARY", typeFactory.createSqlType(SqlTypeName.INTEGER))));
+
+ ImmutableList<ImmutableBitSet> groupSets = ImmutableList.of(ImmutableBitSet.of());
+
+ AggregateCall call = AggregateCall.create(
+ SqlStdOperatorTable.MAX,
+ false,
+ false,
+ false,
+ ImmutableIntList.of(1),
+ -1,
+ RelCollations.EMPTY,
+ typeFactory.createSqlType(SqlTypeName.INTEGER),
+ null);
+
+ WrappersFactoryImpl factory = new WrappersFactoryImpl(ctx, SINGLE, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> agg = new AggregateNode<>(ctx, SINGLE, groupSets, factory, rowHnd);
+ agg.register(scan);
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(agg);
+
+ assertTrue(root.hasNext());
+ assertEquals((Integer)1400, rowHnd.get(0, root.next()));
+ assertFalse(root.hasNext());
+ }
+
+ @Test
+ public void testAggregateCount() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 200),
+ rowHnd.create("Roman", 300),
+ rowHnd.create("Ivan", 1400),
+ rowHnd.create("Alexey", 1000)
+ ));
+
+ IgniteTypeFactory typeFactory = ctx.getTypeFactory();
+
+ RelDataType rowType = typeFactory.createStructType(F.asList(
+ Pair.of("NAME", typeFactory.createSqlType(SqlTypeName.VARCHAR)),
+ Pair.of("SALARY", typeFactory.createSqlType(SqlTypeName.INTEGER))));
+
+ ImmutableList<ImmutableBitSet> groupSets = ImmutableList.of(ImmutableBitSet.of());
+
+ AggregateCall call = AggregateCall.create(
+ SqlStdOperatorTable.COUNT,
+ false,
+ false,
+ false,
+ ImmutableIntList.of(),
+ -1,
+ RelCollations.EMPTY,
+ typeFactory.createSqlType(SqlTypeName.INTEGER),
+ null);
+
+ WrappersFactoryImpl factory = new WrappersFactoryImpl(ctx, SINGLE, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> agg = new AggregateNode<>(ctx, SINGLE, groupSets, factory, rowHnd);
+ agg.register(scan);
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(agg);
+
+ assertTrue(root.hasNext());
+ assertEquals((Integer)4, rowHnd.get(0, root.next()));
+ assertFalse(root.hasNext());
+ }
+
+ @Test
+ public void testAggregateCountByGroup() throws Exception {
+ ExecutionContext ctx = executionContext(F.first(nodes()), UUID.randomUUID(), 0);
+
+ RowHandler<Object[]> rowHnd = ArrayRowHandler.INSTANCE;
+
+ ScanNode scan = new ScanNode(ctx, Arrays.asList(
+ rowHnd.create("Igor", 0, 200),
+ rowHnd.create("Roman", 1, 300),
+ rowHnd.create("Ivan", 1, 1400),
+ rowHnd.create("Alexey", 0, 1000)
+ ));
+
+ IgniteTypeFactory typeFactory = ctx.getTypeFactory();
+
+ RelDataType rowType = typeFactory.createStructType(F.asList(
+ Pair.of("NAME", typeFactory.createSqlType(SqlTypeName.VARCHAR)),
+ Pair.of("PROJECT_ID", typeFactory.createSqlType(SqlTypeName.INTEGER)),
+ Pair.of("SALARY", typeFactory.createSqlType(SqlTypeName.INTEGER))));
+
+ ImmutableList<ImmutableBitSet> groupSets = ImmutableList.of(ImmutableBitSet.of(1));
+
+ AggregateCall call = AggregateCall.create(
+ SqlStdOperatorTable.COUNT,
+ false,
+ false,
+ false,
+ ImmutableIntList.of(),
+ -1,
+ RelCollations.EMPTY,
+ typeFactory.createSqlType(SqlTypeName.INTEGER),
+ null);
+
+ WrappersFactoryImpl factory = new WrappersFactoryImpl(ctx, SINGLE, rowHnd, F.asList(call), rowType);
+
+ AggregateNode<Object[]> agg = new AggregateNode<>(ctx, SINGLE, groupSets, factory, rowHnd);
+ agg.register(scan);
+
+ RootNode root = new RootNode(ctx, c -> {});
+ root.register(agg);
+
+ assertTrue(root.hasNext());
+ // TODO needs a sort, relying on an order in a hash table looks strange
+ Assert.assertArrayEquals(rowHnd.create(1, 2), root.next());
+ Assert.assertArrayEquals(rowHnd.create(0, 2), root.next());
+ assertFalse(root.hasNext());
+ }
}
[ignite] 01/02: Merge branch 'ignite-master' into ignite-12248
Posted by gv...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gvvinblade pushed a commit to branch ignite-12248
in repository https://gitbox.apache.org/repos/asf/ignite.git
commit 10b276845b182040ebf983b5bb3c9ea2e95abe8c
Merge: 4654a3e d580b84
Author: Igor Seliverstov <gv...@gmail.com>
AuthorDate: Mon Apr 6 17:36:16 2020 +0300
Merge branch 'ignite-master' into ignite-12248
examples/pom.xml | 18 +
.../OpenCensusMetricsExporterExample.java | 105 ++
.../OpenCensusMetricsExporterExampleSelfTest.java | 22 +-
.../testsuites/IgniteExamplesSelfTestSuite.java | 4 +
.../internal/jdbc2/JdbcMetadataSelfTest.java | 1 +
.../rest/ChangeStateCommandHandlerTest.java | 13 +-
.../rest/JettyRestProcessorAbstractSelfTest.java | 67 +-
.../rest/JettyRestProcessorCommonSelfTest.java | 25 +-
.../rest/protocols/tcp/TcpRestParserSelfTest.java | 1 +
.../JdbcThinConnectionAdditionalSecurityTest.java | 2 +-
.../jdbc/thin/JdbcThinConnectionSSLTest.java | 2 +-
.../jdbc/thin/JdbcThinLocalQueriesSelfTest.java | 45 +-
.../ignite/jdbc/thin/JdbcThinMetadataSelfTest.java | 20 +-
...nsactionFinishOnDeactivatedClusterSelfTest.java | 4 +-
modules/clients/src/test/keystore/README.txt | 11 +
modules/clients/src/test/keystore/ca/node01.jks | Bin 3719 -> 0 bytes
modules/clients/src/test/keystore/ca/node02.jks | Bin 4599 -> 0 bytes
modules/clients/src/test/keystore/ca/node02old.jks | Bin 4598 -> 0 bytes
modules/clients/src/test/keystore/ca/node03.jks | Bin 3753 -> 0 bytes
.../clients/src/test/keystore/ca/oneca-index.txt | 3 +
modules/clients/src/test/keystore/ca/oneca-serial | 1 +
modules/clients/src/test/keystore/ca/oneca.cnf | 13 +-
modules/clients/src/test/keystore/ca/oneca.key | 52 +-
modules/clients/src/test/keystore/ca/oneca.pem | 21 +
modules/clients/src/test/keystore/ca/oneindex.txt | 1 -
.../clients/src/test/keystore/ca/oneindex.txt.attr | 1 -
modules/clients/src/test/keystore/ca/oneserial | 1 -
.../clients/src/test/keystore/ca/threeca-index.txt | 2 +
.../test/keystore/ca/{twoserial => threeca-serial} | 0
.../test/keystore/ca/{oneca.cnf => threeca.cnf} | 17 +-
modules/clients/src/test/keystore/ca/threeca.key | 28 +
modules/clients/src/test/keystore/ca/threeca.pem | 21 +
.../clients/src/test/keystore/ca/trust-both.jks | Bin 1718 -> 0 bytes
modules/clients/src/test/keystore/ca/trust-one.jks | Bin 877 -> 0 bytes
modules/clients/src/test/keystore/ca/trust-two.jks | Bin 891 -> 0 bytes
.../clients/src/test/keystore/ca/twoca-index.txt | 5 +
modules/clients/src/test/keystore/ca/twoca-serial | 1 +
modules/clients/src/test/keystore/ca/twoca.cnf | 15 +-
modules/clients/src/test/keystore/ca/twoca.key | 52 +-
modules/clients/src/test/keystore/ca/twoca.pem | 21 +
modules/clients/src/test/keystore/ca/twoindex.txt | 2 -
.../clients/src/test/keystore/ca/twoindex.txt.attr | 1 -
modules/clients/src/test/keystore/client.jks | Bin 3232 -> 2324 bytes
modules/clients/src/test/keystore/client.pem | 69 -
modules/clients/src/test/keystore/client.pfx | Bin 3148 -> 0 bytes
.../clients/src/test/keystore/connectorClient.jks | Bin 0 -> 2353 bytes
.../clients/src/test/keystore/connectorServer.jks | Bin 0 -> 2355 bytes
modules/clients/src/test/keystore/generate-ca.sh | 97 ++
modules/clients/src/test/keystore/generate-keys.sh | 92 ++
modules/clients/src/test/keystore/generate.sh | 99 --
modules/clients/src/test/keystore/node01.jks | Bin 0 -> 2325 bytes
modules/clients/src/test/keystore/node0102.jks | Bin 0 -> 4618 bytes
modules/clients/src/test/keystore/node02.jks | Bin 0 -> 2325 bytes
modules/clients/src/test/keystore/node02old.jks | Bin 0 -> 2331 bytes
modules/clients/src/test/keystore/node03.jks | Bin 0 -> 2327 bytes
modules/clients/src/test/keystore/server.jks | Bin 3230 -> 2324 bytes
modules/clients/src/test/keystore/server.pem | 69 -
modules/clients/src/test/keystore/server.pfx | Bin 3148 -> 0 bytes
modules/clients/src/test/keystore/thinClient.jks | Bin 0 -> 2332 bytes
modules/clients/src/test/keystore/thinServer.jks | Bin 0 -> 2332 bytes
modules/clients/src/test/keystore/trust-both.jks | Bin 0 -> 1888 bytes
modules/clients/src/test/keystore/trust-one.jks | Bin 0 -> 960 bytes
modules/clients/src/test/keystore/trust-three.jks | Bin 0 -> 970 bytes
modules/clients/src/test/keystore/trust-two.jks | Bin 0 -> 960 bytes
modules/clients/src/test/keystore/trust.jks | Bin 2432 -> 0 bytes
.../jetty/rest-jetty-ssl-client-auth.xml} | 20 +-
.../src/test/resources/jetty/rest-jetty-ssl.xml | 2 +-
.../src/test/resources/jetty/rest-jetty.xml | 2 +-
.../src/test/resources/jetty/router-jetty-ssl.xml | 2 +-
.../src/test/resources/jetty/router-jetty.xml | 2 +-
.../src/test/resources/spring-router-ssl.xml | 2 +-
.../src/test/resources/spring-server-ssl-node.xml | 2 +-
.../SystemViewRowAttributeWalkerGenerator.java | 2 +
.../ignite/compatibility/cache/LocalCacheTest.java | 136 ++
.../ignite/compatibility/cache/package-info.java} | 14 +-
.../IgniteCompatibilityBasicTestSuite.java | 4 +-
modules/core/pom.xml | 2 +-
.../src/main/java/org/apache/ignite/Ignite.java | 3 +
.../main/java/org/apache/ignite/IgniteCluster.java | 6 +
.../org/apache/ignite/IgniteSystemProperties.java | 8 +
.../cache/store/jdbc/CacheAbstractJdbcStore.java | 8 +-
.../cache/store/jdbc/dialect/BasicJdbcDialect.java | 14 +-
.../cache/store/jdbc/dialect/MySQLDialect.java | 14 +-
.../org/apache/ignite/cluster/ClusterState.java | 7 +-
.../ignite/configuration/CacheConfiguration.java | 34 +
.../PlatformNearCacheConfiguration.java | 113 ++
.../apache/ignite/internal/ComputeMXBeanImpl.java | 92 ++
.../ignite/internal/GridKernalContextImpl.java | 6 +-
.../org/apache/ignite/internal/IgniteFeatures.java | 8 +
.../ignite/internal/IgniteMessagingImpl.java | 20 +-
.../internal/IgniteTooManyOpenFilesException.java | 66 +
.../apache/ignite/internal/QueryMXBeanImpl.java | 210 +++
...reFilter.java => SecurityAwareBiPredicate.java} | 25 +-
.../apache/ignite/internal/ServiceMXBeanImpl.java | 57 +
.../ignite/internal/TransactionsMXBeanImpl.java | 18 +-
.../internal/client/GridClientClusterState.java | 18 +-
.../client/impl/GridClientClusterStateImpl.java | 33 +-
.../impl/connection/GridClientConnection.java | 22 +-
.../connection/GridClientNioTcpConnection.java | 31 +-
.../internal/client/thin/ReliableChannel.java | 28 +-
.../internal/client/util/GridClientUtils.java | 41 +
.../ignite/internal/cluster/IgniteClusterImpl.java | 10 +-
.../internal/commandline/ActivateCommand.java | 3 +-
.../commandline/ClusterStateChangeCommand.java | 25 +-
.../internal/commandline/CommandHandler.java | 37 +-
.../ignite/internal/commandline/CommandList.java | 6 +-
.../internal/commandline/CommonArgParser.java | 13 +-
.../commandline/ConnectionAndSslParameters.java | 16 +-
.../internal/commandline/DeactivateCommand.java | 28 +-
.../ignite/internal/commandline/TxCommands.java | 32 +-
.../commandline/cache/CacheValidateIndexes.java | 37 +-
.../cache/argument/ValidateIndexesCommandArg.java | 3 +
.../internal/commandline/query/KillCommand.java | 214 +++
.../query/KillSubcommand.java} | 65 +-
.../internal/dto/IgniteDataTransferObject.java | 3 +
.../internal/managers/GridManagerAdapter.java | 18 +
.../internal/managers/IgniteMBeansManager.java | 18 +
.../managers/communication/GridIoManager.java | 94 +-
.../communication/IgniteMessageFactoryImpl.java | 37 +-
.../managers/discovery/CustomMessageWrapper.java | 3 +-
.../managers/discovery/DiscoveryCustomMessage.java | 8 -
.../systemview/walker/ComputeJobViewWalker.java | 77 ++
.../systemview/walker/ComputeTaskViewWalker.java | 50 +-
.../internal/pagemem/wal/record/WALRecord.java | 7 +-
...ageInitRootInlineFlagsCreatedVersionRecord.java | 98 ++
.../wal/record/delta/TrackingPageDeltaRecord.java | 30 +-
.../delta/TrackingPageRepairDeltaRecord.java} | 42 +-
.../authentication/UserAcceptedMessage.java | 5 -
.../authentication/UserProposedMessage.java | 5 -
.../cache/CacheAffinityChangeMessage.java | 5 -
.../cache/CacheStatisticsClearMessage.java | 5 -
.../cache/CacheStatisticsModeChangeMessage.java | 5 -
.../cache/ClientCacheChangeDiscoveryMessage.java | 5 -
.../ClientCacheChangeDummyDiscoveryMessage.java | 5 -
.../processors/cache/DynamicCacheChangeBatch.java | 5 -
.../cache/DynamicCacheChangeFailureMessage.java | 5 -
.../processors/cache/GridCacheAdapter.java | 11 +-
.../processors/cache/GridCacheContext.java | 5 +-
.../processors/cache/GridCacheEntryEx.java | 7 +-
.../processors/cache/GridCacheMapEntry.java | 70 +-
.../cache/GridCachePartitionExchangeManager.java | 17 +
.../processors/cache/GridCacheProcessor.java | 59 +-
.../cache/IgniteCacheOffheapManager.java | 5 +
.../cache/IgniteCacheOffheapManagerImpl.java | 61 +-
.../internal/processors/cache/QueryCursorImpl.java | 40 +-
...TimeoutOnPartitionMapExchangeChangeMessage.java | 5 -
.../processors/cache/WalStateFinishMessage.java | 8 +-
.../internal/processors/cache/WalStateManager.java | 2 +-
.../processors/cache/WalStateProposeMessage.java | 10 +-
.../binary/MetadataUpdateAcceptedMessage.java | 5 -
.../binary/MetadataUpdateProposedMessage.java | 5 -
.../distributed/GridDistributedCacheEntry.java | 7 +-
.../cache/distributed/dht/GridDhtCacheEntry.java | 2 +
.../preloader/GridDhtPartitionsExchangeFuture.java | 16 +-
.../dht/preloader/latch/ExchangeLatchManager.java | 13 +-
.../cache/distributed/near/GridNearCacheEntry.java | 10 +
.../cache/local/GridLocalCacheEntry.java | 8 +-
.../CheckpointWriteProgressSupplier.java | 45 -
.../GridCacheDatabaseSharedManager.java | 786 ++++-------
.../cache/persistence/GridCacheOffheapManager.java | 10 +
.../IgniteCacheDatabaseSharedManager.java | 1 +
.../persistence/checkpoint/CheckpointProgress.java | 106 ++
.../checkpoint/CheckpointProgressImpl.java | 271 ++++
.../checkpoint/PartitionDestroyQueue.java | 77 ++
.../checkpoint/PartitionDestroyRequest.java | 127 ++
.../cache/persistence/metastorage/MetaStorage.java | 2 +-
.../cache/persistence/pagemem/PageMemoryImpl.java | 19 +-
.../pagemem/PagesWriteSpeedBasedThrottle.java | 19 +-
.../persistence/pagemem/PagesWriteThrottle.java | 15 +-
.../cache/persistence/tree/BPlusTree.java | 12 +-
.../cache/persistence/tree/io/BPlusMetaIO.java | 175 ++-
.../cache/persistence/tree/io/TrackingPageIO.java | 18 +-
.../wal/serializer/RecordDataV1Serializer.java | 60 +-
.../wal/serializer/RecordDataV2Serializer.java | 18 +
.../cache/query/GridCacheQueryManager.java | 8 +-
.../query/continuous/SecurityAwareFilter.java | 1 +
.../continuous/SecurityAwareFilterFactory.java | 1 +
.../SecurityAwareTransformerFactory.java | 1 +
.../cache/transactions/IgniteTxStateImpl.java | 2 +-
.../processors/cache/tree/CacheDataTree.java | 2 +
.../processors/cache/verify/IdleVerifyUtility.java | 2 +-
.../cluster/ChangeGlobalStateFinishMessage.java | 5 -
.../cluster/ChangeGlobalStateMessage.java | 19 +-
.../cluster/GridClusterStateProcessor.java | 119 +-
.../cluster/IGridClusterStateProcessor.java | 17 +-
.../continuous/AbstractContinuousMessage.java | 5 -
.../continuous/StopRoutineAckDiscoveryMessage.java | 5 -
.../internal/processors/job/GridJobProcessor.java | 35 +-
.../internal/processors/job/GridJobWorker.java | 28 +-
.../marshaller/MappingAcceptedMessage.java | 5 -
.../marshaller/MappingProposedMessage.java | 5 -
.../DistributedMetaStorageUpdateAckMessage.java | 5 -
.../DistributedMetaStorageUpdateMessage.java | 5 -
.../processors/metric/GridMetricManager.java | 6 +
.../processors/metric/impl/MetricUtils.java | 24 +-
.../ClientListenerAbstractConnectionContext.java | 8 +-
.../odbc/ClientListenerConnectionContext.java | 5 +-
.../processors/odbc/ClientListenerNioListener.java | 10 +-
.../odbc/jdbc/JdbcConnectionContext.java | 14 +-
.../odbc/odbc/OdbcConnectionContext.java | 12 +-
.../processors/platform/PlatformContext.java | 29 +
.../processors/platform/PlatformContextImpl.java | 78 +-
.../processors/platform/PlatformNoopProcessor.java | 16 +
.../processors/platform/PlatformProcessor.java | 22 +
.../processors/platform/PlatformProcessorImpl.java | 63 +-
.../processors/platform/cache/PlatformCache.java | 22 +
.../platform/cache/PlatformCacheEntryFilter.java | 9 +
.../cache/PlatformCacheEntryFilterImpl.java | 67 +-
.../platform/cache/PlatformCacheManager.java | 86 ++
.../platform/cache/affinity/PlatformAffinity.java | 30 +
.../platform/callback/PlatformCallbackGateway.java | 74 ++
.../platform/callback/PlatformCallbackOp.java | 12 +
.../platform/client/ClientConnectionContext.java | 6 +-
.../platform/cluster/PlatformClusterGroup.java | 45 +-
.../platform/utils/PlatformConfigurationUtils.java | 28 +
.../processors/platform/utils/PlatformUtils.java | 4 +-
.../internal/processors/query/GridQueryCancel.java | 7 +-
.../processors/query/GridQueryIndexing.java | 11 +
.../processors/query/GridQueryProcessor.java | 9 +-
.../schema/SchemaIndexCachePartitionWorker.java | 72 +-
.../query/schema/SchemaIndexCacheStat.java | 88 ++
.../query/schema/SchemaIndexCacheVisitorImpl.java | 98 +-
.../message/SchemaFinishDiscoveryMessage.java | 5 -
.../message/SchemaProposeDiscoveryMessage.java | 5 -
.../processors/resource/GridResourceIoc.java | 19 +-
.../processors/resource/GridResourceProcessor.java | 9 +-
.../processors/rest/GridRestProcessor.java | 1 +
.../message/GridClientClusterStateRequest.java | 5 +-
...t.java => GridClientClusterStateRequestV2.java} | 52 +-
.../GridChangeClusterStateCommandHandler.java | 3 +-
.../cluster/GridChangeStateCommandHandler.java | 5 +-
.../rest/protocols/tcp/GridTcpRestNioListener.java | 7 +
.../rest/request/GridRestChangeStateRequest.java | 20 +
.../rest/request/GridRestClusterStateRequest.java | 21 +
.../processors/rest/request/GridRestRequest.java | 18 +
.../AbstractSecurityAwareExternalizable.java | 2 +-
.../service/ServiceChangeBatchRequest.java | 5 -
.../ServiceClusterDeploymentResultBatch.java | 5 -
.../processors/task/GridTaskProcessor.java | 4 +-
.../org/apache/ignite/internal/sql/SqlKeyword.java | 12 +
.../org/apache/ignite/internal/sql/SqlParser.java | 24 +
.../sql/command/SqlKillComputeTaskCommand.java | 66 +
.../sql/command/SqlKillContinuousQueryCommand.java | 81 ++
.../internal/sql/command/SqlKillQueryCommand.java | 51 +-
.../sql/command/SqlKillScanQueryCommand.java | 97 ++
.../sql/command/SqlKillServiceCommand.java | 63 +
.../sql/command/SqlKillTransactionCommand.java | 63 +
.../util/GridConcurrentMultiPairQueue.java | 202 +++
.../internal/util/GridReadOnlyArrayView.java | 80 --
.../apache/ignite/internal/util/IgniteUtils.java | 198 +++
.../internal/util/distributed/FullMessage.java | 5 -
.../internal/util/distributed/InitMessage.java | 5 -
.../ignite/internal/util/nio/GridNioServer.java | 14 -
.../ignite/internal/util/nio/GridNioSession.java | 6 +
.../internal/util/nio/GridNioSessionImpl.java | 20 +
.../internal/util/nio/ssl/GridNioSslFilter.java | 2 +
...ask.java => VisorComputeCancelSessionTask.java} | 37 +-
.../compute/VisorComputeCancelSessionTaskArg.java | 73 ++
.../compute/VisorComputeCancelSessionsTask.java | 4 +-
.../VisorContinuousQueryCancelTask.java} | 38 +-
.../query/VisorContinuousQueryCancelTaskArg.java | 89 ++
.../VisorQueryCancelOnInitiatorTask.java} | 43 +-
.../query/VisorQueryCancelOnInitiatorTaskArg.java | 81 ++
.../VisorScanQueryCancelTask.java} | 35 +-
.../visor/query/VisorScanQueryCancelTaskArg.java | 101 ++
.../visor/service/VisorCancelServiceTask.java | 6 +-
.../visor/verify/VisorValidateIndexesTaskArg.java | 34 +-
.../apache/ignite/lang/IgniteProductVersion.java | 14 +-
.../AbstractNodeNameAwareMarshaller.java | 8 +-
.../org/apache/ignite/marshaller/Marshaller.java | 10 +-
.../ComputeMXBean.java} | 26 +-
.../org/apache/ignite/mxbean/IgniteMXBean.java | 9 +-
.../java/org/apache/ignite/mxbean/QueryMXBean.java | 67 +
.../ServiceMXBean.java} | 16 +-
.../apache/ignite/mxbean/TransactionsMXBean.java | 12 +
.../plugin/security/AuthenticationContext.java | 21 +
.../security/SecurityPermissionSetBuilder.java | 7 +-
.../ignite/plugin/security/SecuritySubject.java | 10 +
.../org/apache/ignite/spi/IgniteSpiAdapter.java | 33 +-
.../org/apache/ignite/spi/IgniteSpiContext.java | 36 +
.../tcp/TcpCommunicationMetricsListener.java | 156 ++-
.../spi/communication/tcp/TcpCommunicationSpi.java | 79 +-
.../spi/discovery/DiscoverySpiCustomMessage.java | 4 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 10 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 3 +-
.../ignite/spi/discovery/tcp/TcpDiscoverySpi.java | 18 +-
.../ignite/spi/metric/jmx/MetricRegistryMBean.java | 55 +-
.../ignite/spi/systemview/view/ComputeJobView.java | 197 +++
.../spi/systemview/view/ComputeTaskView.java | 60 +-
.../org/apache/ignite/ssl/SslContextFactory.java | 29 +-
.../main/resources/META-INF/classnames.properties | 2 +
modules/core/src/test/config/tests.properties | 26 +-
.../jdbc/CacheJdbcPojoStoreAbstractSelfTest.java | 42 +-
.../failure/FailureHandlerTriggeredTest.java | 4 -
.../DiagnosticLogForPartitionStatesTest.java | 103 +-
.../ThinClientAbstractPartitionAwarenessTest.java | 22 +
...lientPartitionAwarenessResourceReleaseTest.java | 74 ++
.../commandline/CommandHandlerParsingTest.java | 138 +-
.../IgniteMessageFactoryImplTest.java | 47 +-
.../IncompleteDeserializationExceptionTest.java | 5 -
.../ignite/internal/metric/JmxExporterSpiTest.java | 51 +-
.../ignite/internal/metric/MetricsSelfTest.java | 8 +-
.../internal/metric/SystemViewComputeJobTest.java | 517 ++++++++
.../cache/ClusterStateThinClientAbstractTest.java | 2 +-
.../cache/GridCacheAbstractMetricsSelfTest.java | 27 +-
.../processors/cache/GridCacheTestEntryEx.java | 4 +-
.../dht/GridCacheColocatedDebugTest.java | 109 ++
.../preloader/latch/ExchangeLatchManagerTest.java | 75 ++
.../persistence/IgnitePdsCorruptedStoreTest.java | 3 +-
.../IgniteCheckpointDirtyPagesForLowLoadTest.java | 2 +-
.../db/checkpoint/ProgressWatchdog.java | 6 +-
.../pagemem/BPlusTreePageMemoryImplTest.java | 12 +-
.../BPlusTreeReuseListPageMemoryImplTest.java | 12 +-
.../IgnitePageMemReplaceDelayedWriteUnitTest.java | 12 +-
.../pagemem/IgniteThrottlingUnitTest.java | 14 +-
.../pagemem/IndexStoragePageMemoryImplTest.java | 12 +-
.../pagemem/PageMemoryImplNoLoadTest.java | 12 +-
.../persistence/pagemem/PageMemoryImplTest.java | 17 +-
.../pagemem/PagesWriteThrottleSandboxTest.java | 9 +-
.../cache/query/CacheScanQueryFailoverTest.java | 131 ++
.../CacheContinuousQueryLongP2PTest.java | 6 +-
.../continuous/GridEventConsumeSelfTest.java | 91 +-
.../processors/database/BPlusTreeSelfTest.java | 2 +-
.../GridCommandHandlerSslWithSecurityTest.java | 1 +
...eOperationPermissionCreateDestroyCheckTest.java | 164 +++
.../client/AdditionalSecurityCheckTest.java | 3 +-
.../client/ThinClientSslPermissionCheckTest.java | 237 ++++
.../TestCertificateSecurityPluginProvider.java} | 36 +-
....java => TestCertificateSecurityProcessor.java} | 91 +-
.../security/impl/TestSecurityContext.java | 7 +-
.../security/impl/TestSecurityProcessor.java | 1 +
.../security/impl/TestSecuritySubject.java | 19 +
.../MessagingRemoteSecurityContextCheckTest.java | 148 +++
.../util/GridConcurrentMultiPairQueueTest.java | 155 +++
.../ignite/internal/util/IgniteUtilsSelfTest.java | 137 +-
.../util/nio/impl/GridNioFilterChainSelfTest.java | 139 --
.../internal/util/nio/impl}/MockNioSession.java | 8 +-
.../loadtests/hashmap/GridCacheTestContext.java | 3 +-
.../ignite/platform/PlatformDeployServiceTask.java | 48 +-
.../ignite/platform/PlatformThreadUtils.java | 51 +
.../security/SecurityPermissionSetBuilderTest.java | 22 +-
.../TooManyOpenFilesTcpCommunicationSpiTest.java | 130 ++
.../spi/discovery/ClusterRebalancedMetricTest.java | 187 +++
.../FilterDataForClientNodeDiscoveryTest.java | 5 -
.../discovery/tcp/DiscoveryClientSocketTest.java | 184 +++
.../TcpDiscoveryPendingMessageDeliveryTest.java | 5 -
.../tcp/TcpDiscoverySslTrustedUntrustedTest.java | 16 +
.../apache/ignite/ssl/MultipleSSLContextsTest.java | 263 ++++
.../ignite/testframework/GridSpiTestContext.java | 30 +-
.../testframework/junits/IgniteTestResources.java | 16 +-
.../testframework/wal/record/RecordUtils.java | 8 +
.../testsuites/IgniteCacheMvccTestSuite6.java | 2 +
.../ignite/testsuites/IgniteCacheTestSuite6.java | 2 +
.../ignite/testsuites/IgniteCacheTestSuite9.java | 2 +
.../IgniteSpiCommunicationSelfTestSuite.java | 3 +
.../IgniteSpiDiscoverySelfTestSuite.java | 5 +-
.../ignite/testsuites/IgniteUtilSelfTestSuite.java | 4 +
.../ignite/testsuites/SecurityTestSuite.java | 12 +-
.../util/GridCommandHandlerClusterByClassTest.java | 104 +-
.../apache/ignite/util/GridCommandHandlerTest.java | 58 +
...andHandlerClusterByClassTest_cache_help.output} | 3 +-
...idCommandHandlerClusterByClassTest_help.output} | 45 +-
...lerClusterByClassWithSSLTest_cache_help.output} | 3 +-
...ndHandlerClusterByClassWithSSLTest_help.output} | 45 +-
.../resources/org.apache.ignite.util/bigUtf.txt | Bin 0 -> 430138 bytes
.../resources/org.apache.ignite.util/limitUtf.txt | 10 +
.../query/h2/opt/GridH2SpatialIndex.java | 8 +-
modules/hibernate-4.2/pom.xml | 2 +-
modules/hibernate-5.1/pom.xml | 2 +-
modules/hibernate-5.3/pom.xml | 2 +-
.../cache/query/RegisteredQueryCursor.java | 62 +-
.../processors/query/h2/CommandProcessor.java | 73 +-
.../processors/query/h2/ConcurrentStripedPool.java | 177 +++
.../processors/query/h2/ConnectionManager.java | 426 ++----
.../processors/query/h2/H2CachedStatementKey.java | 27 +-
.../internal/processors/query/h2/H2Connection.java | 207 +++
.../processors/query/h2/H2ConnectionWrapper.java | 133 --
.../processors/query/h2/H2FieldsIterator.java | 20 +-
.../processors/query/h2/H2PooledConnection.java | 129 ++
.../processors/query/h2/H2ResultSetIterator.java | 263 +++-
.../processors/query/h2/H2StatementCache.java | 25 +-
.../internal/processors/query/h2/H2Utils.java | 231 +++-
.../processors/query/h2/IgniteH2Indexing.java | 110 +-
.../internal/processors/query/h2/QueryParser.java | 448 +++----
.../processors/query/h2/SchemaManager.java | 54 +-
.../query/h2/database/H2PkHashIndex.java | 3 +-
.../processors/query/h2/database/H2Tree.java | 150 ++-
.../processors/query/h2/database/H2TreeIndex.java | 60 +-
.../query/h2/database/InlineIndexHelper.java | 2 +-
.../processors/query/h2/dml/UpdatePlan.java | 48 +-
.../processors/query/h2/dml/UpdatePlanBuilder.java | 13 +-
.../processors/query/h2/opt/GridH2IndexBase.java | 5 +-
.../processors/query/h2/opt/GridH2Table.java | 22 +-
.../processors/query/h2/opt/QueryContext.java | 46 +
.../query/h2/opt/QueryContextRegistry.java | 34 +-
.../query/h2/opt/join/DistributedLookupBatch.java | 29 +-
.../query/h2/sql/GridSqlQuerySplitter.java | 30 +-
.../query/h2/twostep/GridMapQueryExecutor.java | 64 +-
.../query/h2/twostep/GridReduceQueryExecutor.java | 260 ++--
.../query/h2/twostep/MapQueryResult.java | 23 +-
.../query/h2/twostep/MapQueryResults.java | 9 -
.../query/h2/twostep/ReduceQueryRun.java | 15 -
.../visor/verify/ValidateIndexesClosure.java | 37 +-
.../visor/verify/VisorValidateIndexesTask.java | 7 +-
.../org/apache/ignite/client/ClientTestSuite.java | 4 +-
.../cache/IgniteCacheQueryH2IndexingLeakTest.java | 14 +-
.../cache/index/H2ConnectionLeaksSelfTest.java | 54 +-
.../cache/metric/SqlViewExporterSpiTest.java | 1 +
...xingMultithreadedLoadContinuousRestartTest.java | 2 +-
.../db/LongDestroyDurableBackgroundTaskTest.java | 2 +-
.../processors/database/RebuildIndexTest.java | 232 ++++
.../RebuildIndexWithHistoricalRebalanceTest.java | 2 +-
...actQueryTableLockAndConnectionPoolSelfTest.java | 123 +-
.../query/IgniteSqlGroupConcatCollocatedTest.java | 26 +
.../IgniteSqlGroupConcatNotCollocatedTest.java | 25 +
.../IgniteSqlSkipReducerOnUpdateDmlSelfTest.java | 2 +-
.../query/IgniteSqlSplitterSelfTest.java | 3 +-
.../internal/processors/query/KillQueryTest.java | 56 +-
.../processors/query/LocalQueryLazyTest.java | 128 +-
.../query/h2/GridIndexFullRebuildTest.java | 2 +-
.../processors/query/h2/QueryDataPageScanTest.java | 7 +-
.../query/h2/sql/GridQueryParsingTest.java | 60 +-
.../org/apache/ignite/sqltests/BaseSqlTest.java | 39 +
.../IgniteBinaryCacheQueryTestSuite.java | 6 +
.../testsuites/IgnitePdsWithIndexingTestSuite.java | 4 +-
...idCommandHandlerIndexingClusterByClassTest.java | 12 +
.../util/GridCommandHandlerIndexingTest.java | 55 +-
.../ignite/util/KillCommandsCommandShTest.java | 176 +++
.../apache/ignite/util/KillCommandsMXBeanTest.java | 181 +++
.../apache/ignite/util/KillCommandsSQLTest.java | 196 +++
.../org/apache/ignite/util/KillCommandsTests.java | 396 ++++++
.../opencensus/OpenCensusMetricExporterSpi.java | 25 +-
modules/platforms/.gitignore | 7 +-
.../ExpiryCacheHolderTest.cs | 5 +
.../Apache.Ignite.Benchmarks.csproj | 3 +
.../Apache.Ignite.Benchmarks/BenchmarkRunner.cs | 17 +-
.../Apache.Ignite.Benchmarks/Config/benchmark.xml | 23 +
.../Interop/GetNearBenchmark.cs | 64 +
.../Interop/PutNearBenchmark.cs | 59 +
.../Interop/ScanQueryBenchmark.cs | 86 ++
.../Apache.Ignite.Core.Tests.DotNetCore.csproj | 59 +
.../Apache.Ignite.Core.Tests.csproj | 10 +-
.../Binary/BinaryStructureTest.cs | 130 ++
.../Binary/TypeNameParserTest.cs | 22 +-
.../Cache/CacheMetricsTest.cs | 82 ++
.../Cache/CacheNearTest.cs | 200 ---
.../Cache/CacheTestAsyncWrapper.cs | 8 +-
.../Cache/Near/CacheNearCreateDestroy.cs | 348 +++++
.../Cache/Near/CacheNearTest.cs | 1371 ++++++++++++++++++++
.../Cache/Near/CacheNearTopologyChangeTest.cs | 614 +++++++++
.../Cache/Near/FailingCacheStore.cs | 72 +
.../Apache.Ignite.Core.Tests/Cache/Near/Foo.cs | 66 +
.../Cache/Near/ScanQueryNearCacheFilter.cs} | 33 +-
.../Cache/Near/ScanQueryNoNearCacheFilter.cs} | 36 +-
.../Cache/Near/StoreNoNearCacheFilter.cs} | 32 +-
.../Cache/Query/Linq/CacheLinqTest.Misc.cs | 7 +-
.../Cache/Store/CacheStoreTest.cs | 3 +-
.../Client/Cache/ListLogger.cs | 19 +-
.../Config/full-config.xml | 1 +
.../IgniteConfigurationSerializerTest.cs | 11 +
.../Apache.Ignite.Core.Tests/IgniteUtilsTest.cs | 71 +
.../Services/ServicesTest.cs | 47 +
.../Apache.Ignite.Core.Tests/TestUtils.Common.cs | 49 +-
.../Apache.Ignite.Core/Apache.Ignite.Core.csproj | 5 +
.../Apache.Ignite.Core/Cache/CachePeekMode.cs | 9 +-
.../Cache/Configuration/CacheConfiguration.cs | 31 +
.../PlatformNearCacheConfiguration.cs | 72 +
.../Cache/Eviction/FifoEvictionPolicy.cs | 2 +-
.../dotnet/Apache.Ignite.Core/Cache/ICache.cs | 5 +
.../Cache/Store/CacheStoreAdapter.cs | 2 +-
.../Apache.Ignite.Core/Cluster/IClusterGroup.cs | 7 +-
.../Apache.Ignite.Core/Events/DiscoveryEvent.cs | 2 +-
.../platforms/dotnet/Apache.Ignite.Core/IIgnite.cs | 53 +
.../IgniteConfigurationSection.xsd | 22 +
.../Impl/Binary/Structure/BinaryStructure.cs | 135 +-
.../Impl/Binary/Structure/BinaryStructureEntry.cs | 2 +-
.../Impl/Cache/CacheAffinityImpl.cs | 18 +-
.../Impl/Cache/CacheEntryFilterHolder.cs | 40 +-
.../Apache.Ignite.Core/Impl/Cache/CacheImpl.cs | 480 ++++++-
.../Apache.Ignite.Core/Impl/Cache/CacheOp.cs | 4 +-
.../Impl/Cache/Near/INearCache.cs | 82 ++
.../Impl/Cache/Near/NearCache.cs | 275 ++++
.../Impl/Cache/Near/NearCacheEntry.cs | 85 ++
.../Impl/Cache/Near/NearCacheManager.cs | 198 +++
.../Impl/Cache/Query/QueryCursor.cs | 2 +-
.../Impl/Cache/Query/QueryCursorBase.cs | 2 +-
.../Impl/Client/Cache/ClientCachePartitionMap.cs | 14 +-
.../Impl/Client/ClientFailoverSocket.cs | 8 +-
.../Impl/Client/Cluster/ClientClusterGroup.cs | 3 +-
.../Apache.Ignite.Core/Impl/Client/IgniteClient.cs | 26 +
.../Impl/Cluster/ClusterGroupImpl.cs | 66 +-
.../Impl/Common/CopyOnWriteConcurrentDictionary.cs | 24 +
.../Apache.Ignite.Core/Impl/Common/TaskRunner.cs | 22 +-
.../Apache.Ignite.Core/Impl/IIgniteInternal.cs | 30 +-
.../Impl/IPlatformTargetInternal.cs | 2 +-
.../dotnet/Apache.Ignite.Core/Impl/Ignite.cs | 118 +-
.../dotnet/Apache.Ignite.Core/Impl/IgniteUtils.cs | 12 +-
.../Apache.Ignite.Core/Impl/PlatformJniTarget.cs | 10 +-
.../Impl/PlatformTargetAdapter.cs | 53 +-
.../Impl/Services/ServiceProxySerializer.cs | 13 +
.../Impl/Unmanaged/UnmanagedCallbackOp.cs | 6 +-
.../Impl/Unmanaged/UnmanagedCallbacks.cs | 61 +-
.../Apache.Ignite.DotNetCore.sln.DotSettings | 1 +
modules/platforms/dotnet/Apache.Ignite.sln | 1 -
.../platforms/dotnet/Apache.Ignite.sln.DotSettings | 1 +
.../Datagrid/NearCacheExample.cs | 10 +-
.../protocols/http/jetty/GridJettyRestHandler.java | 28 +-
.../IgniteSpringDataCrudSelfExpressionTest.java | 21 +
.../misc/PersonExpressionRepository.java | 2 +-
.../IgniteSpringDataCrudSelfExpressionTest.java | 21 +
.../misc/PersonExpressionRepository.java | 2 +-
modules/spring/pom.xml | 7 +
.../visor/commands/open/VisorOpenCommand.scala | 13 +-
.../src/test/resources/jetty-with-ciphers-0.xml | 2 +-
.../src/test/resources/jetty-with-ciphers-1.xml | 2 +-
.../src/test/resources/jetty-with-ciphers-2.xml | 2 +-
.../src/test/resources/jetty-with-ssl.xml | 2 +-
.../ZkCommunicationErrorResolveFinishMessage.java | 1 +
.../ZkCommunicationErrorResolveStartMessage.java | 1 +
.../zk/internal/ZkForceNodeFailMessage.java | 1 +
.../discovery/zk/internal/ZkNoServersMessage.java | 1 +
.../zk/internal/ZookeeperDiscoveryImpl.java | 33 +-
.../ZookeeperDiscoveryCustomEventsTest.java | 339 -----
parent/pom.xml | 4 +-
524 files changed, 19387 insertions(+), 4825 deletions(-)
diff --cc modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/SchemaManager.java
index 7f22861,826fb2d..424f010
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/SchemaManager.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/SchemaManager.java
@@@ -331,12 -326,8 +330,9 @@@ public class SchemaManager
H2Schema schema = schema(schemaName);
- Connection conn = null;
- try {
- conn = connMgr.connectionForThread().connection(schema.schemaName());
-
+ try(H2PooledConnection conn = connMgr.connection(schema.schemaName())) {
GridH2Table h2tbl = createTable(schema.schemaName(), schema, tblDesc, conn);
+ lsnr.onSqlTypeCreate(schemaName, type, cacheInfo);
schema.add(tblDesc);