Support named rules in SchemaStore

- This adds an "optional" name field to the end of schema rules in the schema
  store.
- SchemaRule objects get a generated name if none is otherwise provided.
- The generated name is frozen once observed and cannot be changed afterwards.
- The schema rule names are thus effectively immutable.
- Legacy schema rule records cannot fit this new name anywhere, and thus always
  gets a generated name.
- New schema rule records that don't have a name gets a generated one as well.
- The empty string is not a valid name, but if we somehow find the empty string
  in the store, we will use the generated name instead – this allows us to add
  other fields in the future, without imposing a hard requirement that all
  future schema rule records must have a name.
- The name is stored as a null-terminated string, which is different from other
  strings in the schema store.
- The reason for the null-terminated strings is that its slightly more straight
  forward to handle if the name field is missing.
- Compatibility tests is added for all these new cases.
- Also, obviously, if you set an explicit schema rule name, we require it to
  not be the empty string, and to not contain the null-byte character.
- The name field string is encoded in UTF-8, so the terminating null-byte will
  be the only null-byte in the string.

community/common/src/main/java/org/neo4j/string/UTF8.java

@@ -56,18 +56,58 @@ public static String getDecodedStringFrom( ByteBuffer source )
         return UTF8.decode( data );
     }
 
+    public static String getDecodedNullTerminatedStringFrom( ByteBuffer source )
+    {
+        int start = source.position();
+        int nullPos = -1;
+        int cursor = start;
+        while ( cursor < source.limit() )
+        {
+            if ( source.get( cursor ) == 0 )
+            {
+                nullPos = cursor;
+                break;
+            }
+            cursor++;
+        }
+        if ( nullPos == -1 )
+        {
+            return null;
+        }
+        int length = nullPos - start;
+        if ( length == 0 )
+        {
+            return "";
+        }
+        byte[] data = new byte[length];
+        source.get( data );
+        return UTF8.decode( data );
+    }
+
     public static void putEncodedStringInto( String text, ByteBuffer target )
     {
         byte[] data = encode( text );
         target.putInt( data.length );
         target.put( data );
     }
 
+    public static void putEncodedNullTerminatedStringInto( String text, ByteBuffer target )
+    {
+        byte[] data = encode( text );
+        target.put( data );
+        target.put( (byte) 0 );
+    }
+
     public static int computeRequiredByteBufferSize( String text )
     {
         return encode( text ).length + 4;
     }
 
