thomasmueller commented on code in PR #979: URL: https://github.com/apache/jackrabbit-oak/pull/979#discussion_r1229663890
########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PathElementComparatorStringArray.java: ########## @@ -0,0 +1,70 @@ +/* + * 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.index.indexer.document.flatfile.pipelined; + +import org.apache.jackrabbit.guava.common.collect.ImmutableSet; + +import java.util.Comparator; +import java.util.Set; + +class PathElementComparatorStringArray implements Comparator<String[]> { Review Comment: I think we are missing a unit test here. ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelineMongoDownloadTask.java: ########## @@ -0,0 +1,278 @@ +/* + * 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.index.indexer.document.flatfile.pipelined; + +import com.mongodb.BasicDBObject; +import com.mongodb.MongoException; +import com.mongodb.ReadPreference; +import com.mongodb.client.FindIterable; +import com.mongodb.client.MongoCollection; +import com.mongodb.client.MongoCursor; +import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.index.indexer.document.LastModifiedRange; +import org.apache.jackrabbit.oak.plugins.document.Collection; +import org.apache.jackrabbit.oak.plugins.document.Document; +import org.apache.jackrabbit.oak.plugins.document.NodeDocument; +import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStore; +import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStoreHelper; +import org.apache.jackrabbit.oak.plugins.document.mongo.TraversingRange; +import org.apache.jackrabbit.oak.plugins.index.IndexUpdateCallback; +import org.apache.jackrabbit.oak.plugins.index.NodeTraversalCallback; +import org.apache.jackrabbit.oak.plugins.index.progress.IndexingProgressReporter; +import org.bson.BsonDocument; +import org.bson.BsonNull; +import org.bson.conversions.Bson; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.time.Instant; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.function.Predicate; + +import static com.mongodb.client.model.Sorts.ascending; +import static com.mongodb.client.model.Sorts.descending; + +class PipelineMongoDownloadTask implements Callable<PipelineMongoDownloadTask.Result> { + public static class Result { + private final long documentsDownloaded; + + public Result(long documentsDownloaded) { + this.documentsDownloaded = documentsDownloaded; + } + + public long getDocumentsDownloaded() { + return documentsDownloaded; + } + } + + public static final String OAK_INDEXER_PIPELINED_MONGO_CONNECTION_RETRY_SECONDS = "oak.indexer.pipelined.mongoConnectionRetrySeconds"; + public static final int DEFAULT_OAK_INDEXER_PIPELINED_MONGO_CONNECTION_RETRY_SECONDS = 300; + // Short retrial time, in most cases if the connection to a replica fails, trying again will establish a connection + // to another replica which is up + private final static long retryInitialIntervalMillis = 100; + private final static long retryMaxIntervalMillis = 10_000; + private static final Logger LOG = LoggerFactory.getLogger(PipelineMongoDownloadTask.class); + private static final Duration MONGO_QUEUE_OFFER_TIMEOUT = Duration.ofMinutes(2); + private static final String TRAVERSER_ID_PREFIX = "NSET"; + + private final MongoDocumentStore mongoStore; + private final Collection<? extends Document> collection; + private final Predicate<String> filter; + private final int batchSize; + private final ArrayBlockingQueue<BasicDBObject[]> mongoDocQueue; + private final int retryDuringSeconds; + private final Logger traversalLog = LoggerFactory.getLogger(PipelineMongoDownloadTask.class.getName() + ".traversal"); + private final MongoCollection<BasicDBObject> dbCollection; + private final ReadPreference readPreference; + + private long documentsRead = 0; + private long nextLastModified; + private String lastIdDownloaded = null; + + public <T extends Document> PipelineMongoDownloadTask(MongoDocumentStore mongoStore, + Collection<T> collection, + Predicate<String> filter, + int batchSize, + ArrayBlockingQueue<BasicDBObject[]> queue) { + this.mongoStore = mongoStore; + this.collection = collection; + this.batchSize = batchSize; + // TODO +// this.filter = filter; + this.mongoDocQueue = queue; + + IndexingProgressReporter progressReporterPerTask = + new IndexingProgressReporter(IndexUpdateCallback.NOOP, NodeTraversalCallback.NOOP); + progressReporterPerTask.setMessagePrefix("Dumping from " + TRAVERSER_ID_PREFIX); + + this.filter = (id) -> { + try { + progressReporterPerTask.traversedNode(() -> id); + } catch (CommitFailedException e) { + throw new RuntimeException(e); + } + traversalLog.trace(id); + return true; + }; + + // Default retries for 5 minutes. + this.retryDuringSeconds = ConfigHelper.getSystemPropertyAsInt( + OAK_INDEXER_PIPELINED_MONGO_CONNECTION_RETRY_SECONDS, + DEFAULT_OAK_INDEXER_PIPELINED_MONGO_CONNECTION_RETRY_SECONDS); + this.dbCollection = MongoDocumentStoreHelper.getDBCollection(mongoStore, collection); + + //TODO This may lead to reads being routed to secondary depending on MongoURI + //So caller must ensure that its safe to read from secondary +// this.readPreference = MongoDocumentStoreHelper.getConfiguredReadPreference(mongoStore, collection); Review Comment: This TODO should be resolved ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/SortKeyPathFactory.java: ########## @@ -0,0 +1,65 @@ +/* + * 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.index.indexer.document.flatfile.pipelined; + +import java.util.ArrayList; +import java.util.Set; + +import static org.apache.jackrabbit.oak.commons.PathUtils.elements; + +/** + * Creates String[] representing a path efficiently. It implements the following optimizations: + * <ul> + * <li>Reuse an internal array list to accumulate the parts of the path, before creating a String[] sized to this particular path. + * <li><a href="https://www.baeldung.com/string/intern";>Interns</a> the top level elements of the path. + * These are very likely to be repeated in most of the entries, so they are good candidates to be interned. + * </ul> + * + */ +public final class SortKeyPathFactory { + private final ArrayList<String> arrayBuilder = new ArrayList<>(16); + // Common words that appear in paths. + // TODO: confirm that checking for the common words is a worthwhile optimization. + private static final Set<String> commonWords = Set.of("content", "dam", "product-assets", Review Comment: We are missing a unit test ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedTransformTask.java: ########## @@ -0,0 +1,215 @@ +/* + * 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.index.indexer.document.flatfile.pipelined; + +import com.mongodb.BasicDBObject; +import org.apache.commons.io.FileUtils; +import org.apache.jackrabbit.guava.common.base.Stopwatch; +import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntry; +import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryWriter; +import org.apache.jackrabbit.oak.plugins.document.Collection; +import org.apache.jackrabbit.oak.plugins.document.DocumentNodeState; +import org.apache.jackrabbit.oak.plugins.document.DocumentNodeStore; +import org.apache.jackrabbit.oak.plugins.document.NodeDocument; +import org.apache.jackrabbit.oak.plugins.document.Path; +import org.apache.jackrabbit.oak.plugins.document.RevisionVector; +import org.apache.jackrabbit.oak.plugins.document.cache.NodeDocumentCache; +import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStore; +import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStoreHelper; +import org.apache.jackrabbit.oak.spi.state.NodeStateUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.ByteArrayOutputStream; +import java.io.OutputStreamWriter; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Predicate; + +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.pipelined.PipelinedStrategy.SENTINEL_MONGO_DOCUMENT; + +/** + * Receives batches of Mongo documents, converts them to node state entries, batches them in a {@link NodeStateEntryBatch} + * buffer and when the buffer is full, passes the buffer to the sort-and-save task. + */ +class PipelinedTransformTask implements Callable<PipelinedTransformTask.Result> { + + public static class Result { + private final long entryCount; + + public Result(long entryCount) { + this.entryCount = entryCount; + } + + public long getEntryCount() { + return entryCount; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(PipelinedTransformTask.class); + private static final AtomicInteger threadIdGenerator = new AtomicInteger(); + + private final MongoDocumentStore mongoStore; + private final DocumentNodeStore documentNodeStore; + private final RevisionVector rootRevision; + private final Predicate<String> pathPredicate; + private final ArrayBlockingQueue<BasicDBObject[]> mongoDocQueue; + private final ArrayBlockingQueue<NodeStateEntryBatch> emptyBatchesQueue; + private final ArrayBlockingQueue<NodeStateEntryBatch> nonEmptyBatchesQueue; + private final Collection<NodeDocument> collection; + private final NodeStateEntryWriter entryWriter; + private final int threadId = threadIdGenerator.getAndIncrement(); + + public PipelinedTransformTask(MongoDocumentStore mongoStore, + DocumentNodeStore documentNodeStore, + Collection<NodeDocument> collection, + RevisionVector rootRevision, + Predicate<String> pathPredicate, + NodeStateEntryWriter entryWriter, + ArrayBlockingQueue<BasicDBObject[]> mongoDocQueue, + ArrayBlockingQueue<NodeStateEntryBatch> emptyBatchesQueue, + ArrayBlockingQueue<NodeStateEntryBatch> nonEmptyBatchesQueue + ) { + this.mongoStore = mongoStore; + this.documentNodeStore = documentNodeStore; + this.collection = collection; + this.rootRevision = rootRevision; + this.pathPredicate = pathPredicate; + this.entryWriter = entryWriter; + this.mongoDocQueue = mongoDocQueue; + this.emptyBatchesQueue = emptyBatchesQueue; + this.nonEmptyBatchesQueue = nonEmptyBatchesQueue; + } + + @Override + public Result call() throws Exception { + String originalName = Thread.currentThread().getName(); + Thread.currentThread().setName("mongo-transform-" + threadId); + try { + LOG.info("Starting transform task"); + NodeDocumentCache nodeCache = MongoDocumentStoreHelper.getNodeDocumentCache(mongoStore); + Stopwatch w = Stopwatch.createStarted(); + long totalEntryCount = 0; + long mongoObjectsProcessed = 0; + LOG.info("Waiting for an empty buffer"); + NodeStateEntryBatch nseBatch = emptyBatchesQueue.take(); + ArrayList<SortKey> sortArray = nseBatch.getSortBuffer(); + ByteBuffer nseBuffer = nseBatch.getBuffer(); + + SortKeyPathFactory sortKeyPathFactory = new SortKeyPathFactory(); + // Used to serialize a node state entry before writing it to the buffer + ByteArrayOutputStream baos = new ByteArrayOutputStream(4096); + OutputStreamWriter writer = new OutputStreamWriter(baos, PipelinedStrategy.FLATFILESTORE_CHARSET); + LOG.info("Obtained an empty buffer. Starting to convert Mongo documents to node state entries"); + while (true) { + BasicDBObject[] dbObjectBatch = mongoDocQueue.take(); + if (dbObjectBatch == SENTINEL_MONGO_DOCUMENT) { + //Save the last batch + nseBatch.getBuffer().flip(); + nonEmptyBatchesQueue.put(nseBatch); + LOG.info("Thread terminating. Dumped {} nodestates in json format in {}", totalEntryCount, w); + return new Result(totalEntryCount); + } else { + // Transform object + for (BasicDBObject dbObject : dbObjectBatch) { + mongoObjectsProcessed++; + LOG.debug("Converting: {}", dbObject); + if (mongoObjectsProcessed % 10000 == 0) { + LOG.info("Mongo objects: {}, total entries: {}, current batch: {}, Size: {}/{} MB", + mongoObjectsProcessed, totalEntryCount, sortArray.size(), + nseBuffer.position() / FileUtils.ONE_MB, + nseBuffer.capacity() / FileUtils.ONE_MB + ); + } + //TODO Review the cache update approach where tracker has to track *all* docs + NodeDocument nodeDoc = MongoDocumentStoreHelper.convertFromDBObject(mongoStore, collection, dbObject); + // TODO: should we cache splitDocuments? Maybe this can be moved to after the check for split document + nodeCache.put(nodeDoc); + if (!nodeDoc.isSplitDocument()) { + // LOG.info("Mongo path: {}", nodeDoc.get(Document.ID)); + for (NodeStateEntry nse : getEntries(nodeDoc)) { + String path = nse.getPath(); + if (!NodeStateUtils.isHiddenPath(path) && pathPredicate.test(path)) { + // Serialize entry + entryWriter.writeTo(writer, nse); + writer.flush(); + byte[] entryData = baos.toByteArray(); + baos.reset(); + + if (nseBatch.isAtMaxEntries() || entryData.length + 4 > nseBuffer.remaining()) { + LOG.info("Buffer full, passing buffer to sort task. Total entries: {}, entries in buffer {}, buffer size: {}", + totalEntryCount, sortArray.size(), nseBuffer.position()); + nseBuffer.flip(); + Stopwatch putStart = Stopwatch.createStarted(); + nonEmptyBatchesQueue.put(nseBatch); + LOG.info("Added buffer to queue in {}", putStart); + // Get an empty buffer + nseBatch = emptyBatchesQueue.take(); + sortArray = nseBatch.getSortBuffer(); + nseBuffer = nseBatch.getBuffer(); + } + // Write entry to buffer + int bufferPos = nseBuffer.position(); + nseBuffer.putInt(entryData.length); + nseBuffer.put(entryData); + String[] key = sortKeyPathFactory.genSortKey(nse.getPath()); + sortArray.add(new SortKey(key, bufferPos)); + totalEntryCount++; + } + } + } + } + } + } + } catch (Throwable t) { + LOG.warn("Thread terminating with exception: " + t); Review Comment: Log the exception ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelineMongoDownloadTask.java: ########## @@ -0,0 +1,278 @@ +/* + * 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.index.indexer.document.flatfile.pipelined; + +import com.mongodb.BasicDBObject; +import com.mongodb.MongoException; +import com.mongodb.ReadPreference; +import com.mongodb.client.FindIterable; +import com.mongodb.client.MongoCollection; +import com.mongodb.client.MongoCursor; +import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.index.indexer.document.LastModifiedRange; +import org.apache.jackrabbit.oak.plugins.document.Collection; +import org.apache.jackrabbit.oak.plugins.document.Document; +import org.apache.jackrabbit.oak.plugins.document.NodeDocument; +import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStore; +import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStoreHelper; +import org.apache.jackrabbit.oak.plugins.document.mongo.TraversingRange; +import org.apache.jackrabbit.oak.plugins.index.IndexUpdateCallback; +import org.apache.jackrabbit.oak.plugins.index.NodeTraversalCallback; +import org.apache.jackrabbit.oak.plugins.index.progress.IndexingProgressReporter; +import org.bson.BsonDocument; +import org.bson.BsonNull; +import org.bson.conversions.Bson; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.time.Instant; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.function.Predicate; + +import static com.mongodb.client.model.Sorts.ascending; +import static com.mongodb.client.model.Sorts.descending; + +class PipelineMongoDownloadTask implements Callable<PipelineMongoDownloadTask.Result> { + public static class Result { + private final long documentsDownloaded; + + public Result(long documentsDownloaded) { + this.documentsDownloaded = documentsDownloaded; + } + + public long getDocumentsDownloaded() { + return documentsDownloaded; + } + } + + public static final String OAK_INDEXER_PIPELINED_MONGO_CONNECTION_RETRY_SECONDS = "oak.indexer.pipelined.mongoConnectionRetrySeconds"; + public static final int DEFAULT_OAK_INDEXER_PIPELINED_MONGO_CONNECTION_RETRY_SECONDS = 300; + // Short retrial time, in most cases if the connection to a replica fails, trying again will establish a connection + // to another replica which is up + private final static long retryInitialIntervalMillis = 100; + private final static long retryMaxIntervalMillis = 10_000; + private static final Logger LOG = LoggerFactory.getLogger(PipelineMongoDownloadTask.class); + private static final Duration MONGO_QUEUE_OFFER_TIMEOUT = Duration.ofMinutes(2); + private static final String TRAVERSER_ID_PREFIX = "NSET"; + + private final MongoDocumentStore mongoStore; + private final Collection<? extends Document> collection; + private final Predicate<String> filter; + private final int batchSize; + private final ArrayBlockingQueue<BasicDBObject[]> mongoDocQueue; + private final int retryDuringSeconds; + private final Logger traversalLog = LoggerFactory.getLogger(PipelineMongoDownloadTask.class.getName() + ".traversal"); + private final MongoCollection<BasicDBObject> dbCollection; + private final ReadPreference readPreference; + + private long documentsRead = 0; Review Comment: Initializing to 0 / null is not needed. And it is inconsistent. ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/SortKeyPathFactory.java: ########## @@ -0,0 +1,65 @@ +/* + * 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.index.indexer.document.flatfile.pipelined; + +import java.util.ArrayList; +import java.util.Set; + +import static org.apache.jackrabbit.oak.commons.PathUtils.elements; + +/** + * Creates String[] representing a path efficiently. It implements the following optimizations: + * <ul> + * <li>Reuse an internal array list to accumulate the parts of the path, before creating a String[] sized to this particular path. + * <li><a href="https://www.baeldung.com/string/intern";>Interns</a> the top level elements of the path. + * These are very likely to be repeated in most of the entries, so they are good candidates to be interned. + * </ul> + * + */ +public final class SortKeyPathFactory { + private final ArrayList<String> arrayBuilder = new ArrayList<>(16); + // Common words that appear in paths. + // TODO: confirm that checking for the common words is a worthwhile optimization. + private static final Set<String> commonWords = Set.of("content", "dam", "product-assets", + "jcr:content", "jcr:title", "jcr:lastModified", "jcr:created", "jcr:primaryType", "jcr:uuid" +// "cq:tags", "cq:lastModified", "cq:lastModifiedBy", "cq:template", "cq:templatePath", +// "dc:format", "dc:title", "dc:description", "dc:creator", "dc:modified", "dc:created", +// "dam:sha1", "dam:size", "dam:score", "dam:status", "dam:assetState", "dam:imported", +// "usages", "predictedTags", "imageFeatures", "contentFragment", "pageTitle", "renditions", +// "videoCodec", "audioCodec", "metadata", "original", "profile" + ); + + public String[] genSortKey(String path) { + arrayBuilder.clear(); + int i = 0; + for (String part : elements(path)) { + // This first levels of the path will very likely be similar for most of the entries (e.g. /content/dam/<company>) + // Interning these strings should provide a big reduction in memory usage. + // It is not worth to intern all levels because at lower levels the names are more likely to be less diverse, + // often even unique, so interning them would fill up the interned string hashtable with useless entries. + if (i < 3 || part.length() == 1 || commonWords.contains(part)) { + arrayBuilder.add(part.intern()); Review Comment: Sure we can save a bit of memory, but I think it is not worth the complexity. I think part.length() is not very common. ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelineMongoDownloadTask.java: ########## @@ -0,0 +1,278 @@ +/* + * 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.index.indexer.document.flatfile.pipelined; + +import com.mongodb.BasicDBObject; +import com.mongodb.MongoException; +import com.mongodb.ReadPreference; +import com.mongodb.client.FindIterable; +import com.mongodb.client.MongoCollection; +import com.mongodb.client.MongoCursor; +import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.index.indexer.document.LastModifiedRange; +import org.apache.jackrabbit.oak.plugins.document.Collection; +import org.apache.jackrabbit.oak.plugins.document.Document; +import org.apache.jackrabbit.oak.plugins.document.NodeDocument; +import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStore; +import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStoreHelper; +import org.apache.jackrabbit.oak.plugins.document.mongo.TraversingRange; +import org.apache.jackrabbit.oak.plugins.index.IndexUpdateCallback; +import org.apache.jackrabbit.oak.plugins.index.NodeTraversalCallback; +import org.apache.jackrabbit.oak.plugins.index.progress.IndexingProgressReporter; +import org.bson.BsonDocument; +import org.bson.BsonNull; +import org.bson.conversions.Bson; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.time.Instant; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.function.Predicate; + +import static com.mongodb.client.model.Sorts.ascending; +import static com.mongodb.client.model.Sorts.descending; + +class PipelineMongoDownloadTask implements Callable<PipelineMongoDownloadTask.Result> { + public static class Result { + private final long documentsDownloaded; + + public Result(long documentsDownloaded) { + this.documentsDownloaded = documentsDownloaded; + } + + public long getDocumentsDownloaded() { + return documentsDownloaded; + } + } + + public static final String OAK_INDEXER_PIPELINED_MONGO_CONNECTION_RETRY_SECONDS = "oak.indexer.pipelined.mongoConnectionRetrySeconds"; + public static final int DEFAULT_OAK_INDEXER_PIPELINED_MONGO_CONNECTION_RETRY_SECONDS = 300; + // Short retrial time, in most cases if the connection to a replica fails, trying again will establish a connection + // to another replica which is up + private final static long retryInitialIntervalMillis = 100; + private final static long retryMaxIntervalMillis = 10_000; + private static final Logger LOG = LoggerFactory.getLogger(PipelineMongoDownloadTask.class); + private static final Duration MONGO_QUEUE_OFFER_TIMEOUT = Duration.ofMinutes(2); + private static final String TRAVERSER_ID_PREFIX = "NSET"; + + private final MongoDocumentStore mongoStore; + private final Collection<? extends Document> collection; + private final Predicate<String> filter; + private final int batchSize; + private final ArrayBlockingQueue<BasicDBObject[]> mongoDocQueue; + private final int retryDuringSeconds; + private final Logger traversalLog = LoggerFactory.getLogger(PipelineMongoDownloadTask.class.getName() + ".traversal"); + private final MongoCollection<BasicDBObject> dbCollection; + private final ReadPreference readPreference; + + private long documentsRead = 0; + private long nextLastModified; + private String lastIdDownloaded = null; + + public <T extends Document> PipelineMongoDownloadTask(MongoDocumentStore mongoStore, + Collection<T> collection, + Predicate<String> filter, + int batchSize, + ArrayBlockingQueue<BasicDBObject[]> queue) { + this.mongoStore = mongoStore; + this.collection = collection; + this.batchSize = batchSize; + // TODO +// this.filter = filter; Review Comment: This TODO should be resolved ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedSortBatchTask.java: ########## @@ -0,0 +1,151 @@ +/* + * 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.index.indexer.document.flatfile.pipelined; + +import org.apache.commons.io.FileUtils; +import org.apache.jackrabbit.guava.common.base.Stopwatch; +import org.apache.jackrabbit.oak.commons.Compression; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedOutputStream; +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.Callable; + +import static org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCount; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.createOutputStream; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.pipelined.PipelinedStrategy.SENTINEL_NSE_BUFFER; + +/** + * Receives batches of node state entries, sorts then in memory, and finally writes them to a file. + */ +class PipelinedSortBatchTask implements Callable<PipelinedSortBatchTask.Result> { + public static class Result { + private final long totalEntries; + + public Result(long totalEntries) { + this.totalEntries = totalEntries; + } + + public long getTotalEntries() { + return totalEntries; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(PipelinedSortBatchTask.class); + + private final Comparator<SortKey> pathComparator; + private final Compression algorithm; + private final ArrayBlockingQueue<NodeStateEntryBatch> emptyBuffersQueue; + private final ArrayBlockingQueue<NodeStateEntryBatch> nonEmptyBuffersQueue; + private final ArrayBlockingQueue<File> sortedFilesQueue; + private final File sortWorkDir; + private final byte[] copyBuffer = new byte[4096]; + private long entriesProcessed = 0; + + public PipelinedSortBatchTask(File storeDir, + PathElementComparatorStringArray pathComparator, + Compression algorithm, + ArrayBlockingQueue<NodeStateEntryBatch> emptyBuffersQueue, + ArrayBlockingQueue<NodeStateEntryBatch> nonEmptyBuffersQueue, + ArrayBlockingQueue<File> sortedFilesQueue) throws IOException { + this.pathComparator = (e1, e2) -> pathComparator.compare(e1.getPathElements(), e2.getPathElements()); + this.algorithm = algorithm; + this.emptyBuffersQueue = emptyBuffersQueue; + this.nonEmptyBuffersQueue = nonEmptyBuffersQueue; + this.sortedFilesQueue = sortedFilesQueue; + sortWorkDir = createdSortWorkDir(storeDir); + } + + @Override + public Result call() throws Exception { + String originalName = Thread.currentThread().getName(); + Thread.currentThread().setName("mongo-sort"); + try { + LOG.info("Starting sort and save task"); + while (true) { + LOG.info("Waiting for next batch"); + NodeStateEntryBatch nseBuffer = nonEmptyBuffersQueue.take(); + if (nseBuffer == SENTINEL_NSE_BUFFER) { + LOG.info("Terminating thread, processed {} entries", entriesProcessed); + return new Result(entriesProcessed); + } + sortAndSaveBatch(nseBuffer); + nseBuffer.reset(); + emptyBuffersQueue.put(nseBuffer); + } + } catch (Throwable t) { + LOG.warn("Thread terminating with exception: " + t); Review Comment: Log the exception ", t" ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/SortKeyPathFactory.java: ########## @@ -0,0 +1,65 @@ +/* + * 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.index.indexer.document.flatfile.pipelined; + +import java.util.ArrayList; +import java.util.Set; + +import static org.apache.jackrabbit.oak.commons.PathUtils.elements; + +/** + * Creates String[] representing a path efficiently. It implements the following optimizations: + * <ul> + * <li>Reuse an internal array list to accumulate the parts of the path, before creating a String[] sized to this particular path. + * <li><a href="https://www.baeldung.com/string/intern";>Interns</a> the top level elements of the path. + * These are very likely to be repeated in most of the entries, so they are good candidates to be interned. + * </ul> + * + */ +public final class SortKeyPathFactory { + private final ArrayList<String> arrayBuilder = new ArrayList<>(16); + // Common words that appear in paths. + // TODO: confirm that checking for the common words is a worthwhile optimization. + private static final Set<String> commonWords = Set.of("content", "dam", "product-assets", + "jcr:content", "jcr:title", "jcr:lastModified", "jcr:created", "jcr:primaryType", "jcr:uuid" +// "cq:tags", "cq:lastModified", "cq:lastModifiedBy", "cq:template", "cq:templatePath", +// "dc:format", "dc:title", "dc:description", "dc:creator", "dc:modified", "dc:created", +// "dam:sha1", "dam:size", "dam:score", "dam:status", "dam:assetState", "dam:imported", +// "usages", "predictedTags", "imageFeatures", "contentFragment", "pageTitle", "renditions", +// "videoCodec", "audioCodec", "metadata", "original", "profile" + ); + + public String[] genSortKey(String path) { Review Comment: generateSortKey? ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PathElementComparatorStringArray.java: ########## @@ -0,0 +1,70 @@ +/* + * 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.index.indexer.document.flatfile.pipelined; + +import org.apache.jackrabbit.guava.common.collect.ImmutableSet; + +import java.util.Comparator; +import java.util.Set; + +class PathElementComparatorStringArray implements Comparator<String[]> { + private final Set<String> preferred; + + public PathElementComparatorStringArray(Set<String> preferredPathElements) { + this.preferred = ImmutableSet.copyOf(preferredPathElements); + } + + @Override + public int compare(String[] p1, String[] p2) { + + int i1 = 0; + int i2 = 0; + + //Shorter paths come first i.e. first parent then children + //Also Rank jcr:content higher i.e. first child + while (i1 < p1.length || i2 < p2.length) { + if (i1 >= p1.length) { + return -1; + } + if (i2 >= p2.length) { + return 1; + } + + String pe1 = p1[i1]; i1++; Review Comment: I would use String pe2 = p2[i2++]; as it is less prone to typos -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
