You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@jackrabbit.apache.org by "fabriziofortino (via GitHub)" <gi...@apache.org> on 2023/09/18 09:19:32 UTC
[GitHub] [jackrabbit-oak] fabriziofortino commented on a diff in pull request #1123: OAK-10429: Implementation for IncrementalFFS
fabriziofortino commented on code in PR #1123:
URL: https://github.com/apache/jackrabbit-oak/pull/1123#discussion_r1328370222
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/DocumentStoreIndexerBase.java:
##########
@@ -207,6 +211,30 @@ private List<FlatFileStore> buildFlatFileStoreList(NodeState checkpointedState,
return storeList;
}
+ public IncrementalStore buildIncrementalStore(String initialCheckpoint, String finalCheckpoint) throws IOException, CommitFailedException {
+ IncrementalStoreBuilder builder;
+ IncrementalStore incrementalStore;
+ Set<String> preferredPathElements = new HashSet<>();
+ Set<IndexDefinition> indexDefinitions = getIndexDefinitions();
+ for (IndexDefinition indexDf : indexDefinitions) {
+ preferredPathElements.addAll(indexDf.getRelativeNodeNames());
+ }
Review Comment:
you can use java streams and return an immutable set here
```suggestion
Set<IndexDefinition> indexDefinitions = getIndexDefinitions();
Set<String> preferredPathElements = indexDefinitions.stream()
.flatMap(indexDf -> indexDf.getRelativeNodeNames().stream())
.collect(Collectors.toUnmodifiableSet());
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/indexstore/IndexStoreMetadata.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.indexstore;
+
+import java.util.Set;
+import java.util.function.Predicate;
+
+public class IndexStoreMetadata {
+
+ private String checkpoint;
+ private String storeType;
+ private String strategy;
+ private Set<String> preferredPaths;
+ private Predicate<String> pathPredicate;
Review Comment:
these should be all `final`s
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilder.java:
##########
@@ -244,73 +255,83 @@ public FlatFileStore build() throws IOException, CompositeException {
}
public List<FlatFileStore> buildList(IndexHelper indexHelper, IndexerSupport indexerSupport,
- Set<IndexDefinition> indexDefinitions) throws IOException, CompositeException {
+ Set<IndexDefinition> indexDefinitions) throws IOException, CompositeException {
logFlags();
comparator = new PathElementComparator(preferredPathElements);
entryWriter = new NodeStateEntryWriter(blobStore);
- List<File> fileList = createdSortedStoreFiles();
-
+ Pair<List<File>, List<File>> pair = createdSortedStoreFiles();
+ List<File> fileList = pair.getKey();
+ File metadataFile = pair.getValue() == null || pair.getValue().size() == 0 ? null : pair.getValue().get(0);
Review Comment:
use `isEmpty()`
```suggestion
File metadataFile = pair.getValue() == null || pair.getValue().isEmpty() ? null : pair.getValue().get(0);
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/pipelined/PipelinedStrategy.java:
##########
@@ -223,19 +221,37 @@ public PipelinedStrategy(MongoDocumentStore documentStore,
Compression algorithm,
Predicate<String> pathPredicate,
List<PathFilter> pathFilters) {
+ super(storeDir, algorithm, pathPredicate, preferredPathElements, null);
this.docStore = documentStore;
this.documentNodeStore = documentNodeStore;
this.rootRevision = rootRevision;
this.blobStore = blobStore;
- this.storeDir = storeDir;
this.pathComparator = new PathElementComparator(preferredPathElements);
- this.pathPredicate = pathPredicate;
- this.algorithm = algorithm;
this.pathFilters = pathFilters;
Preconditions.checkState(documentStore.isReadOnly(), "Traverser can only be used with readOnly store");
Review Comment:
this should be replaced with:
```java
this(documentStore, documentNodeStore, rootRevision, preferredPathElements, blobStore, storeDir, algorithm, pathPredicate, pathFilters, null);
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateEntryWriter.java:
##########
@@ -118,14 +122,53 @@ public static String getPath(String entryLine) {
return entryLine.substring(0, getDelimiterPosition(entryLine));
}
+ /*
+ Currently we only have 2 types of store:
+ 1. FlatFileStore
+ 2. IncrementalFlatFileStore
+ */
public static String[] getParts(String line) {
- int pos = getDelimiterPosition(line);
- return new String[]{line.substring(0, pos), line.substring(pos + 1)};
+ // file is FlatFileStore
+ if (line.endsWith("}")) {
Review Comment:
this looks quite hacky. Is there a cleaner way to understand the type? Or even better, should we have different `NodeStateEntryWriter` implementations to avoid this?
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateEntryWriter.java:
##########
@@ -20,6 +20,10 @@
package org.apache.jackrabbit.oak.index.indexer.document.flatfile;
import org.apache.jackrabbit.guava.common.base.Joiner;
+
+import java.util.LinkedList;
+import java.util.List;
+
Review Comment:
these imports should be together with the other `java.util` imports
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalFlatFileStoreStrategy.java:
##########
@@ -0,0 +1,181 @@
+/*
+ * 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.incrementalstore;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Stopwatch;
+import org.apache.commons.io.FileUtils;
+import org.apache.jackrabbit.oak.commons.Compression;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntrySorter;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryWriter;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.PathElementComparator;
+import org.apache.jackrabbit.oak.spi.commit.EditorDiff;
+import org.apache.jackrabbit.oak.spi.commit.VisibleEditor;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.jackrabbit.oak.spi.state.NodeStore;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCount;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileNodeStoreBuilder.OAK_INDEXER_MAX_SORT_MEMORY_IN_GB;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileNodeStoreBuilder.OAK_INDEXER_MAX_SORT_MEMORY_IN_GB_DEFAULT;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.getMetadataFileName;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.getSortedStoreFileName;
+
+public class IncrementalFlatFileStoreStrategy implements IncrementalIndexStoreSortStrategy {
+
+ private final Logger log = LoggerFactory.getLogger(getClass());
+ private final String STORE_TYPE = "IncrementalFlatFileStore";
+ public static final String OAK_INDEXER_DELETE_ORIGINAL = "oak.indexer.deleteOriginal";
+ private final String beforeCheckpoint;
+ private final String afterCheckpoint;
+ private final PathElementComparator comparator;
+ private final NodeStateEntryWriter entryWriter;
+ private final File storeDir;
+ private final NodeStore nodeStore;
+ private final Compression algorithm;
+ private final Predicate<String> pathPredicate;
+ private final boolean deleteOriginal = Boolean.parseBoolean(System.getProperty(OAK_INDEXER_DELETE_ORIGINAL, "true"));
+ private final int maxMemory = Integer.getInteger(OAK_INDEXER_MAX_SORT_MEMORY_IN_GB, OAK_INDEXER_MAX_SORT_MEMORY_IN_GB_DEFAULT);
+ private long textSize = 0;
+ private long entryCount = 0;
+ private final Set<String> preferredPathElements;
+
+ public IncrementalFlatFileStoreStrategy(NodeStore nodeStore, @NotNull String beforeCheckpoint, @NotNull String afterCheckpoint, File storeDir,
+ Set<String> preferredPathElements, Compression algorithm, Predicate<String> pathPredicate, NodeStateEntryWriter entryWriter) {
+ this.nodeStore = nodeStore;
+ this.beforeCheckpoint = beforeCheckpoint;
+ this.afterCheckpoint = afterCheckpoint;
+ this.storeDir = storeDir;
+ this.algorithm = algorithm;
+ this.pathPredicate = pathPredicate;
+ this.entryWriter = entryWriter;
+ this.preferredPathElements = preferredPathElements;
+ this.comparator = new PathElementComparator(preferredPathElements);
+ }
+
+ @Override
+ public File createSortedStoreFile() throws IOException {
+ Stopwatch sw = Stopwatch.createStarted();
+ File file = new File(storeDir, getSortedStoreFileName(algorithm));
+ try (BufferedWriter w = FlatFileStoreUtils.createWriter(file, algorithm)) {
+ NodeState before = Objects.requireNonNull(nodeStore.retrieve(beforeCheckpoint));
+ NodeState after = Objects.requireNonNull(nodeStore.retrieve(afterCheckpoint));
+ EditorDiff.process(VisibleEditor.wrap(new IncrementalFlatFileStoreEditor(w, entryWriter, pathPredicate, this)), before, after);
+ }
+ String sizeStr = algorithm == null || algorithm.equals(Compression.NONE) ? "" : String.format("compressed/%s actual size", humanReadableByteCount(textSize));
Review Comment:
I don't think `algorithm` can be null here
```suggestion
String sizeStr = algorithm.equals(Compression.NONE) ? "" : String.format("compressed/%s actual size", humanReadableByteCount(textSize));```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/TraverseWithSortStrategy.java:
##########
@@ -92,17 +93,22 @@ class TraverseWithSortStrategy implements SortStrategy {
private File sortWorkDir;
private List<File> sortedFiles = new ArrayList<>();
private ArrayList<NodeStateHolder> entryBatch = new ArrayList<>();
- private Predicate<String> pathPredicate;
-
+ TraverseWithSortStrategy(NodeStateEntryTraverserFactory nodeStatesFactory, Set<String> preferredPaths,
+ NodeStateEntryWriter entryWriter, File storeDir, Compression algorithm,
+ Predicate<String> pathPredicate, String checkpoint) {
+ super(storeDir, algorithm, pathPredicate, preferredPaths, checkpoint);
+ this.nodeStatesFactory = nodeStatesFactory;
+ this.entryWriter = entryWriter;
+ this.comparator = (e1, e2) -> new PathElementComparator(preferredPaths).compare(e1.getPathElements(), e2.getPathElements());
+ }
+ @Deprecated
TraverseWithSortStrategy(NodeStateEntryTraverserFactory nodeStatesFactory, PathElementComparator pathComparator,
NodeStateEntryWriter entryWriter, File storeDir, Compression algorithm, Predicate<String> pathPredicate) {
+ super(storeDir, algorithm, pathPredicate,null, null);
this.nodeStatesFactory = nodeStatesFactory;
this.entryWriter = entryWriter;
- this.storeDir = storeDir;
this.comparator = (e1, e2) -> pathComparator.compare(e1.getPathElements(), e2.getPathElements());
- this.pathPredicate = pathPredicate;
- this.algorithm = algorithm;
Review Comment:
this should be replaced with
```java
this(nodeStatesFactory, null, entryWriter, storeDir, algorithm, pathPredicate, null);
```
##########
oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilderTest.java:
##########
@@ -34,6 +34,7 @@
import org.apache.jackrabbit.oak.index.IndexHelper;
import org.apache.jackrabbit.oak.index.IndexerSupport;
import org.apache.jackrabbit.oak.index.indexer.document.CompositeException;
+import org.apache.jackrabbit.oak.index.indexer.document.indexstore.IndexStoreSortStrategy;
Review Comment:
```suggestion
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalIndexStoreMetadata.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * 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.incrementalstore;
+
+import java.util.List;
+import java.util.function.Predicate;
+
+public class IncrementalIndexStoreMetadata {
+
+ private String beforeCheckpoint;
+ private String afterCheckpoint;
+ private String storeType;
+ private String strategy;
+
+ private List<String> preferredPaths;
+ private Predicate<String> pathPredicate;
Review Comment:
finals
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/MergeIncrementalFlatFileStore.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.incrementalstore;
+
+import org.apache.jackrabbit.oak.commons.Compression;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryWriter;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateHolder;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.PathElementComparator;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.SimpleNodeStateHolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.util.Comparator;
+import java.util.Set;
+
+import static org.apache.jackrabbit.guava.common.base.Preconditions.checkState;
+
+
+public class MergeIncrementalFlatFileStore {
+
+ private final Logger log = LoggerFactory.getLogger(getClass());
+ private final File baseFFS;
+ private final File incrementalFFS;
+ private final File merged;
+ private final Set<String> preferredPathElements;
+ private final Compression algorithm;
+
+ public MergeIncrementalFlatFileStore(Set<String> preferredPathElements, File baseFFS, File incrementalFFS,
+ File merged, Compression algorithm) {
+ this.preferredPathElements = preferredPathElements;
+ this.baseFFS = baseFFS;
+ this.incrementalFFS = incrementalFFS;
+ this.merged = merged;
+ this.algorithm = algorithm;
+ }
+
+ public void doMerge() throws IOException {
+
+ log.info("base FFS " + baseFFS.getAbsolutePath());
+ log.info("incremental FFS " + incrementalFFS.getAbsolutePath());
+ log.info("merged FFS " + merged.getAbsolutePath());
+
+ try (BufferedWriter writer = FlatFileStoreUtils.createWriter(merged, algorithm);
+ BufferedReader baseFFSBufferedReader = FlatFileStoreUtils.createReader(baseFFS, algorithm);
+ BufferedReader incrementalFFSBufferedReader = FlatFileStoreUtils.createReader(incrementalFFS, algorithm)) {
+ String baseLine = baseFFSBufferedReader.readLine();
+ String incLine = incrementalFFSBufferedReader.readLine();
+
+ Comparator<NodeStateHolder> comparator = (e1, e2) -> new PathElementComparator(preferredPathElements).compare(e1.getPathElements(), e2.getPathElements());
+ int compared;
+ while (baseLine != null || incLine != null) {
+ if (baseLine != null && incLine != null) {
+ compared = comparator.compare(new SimpleNodeStateHolder(baseLine), new SimpleNodeStateHolder(incLine));
+ if (compared < 0) {
+ writer.write(baseLine);
+ writer.write("\n");
+ baseLine = baseFFSBufferedReader.readLine();
+ } else if (compared > 0) {
+ writer.write(removeOperand(incLine));
+ writer.write("\n");
+ incLine = incrementalFFSBufferedReader.readLine();
+ } else {
+ String operand = NodeStateEntryWriter.getParts(incLine)[3];
+ checkState(IncrementalStoreOperand.DELETE.toString().equals(operand)
+ || IncrementalStoreOperand.ADD.toString().equals(operand)
+ || IncrementalStoreOperand.MODIFY.toString().equals(operand),
+ "wrong operand in incremental ffs: {} ", operand);
+ if (!(IncrementalStoreOperand.DELETE.toString().equals(operand))) {
+ writer.write(removeOperand(incLine));
+ writer.write("\n");
+ }
+ baseLine = baseFFSBufferedReader.readLine();
+ incLine = incrementalFFSBufferedReader.readLine();
+ }
+ } else {
+ if (incLine == null) {
+ writer.write(baseLine);
+ writer.write("\n");
+ writeRestOfFile(writer, baseFFSBufferedReader, false);
+ baseLine = baseFFSBufferedReader.readLine();
+ } else {
+ String operand = getOperand(incLine);
+ if (IncrementalStoreOperand.MODIFY.toString().equals(operand) || IncrementalStoreOperand.DELETE.toString().equals(operand)) {
+ log.error("incremental ffs should not have modify and delete operands: {}", incLine);
+ throw new RuntimeException("incremental ffs should not have modify and delete operands" + incLine);
+ }
+ writer.write(removeOperand(incLine));
+ writer.write("\n");
+ writeRestOfFile(writer, incrementalFFSBufferedReader, true);
+ incLine = incrementalFFSBufferedReader.readLine();
+ }
+ }
+ }
+ }
+ }
+
+ private void writeRestOfFile(BufferedWriter writer, BufferedReader bufferedReader, boolean isIncrementalFile) {
+ if (isIncrementalFile) {
+ bufferedReader.lines().forEach((n) -> {
+ String operand = getOperand(n);
+ if (IncrementalStoreOperand.MODIFY.toString().equals(operand) || IncrementalStoreOperand.DELETE.toString().equals(operand)) {
+ log.error("incremental ffs should not have modify and delete operands: {}", n);
+ throw new RuntimeException("incremental ffs should not have modify and delete operands" + n);
+ }
+ try {
+ writer.write(removeOperand(n));
+ writer.write("\n");
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ } else {
+ bufferedReader.lines().forEach(n -> {
+ try {
+ writer.write(n);
+ writer.write("\n");
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+ }
+
+ private static String getOperand(String incLine) {
+ String[] parts = NodeStateEntryWriter.getParts(incLine);
+ String operand = parts[2];
+ return operand;
Review Comment:
redundant
```suggestion
return parts[2];
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/indexstore/IndexStoreSortStrategyBase.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.indexstore;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.jackrabbit.oak.commons.Compression;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.util.Set;
+import java.util.function.Predicate;
+
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.getMetadataFileName;
+
+public abstract class IndexStoreSortStrategyBase implements IndexStoreSortStrategy {
+ private final Logger log = LoggerFactory.getLogger(getClass());
+ /**
+ * Directory where sorted files will be created.
+ */
+ protected File storeDir;
+ protected Compression algorithm;
+ protected Predicate<String> pathPredicate;
+ protected Set<String> preferredPaths;
+ protected String checkpoint;
+
+ private final String DEFAULT_INDEX_STORE_TYPE = "FlatFileStore";
Review Comment:
static?
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/MergeIncrementalFlatFileStore.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.incrementalstore;
+
+import org.apache.jackrabbit.oak.commons.Compression;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryWriter;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateHolder;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.PathElementComparator;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.SimpleNodeStateHolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.util.Comparator;
+import java.util.Set;
+
+import static org.apache.jackrabbit.guava.common.base.Preconditions.checkState;
+
+
+public class MergeIncrementalFlatFileStore {
+
+ private final Logger log = LoggerFactory.getLogger(getClass());
+ private final File baseFFS;
+ private final File incrementalFFS;
+ private final File merged;
+ private final Set<String> preferredPathElements;
+ private final Compression algorithm;
+
+ public MergeIncrementalFlatFileStore(Set<String> preferredPathElements, File baseFFS, File incrementalFFS,
+ File merged, Compression algorithm) {
+ this.preferredPathElements = preferredPathElements;
+ this.baseFFS = baseFFS;
+ this.incrementalFFS = incrementalFFS;
+ this.merged = merged;
+ this.algorithm = algorithm;
+ }
+
+ public void doMerge() throws IOException {
+
+ log.info("base FFS " + baseFFS.getAbsolutePath());
+ log.info("incremental FFS " + incrementalFFS.getAbsolutePath());
+ log.info("merged FFS " + merged.getAbsolutePath());
+
+ try (BufferedWriter writer = FlatFileStoreUtils.createWriter(merged, algorithm);
+ BufferedReader baseFFSBufferedReader = FlatFileStoreUtils.createReader(baseFFS, algorithm);
+ BufferedReader incrementalFFSBufferedReader = FlatFileStoreUtils.createReader(incrementalFFS, algorithm)) {
+ String baseLine = baseFFSBufferedReader.readLine();
+ String incLine = incrementalFFSBufferedReader.readLine();
+
+ Comparator<NodeStateHolder> comparator = (e1, e2) -> new PathElementComparator(preferredPathElements).compare(e1.getPathElements(), e2.getPathElements());
+ int compared;
+ while (baseLine != null || incLine != null) {
+ if (baseLine != null && incLine != null) {
+ compared = comparator.compare(new SimpleNodeStateHolder(baseLine), new SimpleNodeStateHolder(incLine));
+ if (compared < 0) {
+ writer.write(baseLine);
+ writer.write("\n");
+ baseLine = baseFFSBufferedReader.readLine();
+ } else if (compared > 0) {
+ writer.write(removeOperand(incLine));
+ writer.write("\n");
+ incLine = incrementalFFSBufferedReader.readLine();
+ } else {
+ String operand = NodeStateEntryWriter.getParts(incLine)[3];
+ checkState(IncrementalStoreOperand.DELETE.toString().equals(operand)
+ || IncrementalStoreOperand.ADD.toString().equals(operand)
+ || IncrementalStoreOperand.MODIFY.toString().equals(operand),
+ "wrong operand in incremental ffs: {} ", operand);
+ if (!(IncrementalStoreOperand.DELETE.toString().equals(operand))) {
+ writer.write(removeOperand(incLine));
+ writer.write("\n");
+ }
+ baseLine = baseFFSBufferedReader.readLine();
+ incLine = incrementalFFSBufferedReader.readLine();
+ }
+ } else {
+ if (incLine == null) {
+ writer.write(baseLine);
+ writer.write("\n");
+ writeRestOfFile(writer, baseFFSBufferedReader, false);
+ baseLine = baseFFSBufferedReader.readLine();
+ } else {
+ String operand = getOperand(incLine);
+ if (IncrementalStoreOperand.MODIFY.toString().equals(operand) || IncrementalStoreOperand.DELETE.toString().equals(operand)) {
+ log.error("incremental ffs should not have modify and delete operands: {}", incLine);
+ throw new RuntimeException("incremental ffs should not have modify and delete operands" + incLine);
+ }
+ writer.write(removeOperand(incLine));
+ writer.write("\n");
+ writeRestOfFile(writer, incrementalFFSBufferedReader, true);
+ incLine = incrementalFFSBufferedReader.readLine();
+ }
+ }
+ }
+ }
+ }
+
+ private void writeRestOfFile(BufferedWriter writer, BufferedReader bufferedReader, boolean isIncrementalFile) {
+ if (isIncrementalFile) {
+ bufferedReader.lines().forEach((n) -> {
+ String operand = getOperand(n);
+ if (IncrementalStoreOperand.MODIFY.toString().equals(operand) || IncrementalStoreOperand.DELETE.toString().equals(operand)) {
+ log.error("incremental ffs should not have modify and delete operands: {}", n);
+ throw new RuntimeException("incremental ffs should not have modify and delete operands" + n);
+ }
+ try {
+ writer.write(removeOperand(n));
+ writer.write("\n");
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ } else {
+ bufferedReader.lines().forEach(n -> {
+ try {
+ writer.write(n);
+ writer.write("\n");
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+ }
+
+ private static String getOperand(String incLine) {
+ String[] parts = NodeStateEntryWriter.getParts(incLine);
+ String operand = parts[2];
+ return operand;
+ }
+
+ private String removeOperand(String line) {
+ String[] parts = NodeStateEntryWriter.getParts(line);
+ return new StringBuilder().append(parts[0]).append("|").append(parts[1]).toString();
Review Comment:
I would avoid to use `StringBuilder` here
```suggestion
return parts[0] + "|" + parts[1];
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/FlatFileNodeStoreBuilder.java:
##########
@@ -244,73 +255,83 @@ public FlatFileStore build() throws IOException, CompositeException {
}
public List<FlatFileStore> buildList(IndexHelper indexHelper, IndexerSupport indexerSupport,
- Set<IndexDefinition> indexDefinitions) throws IOException, CompositeException {
+ Set<IndexDefinition> indexDefinitions) throws IOException, CompositeException {
logFlags();
comparator = new PathElementComparator(preferredPathElements);
entryWriter = new NodeStateEntryWriter(blobStore);
- List<File> fileList = createdSortedStoreFiles();
-
+ Pair<List<File>, List<File>> pair = createdSortedStoreFiles();
+ List<File> fileList = pair.getKey();
+ File metadataFile = pair.getValue() == null || pair.getValue().size() == 0 ? null : pair.getValue().get(0);
+
long start = System.currentTimeMillis();
// If not already split, split otherwise skip splitting
if (!fileList.stream().allMatch(FlatFileSplitter.IS_SPLIT)) {
NodeStore nodeStore = new MemoryNodeStore(indexerSupport.retrieveNodeStateForCheckpoint());
FlatFileSplitter splitter = new FlatFileSplitter(fileList.get(0), indexHelper.getWorkDir(),
- new NodeStateNodeTypeInfoProvider(nodeStore.getRoot()), new NodeStateEntryReader(blobStore),
- indexDefinitions);
+ new NodeStateNodeTypeInfoProvider(nodeStore.getRoot()), new NodeStateEntryReader(blobStore),
+ indexDefinitions);
fileList = splitter.split();
log.info("Split flat file to result files '{}' is done, took {} ms", fileList, System.currentTimeMillis() - start);
}
List<FlatFileStore> storeList = new ArrayList<>();
for (File flatFileItem : fileList) {
- FlatFileStore store = new FlatFileStore(blobStore, flatFileItem, new NodeStateEntryReader(blobStore),
+ FlatFileStore store = new FlatFileStore(blobStore, flatFileItem, metadataFile, new NodeStateEntryReader(blobStore),
unmodifiableSet(preferredPathElements), algorithm);
storeList.add(store);
}
return storeList;
}
/**
- * Returns the existing list of store files if it can read from system property OAK_INDEXER_SORTED_FILE_PATH which
- * defines the existing folder where the flat file store files are present. Will throw an exception if it cannot
+ * Returns the existing list of store files if it can read from system property OAK_INDEXER_SORTED_FILE_PATH which
+ * defines the existing folder where the flat file store files are present. Will throw an exception if it cannot
* read or the path in the system property is not a directory.
- * If the system property OAK_INDEXER_SORTED_FILE_PATH in undefined, or it cannot read relevant files it
+ * If the system property OAK_INDEXER_SORTED_FILE_PATH in undefined, or it cannot read relevant files it
* initializes the flat file store.
- *
- * @return list of flat files
- * @throws IOException
+ *
+ * @return pair of "list of flat files" and metadata file
+ * @throws IOException
* @throws CompositeException
*/
- private List<File> createdSortedStoreFiles() throws IOException, CompositeException {
+ private Pair<List<File>, List<File>> createdSortedStoreFiles() throws IOException, CompositeException {
// Check system property defined path
String sortedFilePath = System.getProperty(OAK_INDEXER_SORTED_FILE_PATH);
if (StringUtils.isNotBlank(sortedFilePath)) {
File sortedDir = new File(sortedFilePath);
log.info("Attempting to read from provided sorted files directory [{}] (via system property '{}')",
- sortedDir.getAbsolutePath(), OAK_INDEXER_SORTED_FILE_PATH);
- List<File> files = getFiles(sortedDir);
- if (files != null) {
- return files;
+ sortedDir.getAbsolutePath(), OAK_INDEXER_SORTED_FILE_PATH);
+ // List of storefiles, List of metadatafile
+ Pair<List<File>, List<File>> storeFiles = getIndexStoreFiles(sortedDir);
+ if (storeFiles != null) {
+ return storeFiles;
}
}
// Initialize the flat file store again
createStoreDir();
- SortStrategy strategy = createSortStrategy(flatFileStoreDir);
+ IndexStoreSortStrategy strategy = createSortStrategy(flatFileStoreDir);
File result = strategy.createSortedStoreFile();
+ File metadata = strategy.createMetadataFile();
entryCount = strategy.getEntryCount();
- return Collections.singletonList(result);
+ return new ImmutablePair<>(Collections.singletonList(result), Collections.singletonList(metadata));
}
@Nullable
- private List<File> getFiles(File sortedDir) {
+ private Pair<List<File>, List<File>> getIndexStoreFiles(File sortedDir) {
if (sortedDir.exists() && sortedDir.canRead() && sortedDir.isDirectory()) {
- File[] files = sortedDir.listFiles(
- (dir, name) -> name.endsWith(FlatFileStoreUtils.getSortedStoreFileName(algorithm)));
- if (files != null && files.length != 0) {
- return Arrays.asList(files);
+ File[] storeFiles = sortedDir.listFiles(
+ (dir, name) -> name.endsWith(FlatFileStoreUtils.getSortedStoreFileName(algorithm)));
+ File[] metadataFiles = sortedDir.listFiles(
+ (dir, name) -> name.endsWith(FlatFileStoreUtils.getMetadataFileName(algorithm)));
+ // Not throwing error for backward compatibility
+ if (metadataFiles == null || metadataFiles.length == 0) {
+ log.error("Unable to find metadata file in files directory:{}", sortedDir.getAbsolutePath());
+ }
Review Comment:
I see an Exception is not thrown for backward compatibility. But in that case, I think line 334 will throw a runtime Exception because `Arrays.asList` accept a not nullable varargs
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/MultithreadedTraverseWithSortStrategy.java:
##########
@@ -23,9 +23,11 @@
import org.apache.jackrabbit.guava.common.collect.Lists;
import org.apache.jackrabbit.oak.commons.Compression;
import org.apache.jackrabbit.oak.index.indexer.document.CompositeException;
+import org.apache.jackrabbit.oak.index.indexer.document.indexstore.IndexStoreSortStrategyBase;
import org.apache.jackrabbit.oak.index.indexer.document.LastModifiedRange;
import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntryTraverser;
import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntryTraverserFactory;
+import org.apache.jackrabbit.oak.plugins.document.DocumentNodeState;
Review Comment:
unused
```suggestion
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/StoreAndSortStrategy.java:
##########
@@ -22,10 +22,12 @@
import org.apache.commons.io.FileUtils;
import org.apache.jackrabbit.guava.common.base.Stopwatch;
import org.apache.jackrabbit.oak.commons.Compression;
+import org.apache.jackrabbit.oak.index.indexer.document.indexstore.IndexStoreSortStrategyBase;
import org.apache.jackrabbit.oak.index.indexer.document.LastModifiedRange;
import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntry;
import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntryTraverser;
import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntryTraverserFactory;
+import org.apache.jackrabbit.oak.plugins.document.DocumentNodeState;
Review Comment:
unused
```suggestion
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateEntryWriter.java:
##########
@@ -118,14 +122,53 @@ public static String getPath(String entryLine) {
return entryLine.substring(0, getDelimiterPosition(entryLine));
}
+ /*
+ Currently we only have 2 types of store:
+ 1. FlatFileStore
+ 2. IncrementalFlatFileStore
+ */
public static String[] getParts(String line) {
- int pos = getDelimiterPosition(line);
- return new String[]{line.substring(0, pos), line.substring(pos + 1)};
+ // file is FlatFileStore
+ if (line.endsWith("}")) {
+ int pos = getDelimiterPosition(line);
+ return new String[]{line.substring(0, pos), line.substring(pos + 1)};
+ } else {
+ // file is IncrementalFlatFileStore
+ List<Integer> positions = getDelimiterPositions(line);
+ checkState(positions.size() >= 2, "Invalid path entry [%s]", line);
+ // there are 4 parts in incrementalFFS and default delimiter is |
+ // path|nodeData|checkpoint|operand
+ // Node's data can itself have many | so we split based on first and last 2 |
+ String[] parts = new String[4];
+ parts[0] = line.substring(0, positions.get(0));
+ parts[1] = line.substring(positions.get(0) + 1, positions.get(positions.size() - 2));
+ parts[2] = line.substring(positions.get(positions.size() - 2) + 1, positions.get(positions.size() - 1));
+ parts[3] = line.substring(positions.get(positions.size() - 1) + 1);
+ return parts;
+ }
}
private static int getDelimiterPosition(String entryLine) {
int indexOfPipe = entryLine.indexOf(NodeStateEntryWriter.DELIMITER);
checkState(indexOfPipe > 0, "Invalid path entry [%s]", entryLine);
return indexOfPipe;
}
+
+ private static List<Integer> getDelimiterPositions(String entryLine) {
+ List<Integer> indexPositions = new LinkedList<>();
+ int index = 0;
+ int indexOfPipe;
+ while (true) {
+ indexOfPipe = entryLine.indexOf(NodeStateEntryWriter.DELIMITER, index);
+ if (indexOfPipe > 0) {
+ indexPositions.add(indexOfPipe);
+ index = indexOfPipe + 1;
+ } else {
+ break;
+ }
+ }
+ checkState(indexPositions.size() > 0, "Invalid path entry [%s]", entryLine);
Review Comment:
use `isEmpty()`
```suggestion
checkState(!indexPositions.isEmpty(), "Invalid path entry [%s]", entryLine);
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/StoreAndSortStrategy.java:
##########
@@ -42,31 +45,36 @@
import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileNodeStoreBuilder.OAK_INDEXER_MAX_SORT_MEMORY_IN_GB_DEFAULT;
import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.getSortedStoreFileName;
-class StoreAndSortStrategy implements SortStrategy {
+class StoreAndSortStrategy extends IndexStoreSortStrategyBase {
private static final String OAK_INDEXER_DELETE_ORIGINAL = "oak.indexer.deleteOriginal";
private static final int LINE_SEP_LENGTH = LINE_SEPARATOR.value().length();
private final Logger log = LoggerFactory.getLogger(getClass());
private final NodeStateEntryTraverserFactory nodeStatesFactory;
private final PathElementComparator comparator;
private final NodeStateEntryWriter entryWriter;
- private final File storeDir;
- private final Compression algorithm;
private long entryCount;
private boolean deleteOriginal = Boolean.parseBoolean(System.getProperty(OAK_INDEXER_DELETE_ORIGINAL, "true"));
private int maxMemory = Integer.getInteger(OAK_INDEXER_MAX_SORT_MEMORY_IN_GB, OAK_INDEXER_MAX_SORT_MEMORY_IN_GB_DEFAULT);
private long textSize;
- private Predicate<String> pathPredicate;
+ public StoreAndSortStrategy(NodeStateEntryTraverserFactory nodeStatesFactory, Set<String> preferredPaths,
+ NodeStateEntryWriter entryWriter, File storeDir, Compression algorithm,
+ Predicate<String> pathPredicate, String checkpoint) {
+ super(storeDir, algorithm, pathPredicate,preferredPaths, checkpoint);
Review Comment:
missing space
```suggestion
super(storeDir, algorithm, pathPredicate, preferredPaths, checkpoint);
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/TraverseWithSortStrategy.java:
##########
@@ -23,10 +23,12 @@
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.apache.jackrabbit.oak.index.indexer.document.indexstore.IndexStoreSortStrategyBase;
import org.apache.jackrabbit.oak.index.indexer.document.LastModifiedRange;
import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntry;
import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntryTraverser;
import org.apache.jackrabbit.oak.index.indexer.document.NodeStateEntryTraverserFactory;
+import org.apache.jackrabbit.oak.plugins.document.DocumentNodeState;
Review Comment:
unused
```suggestion
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/StoreAndSortStrategy.java:
##########
@@ -42,31 +45,36 @@
import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileNodeStoreBuilder.OAK_INDEXER_MAX_SORT_MEMORY_IN_GB_DEFAULT;
import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.getSortedStoreFileName;
-class StoreAndSortStrategy implements SortStrategy {
+class StoreAndSortStrategy extends IndexStoreSortStrategyBase {
private static final String OAK_INDEXER_DELETE_ORIGINAL = "oak.indexer.deleteOriginal";
private static final int LINE_SEP_LENGTH = LINE_SEPARATOR.value().length();
private final Logger log = LoggerFactory.getLogger(getClass());
private final NodeStateEntryTraverserFactory nodeStatesFactory;
private final PathElementComparator comparator;
private final NodeStateEntryWriter entryWriter;
- private final File storeDir;
- private final Compression algorithm;
private long entryCount;
private boolean deleteOriginal = Boolean.parseBoolean(System.getProperty(OAK_INDEXER_DELETE_ORIGINAL, "true"));
private int maxMemory = Integer.getInteger(OAK_INDEXER_MAX_SORT_MEMORY_IN_GB, OAK_INDEXER_MAX_SORT_MEMORY_IN_GB_DEFAULT);
private long textSize;
- private Predicate<String> pathPredicate;
+ public StoreAndSortStrategy(NodeStateEntryTraverserFactory nodeStatesFactory, Set<String> preferredPaths,
+ NodeStateEntryWriter entryWriter, File storeDir, Compression algorithm,
+ Predicate<String> pathPredicate, String checkpoint) {
+ super(storeDir, algorithm, pathPredicate,preferredPaths, checkpoint);
+ this.nodeStatesFactory = nodeStatesFactory;
+ this.comparator = new PathElementComparator(preferredPaths);
+ this.entryWriter = entryWriter;
+
+ }
+ @Deprecated
public StoreAndSortStrategy(NodeStateEntryTraverserFactory nodeStatesFactory, PathElementComparator comparator,
NodeStateEntryWriter entryWriter, File storeDir, Compression algorithm, Predicate<String> pathPredicate) {
+ super(storeDir, algorithm, pathPredicate, null, null);
this.nodeStatesFactory = nodeStatesFactory;
this.comparator = comparator;
this.entryWriter = entryWriter;
- this.storeDir = storeDir;
- this.algorithm = algorithm;
- this.pathPredicate = pathPredicate;
Review Comment:
this should be replaced with
```java
this(nodeStatesFactory, null, entryWriter, storeDir, algorithm, pathPredicate, null);
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalFlatFileStoreStrategy.java:
##########
@@ -0,0 +1,181 @@
+/*
+ * 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.incrementalstore;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Stopwatch;
+import org.apache.commons.io.FileUtils;
+import org.apache.jackrabbit.oak.commons.Compression;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntrySorter;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryWriter;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.PathElementComparator;
+import org.apache.jackrabbit.oak.spi.commit.EditorDiff;
+import org.apache.jackrabbit.oak.spi.commit.VisibleEditor;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.jackrabbit.oak.spi.state.NodeStore;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCount;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileNodeStoreBuilder.OAK_INDEXER_MAX_SORT_MEMORY_IN_GB;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileNodeStoreBuilder.OAK_INDEXER_MAX_SORT_MEMORY_IN_GB_DEFAULT;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.getMetadataFileName;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.getSortedStoreFileName;
+
+public class IncrementalFlatFileStoreStrategy implements IncrementalIndexStoreSortStrategy {
+
+ private final Logger log = LoggerFactory.getLogger(getClass());
+ private final String STORE_TYPE = "IncrementalFlatFileStore";
Review Comment:
static?
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/TraverseWithSortStrategy.java:
##########
@@ -92,17 +93,22 @@ class TraverseWithSortStrategy implements SortStrategy {
private File sortWorkDir;
private List<File> sortedFiles = new ArrayList<>();
private ArrayList<NodeStateHolder> entryBatch = new ArrayList<>();
- private Predicate<String> pathPredicate;
-
+ TraverseWithSortStrategy(NodeStateEntryTraverserFactory nodeStatesFactory, Set<String> preferredPaths,
Review Comment:
add a space
```suggestion
TraverseWithSortStrategy(NodeStateEntryTraverserFactory nodeStatesFactory, Set<String> preferredPaths,
```
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalStoreBuilder.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.incrementalstore;
+
+import org.apache.jackrabbit.guava.common.collect.Iterables;
+import org.apache.jackrabbit.oak.commons.Compression;
+import org.apache.jackrabbit.oak.index.IndexHelper;
+import org.apache.jackrabbit.oak.index.indexer.document.CompositeException;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileNodeStoreBuilder;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.LZ4Compression;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryReader;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryWriter;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Predicate;
+
+import static java.util.Collections.unmodifiableSet;
+
+public class IncrementalStoreBuilder {
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+ private final String INCREMENTAL_STORE_DIR_NAME_PREFIX = "inc-store";
Review Comment:
static?
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalFlatFileStoreStrategy.java:
##########
@@ -0,0 +1,181 @@
+/*
+ * 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.incrementalstore;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Stopwatch;
+import org.apache.commons.io.FileUtils;
+import org.apache.jackrabbit.oak.commons.Compression;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntrySorter;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryWriter;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.PathElementComparator;
+import org.apache.jackrabbit.oak.spi.commit.EditorDiff;
+import org.apache.jackrabbit.oak.spi.commit.VisibleEditor;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.apache.jackrabbit.oak.spi.state.NodeStore;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCount;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileNodeStoreBuilder.OAK_INDEXER_MAX_SORT_MEMORY_IN_GB;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileNodeStoreBuilder.OAK_INDEXER_MAX_SORT_MEMORY_IN_GB_DEFAULT;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.getMetadataFileName;
+import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileStoreUtils.getSortedStoreFileName;
+
+public class IncrementalFlatFileStoreStrategy implements IncrementalIndexStoreSortStrategy {
+
+ private final Logger log = LoggerFactory.getLogger(getClass());
+ private final String STORE_TYPE = "IncrementalFlatFileStore";
+ public static final String OAK_INDEXER_DELETE_ORIGINAL = "oak.indexer.deleteOriginal";
+ private final String beforeCheckpoint;
+ private final String afterCheckpoint;
+ private final PathElementComparator comparator;
+ private final NodeStateEntryWriter entryWriter;
+ private final File storeDir;
+ private final NodeStore nodeStore;
+ private final Compression algorithm;
+ private final Predicate<String> pathPredicate;
+ private final boolean deleteOriginal = Boolean.parseBoolean(System.getProperty(OAK_INDEXER_DELETE_ORIGINAL, "true"));
+ private final int maxMemory = Integer.getInteger(OAK_INDEXER_MAX_SORT_MEMORY_IN_GB, OAK_INDEXER_MAX_SORT_MEMORY_IN_GB_DEFAULT);
+ private long textSize = 0;
+ private long entryCount = 0;
+ private final Set<String> preferredPathElements;
+
+ public IncrementalFlatFileStoreStrategy(NodeStore nodeStore, @NotNull String beforeCheckpoint, @NotNull String afterCheckpoint, File storeDir,
+ Set<String> preferredPathElements, Compression algorithm, Predicate<String> pathPredicate, NodeStateEntryWriter entryWriter) {
+ this.nodeStore = nodeStore;
+ this.beforeCheckpoint = beforeCheckpoint;
+ this.afterCheckpoint = afterCheckpoint;
+ this.storeDir = storeDir;
+ this.algorithm = algorithm;
+ this.pathPredicate = pathPredicate;
+ this.entryWriter = entryWriter;
+ this.preferredPathElements = preferredPathElements;
+ this.comparator = new PathElementComparator(preferredPathElements);
+ }
+
+ @Override
+ public File createSortedStoreFile() throws IOException {
+ Stopwatch sw = Stopwatch.createStarted();
+ File file = new File(storeDir, getSortedStoreFileName(algorithm));
+ try (BufferedWriter w = FlatFileStoreUtils.createWriter(file, algorithm)) {
+ NodeState before = Objects.requireNonNull(nodeStore.retrieve(beforeCheckpoint));
+ NodeState after = Objects.requireNonNull(nodeStore.retrieve(afterCheckpoint));
+ EditorDiff.process(VisibleEditor.wrap(new IncrementalFlatFileStoreEditor(w, entryWriter, pathPredicate, this)), before, after);
+ }
+ String sizeStr = algorithm == null || algorithm.equals(Compression.NONE) ? "" : String.format("compressed/%s actual size", humanReadableByteCount(textSize));
+ log.info("Dumped {} nodestates in json format in {} ({} {})", entryCount, sw, humanReadableByteCount(file.length()), sizeStr);
+ return sortStoreFile(file);
+ }
+
+ @Override
+ public File createMetadataFile() throws IOException {
+ File metadataFile = new File(storeDir, getMetadataFileName(algorithm));
+ IncrementalIndexStoreMetadata indexStoreMetadata = new IncrementalIndexStoreMetadata(this);
+ try (BufferedWriter metadataWriter = FlatFileStoreUtils.createWriter(metadataFile, algorithm)) {
+ writeMetadataToFile(metadataWriter, indexStoreMetadata);
+ }
+ log.info("Created metadataFile:{} with strategy:{} ", metadataFile.getPath(), indexStoreMetadata.getStoreType());
+ return metadataFile;
+ }
+
+
+ private void writeMetadataToFile(BufferedWriter w, IncrementalIndexStoreMetadata indexStoreMetadata) throws IOException {
+ ObjectMapper mapper = new ObjectMapper();
+ mapper.writeValue(w, indexStoreMetadata);
+ }
Review Comment:
these methods do pretty much the same thing as the ones in `IndexStoreSortStrategyBase`. I think we could improve this code by introducing generics and avoid to duplicate most of the logic
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/incrementalstore/IncrementalStoreBuilder.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.incrementalstore;
+
+import org.apache.jackrabbit.guava.common.collect.Iterables;
+import org.apache.jackrabbit.oak.commons.Compression;
+import org.apache.jackrabbit.oak.index.IndexHelper;
+import org.apache.jackrabbit.oak.index.indexer.document.CompositeException;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.FlatFileNodeStoreBuilder;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.LZ4Compression;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryReader;
+import org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryWriter;
+import org.apache.jackrabbit.oak.spi.blob.BlobStore;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Predicate;
+
+import static java.util.Collections.unmodifiableSet;
+
+public class IncrementalStoreBuilder {
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+ private final String INCREMENTAL_STORE_DIR_NAME_PREFIX = "inc-store";
+
+ private final File workDir;
+ private final IndexHelper indexHelper;
+ private final String initialCheckpoint;
+ private final String finalCheckpoint;
+ private Predicate<String> pathPredicate = path -> true;
+ private Set<String> preferredPathElements = Collections.emptySet();
+ private BlobStore blobStore;
+
+ private final boolean compressionEnabled = Boolean.parseBoolean(System.getProperty(FlatFileNodeStoreBuilder.OAK_INDEXER_USE_ZIP, "true"));
+ private final boolean useLZ4 = Boolean.parseBoolean(System.getProperty(FlatFileNodeStoreBuilder.OAK_INDEXER_USE_LZ4, "false"));
+ private final Compression algorithm = compressionEnabled ? (useLZ4 ? new LZ4Compression() : Compression.GZIP) :
+ Compression.NONE;
+
+
+ /**
+ * System property name for sort strategy. This takes precedence over {@link #INCREMENTAL_SORT_STRATEGY_TYPE}.
+ * Allowed values are the values from enum {@link IncrementalStoreBuilder.IncrementalSortStrategyType}
+ */
+ public static final String INCREMENTAL_SORT_STRATEGY_TYPE = "oak.indexer.incrementalSortStrategyType";
+
+
+ private final String sortStrategyTypeString = System.getProperty(INCREMENTAL_SORT_STRATEGY_TYPE);
+ private IncrementalStoreBuilder.IncrementalSortStrategyType sortStrategyType = sortStrategyTypeString != null
+ ? IncrementalStoreBuilder.IncrementalSortStrategyType.valueOf(sortStrategyTypeString)
+ : IncrementalSortStrategyType.INCREMENTAL_FFS_STORE;
+
+
+ public enum IncrementalSortStrategyType {
+ /**
+ * Incremental store having nodes updated between initial and final checkpoint
+ */
+
+ INCREMENTAL_FFS_STORE
+ }
+
+ public IncrementalStoreBuilder(File workDir, IndexHelper indexHelper,
+ @NotNull String initialCheckpoint, @NotNull String finalCheckpoint) {
+ this.workDir = workDir;
+ this.indexHelper = indexHelper;
+ this.initialCheckpoint = Objects.requireNonNull(initialCheckpoint);
+ this.finalCheckpoint = Objects.requireNonNull(finalCheckpoint);
+ }
+
+ public IncrementalStoreBuilder withPreferredPathElements(Set<String> preferredPathElements) {
+ this.preferredPathElements = preferredPathElements;
+ return this;
+ }
+
+ public IncrementalStoreBuilder withSortStrategyType(IncrementalStoreBuilder.IncrementalSortStrategyType sortStrategyType) {
+ this.sortStrategyType = sortStrategyType;
+ return this;
+ }
+
+ public IncrementalStoreBuilder withPathPredicate(Predicate<String> pathPredicate) {
+ this.pathPredicate = pathPredicate;
+ return this;
+ }
+
+ public IncrementalStoreBuilder withBlobStore(BlobStore blobStore) {
+ this.blobStore = blobStore;
+ return this;
+ }
+
+
+ public IncrementalStore build() throws IOException, CompositeException {
+ logFlags();
+ File dir = createStoreDir();
+
+ switch (sortStrategyType) {
+ case INCREMENTAL_FFS_STORE:
Review Comment:
this switch has only one case, I would replace it with a if for better readability
```suggestion
if (Objects.requireNonNull(sortStrategyType) == IncrementalSortStrategyType.INCREMENTAL_FFS_STORE) {
```
##########
oak-run-commons/src/test/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/MergeIncrementalFFSTest.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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;
+
+import org.apache.jackrabbit.oak.commons.Compression;
+import org.apache.jackrabbit.oak.index.indexer.document.incrementalstore.MergeIncrementalFlatFileStore;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+
+public class MergeIncrementalFFSTest {
+ private static final String BUILD_TARGET_FOLDER = "target";
+ private static final Compression algorithm = Compression.GZIP;
+ @Rule
+ public TemporaryFolder folder = new TemporaryFolder(new File(BUILD_TARGET_FOLDER));
+ @Test
+ public void test1() throws IOException {
+
+ File base = folder.newFile("base.gz");
+ File inc = folder.newFile("inc.gz");
+ File merged = folder.newFile("merged.gz");
+
+ try(BufferedWriter baseBW = FlatFileStoreUtils.createWriter(base, algorithm)) {
+ baseBW.write("/tmp|{prop1=\"foo\"}");
+ baseBW.newLine();
+ baseBW.write("/tmp/a|{prop2=\"foo\"}");
+ baseBW.newLine();
+ baseBW.write("/tmp/a/b|{prop3=\"foo\"}");
+ baseBW.newLine();
+ baseBW.write("/tmp/b|{prop1=\"foo\"}");
+ baseBW.newLine();
+ baseBW.write("/tmp/b/c|{prop2=\"foo\"}");
+ baseBW.newLine();
+ baseBW.write("/tmp/c|{prop3=\"foo\"}");
+ }
+
+ try(BufferedWriter baseInc = FlatFileStoreUtils.createWriter(inc, algorithm)) {
+ baseInc.write("/tmp/a|{prop2=\"fooModified\"}|r1|M");
+ baseInc.newLine();
+ baseInc.write("/tmp/b|{prop1=\"foo\"}|r1|D");
+ baseInc.newLine();
+ baseInc.write("/tmp/b/c/d|{prop2=\"fooNew\"}|r1|A");
+ baseInc.newLine();
+ baseInc.write("/tmp/c|{prop3=\"fooModified\"}|r1|M");
+ baseInc.newLine();
+ baseInc.write("/tmp/d|{prop3=\"bar\"}|r1|A");
+ baseInc.newLine();
+ baseInc.write("/tmp/e|{prop3=\"bar\"}|r1|A");
+ }
+
+ List<String> expectedList = new LinkedList<>();
+
+ expectedList.add("/tmp|{prop1=\"foo\"}");
+ expectedList.add("/tmp/a|{prop2=\"fooModified\"}");
+ expectedList.add("/tmp/a/b|{prop3=\"foo\"}");
+ expectedList.add("/tmp/b/c|{prop2=\"foo\"}");
+ expectedList.add("/tmp/b/c/d|{prop2=\"fooNew\"}");
+ expectedList.add("/tmp/c|{prop3=\"fooModified\"}");
+ expectedList.add("/tmp/d|{prop3=\"bar\"}");
+ expectedList.add("/tmp/e|{prop3=\"bar\"}");
+
+ MergeIncrementalFlatFileStore merge = new MergeIncrementalFlatFileStore(Collections.emptySet(), base, inc, merged, algorithm);
+
+ merge.doMerge();
+
+ try(BufferedReader br = FlatFileStoreUtils.createReader(merged, algorithm)) {
+ for (String line : expectedList) {
+ String actual = br.readLine();
+ System.out.println(actual);
+ Assert.assertEquals(line, actual);
+
+ }
+ Assert.assertEquals(null, br.readLine());
Review Comment:
use `assertNull()`
```suggestion
Assert.assertNull(br.readLine());
```
--
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: dev-unsubscribe@jackrabbit.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org