use caffeine in CachedStore and adapt ManagedInternalForm for state p…

backend/src/main/java/com/bakdata/conquery/apiv1/QueryProcessor.java

@@ -369,9 +369,12 @@ public FullExecutionStatus getSingleEntityExport(Subject subject, UriBuilder uri
 			throw new ConqueryError.ExecutionProcessingError();
 		}
 
+		// Workaround update our execution as the lastresultcount was set in the background
+		final EntityPreviewExecution executionFinished = (EntityPreviewExecution) execution.getId().resolve();
+		executionFinished.initExecutable(config);
 
 		final FullExecutionStatus status = execution.buildStatusFull(subject, namespace);
-		status.setResultUrls(getResultAssets(config.getResultProviders(), execution, uriBuilder, false));
+		status.setResultUrls(getResultAssets(config.getResultProviders(), executionFinished, uriBuilder, false));
 		return status;
 	}
 
@@ -571,7 +574,7 @@ public Stream<Map<String, String>> resolveEntities(Subject subject, List<FilterV
 					 .filter(Predicate.not(Map::isEmpty));
 	}
 
-	public ResultStatistics getResultStatistics(SingleTableResult managedQuery) {
+	public <E extends ManagedExecution & SingleTableResult> ResultStatistics getResultStatistics(E managedQuery) {
 
 		final Locale locale = I18n.LOCALE.get();
 		final NumberFormat decimalFormat = NumberFormat.getNumberInstance(locale);
@@ -584,6 +587,8 @@ public ResultStatistics getResultStatistics(SingleTableResult managedQuery) {
 				new PrintSettings(true, locale, managedQuery.getNamespace(), config, null, null, decimalFormat, integerFormat);
 		final UniqueNamer uniqueNamer = new UniqueNamer(printSettings);
 
+		managedQuery.initExecutable(config);
+
 		final List<ResultInfo> resultInfos = managedQuery.getResultInfos();
 
 		final Optional<ResultInfo>

backend/src/main/java/com/bakdata/conquery/apiv1/query/Query.java

@@ -6,12 +6,9 @@
 import java.util.stream.Stream;
 
 import com.bakdata.conquery.io.storage.MetaStorage;
-import com.bakdata.conquery.models.execution.ExecutionState;
-import com.bakdata.conquery.models.execution.ManagedExecution;
 import com.bakdata.conquery.models.identifiable.ids.specific.DatasetId;
 import com.bakdata.conquery.models.identifiable.ids.specific.ManagedExecutionId;
 import com.bakdata.conquery.models.identifiable.ids.specific.UserId;
-import com.bakdata.conquery.models.query.ExecutionManager;
 import com.bakdata.conquery.models.query.ManagedQuery;
 import com.bakdata.conquery.models.query.QueryPlanContext;
 import com.bakdata.conquery.models.query.QueryResolveContext;
@@ -27,8 +24,6 @@ public abstract class Query implements QueryDescription {
 
 	public abstract QueryPlan<?> createQueryPlan(QueryPlanContext context);
 
-	public abstract void collectRequiredQueries(Set<ManagedExecutionId> requiredQueries);
-
 	@Override
 	public abstract void resolve(QueryResolveContext context);
 
@@ -38,6 +33,8 @@ public Set<ManagedExecutionId> collectRequiredQueries() {
 		return set;
 	}
 
+	public abstract void collectRequiredQueries(Set<ManagedExecutionId> requiredQueries);
+
 	@JsonIgnore
 	public abstract List<ResultInfo> getResultInfos();
 
@@ -59,7 +56,6 @@ public CQElement getReusableComponents() {
 	 *
 	 * @param results
 	 * @return the number of results in the result List.
-	 * @see ManagedExecution#finish(ExecutionState, ExecutionManager)  for how it's used.
 	 */
 	public long countResults(Stream<EntityResult> results) {
 		return results.map(EntityResult::listResultLines)

backend/src/main/java/com/bakdata/conquery/io/result/csv/ResultCsvProcessor.java

@@ -53,6 +53,9 @@ public <E extends ManagedExecution & SingleTableResult> Response createResult(Su
 		// Check if subject is permitted to download on all datasets that were referenced by the query
 		authorizeDownloadDatasets(subject, exec);
 
+		// Initialize execution so columns can be correctly accounted
+		exec.initExecutable(config);
+
 		final IdPrinter idPrinter = IdColumnUtil.getIdPrinter(subject, exec, namespace, config.getIdColumns().getIds());
 
 		// Get the locale extracted by the LocaleFilter

backend/src/main/java/com/bakdata/conquery/io/storage/xodus/stores/CachedStore.java

@@ -1,71 +1,73 @@
 package com.bakdata.conquery.io.storage.xodus.stores;
 
 import java.io.IOException;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.stream.Stream;
 
 import com.bakdata.conquery.io.jackson.serializer.IdReferenceResolvingException;
 import com.bakdata.conquery.io.storage.Store;
 import com.bakdata.conquery.io.storage.xodus.stores.SerializingStore.IterationStatistic;
 import com.bakdata.conquery.util.io.ProgressBar;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.CaffeineSpec;
+import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.google.common.base.Stopwatch;
 import com.jakewharton.byteunits.BinaryByteUnit;
 import lombok.RequiredArgsConstructor;
+import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
 
 @RequiredArgsConstructor
 @Slf4j
+@ToString(onlyExplicitlyIncluded = true)
 public class CachedStore<KEY, VALUE> implements Store<KEY, VALUE> {
 
 	private static final ProgressBar PROGRESS_BAR = new ProgressBar(0, System.out);
 
-	private ConcurrentHashMap<KEY, VALUE> cache = new ConcurrentHashMap<>();
+	private final LoadingCache<KEY, VALUE> cache;
+	@ToString.Include
 	private final Store<KEY, VALUE> store;
 
+	public CachedStore(Store<KEY, VALUE> store, CaffeineSpec caffeineSpec) {
+		this.store = store;
+		cache = Caffeine.from(caffeineSpec)
+//						.recordStats(() -> new MetricsStatsCounter(metricRegistry, "cache."+store.toString()))
+						.build(this.store::get);
+	}
+
 	@Override
 	public void add(KEY key, VALUE value) {
-		if (cache.putIfAbsent(key, value) != null) {
-			throw new IllegalStateException("The id " + key + " is already part of this store");
-		}
-		store.add(key, value);
+		// We don't distinguish between add and update on this layer. Let a deeper layer complain
+		update(key, value);
 	}
 
 	@Override
-	public VALUE get(KEY key) {
-		return cache.computeIfAbsent(key, store::get);
+	public void update(KEY key, VALUE value) {
+		store.update(key, value);
+		cache.invalidate(key);
 	}
 
 	@Override
-	public IterationStatistic forEach(StoreEntryConsumer<KEY, VALUE> consumer) {
-		return store.forEach(consumer);
+	public VALUE get(KEY key) {
+		return cache.get(key);
 	}
 
 	@Override
-	public void update(KEY key, VALUE value) {
-		cache.put(key, value);
-		store.update(key, value);
+	public IterationStatistic forEach(StoreEntryConsumer<KEY, VALUE> consumer) {
+		store.getAllKeys().forEach( k -> consumer.accept(k, cache.get(k), 0 /*Leaky?*/));
+		return null;
 	}
 
 	@Override
 	public void remove(KEY key) {
-		cache.remove(key);
 		store.remove(key);
-	}
-
-	@Override
-	public int count() {
-		if (cache.isEmpty()) {
-			return store.count();
-		}
-		return cache.size();
+		cache.invalidate(key);
 	}
 
 	@Override
 	public void loadData() {
 		final LongAdder totalSize = new LongAdder();
 		final int count = count();
-		cache = new ConcurrentHashMap<>(count);
 		final ProgressBar bar;
 
 		if (count > 100) {
@@ -108,36 +110,37 @@ public void loadData() {
 		});
 		log.debug("\tloaded store {}: {} entries, {} within {}",
 				this,
-				cache.values().size(),
+			  	count,
 				BinaryByteUnit.format(totalSize.sum()),
 				timer.stop()
 		);
 	}
 
 	@Override
-	public Stream<VALUE> getAll() {
-		return cache.values().stream();
+	public int count() {
+		return store.count();
 	}
 
 	@Override
-	public String toString() {
-		return "cached " + store.toString();
+	public Stream<VALUE> getAll() {
+		return store.getAllKeys().map(cache::get);
 	}
 
 	@Override
 	public Stream<KEY> getAllKeys() {
-		return cache.keySet().stream();
+		return store.getAllKeys();
 	}
 
 	@Override
 	public void clear() {
-		cache.clear();
 		store.clear();
+		cache.invalidateAll();
 	}
 
 	@Override
 	public void removeStore() {
 		store.removeStore();
+		cache.invalidateAll();
 	}
 
 	@Override

backend/src/main/java/com/bakdata/conquery/io/storage/xodus/stores/XodusStore.java

@@ -14,6 +14,7 @@
 import jetbrains.exodus.env.Store;
 import jetbrains.exodus.env.StoreConfig;
 import lombok.Getter;
+import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
@@ -25,6 +26,7 @@ public class XodusStore {
 	private final Consumer<XodusStore> storeCloseHook;
 	private final Consumer<XodusStore> storeRemoveHook;
 	@Getter
+	@ToString.Include
 	private final String name;
 
 	public XodusStore(Environment env, String name, Consumer<XodusStore> storeCloseHook, Consumer<XodusStore> storeRemoveHook) {

backend/src/main/java/com/bakdata/conquery/models/auth/basic/LocalAuthenticationRealm.java

@@ -26,6 +26,7 @@
 import com.bakdata.conquery.models.identifiable.ids.specific.UserId;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.github.benmanes.caffeine.cache.CaffeineSpec;
 import com.google.common.collect.ImmutableList;
 import com.password4j.HashingFunction;
 import com.password4j.Password;
@@ -66,19 +67,16 @@ public class LocalAuthenticationRealm extends AuthenticatingRealm implements Con
 
 	private final XodusConfig passwordStoreConfig;
 	private final String storeName;
-
-	@JsonIgnore
-	private Environment passwordEnvironment;
-	@JsonIgnore
-	private Store<UserId, HashEntry> passwordStore;
-
 	@JsonIgnore
 	private final ConqueryTokenRealm centralTokenRealm;
 	private final Duration validDuration;
 	private final Validator validator;
 	private final ObjectMapper mapper;
-
 	private final HashingFunction defaultHashingFunction;
+	@JsonIgnore
+	private Environment passwordEnvironment;
+	@JsonIgnore
+	private Store<UserId, HashEntry> passwordStore;
 
 	//////////////////// INITIALIZATION ////////////////////
 
@@ -116,7 +114,9 @@ protected void onInit() {
 						false,
 						true,
 						null, Executors.newSingleThreadExecutor()
-				));
+				),
+				CaffeineSpec.parse("") // TODO add configuration
+				);
 	}
 
 	//////////////////// AUTHENTICATION ////////////////////
@@ -145,14 +145,31 @@ public String createAccessToken(String username, String password) {
 			throw new CredentialsException("No password hash was found for user: " + username);
 		}
 
-		final String hash = hashedEntry.getHash();
+		final String hash = hashedEntry.hash();
 		if (!Password.check(password.getBytes(), hash.getBytes()).with(PasswordHelper.getHashingFunction(hash))) {
 			throw new IncorrectCredentialsException("Password was was invalid for user: " + userId);
 		}
 
 		return centralTokenRealm.createTokenForUser(userId, validDuration);
 	}
 
+	@Override
+	public boolean addUser(@NonNull User user, @NonNull CredentialType credential) {
+
+		try {
+			final HashEntry hashEntry = toHashEntry(credential);
+			passwordStore.add(user.getId(), hashEntry);
+			log.debug("Added user to realm: {}", user.getId());
+			return true;
+		}
+		catch (IllegalArgumentException e) {
+			log.warn("Unable to add user '{}'", user.getId(), e);
+		}
+		return false;
+	}
+
+	//////////////////// USER MANAGEMENT ////////////////////
+
 	/**
 	 * Converts the provided password to a Xodus compatible hash.
 	 */
@@ -171,22 +188,6 @@ else if (credential instanceof PasswordHashCredential passwordHashCredential) {
 		throw new IllegalArgumentException("CredentialType not supported yet: " + credential.getClass());
 	}
 
-	//////////////////// USER MANAGEMENT ////////////////////
-
-	@Override
-	public boolean addUser(@NonNull User user, @NonNull CredentialType credential) {
-
-		try {
-			final HashEntry hashEntry = toHashEntry(credential);
-			passwordStore.add(user.getId(), hashEntry);
-			return true;
-		}
-		catch (IllegalArgumentException e) {
-			log.warn("Unable to add user '{}'", user.getId(), e);
-		}
-		return false;
-	}
-
 	@Override
 	public boolean updateUser(User user, CredentialType credential) {
 

backend/src/main/java/com/bakdata/conquery/models/auth/basic/PasswordHasher.java

@@ -1,6 +1,5 @@
 package com.bakdata.conquery.models.auth.basic;
 
-import lombok.Data;
 import lombok.experimental.UtilityClass;
 import lombok.extern.slf4j.Slf4j;
 
@@ -13,12 +12,10 @@
 @Slf4j
 public class PasswordHasher {
 
-	@Data
 	/**
 	 * Container class for the entries in the store consisting of the salted password hash and the corresponding salt.
 	 */
-	public static class HashEntry {
-		final String hash;
+	public record HashEntry(String hash) {
 	}
 
 }