thomasmueller commented on code in PR #1123:
URL: https://github.com/apache/jackrabbit-oak/pull/1123#discussion_r1328472985
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/MultithreadedTraverseWithSortStrategy.java:
##########
@@ -234,18 +230,42 @@ public int getValue() {
* previous runs stopped). If this is not null
and not empty, the {@code lastModifiedBreakPoints} parameter is ignored.
* @param algorithm string representation of the compression algorithm,
use "none" for disable compression.
*/
+ @Deprecated
MultithreadedTraverseWithSortStrategy(NodeStateEntryTraverserFactory
nodeStateEntryTraverserFactory,
List<Long> lastModifiedBreakPoints,
PathElementComparator pathComparator,
BlobStore blobStore, File storeDir,
List<File> existingDataDumpDirs,
Compression algorithm, MemoryManager
memoryManager, long dumpThreshold,
Predicate<String> pathPredicate)
throws IOException {
- this.storeDir = storeDir;
+ super(storeDir, algorithm, pathPredicate, null, null);
this.mergeDir = new File(storeDir, mergeDirName);
- this.algorithm = algorithm;
this.sortedFiles = new LinkedBlockingQueue<>();
this.throwables = new ConcurrentLinkedQueue<>();
this.comparator = (e1, e2) ->
pathComparator.compare(e1.getPathElements(), e2.getPathElements());
- this.pathPredicate = pathPredicate;
+ taskQueue = new LinkedBlockingQueue<>();
+ phaser = new Phaser() {
+ @Override
+ protected boolean onAdvance(int phase, int registeredParties) {
+ //terminate phaser only if it is advancing from last phase and
there are no registered parties
+ return phase == Phases.WAITING_FOR_RESULTS.value &&
registeredParties == 0;
+ }
+ };
+ // arrives on final file sorted
+ mergePhaser = new Phaser(1);
Review Comment:
(It's a bit sad that we have to copy&paste code here... if it can be avoided
easily, that would be great. If not, then OK: anyway we want to retire + remove
MultithreadedTraverseWithSortStrategy soon.)
##########
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) {
Review Comment:
I think it should be a public method, so we can add a few unit tests.
##########
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) {
Review Comment:
I think the JSON can contain the delimiter (NodeStateEntryWriter.DELIMITER).
No?
The path can not. But the JSON, I think it can.
So we would have to get the first delimiter from the left. And the remaining
2 delimiter from the right (from end to beginning).
--
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]
