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 th...@apache.org on 2024/09/23 15:57:21 UTC
(jackrabbit-oak) 01/01: OAK-11139 Allow downloading only recently changed nodes from MongoDB
This is an automated email from the ASF dual-hosted git repository.
thomasm pushed a commit to branch OAK-11139
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
commit ddf145f44d9babb64a419d54bc04365ad2e1c3b8
Author: Thomas Mueller <th...@apache.org>
AuthorDate: Mon Sep 23 17:56:56 2024 +0200
OAK-11139 Allow downloading only recently changed nodes from MongoDB
---
.../apache/jackrabbit/oak/index/IndexerSupport.java | 9 +++++++++
.../indexer/document/DocumentStoreIndexerBase.java | 1 +
.../document/flatfile/FlatFileNodeStoreBuilder.java | 9 ++++++++-
.../pipelined/PipelinedMongoDownloadTask.java | 20 ++++++++++++++------
.../pipelined/PipelinedTreeStoreStrategy.java | 10 ++++++++--
.../flatfile/pipelined/PipelinedTreeStoreIT.java | 1 +
6 files changed, 41 insertions(+), 9 deletions(-)
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/IndexerSupport.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/IndexerSupport.java
index 18173621f0..a588ca5029 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/IndexerSupport.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/IndexerSupport.java
@@ -76,12 +76,21 @@ public class IndexerSupport {
private File indexDefinitions;
private final String checkpoint;
private File existingDataDumpDir;
+ private long minModified;
public IndexerSupport(IndexHelper indexHelper, String checkpoint) {
this.indexHelper = indexHelper;
this.checkpoint = checkpoint;
}
+ public long getMinModified() {
+ return minModified;
+ }
+
+ public void setMinModified(long minModified) {
+ this.minModified = minModified;
+ }
+
public IndexerSupport withExistingDataDumpDir(File existingDataDumpDir) {
this.existingDataDumpDir = existingDataDumpDir;
return this;
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java
index 1a25acc7c1..1f298a806a 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java
@@ -202,6 +202,7 @@ public abstract class DocumentStoreIndexerBase implements Closeable {
.withNodeStateEntryTraverserFactory(new MongoNodeStateEntryTraverserFactory(rootDocumentState.getRootRevision(),
nodeStore, getMongoDocumentStore(), traversalLog))
.withCheckpoint(indexerSupport.getCheckpoint())
+ .withMinModified(indexerSupport.getMinModified())
.withStatisticsProvider(indexHelper.getStatisticsProvider())
.withIndexingReporter(reporter)
.withAheadOfTimeBlobDownloader(true);
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilder.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilder.java
index 38793e61a4..448970c9bd 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilder.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilder.java
@@ -110,6 +110,7 @@ public class FlatFileNodeStoreBuilder {
private MongoDatabase mongoDatabase = null;
private Set<IndexDefinition> indexDefinitions = null;
private String checkpoint;
+ private long minModified;
private StatisticsProvider statisticsProvider = StatisticsProvider.NOOP;
private IndexingReporter indexingReporter = IndexingReporter.NOOP;
private MongoClientURI mongoClientURI;
@@ -191,6 +192,11 @@ public class FlatFileNodeStoreBuilder {
return this;
}
+ public FlatFileNodeStoreBuilder withMinModified(long minModified) {
+ this.minModified = minModified;
+ return this;
+ }
+
public FlatFileNodeStoreBuilder withMongoClientURI(MongoClientURI mongoClientURI) {
this.mongoClientURI = mongoClientURI;
return this;
@@ -408,7 +414,7 @@ public class FlatFileNodeStoreBuilder {
indexingReporter.setIndexNames(indexNames);
return new PipelinedTreeStoreStrategy(mongoClientURI, mongoDocumentStore, nodeStore, rootRevision,
preferredPathElements, blobStore, dir, algorithm, pathPredicate, pathFilters, checkpoint,
- statisticsProvider, indexingReporter);
+ minModified, statisticsProvider, indexingReporter);
}
}
throw new IllegalStateException("Not a valid sort strategy value " + sortStrategyType);
@@ -435,4 +441,5 @@ public class FlatFileNodeStoreBuilder {
public File getFlatFileStoreDir() {
return flatFileStoreDir;
}
+
}
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTask.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTask.java
index 2c7c78cd75..284beff7a3 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTask.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTask.java
@@ -199,7 +199,6 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
private static final int MIN_INTERVAL_BETWEEN_DELAYED_ENQUEUING_MESSAGES = 10;
private static final BsonDocument NATURAL_HINT = BsonDocument.parse("{ $natural: 1 }");
private static final BsonDocument ID_INDEX_HINT = BsonDocument.parse("{ _id: 1 }");
- private static final Bson WITH_MODIFIED_FIELD = Filters.gte(NodeDocument.MODIFIED_IN_SECS, 0);
static final String THREAD_NAME_PREFIX = "mongo-dump";
@@ -228,6 +227,7 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
private final Stopwatch downloadStartWatch = Stopwatch.createUnstarted();
private final DownloadStageStatistics downloadStageStatistics = new DownloadStageStatistics();
private Instant lastDelayedEnqueueWarningMessageLoggedTimestamp = Instant.now();
+ private final long minModified;
public PipelinedMongoDownloadTask(MongoClientURI mongoClientURI,
MongoDocumentStore docStore,
@@ -236,8 +236,10 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
BlockingQueue<NodeDocument[]> queue,
List<PathFilter> pathFilters,
StatisticsProvider statisticsProvider,
- IndexingReporter reporter) {
- this(mongoClientURI, docStore, maxBatchSizeBytes, maxBatchNumberOfDocuments, queue, pathFilters, statisticsProvider, reporter, new ThreadFactoryBuilder().setDaemon(true).build());
+ IndexingReporter reporter,
+ ThreadFactory threadFactory) {
+ this(mongoClientURI, docStore, maxBatchSizeBytes, maxBatchNumberOfDocuments,
+ queue, pathFilters, statisticsProvider, reporter, threadFactory, 0);
}
public PipelinedMongoDownloadTask(MongoClientURI mongoClientURI,
@@ -248,7 +250,8 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
List<PathFilter> pathFilters,
StatisticsProvider statisticsProvider,
IndexingReporter reporter,
- ThreadFactory threadFactory) {
+ ThreadFactory threadFactory,
+ long minModified) {
this.mongoClientURI = mongoClientURI;
this.docStore = docStore;
this.statisticsProvider = statisticsProvider;
@@ -258,6 +261,7 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
this.mongoDocQueue = queue;
this.pathFilters = pathFilters;
this.threadFactory = threadFactory;
+ this.minModified = minModified;
// Default retries for 5 minutes.
this.retryDuringSeconds = ConfigHelper.getSystemPropertyAsInt(
@@ -328,6 +332,10 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
}
}
+ private Bson getModifiedFieldFilter() {
+ return Filters.gte(NodeDocument.MODIFIED_IN_SECS, minModified);
+ }
+
@Override
public Result call() throws Exception {
String originalName = Thread.currentThread().getName();
@@ -406,7 +414,7 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
if (mongoFilter == null) {
LOG.info("Downloading full repository from Mongo with natural order");
FindIterable<NodeDocument> mongoIterable = dbCollection
- .find(WITH_MODIFIED_FIELD) // Download only documents that have _modified set
+ .find(getModifiedFieldFilter()) // Download only documents that have _modified set
.hint(NATURAL_HINT);
DownloadTask downloadTask = new DownloadTask(DownloadOrder.UNDEFINED, downloadStageStatistics);
downloadTask.download(mongoIterable);
@@ -419,7 +427,7 @@ public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownlo
DownloadTask downloadTask = new DownloadTask(DownloadOrder.UNDEFINED, downloadStageStatistics);
LOG.info("Downloading from Mongo with natural order using filter: {}", mongoFilter);
FindIterable<NodeDocument> findIterable = dbCollection
- .find(Filters.and(WITH_MODIFIED_FIELD, mongoFilter))
+ .find(Filters.and(getModifiedFieldFilter(), mongoFilter))
.hint(NATURAL_HINT);
downloadTask.download(findIterable);
downloadTask.reportFinalResults();
diff --git a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedTreeStoreStrategy.java b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedTreeStoreStrategy.java
index b5cb037113..64114caabf 100644
--- a/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedTreeStoreStrategy.java
+++ b/oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedTreeStoreStrategy.java
@@ -123,6 +123,7 @@ public class PipelinedTreeStoreStrategy extends IndexStoreSortStrategyBase {
public static final String OAK_INDEXER_PIPELINED_TRANSFORM_THREADS = "oak.indexer.pipelined.transformThreads";
public static final int DEFAULT_OAK_INDEXER_PIPELINED_TRANSFORM_THREADS = 2;
public static final String OAK_INDEXER_PIPELINED_WORKING_MEMORY_MB = "oak.indexer.pipelined.workingMemoryMB";
+ public static final String OAK_INDEXER_PIPELINED_TREE_MIN_MODIFIED = "oak.indexer.pipelined.tree.minModified";
// 0 means autodetect
public static final int DEFAULT_OAK_INDEXER_PIPELINED_WORKING_MEMORY_MB = 0;
// Between 1 and 100
@@ -179,6 +180,7 @@ public class PipelinedTreeStoreStrategy extends IndexStoreSortStrategyBase {
private final int mongoDocBatchMaxNumberOfDocuments;
private final int nseBuffersCount;
private final int nseBuffersSizeBytes;
+ private final long minModified;
private long nodeStateEntriesExtracted;
@@ -201,9 +203,11 @@ public class PipelinedTreeStoreStrategy extends IndexStoreSortStrategyBase {
Predicate<String> pathPredicate,
List<PathFilter> pathFilters,
String checkpoint,
+ long minModified,
StatisticsProvider statisticsProvider,
IndexingReporter indexingReporter) {
super(storeDir, algorithm, pathPredicate, preferredPathElements, checkpoint);
+ this.minModified = minModified;
this.mongoClientURI = mongoClientURI;
this.docStore = documentStore;
this.documentNodeStore = documentNodeStore;
@@ -361,7 +365,7 @@ public class PipelinedTreeStoreStrategy extends IndexStoreSortStrategyBase {
emptyBatchesQueue.add(NodeStateEntryBatch.createNodeStateEntryBatch(nseBuffersSizeBytes, Integer.MAX_VALUE));
}
- INDEXING_PHASE_LOGGER.info("[TASK:PIPELINED-DUMP:START] Starting to build TreeStore");
+ INDEXING_PHASE_LOGGER.info("[TASK:PIPELINED-DUMP:START] Starting to build TreeStore with minModified {}", minModified);
Stopwatch start = Stopwatch.createStarted();
@SuppressWarnings("unchecked")
@@ -373,7 +377,9 @@ public class PipelinedTreeStoreStrategy extends IndexStoreSortStrategyBase {
mongoDocQueue,
pathFilters,
statisticsProvider,
- indexingReporter
+ indexingReporter,
+ new ThreadFactoryBuilder().setDaemon(true).build(),
+ minModified
));
ArrayList<Future<PipelinedTransformTask.Result>> transformFutures = new ArrayList<>(numberOfTransformThreads);
diff --git a/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedTreeStoreIT.java b/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedTreeStoreIT.java
index 3318cc77c2..7a37186b33 100644
--- a/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedTreeStoreIT.java
+++ b/oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedTreeStoreIT.java
@@ -675,6 +675,7 @@ public class PipelinedTreeStoreIT {
pathPredicate,
mongoRegexPathFilter,
null,
+ 0,
statsProvider,
indexingReporter);
}