Google Git
Sign in
apache / jackrabbit-oak / 0ea42704e4eeee784a496a4bc640ad1e1b0c5307 / . / oak-store-document / src / main / java / org / apache / jackrabbit / oak / plugins / document / NodeDocument.java
blob: e22ac1d616df1ea357f096ca7a41c69cfb5e9a77 [file] [log] [blame]
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.jackrabbit.oak.plugins.document;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.NavigableMap;
import java.util.Queue;
import java.util.Set;
import java.util.SortedMap;
import java.util.SortedSet;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import com.google.common.base.Function;
import com.google.common.base.Predicate;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Ordering;
import com.google.common.collect.Queues;
import org.apache.jackrabbit.oak.api.PropertyState;
import org.apache.jackrabbit.oak.commons.PathUtils;
import org.apache.jackrabbit.oak.commons.json.JsopBuilder;
import org.apache.jackrabbit.oak.commons.json.JsopReader;
import org.apache.jackrabbit.oak.commons.json.JsopTokenizer;
import org.apache.jackrabbit.oak.commons.json.JsopWriter;
import org.apache.jackrabbit.oak.plugins.document.memory.MemoryDocumentStore;
import org.apache.jackrabbit.oak.plugins.document.util.Utils;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import static com.google.common.base.Objects.equal;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.collect.ImmutableList.copyOf;
import static com.google.common.collect.Iterables.filter;
import static com.google.common.collect.Iterables.mergeSorted;
import static com.google.common.collect.Iterables.transform;
import static org.apache.jackrabbit.oak.plugins.document.Collection.NODES;
import static org.apache.jackrabbit.oak.plugins.document.StableRevisionComparator.REVERSE;
import static org.apache.jackrabbit.oak.plugins.document.UpdateOp.Key;
import static org.apache.jackrabbit.oak.plugins.document.UpdateOp.Operation;
import static org.apache.jackrabbit.oak.plugins.document.util.Utils.abortingIterable;
import static org.apache.jackrabbit.oak.plugins.document.util.Utils.resolveCommitRevision;
/**
* A document storing data about a node.
*/
public final class NodeDocument extends Document {
/**
* Marker document, which indicates the document does not exist.
*/
public static final NodeDocument NULL = new NodeDocument(new MemoryDocumentStore());
static {
NULL.seal();
}
static final Logger LOG = LoggerFactory.getLogger(NodeDocument.class);
/**
* All NodeDocument ID value would be greater than this value
* It can be used as startKey in DocumentStore#query methods
*/
public static final String MIN_ID_VALUE = "0000000";
/**
* All NodeDocument ID value would be less than this value
* It can be used as endKey in DocumentStore#query methods
*/
public static final String MAX_ID_VALUE = ";";
/**
* A size threshold after which to consider a document a split candidate.
* TODO: check which value is the best one
*/
static final int SPLIT_CANDIDATE_THRESHOLD = 8 * 1024;
/**
* A document size threshold after which a split is forced even if
* {@link #NUM_REVS_THRESHOLD} is not reached.
*/
static final int DOC_SIZE_THRESHOLD = 1024 * 1024;
/**
* Only split off at least this number of revisions.
*/
static final int NUM_REVS_THRESHOLD = 100;
/**
* Create an intermediate previous document when there are this many
* previous documents of equal height.
*/
static final int PREV_SPLIT_FACTOR = 10;
/**
* Revision collision markers set by commits with modifications, which
* overlap with un-merged branch commits.
* Key: revision, value: always true
*/
public static final String COLLISIONS = "_collisions";
/**
* The modified time in seconds (5 second resolution).
*/
public static final String MODIFIED_IN_SECS = "_modified";
/**
* The resolution of the modified time.
*/
static final int MODIFIED_IN_SECS_RESOLUTION = 5;
private static final NavigableMap<Revision, Range> EMPTY_RANGE_MAP =
Maps.unmodifiableNavigableMap(new TreeMap<Revision, Range>(REVERSE));
/**
* The list of revision to root commit depth mappings to find out if a
* revision is actually committed. Depth 0 means the commit is in the root node,
* depth 1 means one node below the root, and so on.
*/
static final String COMMIT_ROOT = "_commitRoot";
/**
* The number of previous documents (documents that contain old revisions of
* this node). This property is only set if multiple documents per node
* exist. This is the case when a node is updated very often in a short
* time, such that the document gets very big.
* <p>
* Key: high revision
* <p>
* Value: low revision / height (see {@link Range#getLowValue()}
*/
private static final String PREVIOUS = "_prev";
/**
* Whether this node is deleted. Key: revision, value: true/false.
*/
private static final String DELETED = "_deleted";
/**
* Flag indicating that whether this node was ever deleted. Its just used as
* a hint. If set to true then it indicates that node was once deleted.
* <p>
* Note that a true value does not mean that node should be considered
* deleted as it might have been resurrected in later revision. Further note
* that it might get reset by maintenance tasks once they discover that it
* indeed was resurrected.
*/
public static final String DELETED_ONCE = "_deletedOnce";
/**
* The list of recent revisions for this node, where this node is the
* root of the commit.
* <p>
* Key: revision.
* <p>
* Value: "c" for a regular (non-branch) commit,
* "c-" + base revision of the successfully merged branch commit,
* "b" + base revision of an un-merged branch commit
*/
static final String REVISIONS = "_revisions";
/**
* The last revision.
* <p>
* Key: machine id, in the form "r0-0-1".
* <p>
* Value: the revision.
*/
private static final String LAST_REV = "_lastRev";
/**
* Flag indicating that there are child nodes present. Its just used as a hint.
* If false then that indicates that there are no child. However if its true its
* not necessary that there are child nodes. It just means at some moment this
* node had a child node
*/
private static final String CHILDREN_FLAG = "_children";
/**
* The node path, in case the id can not be converted to a path.
*/
public static final String PATH = "_path";
public static final String HAS_BINARY_FLAG = "_bin";
/**
* Contains {@link #PREVIOUS} entries that are considered stale (pointing
* to a previous document that had been deleted) and should be removed
* during the next split run.
*/
private static final String STALE_PREV = "_stalePrev";
/**
* Contains revision entries for changes done by branch commits.
*/
private static final String BRANCH_COMMITS = "_bc";
/**
* The revision set by the background document sweeper. The revision
* indicates up to which revision documents have been cleaned by the sweeper
* and all previous non-branch revisions by this cluster node can be
* considered committed.
*/
static final String SWEEP_REV = "_sweepRev";
//~----------------------------< Split Document Types >
/**
* Defines the type of split document. Its value is an integer whose value is
* defined as per
*
* @see org.apache.jackrabbit.oak.plugins.document.NodeDocument.SplitDocType
*/
public static final String SD_TYPE = "_sdType";
/**
* Property name which refers to timestamp (long) of the latest revision kept
* in the document
*/
public static final String SD_MAX_REV_TIME_IN_SECS = "_sdMaxRevTime";
/**
* The path prefix for previous documents.
*/
private static final Path PREVIOUS_PREFIX = new Path("p");
/**
* Return time in seconds with 5 second resolution
*
* @param timestamp time in millis to convert
* @return the time in seconds with the given resolution.
*/
public static long getModifiedInSecs(long timestamp) {
// 5 second resolution
long timeInSec = TimeUnit.MILLISECONDS.toSeconds(timestamp);
return timeInSec - timeInSec % MODIFIED_IN_SECS_RESOLUTION;
}
/**
* A document which is created from splitting a main document can be classified
* into multiple types depending on the content i.e. weather it contains
* REVISIONS, COMMIT_ROOT, property history etc
*/
public enum SplitDocType {
/**
* Not a split document
*/
NONE(-1),
/**
* A split document which contains all types of data
*/
DEFAULT(10),
/**
* A split document which contains all types of data. In addition
* when the split document was created the main document did not had
* any child.
* This type is deprecated because these kind of documents cannot be
* garbage collected independently. The main document may still
* reference _commitRoot entries in the previous document. See OAK-1794
*/
@Deprecated
DEFAULT_NO_CHILD(20),
/**
* A split document which does not contain REVISIONS history.
* This type is deprecated because these kind of documents cannot be
* garbage collected independently. The main document may still
* reference _commitRoot entries in the previous document. See OAK-1794
*/
@Deprecated
PROP_COMMIT_ONLY(30),
/**
* Its an intermediate split document which only contains version ranges
* and does not contain any other attributes
*/
INTERMEDIATE(40),
/**
* A split document which contains all types of data. In addition
* when the split document was created the main document did not had
* any child.
*/
DEFAULT_LEAF(50),
/**
* A split document which does not contain REVISIONS history.
*/
COMMIT_ROOT_ONLY(60),
/**
* A split document which contains all types of data, but no branch
* commits.
*/
DEFAULT_NO_BRANCH(70),
;
final int type;
SplitDocType(int type){
this.type = type;
}
public int typeCode() {
return type;
}
static SplitDocType valueOf(Integer type){
if(type == null){
return NONE;
}
for(SplitDocType docType : values()){
if(docType.type == type){
return docType;
}
}
throw new IllegalArgumentException("Not a valid SplitDocType :" + type);
}
}
public static final long HAS_BINARY_VAL = 1;
final DocumentStore store;
/**
* Parsed and sorted set of previous revisions (without stale references
* to removed previous documents).
*/
private NavigableMap<Revision, Range> previous;
/**
* Time at which this object was check for cache consistency
*/
private final AtomicLong lastCheckTime = new AtomicLong(Revision.getCurrentTimestamp());
private final long creationTime;
NodeDocument(@NotNull DocumentStore store) {
this(store, Revision.getCurrentTimestamp());
}
/**
* Required for serialization
*
* @param store the document store.
* @param creationTime time at which it was created. Would be different from current time
* in case of being resurrected from a serialized for
*/
public NodeDocument(@NotNull DocumentStore store, long creationTime) {
this.store = checkNotNull(store);
this.creationTime = creationTime;
}
/**
* Gets the value map for the given key. This method is similar to {@link
* #get(String)} but will always return a value map. The returned value map
* may span multiple documents if the values of the given <code>key</code>
* were split off to {@link #PREVIOUS} documents.
*
* @param key a string key.
* @return the map associated with the key.
*/
@NotNull
public Map<Revision, String> getValueMap(@NotNull String key) {
return ValueMap.create(this, key);
}
/**
* @return the system time this object was created.
*/
public long getCreated() {
return creationTime;
}
/**
* See also {@link #MODIFIED_IN_SECS}.
*
* @return the time in seconds this document was last modified with five
* seconds precision. Returns {@code null} if none is set.
*/
@Nullable
public Long getModified() {
return (Long) get(MODIFIED_IN_SECS);
}
/**
* Returns {@code true} if this node possibly has children.
* If false then that indicates that there are no child
*
* @return {@code true} if this node has children
*/
public boolean hasChildren() {
Boolean childrenFlag = (Boolean) get(CHILDREN_FLAG);
return childrenFlag != null && childrenFlag;
}
/**
* Returns {@code true} if this document was ever deleted in past.
*/
public boolean wasDeletedOnce() {
Boolean deletedOnceFlag = (Boolean) get(DELETED_ONCE);
return deletedOnceFlag != null && deletedOnceFlag;
}
/**
* Checks if this document has been modified after the given lastModifiedTime
*
* @param lastModifiedTime time to compare against in millis
* @return {@code true} if this document was modified after the given
* lastModifiedTime
*/
public boolean hasBeenModifiedSince(long lastModifiedTime){
Long modified = (Long) get(MODIFIED_IN_SECS);
return modified != null && modified > TimeUnit.MILLISECONDS.toSeconds(lastModifiedTime);
}
/**
* Checks if revision time of all entries in this document is less than the passed
* time
*
* @param maxRevisionTime timemstamp (in millis) of revision to check
* @return {@code true} if timestamp of maximum revision stored in this document
* is less than than the passed revision timestamp
*/
public boolean hasAllRevisionLessThan(long maxRevisionTime){
Long maxRevTimeStamp = (Long) get(SD_MAX_REV_TIME_IN_SECS);
return maxRevTimeStamp != null && maxRevTimeStamp < TimeUnit.MILLISECONDS.toSeconds(maxRevisionTime);
}
/**
* Determines if this document is a split document
*
* @return {@code true} if this document is a split document
*/
public boolean isSplitDocument(){
return getSplitDocType() != SplitDocType.NONE;
}
/**
* Determines the type of split document
*
* @return type of Split Document
*/
public SplitDocType getSplitDocType() {
Object t = get(SD_TYPE);
return t == null ? SplitDocType.valueOf((Integer) null) : SplitDocType.valueOf(((Number) t).intValue());
}
/**
* Mark this instance as up-to-date (matches the state in persistence
* store).
*
* @param checkTime time at which the check was performed
*/
public void markUpToDate(long checkTime) {
lastCheckTime.set(checkTime);
}
/**
* Returns the last time when this object was checked for consistency.
*
* @return the last check time
*/
public long getLastCheckTime() {
return lastCheckTime.get();
}
public boolean hasBinary() {
Number flag = (Number) get(HAS_BINARY_FLAG);
return flag != null && flag.intValue() == HAS_BINARY_VAL;
}
/**
* Returns the path of the main document if this document is part of a _prev
* history tree. Otherwise this method simply returns {@link #getPath()}.
*
* @return the path of the main document.
*/
@NotNull
public Path getMainPath() {
String p = getPathString();
if (p.startsWith("p")) {
p = PathUtils.getAncestorPath(p, 2);
if (p.length() == 1) {
return Path.ROOT;
} else {
p = p.substring(1);
}
}
return Path.fromString(p);
}
/**
* @return a map of the last known revision for each clusterId.
*/
@NotNull
public Map<Integer, Revision> getLastRev() {
Map<Integer, Revision> map = Maps.newHashMap();
Map<Revision, String> valueMap = getLocalMap(LAST_REV);
for (Map.Entry<Revision, String> e : valueMap.entrySet()) {
int clusterId = e.getKey().getClusterId();
Revision rev = Revision.fromString(e.getValue());
map.put(clusterId, rev);
}
return map;
}
/**
* Returns <code>true</code> if this document contains an entry for the
* given <code>revision</code> in the {@link #REVISIONS} map. Please note
* that an entry in the {@link #REVISIONS} map does not necessarily mean
* the the revision is committed.
* Use {@link RevisionContext#getCommitValue(Revision, NodeDocument)} to get
* the commit state of a revision.
*
* @param revision the revision to check.
* @return <code>true</code> if this document contains the given revision.
*/
public boolean containsRevision(@NotNull Revision revision) {
if (getLocalRevisions().containsKey(revision)) {
return true;
}
for (NodeDocument prev : getPreviousDocs(REVISIONS, revision)) {
if (prev.containsRevision(revision)) {
return true;
}
}
return false;
}
/**
* Purge the uncommitted revisions of this document with the
* local cluster node id as returned by the {@link RevisionContext}. These
* are the {@link #REVISIONS} and {@link #BRANCH_COMMITS} entries where
* {@link Utils#isCommitted(String)} returns false.
*
* <p>
* <bold>Note</bold> - This method should only be invoked upon startup
* as then only we can safely assume that these revisions would not be
* committed
* </p>
*
* @param context the revision context.
* @return count of the revision entries purged
*/
int purgeUncommittedRevisions(RevisionContext context) {
// only look at revisions in this document.
// uncommitted revisions are not split off
Map<Revision, String> localRevisions = getLocalRevisions();
UpdateOp op = new UpdateOp(getId(), false);
Set<Revision> uniqueRevisions = new HashSet<>();
for (Map.Entry<Revision, String> commit : localRevisions.entrySet()) {
if (!Utils.isCommitted(commit.getValue())) {
Revision r = commit.getKey();
if (r.getClusterId() == context.getClusterId()) {
uniqueRevisions.add(r);
removeRevision(op, r);
}
}
}
for (Revision r : getLocalBranchCommits()) {
String commitValue = localRevisions.get(r);
if (!Utils.isCommitted(commitValue)) {
if (r.getClusterId() == context.getClusterId()) {
uniqueRevisions.add(r);
removeBranchCommit(op, r);
}
}
}
if (op.hasChanges()) {
store.findAndUpdate(Collection.NODES, op);
}
return uniqueRevisions.size();
}
/**
* Purge collision markers with the local clusterId on this document. Use
* only on start when there are no ongoing or pending commits.
*
* @param context the revision context.
* @return the number of removed collision markers.
*/
int purgeCollisionMarkers(RevisionContext context) {
Map<Revision, String> valueMap = getLocalMap(COLLISIONS);
UpdateOp op = new UpdateOp(getId(), false);
int purgeCount = 0;
for (Map.Entry<Revision, String> commit : valueMap.entrySet()) {
Revision r = commit.getKey();
if (r.getClusterId() == context.getClusterId()) {
purgeCount++;
removeCollision(op, r);
}
}
if (op.hasChanges()) {
store.findAndUpdate(Collection.NODES, op);
}
return purgeCount;
}
/**
* Returns the conflicts on the given {@code changes} if there are any. The
* returned revisions are the commits, which created the collision markers
* for one of the {@code changes}.
*
* @param changes the changes to check.
* @return the conflict revisions.
*/
@NotNull
Set<Revision> getConflictsFor(@NotNull Iterable<Revision> changes) {
checkNotNull(changes);
Set<Revision> conflicts = Sets.newHashSet();
Map<Revision, String> collisions = getLocalMap(COLLISIONS);
for (Revision r : changes) {
String value = collisions.get(r.asTrunkRevision());
if (value == null) {
continue;
}
try {
conflicts.add(Revision.fromString(value));
} catch (IllegalArgumentException e) {
// backward compatibility: collision marker with value 'true'
}
}
return conflicts;
}
/**
* Returns the commit root path for the given <code>revision</code> or
* <code>null</code> if this document does not have a commit root entry for
* the given <code>revision</code>.
*
* @param revision a revision.
* @return the commit root path or <code>null</code>.
*/
@Nullable
public Path getCommitRootPath(Revision revision) {
String depth = getCommitRootDepth(revision);
if (depth != null) {
return getPathAtDepth(depth);
}
return null;
}
/**
* Get the revision of the latest change made to this node. At the same
* time this method collects all collisions that happened for the given
* {@code changeRev}. The reported latest change takes branches into
* account. This means, if {@code changeRev} is on a branch, the latest
* change is either a change that was done by a preceding branch commit or
* a change that happened before the base of the branch. Changes done after
* the branch base on trunk are not considered in this case. For a trunk
* commit the latest change is reported similarly. In this case, unmerged
* branch commits are not considered as latest change. Only commits to trunk
* are considered.
*
* Collisions include the following cases:
* <ul>
* <li>The other change is not yet committed</li>
* <li>The other change is a branch commit and not yet merged</li>
* <li>The {@code changeRev} is a branch commit and the other change
* happened after the base revision of the branch</li>
* <li>The other change is from another cluster node and not yet
* visible</li>
* </ul>
*
* @param context the revision context.
* @param baseRev the base revision of the current change.
* @param changeRev the revision of the current change.
* @param branch the branch associated with the current change or
* {@code null} if {@code changeRev} is not a branch commit.
* @param collisions changes that happened after {@code baseRev}.
*/
@Nullable
Revision getNewestRevision(final RevisionContext context,
final RevisionVector baseRev,
final Revision changeRev,
final Branch branch,
final Set<Revision> collisions) {
checkArgument(!baseRev.isBranch() || branch != null,
"Branch must be non-null if baseRev is a branch revision");
RevisionVector head = context.getHeadRevision();
RevisionVector lower = branch != null ? branch.getBase() : baseRev;
// the clusterIds to check when walking the changes
Set<Integer> clusterIds = Collections.emptySet();
if (!getPreviousRanges().isEmpty()) {
clusterIds = Sets.newHashSet();
for (Revision prevRev : getPreviousRanges().keySet()) {
if (lower.isRevisionNewer(prevRev) ||
equal(prevRev, lower.getRevision(prevRev.getClusterId()))) {
clusterIds.add(prevRev.getClusterId());
}
}
if (!clusterIds.isEmpty()) {
// add clusterIds of local changes as well
for (Revision r : getLocalCommitRoot().keySet()) {
clusterIds.add(r.getClusterId());
}
for (Revision r : getLocalRevisions().keySet()) {
clusterIds.add(r.getClusterId());
}
}
}
// if we don't have clusterIds, we can use the local changes only
boolean fullScan = true;
Iterable<Revision> changes = Iterables.mergeSorted(
ImmutableList.of(
getLocalRevisions().keySet(),
getLocalCommitRoot().keySet()),
getLocalRevisions().comparator()
);
if (!clusterIds.isEmpty()) {
// there are some previous documents that potentially
// contain changes after 'lower' revision vector
// include previous documents as well (only needed in rare cases)
fullScan = false;
changes = Iterables.mergeSorted(
ImmutableList.of(
changes,
getChanges(REVISIONS, lower),
getChanges(COMMIT_ROOT, lower)
), getLocalRevisions().comparator()
);
if (LOG.isDebugEnabled()) {
LOG.debug("getNewestRevision() with changeRev {} on {}, " +
"_revisions {}, _commitRoot {}",
changeRev, getId(), getLocalRevisions(), getLocalCommitRoot());
}
}
Map<Integer, Revision> newestRevs = Maps.newHashMap();
Map<Revision, String> validRevisions = Maps.newHashMap();
for (Revision r : changes) {
if (r.equals(changeRev)) {
continue;
}
if (!fullScan) {
// check if we can stop going through changes
if (clusterIds.contains(r.getClusterId())
&& !lower.isRevisionNewer(r)
&& newestRevs.containsKey(r.getClusterId())) {
clusterIds.remove(r.getClusterId());
if (clusterIds.isEmpty()) {
// all remaining revisions are older than
// the lower bound
break;
}
}
}
if (newestRevs.containsKey(r.getClusterId())) {
// we already found the newest revision for this clusterId
// from a baseRev point of view
// we still need to find collisions up to the base
// of the branch if this is for a commit on a branch
if (branch != null && !branch.containsCommit(r)) {
// change does not belong to the branch
if (branch.getBase(changeRev).isRevisionNewer(r)) {
// and happened after the base of the branch
collisions.add(r);
}
}
} else {
// we don't yet have the newest committed change
// for this clusterId
// check if change is visible from baseRev
if (isValidRevision(context, r, null, baseRev, validRevisions)) {
// consider for newestRev
newestRevs.put(r.getClusterId(), r);
} else {
// not valid means:
// 1) 'r' is not committed -> collision
// 2) 'r' is on a branch, but not the same as
// changeRev -> collisions
// 3) changeRev is on a branch and 'r' is newer than
// the base of the branch -> collision
// 4) 'r' is committed but not yet visible to current
// cluster node -> collisions
// 5) changeRev is not on a branch, 'r' is committed and
// newer than baseRev -> newestRev
Revision commitRevision = null;
String cv = context.getCommitValue(r, this);
if (Utils.isCommitted(cv)) {
commitRevision = resolveCommitRevision(r, cv);
}
if (commitRevision != null // committed but not yet visible
&& head.isRevisionNewer(commitRevision)) {
// case 4)
collisions.add(r);
} else if (commitRevision != null // committed
&& branch == null // changeRev not on branch
&& baseRev.isRevisionNewer(r)) {
// case 5)
newestRevs.put(r.getClusterId(), r);
} else {
// remaining cases 1), 2) and 3)
collisions.add(r);
}
}
}
}
// select the newest committed change
Revision newestRev = null;
for (Revision r : newestRevs.values()) {
newestRev = Utils.max(newestRev, r, StableRevisionComparator.INSTANCE);
}
if (newestRev == null) {
return null;
}
// the local deleted map contains the most recent revisions
SortedMap<Revision, String> deleted = getLocalDeleted();
String value = deleted.get(newestRev);
if (value == null && deleted.headMap(newestRev).isEmpty()) {
// newestRev is newer than most recent entry in local deleted
// no need to check previous docs
return newestRev;
}
if (value == null) {
// get from complete map
value = getDeleted().get(newestRev);
}
if ("true".equals(value)) {
// deleted in the newest revision
return null;
}
return newestRev;
}
/**
* Checks if the revision is valid for the given document. A revision is
* considered valid if the given document is the root of the commit, or the
* commit root has the revision set. This method may read further documents
* to perform this check.
* This method also takes pending branches into consideration.
* The <code>readRevision</code> identifies the read revision used by the
* client, which may be a branch revision logged in {@link RevisionContext#getBranches()}.
* The revision <code>rev</code> is valid if it is part of the branch
* history of <code>readRevision</code>.
*
* @param rev revision to check.
* @param commitValue the commit value of the revision to check or
* <code>null</code> if unknown.
* @param readRevision the read revision of the client.
* @param validRevisions map of revisions to commit value already checked
* against <code>readRevision</code> and considered
* valid.
* @return <code>true</code> if the revision is valid; <code>false</code>
* otherwise.
*/
private boolean isValidRevision(@NotNull RevisionContext context,
@NotNull Revision rev,
@Nullable String commitValue,
@NotNull RevisionVector readRevision,
@NotNull Map<Revision, String> validRevisions) {
if (validRevisions.containsKey(rev)) {
return true;
}
// get the commit value if it is not yet available
if (commitValue == null) {
commitValue = context.getCommitValue(rev, this);
}
if (commitValue == null) {
// this change is not committed, hence not valid/visible
return false;
}
if (isVisible(context, rev, commitValue, readRevision)) {
validRevisions.put(rev, commitValue);
return true;
}
return false;
}
/**
* Returns a {@link DocumentNodeState} as seen at the given
* <code>readRevision</code>.
*
* @param nodeStore the node store.
* @param readRevision the read revision.
* @param lastModified the revision when this node was last modified, but
* the value is potentially not yet reflected in this
* document.
* See {@link RevisionContext#getPendingModifications()}.
* @return the node or <code>null</code> if the node doesn't exist at the
* given read revision.
*/
@Nullable
public DocumentNodeState getNodeAtRevision(@NotNull DocumentNodeStore nodeStore,
@NotNull RevisionVector readRevision,
@Nullable Revision lastModified) {
Map<Revision, String> validRevisions = Maps.newHashMap();
Branch branch = nodeStore.getBranches().getBranch(readRevision);
LastRevs lastRevs = createLastRevs(readRevision,
nodeStore, branch, lastModified);
Revision min = getLiveRevision(nodeStore, readRevision, validRevisions, lastRevs);
if (min == null) {
// deleted
return null;
}
Path path = getPath();
List<PropertyState> props = Lists.newArrayList();
for (String key : keySet()) {
if (!Utils.isPropertyName(key)) {
continue;
}
// ignore when local map is empty (OAK-2442)
SortedMap<Revision, String> local = getLocalMap(key);
if (local.isEmpty()) {
continue;
}
// first check local map, which contains most recent values
Value value = getLatestValue(nodeStore, local.entrySet(),
readRevision, validRevisions, lastRevs);
// check if there may be more recent values in a previous document
if (value != null
&& !getPreviousRanges().isEmpty()
&& !isMostRecentCommitted(local, value.revision, nodeStore)) {
// not reading the most recent value, we may need to
// consider previous documents as well
for (Revision prev : getPreviousRanges().keySet()) {
if (prev.compareRevisionTimeThenClusterId(value.revision) > 0) {
// a previous document has more recent changes
// than value.revision
value = null;
break;
}
}
}
if (value == null && !getPreviousRanges().isEmpty()) {
// check revision history
value = getLatestValue(nodeStore, getVisibleChanges(key, readRevision),
readRevision, validRevisions, lastRevs);
}
String propertyName = Utils.unescapePropertyName(key);
String v = value != null ? value.value : null;
if (v != null){
props.add(nodeStore.createPropertyState(propertyName, v));
}
}
// when was this node last modified?
RevisionVector lastRevision = new RevisionVector(min);
RevisionVector branchBase = null;
if (branch != null) {
branchBase = branch.getBase(readRevision.getBranchRevision());
}
for (Revision r : lastRevs) {
if (readRevision.isRevisionNewer(r)) {
// the node has a _lastRev which is newer than readRevision
// this means we don't know when this node was
// modified by an operation on a descendant node between
// current lastRevision and readRevision. therefore we have
// to stay on the safe side and use readRevision
Revision rev = readRevision.getRevision(r.getClusterId());
if (rev != null) {
lastRevision = lastRevision.update(rev);
} else {
// readRevision does not have a revision for this
// clusterId -> remove from lastRevision
lastRevision = lastRevision.remove(r.getClusterId());
}
} else if (branchBase != null && branchBase.isRevisionNewer(r)) {
// readRevision is on a branch and the node has a
// _lastRev which is newer than the base of the branch
// we cannot use this _lastRev because it is not visible
// from this branch. highest possible revision of visible
// changes is the base of the branch
Revision rev = branchBase.getRevision(r.getClusterId());
if (rev != null) {
lastRevision = lastRevision.update(rev);
} else {
// branchBase does not have a revision for this
// clusterId -> remove from lastRevision
lastRevision = lastRevision.remove(r.getClusterId());
}
} else if (lastRevision.isRevisionNewer(r)) {
lastRevision = lastRevision.update(r);
}
}
if (branch != null) {
// read from a branch
// -> possibly overlay with unsaved last revs from branch
lastRevs.updateBranch(branch.getUnsavedLastRevision(path, readRevision.getBranchRevision()));
Revision r = lastRevs.getBranchRevision();
if (r != null) {
lastRevision = lastRevision.update(r);
}
}
return new DocumentNodeState(nodeStore, path, readRevision, props, hasChildren(), lastRevision);
}
/**
* Get the earliest (oldest) revision where the node was alive at or before
* the provided revision, if the node was alive at the given revision.
*
* @param context the revision context
* @param readRevision the read revision
* @param validRevisions the map of revisions to commit value already
* checked against readRevision and considered valid.
* @param lastRevs to keep track of the last modification.
* @return the earliest revision, or null if the node is deleted at the
* given revision
*/
@Nullable
public Revision getLiveRevision(RevisionContext context,
RevisionVector readRevision,
Map<Revision, String> validRevisions,
LastRevs lastRevs) {
// check local deleted map first
Value value = getLatestValue(context, getLocalDeleted().entrySet(), readRevision, validRevisions, lastRevs);
if (value == null && !getPreviousRanges().isEmpty()) {
// need to check complete map
value = getLatestValue(context, getDeleted().entrySet(), readRevision, validRevisions, lastRevs);
}
return value != null && "false".equals(value.value) ? value.revision : null;
}
/**
* Returns <code>true</code> if the given operation is conflicting with this
* document.
*
* @param op the update operation.
* @param baseRevision the base revision for the update operation.
* @param commitRevision the commit revision of the update operation.
* @param enableConcurrentAddRemove feature flag for OAK-2673.
* @return <code>true</code> if conflicting, <code>false</code> otherwise.
*/
boolean isConflicting(@NotNull UpdateOp op,
@NotNull RevisionVector baseRevision,
@NotNull Revision commitRevision,
boolean enableConcurrentAddRemove) {
// did existence of node change after baseRevision?
// only check local deleted map, which contains the most
// recent values
Map<Revision, String> deleted = getLocalDeleted();
boolean allowConflictingDeleteChange =
enableConcurrentAddRemove && allowConflictingDeleteChange(op);
for (Map.Entry<Revision, String> entry : deleted.entrySet()) {
if (entry.getKey().equals(commitRevision)) {
continue;
}
if (baseRevision.isRevisionNewer(entry.getKey())) {
boolean newerDeleted = Boolean.parseBoolean(entry.getValue());
if (!allowConflictingDeleteChange || op.isDelete() != newerDeleted) {
return true;
}
}
}
for (Map.Entry<Key, Operation> entry : op.getChanges().entrySet()) {
if (entry.getValue().type != Operation.Type.SET_MAP_ENTRY) {
continue;
}
String name = entry.getKey().getName();
if (DELETED.equals(name) && !allowConflictingDeleteChange) {
// existence of node changed, this always conflicts with
// any other concurrent change
return true;
}
if (!Utils.isPropertyName(name)) {
continue;
}
// was this property touched after baseRevision?
for (Revision rev : getChanges(name, baseRevision)) {
if (rev.equals(commitRevision)) {
continue;
}
if (baseRevision.isRevisionNewer(rev)) {
return true;
}
}
}
return false;
}
/**
* Utility method to check if {@code op} can be allowed to change
* {@link #DELETED} property. Basic idea is that a change in
* {@link #DELETED} property should be consistent if final value is same
* and there are no observation semantic change. Thus, this method tries to
* be very conservative and allows delete iff:
* <ul>
* <li>{@code doc} represents and internal path</li>
* <li>{@code op} represents an add or delete operation</li>
* <li>{@code op} doesn't change add/delete any exposed property</li>
* <li>{@code doc} doesn't have any exposed property</li>
* </ul>
* <i>
* Note: This method is a broad level check if we can allow such conflict
* resolution. Actual cases, like allow-delete-delete, allow-add-add wrt to
* revision are not handled here.
* </i>
* @param op {@link UpdateOp} instance having changes to check {@code doc} against
* @return if conflicting change in {@link #DELETED} property is allowed
*/
private boolean allowConflictingDeleteChange(UpdateOp op) {
String path = getPathString();
if (!Utils.isHiddenPath(path)) {
return false;
}
if (!op.isNew() && !op.isDelete()) {
return false;//only handle added/delete operations
}
for (Key opKey : op.getChanges().keySet()) {
String name = opKey.getName();
if (Utils.isPropertyName(name)) {
return false; //only handle changes to internal properties
}
}
// Only look at local data ...
// even remotely updated properties should have an entry (although invisible)
// by the time we are looking for conflicts
for (String dataKey : keySet()) {
if (Utils.isPropertyName(dataKey)) {
return false; //only handle changes to internal properties
}
}
return true;
}
/**
* Returns update operations to split this document. The implementation may
* decide to not return any operations if no splitting is required. A caller
* must explicitly pass a head revision even though it is available through
* the {@link RevisionContext}. The given head revision must reflect a head
* state before {@code doc} was retrieved from the document store. This is
* important in order to maintain consistency. See OAK-3081 for details.
*
* @param context the revision context.
* @param head the head revision before this document was retrieved from
* the document store.
* @param binarySize a function that returns the binary size of the given
* JSON property value String.
* @return the split operations, whereby the last updateOp is guaranteed to be
* the update of the main document (unless the entire list is empty)
*/
@NotNull
public Iterable<UpdateOp> split(@NotNull RevisionContext context,
@NotNull RevisionVector head,
@NotNull Function<String, Long> binarySize) {
return SplitOperations.forDocument(this, context, head,
binarySize, NUM_REVS_THRESHOLD);
}
/**
* Returns previous revision ranges for this document. The revision keys are
* sorted descending, newest first! The returned map does not include stale
* entries.
* This method is equivalent to calling {@link #getPreviousRanges(boolean)}
* with {@code includeStale} set to false.
*
* @return the previous ranges for this document.
*/
@NotNull
NavigableMap<Revision, Range> getPreviousRanges() {
return getPreviousRanges(false);
}
/**
* Returns previous revision ranges for this document. The revision keys are
* sorted descending, newest first!
*
* @param includeStale whether stale revision ranges are included or not.
* @return the previous ranges for this document.
*/
@NotNull
NavigableMap<Revision, Range> getPreviousRanges(boolean includeStale) {
if (includeStale) {
return createPreviousRanges(true);
} else {
if (previous == null) {
previous = createPreviousRanges(false);
}
return previous;
}
}
/**
* Creates a map with previous revision ranges for this document. The
* revision keys are sorted descending, newest first!
*
* @param includeStale whether stale revision ranges are included or not.
* @return the previous ranges for this document.
*/
@NotNull
private NavigableMap<Revision, Range> createPreviousRanges(boolean includeStale) {
NavigableMap<Revision, Range> ranges;
Map<Revision, String> map = getLocalMap(PREVIOUS);
if (map.isEmpty()) {
ranges = EMPTY_RANGE_MAP;
} else {
Map<Revision, String> stale = Collections.emptyMap();
if (!includeStale) {
stale = getLocalMap(STALE_PREV);
}
NavigableMap<Revision, Range> transformed =
new TreeMap<Revision, Range>(REVERSE);
for (Map.Entry<Revision, String> entry : map.entrySet()) {
Range r = Range.fromEntry(entry.getKey(), entry.getValue());
if (String.valueOf(r.height).equals(stale.get(r.high))) {
continue;
}
transformed.put(r.high, r);
}
ranges = Maps.unmodifiableNavigableMap(transformed);
}
return ranges;
}
/**
* Returns previous {@link NodeDocument}, which include entries for the
* property in the given revision.
* If the <code>revision</code> is <code>null</code>, then all previous
* documents with changes for the given property are returned. The returned
* documents are returned in descending revision order (newest first).
*
* @param property the name of a property.
* @param revision the revision to match or <code>null</code>.
* @return previous documents.
*/
@NotNull
Iterable<NodeDocument> getPreviousDocs(@NotNull final String property,
@Nullable final Revision revision) {
if (getPreviousRanges().isEmpty()) {
return Collections.emptyList();
}
if (revision == null) {
return new PropertyHistory(this, property);
} else {
final Path mainPath = getMainPath();
// first try to lookup revision directly
Map.Entry<Revision, Range> entry = getPreviousRanges().floorEntry(revision);
if (entry != null) {
Revision r = entry.getKey();
int h = entry.getValue().height;
String prevId = Utils.getPreviousIdFor(mainPath, r, h);
NodeDocument prev = getPreviousDocument(prevId);
if (prev != null) {
if (prev.getValueMap(property).containsKey(revision)) {
return Collections.singleton(prev);
}
} else {
previousDocumentNotFound(prevId, revision);
}
}
// didn't find entry -> scan through remaining head ranges
return filter(transform(getPreviousRanges().headMap(revision).entrySet(),
new Function<Map.Entry<Revision, Range>, NodeDocument>() {
@Override
public NodeDocument apply(Map.Entry<Revision, Range> input) {
if (input.getValue().includes(revision)) {
return getPreviousDoc(input.getKey(), input.getValue());
}
return null;
}
}), new Predicate<NodeDocument>() {
@Override
public boolean apply(@Nullable NodeDocument input) {
return input != null && input.getValueMap(property).containsKey(revision);
}
});
}
}
NodeDocument getPreviousDocument(String prevId){
LOG.trace("get previous document {}", prevId);
NodeDocument doc = store.find(Collection.NODES, prevId);
if (doc == null) {
// In case secondary read preference is used and node is not found
// then check with primary again as it might happen that node document has not been
// replicated. We know that document with such an id must exist but possibly dut to
// replication lag it has not reached to secondary. So in that case read again
// from primary
doc = store.find(Collection.NODES, prevId, 0);
}
return doc;
}
@NotNull
Iterator<NodeDocument> getAllPreviousDocs() {
if (getPreviousRanges().isEmpty()) {
return Collections.emptyIterator();
}
//Currently this method would fire one query per previous doc
//If that poses a problem we can try to find all prev doc by relying
//on property that all prevDoc id would starts <depth+2>:p/path/to/node
return new AbstractIterator<NodeDocument>(){
private Queue<Map.Entry<Revision, Range>> previousRanges =
Queues.newArrayDeque(getPreviousRanges().entrySet());
@Override
protected NodeDocument computeNext() {
if(!previousRanges.isEmpty()){
Map.Entry<Revision, Range> e = previousRanges.remove();
NodeDocument prev = getPreviousDoc(e.getKey(), e.getValue());
if(prev != null){
previousRanges.addAll(prev.getPreviousRanges().entrySet());
return prev;
}
}
return endOfData();
}
};
}
/**
* Returns previous leaf documents. Those are the previous documents with
* a type {@code !=} {@link SplitDocType#INTERMEDIATE}. The documents are
* returned in descending order based on the most recent change recorded
* in the previous document. A change is defined as an entry in either the
* {@link #REVISIONS} or {@link #COMMIT_ROOT} map.
*
* @return the leaf documents in descending order.
*/
@NotNull
Iterator<NodeDocument> getPreviousDocLeaves() {
if (getPreviousRanges().isEmpty()) {
return Collections.emptyIterator();
}
// create a mutable copy
final NavigableMap<Revision, Range> ranges = Maps.newTreeMap(getPreviousRanges());
return new AbstractIterator<NodeDocument>() {
@Override
protected NodeDocument computeNext() {
NodeDocument next;
for (;;) {
Map.Entry<Revision, Range> topEntry = ranges.pollFirstEntry();
if (topEntry == null) {
// no more ranges
next = endOfData();
break;
}
NodeDocument prev = getPreviousDoc(topEntry.getKey(), topEntry.getValue());
if (prev == null) {
// move on to next range
continue;
}
if (topEntry.getValue().getHeight() == 0) {
// this is a leaf
next = prev;
break;
} else {
// replace intermediate entry with its previous ranges
ranges.putAll(prev.getPreviousRanges());
}
}
return next;
}
};
}
@Nullable
private NodeDocument getPreviousDoc(Revision rev, Range range){
int h = range.height;
String prevId = Utils.getPreviousIdFor(getMainPath(), rev, h);
NodeDocument prev = getPreviousDocument(prevId);
if (prev != null) {
return prev;
} else {
previousDocumentNotFound(prevId, rev);
}
return null;
}
/**
* Returns the document that contains a reference to the previous document
* identified by {@code revision} and {@code height}. This is either the
* current document or an intermediate split document. This method returns
* {@code null} if there is no such reference.
*
* @param revision the high revision of a range entry in {@link #PREVIOUS}.
* @param height the height of the entry in {@link #PREVIOUS}.
* @return the document with the entry or {@code null} if not found.
*/
@Nullable
NodeDocument findPrevReferencingDoc(Revision revision, int height) {
for (Range range : getPreviousRanges().values()) {
if (range.getHeight() == height && range.high.equals(revision)) {
return this;
} else if (range.includes(revision)) {
String prevId = Utils.getPreviousIdFor(
getMainPath(), range.high, range.height);
NodeDocument prev = store.find(NODES, prevId);
if (prev == null) {
LOG.warn("Split document {} does not exist anymore. Main document is {}",
prevId, Utils.getIdFromPath(getMainPath()));
continue;
}
// recurse into the split hierarchy
NodeDocument doc = prev.findPrevReferencingDoc(revision, height);
if (doc != null) {
return doc;
}
}
}
return null;
}
/**
* Returns an {@link Iterable} of {@link Revision} of all changes performed
* on this document. This covers all entries for {@link #REVISIONS} and
* {@link #COMMIT_ROOT} including previous documents. The revisions are
* returned in descending stable revision order using
* {@link StableRevisionComparator#REVERSE}.
*
* @return revisions of all changes performed on this document.
*/
Iterable<Revision> getAllChanges() {
RevisionVector empty = new RevisionVector();
return Iterables.mergeSorted(ImmutableList.of(
getChanges(REVISIONS, empty),
getChanges(COMMIT_ROOT, empty)
), StableRevisionComparator.REVERSE);
}
/**
* Returns all changes for the given property back to {@code min} revision
* (exclusive). The revisions include committed as well as uncommitted
* changes. The returned revisions are sorted in reverse order (newest
* first).
*
* @param property the name of the property.
* @param min the lower bound revision (exclusive).
* @return changes back to {@code min} revision.
*/
@NotNull
Iterable<Revision> getChanges(@NotNull final String property,
@NotNull final RevisionVector min) {
Predicate<Revision> p = new Predicate<Revision>() {
@Override
public boolean apply(Revision input) {
return min.isRevisionNewer(input);
}
};
List<Iterable<Revision>> changes = Lists.newArrayList();
changes.add(abortingIterable(getLocalMap(property).keySet(), p));
for (Map.Entry<Revision, Range> e : getPreviousRanges().entrySet()) {
if (min.isRevisionNewer(e.getKey())) {
final NodeDocument prev = getPreviousDoc(e.getKey(), e.getValue());
if (prev != null) {
changes.add(abortingIterable(prev.getValueMap(property).keySet(), p));
}
}
}
if (changes.size() == 1) {
return changes.get(0);
} else {
return Iterables.mergeSorted(changes, StableRevisionComparator.REVERSE);
}
}
/**
* Returns all changes for the given property that are visible from the
* {@code readRevision} vector. The revisions include committed as well as
* uncommitted changes. The returned revisions are sorted in reverse order
* (newest first).
*
* @param property the name of the property.
* @param readRevision the read revision vector.
* @return property changes visible from the given read revision vector.
*/
@NotNull
Iterable<Map.Entry<Revision, String>> getVisibleChanges(@NotNull final String property,
@NotNull final RevisionVector readRevision) {
Predicate<Map.Entry<Revision, String>> p = new Predicate<Map.Entry<Revision, String>>() {
@Override
public boolean apply(Map.Entry<Revision, String> input) {
return !readRevision.isRevisionNewer(input.getKey());
}
};
List<Iterable<Map.Entry<Revision, String>>> changes = Lists.newArrayList();
Map<Revision, String> localChanges = getLocalMap(property);
if (!localChanges.isEmpty()) {
changes.add(filter(localChanges.entrySet(), p));
}
for (Revision r : readRevision) {
// collect changes per clusterId
collectVisiblePreviousChanges(property, r, changes);
}
if (changes.size() == 1) {
return changes.get(0);
} else {
return mergeSorted(changes, ValueComparator.REVERSE);
}
}
/**
* Collect changes in previous documents into {@code changes} visible from
* the given {@code readRevision} and for the given {@code property}. The
* {@code Iterable} added to the {@code changes} list must be in descending
* revision order.
*
* @param property the name of the property.
* @param readRevision collect changes for this part of the readRevision.
* @param changes where to add the changes to.
*/
private void collectVisiblePreviousChanges(@NotNull final String property,
@NotNull final Revision readRevision,
@NotNull final List<Iterable<Entry<Revision, String>>> changes) {
List<Iterable<Map.Entry<Revision, String>>> revs = Lists.newArrayList();
RevisionVector readRV = new RevisionVector(readRevision);
List<Range> ranges = new ArrayList<>();
for (Range r : getPreviousRanges().values()) {
if (r.low.getClusterId() == readRevision.getClusterId()
&& readRevision.compareRevisionTime(r.low) >= 0) {
// clusterId matches and range is visible from read revision
ranges.add(r);
}
}
List<Range> batch = new ArrayList<>();
while (!ranges.isEmpty()) {
// create batches of non-overlapping ranges
Range previous = null;
Iterator<Range> it = ranges.iterator();
while (it.hasNext()) {
Range r = it.next();
if (previous == null || r.high.compareRevisionTime(previous.low) < 0) {
// first range or does not overlap with previous in batch
batch.add(r);
it.remove();
previous = r;
}
}
revs.add(changesFor(batch, readRV, property));
batch.clear();
}
if (revs.size() == 1) {
// optimize single batch case
changes.add(revs.get(0));
} else if (!revs.isEmpty()) {
// merge sort them
changes.add(mergeSorted(revs, ValueComparator.REVERSE));
}
}
/**
* Get changes of {@code property} for the given list of {@code ranges}
* visible from {@code readRev}.
*
* @param ranges a list of ranges of previous document where to read the
* changes from.
* @param readRev get changes visible from this read revision.
* @param property the name of the property to read changes.
* @return iterable over the changes.
*/
private Iterable<Map.Entry<Revision, String>> changesFor(final List<Range> ranges,
final RevisionVector readRev,
final String property) {
if (ranges.isEmpty()) {
return Collections.emptyList();
}
final Function<Range, Iterable<Map.Entry<Revision, String>>> rangeToChanges =
new Function<Range, Iterable<Map.Entry<Revision, String>>>() {
@Override
public Iterable<Map.Entry<Revision, String>> apply(Range input) {
NodeDocument doc = getPreviousDoc(input.high, input);
if (doc == null) {
return Collections.emptyList();
}
return doc.getVisibleChanges(property, readRev);
}
};
Iterable<Map.Entry<Revision, String>> changes;
if (ranges.size() == 1) {
final Range range = ranges.get(0);
changes = new Iterable<Entry<Revision, String>>() {
@SuppressWarnings("ConstantConditions")
@Override
public Iterator<Entry<Revision, String>> iterator() {
return rangeToChanges.apply(range).iterator();
}
};
} else {
changes = Iterables.concat(transform(copyOf(ranges), rangeToChanges));
}
return filter(changes, new Predicate<Entry<Revision, String>>() {
@Override
public boolean apply(Entry<Revision, String> input) {
return !readRev.isRevisionNewer(input.getKey());
}
});
}
/**
* Returns the local value map for the given key.
*
* @param key the key.
* @return local value map.
*/
@NotNull
SortedMap<Revision, String> getLocalMap(String key) {
@SuppressWarnings("unchecked")
SortedMap<Revision, String> map = (SortedMap<Revision, String>) data.get(key);
if (map == null) {
map = ValueMap.EMPTY;
}
return map;
}
/**
* @return the {@link #REVISIONS} stored on this document.
*/
@NotNull
SortedMap<Revision, String> getLocalRevisions() {
return getLocalMap(REVISIONS);
}
@NotNull
SortedMap<Revision, String> getLocalCommitRoot() {
return getLocalMap(COMMIT_ROOT);
}
@NotNull
SortedMap<Revision, String> getLocalDeleted() {
return getLocalMap(DELETED);
}
@NotNull
SortedMap<Revision, String> getStalePrev() {
return getLocalMap(STALE_PREV);
}
/**
* Returns the branch commit entries on this document
* ({@link #BRANCH_COMMITS}). This method does not consider previous
* documents, but only returns the entries on this document.
*/
@NotNull
public Set<Revision> getLocalBranchCommits() {
return getLocalMap(BRANCH_COMMITS).keySet();
}
/**
* Resolves the commit value for the change with the given revision on this
* document. If necessary, this method will lookup the commit value on the
* referenced commit root document.
*
* @param revision the revision of a change on this document.
* @return the commit value associated with the change.
*/
@Nullable
String resolveCommitValue(Revision revision) {
NodeDocument commitRoot = getCommitRoot(revision);
if (commitRoot == null) {
return null;
}
return commitRoot.getCommitValue(revision);
}
/**
* Returns the sweep revisions on this document as a {@link RevisionVector}.
* This method will return an empty {@link RevisionVector} if this document
* doesn't have any sweep revisions set.
*
* @return the sweep revisions as a {@link RevisionVector}.
*/
@NotNull
RevisionVector getSweepRevisions() {
return new RevisionVector(transform(getLocalMap(SWEEP_REV).values(),
new Function<String, Revision>() {
@Override
public Revision apply(String s) {
return Revision.fromString(s);
}
}));
}
//-------------------------< UpdateOp modifiers >---------------------------
public static void setChildrenFlag(@NotNull UpdateOp op,
boolean hasChildNode) {
checkNotNull(op).set(CHILDREN_FLAG, hasChildNode);
}
public static void setModified(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).max(MODIFIED_IN_SECS, getModifiedInSecs(checkNotNull(revision).getTimestamp()));
}
public static void setRevision(@NotNull UpdateOp op,
@NotNull Revision revision,
@NotNull String commitValue) {
checkNotNull(op).setMapEntry(REVISIONS,
checkNotNull(revision), checkNotNull(commitValue));
}
public static void unsetRevision(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).unsetMapEntry(REVISIONS, checkNotNull(revision));
}
public static boolean isRevisionsEntry(String name) {
return REVISIONS.equals(name);
}
public static boolean isCommitRootEntry(String name) {
return COMMIT_ROOT.equals(name);
}
public static boolean isDeletedEntry(String name) {
return DELETED.equals(name);
}
public static boolean isLastRevEntry(String name) {
return LAST_REV.equals(name);
}
public static void removeRevision(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).removeMapEntry(REVISIONS, checkNotNull(revision));
}
/**
* Add a collision marker for the given {@code revision}.
*
* @param op the update operation.
* @param revision the commit for which a collision was detected.
* @param other the revision for the commit, which detected the collision.
*/
public static void addCollision(@NotNull UpdateOp op,
@NotNull Revision revision,
@NotNull Revision other) {
checkNotNull(op).setMapEntry(COLLISIONS, checkNotNull(revision),
other.toString());
}
public static void removeCollision(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).removeMapEntry(COLLISIONS, checkNotNull(revision));
}
public static void setLastRev(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).setMapEntry(LAST_REV,
new Revision(0, 0, revision.getClusterId()),
revision.toString());
}
public static void setCommitRoot(@NotNull UpdateOp op,
@NotNull Revision revision,
int commitRootDepth) {
checkNotNull(op).setMapEntry(COMMIT_ROOT, checkNotNull(revision),
String.valueOf(commitRootDepth));
}
public static void removeCommitRoot(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).removeMapEntry(COMMIT_ROOT, revision);
}
public static void unsetCommitRoot(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).unsetMapEntry(COMMIT_ROOT, revision);
}
public static void setDeleted(@NotNull UpdateOp op,
@NotNull Revision revision,
boolean deleted) {
if(deleted) {
//DELETED_ONCE would be set upon every delete.
//possibly we can avoid that
setDeletedOnce(op);
}
checkNotNull(op).setMapEntry(DELETED, checkNotNull(revision), String.valueOf(deleted));
}
public static void setDeletedOnce(@NotNull UpdateOp op) {
checkNotNull(op).set(DELETED_ONCE, Boolean.TRUE);
}
public static void removeDeleted(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).removeMapEntry(DELETED, revision);
}
public static void setPrevious(@NotNull UpdateOp op,
@NotNull Range range) {
checkNotNull(op).setMapEntry(PREVIOUS, checkNotNull(range).high,
range.getLowValue());
}
public static void removePrevious(@NotNull UpdateOp op,
@NotNull Range range) {
removePrevious(op, checkNotNull(range).high);
}
public static void removePrevious(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).removeMapEntry(PREVIOUS, checkNotNull(revision));
}
public static void setStalePrevious(@NotNull UpdateOp op,
@NotNull Revision revision,
int height) {
checkNotNull(op).setMapEntry(STALE_PREV,
checkNotNull(revision), String.valueOf(height));
}
public static void removeStalePrevious(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).removeMapEntry(STALE_PREV, checkNotNull(revision));
}
public static void setHasBinary(@NotNull UpdateOp op) {
checkNotNull(op).set(HAS_BINARY_FLAG, HAS_BINARY_VAL);
}
public static void setBranchCommit(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).setMapEntry(BRANCH_COMMITS,
revision, String.valueOf(true));
}
public static void removeBranchCommit(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).removeMapEntry(BRANCH_COMMITS, revision);
}
public static void setSweepRevision(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).setMapEntry(SWEEP_REV,
new Revision(0, 0, revision.getClusterId()),
revision.toString());
}
public static void hasLastRev(@NotNull UpdateOp op,
@NotNull Revision revision) {
checkNotNull(op).equals(LAST_REV,
new Revision(0, 0, revision.getClusterId()),
revision.toString());
}
//----------------------------< internal >----------------------------------
private void previousDocumentNotFound(String prevId, Revision rev) {
LOG.warn("Document with previous revisions not found: " + prevId);
// main document may be stale, evict it from the cache if it is
// older than one minute. We don't want to invalidate a document
// too frequently if the document structure is really broken.
Path path = getMainPath();
String id = Utils.getIdFromPath(path);
NodeDocument doc = store.getIfCached(NODES, id);
long now = Revision.getCurrentTimestamp();
while (doc != null
&& doc.getCreated() + TimeUnit.MINUTES.toMillis(1) < now) {
LOG.info("Invalidated cached document {}", id);
store.invalidateCache(NODES, id);
// also invalidate intermediate docs if there are any matching
Iterable<Range> ranges = doc.getPreviousRanges().values();
doc = null;
for (Range range : ranges) {
if (range.includes(rev)) {
id = Utils.getPreviousIdFor(path, range.high, range.height);
doc = store.getIfCached(NODES, id);
break;
}
}
}
}
private LastRevs createLastRevs(@NotNull RevisionVector readRevision,
@NotNull RevisionContext context,
@Nullable Branch branch,
@Nullable Revision pendingLastRev) {
LastRevs lastRevs = new LastRevs(getLastRev(), readRevision, branch);
// overlay with unsaved last modified from this instance
lastRevs.update(pendingLastRev);
// collect clusterIds
SortedSet<Revision> mostRecentChanges = Sets.newTreeSet(REVERSE);
mostRecentChanges.addAll(getLocalRevisions().keySet());
mostRecentChanges.addAll(getLocalCommitRoot().keySet());
Set<Integer> clusterIds = Sets.newHashSet();
for (Revision r : getLocalRevisions().keySet()) {
clusterIds.add(r.getClusterId());
}
for (Revision r : getLocalCommitRoot().keySet()) {
clusterIds.add(r.getClusterId());
}
for (Revision r : mostRecentChanges) {
if (!clusterIds.contains(r.getClusterId())) {
// already found most recent change from this cluster node
continue;
}
String commitValue = context.getCommitValue(r, this);
if (commitValue == null) {
continue;
}
// resolve revision
Revision commitRev = resolveCommitRevision(r, commitValue);
if (Utils.isCommitted(commitValue)) {
lastRevs.update(commitRev);
clusterIds.remove(r.getClusterId());
} else if (branch != null) {
Revision branchRev = commitRev.asBranchRevision();
if (branch.containsCommit(branchRev)) {
lastRevs.updateBranch(branchRev);
clusterIds.remove(r.getClusterId());
}
}
}
return lastRevs;
}
/**
* Returns {@code true} if the given {@code revision} is more recent or
* equal to the committed revision in {@code valueMap}. This method assumes
* the given {@code revision} is committed.
*
* @param valueMap the value map sorted most recent first.
* @param revision a committed revision.
* @param context the revision context.
* @return if {@code revision} is the most recent committed revision in the
* {@code valueMap}.
*/
private boolean isMostRecentCommitted(SortedMap<Revision, String> valueMap,
Revision revision,
RevisionContext context) {
if (valueMap.isEmpty()) {
return true;
}
// shortcut when revision is the first key
Revision first = valueMap.firstKey();
if (first.compareRevisionTimeThenClusterId(revision) <= 0) {
return true;
}
// need to check commit status
for (Revision r : valueMap.keySet()) {
String cv = context.getCommitValue(r, this);
if (Utils.isCommitted(cv)) {
Revision c = resolveCommitRevision(r, cv);
return c.compareRevisionTimeThenClusterId(revision) <= 0;
}
}
// no committed revision found in valueMap
return true;
}
/**
* Returns the commit root document for the given revision. This may either
* be this document or another one.
*
* @param rev a revision.
* @return the commit root or <code>null</code> if there is none.
*/
@Nullable
private NodeDocument getCommitRoot(@NotNull Revision rev) {
// check local revisions and commitRoot first
if (getLocalRevisions().containsKey(rev)) {
return this;
}
Path commitRootPath;
String depth = getLocalCommitRoot().get(rev);
if (depth != null) {
commitRootPath = getPathAtDepth(depth);
} else {
// fall back to complete check, including previous documents
if (containsRevision(rev)) {
return this;
}
commitRootPath = getCommitRootPath(rev);
if (commitRootPath == null) {
// may happen for a commit root document, which hasn't been
// updated with the commit revision yet
return null;
}
}
// get root of commit
return store.find(Collection.NODES, Utils.getIdFromPath(commitRootPath));
}
/**
* Returns the path at the given {@code depth} based on the path of this
* document.
*
* @param depth the depth as a string.
* @return the path.
* @throws NumberFormatException if {@code depth} cannot be parsed as an
* integer.
*/
@NotNull
private Path getPathAtDepth(@NotNull String depth) {
if (checkNotNull(depth).equals("0")) {
return Path.ROOT;
}
Path p = getPath();
return p.getAncestor(p.getDepth() - Integer.parseInt(depth));
}
/**
* Returns the commit root depth for the given revision. This method also
* takes previous documents into account.
*
* @param revision get the commit root depth for this revision.
* @return the depth or <code>null</code> if there is no commit root entry
* for the given revision on this document or previous documents.
*/
@Nullable
private String getCommitRootDepth(@NotNull Revision revision) {
// check local map first
Map<Revision, String> local = getLocalCommitRoot();
String depth = local.get(revision);
if (depth == null) {
// check previous
for (NodeDocument prev : getPreviousDocs(COMMIT_ROOT, revision)) {
depth = prev.getCommitRootDepth(revision);
if (depth != null) {
break;
}
}
}
return depth;
}
/**
* Returns <code>true</code> if the given revision
* {@link Utils#isCommitted(String)} in the revisions map (including
* revisions split off to previous documents) and is visible from the
* <code>readRevision</code>. This includes branch commits if the read
* revision is on the same branch and is equal or newer than the revision
* to check.
*
* @param revision the revision to check.
* @param commitValue the commit value of the revision to check.
* @param readRevision the read revision.
* @return <code>true</code> if the revision is visible, otherwise
* <code>false</code>.
*/
private boolean isVisible(@NotNull RevisionContext context,
@NotNull Revision revision,
@NotNull String commitValue,
@NotNull RevisionVector readRevision) {
if (Utils.isCommitted(commitValue)) {
Branch b = context.getBranches().getBranch(readRevision);
if (b == null) {
// readRevision is not from a branch commit, though it may
// still be a branch revision when it references the base
// of a new branch that has not yet been created. In that
// case the branch revision is equivalent with the trunk
// revision.
// resolve commit revision
revision = resolveCommitRevision(revision, commitValue);
// compare resolved revision as is
return !readRevision.isRevisionNewer(revision);
} else {
// read revision is on a branch and the change is committed
// get the base revision of the branch and check
// if change is visible from there
RevisionVector baseRev = b.getBase(readRevision.getBranchRevision());
revision = resolveCommitRevision(revision, commitValue);
return !baseRev.isRevisionNewer(revision);
}
} else {
// branch commit (not merged)
// read as RevisionVector, even though this should be
// a Revision only. See OAK-4840
RevisionVector branchCommit = RevisionVector.fromString(commitValue);
if (branchCommit.getBranchRevision().getClusterId() != context.getClusterId()) {
// this is an unmerged branch commit from another cluster node,
// hence never visible to us
return false;
} else {
// unmerged branch change with local clusterId
Branch b = context.getBranches().getBranch(readRevision);
if (b == null) {
// reading on trunk never sees changes on an unmerged branch
return false;
} else if (b.containsCommit(revision)) {
// read revision is on the same branch as the
// unmerged branch changes -> compare revisions as is
return !readRevision.isRevisionNewer(revision);
} else {
// read revision is on a different branch than the
// unmerged branch changes -> never visible
return false;
}
}
}
}
/**
* Returns the commit value for the given <code>revision</code>.
*
* @param revision a revision.
* @return the commit value or <code>null</code> if the revision is unknown.
*/
@Nullable
private String getCommitValue(Revision revision) {
String value = getLocalRevisions().get(revision);
if (value == null) {
// check previous
for (NodeDocument prev : getPreviousDocs(REVISIONS, revision)) {
value = prev.getCommitValue(revision);
if (value != null) {
break;
}
}
}
return value;
}
/**
* Get the latest property value smaller or equal the readRevision revision.
*
* @param valueMap the sorted revision-value map
* @param readRevision the maximum revision
* @param validRevisions map of revision to commit value considered valid
* against the given readRevision.
* @param lastRevs to keep track of the most recent modification.
* @return the latest value from the {@code readRevision} point of view.
*/
@Nullable
private Value getLatestValue(@NotNull RevisionContext context,
@NotNull Iterable<Map.Entry<Revision, String>> valueMap,
@NotNull RevisionVector readRevision,
@NotNull Map<Revision, String> validRevisions,
@NotNull LastRevs lastRevs) {
for (Map.Entry<Revision, String> entry : valueMap) {
Revision propRev = entry.getKey();
String commitValue = validRevisions.get(propRev);
if (commitValue == null) {
commitValue = context.getCommitValue(propRev, this);
}
if (commitValue == null) {
continue;
}
// resolve revision
Revision commitRev = resolveCommitRevision(propRev, commitValue);
if (Utils.isCommitted(commitValue)) {
lastRevs.update(commitRev);
} else {
// branch commit
lastRevs.updateBranch(commitRev.asBranchRevision());
}
if (isValidRevision(context, propRev, commitValue, readRevision, validRevisions)) {
return new Value(commitRev, entry.getValue());
}
}
return null;
}
@NotNull
public Path getPath() {
return Path.fromString(getPathString());
}
@NotNull
private String getPathString() {
String p = (String) get(PATH);
if (p != null) {
return p;
}
return Utils.getPathFromId(getId());
}
@NotNull
Map<Revision, String> getDeleted() {
return ValueMap.create(this, DELETED);
}
public String asString() {
JsopWriter json = new JsopBuilder();
toJson(json, data);
return json.toString();
}
@SuppressWarnings("unchecked")
private static void toJson(JsopWriter json, Map<?, Object> map) {
for (Entry<?, Object>e : map.entrySet()) {
json.key(e.getKey().toString());
Object value = e.getValue();
if (value == null) {
json.value(null);
} else if (value instanceof Boolean) {
json.value((Boolean) value);
} else if (value instanceof Long) {
json.value((Long) value);
} else if (value instanceof Integer) {
json.value((Integer) value);
} else if (value instanceof Map) {
json.object();
toJson(json, (Map<Object, Object>) value);
json.endObject();
} else if (value instanceof Revision) {
json.value(value.toString());
} else {
json.value((String) value);
}
}
}
public static NodeDocument fromString(DocumentStore store, String s) {
JsopTokenizer json = new JsopTokenizer(s);
NodeDocument doc = new NodeDocument(store);
while (true) {
if (json.matches(JsopReader.END)) {
break;
}
String k = json.readString();
json.read(':');
if (json.matches(JsopReader.END)) {
break;
}
doc.put(k, fromJson(json));
json.matches(',');
}
doc.seal();
return doc;
}
private static Object fromJson(JsopTokenizer json) {
switch (json.read()) {
case JsopReader.NULL:
return null;
case JsopReader.TRUE:
return true;
case JsopReader.FALSE:
return false;
case JsopReader.NUMBER:
return Long.parseLong(json.getToken());
case JsopReader.STRING:
return json.getToken();
case '{':
TreeMap<Revision, Object> map = new TreeMap<Revision, Object>(REVERSE);
while (true) {
if (json.matches('}')) {
break;
}
String k = json.readString();
json.read(':');
map.put(Revision.fromString(k), fromJson(json));
json.matches(',');
}
return map;
}
throw new IllegalArgumentException(json.readRawValue());
}
/**
* A property value / revision combination.
*/
private static final class Value {
final Revision revision;
/**
* The value of a property at the given revision. A {@code null} value
* indicates the property was removed.
*/
final String value;
Value(@NotNull Revision revision, @Nullable String value) {
this.revision = checkNotNull(revision);
this.value = value;
}
}
private static final class ValueComparator implements
Comparator<Entry<Revision, String>> {
static final Comparator<Entry<Revision, String>> INSTANCE = new ValueComparator();
static final Comparator<Entry<Revision, String>> REVERSE = Collections.reverseOrder(INSTANCE);
private static final Ordering<String> STRING_ORDERING = Ordering.natural().nullsFirst();
@Override
public int compare(Entry<Revision, String> o1,
Entry<Revision, String> o2) {
int cmp = StableRevisionComparator.INSTANCE.compare(o1.getKey(), o2.getKey());
if (cmp != 0) {
return cmp;
}
return STRING_ORDERING.compare(o1.getValue(), o2.getValue());
}
}
}