OPENDJ-2016 Implement new on disk merge import strategy based on storage en...

Jean-Noel Rouvignac

26.01.2015 c5d246665c8d72aa524009a12af556f8fba76fe4

 opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/DefaultIndex.java

@@ -60,12 +60,9 @@
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /** The limit on the number of entry IDs that may be indexed by one key. */
  private int indexEntryLimit;

  private final State state;

  private final EntryContainer entryContainer;

  private int indexEntryLimit;
  private EntryIDSetCodec codec;

  /**
@@ -124,11 +121,32 @@
          @Override
          public EntryIDSet transform(ByteString key, ByteString value) throws NeverThrowsException
          {
            return codec.decode(key, value);
            return decodeValue(key, value);
          }
        });
  }

  EntryIDSet decodeValue(ByteSequence key, ByteString value)
  {
    return codec.decode(key, value);
  }

  ByteString toValue(EntryID entryID)
  {
    return codec.encode(newDefinedSet(entryID.longValue()));
  }

  ByteString toValue(EntryIDSet entryIDSet)
  {
    return codec.encode(entryIDSet);
  }

  ByteString toValue(ImportIDSet importIDSet)
  {
    return importIDSet.valueToByteString(codec);
  }

  // TODO JNR rename to importUpsert() ?
  @Override
  public final void importPut(Importer importer, ImportIDSet idsToBeAdded) throws StorageRuntimeException
  {
@@ -138,14 +156,14 @@
    ByteString value = importer.read(getName(), key);
    if (value != null)
    {
      final EntryIDSet entryIDSet = codec.decode(key, value);
      final EntryIDSet entryIDSet = decodeValue(key, value);
      final ImportIDSet importIDSet = new ImportIDSet(key, entryIDSet, indexEntryLimit);
      importIDSet.merge(idsToBeAdded);
      importer.put(getName(), key, importIDSet.valueToByteString(codec));
      importer.put(getName(), key, toValue(importIDSet));
    }
    else
    {
      importer.put(getName(), key, idsToBeAdded.valueToByteString(codec));
      importer.put(getName(), key, toValue(idsToBeAdded));
    }
  }

@@ -162,7 +180,7 @@
      throw new IllegalStateException("Expected to have a value associated to key " + key + " for index " + getName());
    }

    final EntryIDSet entryIDSet = codec.decode(key, value);
    final EntryIDSet entryIDSet = decodeValue(key, value);
    final ImportIDSet importIDSet = new ImportIDSet(key, entryIDSet, indexEntryLimit);
    importIDSet.remove(idsToBeRemoved);
    if (importIDSet.isDefined() && importIDSet.size() == 0)
@@ -171,7 +189,7 @@
    }
    else
    {
      importer.put(getName(), key, importIDSet.valueToByteString(codec));
      importer.put(getName(), key, toValue(importIDSet));
    }
  }

@@ -218,7 +236,7 @@
        if (oldValue != null)
        {
          EntryIDSet entryIDSet = computeEntryIDSet(key, oldValue.toByteString(), deletedIDs, addedIDs);
          ByteString after = codec.encode(entryIDSet);
          ByteString after = toValue(entryIDSet);
          /*
           * If there are no more IDs then return null indicating that the record should be removed.
           * If index is not trusted then this will cause all subsequent reads for this key to
@@ -234,7 +252,7 @@
          }
          if (isNotEmpty(addedIDs))
          {
            return codec.encode(addedIDs);
            return toValue(addedIDs);
          }
        }
        return null; // no change.
@@ -254,7 +272,7 @@

  private EntryIDSet computeEntryIDSet(ByteString key, ByteString value, EntryIDSet deletedIDs, EntryIDSet addedIDs)
  {
    EntryIDSet entryIDSet = codec.decode(key, value);
    EntryIDSet entryIDSet = decodeValue(key, value);
    if (addedIDs != null)
    {
      if (entryIDSet.isDefined() && indexEntryLimit > 0)
@@ -300,7 +318,7 @@
      ByteString value = txn.read(getName(), key);
      if (value != null)
      {
        return codec.decode(key, value);
        return decodeValue(key, value);
      }
      return trusted ? newDefinedSet() : newUndefinedSet();
    }