| /* |
| * 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; |
| |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.Callable; |
| import javax.cache.Cache; |
| import javax.cache.CacheException; |
| import javax.cache.processor.EntryProcessor; |
| import javax.cache.processor.EntryProcessorException; |
| import javax.cache.processor.EntryProcessorResult; |
| import javax.cache.processor.MutableEntry; |
| import org.apache.ignite.Ignite; |
| import org.apache.ignite.IgniteCache; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.cache.CacheAtomicityMode; |
| import org.apache.ignite.cache.CacheInterceptor; |
| import org.apache.ignite.cache.CacheMode; |
| import org.apache.ignite.cache.CacheWriteSynchronizationMode; |
| import org.apache.ignite.cache.QueryEntity; |
| import org.apache.ignite.cache.query.SqlFieldsQuery; |
| import org.apache.ignite.cache.query.annotations.QuerySqlField; |
| import org.apache.ignite.cache.store.CacheStoreAdapter; |
| import org.apache.ignite.configuration.CacheConfiguration; |
| import org.apache.ignite.configuration.DataStorageConfiguration; |
| import org.apache.ignite.configuration.IgniteConfiguration; |
| import org.apache.ignite.configuration.NearCacheConfiguration; |
| import org.apache.ignite.internal.IgniteEx; |
| import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor; |
| import org.apache.ignite.internal.processors.cache.index.AbstractIndexingCommonTest; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.X; |
| import org.apache.ignite.lang.IgniteBiInClosure; |
| import org.apache.ignite.lang.IgniteBiTuple; |
| import org.apache.ignite.testframework.GridTestUtils; |
| import org.apache.ignite.transactions.Transaction; |
| import org.apache.ignite.transactions.TransactionConcurrency; |
| import org.apache.ignite.transactions.TransactionIsolation; |
| import org.jetbrains.annotations.Nullable; |
| import org.junit.Test; |
| |
| /** */ |
| public class IgniteSqlNotNullConstraintTest extends AbstractIndexingCommonTest { |
| /** Name of client node. */ |
| private static final String NODE_CLIENT = "client"; |
| |
| /** Number of server nodes. */ |
| private static final int NODE_COUNT = 2; |
| |
| /** Cache prefix. */ |
| private static final String CACHE_PREFIX = "person"; |
| |
| /** Transactional person cache. */ |
| private static final String CACHE_PERSON = "person-PARTITIONED-TRANSACTIONAL"; |
| |
| /** Name of SQL table. */ |
| private static final String TABLE_PERSON = "\"" + CACHE_PERSON + "\".\"PERSON\""; |
| |
| /** Template of cache with read-through setting. */ |
| private static final String CACHE_READ_THROUGH = "cacheReadThrough"; |
| |
| /** Template of cache with interceptor setting. */ |
| private static final String CACHE_INTERCEPTOR = "cacheInterceptor"; |
| |
| /** Expected error message. */ |
| private static final String ERR_MSG = "Null value is not allowed for column 'NAME'"; |
| |
| /** Expected error message for read-through restriction. */ |
| private static final String READ_THROUGH_ERR_MSG = "NOT NULL constraint is not supported when " + |
| "CacheConfiguration.readThrough is enabled."; |
| |
| /** Expected error message for cache interceptor restriction. */ |
| private static final String INTERCEPTOR_ERR_MSG = "NOT NULL constraint is not supported when " + |
| "CacheConfiguration.interceptor is set."; |
| |
| /** Name of the node which configuration includes restricted cache config. */ |
| private static final String READ_THROUGH_CFG_NODE_NAME = "nodeCacheReadThrough"; |
| |
| /** Name of the node which configuration includes restricted cache config. */ |
| private static final String INTERCEPTOR_CFG_NODE_NAME = "nodeCacheInterceptor"; |
| |
| /** OK value. */ |
| private final Person okValue = new Person("Name", 18); |
| |
| /** Bad value, violating constraint. */ |
| private final Person badValue = new Person(null, 25); |
| |
| /** {@inheritDoc} */ |
| @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { |
| IgniteConfiguration c = super.getConfiguration(gridName); |
| |
| List<CacheConfiguration> ccfgs = new ArrayList<>(); |
| |
| ccfgs.addAll(cacheConfigurations()); |
| |
| if (gridName.equals(READ_THROUGH_CFG_NODE_NAME)) |
| ccfgs.add(buildCacheConfigurationRestricted("BadCfgTestCacheRT", true, false, true)); |
| |
| if (gridName.equals(INTERCEPTOR_CFG_NODE_NAME)) |
| ccfgs.add(buildCacheConfigurationRestricted("BadCfgTestCacheINT", false, true, true)); |
| |
| c.setCacheConfiguration(ccfgs.toArray(new CacheConfiguration[ccfgs.size()])); |
| |
| if (gridName.equals(NODE_CLIENT)) { |
| // Not allowed to have local cache on client without memory config |
| c.setDataStorageConfiguration(new DataStorageConfiguration()); |
| } |
| |
| return c; |
| } |
| |
| /** */ |
| private List<CacheConfiguration> cacheConfigurations() { |
| List<CacheConfiguration> res = new ArrayList<>(); |
| |
| for (boolean wrt : new boolean[] {false, true}) { |
| for (boolean annot : new boolean[] {false, true}) { |
| res.add(buildCacheConfiguration(CacheMode.REPLICATED, CacheAtomicityMode.ATOMIC, false, wrt, annot)); |
| res.add(buildCacheConfiguration(CacheMode.REPLICATED, CacheAtomicityMode.TRANSACTIONAL, false, wrt, annot)); |
| |
| res.add(buildCacheConfiguration(CacheMode.PARTITIONED, CacheAtomicityMode.ATOMIC, false, wrt, annot)); |
| res.add(buildCacheConfiguration(CacheMode.PARTITIONED, CacheAtomicityMode.ATOMIC, true, wrt, annot)); |
| res.add(buildCacheConfiguration(CacheMode.PARTITIONED, CacheAtomicityMode.TRANSACTIONAL, false, wrt, annot)); |
| res.add(buildCacheConfiguration(CacheMode.PARTITIONED, CacheAtomicityMode.TRANSACTIONAL, true, wrt, annot)); |
| } |
| } |
| |
| return res; |
| } |
| |
| /** */ |
| private CacheConfiguration buildCacheConfiguration(CacheMode mode, |
| CacheAtomicityMode atomicityMode, boolean hasNear, boolean writeThrough, boolean notNullAnnotated) { |
| |
| CacheConfiguration cfg = new CacheConfiguration(CACHE_PREFIX + "-" + |
| mode.name() + "-" + atomicityMode.name() + (hasNear ? "-near" : "") + |
| (writeThrough ? "-writethrough" : "") + (notNullAnnotated ? "-annot" : "")); |
| |
| cfg.setCacheMode(mode); |
| cfg.setAtomicityMode(atomicityMode); |
| cfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC); |
| |
| QueryEntity qe = new QueryEntity(new QueryEntity(Integer.class, Person.class)); |
| |
| if (!notNullAnnotated) |
| qe.setNotNullFields(Collections.singleton("name")); |
| |
| cfg.setQueryEntities(F.asList(qe)); |
| |
| if (hasNear) |
| cfg.setNearConfiguration(new NearCacheConfiguration().setNearStartSize(100)); |
| |
| if (writeThrough) { |
| cfg.setCacheStoreFactory(singletonFactory(new TestStore())); |
| cfg.setWriteThrough(true); |
| } |
| |
| return cfg; |
| } |
| |
| /** */ |
| private CacheConfiguration buildCacheConfigurationRestricted(String cacheName, boolean readThrough, |
| boolean interceptor, boolean hasQueryEntity) { |
| CacheConfiguration cfg = new CacheConfiguration<Integer, Person>() |
| .setName(cacheName) |
| .setCacheMode(CacheMode.PARTITIONED) |
| .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL); |
| |
| if (readThrough) { |
| cfg.setCacheStoreFactory(singletonFactory(new TestStore())); |
| cfg.setReadThrough(true); |
| } |
| |
| if (interceptor) |
| cfg.setInterceptor(new TestInterceptor()); |
| |
| if (hasQueryEntity) { |
| cfg.setQueryEntities(F.asList(new QueryEntity(Integer.class, Person.class) |
| .setNotNullFields(Collections.singleton("name")))); |
| } |
| |
| return cfg; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void beforeTestsStarted() throws Exception { |
| super.beforeTestsStarted(); |
| |
| startGrids(NODE_COUNT); |
| |
| startClientGrid(NODE_CLIENT); |
| |
| // Add cache template with read-through cache store. |
| grid(NODE_CLIENT).addCacheConfiguration( |
| buildCacheConfigurationRestricted(CACHE_READ_THROUGH, true, false, false)); |
| |
| // Add cache template with cache interceptor. |
| grid(NODE_CLIENT).addCacheConfiguration( |
| buildCacheConfigurationRestricted(CACHE_INTERCEPTOR, false, true, false)); |
| |
| awaitPartitionMapExchange(); |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void afterTest() throws Exception { |
| super.afterTest(); |
| |
| cleanup(); |
| } |
| |
| /** |
| * We need to keep serialized cache store factories to share them between tests. |
| */ |
| @Override protected boolean keepSerializedObjects() { |
| return true; |
| } |
| |
| /** */ |
| @Test |
| public void testQueryEntityGetSetNotNullFields() throws Exception { |
| QueryEntity qe = new QueryEntity(); |
| |
| assertNull(qe.getNotNullFields()); |
| |
| Set<String> val = Collections.singleton("test"); |
| |
| qe.setNotNullFields(val); |
| |
| assertEquals(val, Collections.singleton("test")); |
| |
| qe.setNotNullFields(null); |
| |
| assertNull(qe.getNotNullFields()); |
| } |
| |
| /** */ |
| @Test |
| public void testQueryEntityEquals() throws Exception { |
| QueryEntity a = new QueryEntity(); |
| |
| QueryEntity b = new QueryEntity(); |
| |
| assertEquals(a, b); |
| |
| a.setNotNullFields(Collections.singleton("test")); |
| |
| assertFalse(a.equals(b)); |
| |
| b.setNotNullFields(Collections.singleton("test")); |
| |
| assertTrue(a.equals(b)); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxPut() throws Exception { |
| executeWithAllCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| cache.put(key1, okValue); |
| |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| cache.put(key2, badValue); |
| |
| return null; |
| } |
| }, IgniteCheckedException.class, ERR_MSG); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxPutIfAbsent() throws Exception { |
| executeWithAllCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| cache.putIfAbsent(key1, badValue); |
| |
| return null; |
| } |
| }, IgniteCheckedException.class, ERR_MSG); |
| |
| assertEquals(0, cache.size()); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxGetAndPut() throws Exception { |
| executeWithAllCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| cache.getAndPut(key1, badValue); |
| |
| return null; |
| } |
| }, IgniteCheckedException.class, ERR_MSG); |
| |
| assertEquals(0, cache.size()); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxGetAndPutIfAbsent() throws Exception { |
| executeWithAllCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| assertEquals(0, cache.size()); |
| |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return cache.getAndPutIfAbsent(key1, badValue); |
| } |
| }, IgniteCheckedException.class, ERR_MSG); |
| |
| assertEquals(0, cache.size()); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxReplace() throws Exception { |
| executeWithAllCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| cache.put(key1, okValue); |
| |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return cache.replace(key1, badValue); |
| } |
| }, IgniteCheckedException.class, ERR_MSG); |
| |
| assertEquals(1, cache.size()); |
| assertEquals(okValue, cache.get(key1)); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxGetAndReplace() throws Exception { |
| executeWithAllCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| cache.put(key1, okValue); |
| |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return cache.getAndReplace(key1, badValue); |
| } |
| }, IgniteCheckedException.class, ERR_MSG); |
| |
| assertEquals(1, cache.size()); |
| assertEquals(okValue, cache.get(key1)); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxPutAll() throws Exception { |
| doAtomicOrImplicitTxPutAll(F.asMap(1, okValue, 5, badValue), 1); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxPutAllForSingleValue() throws Exception { |
| doAtomicOrImplicitTxPutAll(F.asMap(5, badValue), 0); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxInvoke() throws Exception { |
| executeWithAllCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return cache.invoke(key1, new TestEntryProcessor(badValue)); |
| } |
| }, IgniteCheckedException.class, ERR_MSG); |
| |
| assertEquals(0, cache.size()); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxInvokeAll() throws Exception { |
| executeWithAllCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| final Map<Integer, EntryProcessorResult<Object>> r = cache.invokeAll(F.asMap( |
| key1, new TestEntryProcessor(okValue), |
| key2, new TestEntryProcessor(badValue))); |
| |
| assertNotNull(r); |
| |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return r.get(key2).get(); |
| } |
| }, IgniteCheckedException.class, ERR_MSG); |
| |
| assertEquals(1, cache.size()); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testTxPutCreate() throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| GridTestUtils.assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| cache.put(key1, okValue); |
| cache.put(key2, badValue); |
| |
| tx.commit(); |
| } |
| |
| assertEquals(0, cache.size()); |
| |
| return null; |
| } |
| }, CacheException.class, ERR_MSG); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testTxPutUpdate() throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| GridTestUtils.assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| cache.put(key1, okValue); |
| cache.put(key2, okValue); |
| cache.put(key2, badValue); |
| |
| tx.commit(); |
| } |
| |
| assertEquals(0, cache.size()); |
| |
| return null; |
| } |
| }, CacheException.class, ERR_MSG); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testTxPutIfAbsent() throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| GridTestUtils.assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| cache.putIfAbsent(key1, badValue); |
| |
| tx.commit(); |
| } |
| |
| assertEquals(0, cache.size()); |
| |
| return null; |
| } |
| }, CacheException.class, ERR_MSG); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testTxGetAndPut() throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| GridTestUtils.assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| cache.getAndPut(key1, badValue); |
| |
| tx.commit(); |
| } |
| |
| assertEquals(0, cache.size()); |
| |
| return null; |
| } |
| }, CacheException.class, ERR_MSG); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testTxGetAndPutIfAbsent() throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| GridTestUtils.assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| cache.getAndPutIfAbsent(key1, badValue); |
| |
| tx.commit(); |
| } |
| |
| assertEquals(0, cache.size()); |
| |
| return null; |
| } |
| }, CacheException.class, ERR_MSG); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testTxReplace() throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| GridTestUtils.assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| cache.put(1, okValue); |
| |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| cache.replace(key1, badValue); |
| |
| tx.commit(); |
| } |
| |
| assertEquals(1, cache.size()); |
| assertEquals(okValue, cache.get(key1)); |
| |
| return null; |
| } |
| }, CacheException.class, ERR_MSG); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testTxGetAndReplace() throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| GridTestUtils.assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| cache.put(key1, okValue); |
| |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| cache.getAndReplace(key1, badValue); |
| |
| tx.commit(); |
| } |
| |
| assertEquals(1, cache.size()); |
| assertEquals(okValue, cache.get(key1)); |
| |
| return null; |
| } |
| }, CacheException.class, ERR_MSG); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testTxPutAll() throws Exception { |
| doTxPutAll(F.asMap(1, okValue, 5, badValue)); |
| } |
| |
| /** */ |
| @Test |
| public void testTxPutAllForSingleValue() throws Exception { |
| doTxPutAll(F.asMap(5, badValue)); |
| } |
| |
| /** */ |
| @Test |
| public void testTxInvoke() throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| cache.put(key1, okValue); |
| |
| GridTestUtils.assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| cache.invoke(key1, new TestEntryProcessor(badValue)); |
| |
| tx.commit(); |
| } |
| |
| return null; |
| } |
| }, EntryProcessorException.class, ERR_MSG); |
| |
| assertEquals(1, cache.size()); |
| assertEquals(okValue, cache.get(key1)); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testTxInvokeAll() throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| final Map<Integer, EntryProcessorResult<Object>> r = cache.invokeAll(F.asMap( |
| key1, new TestEntryProcessor(okValue), |
| key2, new TestEntryProcessor(badValue))); |
| |
| assertNotNull(r); |
| |
| GridTestUtils.assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return r.get(key2).get(); |
| } |
| }, EntryProcessorException.class, ERR_MSG); |
| |
| tx.rollback(); |
| } |
| |
| assertEquals(0, cache.size()); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxInvokeDelete() throws Exception { |
| executeWithAllCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| cache.put(key1, okValue); |
| |
| cache.invoke(key1, new TestEntryProcessor(null)); |
| |
| assertEquals(0, cache.size()); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicOrImplicitTxInvokeAllDelete() throws Exception { |
| executeWithAllCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| cache.put(key1, okValue); |
| cache.put(key2, okValue); |
| |
| cache.invokeAll(F.asMap( |
| key1, new TestEntryProcessor(null), |
| key2, new TestEntryProcessor(null))); |
| |
| assertEquals(0, cache.size()); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testTxInvokeDelete() throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| cache.put(key1, okValue); |
| |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| cache.invoke(key1, new TestEntryProcessor(null)); |
| |
| tx.commit(); |
| } |
| |
| assertEquals(0, cache.size()); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testTxInvokeAllDelete() throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| cache.put(key1, okValue); |
| cache.put(key2, okValue); |
| |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| cache.invokeAll(F.asMap( |
| key1, new TestEntryProcessor(null), |
| key2, new TestEntryProcessor(null))); |
| |
| tx.commit(); |
| } |
| |
| assertEquals(0, cache.size()); |
| } |
| }); |
| } |
| |
| /** */ |
| @Test |
| public void testDynamicTableCreateNotNullFieldsAllowed() throws Exception { |
| executeSql("CREATE TABLE test(id INT PRIMARY KEY, field INT NOT NULL)"); |
| |
| String cacheName = QueryUtils.createTableCacheName("PUBLIC", "TEST"); |
| |
| IgniteEx client = grid(NODE_CLIENT); |
| |
| CacheConfiguration ccfg = client.context().cache().cache(cacheName).configuration(); |
| |
| QueryEntity qe = (QueryEntity)F.first(ccfg.getQueryEntities()); |
| |
| assertEquals(Collections.singleton("FIELD"), qe.getNotNullFields()); |
| |
| checkState("PUBLIC", "TEST", "FIELD"); |
| } |
| |
| /** */ |
| @Test |
| public void testAlterTableAddColumnNotNullFieldAllowed() throws Exception { |
| executeSql("CREATE TABLE test(id INT PRIMARY KEY, age INT)"); |
| |
| executeSql("ALTER TABLE test ADD COLUMN name CHAR NOT NULL"); |
| |
| checkState("PUBLIC", "TEST", "NAME"); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicNotNullCheckDmlInsertValues() throws Exception { |
| checkNotNullCheckDmlInsertValues(CacheAtomicityMode.ATOMIC); |
| } |
| |
| /** */ |
| @Test |
| public void testTransactionalNotNullCheckDmlInsertValues() throws Exception { |
| checkNotNullCheckDmlInsertValues(CacheAtomicityMode.TRANSACTIONAL); |
| } |
| |
| /** */ |
| private void checkNotNullCheckDmlInsertValues(CacheAtomicityMode atomicityMode) throws Exception { |
| executeSql("CREATE TABLE test(id INT PRIMARY KEY, name VARCHAR NOT NULL, age INT) WITH \"atomicity=" |
| + atomicityMode.name() + "\""); |
| |
| checkNotNullInsertValues(); |
| } |
| |
| /** */ |
| @Test |
| public void testAtomicAddColumnNotNullCheckDmlInsertValues() throws Exception { |
| checkAddColumnNotNullCheckDmlInsertValues(CacheAtomicityMode.ATOMIC); |
| } |
| |
| /** */ |
| @Test |
| public void testTransactionalAddColumnNotNullCheckDmlInsertValues() throws Exception { |
| checkAddColumnNotNullCheckDmlInsertValues(CacheAtomicityMode.TRANSACTIONAL); |
| } |
| |
| /** */ |
| private void checkAddColumnNotNullCheckDmlInsertValues(CacheAtomicityMode atomicityMode) throws Exception { |
| executeSql("CREATE TABLE test(id INT PRIMARY KEY, age INT) WITH \"atomicity=" |
| + atomicityMode.name() + "\""); |
| |
| executeSql("ALTER TABLE test ADD COLUMN name VARCHAR NOT NULL"); |
| |
| checkNotNullInsertValues(); |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| private void checkNotNullInsertValues() throws Exception { |
| GridTestUtils.assertThrows(log(), new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| executeSql("INSERT INTO test(id, name, age) VALUES (2, NULLIF('a', 'a'), 2)"); |
| |
| return null; |
| } |
| }, IgniteSQLException.class, ERR_MSG); |
| |
| List<List<?>> result = executeSql("SELECT id, name, age FROM test ORDER BY id"); |
| |
| assertEquals(0, result.size()); |
| |
| executeSql("INSERT INTO test(id, name) VALUES (1, 'ok'), (2, 'ok2'), (3, 'ok3')"); |
| |
| result = executeSql("SELECT id, name FROM test ORDER BY id"); |
| |
| assertEquals(3, result.size()); |
| } |
| |
| /** */ |
| @Test |
| public void testNotNullCheckDmlInsertFromSelect() throws Exception { |
| executeSql("CREATE TABLE test(id INT PRIMARY KEY, name VARCHAR, age INT)"); |
| |
| executeSql("INSERT INTO test(id, name, age) VALUES (2, null, 25)"); |
| |
| GridTestUtils.assertThrows(log(), new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return executeSql("INSERT INTO " + TABLE_PERSON + |
| "(_key, name, age) " + |
| "SELECT id, name, age FROM test"); |
| } |
| }, IgniteSQLException.class, ERR_MSG); |
| |
| List<List<?>> result = executeSql("SELECT _key, name FROM " + TABLE_PERSON + " ORDER BY _key"); |
| |
| assertEquals(0, result.size()); |
| |
| executeSql("INSERT INTO test(id, name, age) VALUES (1, 'Macy', 25), (3, 'John', 30)"); |
| executeSql("DELETE FROM test WHERE id = 2"); |
| |
| result = executeSql("INSERT INTO " + TABLE_PERSON + "(_key, name, age) " + "SELECT id, name, age FROM test"); |
| |
| assertEquals(2L, result.get(0).get(0)); |
| } |
| |
| /** */ |
| @Test |
| public void testNotNullCheckDmlUpdateValues() throws Exception { |
| executeSql("CREATE TABLE test(id INT PRIMARY KEY, name VARCHAR NOT NULL)"); |
| |
| executeSql("INSERT INTO test(id, name) VALUES (1, 'John')"); |
| |
| GridTestUtils.assertThrows(log(), new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return executeSql("UPDATE test SET name = NULLIF(id, 1) WHERE id = 1"); |
| } |
| }, IgniteSQLException.class, ERR_MSG); |
| |
| List<List<?>> result = executeSql("SELECT id, name FROM test"); |
| |
| assertEquals(1, result.size()); |
| assertEquals(1, result.get(0).get(0)); |
| assertEquals("John", result.get(0).get(1)); |
| |
| executeSql("UPDATE test SET name = 'James' WHERE id = 1"); |
| |
| result = executeSql("SELECT id, name FROM test"); |
| |
| assertEquals(1, result.get(0).get(0)); |
| assertEquals("James", result.get(0).get(1)); |
| } |
| |
| /** */ |
| @Test |
| public void testNotNullCheckDmlUpdateFromSelect() throws Exception { |
| executeSql("CREATE TABLE src(id INT PRIMARY KEY, name VARCHAR)"); |
| executeSql("CREATE TABLE dest(id INT PRIMARY KEY, name VARCHAR NOT NULL)"); |
| |
| executeSql("INSERT INTO dest(id, name) VALUES (1, 'William'), (2, 'Warren'), (3, 'Robert')"); |
| executeSql("INSERT INTO src(id, name) VALUES (1, 'Bill'), (2, null), (3, 'Bob')"); |
| |
| GridTestUtils.assertThrows(log(), new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return executeSql("UPDATE dest p " + |
| "SET (name) = (SELECT name FROM src t WHERE p.id = t.id) " + |
| "WHERE p.id = 2"); |
| } |
| }, IgniteSQLException.class, ERR_MSG); |
| |
| List<List<?>> result = executeSql("SELECT id, name FROM dest ORDER BY id"); |
| |
| assertEquals(3, result.size()); |
| |
| assertEquals(1, result.get(0).get(0)); |
| assertEquals("William", result.get(0).get(1)); |
| |
| assertEquals(2, result.get(1).get(0)); |
| assertEquals("Warren", result.get(1).get(1)); |
| |
| assertEquals(3, result.get(2).get(0)); |
| assertEquals("Robert", result.get(2).get(1)); |
| |
| executeSql("UPDATE src SET name = 'Ren' WHERE id = 2"); |
| |
| executeSql("UPDATE dest p SET (name) = (SELECT name FROM src t WHERE p.id = t.id)"); |
| |
| result = executeSql("SELECT id, name FROM dest ORDER BY id"); |
| |
| assertEquals(3, result.size()); |
| |
| assertEquals(1, result.get(0).get(0)); |
| assertEquals("Bill", result.get(0).get(1)); |
| |
| assertEquals(2, result.get(1).get(0)); |
| assertEquals("Ren", result.get(1).get(1)); |
| |
| assertEquals(3, result.get(2).get(0)); |
| assertEquals("Bob", result.get(2).get(1)); |
| } |
| |
| /** Check QueryEntity configuration fails with NOT NULL field and read-through. */ |
| @Test |
| public void testReadThroughRestrictionQueryEntity() throws Exception { |
| // Node start-up failure (read-through cache store). |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return startClientGrid(READ_THROUGH_CFG_NODE_NAME); |
| } |
| }, IgniteCheckedException.class, READ_THROUGH_ERR_MSG); |
| |
| // Dynamic cache start-up failure (read-through cache store) |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return grid(NODE_CLIENT).createCache( |
| buildCacheConfigurationRestricted("dynBadCfgCacheRT", true, false, true)); |
| } |
| }, IgniteCheckedException.class, READ_THROUGH_ERR_MSG); |
| } |
| |
| /** Check QueryEntity configuration fails with NOT NULL field and cache interceptor. */ |
| @Test |
| public void testInterceptorRestrictionQueryEntity() throws Exception { |
| // Node start-up failure (interceptor). |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return startClientGrid(INTERCEPTOR_CFG_NODE_NAME); |
| } |
| }, IgniteCheckedException.class, INTERCEPTOR_ERR_MSG); |
| |
| // Dynamic cache start-up failure (interceptor) |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return grid(NODE_CLIENT).createCache( |
| buildCacheConfigurationRestricted("dynBadCfgCacheINT", false, true, true)); |
| } |
| }, IgniteCheckedException.class, INTERCEPTOR_ERR_MSG); |
| } |
| |
| /** Check create table fails with NOT NULL field and read-through. */ |
| @Test |
| public void testReadThroughRestrictionCreateTable() throws Exception { |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return executeSql("CREATE TABLE test(id INT PRIMARY KEY, name char NOT NULL) " + |
| "WITH \"template=" + CACHE_READ_THROUGH + "\""); |
| } |
| }, IgniteSQLException.class, READ_THROUGH_ERR_MSG); |
| } |
| |
| /** Check create table fails with NOT NULL field and cache interceptor. */ |
| @Test |
| public void testInterceptorRestrictionCreateTable() throws Exception { |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return executeSql("CREATE TABLE test(id INT PRIMARY KEY, name char NOT NULL) " + |
| "WITH \"template=" + CACHE_INTERCEPTOR + "\""); |
| } |
| }, IgniteSQLException.class, INTERCEPTOR_ERR_MSG); |
| } |
| |
| /** Check alter table fails with NOT NULL field and read-through. */ |
| @Test |
| public void testReadThroughRestrictionAlterTable() throws Exception { |
| executeSql("CREATE TABLE test(id INT PRIMARY KEY, age INT) " + |
| "WITH \"template=" + CACHE_READ_THROUGH + "\""); |
| |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return executeSql("ALTER TABLE test ADD COLUMN name char NOT NULL"); |
| } |
| }, IgniteSQLException.class, READ_THROUGH_ERR_MSG); |
| } |
| |
| /** Check alter table fails with NOT NULL field and cache interceptor. */ |
| @Test |
| public void testInterceptorRestrictionAlterTable() throws Exception { |
| executeSql("CREATE TABLE test(id INT PRIMARY KEY, age INT) " + |
| "WITH \"template=" + CACHE_INTERCEPTOR + "\""); |
| |
| GridTestUtils.assertThrowsAnyCause(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| return executeSql("ALTER TABLE test ADD COLUMN name char NOT NULL"); |
| } |
| }, IgniteSQLException.class, INTERCEPTOR_ERR_MSG); |
| } |
| |
| /** */ |
| private void doAtomicOrImplicitTxPutAll(final Map<Integer, Person> values, int expAtomicCacheSize) throws Exception { |
| executeWithAllCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| Throwable t = GridTestUtils.assertThrowsWithCause(new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| cache.putAll(values); |
| |
| return null; |
| } |
| }, IgniteSQLException.class); |
| |
| IgniteSQLException ex = X.cause(t, IgniteSQLException.class); |
| |
| assertNotNull(ex); |
| |
| assertTrue(ex.getMessage().contains(ERR_MSG)); |
| |
| assertEquals(0, cache.size()); |
| } |
| }); |
| } |
| |
| /** */ |
| private void doTxPutAll(Map<Integer, Person> values) throws Exception { |
| executeWithAllTxCaches(new TestClosure() { |
| @Override public void run() throws Exception { |
| GridTestUtils.assertThrows(log, new Callable<Object>() { |
| @Override public Object call() throws Exception { |
| try (Transaction tx = ignite.transactions().txStart(concurrency, isolation)) { |
| cache.putAll(values); |
| |
| tx.commit(); |
| } |
| |
| assertEquals(0, cache.size()); |
| |
| return null; |
| } |
| }, CacheException.class, ERR_MSG); |
| } |
| }); |
| } |
| |
| /** */ |
| private void executeWithAllCaches(TestClosure clo) throws Exception { |
| for (CacheConfiguration ccfg : cacheConfigurations()) |
| executeForCache(ccfg, clo, TransactionConcurrency.OPTIMISTIC, TransactionIsolation.READ_COMMITTED); |
| } |
| |
| /** */ |
| private void executeWithAllTxCaches(final TestClosure clo) throws Exception { |
| for (CacheConfiguration ccfg : cacheConfigurations()) { |
| if (ccfg.getAtomicityMode() == CacheAtomicityMode.TRANSACTIONAL) { |
| for (TransactionConcurrency con : TransactionConcurrency.values()) { |
| for (TransactionIsolation iso : TransactionIsolation.values()) |
| executeForCache(ccfg, clo, con, iso); |
| } |
| } |
| } |
| } |
| |
| /** */ |
| private void executeForCache(CacheConfiguration ccfg, TestClosure clo, TransactionConcurrency concurrency, |
| TransactionIsolation isolation) throws Exception { |
| |
| Ignite ignite = grid(NODE_CLIENT); |
| executeForNodeAndCache(ccfg, ignite, clo, concurrency, isolation); |
| |
| for (int node = 0; node < NODE_COUNT; node++) { |
| ignite = grid(node); |
| |
| executeForNodeAndCache(ccfg, ignite, clo, concurrency, isolation); |
| } |
| } |
| |
| /** */ |
| private void executeForNodeAndCache(CacheConfiguration ccfg, Ignite ignite, TestClosure clo, |
| TransactionConcurrency concurrency, TransactionIsolation isolation) throws Exception { |
| String cacheName = ccfg.getName(); |
| |
| IgniteCache cache; |
| |
| if (ignite.configuration().isClientMode() && |
| ccfg.getCacheMode() == CacheMode.PARTITIONED && |
| ccfg.getNearConfiguration() != null) |
| cache = ignite.getOrCreateNearCache(ccfg.getName(), ccfg.getNearConfiguration()); |
| else |
| cache = ignite.cache(ccfg.getName()); |
| |
| cache.removeAll(); |
| |
| assertEquals(0, cache.size()); |
| |
| clo.configure(ignite, cache, concurrency, isolation); |
| |
| log.info("Running test with node " + ignite.name() + ", cache " + cacheName); |
| |
| clo.key1 = 1; |
| clo.key2 = 4; |
| |
| clo.run(); |
| } |
| |
| /** */ |
| private List<List<?>> executeSql(String sqlText) throws Exception { |
| GridQueryProcessor qryProc = grid(NODE_CLIENT).context().query(); |
| |
| return qryProc.querySqlFields(new SqlFieldsQuery(sqlText), true).getAll(); |
| } |
| |
| /** */ |
| private void cleanup() throws Exception { |
| for (CacheConfiguration ccfg : cacheConfigurations()) { |
| String cacheName = ccfg.getName(); |
| |
| if (ccfg.getCacheMode() == CacheMode.PARTITIONED && ccfg.getNearConfiguration() != null) { |
| IgniteCache cache = grid(NODE_CLIENT).getOrCreateNearCache(cacheName, ccfg.getNearConfiguration()); |
| |
| cache.clear(); |
| } |
| |
| grid(NODE_CLIENT).cache(cacheName).clear(); |
| } |
| |
| executeSql("DROP TABLE test IF EXISTS"); |
| } |
| |
| /** */ |
| private void checkState(String schemaName, String tableName, String fieldName) { |
| IgniteEx client = grid(NODE_CLIENT); |
| |
| checkNodeState(client, schemaName, tableName, fieldName); |
| |
| for (int i = 0; i < NODE_COUNT; i++) |
| checkNodeState(grid(i), schemaName, tableName, fieldName); |
| } |
| |
| /** */ |
| private void checkNodeState(IgniteEx node, String schemaName, String tableName, String fieldName) { |
| String cacheName = F.eq(schemaName, QueryUtils.DFLT_SCHEMA) ? |
| QueryUtils.createTableCacheName(schemaName, tableName) : schemaName; |
| |
| DynamicCacheDescriptor desc = node.context().cache().cacheDescriptor(cacheName); |
| |
| assertNotNull("Cache descriptor not found", desc); |
| |
| QuerySchema schema = desc.schema(); |
| |
| assertNotNull(schema); |
| |
| QueryEntity entity = null; |
| |
| for (QueryEntity e : schema.entities()) { |
| if (F.eq(tableName, e.getTableName())) { |
| entity = e; |
| |
| break; |
| } |
| } |
| |
| assertNotNull(entity); |
| |
| assertNotNull(entity.getNotNullFields()); |
| |
| assertTrue(entity.getNotNullFields().contains(fieldName)); |
| } |
| |
| /** */ |
| public static class Person { |
| /** */ |
| @QuerySqlField(notNull = true) |
| private String name; |
| |
| /** */ |
| @QuerySqlField |
| private int age; |
| |
| /** */ |
| public Person(String name, int age) { |
| this.name = name; |
| this.age = age; |
| } |
| |
| /** */ |
| @Override public int hashCode() { |
| return (name == null ? 0 : name.hashCode()) ^ age; |
| } |
| |
| /** */ |
| @Override public boolean equals(Object o) { |
| if (this == o) |
| return true; |
| |
| if (!(o instanceof Person)) |
| return false; |
| |
| Person other = (Person)o; |
| |
| return F.eq(other.name, name) && other.age == age; |
| } |
| } |
| |
| /** */ |
| public class TestEntryProcessor implements EntryProcessor<Integer, Person, Object> { |
| /** Value to set. */ |
| private Person value; |
| |
| /** */ |
| public TestEntryProcessor(Person value) { |
| this.value = value; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Object process(MutableEntry<Integer, Person> entry, |
| Object... objects) throws EntryProcessorException { |
| |
| if (value == null) |
| entry.remove(); |
| else |
| entry.setValue(value); |
| |
| return null; |
| } |
| } |
| |
| /** */ |
| public abstract class TestClosure { |
| /** */ |
| protected Ignite ignite; |
| |
| /** */ |
| protected IgniteCache<Integer, Person> cache; |
| |
| /** */ |
| protected TransactionConcurrency concurrency; |
| |
| /** */ |
| protected TransactionIsolation isolation; |
| |
| /** */ |
| public int key1; |
| |
| /** */ |
| public int key2; |
| |
| /** */ |
| public void configure(Ignite ignite, IgniteCache<Integer, Person> cache, TransactionConcurrency concurrency, |
| TransactionIsolation isolation) { |
| this.ignite = ignite; |
| this.cache = cache; |
| this.concurrency = concurrency; |
| this.isolation = isolation; |
| } |
| |
| /** */ |
| public abstract void run() throws Exception; |
| } |
| |
| /** |
| * Test cache store stub. |
| */ |
| private static class TestStore extends CacheStoreAdapter<Integer, Person> { |
| /** {@inheritDoc} */ |
| @Override public void loadCache(IgniteBiInClosure<Integer, Person> clo, @Nullable Object... args) { |
| // No-op |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public Person load(Integer key) { |
| return null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void write(javax.cache.Cache.Entry<? extends Integer, ? extends Person> e) { |
| // No-op |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void delete(Object key) { |
| // No-op |
| } |
| } |
| |
| /** |
| * Test interceptor stub. |
| */ |
| private static class TestInterceptor implements CacheInterceptor<Integer, Person> { |
| /** {@inheritDoc} */ |
| @Nullable @Override public Person onGet(Integer key, @Nullable Person val) { |
| return null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public Person onBeforePut(Cache.Entry<Integer, Person> entry, Person newVal) { |
| return null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onAfterPut(Cache.Entry<Integer, Person> entry) { |
| // No-op |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public IgniteBiTuple<Boolean, Person> onBeforeRemove(Cache.Entry<Integer, Person> entry) { |
| return null; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void onAfterRemove(Cache.Entry<Integer, Person> entry) { |
| // No-op |
| } |
| } |
| } |
