More compact serialization of numbers in generic native index

Previously all numbers where serialized in 8 bytes (long). Now the prop number of bytes are used for each type. This also applies for number arrays.
neo4j · Aug 16, 2018 · fdb5523 · fdb5523
1 parent fdfd068
commit fdb5523
Showing 1 changed file with 107 additions and 25 deletions.
diff --git a/community/kernel/src/main/java/org/neo4j/kernel/impl/index/schema/GenericKeyState.java b/community/kernel/src/main/java/org/neo4j/kernel/impl/index/schema/GenericKeyState.java
@@ -836,7 +836,7 @@ void put( PageCursor cursor )
             putBoolean( cursor, long0 );
             break;
         case NUMBER:
-            putNumber( cursor, long0, long1 );
+            putNumberIncludingType( cursor, long0, long1 );
             break;
         case ZONED_DATE_TIME_ARRAY:
             putArray( cursor, ( c, i ) -> putZonedDateTime( c, long0Array[i], long1Array[i], long2Array[i], long3Array[i] ) );
@@ -864,13 +864,32 @@ void put( PageCursor cursor )
             break;
         case NUMBER_ARRAY:
             cursor.putByte( (byte) long1 );
-            putArray( cursor, ( c, i ) -> c.putLong( long0Array[i] ) );
+            putArray( cursor, numberArrayElementWriter( long1 ) );
             break;
         default:
             throw new IllegalArgumentException( "Unknown type " + type );
         }
     }
 
+    private ArrayElementWriter numberArrayElementWriter( long long1 )
+    {
+        switch ( (int) long1 )
+        {
+        case RawBits.BYTE:
+            return ( c, i ) -> c.putByte( (byte) long0Array[i] );
+        case RawBits.SHORT:
+            return ( c, i ) -> c.putShort( (short) long0Array[i] );
+        case RawBits.INT:
+        case RawBits.FLOAT:
+            return ( c, i ) -> c.putInt( (int) long0Array[i] );
+        case RawBits.LONG:
+        case RawBits.DOUBLE:
+            return ( c, i ) -> c.putLong( long0Array[i] );
+        default:
+            throw new IllegalArgumentException( "Unknown number type " + long1 );
+        }
+    }
+
     interface ArrayElementWriter
     {
         void write( PageCursor cursor, int i );
@@ -885,10 +904,28 @@ private void putArray( PageCursor cursor, ArrayElementWriter writer )
         }
     }
 
-    private static void putNumber( PageCursor cursor, long long0, long long1 )
+    private static void putNumberIncludingType( PageCursor cursor, long long0, long long1 )
     {
         cursor.putByte( (byte) long1 );
-        cursor.putLong( long0 );
+        switch ( (int) long1 )
+        {
+        case RawBits.BYTE:
+            cursor.putByte( (byte) long0 );
+            break;
+        case RawBits.SHORT:
+            cursor.putShort( (short) long0 );
+            break;
+        case RawBits.INT:
+        case RawBits.FLOAT:
+            cursor.putInt( (int) long0 );
+            break;
+        case RawBits.LONG:
+        case RawBits.DOUBLE:
+            cursor.putLong( long0 );
+            break;
+        default:
+            throw new IllegalArgumentException( "Unknown number type " + long1 );
+        }
     }
 
     private static void putBoolean( PageCursor cursor, long long0 )
@@ -1002,7 +1039,13 @@ boolean read( PageCursor cursor, int size )
         case BOOLEAN_ARRAY:
             return readArray( cursor, ArrayType.BOOLEAN, this::readBoolean );
         case NUMBER_ARRAY:
-            return readNumberArray( cursor );
+            long1 = cursor.getByte(); // number type, like: byte, int, short a.s.o.
+            ArrayType numberType = numberArrayTypeOf( (byte) long1 );
+            if ( numberType == null )
+            {
+                return false;
+            }
+            return readArray( cursor, numberType, numberArrayElementReader( long1 ) );
         default:
             return false;
         }
@@ -1026,27 +1069,49 @@ private boolean readArray( PageCursor cursor, ArrayType type, ArrayElementReader
         return true;
     }
 
-    private boolean readNumberArray( PageCursor cursor )
+    private ArrayElementReader numberArrayElementReader( long long1 )
     {
-        long1 = cursor.getByte(); // number type, like: byte, int, short a.s.o.
-        if ( !setArrayLengthWhenReading( cursor ) )
-        {
-            return false;
-        }
-        initializeNumberArray( arrayLength );
-        ArrayType numberType = numberArrayTypeOf( (byte) long1 );
-        if ( numberType == null )
-        {
-            return false;
-        }
-
-        beginArray( arrayLength, numberType );
-        for ( int i = 0; i < arrayLength; i++ )
+        switch ( (int) long1 )
         {
-            long0Array[i] = cursor.getLong();
+        case RawBits.BYTE:
+            return c ->
+            {
+                writeInteger( c.getByte() );
+                return true;
+            };
+        case RawBits.SHORT:
+            return c ->
+            {
+                writeInteger( c.getShort() );
+                return true;
+            };
+        case RawBits.INT:
+            return c->
+            {
+                writeInteger( c.getInt() );
+                return true;
+            };
+        case RawBits.LONG:
+            return c->
+            {
+                writeInteger( c.getLong() );
+                return true;
+            };
+        case RawBits.FLOAT:
+            return c->
+            {
+                writeFloatingPoint( Float.intBitsToFloat( c.getInt() ) );
+                return true;
+            };
+        case RawBits.DOUBLE:
+            return c->
+            {
+                writeFloatingPoint( Double.longBitsToDouble( c.getLong() ) );
+                return true;
+            };
+        default:
+            throw new IllegalArgumentException( "Unknown number type " + long1 );
         }
-        endArray();
-        return true;
     }
 
     private static ArrayType numberArrayTypeOf( byte numberType )
@@ -1103,8 +1168,25 @@ private boolean setArrayLengthWhenReading( PageCursor cursor )
     private boolean readNumber( PageCursor cursor )
     {
         long1 = cursor.getByte();
-        long0 = cursor.getLong();
-        return true;
+        switch ( (int) long1 )
+        {
+        case RawBits.BYTE:
+            long0 = cursor.getByte();
+            return true;
+        case RawBits.SHORT:
+            long0 = cursor.getShort();
+            return true;
+        case RawBits.INT:
+        case RawBits.FLOAT:
+            long0 = cursor.getInt();
+            return true;
+        case RawBits.LONG:
+        case RawBits.DOUBLE:
+            long0 = cursor.getLong();
+            return true;
+        default:
+            return false;
+        }
     }
 
     private boolean readBoolean( PageCursor cursor )