Skip to content

fix: update matching-stats don't resolve unneeded objects #3824

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
thoniTUB merged 4 commits into from
Jan 6, 2026
Merged

fix: update matching-stats don't resolve unneeded objects #3824

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
579982c
update matching-stats don't resolve unneeded objects
thoniTUB Dec 10, 2025
990d01a
Merge branch 'develop' into feat/improve-matching-stats
thoniTUB Dec 15, 2025
8a267e9
Merge branch 'develop' into feat/improve-matching-stats
thoniTUB Jan 5, 2026
411e14e
Merge branch 'develop' into feat/improve-matching-stats
thoniTUB Jan 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 21 additions & 18 deletions .../com/bakdata/conquery/models/messages/namespaces/specific/UpdateMatchingStatsMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import com.bakdata.conquery.models.datasets.concepts.tree.TreeConcept;
import com.bakdata.conquery.models.events.Bucket;
import com.bakdata.conquery.models.events.CBlock;
import com.bakdata.conquery.models.identifiable.ids.specific.CBlockId;
import com.bakdata.conquery.models.identifiable.ids.specific.ConceptElementId;
import com.bakdata.conquery.models.identifiable.ids.specific.ConceptId;
import com.bakdata.conquery.models.jobs.Job;
Expand Down Expand Up @@ -73,13 +74,9 @@ public void execute() throws Exception {
subJobs =
concepts.stream()
.collect(Collectors.toMap(Functions.identity(),
concept -> CompletableFuture.runAsync(() -> {
final Concept<?> resolved = concept.resolve();
final Map<ConceptElementId<?>, MatchingStats.Entry>
matchingStats =
new HashMap<>(resolved.countElements());
conceptId -> CompletableFuture.runAsync(() -> {

calculateConceptMatches(resolved, matchingStats, worker);
Map<ConceptElementId<?>, MatchingStats.Entry> matchingStats = calculateConceptMatches(conceptId, worker);

final WriteFuture writeFuture = worker.send(new UpdateElementMatchingStats(worker.getInfo().getId(), matchingStats));

Expand Down Expand Up @@ -127,18 +124,24 @@ public String getLabel() {
return String.format("Calculate Matching Stats for %s", worker.getInfo().getDataset());
}

private static void calculateConceptMatches(Concept<?> concept, Map<ConceptElementId<?>, MatchingStats.Entry> results, Worker worker) {
log.debug("BEGIN calculating for `{}`", concept.getId());
private static Map<ConceptElementId<?>, MatchingStats.Entry> calculateConceptMatches(ConceptId conceptId, Worker worker) {

try(Stream<CBlock> allCBlocks = worker.getStorage().getAllCBlocks()) {
Copy link
Collaborator Author

@thoniTUB thoniTUB Dec 10, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@awildturtok das getAllCBlocks() habe ich durch ein getAllCBlockIds() ersetzt

Copy link
Collaborator

@awildturtok awildturtok Jan 5, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ist das der Teil von dem du dir die Performance gains erhoffst?

Ich bin mir nicht sicher, was es erreichen soll. Du verschiebst die Operation ja nur. Wobei du vlt durch die äußere Concurrency eine bessere IO Verteilung hast?

Copy link
Collaborator Author

@thoniTUB thoniTUB Jan 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bisher sind wir für jedes Concept über alle CBlocks gelaufen nur um hier die meisten zu skippen.

Jetzt skippen wir über die meisten Ids und resolven nur die CBlocks die auch zu den Connectoren des Concepts passen für das gerade die matching Stats laufen. Da das alles async/parallel ausgeführt wird hatten wir vorher auch nicht umbedingt einen seriellen IO weil jeder Task zu unterschiedlicher Zeit den Storage von vorne resolved hat.

Um den IO seriell zu machen müssten wir nicht in der äußersten Loop/Stream über die Concepts laufen sondern über die Cblocks. Dann könnten wir aber erst beim letzten CBlock die ersten MatchingStats zurückliefern.

Copy link
Collaborator

@awildturtok awildturtok Jan 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ahhh, ich hatte nicht gesehen, dass das in einer schleife ist. Super

Concept<?> concept = conceptId.resolve();

final Map<ConceptElementId<?>, MatchingStats.Entry> matchingStats = new HashMap<>(concept.countElements());

log.debug("BEGIN calculating for `{}`", conceptId);

try (Stream<CBlockId> allCBlocks = worker.getStorage().getAllCBlockIds()) {

for (CBlock cBlock : allCBlocks.toList()) {
for (CBlockId cBlockId : allCBlocks.toList()) {

if (!cBlock.getConnector().getConcept().equals(concept.getId())) {
if (!cBlockId.getConnector().getConcept().equals(conceptId)) {
continue;
}

try {
CBlock cBlock = cBlockId.resolve();
final Bucket bucket = cBlock.getBucket().resolve();
final Table table = bucket.getTable().resolve();

Expand All @@ -152,7 +155,7 @@ private static void calculateConceptMatches(Concept<?> concept, Map<ConceptEleme


if (!(concept instanceof TreeConcept) || localIds == null) {
results.computeIfAbsent(concept.getId(), (ignored) -> new MatchingStats.Entry()).addEvent(table, bucket, event, entity);
matchingStats.computeIfAbsent(conceptId, (ignored) -> new MatchingStats.Entry()).addEvent(table, bucket, event, entity);
continue;
}

Expand All @@ -163,24 +166,24 @@ private static void calculateConceptMatches(Concept<?> concept, Map<ConceptEleme
ConceptElement<?> element = ((TreeConcept) concept).getElementByLocalIdPath(localIds);

while (element != null) {
results.computeIfAbsent(element.getId(), (ignored) -> new MatchingStats.Entry())
.addEvent(table, bucket, event, entity);
matchingStats.computeIfAbsent(element.getId(), (ignored) -> new MatchingStats.Entry())
.addEvent(table, bucket, event, entity);
element = element.getParent();
}
}
}

}
catch (Exception e) {
log.error("Failed to collect the matching stats for {}", cBlock, e);
log.error("Failed to collect the matching stats for {}", cBlockId, e);
}
}
}

log.trace("DONE calculating for `{}`", concept.getId());
log.trace("DONE calculating for `{}`", conceptId);

return matchingStats;
}

}


}
Loading