<commit_before>package se.l4.silo.engine.internal.query;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;

import org.h2.mvstore.MVMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.ImmutableList;

import se.l4.commons.io.Bytes;
import se.l4.silo.engine.MVStoreManager;
import se.l4.silo.engine.QueryEngine;
import se.l4.silo.engine.internal.DataEncounterImpl;
import se.l4.silo.engine.internal.StorageImpl;
import se.l4.silo.engine.types.LongFieldType;
import se.l4.silo.engine.types.StringFieldType;
import se.l4.silo.engine.types.VersionedType;

/**
 * Updater that ensures that {@link QueryEngine}s have been updated with
 * the correct data.
 * 
 * @author <NAME>
 *
 */
public class QueryEngineUpdater
{
	private static final Logger log = LoggerFactory.getLogger(QueryEngineUpdater.class);
	
	private static final Long DEFAULT = 0l;
	
	private final StorageImpl storage;
	private final String name;
	
	private final List<EngineDef> engines;
	private final MVMap<String, Long> state;

	public QueryEngineUpdater(MVStoreManager store,
			StorageImpl storage, 
			ScheduledExecutorService executor,
			String name,
			Map<String, QueryEngine<?>> engines)
	{
		this.storage = storage;
		this.name = name;
		
		ImmutableList.Builder<EngineDef> builder = ImmutableList.builder();
		for(Map.Entry<String, QueryEngine<?>> e : engines.entrySet())
		{
			builder.add(new EngineDef(e.getKey(), e.getValue()));
		}
		this.engines = builder.build();
		
		
		state = store.openMap("storage.query-engine." + name,
			StringFieldType.INSTANCE,
			VersionedType.singleVersion(LongFieldType.INSTANCE)
		);
		
		// Check the state of all indexes and start building new ones
		long latest = storage.getLatest();
		
		log.debug("Updater created for storage {}, latest entry in storage is {}", name, latest);
		
		for(EngineDef def : this.engines)
		{
			long current = state.getOrDefault(def.name, DEFAULT);
			
			if(log.isTraceEnabled())
			{
				log.trace("  " + def.name + " is currently at " + current + ", " + (current < latest ? "updating" : "skipping update"));
			}
			
			if(current < latest)
			{
				executor.submit(() -> updateEngine(def));
			}
		}
	}
	
	public void store(long previous, long id, Bytes bytes)
	{
		if(log.isTraceEnabled())
		{
			log.trace("[" + name + "] Store request with previous=" + previous + ", id=" + id);
		}
		
		for(EngineDef def : engines)
		{
			long previousForEngine = state.getOrDefault(def.name, DEFAULT);
			if(previousForEngine == 0l || previousForEngine == previous)
			{
				if(log.isTraceEnabled())
				{
					log.trace("[" + name + "] " + def.name + " is at " + previousForEngine + " and is up to date, updating");
				}
				
				// This query engine is up to date, continue indexing
				def.engine.update(id, new DataEncounterImpl(bytes));
				state.put(def.name, Math.max(previous, id));
			}
			else if(log.isTraceEnabled())
			{
				log.trace("[" + name + "] " + def.name + " is at " + previousForEngine + ", and is not up to date, skipping");
			}
		}
	}
	
