thomasmueller commented on code in PR #1123:
URL: https://github.com/apache/jackrabbit-oak/pull/1123#discussion_r1331155586
##########
oak-run-commons/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/flatfile/StoreAndSortStrategy.java:
##########
@@ -42,39 +44,43 @@
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 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;
- 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;
}
@Override
public File createSortedStoreFile() throws IOException {
try (NodeStateEntryTraverser nodeStates = nodeStatesFactory.create(
new TraversingRange(new LastModifiedRange(0, Long.MAX_VALUE),
null))
) {
- File storeFile = writeToStore(nodeStates, storeDir,
getSortedStoreFileName(algorithm));
+ File storeFile = writeToStore(nodeStates, this.getStoreDir(),
getSortedStoreFileName(this.getAlgorithm()));
Review Comment:
this. shouldn't be needed here
##########
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:
I don't have a strong opinion; if you think it is better that way then let's
keep it.
--
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]
