You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by jo...@apache.org on 2024/03/14 08:25:12 UTC
(jackrabbit-oak) 01/04: OAK-10535: Prototype code for Old Revisions Cleanup
This is an automated email from the ASF dual-hosted git repository.
joscorbe pushed a commit to branch DetailedGC/OAK-10199
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
commit cf4147c2daad35ea1340658554ca86e1653e040c
Author: Jose Cordero <co...@adobe.com>
AuthorDate: Fri Jun 30 18:58:40 2023 +0200
OAK-10535: Prototype code for Old Revisions Cleanup
---
.../jackrabbit/oak/console/GroovyConsole.groovy | 1 +
.../oak/console/commands/RevisionCommand.groovy | 290 +++++++++++
.../console/commands/RevisionCommand.properties | 21 +
.../jackrabbit/oak/run/RevisionsCommand.java | 125 +++++
oak-store-document/pom.xml | 6 +
.../document/NodeDocumentRevisionCleaner.java | 270 ++++++++++
.../document/NodeDocumentRevisionCleanerTest.java | 549 +++++++++++++++++++++
7 files changed, 1262 insertions(+)
diff --git a/oak-run/src/main/groovy/org/apache/jackrabbit/oak/console/GroovyConsole.groovy b/oak-run/src/main/groovy/org/apache/jackrabbit/oak/console/GroovyConsole.groovy
index 904ab15a60..6124cc8a16 100644
--- a/oak-run/src/main/groovy/org/apache/jackrabbit/oak/console/GroovyConsole.groovy
+++ b/oak-run/src/main/groovy/org/apache/jackrabbit/oak/console/GroovyConsole.groovy
@@ -124,6 +124,7 @@ class GroovyConsole {
new CheckpointCommand(shell),
new LsCommand(shell),
new PnCommand(shell),
+ new RevisionCommand(shell),
new RefreshCommand(shell),
new RetrieveCommand(shell),
new LuceneCommand(shell),
diff --git a/oak-run/src/main/groovy/org/apache/jackrabbit/oak/console/commands/RevisionCommand.groovy b/oak-run/src/main/groovy/org/apache/jackrabbit/oak/console/commands/RevisionCommand.groovy
new file mode 100644
index 0000000000..81ddc2af8d
--- /dev/null
+++ b/oak-run/src/main/groovy/org/apache/jackrabbit/oak/console/commands/RevisionCommand.groovy
@@ -0,0 +1,290 @@
+/*
+ * 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.console.commands
+
+
+import groovy.transform.CompileStatic
+import org.apache.jackrabbit.oak.console.ConsoleSession
+import org.apache.jackrabbit.oak.plugins.document.DocumentNodeStore
+import org.apache.jackrabbit.oak.plugins.document.DocumentStore
+import org.apache.jackrabbit.oak.plugins.document.NodeDocumentRevisionCleaner
+import org.apache.jackrabbit.oak.plugins.document.Revision
+import org.apache.jackrabbit.oak.spi.state.NodeStore
+import org.codehaus.groovy.tools.shell.CommandSupport
+import org.codehaus.groovy.tools.shell.Groovysh
+
+import java.util.stream.Collectors
+
+import static java.lang.Integer.getInteger
+
+@CompileStatic
+class RevisionCommand extends CommandSupport {
+ public static final String COMMAND_NAME = 'revision'
+
+ private static final int REVISION_CAP = getInteger("oak.revision.cap", 250)
+
+ private static final String ARGS_SUBCOMMAND_LIST = "list"
+ private static final String ARGS_SUBCOMMAND_INIT = "init"
+ private static final String ARGS_SUBCOMMAND_SHOW = "show"
+ private static final String ARGS_SUBCOMMAND_CLEANUP = "cleanup"
+
+ private static final String ARGS_NO_CAP_RESULTS = "--nocap"
+ private static final String ARGS_NO_CAP_RESULTS_SHORT = "-n"
+ private static final String ARGS_ORDER_DESC = "--desc"
+ private static final String ARGS_ORDER_DESC_SHORT = "-d"
+
+ private static final int ORDER_ASC = 0
+ private static final int ORDER_DESC = 1
+
+ private static int order = ORDER_ASC
+ private static boolean capResults = true
+ private static boolean showPropertiesUse = false
+
+ private NodeDocumentRevisionCleaner cleanupHelper;
+
+ RevisionCommand(Groovysh shell) {
+ super(shell, COMMAND_NAME, 'rev')
+ }
+
+ @Override
+ Object execute(List<String> args) {
+ if (args.isEmpty()) {
+ io.err.println("Missing subcommand: [<list>|<init>|<show>|<cleanup>]")
+ return null
+ }
+
+ parseArgs(args)
+ switch(args[0].toLowerCase()) {
+ case ARGS_SUBCOMMAND_LIST:
+ list()
+ break
+ case ARGS_SUBCOMMAND_INIT:
+ initialize()
+ break
+ case ARGS_SUBCOMMAND_SHOW:
+ show(args.contains("candidates"), args.contains("blocked"))
+ break
+ case ARGS_SUBCOMMAND_CLEANUP:
+ io.err.println("This command is read-only. To execute the cleanup, use 'oak-run revisions' command:")
+ io.err.println(" oak-run revisions {<jdbc-uri> | <mongodb-uri>} cleanup [options]")
+ io.err.println(" Options:")
+ io.err.println(" --path <path> Path to the node to cleanup. This is required.")
+ io.err.println(" --number <number> Number of revisions to cleanup")
+ io.err.println(" --clusterId <clusterId> ClusterId to cleanup")
+ io.err.println(" Example:")
+ io.err.println(" oak-run revisions {<jdbc-uri> | <mongodb-uri>} cleanup --path /content --number 10 --clusterId 1")
+ io.err.println()
+ break
+ default:
+ io.err.println("Unrecognized subcommand: " + args[0])
+ }
+
+ io.out.flush()
+ return null
+ }
+
+ /**
+ * List command.
+ * Prints the list of revisions for the working document.
+ */
+ void list() {
+ int count = 0
+ SortedMap<Revision, String> allRevisions = cleanupHelper.getAllRevisions();
+
+ for (Map.Entry<Revision, String> revisionEntry : allRevisions.entrySet()) {
+ Revision revision = revisionEntry.getKey()
+ String value = revisionEntry.getValue()
+ io.out.println(revision.toReadableString() + " " + value)
+ count++
+ if (capResults && count >= REVISION_CAP) {
+ io.out.printf("-- Reached revision cap of %d elements. This document has %d revisions --%n", REVISION_CAP, allRevisions.size())
+ break
+ }
+ }
+ }
+
+ /**
+ * Outputs the candidate and blocked revision sets. These sets are populated during a previous execution of
+ * the info() command.
+ * @param args
+ */
+ void show(boolean candidates, boolean blocked) {
+ if (candidates) {
+ SortedMap<Integer, TreeSet<Revision>> candidateRevisionsToClean = cleanupHelper.getCandidateRevisionsToClean()
+ if (candidateRevisionsToClean.isEmpty()) {
+ io.out.println("-- Candidates revision list is empty. Run 'rev init' before --")
+ }
+ for (Map.Entry<Integer, TreeSet<Revision>> clusterEntry : candidateRevisionsToClean) {
+ int count = 0
+ io.out.println("ClusterId " + clusterEntry.key)
+ for (Revision revision : order == ORDER_ASC ? clusterEntry.value : clusterEntry.value.descendingSet()) {
+ io.out.println(" " + revision.toReadableString())
+ count++
+ if (capResults && count >= REVISION_CAP) {
+ io.out.printf("-- Reached revision cap of %d elements. There are %d revisions in the list --%n", REVISION_CAP, clusterEntry.value.size())
+ break
+ }
+ }
+ }
+ }
+
+ if (blocked) {
+ SortedMap<Integer, TreeSet<Revision>> usedRevisionsToKeep = cleanupHelper.getBlockedRevisionsToKeep()
+ if (cleanupHelper.getBlockedRevisionsToKeep().isEmpty()) {
+ io.out.println("-- Blocked revision list is empty. Run 'rev init' before --")
+ }
+ for (Map.Entry<Integer, TreeSet<Revision>> clusterEntry : usedRevisionsToKeep) {
+ int count = 0
+ io.out.println("ClusterId " + clusterEntry.key)
+ for (Revision revision : order == ORDER_ASC ? clusterEntry.value : clusterEntry.value.descendingSet()) {
+ io.out.println(revision.toReadableString())
+ count++
+ if (capResults && count >= REVISION_CAP) {
+ io.out.printf("-- Reached revision cap of %d elements. There are %d revisions in the list --%n", REVISION_CAP, clusterEntry.value.size())
+ break
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Performs the cleanup of a certain number of revisions for the specified clusterId, starting from oldest.
+ * @param numberToCleanup
+ * @param clusterToCleanup
+ */
+ void cleanup(int numberToCleanup, int clusterToCleanup) {
+ io.out.println("This will delete the following revisions and the property values permanently:")
+ SortedMap<Integer, TreeSet<Revision>> candidateRevisionsToClean = cleanupHelper.getCandidateRevisionsToClean()
+ SortedMap<Revision, TreeSet<String>> propertiesModifiedByRevision = null;//cleanupHelper.getPropertiesModifiedByRevision()
+ TreeSet<Revision> revisions = candidateRevisionsToClean.get(clusterToCleanup)
+ if (revisions != null) {
+ int count = 0
+ for (Revision revision : revisions) {
+ io.out.println(revision.toReadableString() + " => " + propertiesModifiedByRevision.get(revision))
+ count++
+ if (count >= numberToCleanup) {
+ break
+ }
+ }
+ } else {
+ io.err.println("Invalid clusterId")
+ return
+ }
+
+ io.out.println("Are you sure to proceed? [y/N]")
+ int confirmation = io.in.read()
+ io.in.readLine()
+ if (confirmation == (int)('y' as char)) {
+ // Start the cleanup
+ int count = 0
+ for (Revision revision : candidateRevisionsToClean.get(clusterToCleanup)) {
+ // Remove the revision from _revisions and all the properties
+ /*UpdateOp update = new UpdateOp(workingDocument.path.toString(), false)
+ update.removeMapEntry("_revisions", revision)
+ for (String property : propertiesModifiedByRevision.get(revision)) {
+ update.removeMapEntry(property, revision)
+ }
+ io.out.println("Executing UpdateOp: " + update)
+ try {
+ documentStore.findAndUpdate(NODES, update)
+ } catch (DocumentStoreException ex) {
+ io.out.println("Operation failed: " + ex)
+ }*/
+
+ count++
+ if (count >= numberToCleanup) {
+ break
+ }
+ }
+ io.out.println("-- Executed " + count + " operations --")
+ }
+ }
+
+ void initialize() {
+ NodeStore nodeStore = getSession().getStore()
+ assert nodeStore instanceof DocumentNodeStore
+ DocumentStore documentStore = nodeStore.getDocumentStore()
+ DocumentNodeStore documentNodeStore = (DocumentNodeStore) nodeStore
+
+
+ //cleanupHelper = new NodeDocumentRevisionCleaner(documentStore, documentNodeStore, session.get)
+
+ //cleanupHelper.initializeCleanupProcess()
+
+ SortedMap<Revision, String> allRevisions = cleanupHelper.getAllRevisions()
+ TreeMap<Integer, Integer> revisionsByClusterId = allRevisions.keySet().groupBy { it.clusterId }
+ .collectEntries { clusterId, revisions ->
+ [clusterId, revisions.size()]
+ } as TreeMap<Integer, Integer>
+
+ /*io.out.println("=== Last Revision by clusterId ===")
+ for (Map.Entry<Integer, Revision> entry : cleanupHelper.getLastRev()) {
+ io.out.printf(" [%d] -> %s%n", entry.key, entry.value.toReadableString())
+ }
+ io.out.println()
+
+ io.out.println("=== Sweep Revision by clusterId ===")
+ for (Map.Entry<Integer, Revision> entry : cleanupHelper.getSweepRev()) {
+ io.out.printf(" [%d] -> %s%n", entry.key, entry.value.toReadableString())
+ }
+ io.out.println()*/
+
+ io.out.println("=== Total Revisions by clusterId ===")
+ for (Map.Entry<Integer, Integer> entry : revisionsByClusterId) {
+ io.out.printf(" [%d] -> %d revisions%n", entry.key, entry.value)
+ }
+ io.out.println()
+
+ io.out.println("=== Candidates to cleanup ===")
+ cleanupHelper.getCandidateRevisionsToClean().each { clusterId, revisions ->
+ int blocked = cleanupHelper.getBlockedRevisionsToKeep().get(clusterId)?.size() ?: 0
+ io.out.printf("ClusterId [%d] has %d Candidates and %d Blocked%n", clusterId, revisions.size(), blocked)
+ }
+
+ io.out.println("The lists are stored temporarily. You can print them using:")
+ io.out.println(" > rev show candidates")
+ io.out.println(" > rev show blocked")
+ io.out.println()
+ }
+
+ void parseArgs(List<String> args) {
+ // Defaults
+ order = ORDER_ASC
+ capResults = true
+ showPropertiesUse = false
+
+ for (String arg : args) {
+ switch(arg) {
+ case ARGS_NO_CAP_RESULTS:
+ case ARGS_NO_CAP_RESULTS_SHORT:
+ capResults = false
+ break
+ case ARGS_ORDER_DESC:
+ case ARGS_ORDER_DESC_SHORT:
+ order = ORDER_DESC
+ break
+ }
+ }
+ }
+
+ ConsoleSession getSession(){
+ return (ConsoleSession)variables.session
+ }
+}
diff --git a/oak-run/src/main/groovy/org/apache/jackrabbit/oak/console/commands/RevisionCommand.properties b/oak-run/src/main/groovy/org/apache/jackrabbit/oak/console/commands/RevisionCommand.properties
new file mode 100644
index 0000000000..5bc67c2261
--- /dev/null
+++ b/oak-run/src/main/groovy/org/apache/jackrabbit/oak/console/commands/RevisionCommand.properties
@@ -0,0 +1,21 @@
+#
+# 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.
+#
+command.description=Revision management.
+command.usage=[<list>|<show>|<init>|<cleanup>]
+command.help=Revision management.
diff --git a/oak-run/src/main/java/org/apache/jackrabbit/oak/run/RevisionsCommand.java b/oak-run/src/main/java/org/apache/jackrabbit/oak/run/RevisionsCommand.java
index 5c6aba2e7e..ac72509b26 100644
--- a/oak-run/src/main/java/org/apache/jackrabbit/oak/run/RevisionsCommand.java
+++ b/oak-run/src/main/java/org/apache/jackrabbit/oak/run/RevisionsCommand.java
@@ -23,6 +23,11 @@ import org.apache.jackrabbit.guava.common.io.Closer;
import java.io.IOException;
import java.util.List;
import java.util.Locale;
+import java.util.Map;
+import java.util.Scanner;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -42,10 +47,15 @@ import org.apache.jackrabbit.oak.plugins.document.ClusterNodeInfoDocument;
import org.apache.jackrabbit.oak.plugins.document.DocumentNodeStore;
import org.apache.jackrabbit.oak.plugins.document.DocumentNodeStoreBuilder;
import org.apache.jackrabbit.oak.plugins.document.DocumentStore;
+import org.apache.jackrabbit.oak.plugins.document.DocumentStoreException;
import org.apache.jackrabbit.oak.plugins.document.FormatVersion;
import org.apache.jackrabbit.oak.plugins.document.MissingLastRevSeeker;
+import org.apache.jackrabbit.oak.plugins.document.NodeDocument;
+import org.apache.jackrabbit.oak.plugins.document.NodeDocumentRevisionCleaner;
+import org.apache.jackrabbit.oak.plugins.document.Revision;
import org.apache.jackrabbit.oak.plugins.document.RevisionContextWrapper;
import org.apache.jackrabbit.oak.plugins.document.SweepHelper;
+import org.apache.jackrabbit.oak.plugins.document.UpdateOp;
import org.apache.jackrabbit.oak.plugins.document.VersionGCSupport;
import org.apache.jackrabbit.oak.plugins.document.VersionGarbageCollector.VersionGCInfo;
import org.apache.jackrabbit.oak.plugins.document.VersionGarbageCollector.VersionGCStats;
@@ -57,7 +67,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static java.lang.Boolean.TRUE;
+import static java.lang.Integer.getInteger;
+import static java.util.Objects.requireNonNull;
import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.jackrabbit.oak.plugins.document.Collection.NODES;
import static org.apache.jackrabbit.oak.plugins.document.DocumentNodeStoreHelper.createVersionGC;
import static org.apache.jackrabbit.oak.plugins.document.FormatVersion.versionOf;
import static org.apache.jackrabbit.oak.plugins.document.util.Utils.getRootDocument;
@@ -74,6 +87,8 @@ public class RevisionsCommand implements Command {
private static final Logger LOG = LoggerFactory.getLogger(RevisionsCommand.class);
+ private static final int REVISION_CAP = getInteger("oak.revision.cap", 250);
+
private static final String USAGE = Joiner.on(System.lineSeparator()).join(
"revisions {<jdbc-uri> | <mongodb-uri>} <sub-command> [options]",
"where sub-command is one of",
@@ -82,6 +97,7 @@ public class RevisionsCommand implements Command {
" collect perform garbage collection",
" reset clear all persisted metadata",
" sweep clean up uncommitted changes",
+ " cleanup clean up old/unused revisions",
" detailedGC perform detailed garbage collection i.e. remove unmerged branch commits, old revisions, deleted properties etc"
);
@@ -96,6 +112,7 @@ public class RevisionsCommand implements Command {
static final String CMD_COLLECT = "collect";
static final String CMD_RESET = "reset";
static final String CMD_SWEEP = "sweep";
+ static final String CMD_CLEANUP = "cleanup";
static final String CMD_DETAILED_GC = "detailedGC";
final OptionSpec<?> once;
@@ -107,6 +124,8 @@ public class RevisionsCommand implements Command {
final OptionSpec<?> detailedGC;
final OptionSpec<?> detailedGCOnly;
final OptionSpec<?> verbose;
+ final OptionSpec<String> path;
+ final OptionSpec<Integer> number;
final OptionSpec<Boolean> dryRun;
final OptionSpec<Boolean> embeddedVerification;
@@ -139,6 +158,11 @@ public class RevisionsCommand implements Command {
.accepts("detailedGCOnly", "apply only to detailedGC (reset only)");
verbose = parser
.accepts("verbose", "print INFO messages to the console");
+ path = parser
+ .accepts("path", "path to the document to be cleaned up").withRequiredArg();
+ number = parser
+ .accepts("number", "number of revisions to clean").withRequiredArg()
+ .ofType(Integer.class).defaultsTo(0);
}
public RevisionsOptions parse(String[] args) {
@@ -186,6 +210,14 @@ public class RevisionsCommand implements Command {
return options.has(continuous);
}
+ String getPath() {
+ return path.value(options);
+ }
+
+ int getNumber() {
+ return number.value(options);
+ }
+
boolean isDetailedGCEnabled() {
return options.has(detailedGC);
}
@@ -215,6 +247,8 @@ public class RevisionsCommand implements Command {
reset(options, closer);
} else if (RevisionsOptions.CMD_SWEEP.equals(subCmd)) {
sweep(options, closer);
+ } else if (RevisionsOptions.CMD_CLEANUP.equals(subCmd)) {
+ cleanup(options, closer);
} else if (RevisionsOptions.CMD_DETAILED_GC.equals(subCmd)) {
collect(options, closer, true);
} else {
@@ -460,6 +494,97 @@ public class RevisionsCommand implements Command {
SweepHelper.sweep(store, new RevisionContextWrapper(ns, clusterId), seeker);
}
+ private void cleanup(RevisionsOptions options, Closer closer) throws IOException {
+ String path = options.getPath();
+ if (path == null || path.isEmpty()) {
+ System.err.println("path option is required for " + RevisionsOptions.CMD_CLEANUP + " command");
+ return;
+ }
+
+ DocumentNodeStoreBuilder<?> builder = createDocumentMKBuilder(options, closer);
+ if (builder == null) {
+ System.err.println("revisions mode only available for DocumentNodeStore");
+ return;
+ }
+
+ DocumentStore documentStore = builder.getDocumentStore();
+ builder.setReadOnlyMode();
+ useMemoryBlobStore(builder);
+ DocumentNodeStore documentNodeStore = builder.build();
+ String id = org.apache.jackrabbit.oak.plugins.document.util.Utils.getIdFromPath(path);
+ NodeDocument workingDocument = documentStore.find(NODES, id);
+ NodeDocumentRevisionCleaner revisionCleaner = new NodeDocumentRevisionCleaner(documentStore, documentNodeStore, workingDocument);
+
+ final UpdateOp op = new UpdateOp(requireNonNull(id), false);
+ revisionCleaner.initializeCleanupProcess(op);
+
+ SortedMap<Revision, String> allRevisions = revisionCleaner.getAllRevisions();
+ SortedMap<Integer, Integer> revisionsByClusterId = new TreeMap<>();
+ for (Revision revision : allRevisions.keySet()) {
+ Integer cid = revision.getClusterId();
+ Integer count = revisionsByClusterId.get(cid);
+ if (count == null) {
+ count = 0;
+ }
+ revisionsByClusterId.put(cid, count + 1);
+ }
+
+ System.out.println();
+
+ int count = 0;
+ System.out.println("=== Total Revisions by clusterId ===");
+ for (Map.Entry<Integer, Integer> entry : revisionsByClusterId.entrySet()) {
+ System.out.printf(" [%d] -> %d revisions%n", entry.getKey(), entry.getValue());
+ count++;
+ if (count >= REVISION_CAP) {
+ System.out.printf(" ...%n");
+ break;
+ }
+ }
+
+ for (Map.Entry<Integer, TreeSet<Revision>> entry : revisionCleaner.getCandidateRevisionsToClean().entrySet()) {
+ Integer cid = entry.getKey();
+ TreeSet<Revision> revisions = entry.getValue();
+ int blocked = revisionCleaner.getBlockedRevisionsToKeep().get(cid) != null ? revisionCleaner.getBlockedRevisionsToKeep().get(cid).size() : 0;
+ System.out.printf("ClusterId [%d] has %d Candidates and %d Blocked%n", cid, revisions.size(), blocked);
+ }
+
+ System.out.println("=== Operations to be executed ===");
+ System.out.println("Total of operations: " + op.getChanges().size());
+ System.out.println(op.getChanges());
+
+ /*TreeSet<Revision> revisionsToClean = revisionCleaner.getCandidateRevisionsToClean().get(clusterToCleanup)
+ .stream().limit(numberToCleanup).collect(Collectors.toCollection(TreeSet::new));
+ if (revisionsToClean.isEmpty()) {
+ System.out.println("No revisions to clean");
+ return;
+ } else {
+ count = 0;
+ for (Revision revision : revisionsToClean) {
+ System.out.printf(" %s%n", revision.toReadableString());
+ count++;
+ if (count >= REVISION_CAP) {
+ System.out.print(" ...%n");
+ System.out.printf(" %s%n", revisionsToClean.last().toReadableString());
+ break;
+ }
+ }
+ }*/
+
+ Scanner scanner = new Scanner(System.in);
+ System.out.println("The revisions will be deleted permanently. Are you sure to proceed? [y/N]");
+ String confirmation = scanner.nextLine().trim().toLowerCase();
+ if (confirmation.equals("y") || confirmation.equals("yes")) {
+ try {
+ count++;
+ documentStore.findAndUpdate(NODES, op);
+ } catch (DocumentStoreException ex) {
+ System.out.println("Operation failed: " + ex);
+ }
+ System.out.println("-- Executed " + count + " operations --");
+ }
+ }
+
private String fmtTimestamp(long ts) {
return timestampToString(ts);
}
diff --git a/oak-store-document/pom.xml b/oak-store-document/pom.xml
index 96ecff9c9e..8ee078a8fd 100644
--- a/oak-store-document/pom.xml
+++ b/oak-store-document/pom.xml
@@ -349,5 +349,11 @@
<version>${testcontainers.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${jackson.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/NodeDocumentRevisionCleaner.java b/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/NodeDocumentRevisionCleaner.java
new file mode 100644
index 0000000000..b55db4afb6
--- /dev/null
+++ b/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/NodeDocumentRevisionCleaner.java
@@ -0,0 +1,270 @@
+/*
+ * 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 org.apache.jackrabbit.oak.plugins.document.util.Utils;
+
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.TreeSet;
+
+public class NodeDocumentRevisionCleaner {
+
+ private final DocumentStore documentStore;
+ private final DocumentNodeStore documentNodeStore;
+ private final NodeDocument workingDocument;
+ private final RevisionPropertiesClassifier revisionClassifier;
+ private final RevisionCleanerUtility revisionCleaner;
+
+ /**
+ * Constructor for NodeDocumentRevisionCleaner.
+ * @param documentStore The DocumentStore instance. Must be writable to perform cleanup.
+ * @param documentNodeStore The DocumentNodeStore instance.
+ * @param workingDocument The document to clean up.
+ */
+ public NodeDocumentRevisionCleaner(DocumentStore documentStore, DocumentNodeStore documentNodeStore,
+ NodeDocument workingDocument) {
+ this.workingDocument = workingDocument;
+ this.documentStore = documentStore;
+ this.documentNodeStore = documentNodeStore;
+
+ revisionClassifier = new RevisionPropertiesClassifier(workingDocument);
+ revisionCleaner = new RevisionCleanerUtility(revisionClassifier);
+ }
+
+ /**
+ * Collects cleanable old revisions for the given document.
+ */
+ public void collectOldRevisions(UpdateOp op) {
+ revisionClassifier.classifyRevisionsAndProperties();
+ revisionCleaner.preserveRevisionsNewerThanThreshold(24, ChronoUnit.HOURS);
+ revisionCleaner.preserveLastRevisionForEachProperty();
+ revisionCleaner.preserveRevisionsReferencedByCheckpoints();
+ revisionCleaner.removeCandidatesInList();
+
+ /*for (Map.Entry<Integer, TreeSet<Revision>> entry : revisionCleaner.candidateRevisionsToClean.entrySet()) {
+ for (Revision revision : entry.getValue()) {
+ System.out.println("Removing revision " + revision);
+ TreeSet<String> properties = revisionClassifier.propertiesModifiedByRevision.get(revision);
+ System.out.println("Properties modified by this revision: " + properties);
+ if (properties != null) {
+ for (String property : properties) {
+ op.removeMapEntry(property, revision);
+ }
+ }
+
+ op.removeMapEntry("_revisions", revision);
+ }
+ }*/
+ }
+
+ protected class RevisionPropertiesClassifier {
+ private final NodeDocument workingDocument;
+ private SortedMap<String, SortedMap<Integer, TreeSet<Revision>>> revisionsModifyingPropertyByCluster;
+ private SortedMap<String, TreeSet<Revision>> revisionsModifyingProperty;
+ private SortedMap<Revision, TreeSet<String>> propertiesModifiedByRevision;
+
+ private RevisionPropertiesClassifier(NodeDocument workingDocument) {
+ this.workingDocument = workingDocument;
+
+ this.revisionsModifyingPropertyByCluster = new TreeMap<>();
+ this.revisionsModifyingProperty = new TreeMap<>();
+ this.propertiesModifiedByRevision = new TreeMap<>(StableRevisionComparator.INSTANCE);
+ }
+
+ /**
+ * This method processes the revisions of the working document, creating maps to
+ * track the relationships between revisions and modified properties.
+ */
+ protected void classifyRevisionsAndProperties() {
+ SortedMap<Revision, String> deletedRevisions = workingDocument.getLocalDeleted();
+ // Always keep the first "_deleted" entry, as is when the document was created
+ if (!deletedRevisions.isEmpty()) {
+ Revision createdRevision = deletedRevisions.firstKey();
+ revisionCleaner.addBlockedRevisionToKeep(createdRevision);
+ }
+
+ SortedMap<Revision, String> documentRevisions = workingDocument.getLocalRevisions();
+ for (Map.Entry<Revision, String> revisionEntry : documentRevisions.entrySet()) {
+ Revision revision = revisionEntry.getKey();
+ String revisionValue = revisionEntry.getValue();
+
+ // Only check committed revisions
+ if (Utils.isCommitted(revisionValue)) {
+ // Candidate to clean up
+ revisionCleaner.addCandidateRevisionToClean(revision);
+ // Store properties usage
+ classifyPropertiesModifiedByRevision(revision);
+ }
+ }
+ }
+
+ private void classifyPropertiesModifiedByRevision(Revision revision) {
+ for (Map.Entry<String, Object> propertyEntry : workingDocument.entrySet()) {
+ if (Utils.isPropertyName(propertyEntry.getKey()) || NodeDocument.isDeletedEntry(propertyEntry.getKey())) {
+ Map<Revision, String> valueMap = (Map) propertyEntry.getValue();
+ if (valueMap.containsKey(revision)) {
+ propertiesModifiedByRevision.computeIfAbsent(revision, key ->
+ new TreeSet<>()).add(propertyEntry.getKey()
+ );
+
+ revisionsModifyingPropertyByCluster.computeIfAbsent(propertyEntry.getKey(), key ->
+ new TreeMap<>()
+ ).computeIfAbsent(revision.getClusterId(), key ->
+ new TreeSet<>(StableRevisionComparator.INSTANCE)
+ ).add(revision);
+
+ revisionsModifyingProperty.computeIfAbsent(propertyEntry.getKey(), key ->
+ new TreeSet<>(StableRevisionComparator.INSTANCE)
+ ).add(revision);
+ }
+ }
+ }
+ }
+ }
+
+ protected class RevisionCleanerUtility {
+
+ private final SortedMap<Integer, TreeSet<Revision>> blockedRevisionsToKeep;
+ private final SortedMap<Integer, TreeSet<Revision>> candidateRevisionsToClean;
+ private final RevisionPropertiesClassifier revisionClassifier;
+
+ private RevisionCleanerUtility(RevisionPropertiesClassifier revisionClassifier) {
+ this.revisionClassifier = revisionClassifier;
+ this.candidateRevisionsToClean = new TreeMap<>();
+ this.blockedRevisionsToKeep = new TreeMap<>();
+ }
+
+ protected void preserveLastRevisionForEachProperty() {
+ for (SortedMap<Integer, TreeSet<Revision>> revisionsByCluster : revisionClassifier.revisionsModifyingPropertyByCluster.values()) {
+ for (TreeSet<Revision> revisions : revisionsByCluster.values()) {
+ Revision lastRevision = revisions.last();
+ addBlockedRevisionToKeep(lastRevision);
+ }
+ }
+ }
+
+ protected void preserveRevisionsNewerThanThreshold(long amount, ChronoUnit unit) {
+ long thresholdToPreserve = Instant.now().minus(amount, unit).toEpochMilli();
+ for (TreeSet<Revision> revisionSet : candidateRevisionsToClean.values()) {
+ for (Revision revision : revisionSet) {
+ if (revision.getTimestamp() > thresholdToPreserve) {
+ addBlockedRevisionToKeep(revision);
+ }
+ }
+ }
+ }
+
+ protected void preserveRevisionsReferencedByCheckpoints() {
+ SortedMap<Revision, Checkpoints.Info> checkpoints = documentNodeStore.getCheckpoints().getCheckpoints();
+ checkpoints.forEach((revision, info) -> {
+ // For each checkpoint, keep the last revision that modified a property prior to checkpoint
+ revisionClassifier.revisionsModifyingProperty.forEach((propertyName, revisionsSet) -> {
+ // Traverse the revisionVector of the checkpoint and find the last revision that modified the property
+ info.getCheckpoint().forEach(revisionToFind -> {
+ // If the exact revision exists, keep it. If not, find the previous one that modified that property
+ if (revisionsSet.contains(revisionToFind)) {
+ addBlockedRevisionToKeep(revisionToFind);
+ } else {
+ Revision previousRevision = revisionsSet.descendingSet().ceiling(revisionToFind);
+ if (previousRevision != null) {
+ addBlockedRevisionToKeep(previousRevision);
+ }
+ }
+ });
+ });
+ });
+ }
+
+ /**
+ * Adds a revision to the list of candidates to delete.
+ * @param revision
+ */
+ private void addCandidateRevisionToClean(Revision revision) {
+ candidateRevisionsToClean.computeIfAbsent(revision.getClusterId(), key ->
+ new TreeSet<>(StableRevisionComparator.INSTANCE)
+ ).add(revision);
+ }
+
+ /**
+ * Adds a revision to the list of revisions to keep.
+ * @param revision
+ */
+ private void addBlockedRevisionToKeep(Revision revision) {
+ blockedRevisionsToKeep.computeIfAbsent(revision.getClusterId(), key ->
+ new TreeSet<>(StableRevisionComparator.INSTANCE)
+ ).add(revision);
+ }
+
+ protected void removeCandidatesInList() {
+ revisionCleaner.blockedRevisionsToKeep.forEach((key, value) -> {
+ if (revisionCleaner.candidateRevisionsToClean.containsKey(key)) {
+ revisionCleaner.candidateRevisionsToClean.get(key).removeAll(value);
+ }
+ });
+ }
+ }
+
+ public SortedMap<Revision, String> getAllRevisions() {
+ return workingDocument.getLocalRevisions();
+ }
+
+ protected SortedMap<String, SortedMap<Integer, TreeSet<Revision>>> getRevisionsModifyingPropertyByCluster() {
+ return revisionClassifier.revisionsModifyingPropertyByCluster;
+ }
+
+ protected SortedMap<String, TreeSet<Revision>> getRevisionsModifyingProperty() {
+ return revisionClassifier.revisionsModifyingProperty;
+ }
+
+ protected SortedMap<Revision, TreeSet<String>> getPropertiesModifiedByRevision() {
+ return revisionClassifier.propertiesModifiedByRevision;
+ }
+
+ public SortedMap<Integer, TreeSet<Revision>> getBlockedRevisionsToKeep() {
+ return revisionCleaner.blockedRevisionsToKeep;
+ }
+
+ public SortedMap<Integer, TreeSet<Revision>> getCandidateRevisionsToClean() {
+ return revisionCleaner.candidateRevisionsToClean;
+ }
+
+ protected void classifyRevisionsAndProperties() {
+ revisionClassifier.classifyRevisionsAndProperties();
+ }
+
+ protected void markLastRevisionForEachProperty() {
+ revisionCleaner.preserveLastRevisionForEachProperty();
+ }
+
+ protected void markRevisionsNewerThanThresholdToPreserve(long amount, ChronoUnit unit) {
+ revisionCleaner.preserveRevisionsNewerThanThreshold(amount, unit);
+ }
+
+ protected void markCheckpointRevisionsToPreserve() {
+ revisionCleaner.preserveRevisionsReferencedByCheckpoints();
+ }
+
+ protected void removeCandidatesInList() {
+ revisionCleaner.removeCandidatesInList();
+ }
+}
diff --git a/oak-store-document/src/test/java/org/apache/jackrabbit/oak/plugins/document/NodeDocumentRevisionCleanerTest.java b/oak-store-document/src/test/java/org/apache/jackrabbit/oak/plugins/document/NodeDocumentRevisionCleanerTest.java
new file mode 100644
index 0000000000..ab0880ea37
--- /dev/null
+++ b/oak-store-document/src/test/java/org/apache/jackrabbit/oak/plugins/document/NodeDocumentRevisionCleanerTest.java
@@ -0,0 +1,549 @@
+package org.apache.jackrabbit.oak.plugins.document;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.jackrabbit.oak.plugins.document.Collection.NODES;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class NodeDocumentRevisionCleanerTest {
+
+ @Mock
+ DocumentStore documentStore;
+
+ @Mock
+ DocumentNodeStore documentNodeStore;
+
+ @Mock
+ Checkpoints checkpoints;
+
+ NodeDocument workingDocument;
+
+ NodeDocumentRevisionCleaner nodeDocumentRevisionCleaner;
+
+ @Before
+ public void setUp() {
+ MockitoAnnotations.initMocks(this);
+ workingDocument = new NodeDocument(documentStore);
+
+ Mockito.when(documentStore.find(Mockito.eq(NODES), Mockito.anyString())).thenReturn(workingDocument);
+ Mockito.when(documentNodeStore.getBranches()).thenReturn(new UnmergedBranches());
+ nodeDocumentRevisionCleaner = new NodeDocumentRevisionCleaner(documentStore, documentNodeStore, workingDocument);
+ }
+
+ @Test
+ public void testMarkCheckpointRevisionsToPreserveOnePropertyOneCluster() throws IOException {
+ Revision revisionA = Revision.fromString("r100000000-0-1");
+ Revision revisionB = Revision.fromString("r105000000-0-1");
+ Revision revisionC = Revision.fromString("r110000000-0-1");
+ Revision revisionD = Revision.fromString("r115000000-0-1");
+ Revision revisionE = Revision.fromString("r120000000-0-1");
+ Revision revisionF = Revision.fromString("r125000000-0-1");
+
+ Revision checkpoint1 = Revision.fromString("r109000000-0-1");
+ Revision checkpoint2 = Revision.fromString("r119000000-0-1");
+
+ String jsonProperties = "{" +
+ "'prop1': {'" + revisionA + "': 'value1', '" + revisionB + "': 'value2', '" + revisionC + "': 'value3', '" + revisionD + "': 'value4', '" + revisionE + "': 'value5', '" + revisionF + "': 'value6'}, " +
+ "'_revisions': {'" + revisionA + "': 'c', '" + revisionB + "': 'c', '" + revisionC + "': 'c', '" + revisionC + "': 'c', '" + revisionD + "': 'c', '" + revisionE + "': 'c', '" + revisionF + "': 'c'}" +
+ "}";
+ String jsonCheckpoints = "{" +
+ "'" + checkpoint1 + "': {'expires':'200000000','rv':'r109000000-0-1'}," +
+ "'" + checkpoint2 + "': {'expires':'200000000','rv':'r119000000-0-1'}" +
+ "}";
+
+
+ prepareDocumentMock(jsonProperties);
+ prepareCheckpointsMock(jsonCheckpoints);
+
+ nodeDocumentRevisionCleaner.classifyRevisionsAndProperties();
+ nodeDocumentRevisionCleaner.markCheckpointRevisionsToPreserve();
+ nodeDocumentRevisionCleaner.removeCandidatesInList();
+
+ // The revisions blocked should be:
+ // - r105000000-0-1 (blocked by checkpoint r109000000-0-1)
+ // - r115000000-0-1 (blocked by checkpoint r119000000-0-1)
+ assertEquals(Set.of(revisionB, revisionD), nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(1));
+ assertEquals(Set.of(revisionA, revisionC, revisionE, revisionF), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(1));
+ }
+
+ @Test
+ public void testRecentRevisionsArePreserved() throws IOException {
+ StringBuilder jsonPropBuilder = new StringBuilder("'prop1': {");
+ StringBuilder jsonRevisionsBuilder = new StringBuilder("'_revisions': {");
+
+ Instant currentTime = Instant.now();
+ List<Revision> revisions = new ArrayList<>();
+ for (int i = 29; i >= 0; i--) {
+ long timestamp = currentTime.minus(i, ChronoUnit.HOURS).toEpochMilli();
+ Revision revision = new Revision(timestamp, 0, 1);
+ revisions.add(revision);
+
+ jsonPropBuilder.append("'").append(revision).append("': 'value").append(i).append("', ");
+ jsonRevisionsBuilder.append("'").append(revision).append("': 'c', ");
+ }
+
+ // 23 hours and 59 minutes ago -> Should be preserved
+ long timestamp = currentTime.minus(23, ChronoUnit.HOURS).minus(59, ChronoUnit.MINUTES).toEpochMilli();
+ Revision revision = new Revision(timestamp, 0, 1);
+ revisions.add(revision); // revisions[30]
+ jsonPropBuilder.append("'").append(revision).append("': 'value").append(30).append("', ");
+ jsonRevisionsBuilder.append("'").append(revision).append("': 'c', ");
+
+ // 24 hours and 1 minute ago -> Should be candidate to clean
+ timestamp = currentTime.minus(24, ChronoUnit.HOURS).minus(1, ChronoUnit.MINUTES).toEpochMilli();
+ revision = new Revision(timestamp, 0, 1);
+ revisions.add(revision); // revisions[31]
+ jsonPropBuilder.append("'").append(revision).append("': 'value").append(30).append("'");
+ jsonRevisionsBuilder.append("'").append(revision).append("': 'c'");
+
+ jsonPropBuilder.append("}");
+ jsonRevisionsBuilder.append("}");
+ StringBuilder jsonBuilder = new StringBuilder("{");
+ jsonBuilder.append(jsonPropBuilder).append(", ").append(jsonRevisionsBuilder).append("}");
+
+ // revisions.forEach(rev -> System.out.println(rev.toReadableString()));
+
+ prepareDocumentMock(jsonBuilder.toString());
+
+ nodeDocumentRevisionCleaner.classifyRevisionsAndProperties();
+ nodeDocumentRevisionCleaner.markRevisionsNewerThanThresholdToPreserve(24, ChronoUnit.HOURS);
+ nodeDocumentRevisionCleaner.removeCandidatesInList();
+
+ // The candidate revisions should be the 6 oldest ones (current time -29, -28, -27, -26, -25, -24)
+ // and the one created 24 hours and 1 minute ago
+ Set<Revision> expectedCandidateRevisions = Stream.concat(
+ revisions.subList(0, 6).stream(), Stream.of(revisions.get(31))
+ ).collect(Collectors.toSet());
+
+ // Blocked revisions are all the 24 revisions created in the last 24 hours, and the one created 23 hours and 59 minutes ago
+ Set<Revision> expectedBlockedRevisions = revisions.subList(6, 31).stream().collect(Collectors.toSet());
+
+ assertEquals(expectedCandidateRevisions, nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(1));
+ assertEquals(expectedBlockedRevisions, nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(1));
+ }
+
+ @Test
+ public void testInitializeCleanupProcessMultipleClusters() throws IOException {
+ List<Revision> revs = new ArrayList<>();
+
+ // Some initial revisions (0-2)
+ revs.add(Revision.fromString("r100000000-0-1"));
+ revs.add(Revision.fromString("r100000001-0-2"));
+ revs.add(Revision.fromString("r100000002-0-3"));
+
+ // Blocked by first checkpoint r109000000 (3-5)
+ revs.add(Revision.fromString("r106000003-0-1"));
+ revs.add(Revision.fromString("r107000004-0-2"));
+ revs.add(Revision.fromString("r108000005-0-3"));
+
+ // Some unblocked revisions in middle (6-10)
+ revs.add(Revision.fromString("r110000006-0-3"));
+ revs.add(Revision.fromString("r110000006-1-3"));
+ revs.add(Revision.fromString("r114000007-0-2"));
+ revs.add(Revision.fromString("r115000008-0-1"));
+ revs.add(Revision.fromString("r118000009-0-1"));
+
+ // Blocked by second checkpoint r118000000 (11-13)
+ revs.add(Revision.fromString("r118000009-1-1"));
+ revs.add(Revision.fromString("r118000010-0-2"));
+ revs.add(Revision.fromString("r118000011-0-3"));
+
+ // Some more unblocked revisions (14-19)
+ revs.add(Revision.fromString("r120000012-0-1"));
+ revs.add(Revision.fromString("r122000013-0-2"));
+ revs.add(Revision.fromString("r122000013-1-2"));
+ revs.add(Revision.fromString("r123000014-0-2"));
+ revs.add(Revision.fromString("r125000015-0-3"));
+ revs.add(Revision.fromString("r130000016-0-1"));
+
+ // Last revision (20-22)
+ revs.add(Revision.fromString("r130000016-1-1"));
+ revs.add(Revision.fromString("r130000017-0-2"));
+ revs.add(Revision.fromString("r130000018-0-3"));
+
+ // Checkpoint revisions
+ Revision checkpoint1 = Revision.fromString("r109000000-0-1");
+ Revision checkpoint2 = Revision.fromString("r119000000-0-1");
+
+ StringBuilder jsonPropBuilder = new StringBuilder("'prop1': {");
+ StringBuilder jsonRevisionsBuilder = new StringBuilder("'_revisions': {");
+
+ for (int i = 0; i < revs.size(); i++) {
+ jsonPropBuilder.append("'").append(revs.get(i)).append("': 'value").append(i).append("'");
+ jsonRevisionsBuilder.append("'").append(revs.get(i)).append("': 'c'");
+ if (i < revs.size() - 1) {
+ jsonPropBuilder.append(", ");
+ jsonRevisionsBuilder.append(", ");
+ }
+ }
+
+ jsonPropBuilder.append("}");
+ jsonRevisionsBuilder.append("}");
+ StringBuilder jsonBuilder = new StringBuilder("{");
+ jsonBuilder.append(jsonPropBuilder).append(", ").append(jsonRevisionsBuilder).append("}");
+
+ String jsonCheckpoints = "{" +
+ "'" + checkpoint1 + "': {'expires':'200000000','rv':'r106500003-0-1,r107500004-0-2,r108500005-0-3'}," +
+ "'" + checkpoint2 + "': {'expires':'200000000','rv':'r118000009-1-1,r118000010-0-2,r118000011-0-3'}" +
+ "}";
+
+ prepareDocumentMock(jsonBuilder.toString());
+ prepareCheckpointsMock(jsonCheckpoints);
+
+ nodeDocumentRevisionCleaner.collectOldRevisions(null);
+
+ // The revisions blocked should be:
+ // - r106000003-0-3, r118000004-0-2, r108000005-0-3 (referenced by checkpoint 1)
+ // - r118000009-1-1, r118000010-0-2, r118000011-0-3 (referenced by checkpoint 2)
+ // - r130000016-1-1, r130000017-0-2, r130000018-0-3 (last revision)
+ assertEquals(Set.of(revs.get(3), revs.get(11), revs.get(20)), nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(1));
+ assertEquals(Set.of(revs.get(4), revs.get(12), revs.get(21)), nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(2));
+ assertEquals(Set.of(revs.get(5), revs.get(13), revs.get(22)), nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(3));
+
+ // Rest of revisions are candidates to clean
+ assertEquals(Set.of(revs.get(0), revs.get(9), revs.get(10), revs.get(14), revs.get(19)), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(1));
+ assertEquals(Set.of(revs.get(1), revs.get(8), revs.get(15), revs.get(16), revs.get(17)), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(2));
+ assertEquals(Set.of(revs.get(2), revs.get(6), revs.get(7), revs.get(18)), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(3));
+
+ assertTrue(Collections.disjoint(nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(1), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(1)));
+ assertTrue(Collections.disjoint(nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(2), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(2)));
+ assertTrue(Collections.disjoint(nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(3), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(3)));
+ }
+
+ @Test
+ public void testCheckpointedRevisionFallback() throws IOException {
+ List<Revision> revs = new ArrayList<>();
+
+ revs.add(Revision.fromString("r100000000-0-1"));
+ revs.add(Revision.fromString("r100000001-0-2"));
+ revs.add(Revision.fromString("r100000002-0-3"));
+
+ revs.add(Revision.fromString("r101100003-0-1"));
+ revs.add(Revision.fromString("r101200004-0-2"));
+ revs.add(Revision.fromString("r101300005-0-3"));
+
+ revs.add(Revision.fromString("r102000006-0-1"));
+ revs.add(Revision.fromString("r102000007-0-3"));
+
+ revs.add(Revision.fromString("r103000008-0-1"));
+ revs.add(Revision.fromString("r103000009-0-2"));
+ revs.add(Revision.fromString("r103000010-0-3"));
+
+ // Checkpoint
+ Revision checkpoint1 = Revision.fromString("r102000000-0-1");
+
+ StringBuilder jsonPropBuilder = new StringBuilder("'prop1': {");
+ StringBuilder jsonRevisionsBuilder = new StringBuilder("'_revisions': {");
+
+ for (int i = 0; i < revs.size(); i++) {
+ jsonPropBuilder.append("'").append(revs.get(i)).append("': 'value").append(i).append("'");
+ jsonRevisionsBuilder.append("'").append(revs.get(i)).append("': 'c'");
+ if (i < revs.size() - 1) {
+ jsonPropBuilder.append(", ");
+ jsonRevisionsBuilder.append(", ");
+ }
+ }
+
+ jsonPropBuilder.append("}");
+ jsonRevisionsBuilder.append("}");
+ StringBuilder jsonBuilder = new StringBuilder("{");
+ jsonBuilder.append(jsonPropBuilder).append(", ").append(jsonRevisionsBuilder).append("}");
+
+ String jsonCheckpoints = "{" +
+ "'" + checkpoint1 + "': {'expires':'200000000','rv':'r102000007-0-1,r102000000-0-2,r102000008-0-3'}" +
+ "}";
+
+ prepareDocumentMock(jsonBuilder.toString());
+ prepareCheckpointsMock(jsonCheckpoints);
+
+ nodeDocumentRevisionCleaner.collectOldRevisions(null);
+
+ // The revisions blocked should be:
+ // - r103000008-0-1, r103000009-0-2, r103000010-0-3 (last revisions)
+ // - r102000006-0-1, r101300005-0-3, r102000007-0-3 (referenced by checkpoint for clusters 1, 2 and 3 respectively)
+ assertEquals(Set.of(revs.get(6), revs.get(8)), nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(1));
+ assertEquals(Set.of(revs.get(9)), nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(2));
+ assertEquals(Set.of(revs.get(5), revs.get(7), revs.get(10)), nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(3));
+
+ // Rest of revisions are candidates to clean
+ assertEquals(Set.of(revs.get(0), revs.get(3)), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(1));
+ assertEquals(Set.of(revs.get(1), revs.get(4)), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(2));
+ assertEquals(Set.of(revs.get(2)), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(3));
+ }
+
+ @Test
+ public void testLastRevisionIsBlocked() throws IOException {
+ Revision revisionA = new Revision(111111111L, 0, 1);
+ Revision revisionB = new Revision(222222222L, 0, 1);
+ Revision revisionC = new Revision(333333333L, 0, 1);
+
+ String jsonProperties = "{" +
+ "'prop1': {'" + revisionA + "': 'value1', '" + revisionB + "': 'value2', '" + revisionC + "': 'value3'}, " +
+ "'_revisions': {'" + revisionA + "': 'c', '" + revisionB + "': 'c', '" + revisionC + "': 'c'}" +
+ "}";
+ prepareDocumentMock(jsonProperties);
+
+ nodeDocumentRevisionCleaner.classifyRevisionsAndProperties();
+ nodeDocumentRevisionCleaner.markLastRevisionForEachProperty();
+
+ assertFalse(nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(1).contains(revisionA));
+ assertFalse(nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(1).contains(revisionB));
+ assertEquals(Set.of(revisionC), nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(1));
+ assertEquals(Set.of(revisionA, revisionB, revisionC), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(1));
+ }
+
+ @Test
+ public void testFirstDeletedRevisionIsBlocked() throws Exception {
+ Revision revisionA = new Revision(111111111L, 0, 1);
+ Revision revisionB = new Revision(222222222L, 0, 1);
+ Revision revisionC = new Revision(333333333L, 0, 1);
+
+ String jsonProperties = "{" +
+ "'_deleted': {'" + revisionA + "': 'false', '" + revisionB + "': 'true', '" + revisionC + "': 'false'}," +
+ "'_revisions': {'" + revisionA + "': 'c', '" + revisionB + "': 'c', '" + revisionC + "': 'c'}" +
+ "}";
+ prepareDocumentMock(jsonProperties);
+
+ nodeDocumentRevisionCleaner.classifyRevisionsAndProperties();
+
+ assertTrue(nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(1).contains(revisionA));
+ assertEquals(Set.of(revisionA, revisionB, revisionC), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(1));
+ }
+
+ @Test
+ public void testFirstDeletedRevisionIsBlockedUnordered() throws Exception {
+ Revision revisionA = new Revision(111111111L, 0, 1);
+ Revision revisionB = new Revision(222222222L, 0, 1);
+ Revision revisionC = new Revision(333333333L, 0, 1);
+
+ String jsonProperties = "{" +
+ "'_deleted': {'" + revisionC + "': 'false', '" + revisionA + "': 'true', '" + revisionB + "': 'false'}," +
+ "'_revisions': {'" + revisionC + "': 'c', '" + revisionB + "': 'c', '" + revisionA + "': 'c'}" +
+ "}";
+ prepareDocumentMock(jsonProperties);
+
+ nodeDocumentRevisionCleaner.classifyRevisionsAndProperties();
+
+ assertTrue(nodeDocumentRevisionCleaner.getBlockedRevisionsToKeep().get(1).contains(revisionA));
+ assertEquals(Set.of(revisionA, revisionB, revisionC), nodeDocumentRevisionCleaner.getCandidateRevisionsToClean().get(1));
+ }
+
+ @Test
+ public void testClassifyAndMapRevisionsAndPropertiesWithDeleted() throws IOException {
+ Revision revisionA = new Revision(111111111L, 0, 1);
+ Revision revisionB = new Revision(222222222L, 0, 1);
+ Revision revisionC = new Revision(333333333L, 0, 1);
+
+ String jsonProperties = "{" +
+ "'prop1': {'" + revisionA + "': 'value1', '" + revisionB + "': 'value2', '" + revisionC + "': 'value3'}, " +
+ "'prop2': {'" + revisionB + "': 'value4'}, " +
+ "'_deleted': {'" + revisionA + "': 'false'}," +
+ "'_revisions': {'" + revisionA + "': 'c', '" + revisionB + "': 'c', '" + revisionC + "': 'c'}" +
+ "}";
+ prepareDocumentMock(jsonProperties);
+
+ nodeDocumentRevisionCleaner.classifyRevisionsAndProperties();
+
+ SortedMap<Revision, TreeSet<String>> propertiesModifiedByRevision = nodeDocumentRevisionCleaner.getPropertiesModifiedByRevision();
+ assertEquals(Set.of("prop1", "_deleted"), propertiesModifiedByRevision.get(revisionA));
+ assertEquals(Set.of("prop1", "prop2"), propertiesModifiedByRevision.get(revisionB));
+ assertEquals(Set.of("prop1"), propertiesModifiedByRevision.get(revisionC));
+
+ SortedMap<String, SortedMap<Integer, TreeSet<Revision>>> revisionsModifyingProperty = nodeDocumentRevisionCleaner.getRevisionsModifyingPropertyByCluster();
+ assertEquals(Set.of(revisionA, revisionB, revisionC), revisionsModifyingProperty.get("prop1").get(1));
+ assertEquals(Set.of(revisionB), revisionsModifyingProperty.get("prop2").get(1));
+ assertEquals(Set.of(revisionA), revisionsModifyingProperty.get("_deleted").get(1));
+ }
+
+ @Test
+ public void testClassifyAndMapRevisionsMultipleDeleted() throws IOException {
+ Revision revisionA = new Revision(111111111L, 0, 1);
+ Revision revisionB = new Revision(222222222L, 0, 1);
+ Revision revisionC = new Revision(333333333L, 0, 1);
+
+ String jsonProperties = "{" +
+ "'_deleted': {'" + revisionA + "': 'false', '" + revisionB + "': 'true', '" + revisionC + "': 'false'}," +
+ "'_revisions': {'" + revisionA + "': 'c', '" + revisionB + "': 'c', '" + revisionC + "': 'c'}" +
+ "}";
+ prepareDocumentMock(jsonProperties);
+
+ nodeDocumentRevisionCleaner.classifyRevisionsAndProperties();
+
+ SortedMap<Revision, TreeSet<String>> propertiesModifiedByRevision = nodeDocumentRevisionCleaner.getPropertiesModifiedByRevision();
+ assertEquals(Set.of("_deleted"), propertiesModifiedByRevision.get(revisionA));
+ assertEquals(Set.of("_deleted"), propertiesModifiedByRevision.get(revisionB));
+ assertEquals(Set.of("_deleted"), propertiesModifiedByRevision.get(revisionC));
+
+ SortedMap<String, SortedMap<Integer, TreeSet<Revision>>> revisionsModifyingProperty = nodeDocumentRevisionCleaner.getRevisionsModifyingPropertyByCluster();
+ assertEquals(Set.of(revisionA, revisionB, revisionC), revisionsModifyingProperty.get("_deleted").get(1));
+ }
+
+ @Test
+ public void testClassifyAndMapRevisionsAndPropertiesWithoutDeleted() throws IOException {
+ Revision revisionA = new Revision(111111111L, 0, 1);
+ Revision revisionB = new Revision(222222222L, 0, 1);
+ Revision revisionC = new Revision(333333333L, 0, 1);
+
+ String jsonProperties = "{" +
+ "'prop1': {'" + revisionA + "': 'value1', '" + revisionB + "': 'value2', '" + revisionC + "': 'value3'}, " +
+ "'prop2': {'" + revisionB + "': 'value4'}, " +
+ "'_revisions': {'" + revisionA + "': 'c', '" + revisionB + "': 'c', '" + revisionC + "': 'c'}" +
+ "}";
+ prepareDocumentMock(jsonProperties);
+
+ nodeDocumentRevisionCleaner.classifyRevisionsAndProperties();
+
+ SortedMap<Revision, TreeSet<String>> propertiesModifiedByRevision = nodeDocumentRevisionCleaner.getPropertiesModifiedByRevision();
+ assertEquals(Set.of("prop1"), propertiesModifiedByRevision.get(revisionA));
+ assertEquals(Set.of("prop1", "prop2"), propertiesModifiedByRevision.get(revisionB));
+ assertEquals(Set.of("prop1"), propertiesModifiedByRevision.get(revisionC));
+
+ SortedMap<String, SortedMap<Integer, TreeSet<Revision>>> revisionsModifyingProperty = nodeDocumentRevisionCleaner.getRevisionsModifyingPropertyByCluster();
+ assertEquals(Set.of(revisionA, revisionB, revisionC), revisionsModifyingProperty.get("prop1").get(1));
+ assertEquals(Set.of(revisionB), revisionsModifyingProperty.get("prop2").get(1));
+ }
+
+ @Test
+ public void testClassifyAndMapRevisionsAndPropertiesNotCommitted() throws IOException {
+ Revision revisionA = new Revision(111111111L, 0, 1);
+ Revision revisionB = new Revision(222222222L, 0, 1);
+ Revision revisionC = new Revision(333333333L, 0, 1);
+ Revision revisionD = new Revision(444444444L, 0, 1);
+
+ String jsonProperties = "{" +
+ "'prop1': {'" + revisionA + "': 'value1', '" + revisionB + "': 'value2', '" + revisionC + "': 'value3', '" + revisionD + "': 'value5'}, " +
+ "'prop2': {'" + revisionB + "': 'value4', '" + revisionD + "': 'value5'}, " +
+ "'_revisions': {'" + revisionA + "': 'c', '" + revisionB + "': 'c', '" + revisionC + "': 'c', '" + revisionD + "': 'nc'}" +
+ "}";
+ prepareDocumentMock(jsonProperties);
+
+ nodeDocumentRevisionCleaner.classifyRevisionsAndProperties();
+
+ // Modifications done in revisionD should be ignored, as it is not committed
+ SortedMap<Revision, TreeSet<String>> propertiesModifiedByRevision = nodeDocumentRevisionCleaner.getPropertiesModifiedByRevision();
+ assertEquals(Set.of("prop1"), propertiesModifiedByRevision.get(revisionA));
+ assertEquals(Set.of("prop1", "prop2"), propertiesModifiedByRevision.get(revisionB));
+ assertEquals(Set.of("prop1"), propertiesModifiedByRevision.get(revisionC));
+ assertNull(propertiesModifiedByRevision.get(revisionD));
+
+ SortedMap<String, SortedMap<Integer, TreeSet<Revision>>> revisionsModifyingProperty = nodeDocumentRevisionCleaner.getRevisionsModifyingPropertyByCluster();
+ assertEquals(Set.of(revisionA, revisionB, revisionC), revisionsModifyingProperty.get("prop1").get(1));
+ assertEquals(Set.of(revisionB), revisionsModifyingProperty.get("prop2").get(1));
+ }
+
+ @Test
+ public void testClassifyAndMapRevisionsAndPropertiesDifferentClusters() throws IOException {
+ Revision revisionA = new Revision(111111111L, 0, 1);
+ Revision revisionB = new Revision(222222222L, 0, 2);
+ Revision revisionC = new Revision(333333333L, 0, 3);
+ Revision revisionD = new Revision(444444444L, 0, 1);
+
+ String jsonProperties = "{" +
+ "'prop1': {'" + revisionA + "': 'value1', '" + revisionB + "': 'value2', '" + revisionC + "': 'value3', '" + revisionD + "': 'value5'}, " +
+ "'prop2': {'" + revisionB + "': 'value4', '" + revisionD + "': 'value5'}, " +
+ "'_revisions': {'" + revisionA + "': 'c', '" + revisionB + "': 'c', '" + revisionC + "': 'c', '" + revisionD + "': 'c'}" +
+ "}";
+ prepareDocumentMock(jsonProperties);
+
+ nodeDocumentRevisionCleaner.classifyRevisionsAndProperties();
+
+ SortedMap<Revision, TreeSet<String>> propertiesModifiedByRevision = nodeDocumentRevisionCleaner.getPropertiesModifiedByRevision();
+ assertEquals(Set.of("prop1"), propertiesModifiedByRevision.get(revisionA));
+ assertEquals(Set.of("prop1", "prop2"), propertiesModifiedByRevision.get(revisionB));
+ assertEquals(Set.of("prop1"), propertiesModifiedByRevision.get(revisionC));
+ assertEquals(Set.of("prop1", "prop2"), propertiesModifiedByRevision.get(revisionD));
+
+ SortedMap<String, SortedMap<Integer, TreeSet<Revision>>> revisionsModifyingProperty = nodeDocumentRevisionCleaner.getRevisionsModifyingPropertyByCluster();
+ assertEquals(Set.of(revisionA, revisionD), revisionsModifyingProperty.get("prop1").get(1));
+ assertEquals(Set.of(revisionB), revisionsModifyingProperty.get("prop1").get(2));
+ assertEquals(Set.of(revisionC), revisionsModifyingProperty.get("prop1").get(3));
+ assertEquals(Set.of(revisionD), revisionsModifyingProperty.get("prop2").get(1));
+ assertEquals(Set.of(revisionB), revisionsModifyingProperty.get("prop2").get(2));
+ assertNull(revisionsModifyingProperty.get("prop2").get(3));
+ }
+
+ private void prepareDocumentMock(String jsonProperties) throws IOException {
+ String json = jsonProperties.replace("'", "\"");
+
+ ObjectMapper objectMapper = new ObjectMapper();
+ Map<String, Map<String, String>> data = objectMapper.readValue(json, new TypeReference<>() {});
+
+ SortedMap<String, Object> entries = new TreeMap<>();
+ SortedMap<Revision, String> allRevisions = new TreeMap<>(StableRevisionComparator.INSTANCE);
+ for (Map.Entry<String, Map<String, String>> entry : data.entrySet()) {
+ String property = entry.getKey();
+ Map<String, String> revisions = entry.getValue();
+
+ SortedMap<Revision, String> sortedRevisions = new TreeMap<>(StableRevisionComparator.INSTANCE);
+ for (Map.Entry<String, String> revisionEntry : revisions.entrySet()) {
+ String revisionStr = revisionEntry.getKey();
+ String value = revisionEntry.getValue();
+
+ String[] parts = revisionStr.split("-");
+ // The timestamp part of the revision string (first part) is parsed as hexadecimal (radix 16)
+ long timestamp = Long.parseLong(parts[0].substring(1), 16);
+ int counter = Integer.parseInt(parts[1]);
+ int clusterId = Integer.parseInt(parts[2]);
+
+ Revision revision = new Revision(timestamp, counter, clusterId);
+ sortedRevisions.put(revision, value);
+
+ // Add all revisions to the "_revisions" list
+ allRevisions.put(revision, value);
+ }
+ entries.put(property, sortedRevisions);
+ }
+
+ workingDocument.data = entries;
+ }
+
+ private void prepareCheckpointsMock(String jsonCheckpoints) throws IOException {
+ String json = jsonCheckpoints.replace("'", "\"");
+
+ ObjectMapper objectMapper = new ObjectMapper();
+ Map<String, Map<String, Object>> data = objectMapper.readValue(json, new TypeReference<>() {});
+
+ SortedMap<Revision, Checkpoints.Info> checkpoints = new TreeMap<>(StableRevisionComparator.REVERSE);
+ for (Map.Entry<String, Map<String, Object>> entry : data.entrySet()) {
+ String checkpointStr = entry.getKey();
+ Map<String, Object> checkpointData = entry.getValue();
+
+ String[] parts = checkpointStr.split("-");
+ long timestamp = Long.parseLong(parts[0].substring(1), 16);
+ int counter = Integer.parseInt(parts[1]);
+ int clusterId = Integer.parseInt(parts[2]);
+
+ Revision checkpoint = new Revision(timestamp, counter, clusterId);
+ String checkpointDataJson = objectMapper.writeValueAsString(checkpointData);
+
+ Checkpoints.Info info = Checkpoints.Info.fromString(checkpointDataJson);
+ checkpoints.put(checkpoint, info);
+ }
+
+ Mockito.when(documentNodeStore.getCheckpoints()).thenReturn(this.checkpoints);
+ Mockito.when(documentNodeStore.getCheckpoints().getCheckpoints()).thenReturn(checkpoints);
+ }
+}
\ No newline at end of file