diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/CompressedStateHolder.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/CompressedStateHolder.java new file mode 100644 index 0000000..84a6713 --- /dev/null +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/CompressedStateHolder.java @@ -0,0 +1,115 @@ +/* + * 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 java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.util.List; +import java.util.zip.Deflater; +import java.util.zip.DeflaterOutputStream; +import java.util.zip.InflaterInputStream; + +import org.apache.jackrabbit.oak.commons.StringUtils; + +import static com.google.common.base.Charsets.UTF_8; +import static com.google.common.collect.ImmutableList.copyOf; +import static org.apache.jackrabbit.oak.commons.PathUtils.elements; + +public class CompressedStateHolder implements NodeStateHolder { + private final List pathElements; + private final byte[] content; + private final boolean compressed; + + public CompressedStateHolder(String path, String line) { + this.pathElements = copyOf(elements(path)); + byte[] textBytes = line.getBytes(UTF_8); + byte[] compressed = compress(textBytes); + if (compressed.length < textBytes.length) { + this.compressed = true; + this.content = compressed; + } else { + this.compressed = false; + this.content = textBytes; + } + } + + @Override + public List getPathElements() { + return pathElements; + } + + /** + * Line here does not include the path + */ + @Override + public String getLine() { + return toString(compressed ? decompress(content) : content); + } + + @Override + public int getMemorySize() { + int memoryUsed = 0; + for (String e : pathElements) { + memoryUsed += StringUtils.estimateMemoryUsage(e); + } + memoryUsed += content.length; + return memoryUsed; + } + + static byte[] compress(byte[] bytes) { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (OutputStream out = newDeflaterStream(baos)) { + out.write(bytes); + } catch (IOException e) { + throw new RuntimeException(e); + } + return baos.toByteArray(); + } + + private static DeflaterOutputStream newDeflaterStream(ByteArrayOutputStream baos) { + return new DeflaterOutputStream(baos){ + { + def.setLevel(Deflater.BEST_SPEED); + } + }; + } + + static byte[] decompress(byte[] bytes) { + InputStream in = new InflaterInputStream(new ByteArrayInputStream(bytes)); + try { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + byte[] buffer = new byte[8192]; + int len; + while ((len = in.read(buffer)) > 0) { + baos.write(buffer, 0, len); + } + return baos.toByteArray(); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + private static String toString(byte[] bytes) { + return new String(bytes, UTF_8); + } +} diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateEntryHolder.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateEntryHolder.java deleted file mode 100644 index 731cf8b..0000000 --- oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateEntryHolder.java +++ /dev/null @@ -1,40 +0,0 @@ -/* - * 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 java.util.List; - -import static com.google.common.collect.ImmutableList.copyOf; -import static org.apache.jackrabbit.oak.commons.PathUtils.elements; -import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryWriter.getPath; - -class NodeStateEntryHolder { - final String line; - final List pathElements; - - public NodeStateEntryHolder(String line) { - this(getPath(line), line); - } - - public NodeStateEntryHolder(String path, String line) { - this.pathElements = copyOf(elements(path)); - this.line = line; - } -} diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateEntrySorter.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateEntrySorter.java index f15d8ed..7b0d17c 100644 --- oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateEntrySorter.java +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateEntrySorter.java @@ -89,9 +89,9 @@ public class NodeStateEntrySorter { log.info("Sorting with memory {} (estimated {})", humanReadableByteCount(memory), humanReadableByteCount(estimatedMemory)); Stopwatch w = Stopwatch.createStarted(); - Comparator comparator = (e1, e2) -> pathComparator.compare(e1.pathElements, e2.pathElements); - Function func1 = (line) -> line == null ? null : new NodeStateEntryHolder(line); - Function func2 = holder -> holder == null ? null : holder.line; + Comparator comparator = (e1, e2) -> pathComparator.compare(e1.getPathElements(), e2.getPathElements()); + Function func1 = (line) -> line == null ? null : new SimpleNodeStateHolder(line); + Function func2 = holder -> holder == null ? null : holder.getLine(); List sortedFiles = sortInBatch(memory, comparator, func1, func2); @@ -111,8 +111,8 @@ public class NodeStateEntrySorter { log.info("Sorting completed in {}", w); } - private void mergeSortedFiles(Comparator comparator, Function func1, - Function func2, List sortedFiles) throws IOException { + private void mergeSortedFiles(Comparator comparator, Function func1, + Function func2, List sortedFiles) throws IOException { try(BufferedWriter writer = createWriter(sortedFile, useZip)) { ExternalSort.mergeSortedFiles(sortedFiles, writer, @@ -127,9 +127,9 @@ public class NodeStateEntrySorter { } } - private List sortInBatch(long memory, Comparator comparator, - Function func1, - Function func2) throws IOException { + private List sortInBatch(long memory, Comparator comparator, + Function func1, + Function func2) throws IOException { if (useZip) { try (BufferedReader reader = createReader(nodeStateFile, useZip)) { return ExternalSort.sortInBatch(reader, diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateHolder.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateHolder.java new file mode 100644 index 0000000..ccfbb31 --- /dev/null +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/NodeStateHolder.java @@ -0,0 +1,31 @@ +/* + * 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 java.util.List; + +public interface NodeStateHolder { + + List getPathElements(); + + String getLine(); + + int getMemorySize(); +} diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/SimpleNodeStateHolder.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/SimpleNodeStateHolder.java new file mode 100644 index 0000000..4b6f220 --- /dev/null +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/SimpleNodeStateHolder.java @@ -0,0 +1,61 @@ +/* + * 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 java.util.List; + +import org.apache.jackrabbit.oak.commons.StringUtils; + +import static com.google.common.collect.ImmutableList.copyOf; +import static org.apache.jackrabbit.oak.commons.PathUtils.elements; +import static org.apache.jackrabbit.oak.index.indexer.document.flatfile.NodeStateEntryWriter.getPath; + +class SimpleNodeStateHolder implements NodeStateHolder{ + private final String line; + private final List pathElements; + + public SimpleNodeStateHolder(String line) { + this.pathElements = copyOf(elements(getPath(line))); + this.line = line; + } + + @Override + public List getPathElements() { + return pathElements; + } + + /** + * Line here may contain the path also + */ + @Override + public String getLine() { + return line; + } + + @Override + public int getMemorySize() { + int memoryUsed = 0; + for (String e : pathElements) { + memoryUsed += StringUtils.estimateMemoryUsage(e); + } + memoryUsed += StringUtils.estimateMemoryUsage(line); + return memoryUsed; + } +} diff --git oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/TraverseWithSortStrategy.java oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/TraverseWithSortStrategy.java index 5386e59..dac9403 100644 --- oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/TraverseWithSortStrategy.java +++ oak-run/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/TraverseWithSortStrategy.java @@ -66,7 +66,7 @@ class TraverseWithSortStrategy implements SortStrategy { private final File storeDir; private final boolean compressionEnabled; private final Charset charset = UTF_8; - private final Comparator comparator; + private final Comparator comparator; private NotificationEmitter emitter; private MemoryListener listener; private final int maxMemory = Integer.getInteger(OAK_INDEXER_MAX_SORT_MEMORY_IN_GB, OAK_INDEXER_MAX_SORT_MEMORY_IN_GB_DEFAULT); @@ -78,7 +78,7 @@ class TraverseWithSortStrategy implements SortStrategy { private long memoryUsed; private File sortWorkDir; private List sortedFiles = new ArrayList<>(); - private ArrayList entryBatch = new ArrayList<>(); + private ArrayList entryBatch = new ArrayList<>(); TraverseWithSortStrategy(Iterable nodeStates, PathElementComparator pathComparator, @@ -87,7 +87,7 @@ class TraverseWithSortStrategy implements SortStrategy { this.entryWriter = entryWriter; this.storeDir = storeDir; this.compressionEnabled = compressionEnabled; - this.comparator = (e1, e2) -> pathComparator.compare(e1.pathElements, e2.pathElements); + this.comparator = (e1, e2) -> pathComparator.compare(e1.getPathElements(), e2.getPathElements()); } @Override @@ -108,8 +108,8 @@ class TraverseWithSortStrategy implements SortStrategy { Stopwatch w = Stopwatch.createStarted(); File sortedFile = new File(storeDir, getSortedStoreFileName(compressionEnabled)); try(BufferedWriter writer = createWriter(sortedFile, compressionEnabled)) { - Function func1 = (line) -> line == null ? null : new NodeStateEntryHolder(line); - Function func2 = holder -> holder == null ? null : holder.line; + Function func1 = (line) -> line == null ? null : new SimpleNodeStateHolder(line); + Function func2 = holder -> holder == null ? null : holder.getLine(); ExternalSort.mergeSortedFiles(sortedFiles, writer, comparator, @@ -152,7 +152,7 @@ class TraverseWithSortStrategy implements SortStrategy { String jsonText = entryWriter.asJson(e.getNodeState()); //Here logic differs from NodeStateEntrySorter in sense that //Holder line consist only of json and not 'path|json' - NodeStateEntryHolder h = new NodeStateEntryHolder(e.getPath(), jsonText); + NodeStateHolder h = new CompressedStateHolder(e.getPath(), jsonText); entryBatch.add(h); updateMemoryUsed(h); @@ -173,9 +173,9 @@ class TraverseWithSortStrategy implements SortStrategy { File newtmpfile = File.createTempFile("sortInBatch", "flatfile", sortWorkDir); long textSize = 0; try (BufferedWriter writer = FlatFileStoreUtils.createWriter(newtmpfile, compressionEnabled)) { - for (NodeStateEntryHolder h : entryBatch) { + for (NodeStateHolder h : entryBatch) { //Here holder line only contains nodeState json - String text = entryWriter.toString(h.pathElements, h.line); + String text = entryWriter.toString(h.getPathElements(), h.getLine()); writer.write(text); writer.newLine(); textSize += text.length() + 1; @@ -193,11 +193,8 @@ class TraverseWithSortStrategy implements SortStrategy { return !sufficientMemory.get(); } - private void updateMemoryUsed(NodeStateEntryHolder h) { - for (String e : h.pathElements) { - memoryUsed += StringUtils.estimateMemoryUsage(e); - } - memoryUsed += StringUtils.estimateMemoryUsage(h.line); + private void updateMemoryUsed(NodeStateHolder h) { + memoryUsed += h.getMemorySize(); } private static File createdSortWorkDir(File storeDir) throws IOException {