matching stats as join table

backend/src/main/java/com/bakdata/conquery/mode/cluster/ClusterStorageListener.java

@@ -18,49 +18,43 @@
 import com.bakdata.conquery.models.worker.DatasetRegistry;
 import com.bakdata.conquery.models.worker.DistributedNamespace;
 import com.bakdata.conquery.models.worker.WorkerHandler;
-import lombok.AllArgsConstructor;
 
 /**
  * Propagates changes of stored entities to relevant ConQuery shards in the cluster.
  */
-@AllArgsConstructor
-public
-class ClusterStorageListener implements StorageListener {
-
-	private final JobManager jobManager;
-	private final DatasetRegistry<DistributedNamespace> datasetRegistry;
+public record ClusterStorageListener(JobManager jobManager, DatasetRegistry<DistributedNamespace> datasetRegistry) implements StorageListener {
 
 	@Override
 	public void onAddSecondaryId(SecondaryIdDescription secondaryId) {
-		datasetRegistry.get(secondaryId.getDataset()).getWorkerHandler().sendToAll(new UpdateSecondaryId(secondaryId));
+		datasetRegistry().get(secondaryId.getDataset()).getWorkerHandler().sendToAll(new UpdateSecondaryId(secondaryId));
 	}
 
 	@Override
 	public void onDeleteSecondaryId(SecondaryIdDescriptionId secondaryId) {
-		datasetRegistry.get(secondaryId.getDataset()).getWorkerHandler().sendToAll(new RemoveSecondaryId(secondaryId));
+		datasetRegistry().get(secondaryId.getDataset()).getWorkerHandler().sendToAll(new RemoveSecondaryId(secondaryId));
 	}
 
 	@Override
 	public void onAddTable(Table table) {
-		datasetRegistry.get(table.getDataset()).getWorkerHandler().sendToAll(new UpdateTable(table));
+		datasetRegistry().get(table.getDataset()).getWorkerHandler().sendToAll(new UpdateTable(table));
 	}
 
 	@Override
 	public void onRemoveTable(TableId table) {
-		datasetRegistry.get(table.getDataset()).getWorkerHandler().sendToAll(new RemoveTable(table));
+		datasetRegistry().get(table.getDataset()).getWorkerHandler().sendToAll(new RemoveTable(table));
 	}
 
 	@Override
 	public void onAddConcept(Concept<?> concept) {
-		WorkerHandler handler = datasetRegistry.get(concept.getDataset()).getWorkerHandler();
+		WorkerHandler handler = datasetRegistry().get(concept.getDataset()).getWorkerHandler();
 		SimpleJob simpleJob = new SimpleJob(String.format("sendToAll : Add %s ", concept.getId()), () -> handler.sendToAll(new UpdateConcept(concept)));
-		jobManager.addSlowJob(simpleJob);
+		jobManager().addSlowJob(simpleJob);
 	}
 
 	@Override
 	public void onDeleteConcept(ConceptId concept) {
-		WorkerHandler handler = datasetRegistry.get(concept.getDataset()).getWorkerHandler();
+		WorkerHandler handler = datasetRegistry().get(concept.getDataset()).getWorkerHandler();
 		SimpleJob simpleJob = new SimpleJob("sendToAll: remove " + concept, () -> handler.sendToAll(new RemoveConcept(concept)));
-		jobManager.addSlowJob(simpleJob);
+		jobManager().addSlowJob(simpleJob);
 	}
 }

backend/src/main/java/com/bakdata/conquery/mode/local/LocalManagerProvider.java

@@ -12,6 +12,7 @@
 import com.bakdata.conquery.mode.NamespaceHandler;
 import com.bakdata.conquery.mode.cluster.InternalMapperFactory;
 import com.bakdata.conquery.models.config.ConqueryConfig;
+import com.bakdata.conquery.models.jobs.JobManager;
 import com.bakdata.conquery.models.worker.DatasetRegistry;
 import com.bakdata.conquery.models.worker.LocalNamespace;
 import com.bakdata.conquery.models.worker.ShardNodeInformation;
@@ -29,6 +30,8 @@ public DelegateManager<LocalNamespace> provideManager(ConqueryConfig config, Env
 
 		final ConnectionManager connectionManager = config.getSqlConnectorConfig().toConnectionManager(environment);
 
+		final JobManager jobManager = ManagerProvider.newJobManager(config);
+
 		final MetaStorage storage = new MetaStorage(config.getStorage());
 		final InternalMapperFactory internalMapperFactory = new InternalMapperFactory(config, environment.getValidator());
 		final NamespaceHandler<LocalNamespace> namespaceHandler = new LocalNamespaceHandler(config, internalMapperFactory, connectionManager, clock);
@@ -40,7 +43,7 @@ public DelegateManager<LocalNamespace> provideManager(ConqueryConfig config, Env
 				datasetRegistry,
 				storage,
 				new FailingImportHandler(),
-				new LocalStorageListener(),
+				new LocalStorageListener(jobManager, datasetRegistry),
 				EMPTY_NODE_PROVIDER,
 				List.of(),
 				internalMapperFactory,

backend/src/main/java/com/bakdata/conquery/mode/local/LocalNamespaceHandler.java

@@ -13,6 +13,7 @@
 import com.bakdata.conquery.models.worker.DatasetRegistry;
 import com.bakdata.conquery.models.worker.LocalNamespace;
 import com.bakdata.conquery.sql.conquery.SqlExecutionManager;
+import com.bakdata.conquery.sql.conquery.SqlMatchingStats;
 import com.bakdata.conquery.sql.conversion.NodeConversions;
 import com.bakdata.conquery.sql.conversion.SqlConverter;
 import com.bakdata.conquery.sql.conversion.dialect.DialectBundle;
@@ -62,7 +63,8 @@ public LocalNamespace createNamespace(
 				dslContext, sqlStorageHandler,
 				namespaceData.jobManager(),
 				namespaceData.filterSearch(),
-				sqlEntityResolver
+				sqlEntityResolver,
+				connection.getConnection()
 		);
 	}
 

backend/src/main/java/com/bakdata/conquery/mode/local/LocalStorageListener.java

@@ -4,13 +4,15 @@
 import com.bakdata.conquery.models.datasets.SecondaryIdDescription;
 import com.bakdata.conquery.models.datasets.Table;
 import com.bakdata.conquery.models.datasets.concepts.Concept;
+import com.bakdata.conquery.models.datasets.concepts.tree.TreeConcept;
 import com.bakdata.conquery.models.identifiable.ids.specific.ConceptId;
 import com.bakdata.conquery.models.identifiable.ids.specific.SecondaryIdDescriptionId;
 import com.bakdata.conquery.models.identifiable.ids.specific.TableId;
-import lombok.Data;
+import com.bakdata.conquery.models.jobs.JobManager;
+import com.bakdata.conquery.models.worker.DatasetRegistry;
+import com.bakdata.conquery.models.worker.LocalNamespace;
 
-@Data
-public class LocalStorageListener implements StorageListener {
+public record LocalStorageListener(JobManager jobManager, DatasetRegistry<LocalNamespace> datasetRegistry) implements StorageListener {
 
 	@Override
 	public void onAddSecondaryId(SecondaryIdDescription secondaryId) {
@@ -31,9 +33,13 @@ public void onRemoveTable(TableId table) {
 
 	@Override
 	public void onAddConcept(Concept<?> concept) {
+		LocalNamespace namespace = datasetRegistry().get(concept.getDataset());
+		namespace.getMatchingStats().createConceptIdJoinTable((TreeConcept) concept);
 	}
 
 	@Override
 	public void onDeleteConcept(ConceptId concept) {
+		LocalNamespace namespace = datasetRegistry().get(concept.getDataset());
+		namespace.getMatchingStats().deleteConceptIdJoinTable(concept);
 	}
 }

backend/src/main/java/com/bakdata/conquery/mode/local/ManagedConnection.java

@@ -1,7 +1,6 @@
 package com.bakdata.conquery.mode.local;
 
 import java.sql.SQLException;
-
 import javax.annotation.CheckForNull;
 
 import com.bakdata.conquery.models.config.DatabaseConnectionConfig;

backend/src/main/java/com/bakdata/conquery/mode/local/UpdateMatchingStatsSqlJob.java

@@ -0,0 +1,78 @@
+package com.bakdata.conquery.mode.local;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.*;
+
+import com.bakdata.conquery.models.datasets.Dataset;
+import com.bakdata.conquery.models.datasets.concepts.Concept;
+import com.bakdata.conquery.models.datasets.concepts.tree.TreeConcept;
+import com.bakdata.conquery.models.jobs.Job;
+import com.bakdata.conquery.sql.conquery.SqlMatchingStats;
+import com.google.common.base.Stopwatch;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import lombok.Data;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.builder.ToStringExclude;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+@Slf4j
+@Data
+public class UpdateMatchingStatsSqlJob extends Job {
+
+    @ToString.Exclude
+    private final List<Concept<?>> concepts;
+    private final Dataset dataset;
+
+    @ToString.Exclude
+    private final SqlMatchingStats matchingStats;
+
+
+    @Override
+    public void execute() throws Exception {
+
+        log.info("BEGIN collecting SQL matching stats for {}", dataset);
+
+        Stopwatch stopwatch = Stopwatch.createStarted();
+
+        ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newVirtualThreadPerTaskExecutor());
+
+        List<ListenableFuture<?>> jobs = new ArrayList<>();
+
+        for (Concept<?> concept : concepts) {
+            if (!(concept instanceof TreeConcept)) {
+                continue;
+            }
+            jobs.add(matchingStats.collectMatchingStatsForConcept((TreeConcept) concept, executorService));
+        }
+
+        ListenableFuture<List<@Nullable Object>> all = Futures.allAsList(jobs);
+
+        while (!all.isDone()) {
+            if (isCancelled()) {
+                all.cancel(true);
+                log.debug("CANCELLED update matching stats for {}", getDataset(), all.exceptionNow());
+                return;
+            }
+
+            all.get(5, TimeUnit.SECONDS);
+            log.trace("WAITING for matching stats to finish {}", getDataset());
+
+            if (all.state().equals(Future.State.FAILED)) {
+                log.error("FAILED update matching stats for {}", getDataset(), all.exceptionNow());
+                return;
+            }
+        }
+
+        log.debug("DONE collecting SQL matching stats for {} within {}", dataset, stopwatch);
+    }
+
+    @Override
+    public String getLabel() {
+        return "Collect matching stats for %s (%s concepts)".formatted(dataset.getName(), concepts.size());
+    }
+}