fabriziofortino commented on code in PR #979: URL: https://github.com/apache/jackrabbit-oak/pull/979#discussion_r1244991492
########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/NodeStateHolder.java: ########## @@ -0,0 +1,54 @@ +/* + * 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.oak.commons.PathUtils; + +import java.util.Iterator; + +import static org.apache.jackrabbit.oak.commons.PathUtils.elements; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryWriter.getPath; + +final class NodeStateHolder { + + private final String line; + private final String[] pathElements; + + public NodeStateHolder(String line) { + this.line = line; + String path = getPath(line); + int depth = PathUtils.getDepth(path); Review Comment: minor: for consistency you might want to statically import `getDepth`. ########## oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedIT.java: ########## @@ -0,0 +1,149 @@ +/* + * 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.lang3.tuple.ImmutablePair; +import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.commons.Compression; +import org.apache.jackrabbit.oak.plugins.document.DocumentMK; +import org.apache.jackrabbit.oak.plugins.document.DocumentMKBuilderProvider; +import org.apache.jackrabbit.oak.plugins.document.DocumentNodeStore; +import org.apache.jackrabbit.oak.plugins.document.MongoConnectionFactory; +import org.apache.jackrabbit.oak.plugins.document.MongoUtils; +import org.apache.jackrabbit.oak.plugins.document.RevisionVector; +import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStore; +import org.apache.jackrabbit.oak.plugins.document.util.MongoConnection; +import org.apache.jackrabbit.oak.spi.blob.MemoryBlobStore; +import org.apache.jackrabbit.oak.spi.commit.CommitInfo; +import org.apache.jackrabbit.oak.spi.commit.EmptyHook; +import org.apache.jackrabbit.oak.spi.state.NodeBuilder; +import org.apache.jackrabbit.oak.spi.state.NodeStore; +import org.jetbrains.annotations.NotNull; +import org.junit.After; +import org.junit.Assume; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; +import org.junit.contrib.java.lang.system.RestoreSystemProperties; +import org.junit.rules.TemporaryFolder; +import org.junit.rules.TestRule; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.util.Set; +import java.util.function.Predicate; + +import static org.junit.Assert.assertEquals; + +public class PipelinedIT { Review Comment: can we have a test where the `pathPredicate` does not match any node in the repository? In the past, this has caused issues because the files where empty. ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMongoDownloadTask.java: ########## @@ -0,0 +1,304 @@ +/* + * 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.MongoIncompatibleDriverException; +import com.mongodb.MongoInterruptedException; +import com.mongodb.ReadPreference; +import com.mongodb.client.FindIterable; +import com.mongodb.client.MongoCollection; +import com.mongodb.client.MongoCursor; +import org.apache.jackrabbit.guava.common.base.Preconditions; +import org.apache.jackrabbit.guava.common.base.Stopwatch; +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.bson.BsonDocument; +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.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static com.mongodb.client.model.Sorts.ascending; + +public class PipelinedMongoDownloadTask implements Callable<PipelinedMongoDownloadTask.Result> { + public static class Result { + private final long documentsDownloaded; + + public Result(long documentsDownloaded) { + this.documentsDownloaded = documentsDownloaded; + } + + public long getDocumentsDownloaded() { + return documentsDownloaded; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(PipelinedMongoDownloadTask.class); + + /** + * Whether to retry on connection errors to MongoDB. + * This property affects the query that is used to download the documents from MongoDB. If set to true, the query + * will traverse the results by order of the _modfied property (does an index scan), which allows it to resume after Review Comment: typo ```suggestion * will traverse the results by order of the _modified property (does an index scan), which allows it to resume after ``` ########## oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/NodeStateEntryBatchTest.java: ########## @@ -0,0 +1,97 @@ +/* + * 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.junit.Test; + +import java.nio.BufferOverflowException; +import java.nio.ByteBuffer; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class NodeStateEntryBatchTest { + + @Test + public void testMaximumNumberOfEntries() { + NodeStateEntryBatch batch = NodeStateEntryBatch.createNodeStateEntryBatch(1024, 2); + assertFalse(batch.isAtMaxEntries()); + batch.addEntry("a", new byte[1]); + assertFalse(batch.isAtMaxEntries()); + batch.addEntry("b", new byte[1]); + assertEquals(2, batch.numberOfEntries()); + assertTrue(batch.isAtMaxEntries()); + try { + batch.addEntry("c", new byte[1]); + throw new AssertionError("Expected exception"); + } catch (IllegalStateException e) { + // expected + } + } Review Comment: what about using `expected`? Same for the test below. ```suggestion @Test(expected = IllegalStateException.class) public void testMaximumNumberOfEntries() { NodeStateEntryBatch batch = NodeStateEntryBatch.createNodeStateEntryBatch(1024, 2); assertFalse(batch.isAtMaxEntries()); batch.addEntry("a", new byte[1]); assertFalse(batch.isAtMaxEntries()); batch.addEntry("b", new byte[1]); assertEquals(2, batch.numberOfEntries()); assertTrue(batch.isAtMaxEntries()); batch.addEntry("c", new byte[1]); }``` ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMergeSortTask.java: ########## @@ -0,0 +1,140 @@ +/* + * 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.base.Stopwatch; +import org.apache.jackrabbit.oak.commons.Compression; +import org.apache.jackrabbit.oak.commons.sort.ExternalSort; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedWriter; +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.function.Function; + +import static org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCountBin; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.createWriter; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.getSortedStoreFileName; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.sizeOf; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.pipelined.PipelinedStrategy.FLATFILESTORE_CHARSET; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.pipelined.PipelinedStrategy.SENTINEL_SORTED_FILES_QUEUE; + +/** + * Accumulates the intermediate sorted files and, when all files are genereted, merges them into a single sorted file, Review Comment: typo ```suggestion * Accumulates the intermediate sorted files and, when all files are generated, merges them into a single sorted file, ``` ########## oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedMergeSortTask.java: ########## @@ -0,0 +1,140 @@ +/* + * 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.base.Stopwatch; +import org.apache.jackrabbit.oak.commons.Compression; +import org.apache.jackrabbit.oak.commons.sort.ExternalSort; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedWriter; +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.function.Function; + +import static org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCountBin; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.createWriter; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.getSortedStoreFileName; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.sizeOf; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.pipelined.PipelinedStrategy.FLATFILESTORE_CHARSET; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.pipelined.PipelinedStrategy.SENTINEL_SORTED_FILES_QUEUE; + +/** + * Accumulates the intermediate sorted files and, when all files are genereted, merges them into a single sorted file, + * the flat file store + */ +class PipelinedMergeSortTask implements Callable<PipelinedMergeSortTask.Result> { + // TODO: start merging small files into larger files to avoid having too many "small" files at the end. + // Idea: when there are more than k (for instance, 10) intermediate files whose size is under a certain limit + // (for instance, 1GB), merge them into a single file. And repeat the test whenever this task receives a new + // intermediate file. + public static class Result { + private final File flatFileStoreFile; + private final int filesMerged; + + public Result(File flatFileStoreFile, int filesMerged) { + this.flatFileStoreFile = flatFileStoreFile; + this.filesMerged = filesMerged; + } + + public File getFlatFileStoreFile() { + return flatFileStoreFile; + } + + public int getFilesMerged() { + return filesMerged; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(PipelinedMergeSortTask.class); + + private final File storeDir; + private final Comparator<NodeStateHolder> comparator; + private final Compression algorithm; + private final BlockingQueue<File> sortedFilesQueue; + private final ArrayList<File> sortedFiles = new ArrayList<>(); + + public PipelinedMergeSortTask(File storeDir, + PathElementComparator pathComparator, + Compression algorithm, + BlockingQueue<File> sortedFilesQueue) { + this.storeDir = storeDir; + this.comparator = (e1, e2) -> pathComparator.compare(e1.getPathElements(), e2.getPathElements()); + this.algorithm = algorithm; + this.sortedFilesQueue = sortedFilesQueue; + } + + @Override + public Result call() throws Exception { + String originalName = Thread.currentThread().getName(); + Thread.currentThread().setName("mongo-merge-sort-files"); + try { + LOG.info("Starting merge sort thread"); + while (true) { + LOG.info("Waiting for next intermediate sorted file"); + File sortedIntermediateFile = sortedFilesQueue.take(); + if (sortedIntermediateFile == SENTINEL_SORTED_FILES_QUEUE) { + LOG.info("Going to sort {} files, total size {}", sortedFiles.size(), humanReadableByteCountBin(sizeOf(sortedFiles))); + File flatFileStore = sortStoreFile(sortedFiles); + LOG.info("Terminating sort task. Merged {} files to create the FFS: {} of size {}", + sortedFiles.size(), flatFileStore.getAbsolutePath(), humanReadableByteCountBin(flatFileStore.length())); + return new Result(flatFileStore, sortedFiles.size()); + } + sortedFiles.add(sortedIntermediateFile); + LOG.info("Received new intermediate sorted file {}. Size: {}. Total files: {} of size {}", + sortedIntermediateFile, humanReadableByteCountBin(sortedIntermediateFile.length()), + sortedFiles.size(), humanReadableByteCountBin(sizeOf(sortedFiles))); + } + } catch (InterruptedException t) { + LOG.warn("Thread interrupted", t); + throw t; + } catch (Throwable t) { + LOG.warn("Thread terminating with exception.", t); + throw t; + } finally { + Thread.currentThread().setName(originalName); + } + } + + private File sortStoreFile(List<File> sortedFilesBatch) throws IOException { + Stopwatch w = Stopwatch.createStarted(); + File sortedFile = new File(storeDir, getSortedStoreFileName(algorithm)); + try (BufferedWriter writer = createWriter(sortedFile, algorithm)) { + Function<String, NodeStateHolder> func1 = (line) -> line == null ? null : new NodeStateHolder(line); + Function<NodeStateHolder, String> func2 = holder -> holder == null ? null : holder.getLine(); Review Comment: can you rename these with more meaningful names? -- 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]
