ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject ignite git commit: ignite-2982 Unwrap offheap object before pass it to indexing (this closes #693)
Date Fri, 13 May 2016 07:20:13 GMT
Repository: ignite
Updated Branches:
  refs/heads/ignite-1.6 9b606b887 -> b4a5d3766


ignite-2982 Unwrap offheap object before pass it to indexing (this closes #693)


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/b4a5d376
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/b4a5d376
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/b4a5d376

Branch: refs/heads/ignite-1.6
Commit: b4a5d376608fdc36ab0a5dd8157834771a006d37
Parents: 9b606b8
Author: sboikov <sboikov@gridgain.com>
Authored: Fri May 13 10:20:00 2016 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Fri May 13 10:20:00 2016 +0300

----------------------------------------------------------------------
 .../processors/cache/GridCacheMapEntry.java     |   2 +-
 .../processors/cache/GridCacheProcessor.java    |   2 +-
 .../IgniteCacheBinaryObjectsScanSelfTest.java   |   2 +
 .../distributed/IgniteCacheCreatePutTest.java   |   6 +-
 .../processors/query/h2/opt/GridH2Table.java    |   1 -
 .../cache/AtomicBinaryOffheapBatchTest.java     | 384 -------------------
 .../CacheOffheapBatchIndexingBaseTest.java      | 313 +++++++++++++++
 .../CacheOffheapBatchIndexingMultiTypeTest.java | 108 ++++++
 ...CacheOffheapBatchIndexingSingleTypeTest.java | 161 ++++++++
 .../IgniteCacheQuerySelfTestSuite2.java         |   2 +
 10 files changed, 593 insertions(+), 388 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/b4a5d376/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index 0ffaa44..2a29657 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -3869,7 +3869,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter
impleme
             GridCacheQueryManager<?, ?> qryMgr = cctx.queries();
 
             if (qryMgr.enabled())
-                qryMgr.remove(key(), prevVal);
+                qryMgr.remove(key(), (CacheObject)cctx.unwrapTemporary(prevVal));
         }
         catch (IgniteCheckedException e) {
             throw new GridCacheIndexUpdateException(e);

http://git-wip-us.apache.org/repos/asf/ignite/blob/b4a5d376/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
index 468d357..0f8e3f6 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
@@ -336,7 +336,7 @@ public class GridCacheProcessor extends GridProcessorAdapter {
             perf.add("Disable near cache (set 'nearConfiguration' to null)", cfg.getNearConfiguration()
== null);
 
             if (cfg.getAffinity() != null)
-                perf.add("Decrease number of backups (set 'keyBackups' to 0)", cfg.getBackups()
== 0);
+                perf.add("Decrease number of backups (set 'backups' to 0)", cfg.getBackups()
== 0);
         }
 
         // Suppress warning if at least one ATOMIC cache found.

http://git-wip-us.apache.org/repos/asf/ignite/blob/b4a5d376/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheBinaryObjectsScanSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheBinaryObjectsScanSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheBinaryObjectsScanSelfTest.java
index 07f3833..7743882 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheBinaryObjectsScanSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheBinaryObjectsScanSelfTest.java
@@ -60,6 +60,8 @@ public class IgniteCacheBinaryObjectsScanSelfTest extends GridCommonAbstractTest
     /** {@inheritDoc} */
     @Override protected void afterTestsStopped() throws Exception {
         ldr = null;
+
+        stopAllGrids();
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/b4a5d376/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheCreatePutTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheCreatePutTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheCreatePutTest.java
index efba34a..2f700f3 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheCreatePutTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheCreatePutTest.java
@@ -28,6 +28,7 @@ import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.marshaller.optimized.OptimizedMarshaller;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
 import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
 import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -57,6 +58,8 @@ public class IgniteCacheCreatePutTest extends GridCommonAbstractTest {
     protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
         IgniteConfiguration cfg = super.getConfiguration(gridName);
 
+        ((TcpCommunicationSpi)cfg.getCommunicationSpi()).setSharedMemoryPort(-1);
+
         cfg.setPeerClassLoadingEnabled(false);
 
         TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
@@ -74,6 +77,7 @@ public class IgniteCacheCreatePutTest extends GridCommonAbstractTest {
         ccfg.setName("cache*");
         ccfg.setCacheMode(PARTITIONED);
         ccfg.setBackups(1);
+        ccfg.setWriteSynchronizationMode(FULL_SYNC);
 
         cfg.setCacheConfiguration(ccfg);
 
@@ -84,7 +88,7 @@ public class IgniteCacheCreatePutTest extends GridCommonAbstractTest {
 
     /** {@inheritDoc} */
     @Override protected long getTestTimeout() {
-        return 3 * 60 * 1000L;
+        return 5 * 60 * 1000L;
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/b4a5d376/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Table.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Table.java
b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Table.java
index f1e5b16..bea4dd8 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Table.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2Table.java
@@ -25,7 +25,6 @@ import java.util.Collections;
 import java.util.Set;
 import java.util.concurrent.ConcurrentNavigableMap;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import org.apache.ignite.IgniteCheckedException;

http://git-wip-us.apache.org/repos/asf/ignite/blob/b4a5d376/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/AtomicBinaryOffheapBatchTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/AtomicBinaryOffheapBatchTest.java
b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/AtomicBinaryOffheapBatchTest.java
deleted file mode 100644
index fb2d95d..0000000
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/AtomicBinaryOffheapBatchTest.java
+++ /dev/null
@@ -1,384 +0,0 @@
-/*
- * 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.cache;
-
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.IgniteDataStreamer;
-import org.apache.ignite.binary.BinaryObjectException;
-import org.apache.ignite.binary.BinaryReader;
-import org.apache.ignite.binary.BinaryWriter;
-import org.apache.ignite.binary.Binarylizable;
-import org.apache.ignite.cache.CacheAtomicWriteOrderMode;
-import org.apache.ignite.cache.CacheAtomicityMode;
-import org.apache.ignite.cache.CacheMemoryMode;
-import org.apache.ignite.cache.CacheMode;
-import org.apache.ignite.cache.CacheWriteSynchronizationMode;
-import org.apache.ignite.cache.query.annotations.QuerySqlField;
-import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.configuration.NearCacheConfiguration;
-import org.apache.ignite.internal.binary.BinaryMarshaller;
-
-import javax.cache.processor.EntryProcessor;
-import javax.cache.processor.EntryProcessorException;
-import javax.cache.processor.MutableEntry;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
-import java.util.TreeSet;
-
-/**
- * The test are checking batch operation onto atomic offheap cache with per certain key and
value types.
- */
-public class AtomicBinaryOffheapBatchTest extends IgniteCacheAbstractTest {
-
-    /**
-     * Size of batch in operation
-     */
-    public static final int BATCH_SIZE = 500;
-
-    @Override
-    protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
-        IgniteConfiguration cfg = super.getConfiguration(gridName);
-
-        cfg.setMarshaller(new BinaryMarshaller());
-        cfg.setPeerClassLoadingEnabled(false);
-
-        return cfg;
-    }
-
-    @Override
-    protected CacheConfiguration cacheConfiguration(String gridName) throws Exception {
-        CacheConfiguration cfg = super.cacheConfiguration(gridName);
-
-        cfg.setMemoryMode(memoryMode());
-
-        cfg.setIndexedTypes(Integer.class, Person.class, Integer.class, Organization.class);
-
-        cfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.PRIMARY_SYNC);
-
-        return cfg;
-    }
-
-    /** {@inheritDoc} */
-    @Override
-    protected int gridCount() {
-        return 1;
-    }
-
-    /** {@inheritDoc} */
-    @Override
-    protected CacheMode cacheMode() {
-        return CacheMode.PARTITIONED;
-    }
-
-    /** {@inheritDoc} */
-    @Override
-    protected CacheAtomicityMode atomicityMode() {
-        return CacheAtomicityMode.ATOMIC;
-    }
-
-    /** {@inheritDoc} */
-    @Override
-    protected NearCacheConfiguration nearConfiguration() {
-        return null;
-    }
-
-    /**
-     * @return Cache memory mode.
-     */
-    protected CacheMemoryMode memoryMode() {
-        return CacheMemoryMode.OFFHEAP_TIERED;
-    }
-
-    /** {@inheritDoc} */
-    @Override
-    protected CacheAtomicWriteOrderMode atomicWriteOrderMode() {
-        return CacheAtomicWriteOrderMode.PRIMARY;
-    }
-
-    /**
-     * Test method.
-     *
-     * @throws Exception If fail.
-     */
-    public void testBatchOperations() throws Exception {
-        try (IgniteCache defaultCache = ignite(0).cache(null)) {
-            loadingCacheAnyDate();
-
-            for (int cnt = 0; cnt < 200; cnt++) {
-                Map<Integer, Person> putMap1 = new TreeMap<>();
-                for (int i = 0; i < BATCH_SIZE; i++)
-                    putMap1.put(i, new Person(i, i + 1, String.valueOf(i), String.valueOf(i
+ 1), i / 0.99));
-
-                defaultCache.putAll(putMap1);
-
-                Map<Integer, Organization> putMap2 = new TreeMap<>();
-                for (int i = BATCH_SIZE / 2; i < BATCH_SIZE * 3 / 2; i++)
-                    putMap2.put(i, new Organization(i, String.valueOf(i)));
-
-                defaultCache.putAll(putMap2);
-
-                Set<Integer> keySet = new TreeSet<>();
-
-                for (int i = 0; i < BATCH_SIZE * 3 / 2; i += 2)
-                    keySet.add(i);
-
-                defaultCache.removeAll(keySet);
-
-                keySet = new TreeSet<>();
-
-                for (int i = 1; i < BATCH_SIZE * 3 / 2; i += 2)
-                    keySet.add(i);
-
-                defaultCache.invokeAll(keySet, new EntryProcessor() {
-                    @Override
-                    public Object process(MutableEntry entry, Object... arguments) throws
EntryProcessorException {
-                        Object value = entry.getValue();
-                        entry.remove();
-                        return value;
-                    }
-                });
-            }
-        }
-    }
-
-    /**
-     * Loading date into cache
-     */
-    private void loadingCacheAnyDate() {
-        try (IgniteDataStreamer streamer = ignite(0).dataStreamer(null)) {
-            for (int i = 0; i < 30_000; i++) {
-                if (i % 2 == 0)
-                    streamer.addData(i, new Person(i, i + 1, String.valueOf(i), String.valueOf(i
+ 1), i / 0.99));
-                else
-                    streamer.addData(i, new Organization(i, String.valueOf(i)));
-            }
-        }
-    }
-
-    /**
-     * Ignite cache value class.
-     */
-    private static class Person implements Binarylizable {
-
-        /** Person ID. */
-        @QuerySqlField(index = true)
-        private int id;
-
-        /** Organization ID. */
-        @QuerySqlField(index = true)
-        private int orgId;
-
-        /** First name (not-indexed). */
-        @QuerySqlField
-        private String firstName;
-
-        /** Last name (not indexed). */
-        @QuerySqlField
-        private String lastName;
-
-        /** Salary. */
-        @QuerySqlField(index = true)
-        private double salary;
-
-        /**
-         * Constructs empty person.
-         */
-        public Person() {
-            // No-op.
-        }
-
-        /**
-         * Constructs person record.
-         *
-         * @param id Person ID.
-         * @param orgId Organization ID.
-         * @param firstName First name.
-         * @param lastName Last name.
-         * @param salary Salary.
-         */
-        public Person(int id, int orgId, String firstName, String lastName, double salary)
{
-            this.id = id;
-            this.orgId = orgId;
-            this.firstName = firstName;
-            this.lastName = lastName;
-            this.salary = salary;
-        }
-
-        /**
-         * @return Person id.
-         */
-        public int getId() {
-            return id;
-        }
-
-        /**
-         * @param id Person id.
-         */
-        public void setId(int id) {
-            this.id = id;
-        }
-
-        /**
-         * @return Organization id.
-         */
-        public int getOrganizationId() {
-            return orgId;
-        }
-
-        /**
-         * @param orgId Organization id.
-         */
-        public void setOrganizationId(int orgId) {
-            this.orgId = orgId;
-        }
-
-        /**
-         * @return Person first name.
-         */
-        public String getFirstName() {
-            return firstName;
-        }
-
-        /**
-         * @param firstName Person first name.
-         */
-        public void setFirstName(String firstName) {
-            this.firstName = firstName;
-        }
-
-        /**
-         * @return Person last name.
-         */
-        public String getLastName() {
-            return lastName;
-        }
-
-        /**
-         * @param lastName Person last name.
-         */
-        public void setLastName(String lastName) {
-            this.lastName = lastName;
-        }
-
-        /**
-         * @return Salary.
-         */
-        public double getSalary() {
-            return salary;
-        }
-
-        /**
-         * @param salary Salary.
-         */
-        public void setSalary(double salary) {
-            this.salary = salary;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void writeBinary(BinaryWriter writer) throws BinaryObjectException
{
-            writer.writeInt("id", id);
-            writer.writeInt("orgId", orgId);
-            writer.writeString("firstName", firstName);
-            writer.writeString("lastName", lastName);
-            writer.writeDouble("salary", salary);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readBinary(BinaryReader reader) throws BinaryObjectException
{
-            id = reader.readInt("id");
-            orgId = reader.readInt("orgId");
-            firstName = reader.readString("firstName");
-            lastName = reader.readString("lastName");
-            salary = reader.readDouble("salary");
-        }
-    }
-
-    /**
-     * Ignite cache value class with indexed field.
-     */
-    private static class Organization implements Binarylizable {
-
-        /** Organization ID. */
-        @QuerySqlField(index = true)
-        private int id;
-
-        /** Organization name. */
-        @QuerySqlField(index = true)
-        private String name;
-
-        /**
-         * Constructs empty organization.
-         */
-        public Organization() {
-            // No-op.
-        }
-
-        /**
-         * Constructs organization with given ID.
-         *
-         * @param id Organization ID.
-         * @param name Organization name.
-         */
-        public Organization(int id, String name) {
-            this.id = id;
-            this.name = name;
-        }
-
-        /**
-         * @return Organization id.
-         */
-        public int getId() {
-            return id;
-        }
-
-        /**
-         * @param id Organization id.
-         */
-        public void setId(int id) {
-            this.id = id;
-        }
-
-        /**
-         * @return Organization name.
-         */
-        public String getName() {
-            return name;
-        }
-
-        /**
-         * @param name Organization name.
-         */
-        public void setName(String name) {
-            this.name = name;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void writeBinary(BinaryWriter writer) throws BinaryObjectException
{
-            writer.writeInt("id", id);
-            writer.writeString("name", name);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readBinary(BinaryReader reader) throws BinaryObjectException
{
-            id = reader.readInt("id");
-            name = reader.readString("name");
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/ignite/blob/b4a5d376/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingBaseTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingBaseTest.java
b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingBaseTest.java
new file mode 100644
index 0000000..b550b1e
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingBaseTest.java
@@ -0,0 +1,313 @@
+/*
+ * 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.cache;
+
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.binary.BinaryObjectException;
+import org.apache.ignite.binary.BinaryReader;
+import org.apache.ignite.binary.BinaryWriter;
+import org.apache.ignite.binary.Binarylizable;
+import org.apache.ignite.cache.query.annotations.QuerySqlField;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheMemoryMode.OFFHEAP_TIERED;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+
+/**
+ * Tests various cache operations with indexing enabled.
+ */
+public abstract class CacheOffheapBatchIndexingBaseTest extends GridCommonAbstractTest {
+    /** */
+    private final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+    /**
+     * Load data into cache
+     *
+     * @param name Cache name.
+     */
+    protected void preload(String name) {
+        try (IgniteDataStreamer<Object, Object> streamer = ignite(0).dataStreamer(name))
{
+            for (int i = 0; i < 30_000; i++) {
+                if (i % 2 == 0)
+                    streamer.addData(i, new Person(i, i + 1, String.valueOf(i), String.valueOf(i
+ 1), salary(i)));
+                else
+                    streamer.addData(i, new Organization(i, String.valueOf(i)));
+            }
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception
{
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        cfg.setPeerClassLoadingEnabled(false);
+
+        ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+        return cfg;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTestsStarted() throws Exception {
+        startGrid(0);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTestsStopped() throws Exception {
+        stopAllGrids();
+    }
+
+    /**
+     * @param base Base.
+     * @return Salary.
+     */
+    protected double salary(int base) {
+        return base * 100.;
+    }
+
+    /**
+     * @param onHeapRowCacheSize on heap row cache size.
+     * @param indexedTypes indexed types for cache.
+     * @return Cache configuration.
+     */
+    protected CacheConfiguration<Object, Object> cacheConfiguration(int onHeapRowCacheSize,
Class<?>[] indexedTypes) {
+        CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>();
+
+        ccfg.setAtomicityMode(ATOMIC);
+        ccfg.setWriteSynchronizationMode(FULL_SYNC);
+        ccfg.setMemoryMode(OFFHEAP_TIERED);
+        ccfg.setSqlOnheapRowCacheSize(onHeapRowCacheSize);
+        ccfg.setIndexedTypes(indexedTypes);
+
+        return ccfg;
+    }
+
+    /**
+     * Ignite cache value class.
+     */
+    protected static class Person implements Binarylizable {
+        /** Person ID. */
+        @QuerySqlField(index = true)
+        private int id;
+
+        /** Organization ID. */
+        @QuerySqlField(index = true)
+        private int orgId;
+
+        /** First name (not-indexed). */
+        @QuerySqlField
+        private String firstName;
+
+        /** Last name (not indexed). */
+        @QuerySqlField
+        private String lastName;
+
+        /** Salary. */
+        @QuerySqlField(index = true)
+        private double salary;
+
+        /**
+         * Constructs empty person.
+         */
+        public Person() {
+            // No-op.
+        }
+
+        /**
+         * Constructs person record.
+         *
+         * @param id Person ID.
+         * @param orgId Organization ID.
+         * @param firstName First name.
+         * @param lastName Last name.
+         * @param salary Salary.
+         */
+        public Person(int id, int orgId, String firstName, String lastName, double salary)
{
+            this.id = id;
+            this.orgId = orgId;
+            this.firstName = firstName;
+            this.lastName = lastName;
+            this.salary = salary;
+        }
+
+        /**
+         * @return Person id.
+         */
+        public int getId() {
+            return id;
+        }
+
+        /**
+         * @param id Person id.
+         */
+        public void setId(int id) {
+            this.id = id;
+        }
+
+        /**
+         * @return Organization id.
+         */
+        public int getOrganizationId() {
+            return orgId;
+        }
+
+        /**
+         * @param orgId Organization id.
+         */
+        public void setOrganizationId(int orgId) {
+            this.orgId = orgId;
+        }
+
+        /**
+         * @return Person first name.
+         */
+        public String getFirstName() {
+            return firstName;
+        }
+
+        /**
+         * @param firstName Person first name.
+         */
+        public void setFirstName(String firstName) {
+            this.firstName = firstName;
+        }
+
+        /**
+         * @return Person last name.
+         */
+        public String getLastName() {
+            return lastName;
+        }
+
+        /**
+         * @param lastName Person last name.
+         */
+        public void setLastName(String lastName) {
+            this.lastName = lastName;
+        }
+
+        /**
+         * @return Salary.
+         */
+        public double getSalary() {
+            return salary;
+        }
+
+        /**
+         * @param salary Salary.
+         */
+        public void setSalary(double salary) {
+            this.salary = salary;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void writeBinary(BinaryWriter writer) throws BinaryObjectException
{
+            writer.writeInt("id", id);
+            writer.writeInt("orgId", orgId);
+            writer.writeString("firstName", firstName);
+            writer.writeString("lastName", lastName);
+            writer.writeDouble("salary", salary);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void readBinary(BinaryReader reader) throws BinaryObjectException
{
+            id = reader.readInt("id");
+            orgId = reader.readInt("orgId");
+            firstName = reader.readString("firstName");
+            lastName = reader.readString("lastName");
+            salary = reader.readDouble("salary");
+        }
+    }
+
+    /**
+     * Ignite cache value class with indexed field.
+     */
+    protected static class Organization implements Binarylizable {
+        /** Organization ID. */
+        @QuerySqlField(index = true)
+        private int id;
+
+        /** Organization name. */
+        @QuerySqlField(index = true)
+        private String name;
+
+        /**
+         * Constructs empty organization.
+         */
+        public Organization() {
+            // No-op.
+        }
+
+        /**
+         * Constructs organization with given ID.
+         *
+         * @param id Organization ID.
+         * @param name Organization name.
+         */
+        public Organization(int id, String name) {
+            this.id = id;
+            this.name = name;
+        }
+
+        /**
+         * @return Organization id.
+         */
+        public int getId() {
+            return id;
+        }
+
+        /**
+         * @param id Organization id.
+         */
+        public void setId(int id) {
+            this.id = id;
+        }
+
+        /**
+         * @return Organization name.
+         */
+        public String getName() {
+            return name;
+        }
+
+        /**
+         * @param name Organization name.
+         */
+        public void setName(String name) {
+            this.name = name;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void writeBinary(BinaryWriter writer) throws BinaryObjectException
{
+            writer.writeInt("id", id);
+            writer.writeString("name", name);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void readBinary(BinaryReader reader) throws BinaryObjectException
{
+            id = reader.readInt("id");
+            name = reader.readString("name");
+        }
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/b4a5d376/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingMultiTypeTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingMultiTypeTest.java
b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingMultiTypeTest.java
new file mode 100644
index 0000000..602fc1c
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingMultiTypeTest.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.cache;
+
+import java.util.Map;
+import java.util.TreeMap;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.configuration.CacheConfiguration;
+
+/**
+ * Tests various cache operations with indexing enabled.
+ * Cache contain multiple types.
+ */
+public class CacheOffheapBatchIndexingMultiTypeTest extends CacheOffheapBatchIndexingBaseTest
{
+    /**
+     * Tests putAll with multiple indexed entities and streamer pre-loading with low off-heap
cache size.
+     */
+    public void testPutAllMultupleEntitiesAndStreamer() {
+        doStreamerBatchTest(50, 1_000, new Class<?>[] {
+            Integer.class, CacheOffheapBatchIndexingBaseTest.Person.class,
+            Integer.class, CacheOffheapBatchIndexingBaseTest.Organization.class},
+            1,
+            true);
+    }
+
+    /**
+     * Tests putAll with multiple indexed entities and streamer preloading with default off-heap
cache size.
+     */
+    public void testPutAllMultupleEntitiesAndStreamerDfltOffHeapRowCacheSize() {
+        doStreamerBatchTest(50, 1_000, new Class<?>[] {
+            Integer.class, CacheOffheapBatchIndexingBaseTest.Person.class,
+            Integer.class, CacheOffheapBatchIndexingBaseTest.Organization.class},
+            CacheConfiguration.DFLT_SQL_ONHEAP_ROW_CACHE_SIZE,
+            true);
+    }
+
+    /**
+     * Tests putAll after with streamer batch load with one entity.
+     */
+    public void testPuAllSingleEntity() {
+        doStreamerBatchTest(50,
+            1_000,
+            new Class<?>[] {Integer.class, CacheOffheapBatchIndexingBaseTest.Organization.class},
+            1,
+            false);
+    }
+
+    /**
+     * @param iterations Number of iterations.
+     * @param entitiesCnt Number of entities to put.
+     * @param entityClasses Entity classes.
+     * @param onHeapRowCacheSize Cache size.
+     * @param preloadInStreamer Data preload flag.
+     */
+    private void doStreamerBatchTest(int iterations,
+        int entitiesCnt,
+        Class<?>[] entityClasses,
+        int onHeapRowCacheSize,
+        boolean preloadInStreamer) {
+        Ignite ignite = grid(0);
+
+        final IgniteCache<Object, Object> cache =
+            ignite.createCache(cacheConfiguration(onHeapRowCacheSize, entityClasses));
+
+        try {
+            if (preloadInStreamer)
+                preload(cache.getName());
+
+            while (iterations-- >= 0) {
+                Map<Integer, Person> putMap1 = new TreeMap<>();
+
+                for (int i = 0; i < entitiesCnt; i++)
+                    putMap1.put(i, new Person(i, i + 1, String.valueOf(i), String.valueOf(i
+ 1), salary(i)));
+
+                cache.putAll(putMap1);
+
+                Map<Integer, Organization> putMap2 = new TreeMap<>();
+
+                for (int i = entitiesCnt / 2; i < entitiesCnt * 3 / 2; i++) {
+                    cache.remove(i);
+
+                    putMap2.put(i, new Organization(i, String.valueOf(i)));
+                }
+
+                cache.putAll(putMap2);
+            }
+        }
+        finally {
+            cache.destroy();
+        }
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/b4a5d376/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingSingleTypeTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingSingleTypeTest.java
b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingSingleTypeTest.java
new file mode 100644
index 0000000..c59e5fe
--- /dev/null
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOffheapBatchIndexingSingleTypeTest.java
@@ -0,0 +1,161 @@
+/*
+ * 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.cache;
+
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheEntryProcessor;
+import org.apache.ignite.cache.query.QueryCursor;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.configuration.CacheConfiguration;
+
+/**
+ * Tests various cache operations with indexing enabled.
+ * Cache contains single type.
+ */
+public class CacheOffheapBatchIndexingSingleTypeTest extends CacheOffheapBatchIndexingBaseTest
{
+    /**
+     * Tests removal using EntryProcessor.
+     *
+     * @throws Exception If failed.
+     */
+    public void testBatchRemove() throws Exception {
+        Ignite ignite = grid(0);
+
+        CacheConfiguration<Object, Object> ccfg = cacheConfiguration(1,
+            new Class<?>[] {Integer.class, CacheOffheapBatchIndexingBaseTest.Organization.class});
+
+        final IgniteCache<Object, Object> cache = ignite.createCache(ccfg);
+
+        try {
+            int iterations = 50;
+
+            while (iterations-- >= 0) {
+                int total = 1000;
+
+                for (int id = 0; id < total; id++)
+                    cache.put(id, new CacheOffheapBatchIndexingBaseTest.Organization(id,
"Organization " + id));
+
+                cache.invoke(0, new CacheEntryProcessor<Object, Object, Object>() {
+                    @Override public Object process(MutableEntry<Object, Object> entry,
Object... args) {
+                        entry.remove();
+
+                        return null;
+                    }
+                });
+
+                QueryCursor<List<?>> q = cache.query(new SqlFieldsQuery("select
_key,_val from Organization where id=0"));
+
+                assertEquals(0, q.getAll().size());
+
+                q = cache.query(new SqlFieldsQuery("select _key,_val from Organization where
id=1"));
+
+                assertEquals(1, q.getAll().size());
+
+                assertEquals(total - 1, cache.size());
+
+                cache.removeAll();
+            }
+        }
+        finally {
+            cache.destroy();
+        }
+    }
+
+    /**
+     *
+     */
+    public void testPutAllAndStreamer() {
+        doStreamerBatchTest(50,
+            1_000,
+            new Class<?>[] {Integer.class, CacheOffheapBatchIndexingBaseTest.Organization.class},
+            1,
+            true);
+    }
+
+    /**
+     *
+     */
+    public void testPutAllAndStreamerDfltOffHeapRowCacheSize() {
+        doStreamerBatchTest(50,
+            1_000,
+            new Class<?>[] {Integer.class, CacheOffheapBatchIndexingBaseTest.Organization.class},
+            CacheConfiguration.DFLT_SQL_ONHEAP_ROW_CACHE_SIZE,
+            true);
+    }
+
+    /**
+     *
+     */
+    public void testPuAllSingleEntity() {
+        doStreamerBatchTest(50,
+            1_000,
+            new Class<?>[] {Integer.class, CacheOffheapBatchIndexingBaseTest.Organization.class},
+            1,
+            false);
+    }
+
+    /**
+     * @param iterations Number of iterations.
+     * @param entitiesCnt Number of entities to put.
+     * @param entityClasses Entity classes.
+     * @param onHeapRowCacheSize Cache size.
+     * @param preloadInStreamer Data preload flag.
+     */
+    private void doStreamerBatchTest(int iterations,
+        int entitiesCnt,
+        Class<?>[] entityClasses,
+        int onHeapRowCacheSize,
+        boolean preloadInStreamer) {
+        Ignite ignite = grid(0);
+
+        final IgniteCache<Object, Object> cache =
+            ignite.createCache(cacheConfiguration(onHeapRowCacheSize, entityClasses));
+
+        try {
+            if (preloadInStreamer)
+                preload(cache.getName());
+
+            while (iterations-- >= 0) {
+                Map<Integer, Organization> putMap1 = new TreeMap<>();
+
+                for (int i = 0; i < entitiesCnt; i++)
+                    putMap1.put(i, new Organization(i, String.valueOf(i)));
+
+                cache.putAll(putMap1);
+
+                Map<Integer, Organization> putMap2 = new TreeMap<>();
+
+                for (int i = entitiesCnt / 2; i < entitiesCnt * 3 / 2; i++) {
+                    cache.remove(i);
+
+                    putMap2.put(i, new Organization(i, String.valueOf(i)));
+                }
+
+                cache.putAll(putMap2);
+            }
+        }
+        finally {
+            cache.destroy();
+        }
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/b4a5d376/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite2.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite2.java
b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite2.java
index e9fcb79..e5f4027 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite2.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite2.java
@@ -19,6 +19,7 @@ package org.apache.ignite.testsuites;
 
 import junit.framework.TestSuite;
 import org.apache.ignite.internal.processors.cache.CacheLocalQueryMetricsSelfTest;
+import org.apache.ignite.internal.processors.cache.CacheOffheapBatchIndexingSingleTypeTest;
 import org.apache.ignite.internal.processors.cache.CachePartitionedQueryMetricsDistributedSelfTest;
 import org.apache.ignite.internal.processors.cache.CachePartitionedQueryMetricsLocalSelfTest;
 import org.apache.ignite.internal.processors.cache.CacheQueryNewClientSelfTest;
@@ -103,6 +104,7 @@ public class IgniteCacheQuerySelfTestSuite2 extends TestSuite {
 
         // Other.
         suite.addTestSuite(CacheQueryNewClientSelfTest.class);
+        suite.addTestSuite(CacheOffheapBatchIndexingSingleTypeTest.class);
 
         return suite;
     }


Mime
View raw message