	public void delete(long id)
	{
		if(log.isTraceEnabled())
		{
			log.trace("[" + name + "] Delete request for " + id);
		}
		
		for(EngineDef def : engines)
		{
			long latest = state.getOrDefault(def.name, DEFAULT);
			
			if(log.isTraceEnabled())
			{
				log.trace("[" + name + "] " + def.name + " is at " + latest + ", " + (latest>id ? "updating" : "skipping"));
			}
			
			if(latest > id)
			{
				/*
				 * If we have already indexed the item we can delete it
				 * otherwise do not nothing as it should not be indexed at
				 * all.
				 */
				def.engine.delete(id);<commit_msg>Fixing issue where latest data was not removed from query engines<commit_after>package se.l4.silo.engine.internal.query;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;

import org.h2.mvstore.MVMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.ImmutableList;

import se.l4.commons.io.Bytes;
import se.l4.silo.engine.MVStoreManager;
import se.l4.silo.engine.QueryEngine;
import se.l4.silo.engine.internal.DataEncounterImpl;
import se.l4.silo.engine.internal.StorageImpl;
import se.l4.silo.engine.types.LongFieldType;
import se.l4.silo.engine.types.StringFieldType;
import se.l4.silo.engine.types.VersionedType;

/**
 * Updater that ensures that {@link QueryEngine}s have been updated with
 * the correct data.
 * 
 * @author <NAME>
 *
 */
public class QueryEngineUpdater
{
	private static final Logger log = LoggerFactory.getLogger(QueryEngineUpdater.class);
	
	private static final Long DEFAULT = 0l;
	
	private final StorageImpl storage;
	private final String name;
	
	private final List<EngineDef> engines;
	private final MVMap<String, Long> state;

	public QueryEngineUpdater(MVStoreManager store,
			StorageImpl storage, 
			ScheduledExecutorService executor,
			String name,
			Map<String, QueryEngine<?>> engines)
	{
		this.storage = storage;
		this.name = name;
		
		ImmutableList.Builder<EngineDef> builder = ImmutableList.builder();
		for(Map.Entry<String, QueryEngine<?>> e : engines.entrySet())
		{
			builder.add(new EngineDef(e.getKey(), e.getValue()));
		}
		this.engines = builder.build();
		
		
		state = store.openMap("storage.query-engine." + name,
			StringFieldType.INSTANCE,
			VersionedType.singleVersion(LongFieldType.INSTANCE)
		);
		
		// Check the state of all indexes and start building new ones
		long latest = storage.getLatest();
		
		log.debug("Updater created for storage {}, latest entry in storage is {}", name, latest);
		
		for(EngineDef def : this.engines)
		{
			long current = state.getOrDefault(def.name, DEFAULT);
			
			if(log.isTraceEnabled())
			{
				log.trace("  " + def.name + " is currently at " + current + ", " + (current < latest ? "updating" : "skipping update"));
			}
			
			if(current < latest)
			{
				executor.submit(() -> updateEngine(def));
			}
		}
	}
	
	public void store(long previous, long id, Bytes bytes)
	{
		if(log.isTraceEnabled())
		{
			log.trace("[" + name + "] Store request with previous=" + previous + ", id=" + id);
		}
		
		for(EngineDef def : engines)
		{
			long previousForEngine = state.getOrDefault(def.name, DEFAULT);
			if(previousForEngine == 0l || previousForEngine == previous)
			{
				if(log.isTraceEnabled())
				{
					log.trace("[" + name + "] " + def.name + " is at " + previousForEngine + " and is up to date, updating");
				}
				
				// This query engine is up to date, continue indexing
				def.engine.update(id, new DataEncounterImpl(bytes));
				state.put(def.name, Math.max(previous, id));
			}
			else if(log.isTraceEnabled())
			{
				log.trace("[" + name + "] " + def.name + " is at " + previousForEngine + ", and is not up to date, skipping");
			}
		}
	}
	
	public void delete(long id)
	{
		if(log.isTraceEnabled())
		{
			log.trace("[" + name + "] Delete request for " + id);
		}
		
		for(EngineDef def : engines)
		{
			long latest = state.getOrDefault(def.name, DEFAULT);
			
			if(log.isTraceEnabled())
			{
				log.trace("[" + name + "] " + def.name + " is at " + latest + ", " + (latest>id ? "updating" : "skipping"));
			}
			
			if(latest >= id)
			{
				/*
				 * If we have already indexed the item we can delete it
				 * otherwise do not nothing as it should not be indexed at
				 * all.
				 */
				def.engine.delete(id);
{'fraction_non_alphanumeric': 0.1348148148148148, 'fraction_numerical': 0.003771043771043771, 'mean_word_length': 3.4123588829471183, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4157643', 'n_tokens_mistral': 2786, 'n_tokens_neox': 2399, 'n_words': 581}