You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tl...@apache.org on 2021/07/19 09:15:40 UTC
[ignite] 02/02: fix merge conflicts
This is an automated email from the ASF dual-hosted git repository.
tledkov pushed a commit to branch sql-calcite
in repository https://gitbox.apache.org/repos/asf/ignite.git
commit ea6c392628d9b99f7f5319d4cc11d12f1e62f2a4
Author: tledkov <tl...@gridgain.com>
AuthorDate: Mon Jul 19 12:14:28 2021 +0300
fix merge conflicts
---
.../calcite/exec/exp/IgniteSqlFunctionsTest.java | 17 ++
.../processors/query/GridQueryProcessor.java | 158 ++++++++----
.../testframework/wal/record/RecordUtils.java | 7 +
.../processors/query/h2/CommandProcessor.java | 268 ++++++++++++++++++++-
.../internal/processors/query/h2/QueryParser.java | 3 +-
.../processors/query/h2/SchemaManager.java | 135 ++++++++---
6 files changed, 504 insertions(+), 84 deletions(-)
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/IgniteSqlFunctionsTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/IgniteSqlFunctionsTest.java
index 993de3b..90cf6f9 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/IgniteSqlFunctionsTest.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/IgniteSqlFunctionsTest.java
@@ -1,3 +1,20 @@
+/*
+ * 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;
import java.math.BigDecimal;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
index 46227b5..eec8ccb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
@@ -89,6 +89,8 @@ import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor;
import org.apache.ignite.internal.processors.platform.PlatformContext;
import org.apache.ignite.internal.processors.platform.PlatformProcessor;
+import org.apache.ignite.internal.processors.query.aware.IndexBuildStatusStorage;
+import org.apache.ignite.internal.processors.query.aware.IndexRebuildFutureStorage;
import org.apache.ignite.internal.processors.query.property.QueryBinaryProperty;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitor;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
@@ -252,8 +254,11 @@ public class GridQueryProcessor extends GridProcessorAdapter {
/** Cache name - value typeId pairs for which type mismatch message was logged. */
private final Set<Long> missedCacheTypes = ConcurrentHashMap.newKeySet();
- /** Index rebuild aware. */
- private final IndexRebuildAware idxRebuildAware = new IndexRebuildAware();
+ /** Index rebuild futures. */
+ private final IndexRebuildFutureStorage idxRebuildFutStorage = new IndexRebuildFutureStorage();
+
+ /** Index build statuses. */
+ private final IndexBuildStatusStorage idxBuildStatusStorage;
/** Experimental calcite query engine. */
private QueryEngine experimentalQueryEngine;
@@ -265,6 +270,8 @@ public class GridQueryProcessor extends GridProcessorAdapter {
private final boolean useExperimentalSqlEngine;
/**
+ * Constructor.
+ *
* @param ctx Kernal context.
*/
public GridQueryProcessor(GridKernalContext ctx) throws IgniteCheckedException {
@@ -276,24 +283,22 @@ public class GridQueryProcessor extends GridProcessorAdapter {
idxCls = null;
}
else
- idx = INDEXING.inClassPath() ? U.<GridQueryIndexing>newInstance(INDEXING.className()) : null;
+ idx = INDEXING.inClassPath() ? U.newInstance(INDEXING.className()) : null;
idxProc = ctx.indexProcessor();
valCtx = new CacheQueryObjectValueContext(ctx);
- ioLsnr = new GridMessageListener() {
- @Override public void onMessage(UUID nodeId, Object msg, byte plc) {
- if (msg instanceof SchemaOperationStatusMessage) {
- SchemaOperationStatusMessage msg0 = (SchemaOperationStatusMessage)msg;
+ ioLsnr = (nodeId, msg, plc) -> {
+ if (msg instanceof SchemaOperationStatusMessage) {
+ SchemaOperationStatusMessage msg0 = (SchemaOperationStatusMessage)msg;
- msg0.senderNodeId(nodeId);
+ msg0.senderNodeId(nodeId);
- processStatusMessage(msg0);
- }
- else
- U.warn(log, "Unsupported IO message: " + msg);
+ processStatusMessage(msg0);
}
+ else
+ U.warn(log, "Unsupported IO message: " + msg);
};
for (GridComponent cmp : ctx.components()) {
@@ -306,6 +311,8 @@ public class GridQueryProcessor extends GridProcessorAdapter {
}
useExperimentalSqlEngine = getBoolean(IGNITE_EXPERIMENTAL_SQL_ENGINE);
+
+ idxBuildStatusStorage = new IndexBuildStatusStorage(ctx);
}
/** {@inheritDoc} */
@@ -321,12 +328,14 @@ public class GridQueryProcessor extends GridProcessorAdapter {
ctx.io().addMessageListener(TOPIC_SCHEMA, ioLsnr);
// Schedule queries detail metrics eviction.
- qryDetailMetricsEvictTask = ctx.timeout().schedule(new Runnable() {
- @Override public void run() {
- for (GridCacheContext ctxs : ctx.cache().context().cacheContexts())
- ctxs.queries().evictDetailMetrics();
- }
+ qryDetailMetricsEvictTask = ctx.timeout().schedule(() -> {
+ for (GridCacheContext ctxs : ctx.cache().context().cacheContexts())
+ ctxs.queries().evictDetailMetrics();
}, QRY_DETAIL_METRICS_EVICTION_FREQ, QRY_DETAIL_METRICS_EVICTION_FREQ);
+
+ idxBuildStatusStorage.start();
+
+ registerMetadataForRegisteredCaches(false);
}
/** {@inheritDoc} */
@@ -348,6 +357,8 @@ public class GridQueryProcessor extends GridProcessorAdapter {
}
busyLock.block();
+
+ idxBuildStatusStorage.stop();
}
/** {@inheritDoc} */
@@ -368,8 +379,6 @@ public class GridQueryProcessor extends GridProcessorAdapter {
* @throws IgniteCheckedException If failed.
*/
public void onCacheKernalStart() throws IgniteCheckedException {
- registerMetadataForRegisteredCaches();
-
synchronized (stateMux) {
exchangeReady = true;
@@ -377,6 +386,8 @@ public class GridQueryProcessor extends GridProcessorAdapter {
for (SchemaOperation schemaOp : schemaOps.values())
onSchemaPropose(schemaOp.proposeMessage());
}
+
+ idxBuildStatusStorage.onCacheKernalStart();
}
/**
@@ -489,7 +500,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
public void beforeExchange(GridDhtPartitionsExchangeFuture fut) {
Set<Integer> cacheIds = rebuildIndexCacheIds(fut);
- Set<Integer> rejected = idxRebuildAware.prepareRebuildIndexes(cacheIds, fut.initialVersion());
+ Set<Integer> rejected = idxRebuildFutStorage.prepareRebuildIndexes(cacheIds, fut.initialVersion());
if (log.isDebugEnabled()) {
log.debug("Preparing features of rebuilding indexes for caches on exchange [requested=" + cacheIds +
@@ -1159,10 +1170,12 @@ public class GridQueryProcessor extends GridProcessorAdapter {
/**
* Register metadata locally for already registered caches.
+ *
+ * @param platformOnly Whether to register non-Java platformOnly types only.
*/
- private void registerMetadataForRegisteredCaches() {
+ public void registerMetadataForRegisteredCaches(boolean platformOnly) {
for (DynamicCacheDescriptor cacheDescriptor : ctx.cache().cacheDescriptors().values())
- registerBinaryMetadata(cacheDescriptor.cacheConfiguration(), cacheDescriptor.schema());
+ registerBinaryMetadata(cacheDescriptor.cacheConfiguration(), cacheDescriptor.schema(), platformOnly);
}
/**
@@ -1176,7 +1189,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
continue;
try {
- registerBinaryMetadata(req.startCacheConfiguration(), req.schema());
+ registerBinaryMetadata(req.startCacheConfiguration(), req.schema(), false);
}
catch (BinaryObjectException e) {
ctx.cache().completeCacheStartFuture(req, false, e);
@@ -1189,9 +1202,10 @@ public class GridQueryProcessor extends GridProcessorAdapter {
*
* @param ccfg Cache configuration.
* @param schema Schema for which register metadata is required.
+ * @param platformOnly Whether to register non-Java platformOnly types only.
* @throws BinaryObjectException if register was failed.
*/
- private void registerBinaryMetadata(CacheConfiguration ccfg, QuerySchema schema) throws BinaryObjectException {
+ private void registerBinaryMetadata(CacheConfiguration ccfg, QuerySchema schema, boolean platformOnly) throws BinaryObjectException {
if (schema != null) {
Collection<QueryEntity> qryEntities = schema.entities();
@@ -1200,8 +1214,8 @@ public class GridQueryProcessor extends GridProcessorAdapter {
if (binaryEnabled) {
for (QueryEntity qryEntity : qryEntities) {
- registerTypeLocally(qryEntity.findKeyType());
- registerTypeLocally(qryEntity.findValueType());
+ registerTypeLocally(qryEntity.findKeyType(), platformOnly);
+ registerTypeLocally(qryEntity.findValueType(), platformOnly);
}
}
}
@@ -1289,9 +1303,10 @@ public class GridQueryProcessor extends GridProcessorAdapter {
* Register class metadata locally if it didn't do it earlier.
*
* @param clsName Class name for which the metadata should be registered.
+ * @param platformOnly Whether to only register non-Java platform types only.
* @throws BinaryObjectException if register was failed.
*/
- private void registerTypeLocally(String clsName) throws BinaryObjectException {
+ private void registerTypeLocally(String clsName, boolean platformOnly) throws BinaryObjectException {
if (clsName == null)
return;
@@ -1302,8 +1317,10 @@ public class GridQueryProcessor extends GridProcessorAdapter {
Class<?> cls = U.box(U.classForName(clsName, null, true));
- if (cls != null)
- binProc.binaryContext().registerClass(cls, true, false, true);
+ if (cls != null) {
+ if (!platformOnly)
+ binProc.binaryContext().registerClass(cls, true, false, true);
+ }
else
registerPlatformTypeLocally(clsName, binProc);
}
@@ -1318,9 +1335,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
private void registerPlatformTypeLocally(String clsName, CacheObjectBinaryProcessorImpl binProc) {
PlatformProcessor platformProc = ctx.platform();
- assert platformProc != null : "Platform processor must be initialized";
-
- if (!platformProc.hasContext())
+ if (platformProc == null || !platformProc.hasContext())
return;
PlatformContext platformCtx = platformProc.context();
@@ -1919,20 +1934,26 @@ public class GridQueryProcessor extends GridProcessorAdapter {
GridFutureAdapter<Void> createIdxFut = new GridFutureAdapter<>();
+ GridCacheContext<?, ?> cacheCtx = cacheInfo.cacheContext();
+
visitor = new SchemaIndexCacheVisitorImpl(
- cacheInfo.cacheContext(),
+ cacheCtx,
cancelTok,
createIdxFut
) {
/** {@inheritDoc} */
@Override public void visit(SchemaIndexCacheVisitorClosure clo) {
- super.visit(clo);
+ idxBuildStatusStorage.onStartBuildNewIndex(cacheCtx);
try {
+ super.visit(clo);
+
buildIdxFut.get();
}
catch (Exception e) {
throw new IgniteException(e);
+ } finally {
+ idxBuildStatusStorage.onFinishBuildNewIndex(cacheName);
}
}
};
@@ -1977,7 +1998,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
registerCache0(op0.cacheName(), op.schemaName(), cacheInfo, candRes.get1(), false);
}
- if (idxRebuildAware.prepareRebuildIndexes(singleton(cacheInfo.cacheId()), null).isEmpty())
+ if (idxRebuildFutStorage.prepareRebuildIndexes(singleton(cacheInfo.cacheId()), null).isEmpty())
rebuildIndexesFromHash0(cacheInfo.cacheContext(), false);
else {
if (log.isInfoEnabled())
@@ -2407,7 +2428,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
@Nullable IgniteInternalFuture<?> idxFut,
GridCacheContext<?, ?> cctx
) {
- GridFutureAdapter<Void> res = idxRebuildAware.indexRebuildFuture(cctx.cacheId());
+ GridFutureAdapter<Void> res = idxRebuildFutStorage.indexRebuildFuture(cctx.cacheId());
assert res != null;
@@ -2425,13 +2446,13 @@ public class GridQueryProcessor extends GridProcessorAdapter {
else if (!(err instanceof NodeStoppingException))
log.error("Failed to rebuild indexes for cache " + cacheInfo, err);
- idxRebuildAware.onFinishRebuildIndexes(cctx.cacheId(), err);
+ idxRebuildFutStorage.onFinishRebuildIndexes(cctx.cacheId(), err);
});
return res;
}
else {
- idxRebuildAware.onFinishRebuildIndexes(cctx.cacheId(), null);
+ idxRebuildFutStorage.onFinishRebuildIndexes(cctx.cacheId(), null);
return null;
}
@@ -2441,7 +2462,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
* @return Future that will be completed when indexes for given cache are restored.
*/
@Nullable public IgniteInternalFuture<?> indexRebuildFuture(int cacheId) {
- return idxRebuildAware.indexRebuildFuture(cacheId);
+ return idxRebuildFutStorage.indexRebuildFuture(cacheId);
}
/**
@@ -3856,7 +3877,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
GridDhtPartitionsExchangeFuture fut,
@Nullable Set<Integer> cacheIds
) {
- idxRebuildAware.cancelRebuildIndexesOnExchange(
+ idxRebuildFutStorage.cancelRebuildIndexesOnExchange(
cacheIds != null ? cacheIds : rebuildIndexCacheIds(fut),
fut.initialVersion()
);
@@ -3870,7 +3891,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
* @return {@code True} if need to rebuild.
*/
public boolean rebuildIndexOnExchange(int cacheId, GridDhtPartitionsExchangeFuture fut) {
- return idxRebuildAware.rebuildIndexesOnExchange(cacheId, fut.initialVersion());
+ return idxRebuildFutStorage.rebuildIndexesOnExchange(cacheId, fut.initialVersion());
}
/**
@@ -3881,7 +3902,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
* @return Cache ids for which features have not been added.
*/
public Set<Integer> prepareRebuildIndexes(Set<Integer> cacheIds) {
- return idxRebuildAware.prepareRebuildIndexes(cacheIds, null);
+ return idxRebuildFutStorage.prepareRebuildIndexes(cacheIds, null);
}
/**
@@ -3910,4 +3931,57 @@ public class GridQueryProcessor extends GridProcessorAdapter {
return cacheIds;
}
+
+ /**
+ * Callback on start of rebuild cache indexes.
+ * <p/>
+ * Adding an entry that rebuilding the cache indexes in progress.
+ * If the cache is persistent, then add this entry to the MetaStorage.
+ * <p/>
+ * When restarting/reactivating the node, it will be possible to check if
+ * the rebuilding of the indexes has been {@link #rebuildIndexesCompleted}.
+ *
+ * @param cacheCtx Cache context.
+ * @see #onFinishRebuildIndexes
+ * @see #rebuildIndexesCompleted
+ */
+ public void onStartRebuildIndexes(GridCacheContext cacheCtx) {
+ idxBuildStatusStorage.onStartRebuildIndexes(cacheCtx);
+ }
+
+ /**
+ * Callback on finish of rebuild cache indexes.
+ * <p/>
+ * If the cache is persistent, then we mark that the rebuilding of the
+ * indexes is completed and the entry will be deleted from the MetaStorage
+ * at the end of the checkpoint. Otherwise, delete the index rebuild entry.
+ *
+ * @param cacheCtx Cache context.
+ */
+ public void onFinishRebuildIndexes(GridCacheContext cacheCtx) {
+ idxBuildStatusStorage.onFinishRebuildIndexes(cacheCtx.name());
+ }
+
+ /**
+ * Check if rebuilding of indexes for the cache has been completed.
+ *
+ * @param cacheCtx Cache context.
+ * @return {@code True} if completed.
+ */
+ public boolean rebuildIndexesCompleted(GridCacheContext cacheCtx) {
+ return idxBuildStatusStorage.rebuildCompleted(cacheCtx.name());
+ }
+
+ /**
+ * Force a mark that the index rebuild for the cache has completed.
+ * <p/>
+ * If the cache is persistent, then we mark that the rebuilding of the
+ * indexes is completed and the entry will be deleted from the MetaStorage
+ * at the end of the checkpoint. Otherwise, delete the index rebuild entry.
+ *
+ * @param cacheName Cache name.
+ */
+ public void completeRebuildIndexes(String cacheName) {
+ idxBuildStatusStorage.onFinishRebuildIndexes(cacheName);
+ }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/wal/record/RecordUtils.java b/modules/core/src/test/java/org/apache/ignite/testframework/wal/record/RecordUtils.java
index 11efd68..5317b48 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/wal/record/RecordUtils.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/wal/record/RecordUtils.java
@@ -149,6 +149,7 @@ import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType
import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.PAGE_RECORD;
import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.PARTITION_DESTROY;
import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.PARTITION_META_PAGE_DELTA_RECORD_V3;
+import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.PARTITION_META_PAGE_DELTA_RECORD_V4;
import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.PARTITION_META_PAGE_UPDATE_COUNTERS;
import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.PARTITION_META_PAGE_UPDATE_COUNTERS_V2;
import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.PART_META_UPDATE_STATE;
@@ -227,6 +228,7 @@ public class RecordUtils {
put(ROLLBACK_TX_RECORD, RecordUtils::buildRollbackRecord);
put(PARTITION_META_PAGE_UPDATE_COUNTERS_V2, RecordUtils::buildMetaPageUpdatePartitionDataRecordV2);
put(PARTITION_META_PAGE_DELTA_RECORD_V3, RecordUtils::buildMetaPageUpdatePartitionDataRecordV3);
+ put(PARTITION_META_PAGE_DELTA_RECORD_V4, RecordUtils::buildMetaPageUpdatePartitionDataRecordV4);
put(MASTER_KEY_CHANGE_RECORD, RecordUtils::buildMasterKeyChangeRecord);
put(MASTER_KEY_CHANGE_RECORD_V2, RecordUtils::buildMasterKeyChangeRecordV2);
put(REENCRYPTION_START_RECORD, RecordUtils::buildEncryptionStatusRecord);
@@ -548,6 +550,11 @@ public class RecordUtils {
}
/** **/
+ public static UnsupportedWalRecord buildMetaPageUpdatePartitionDataRecordV4() {
+ return new UnsupportedWalRecord(PARTITION_META_PAGE_DELTA_RECORD_V4);
+ }
+
+ /** **/
public static UnsupportedWalRecord buildMasterKeyChangeRecord() {
return new UnsupportedWalRecord(MASTER_KEY_CHANGE_RECORD);
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java
index db5f715..cb70f8c 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java
@@ -33,10 +33,12 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteCluster;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.QueryIndex;
+import org.apache.ignite.cache.QueryIndexType;
import org.apache.ignite.cache.query.BulkLoadContextCursor;
import org.apache.ignite.cache.query.FieldsQueryCursor;
import org.apache.ignite.cluster.ClusterNode;
@@ -44,9 +46,13 @@ import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.events.EventType;
+import org.apache.ignite.internal.ComputeMXBeanImpl;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.GridTopic;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.QueryMXBeanImpl;
+import org.apache.ignite.internal.ServiceMXBeanImpl;
+import org.apache.ignite.internal.TransactionsMXBeanImpl;
import org.apache.ignite.internal.managers.communication.GridIoPolicy;
import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
import org.apache.ignite.internal.processors.bulkload.BulkLoadAckClientParameters;
@@ -81,9 +87,13 @@ import org.apache.ignite.internal.processors.query.h2.sql.GridSqlStatement;
import org.apache.ignite.internal.processors.query.messages.GridQueryKillRequest;
import org.apache.ignite.internal.processors.query.messages.GridQueryKillResponse;
import org.apache.ignite.internal.processors.query.schema.SchemaOperationException;
+import org.apache.ignite.internal.processors.query.stat.StatisticsKey;
+import org.apache.ignite.internal.processors.query.stat.StatisticsTarget;
+import org.apache.ignite.internal.processors.query.stat.config.StatisticsObjectConfiguration;
import org.apache.ignite.internal.sql.SqlCommandProcessor;
import org.apache.ignite.internal.sql.command.SqlAlterTableCommand;
import org.apache.ignite.internal.sql.command.SqlAlterUserCommand;
+import org.apache.ignite.internal.sql.command.SqlAnalyzeCommand;
import org.apache.ignite.internal.sql.command.SqlBeginTransactionCommand;
import org.apache.ignite.internal.sql.command.SqlBulkLoadCommand;
import org.apache.ignite.internal.sql.command.SqlCommand;
@@ -91,8 +101,16 @@ import org.apache.ignite.internal.sql.command.SqlCommitTransactionCommand;
import org.apache.ignite.internal.sql.command.SqlCreateIndexCommand;
import org.apache.ignite.internal.sql.command.SqlCreateUserCommand;
import org.apache.ignite.internal.sql.command.SqlDropIndexCommand;
+import org.apache.ignite.internal.sql.command.SqlDropStatisticsCommand;
import org.apache.ignite.internal.sql.command.SqlDropUserCommand;
+import org.apache.ignite.internal.sql.command.SqlIndexColumn;
+import org.apache.ignite.internal.sql.command.SqlKillComputeTaskCommand;
+import org.apache.ignite.internal.sql.command.SqlKillContinuousQueryCommand;
import org.apache.ignite.internal.sql.command.SqlKillQueryCommand;
+import org.apache.ignite.internal.sql.command.SqlKillScanQueryCommand;
+import org.apache.ignite.internal.sql.command.SqlKillServiceCommand;
+import org.apache.ignite.internal.sql.command.SqlKillTransactionCommand;
+import org.apache.ignite.internal.sql.command.SqlRefreshStatitsicsCommand;
import org.apache.ignite.internal.sql.command.SqlRollbackTransactionCommand;
import org.apache.ignite.internal.sql.command.SqlSetStreamingCommand;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
@@ -363,7 +381,10 @@ public class CommandProcessor extends SqlCommandProcessor {
|| cmd instanceof SqlAlterTableCommand
|| cmd instanceof SqlCreateUserCommand
|| cmd instanceof SqlAlterUserCommand
- || cmd instanceof SqlDropUserCommand;
+ || cmd instanceof SqlDropUserCommand
+ || cmd instanceof SqlAnalyzeCommand
+ || cmd instanceof SqlRefreshStatitsicsCommand
+ || cmd instanceof SqlDropStatisticsCommand;
}
/**
@@ -513,6 +534,251 @@ public class CommandProcessor extends SqlCommandProcessor {
}
/**
+ * Process kill scan query cmd.
+ *
+ * @param cmd Command.
+ */
+ private void processKillScanQueryCommand(SqlKillScanQueryCommand cmd) {
+ new QueryMXBeanImpl(ctx)
+ .cancelScan(cmd.getOriginNodeId(), cmd.getCacheName(), cmd.getQryId());
+ }
+
+ /**
+ * Process kill compute task command.
+ *
+ * @param cmd Command.
+ */
+ private void processKillComputeTaskCommand(SqlKillComputeTaskCommand cmd) {
+ new ComputeMXBeanImpl(ctx).cancel(cmd.getSessionId());
+ }
+
+ /**
+ * Process kill transaction cmd.
+ *
+ * @param cmd Command.
+ */
+ private void processKillTxCommand(SqlKillTransactionCommand cmd) {
+ new TransactionsMXBeanImpl(ctx).cancel(cmd.getXid());
+ }
+
+ /**
+ * Process kill service command.
+ *
+ * @param cmd Command.
+ */
+ private void processKillServiceTaskCommand(SqlKillServiceCommand cmd) {
+ new ServiceMXBeanImpl(ctx).cancel(cmd.getName());
+ }
+
+ /**
+ * Process kill continuous query cmd.
+ *
+ * @param cmd Command.
+ */
+ private void processKillContinuousQueryCommand(SqlKillContinuousQueryCommand cmd) {
+ new QueryMXBeanImpl(ctx).cancelContinuous(cmd.getOriginNodeId(), cmd.getRoutineId());
+ }
+
+ /**
+ * Process analyze command.
+ *
+ * @param cmd Sql analyze command.
+ */
+ private void processAnalyzeCommand(SqlAnalyzeCommand cmd) throws IgniteCheckedException {
+ ctx.security().authorize(SecurityPermission.CHANGE_STATISTICS);
+
+ IgniteH2Indexing indexing = (IgniteH2Indexing)ctx.query().getIndexing();
+
+ StatisticsObjectConfiguration objCfgs[] = cmd.configurations().stream()
+ .map(t -> {
+ if (t.key().schema() == null) {
+ StatisticsKey key = new StatisticsKey(cmd.schemaName(), t.key().obj());
+ return new StatisticsObjectConfiguration(key, t.columns().values(),
+ t.maxPartitionObsolescencePercent());
+ }
+ else
+ return t;
+ }).toArray(StatisticsObjectConfiguration[]::new);
+
+ indexing.statsManager().collectStatistics(objCfgs);
+ }
+
+ /**
+ * Process refresh statistics command.
+ *
+ * @param cmd Refresh statistics command.
+ */
+ private void processRefreshStatisticsCommand(SqlRefreshStatitsicsCommand cmd) throws IgniteCheckedException {
+ ctx.security().authorize(SecurityPermission.REFRESH_STATISTICS);
+
+ IgniteH2Indexing indexing = (IgniteH2Indexing)ctx.query().getIndexing();
+
+ StatisticsTarget[] targets = cmd.targets().stream()
+ .map(t -> (t.schema() == null) ? new StatisticsTarget(cmd.schemaName(), t.obj(), t.columns()) : t)
+ .toArray(StatisticsTarget[]::new);
+
+ indexing.statsManager().refreshStatistics(targets);
+ }
+
+ /**
+ * Process drop statistics command.
+ *
+ * @param cmd Drop statistics command.
+ */
+ private void processDropStatisticsCommand(SqlDropStatisticsCommand cmd) throws IgniteCheckedException {
+ ctx.security().authorize(SecurityPermission.CHANGE_STATISTICS);
+
+ IgniteH2Indexing indexing = (IgniteH2Indexing)ctx.query().getIndexing();
+
+ StatisticsTarget[] targets = cmd.targets().stream()
+ .map(t -> (t.schema() == null) ? new StatisticsTarget(cmd.schemaName(), t.obj(), t.columns()) : t)
+ .toArray(StatisticsTarget[]::new);
+
+ indexing.statsManager().dropStatistics(targets);
+ }
+
+ /**
+ * Run DDL statement.
+ *
+ * @param sql Original SQL.
+ * @param cmd Command.
+ */
+ private void runCommandNativeDdl(String sql, SqlCommand cmd) {
+ IgniteInternalFuture fut = null;
+
+ try {
+ isDdlOnSchemaSupported(cmd.schemaName());
+
+ finishActiveTxIfNecessary();
+
+ if (cmd instanceof SqlCreateIndexCommand) {
+ SqlCreateIndexCommand cmd0 = (SqlCreateIndexCommand)cmd;
+
+ GridH2Table tbl = schemaMgr.dataTable(cmd0.schemaName(), cmd0.tableName());
+
+ if (tbl == null)
+ throw new SchemaOperationException(SchemaOperationException.CODE_TABLE_NOT_FOUND, cmd0.tableName());
+
+ assert tbl.rowDescriptor() != null;
+
+ ensureDdlSupported(tbl.cacheInfo());
+
+ QueryIndex newIdx = new QueryIndex();
+
+ newIdx.setName(cmd0.indexName());
+
+ newIdx.setIndexType(cmd0.spatial() ? QueryIndexType.GEOSPATIAL : QueryIndexType.SORTED);
+
+ LinkedHashMap<String, Boolean> flds = new LinkedHashMap<>();
+
+ // Let's replace H2's table and property names by those operated by GridQueryProcessor.
+ GridQueryTypeDescriptor typeDesc = tbl.rowDescriptor().type();
+
+ for (SqlIndexColumn col : cmd0.columns()) {
+ GridQueryProperty prop = typeDesc.property(col.name());
+
+ if (prop == null)
+ throw new SchemaOperationException(SchemaOperationException.CODE_COLUMN_NOT_FOUND, col.name());
+
+ flds.put(prop.name(), !col.descending());
+ }
+
+ newIdx.setFields(flds);
+ newIdx.setInlineSize(cmd0.inlineSize());
+
+ fut = ctx.query().dynamicIndexCreate(tbl.cacheName(), cmd.schemaName(), typeDesc.tableName(),
+ newIdx, cmd0.ifNotExists(), cmd0.parallel());
+ }
+ else if (cmd instanceof SqlDropIndexCommand) {
+ SqlDropIndexCommand cmd0 = (SqlDropIndexCommand)cmd;
+
+ GridH2Table tbl = schemaMgr.dataTableForIndex(cmd0.schemaName(), cmd0.indexName());
+
+ if (tbl != null) {
+ ensureDdlSupported(tbl.cacheInfo());
+
+ fut = ctx.query().dynamicIndexDrop(tbl.cacheName(), cmd0.schemaName(), cmd0.indexName(),
+ cmd0.ifExists());
+ }
+ else {
+ if (cmd0.ifExists())
+ fut = new GridFinishedFuture();
+ else
+ throw new SchemaOperationException(SchemaOperationException.CODE_INDEX_NOT_FOUND,
+ cmd0.indexName());
+ }
+ }
+ else if (cmd instanceof SqlAlterTableCommand) {
+ SqlAlterTableCommand cmd0 = (SqlAlterTableCommand)cmd;
+
+ GridH2Table tbl = schemaMgr.dataTable(cmd0.schemaName(), cmd0.tableName());
+
+ if (tbl == null) {
+ throw new SchemaOperationException(SchemaOperationException.CODE_TABLE_NOT_FOUND,
+ cmd0.tableName());
+ }
+
+ Boolean logging = cmd0.logging();
+
+ assert logging != null : "Only LOGGING/NOLOGGING are supported at the moment.";
+
+ IgniteCluster cluster = ctx.grid().cluster();
+
+ if (logging) {
+ boolean res = cluster.enableWal(tbl.cacheName());
+
+ if (!res)
+ throw new IgniteSQLException("Logging already enabled for table: " + cmd0.tableName());
+ }
+ else {
+ boolean res = cluster.disableWal(tbl.cacheName());
+
+ if (!res)
+ throw new IgniteSQLException("Logging already disabled for table: " + cmd0.tableName());
+ }
+
+ fut = new GridFinishedFuture();
+ }
+ else if (cmd instanceof SqlCreateUserCommand) {
+ SqlCreateUserCommand addCmd = (SqlCreateUserCommand)cmd;
+
+ ctx.security().createUser(addCmd.userName(), addCmd.password().toCharArray());
+ }
+ else if (cmd instanceof SqlAlterUserCommand) {
+ SqlAlterUserCommand altCmd = (SqlAlterUserCommand)cmd;
+
+ ctx.security().alterUser(altCmd.userName(), altCmd.password().toCharArray());
+ }
+ else if (cmd instanceof SqlDropUserCommand) {
+ SqlDropUserCommand dropCmd = (SqlDropUserCommand)cmd;
+
+ ctx.security().dropUser(dropCmd.userName());
+ }
+ else if (cmd instanceof SqlAnalyzeCommand)
+ processAnalyzeCommand((SqlAnalyzeCommand) cmd);
+ else if (cmd instanceof SqlRefreshStatitsicsCommand)
+ processRefreshStatisticsCommand((SqlRefreshStatitsicsCommand) cmd);
+ else if (cmd instanceof SqlDropStatisticsCommand)
+ processDropStatisticsCommand((SqlDropStatisticsCommand) cmd);
+ else
+ throw new IgniteSQLException("Unsupported DDL operation: " + sql,
+ IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
+
+ if (fut != null)
+ fut.get();
+ }
+ catch (SchemaOperationException e) {
+ throw convert(e);
+ }
+ catch (IgniteSQLException e) {
+ throw e;
+ }
+ catch (Exception e) {
+ throw new IgniteSQLException(e.getMessage(), e);
+ }
+ }
+
+ /**
* Execute DDL statement.
*
* @param sql SQL.
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
index 2eec586..0ecb90c 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
@@ -80,7 +80,8 @@ public class QueryParser {
/** A pattern for commands having internal implementation in Ignite. */
private static final Pattern INTERNAL_CMD_RE = Pattern.compile(
- "^(create|drop)\\s+index|^alter\\s+table|^copy|^set|^begin|^commit|^rollback|^(create|alter|drop)\\s+user" +
+ "^(create|drop)\\s+index|^analyze\\s|^refresh\\sstatistics|^drop\\sstatistics|^alter\\s+table|^copy" +
+ "|^set|^begin|^commit|^rollback|^(create|alter|drop)\\s+user" +
"|^kill\\s+(query|scan|continuous|compute|service|transaction)|show|help|grant|revoke",
Pattern.CASE_INSENSITIVE);
diff --git 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
index 67ffad5..d43c6c2 100644
--- 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
@@ -31,6 +31,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.ignite.IgniteCheckedException;
@@ -63,10 +64,6 @@ import org.apache.ignite.internal.processors.query.h2.opt.GridH2RowDescriptor;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2Table;
import org.apache.ignite.internal.processors.query.h2.sys.SqlSystemTableEngine;
import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemView;
-import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewBaselineNodes;
-import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewCacheGroupsIOStatistics;
-import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewNodeAttributes;
-import org.apache.ignite.internal.processors.query.h2.sys.view.SqlSystemViewNodeMetrics;
import org.apache.ignite.internal.processors.query.schema.SchemaChangeListener;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitor;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
@@ -80,8 +77,10 @@ import org.apache.ignite.spi.systemview.view.SqlViewColumnView;
import org.apache.ignite.spi.systemview.view.SqlViewView;
import org.h2.index.Index;
import org.h2.table.Column;
+import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
+import static java.util.Objects.requireNonNull;
import static org.apache.ignite.internal.processors.metric.impl.MetricUtils.metricName;
/**
@@ -151,6 +150,12 @@ public class SchemaManager implements GridQuerySchemaManager {
/** Logger. */
private final IgniteLogger log;
+ /** Drop column listeners. */
+ private final Set<BiConsumer<GridH2Table, List<String>>> dropColsLsnrs = ConcurrentHashMap.newKeySet();
+
+ /** Drop table listeners. */
+ private final Set<BiConsumer<String, String>> dropTblLsnrs = ConcurrentHashMap.newKeySet();
+
/**
* Constructor.
*
@@ -207,9 +212,6 @@ public class SchemaManager implements GridQuerySchemaManager {
// Register PUBLIC schema which is always present.
schemas.put(QueryUtils.DFLT_SCHEMA, new H2Schema(QueryUtils.DFLT_SCHEMA, true));
- // Create system views.
- createSystemViews();
-
// Create schemas listed in node's configuration.
createPredefinedSchemas(schemaNames);
}
@@ -258,29 +260,6 @@ public class SchemaManager implements GridQuerySchemaManager {
}
/**
- * Create system views.
- */
- private void createSystemViews() throws IgniteCheckedException {
- for (SqlSystemView view : systemViews(ctx))
- createSystemView(QueryUtils.SCHEMA_SYS, view);
- }
-
- /**
- * @param ctx Context.
- * @return Predefined system views.
- */
- private Collection<SqlSystemView> systemViews(GridKernalContext ctx) {
- Collection<SqlSystemView> views = new ArrayList<>();
-
- views.add(new SqlSystemViewNodeAttributes(ctx));
- views.add(new SqlSystemViewBaselineNodes(ctx));
- views.add(new SqlSystemViewNodeMetrics(ctx));
- views.add(new SqlSystemViewCacheGroupsIOStatistics(ctx));
-
- return views;
- }
-
- /**
* Create predefined schemas.
*
* @param schemaNames Schema names.
@@ -376,7 +355,7 @@ public class SchemaManager implements GridQuerySchemaManager {
try {
tbl.table().setRemoveIndexOnDestroy(rmvIdx);
- dropTable(tbl);
+ dropTable(tbl, rmvIdx);
lsnr.onSqlTypeDropped(schemaName, tbl.type());
}
catch (Exception e) {
@@ -603,8 +582,9 @@ public class SchemaManager implements GridQuerySchemaManager {
* Drops table form h2 database and clear all related indexes (h2 text, lucene).
*
* @param tbl Table to unregister.
+ * @param destroy {@code true} when table destroyed (cache destroyed) otherwise {@code false}.
*/
- private void dropTable(H2TableDescriptor tbl) {
+ private void dropTable(H2TableDescriptor tbl, boolean destroy) {
assert tbl != null;
if (log.isDebugEnabled())
@@ -622,6 +602,9 @@ public class SchemaManager implements GridQuerySchemaManager {
log.debug("Dropping database index table with SQL: " + sql);
stmt.executeUpdate(sql);
+
+ if (destroy)
+ afterDropTable(tbl.schemaName(), tbl.tableName());
}
catch (SQLException e) {
throw new IgniteSQLException("Failed to drop database index table [type=" + tbl.type().name() +
@@ -805,6 +788,8 @@ public class SchemaManager implements GridQuerySchemaManager {
desc.table().dropColumns(cols, ifColExists);
lsnr.onSqlTypeUpdated(schemaName, desc.type(), desc.table().cacheInfo());
+
+ dropColsLsnrs.forEach(l -> l.accept(desc.table(), cols));
}
/**
@@ -957,7 +942,9 @@ public class SchemaManager implements GridQuerySchemaManager {
/** */
private static final class CompoundSchemaChangeListener implements SchemaChangeListener {
- /** */
+ /**
+ *
+ */
private final List<SchemaChangeListener> lsnrs;
/**
@@ -967,17 +954,23 @@ public class SchemaManager implements GridQuerySchemaManager {
this.lsnrs = lsnrs;
}
- /** {@inheritDoc} */
+ /**
+ * {@inheritDoc}
+ */
@Override public void onSchemaCreated(String schemaName) {
lsnrs.forEach(lsnr -> lsnr.onSchemaCreated(schemaName));
}
- /** {@inheritDoc} */
+ /**
+ * {@inheritDoc}
+ */
@Override public void onSchemaDropped(String schemaName) {
lsnrs.forEach(lsnr -> lsnr.onSchemaCreated(schemaName));
}
- /** {@inheritDoc} */
+ /**
+ * {@inheritDoc}
+ */
@Override public void onSqlTypeCreated(
String schemaName,
GridQueryTypeDescriptor typeDesc,
@@ -986,7 +979,9 @@ public class SchemaManager implements GridQuerySchemaManager {
lsnrs.forEach(lsnr -> lsnr.onSqlTypeCreated(schemaName, typeDesc, cacheInfo));
}
- /** {@inheritDoc} */
+ /**
+ * {@inheritDoc}
+ */
@Override public void onSqlTypeUpdated(
String schemaName,
GridQueryTypeDescriptor typeDesc,
@@ -995,20 +990,80 @@ public class SchemaManager implements GridQuerySchemaManager {
lsnrs.forEach(lsnr -> lsnr.onSqlTypeUpdated(schemaName, typeDesc, cacheInfo));
}
- /** {@inheritDoc} */
+ /**
+ * {@inheritDoc}
+ */
@Override public void onSqlTypeDropped(String schemaName, GridQueryTypeDescriptor typeDescriptor) {
lsnrs.forEach(lsnr -> lsnr.onSqlTypeDropped(schemaName, typeDescriptor));
}
- /** {@inheritDoc} */
+ /**
+ * {@inheritDoc}
+ */
@Override public void onIndexCreated(String schemaName, String tblName, String idxName,
GridQueryIndexDescriptor idxDesc, org.apache.ignite.internal.cache.query.index.Index idx) {
lsnrs.forEach(lsnr -> lsnr.onIndexCreated(schemaName, tblName, idxName, idxDesc, idx));
}
- /** {@inheritDoc} */
+ /**
+ * {@inheritDoc}
+ */
@Override public void onIndexDropped(String schemaName, String tblName, String idxName) {
lsnrs.forEach(lsnr -> lsnr.onIndexDropped(schemaName, tblName, idxName));
}
}
+
+ /**
+ * Register listener for drop columns event.
+ *
+ * @param lsnr Drop columns event listener.
+ */
+ public void registerDropColumnsListener(@NotNull BiConsumer<GridH2Table, List<String>> lsnr) {
+ requireNonNull(lsnr, "Drop columns listener should be not-null.");
+
+ dropColsLsnrs.add(lsnr);
+ }
+
+ /**
+ * Unregister listener for drop columns event.
+ *
+ * @param lsnr Drop columns event listener.
+ */
+ public void unregisterDropColumnsListener(@NotNull BiConsumer<GridH2Table, List<String>> lsnr) {
+ requireNonNull(lsnr, "Drop columns listener should be not-null.");
+
+ dropColsLsnrs.remove(lsnr);
+ }
+
+ /**
+ * Register listener for drop table event.
+ *
+ * @param lsnr Drop table event listener.
+ */
+ public void registerDropTableListener(@NotNull BiConsumer<String, String> lsnr) {
+ requireNonNull(lsnr, "Drop table listener should be not-null.");
+
+ dropTblLsnrs.add(lsnr);
+ }
+
+ /**
+ * Unregister listener for drop table event.
+ *
+ * @param lsnr Drop table event listener.
+ */
+ public void unregisterDropTableListener(@NotNull BiConsumer<String, String> lsnr) {
+ requireNonNull(lsnr, "Drop table listener should be not-null.");
+
+ dropTblLsnrs.remove(lsnr);
+ }
+
+ /**
+ * Fire each listener after table drop.
+ *
+ * @param schema Dropped table schema.
+ * @param tblName Dropped table name.
+ */
+ private void afterDropTable(String schema, String tblName) {
+ dropTblLsnrs.forEach(l -> l.accept(schema, tblName));
+ }
}