+    public static int computeRequiredNullTerminatedByteBufferSize( String text )
+    {
+        return encode( text ).length + 1;
+    }
+
     private UTF8()
     {
         // No need to instantiate, all methods are static

community/kernel/src/main/java/org/neo4j/kernel/impl/store/record/ConstraintRule.java

@@ -28,18 +28,11 @@
 
 import static org.neo4j.kernel.api.schema_new.SchemaUtil.idTokenNameLookup;
 
-public class ConstraintRule implements SchemaRule, ConstraintDescriptor.Supplier
+public class ConstraintRule extends SchemaRule implements ConstraintDescriptor.Supplier
 {
-    private final long id;
     private final Long ownedIndexRule;
     private final ConstraintDescriptor descriptor;
 
-    @Override
-    public final long getId()
-    {
-        return this.id;
-    }
-
     public static ConstraintRule constraintRule(
             long id, ConstraintDescriptor descriptor )
     {
@@ -54,7 +47,7 @@ public static ConstraintRule constraintRule(
 
     ConstraintRule( long id, ConstraintDescriptor descriptor, Long ownedIndexRule )
     {
-        this.id = id;
+        super( id );
         this.descriptor = descriptor;
         this.ownedIndexRule = ownedIndexRule;
     }

community/kernel/src/main/java/org/neo4j/kernel/impl/store/record/IndexRule.java

@@ -32,9 +32,8 @@
 /**
  * A {@link Label} can have zero or more index rules which will have data specified in the rules indexed.
  */
-public class IndexRule implements SchemaRule, NewIndexDescriptor.Supplier
+public class IndexRule extends SchemaRule implements NewIndexDescriptor.Supplier
 {
-    private final long id;
     private final SchemaIndexProvider.Descriptor providerDescriptor;
     private final NewIndexDescriptor descriptor;
     private final Long owningConstraint;
@@ -55,7 +54,7 @@ public static IndexRule constraintIndexRule( long id, NewIndexDescriptor descrip
     IndexRule( long id, SchemaIndexProvider.Descriptor providerDescriptor,
             NewIndexDescriptor descriptor, Long owningConstraint )
     {
-        this.id = id;
+        super( id );
         if ( providerDescriptor == null )
         {
             throw new IllegalArgumentException( "null provider descriptor prohibited" );
@@ -107,12 +106,6 @@ public IndexRule withOwningConstraint( long constraintId )
         return constraintIndexRule( id, descriptor, providerDescriptor, constraintId );
     }
 
-    @Override
-    public long getId()
-    {
-        return id;
-    }
-
     @Override
     public int length()
     {

community/kernel/src/main/java/org/neo4j/kernel/impl/store/record/SchemaRuleSerialization.java

@@ -31,6 +31,7 @@
 import org.neo4j.kernel.api.schema_new.SchemaProcessor;
 import org.neo4j.kernel.api.schema_new.constaints.ConstraintDescriptor;
 import org.neo4j.kernel.api.schema_new.constaints.ConstraintDescriptorFactory;
+import org.neo4j.kernel.api.schema_new.constaints.UniquenessConstraintDescriptor;
 import org.neo4j.kernel.api.schema_new.index.NewIndexDescriptor;
 import org.neo4j.kernel.api.schema_new.index.NewIndexDescriptorFactory;
 import org.neo4j.storageengine.api.schema.SchemaRule;
@@ -126,6 +127,7 @@ public static void serialize( IndexRule indexRule, ByteBuffer target )
         }
 
         indexDescriptor.schema().processWith( new SchemaDescriptorSerializer( target ) );
+        UTF8.putEncodedNullTerminatedStringInto( indexRule.getName(), target );
     }
 
     /**
@@ -157,6 +159,7 @@ public static void serialize( ConstraintRule constraintRule, ByteBuffer target )
         }
 
         constraintDescriptor.schema().processWith( new SchemaDescriptorSerializer( target ) );
+        UTF8.putEncodedNullTerminatedStringInto( constraintRule.getName(), target );
     }
 
     /**
@@ -181,6 +184,7 @@ public static int lengthOf( IndexRule indexRule )
         }
 
         length += indexDescriptor.schema().computeWith( schemaSizeComputer );
+        length += UTF8.computeRequiredNullTerminatedByteBufferSize( indexRule.getName() );
         return length;
     }
 
@@ -202,6 +206,7 @@ public static int lengthOf( ConstraintRule constraintRule )
         }
 
         length += constraintDescriptor.schema().computeWith( schemaSizeComputer );
+        length += UTF8.computeRequiredNullTerminatedByteBufferSize( constraintRule.getName() );
         return length;
     }
 
@@ -214,26 +219,23 @@ private static IndexRule readIndexRule( long id, ByteBuffer source ) throws Malf
         SchemaIndexProvider.Descriptor indexProvider = readIndexProviderDescriptor( source );
         LabelSchemaDescriptor schema;
         byte indexRuleType = source.get();
+        IndexRule rule;
         switch ( indexRuleType )
         {
         case GENERAL_INDEX:
             schema = readLabelSchema( source );
-            return IndexRule.indexRule(
-                    id,
-                    NewIndexDescriptorFactory.forSchema( schema ),
-                    indexProvider
-                );
+            rule = IndexRule.indexRule( id, NewIndexDescriptorFactory.forSchema( schema ), indexProvider );
+            readRuleName( rule, source );
+            return rule;
 
         case UNIQUE_INDEX:
             long owningConstraint = source.getLong();
             schema = readLabelSchema( source );
-
-            return IndexRule.constraintIndexRule(
-                    id,
-                    NewIndexDescriptorFactory.uniqueForSchema( schema ),
-                    indexProvider,
-                    owningConstraint == NO_OWNING_CONSTRAINT_YET ? null : owningConstraint
-                );
+            NewIndexDescriptor descriptor = NewIndexDescriptorFactory.uniqueForSchema( schema );
+            rule = IndexRule.constraintIndexRule( id, descriptor, indexProvider,
+                    owningConstraint == NO_OWNING_CONSTRAINT_YET ? null : owningConstraint );
+            readRuleName( rule, source );
+            return rule;
 
         default:
             throw new MalformedSchemaRuleException( format( "Got unknown index rule type '%d'.", indexRuleType ) );
@@ -264,29 +266,38 @@ private static ConstraintRule readConstraintRule( long id, ByteBuffer source ) t
     {
         SchemaDescriptor schema;
         byte constraintRuleType = source.get();
+        ConstraintRule rule;
         switch ( constraintRuleType )
         {
         case EXISTS_CONSTRAINT:
             schema = readSchema( source );
-            return ConstraintRule.constraintRule(
-                    id,
-                    ConstraintDescriptorFactory.existsForSchema( schema )
-            );
+            rule = ConstraintRule.constraintRule( id, ConstraintDescriptorFactory.existsForSchema( schema ) );
+            readRuleName( rule, source );
+            return rule;
 
         case UNIQUE_CONSTRAINT:
             long ownedIndex = source.getLong();
             schema = readSchema( source );
-            return ConstraintRule.constraintRule(
-                    id,
-                    ConstraintDescriptorFactory.uniqueForSchema( schema ),
-                    ownedIndex
-            );
+            UniquenessConstraintDescriptor descriptor = ConstraintDescriptorFactory.uniqueForSchema( schema );
+            rule = ConstraintRule.constraintRule( id, descriptor, ownedIndex );
+            readRuleName( rule, source );
+            return rule;
 
         default:
             throw new MalformedSchemaRuleException( format( "Got unknown constraint rule type '%d'.", constraintRuleType ) );
         }
     }
 
+    private static void readRuleName( SchemaRule rule, ByteBuffer source )
+    {
+        String ruleName = UTF8.getDecodedNullTerminatedStringFrom( source );
+        if ( ruleName == null || ruleName.isEmpty() )
+        {
+            ruleName = SchemaRule.generateName( rule );
+        }
+        rule.setName( ruleName );
+    }
+
     // READ HELP
 
     private static SchemaDescriptor readSchema( ByteBuffer source ) throws MalformedSchemaRuleException

community/kernel/src/main/java/org/neo4j/storageengine/api/schema/SchemaRule.java

@@ -26,23 +26,97 @@
 import org.neo4j.kernel.api.schema_new.SchemaDescriptor;
 import org.neo4j.kernel.api.schema_new.constaints.ConstraintDescriptor;
 import org.neo4j.kernel.api.schema_new.index.NewIndexDescriptor;
+import org.neo4j.kernel.impl.store.record.ConstraintRule;
+import org.neo4j.kernel.impl.store.record.IndexRule;
 import org.neo4j.kernel.impl.store.record.RecordSerializable;
 
 /**
  * Represents a stored schema rule.
  */
-public interface SchemaRule extends SchemaDescriptor.Supplier, RecordSerializable
+public abstract class SchemaRule implements SchemaDescriptor.Supplier, RecordSerializable
 {
+    protected final long id;
+    protected String name;
+
+    protected SchemaRule( long id )
+    {
+        this.id = id;
+    }
+
     /**
      * The persistence id for this rule.
      */
-    long getId();
+    public final long getId()
+    {
+        return this.id;
+    }
+
+    /**
+     * @return The (possibly user supplied) name of this schema rule.
+     */
+    public final String getName()
+    {
+        if ( name == null )
+        {
+            name = SchemaRule.generateName( this );
+        }
+        return name;
+    }
+
+    /**
+     * Set the name of this rule, if it has not been set already.
+     *
+     * Note that once a name has been observed - either via this method or via {@link #getName()} – for this rule,
+     * it cannot be changed.
+     * @param name The name desired for this rule.
+     * @return {@code true} if the name was set, {@code false} if this rule already has a name.
+     */
+    public final boolean setName( String name )
+    {
+        if ( this.name == null )
+        {
+            if ( name == null )
+            {
+                return true;
+            }
+            if ( name.length() == 0 )
+            {
+                throw new IllegalArgumentException( "Rule name cannot be the empty string" );
+            }
+            for ( int i = 0; i < name.length(); i++ )
+            {
+                char ch = name.charAt( i );
+                if ( ch == 0 )
+                {
+                    throw new IllegalArgumentException( "Illegal rule name: '" + name + "'" );
+                }
+            }
+            this.name = name;
+            return true;
+        }
+        return false;
+    }
+
+    public static String generateName( SchemaRule rule )
+    {
+        Class<? extends SchemaRule> type = rule.getClass();
+        long id = rule.getId();
+        if ( type == IndexRule.class )
+        {
+            return "index_" + id;
+        }
+        if ( type == ConstraintRule.class )
+        {
+            return "constraint_" + id;
+        }
+        return "schema_" + id;
+    }
 
     /**
      * This enum is used for the legacy schema store, and should not be extended.
      */
     @Deprecated
-    enum Kind
+    public enum Kind
     {
         INDEX_RULE( "Index" ),
         CONSTRAINT_INDEX_RULE( "Constraint index" ),