Page memory integration WIP

modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryEnumObjectImpl.java

@@ -112,7 +112,7 @@ public BinaryEnumObjectImpl(BinaryContext ctx, int typeId, @Nullable String clsN
     @Override public <T> T deserialize() throws BinaryObjectException {
         Class cls = BinaryUtils.resolveClass(ctx, typeId, clsName, ctx.configuration().getClassLoader(), true);
 
-        return BinaryEnumCache.get(cls, ord);
+        return (T)BinaryEnumCache.get(cls, ord);
     }
 
     /** {@inheritDoc} */
@@ -210,6 +210,11 @@ public BinaryEnumObjectImpl(BinaryContext ctx, int typeId, @Nullable String clsN
         throw new UnsupportedOperationException("TODO implement.");
     }
 
+    /** {@inheritDoc} */
+    @Override public int valueBytesLength(CacheObjectContext ctx) throws IgniteCheckedException {
+        throw new UnsupportedOperationException("TODO implement.");
+    }
+
     /** {@inheritDoc} */
     @Override public byte cacheObjectType() {
         return TYPE_BINARY;

modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java

@@ -133,6 +133,7 @@ public BinaryObjectImpl(BinaryContext ctx, byte[] arr, int start) {
         return arr0;
     }
 
+    /** {@inheritDoc} */
     @Override public boolean putValue(ByteBuffer buf, CacheObjectContext ctx) throws IgniteCheckedException {
         int len = length();
 
@@ -146,6 +147,11 @@ public BinaryObjectImpl(BinaryContext ctx, byte[] arr, int start) {
         return true;
     }
 
+    /** {@inheritDoc} */
+    @Override public int valueBytesLength(CacheObjectContext ctx) throws IgniteCheckedException {
+        return length() + 5;
+    }
+
     /** {@inheritDoc} */
     @Override public CacheObject prepareForCache(CacheObjectContext ctx) {
         if (detached())

modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectOffheapImpl.java

@@ -128,6 +128,11 @@ public BinaryObject heapCopy() {
         throw new UnsupportedOperationException("TODO implement");
     }
 
+    /** {@inheritDoc} */
+    @Override public int valueBytesLength(CacheObjectContext ctx) throws IgniteCheckedException {
+        throw new UnsupportedOperationException("TODO implement");
+    }
+
     /** {@inheritDoc} */
     @Override public long offheapAddress() {
         return ptr;

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObject.java

@@ -49,6 +49,20 @@ public interface CacheObject extends Message {
      */
     public byte[] valueBytes(CacheObjectContext ctx) throws IgniteCheckedException;
 
+    /**
+     * @param ctx Cache object context.
+     * @return Size required to store this value object.
+     * @throws IgniteCheckedException If failed.
+     */
+    public int valueBytesLength(CacheObjectContext ctx) throws IgniteCheckedException;
+
+    /**
+     * @param buf Buffer to write value to.
+     * @param ctx Cache object context.
+     * @return {@code True} if value was successfully written, {@code false} if there was not enough space in the
+     *      buffer.
+     * @throws IgniteCheckedException If failed.
+     */
     public boolean putValue(ByteBuffer buf, CacheObjectContext ctx) throws IgniteCheckedException;
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectAdapter.java

@@ -84,6 +84,14 @@ protected boolean needCopy(CacheObjectContext ctx) {
         return true;
     }
 
+    /** {@inheritDoc} */
+    @Override public int valueBytesLength(CacheObjectContext ctx) throws IgniteCheckedException {
+        if (valBytes == null)
+            valueBytes(ctx);
+
+        return valBytes.length + 5;
+    }
+
     /** {@inheritDoc} */
     @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
         reader.setBuffer(buf);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheObjectByteArrayImpl.java

@@ -86,6 +86,10 @@ public CacheObjectByteArrayImpl(byte[] val) {
         return true;
     }
 
+    @Override public int valueBytesLength(CacheObjectContext ctx) throws IgniteCheckedException {
+        return val.length + 5;
+    }
+
     /** {@inheritDoc} */
     @Override public byte cacheObjectType() {
         return TYPE_BYTE_ARR;

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java

@@ -1861,7 +1861,7 @@ public final <K1, V1> IgniteInternalFuture<Map<K1, V1>> getAllAsync0(@Nullable f
 
                 final boolean storeEnabled = !skipVals && readThrough && ctx.readThrough();
 
-                final boolean needEntry = storeEnabled || ctx.isSwapOrOffheapEnabled();
+                final boolean needEntry = storeEnabled || ctx.isSwapOrOffheapEnabled() || ctx.isDatabaseEnabled();
 
                 Map<KeyCacheObject, GridCacheVersion> misses = null;
 
@@ -1878,7 +1878,7 @@ public final <K1, V1> IgniteInternalFuture<Map<K1, V1>> getAllAsync0(@Nullable f
 
                         try {
                             T2<CacheObject, GridCacheVersion> res = entry.innerGetVersioned(null,
-                                ctx.isSwapOrOffheapEnabled(),
+                                ctx.isSwapOrOffheapEnabled() || ctx.isDatabaseEnabled(),
                                 /*unmarshal*/true,
                                 /*update-metrics*/!skipVals,
                                 /*event*/!skipVals,

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheContext.java

@@ -1428,6 +1428,13 @@ public boolean isOffHeapEnabled() {
         return swapMgr.offHeapEnabled();
     }
 
+    /**
+     * @return If database is enabled.
+     */
+    public boolean isDatabaseEnabled() {
+        return storeMgr.isDatabaseEnabled();
+    }
+
     /**
      * @return {@code True} if store read-through mode is enabled.
      */

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java

@@ -791,7 +791,7 @@ public void touch(GridCacheEntryEx e, AffinityTopologyVersion topVer) {
             U.error(log, "Failed to evict entry from cache: " + e, ex);
         }
 
-        if (!cctx.isNear() && memoryMode == OFFHEAP_TIERED) {
+        if (!cctx.isNear() && (memoryMode == OFFHEAP_TIERED || cctx.isDatabaseEnabled())) {
             try {
                 evict0(cctx.cache(), e, cctx.versions().next(), null, false);
             }

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java

@@ -475,62 +475,78 @@ public boolean isStartVersion() {
         throws IgniteCheckedException, GridCacheEntryRemovedException {
         boolean swapEnabled = cctx.swap().swapEnabled();
 
-        if (!swapEnabled && !cctx.isOffHeapEnabled())
+        if (!swapEnabled && !cctx.isOffHeapEnabled() && !cctx.isDatabaseEnabled())
             return null;
 
         synchronized (this) {
             checkObsolete();
 
             if (isStartVersion() && ((flags & IS_UNSWAPPED_MASK) == 0)) {
-                GridCacheSwapEntry e;
+                if (cctx.isDatabaseEnabled()) {
+                    IgniteBiTuple<CacheObject, GridCacheVersion> read = cctx.queries().read(key);
 
-                if (cctx.offheapTiered()) {
-                    e = cctx.swap().readOffheapPointer(this);
+                    flags |= IS_UNSWAPPED_MASK;
 
-                    if (e != null) {
-                        if (e.offheapPointer() > 0) {
-                            offHeapPointer(e.offheapPointer());
+                    if (read != null) {
+                        CacheObject idxVal = read.get1();
 
-                            flags |= IS_OFFHEAP_PTR_MASK;
+                        // Set unswapped value.
+                        update(idxVal, 0, 0, read.get2());
 
-                            if (needVal) {
-                                CacheObject val = cctx.fromOffheap(offHeapPointer(), false);
+                        return idxVal;
+                    }
+                }
+                else {
+                    GridCacheSwapEntry e;
 
-                                e.value(val);
+                    if (cctx.offheapTiered()) {
+                        e = cctx.swap().readOffheapPointer(this);
+
+                        if (e != null) {
+                            if (e.offheapPointer() > 0) {
+                                offHeapPointer(e.offheapPointer());
+
+                                flags |= IS_OFFHEAP_PTR_MASK;
+
+                                if (needVal) {
+                                    CacheObject val = cctx.fromOffheap(offHeapPointer(), false);
+
+                                    e.value(val);
+                                }
                             }
+                            else // Read from swap.
+                                offHeapPointer(0);
                         }
-                        else // Read from swap.
-                            offHeapPointer(0);
                     }
-                }
-                else
-                    e = detached() ? cctx.swap().read(this, true, true, true, false) : cctx.swap().readAndRemove(this);
+                    else
+                        e = detached() ? cctx.swap().read(this, true, true, true, false) : cctx.swap().readAndRemove(this);
 
-                if (log.isDebugEnabled())
-                    log.debug("Read swap entry [swapEntry=" + e + ", cacheEntry=" + this + ']');
+                    if (log.isDebugEnabled())
+                        log.debug("Read swap entry [swapEntry=" + e + ", cacheEntry=" + this + ']');
 
-                flags |= IS_UNSWAPPED_MASK;
+                    flags |= IS_UNSWAPPED_MASK;
 
-                // If there is a value.
-                if (e != null) {
-                    long delta = e.expireTime() == 0 ? 0 : e.expireTime() - U.currentTimeMillis();
+                    // If there is a value.
+                    if (e != null) {
+                        long delta = e.expireTime() == 0 ? 0 : e.expireTime() - U.currentTimeMillis();
 
-                    if (delta >= 0) {
-                        CacheObject val = e.value();
+                        if (delta >= 0) {
+                            CacheObject val = e.value();
 
-                        val = cctx.kernalContext().cacheObjects().prepareForCache(val, cctx);
+                            val = cctx.kernalContext().cacheObjects().prepareForCache(val, cctx);
 
-                        // Set unswapped value.
-                        update(val, e.expireTime(), e.ttl(), e.version());
+                            // Set unswapped value.
+                            update(val, e.expireTime(), e.ttl(), e.version());
 
-                        // Must update valPtr again since update() will reset it.
-                        if (cctx.offheapTiered() && e.offheapPointer() > 0)
-                            offHeapPointer(e.offheapPointer());
+                            // Must update valPtr again since update() will reset it.
+                            if (cctx.offheapTiered() && e.offheapPointer() > 0)
+                                offHeapPointer(e.offheapPointer());
 
-                        return val;
+                            return val;
+                        }
+                        else
+                            clearIndex(e.value(), e.version());
                     }
-                    else
-                        clearIndex(e.value(), e.version());
                 }
             }
         }

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/binary/CacheObjectBinaryProcessorImpl.java

@@ -815,6 +815,14 @@ public BinaryContext binaryContext() {
         return super.toCacheObject(ctx, type, bytes);
     }
 
+    /** {@inheritDoc} */
+    @Override public KeyCacheObject toKeyCacheObject(CacheObjectContext ctx, byte type, byte[] bytes) throws IgniteCheckedException {
+        if (type == BinaryObjectImpl.TYPE_BINARY)
+            return new BinaryObjectImpl(binaryContext(), bytes, 0);
+
+        return super.toKeyCacheObject(ctx, type, bytes);
+    }
+
     /** {@inheritDoc} */
     @Override public CacheObject toCacheObject(GridCacheContext ctx, long valPtr, boolean tmp)
         throws IgniteCheckedException {

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java

@@ -342,7 +342,7 @@ void processQueryRequest(UUID sndId, GridCacheQueryRequest req) {
      * @param key Key.
      * @throws IgniteCheckedException If failed.
      */
-    public void onSwap(CacheObject key) throws IgniteCheckedException {
+    public void onSwap(KeyCacheObject key) throws IgniteCheckedException {
         if (!enterBusy())
             return; // Ignore index update when node is stopping.
 
@@ -361,7 +361,7 @@ public void onSwap(CacheObject key) throws IgniteCheckedException {
      * @param val Value
      * @throws IgniteCheckedException If failed.
      */
-    public void onUnswap(CacheObject key, CacheObject val) throws IgniteCheckedException {
+    public void onUnswap(KeyCacheObject key, CacheObject val) throws IgniteCheckedException {
         if (!enterBusy())
             return; // Ignore index update when node is stopping.
 
@@ -390,7 +390,7 @@ private void invalidateResultCache() {
      * @param expirationTime Expiration time or 0 if never expires.
      * @throws IgniteCheckedException In case of error.
      */
-    public void store(CacheObject key, CacheObject val, GridCacheVersion ver, long expirationTime)
+    public void store(KeyCacheObject key, CacheObject val, GridCacheVersion ver, long expirationTime)
         throws IgniteCheckedException {
         assert key != null;
         assert val != null;
@@ -412,13 +412,29 @@ public void store(CacheObject key, CacheObject val, GridCacheVersion ver, long e
         }
     }
 
+    /**
+     * @param key Key to read.
+     * @return Value tuple, if available.
+     */
+    public IgniteBiTuple<CacheObject, GridCacheVersion> read(KeyCacheObject key) throws IgniteCheckedException {
+        if (!enterBusy())
+            return null; // Ignore index update when node is stopping.
+
+        try {
+            return qryProc.read(space, key);
+        }
+        finally {
+            leaveBusy();
+        }
+    }
+
     /**
      * @param key Key.
      * @param val Value.
      * @throws IgniteCheckedException Thrown in case of any errors.
      */
     @SuppressWarnings("SimplifiableIfStatement")
-    public void remove(CacheObject key, CacheObject val, GridCacheVersion ver) throws IgniteCheckedException {
+    public void remove(KeyCacheObject key, CacheObject val, GridCacheVersion ver) throws IgniteCheckedException {
         assert key != null;
 
         if (!GridQueryProcessor.isEnabled(cctx.config()) && !(key instanceof GridCacheInternal))

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/CacheStoreManager.java

@@ -50,6 +50,11 @@ public void initialize(@Nullable CacheStore<?, ?> cfgStore, Map<CacheStore, Thre
      */
     public boolean configured();
 
+    /**
+     * @return {@code True} if database is enabled.
+     */
+    public boolean isDatabaseEnabled();
+
     /**
      * @return Wrapped store.
      */

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java

@@ -254,6 +254,11 @@ private CacheStore cacheStoreWrapper(GridKernalContext ctx,
         return store != null;
     }
 
+    /** {@inheritDoc} */
+    @Override public boolean isDatabaseEnabled() {
+        return false;
+    }
+
     /** {@inheritDoc} */
     @Override public CacheStore<?, ?> configuredStore() {
         return cfgStore;

modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessor.java

@@ -152,13 +152,28 @@ public interface IgniteCacheObjectProcessor extends GridProcessor {
      */
     public CacheObject toCacheObject(CacheObjectContext ctx, byte type, byte[] bytes);
 
+    /**
+     * @param ctx Cache context.
+     * @param type Object type.
+     * @param bytes Object bytes.
+     * @return Cache object.
+     */
+    public KeyCacheObject toKeyCacheObject(CacheObjectContext ctx, byte type, byte[] bytes) throws IgniteCheckedException;
+
     /**
      * @param ctx Cache context.
      * @param buf Buffer to read from.
      * @return Cache object.
      */
     public CacheObject toCacheObject(CacheObjectContext ctx, ByteBuffer buf);
 
+    /**
+     * @param ctx Cache context.
+     * @param buf Buffer to read from.
+     * @return Cache object.
+     */
+    public KeyCacheObject toKeyCacheObject(CacheObjectContext ctx, ByteBuffer buf) throws IgniteCheckedException;
+
     /**
      * @param ctx Context.
      * @param valPtr Value pointer.