IGNITE-20688 Fixed broken handles after binary object detached (lightweighted).

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

@@ -50,41 +50,50 @@ public BinaryArrayIdentityResolver() {
 
     /** {@inheritDoc} */
     @Override protected int hashCode0(BinaryObject obj) {
-        int hash = 1;
-
         if (obj instanceof BinaryObjectExImpl) {
             BinaryObjectExImpl ex = (BinaryObjectExImpl)obj;
 
             int start = ex.dataStartOffset();
             int end = ex.footerStartOffset();
 
-            if (ex.hasArray()) {
-                // Handle heap object.
-                byte[] data = ex.array();
-
-                for (int i = start; i < end; i++)
-                    hash = 31 * hash + data[i];
-            }
+            if (ex.hasArray())
+                return hashCode(ex.array(), start, end);
             else {
                 // Handle offheap object.
+                int hash = 1;
+
                 long ptr = ex.offheapAddress();
 
                 for (int i = start; i < end; i++)
                     hash = 31 * hash + BinaryPrimitives.readByte(ptr, i);
+
+                return hash;
             }
         }
         else if (obj instanceof BinaryEnumObjectImpl) {
+            int hash = 1;
+
             int ord = obj.enumOrdinal();
 
             // Construct hash as if it was an int serialized in little-endian form.
             hash = 31 * hash + (ord & 0x000000FF);
             hash = 31 * hash + (ord & 0x0000FF00);
             hash = 31 * hash + (ord & 0x00FF0000);
             hash = 31 * hash + (ord & 0xFF000000);
+
+            return hash;
         }
         else
             throw new BinaryObjectException("Array identity resolver cannot be used with provided BinaryObject " +
                 "implementation: " + obj.getClass().getName());
+    }
+
+    /** */
+    public int hashCode(byte[] data, int startPos, int endPos) {
+        int hash = 1;
+
+        for (int i = startPos; i < endPos; i++)
+            hash = 31 * hash + data[i];
 
         return hash;
     }

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

@@ -35,6 +35,8 @@
 import org.apache.ignite.internal.GridDirectTransient;
 import org.apache.ignite.internal.IgniteCodeGeneratingFail;
 import org.apache.ignite.internal.binary.streams.BinaryHeapInputStream;
+import org.apache.ignite.internal.binary.streams.BinaryHeapOutputStream;
+import org.apache.ignite.internal.binary.streams.BinaryOutputStream;
 import org.apache.ignite.internal.processors.cache.CacheObject;
 import org.apache.ignite.internal.processors.cache.CacheObjectAdapter;
 import org.apache.ignite.internal.processors.cache.CacheObjectContext;
@@ -282,16 +284,33 @@ private byte[] valueBytesFromArray(CacheObjectValueContext ctx) {
      * @return Detached binary object.
      */
     public BinaryObjectImpl detach() {
+        return detach(true);
+    }
+
+    /** */
+    public BinaryObjectImpl detach(boolean resolveCrossObjectReferences) {
         if (!detachAllowed || detached())
             return this;
 
         int len = length();
 
-        byte[] arr0 = new byte[len];
+        if (resolveCrossObjectReferences) {
+            ObjectDetachHelper detachHelper = ObjectDetachHelper.create(arr, start);
+
+            if (detachHelper.isCrossObjectReferencesDetected()) {
+                try (BinaryOutputStream out = new BinaryHeapOutputStream(2 * len)) {
+                    detachHelper.detach(out);
+
+                    return new BinaryObjectImpl(ctx, out.arrayCopy(), 0);
+                }
+            }
+        }
+
+        byte[] detachedData = new byte[len];
 
-        U.arrayCopy(arr, start, arr0, 0, len);
+        U.arrayCopy(arr, start, detachedData, 0, len);
 
-        return new BinaryObjectImpl(ctx, arr0, 0);
+        return new BinaryObjectImpl(ctx, detachedData, 0);
     }
 
     /**

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

@@ -1338,7 +1338,7 @@ float readFloat(int fieldId) throws BinaryObjectException {
 
     /** {@inheritDoc} */
     @Nullable @Override public Object readObjectDetached(boolean deserialize) throws BinaryObjectException {
-        return BinaryUtils.unmarshal(in, ctx, ldr, this, true, deserialize);
+        return BinaryUtils.unmarshal(in, ctx, ldr, new BinaryReaderHandlesHolderImpl(), true, deserialize);
     }
 
     /** {@inheritDoc} */
@@ -2369,6 +2369,11 @@ private void streamPositionRandom(int pos) {
         return in.remaining();
     }
 
+    /** {@inheritDoc} */
+    @Override public boolean isEmpty() {
+        return hnds == null || hnds.isEmpty();
+    }
+
     /** {@inheritDoc} */
     @Override public void close() throws IOException {
         // No-op.

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

@@ -103,4 +103,9 @@ public void put(int pos, Object obj) {
                 data0.put(pos, obj);
         }
     }
+
+    /** */
+    public boolean isEmpty() {
+        return mode == MODE_EMPTY;
+    }
 }

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

@@ -43,4 +43,7 @@ public interface BinaryReaderHandlesHolder {
      * @return Handles.
      */
     public BinaryReaderHandles handles();
+
+    /** */
+    public boolean isEmpty();
 }

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

@@ -41,4 +41,9 @@ public class BinaryReaderHandlesHolderImpl implements BinaryReaderHandlesHolder
 
         return hnds;
     }
+
+    /** {@inheritDoc} */
+    @Override public boolean isEmpty() {
+        return hnds == null || hnds.isEmpty();
+    }
 }

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

@@ -836,6 +836,20 @@ public static int length(BinaryPositionReadable in, int start) {
         return in.readIntPositioned(start + GridBinaryMarshaller.TOTAL_LEN_POS);
     }
 
+    /** */
+    public static int dataStartRelative(BinaryPositionReadable in, int start) {
+        int typeId = in.readIntPositioned(start + GridBinaryMarshaller.TYPE_ID_POS);
+
+        if (typeId == GridBinaryMarshaller.UNREGISTERED_TYPE_ID) {
+            // Gets the length of the type name which is stored as string.
+            int len = in.readIntPositioned(start + GridBinaryMarshaller.DFLT_HDR_LEN + /** object type */1);
+
+            return GridBinaryMarshaller.DFLT_HDR_LEN + /** object type */1 + /** string length */ 4 + len;
+        }
+        else
+            return GridBinaryMarshaller.DFLT_HDR_LEN;
+    }
+
     /**
      * Get footer start of the object.
      *
@@ -1929,21 +1943,22 @@ public static Object doReadOptimized(BinaryInputStream in, BinaryContext ctx, @N
                 BinaryObjectExImpl po;
 
                 if (detach) {
-                    // In detach mode we simply copy object's content.
-                    in.position(start);
+                    BinaryObjectImpl binObj = new BinaryObjectImpl(ctx, in.array(), start);
+
+                    binObj.detachAllowed(true);
 
-                    po = new BinaryObjectImpl(ctx, in.readByteArray(len), 0);
+                    po = binObj.detach(!handles.isEmpty());
                 }
                 else {
                     if (in.offheapPointer() == 0)
                         po = new BinaryObjectImpl(ctx, in.array(), start);
                     else
                         po = new BinaryObjectOffheapImpl(ctx, in.offheapPointer(), start,
                             in.remaining() + in.position());
-
-                    in.position(start + po.length());
                 }
 
+                in.position(start + len);
+
                 handles.setHandle(po, start);
 
                 return po;