content: package io.activej.cube;
  
  import io.activej.cube.CubeStructure.AggregationConfig;
  import io.activej.cube.aggregation.AggregationChunk;
  import io.activej.cube.aggregation.PrimaryKey;
  import io.activej.cube.aggregation.ot.AggregationDiff;
  import io.activej.cube.aggregation.predicate.AggregationPredicate;
  import io.activej.cube.aggregation.predicate.AggregationPredicates;
  import io.activej.cube.ot.CubeDiff;
  import io.activej.etl.LogDiff;
  import io.activej.etl.LogState;
  import io.activej.fs.FileSystem;
  import io.activej.ot.StateManager;
  import org.junit.Before;
  import org.junit.Test;
  
  import java.nio.file.Files;
  import java.nio.file.Path;
  import java.time.LocalDate;
  import java.time.Month;
  import java.util.*;
  
  import static io.activej.cube.CubeStructure.AggregationConfig.id;
  import static io.activej.cube.aggregation.PrimaryKey.ofArray;
  import static io.activej.cube.aggregation.fieldtype.FieldTypes.*;
  import static io.activej.cube.aggregation.measure.Measures.sum;
  import static io.activej.cube.aggregation.predicate.AggregationPredicates.gt;
  import static io.activej.cube.aggregation.predicate.AggregationPredicates.notEq;
  import static io.activej.promise.TestUtils.await;
  import static java.util.stream.Collectors.toSet;
  import static org.junit.Assert.assertEquals;
  
  public final class CubeGetIrrelevantChunksTest extends CubeTestBase {
  	private static final int NUMBER_MIN = 0;
  	private static final int NUMBER_MAX = 100;
  
  	private static final int DATE_MIN_DAYS = (int) LocalDate.of(2020, Month.JANUARY, 1).toEpochDay();
  	private static final int DATE_MAX_DAYS = (int) LocalDate.of(2021, Month.JANUARY, 1).toEpochDay();
  
  	private static final LocalDate LOWER_DATE_BOUNDARY = LocalDate.of(2020, Month.JULY, 31);
  	private static final int LOWER_DATE_BOUNDARY_DAYS = (int) LOWER_DATE_BOUNDARY.toEpochDay();
  	private static final AggregationPredicate DATE_PREDICATE = gt("date", LOWER_DATE_BOUNDARY);
  	private static final int LOWER_NUMBER_BOUNDARY = 50;
  	private static final AggregationPredicate ADVERTISER_PREDICATE = gt("advertiser", LOWER_NUMBER_BOUNDARY);
  
  	private AggregationConfig dateAggregation;
  	private AggregationConfig advertiserDateAggregation;
  	private AggregationConfig enumAggregation;
  	private StateManager<LogDiff<CubeDiff>, LogState<CubeDiff, CubeState>> stateManager;
  	private CubeStructure cubeStructure;
  
  	private long chunkId;
  
  	private final Set<Object> toBePreserved = new HashSet<>();
  	private final Set<Object> toBeCleanedUp = new HashSet<>();
  
  	private List<LogDiff<CubeDiff>> diffs;
  
  	@Before
  	public void before() throws Exception {
  		diffs = new ArrayList<>();
  		chunkId = 0;
  		toBeCleanedUp.clear();
  		toBePreserved.clear();
  		Path aggregationsDir = temporaryFolder.newFolder().toPath();
  
  		FileSystem fs = FileSystem.builder(reactor, EXECUTOR, aggregationsDir)
  			.withTempDir(Files.createTempDirectory(""))
  			.build();
  		await(fs.start());
  
  		dateAggregation = id("date")
  			.withDimensions("date")
  			.withMeasures("impressions", "clicks", "conversions", "revenue");
  
  		advertiserDateAggregation = id("advertiser-date")
  			.withDimensions("advertiser", "date")
  			.withMeasures("impressions", "clicks", "conversions", "revenue");
  
  		enumAggregation = id("enum-date")
  			.withDimensions("enum", "date")
  			.withMeasures("impressions", "clicks", "conversions", "revenue");
  
  		CubeStructure basicCubeStructure = builderOfBasicCubeStructure()
  			.withAggregation(dateAggregation)
  			.withAggregation(advertiserDateAggregation)
  			.withAggregation(enumAggregation)
  			.build();
  
  		stateManager = stateManagerFactory.create(basicCubeStructure, description);
  	}
  
  	@Test
  	public void date() throws Exception {
  		cubeStructure = builderOfBasicCubeStructure()
  			.withAggregation(dateAggregation.withPredicate(DATE_PREDICATE))
  			.build();
  
  		toBePreserved.add(addChunk("date", ofArray(DATE_MIN_DAYS), ofArray(DATE_MAX_DAYS)));
  		toBePreserved.add(addChunk("date", ofArray(DATE_MIN_DAYS + 50), ofArray(DATE_MAX_DAYS - 50)));
  
  		toBeCleanedUp.add(addChunk("date", ofArray(DATE_MIN_DAYS), ofArray(LOWER_DATE_BOUNDARY_DAYS)));
  		toBeCleanedUp.add(addChunk("date", ofArray(DATE_MIN_DAYS), ofArray(DATE_MIN_DAYS + 50)));
  
  		doTest();
  	}
  
  	@Test
  	public void anEnum() {
  		cubeStructure = builderOfBasicCubeStructure()
  			.withAggregation(enumAggregation.withPredicate(notEq("enum", null)))
  			.build();
  
  		toBePreserved.add(addChunk("enum-date", ofArray(MyEnum.ONE, DATE_MIN_DAYS), ofArray(MyEnum.ONE, DATE_MAX_DAYS)));
  		toBePreserved.add(addChunk("enum-date", ofArray(MyEnum.ONE, DATE_MIN_DAYS + 50), ofArray(MyEnum.TWO, DATE_MAX_DAYS - 50)));
  
  		doTest();
  	}
  
  	@Test
  	public void advertiserDate_DatePredicate() throws Exception {
  		cubeStructure = builderOfBasicCubeStructure()
  			.withAggregation(advertiserDateAggregation.withPredicate(DATE_PREDICATE))
  			.build();
  		stateManager = stateManagerFactory.create(cubeStructure, description);
  
  		toBePreserved.add(addChunk("advertiser-date", ofArray(5, DATE_MIN_DAYS), ofArray(6, DATE_MAX_DAYS)));
  		toBePreserved.add(addChunk("advertiser-date", ofArray(1, DATE_MIN_DAYS + 50), ofArray(20, DATE_MAX_DAYS - 50)));
  		toBePreserved.add(addChunk("advertiser-date", ofArray(6, DATE_MIN_DAYS), ofArray(7, DATE_MIN_DAYS + 1)));
  		toBePreserved.add(addChunk("advertiser-date", ofArray(10, DATE_MIN_DAYS), ofArray(10, LOWER_DATE_BOUNDARY_DAYS + 1)));
  
  		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(5, DATE_MIN_DAYS), ofArray(5, DATE_MIN_DAYS + 50)));
  		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(10, DATE_MIN_DAYS + 50), ofArray(10, LOWER_DATE_BOUNDARY_DAYS)));
  
  		doTest();
  	}
  
  	@Test
  	public void advertiserDate_AdvertiserPredicate() throws Exception {
  		cubeStructure = builderOfBasicCubeStructure()
  			.withAggregation(advertiserDateAggregation.withPredicate(ADVERTISER_PREDICATE))
  			.build();
  		stateManager = stateManagerFactory.create(cubeStructure, description);
  
  		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MIN, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY + 1, DATE_MAX_DAYS)));
  		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MIN + 50, DATE_MIN_DAYS + 50), ofArray(NUMBER_MAX - 10, DATE_MAX_DAYS - 50)));
  		toBePreserved.add(addChunk("advertiser-date", ofArray(LOWER_NUMBER_BOUNDARY - 1, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY + 1, DATE_MIN_DAYS + 1)));
  
  		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(NUMBER_MIN, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY, DATE_MIN_DAYS + 50)));
  		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(LOWER_NUMBER_BOUNDARY - 10, DATE_MIN_DAYS + 50), ofArray(LOWER_NUMBER_BOUNDARY - 5, LOWER_DATE_BOUNDARY_DAYS)));
  
  		doTest();
  	}
  
  	@Test
  	public void advertiserDate_AdvertiserPredicateAndDatePredicate() throws Exception {
  		cubeStructure = builderOfBasicCubeStructure()
  			.withAggregation(advertiserDateAggregation
  				.withPredicate(AggregationPredicates.and(ADVERTISER_PREDICATE, DATE_PREDICATE)))
  			.build();
  		stateManager = stateManagerFactory.create(cubeStructure, description);
  
  		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MIN, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY + 1, DATE_MAX_DAYS)));
  		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MIN + 50, DATE_MIN_DAYS + 50), ofArray(NUMBER_MAX - 10, DATE_MAX_DAYS - 50)));
  		toBePreserved.add(addChunk("advertiser-date", ofArray(LOWER_NUMBER_BOUNDARY - 1, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY + 1, DATE_MIN_DAYS + 1)));
  		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MAX - 10, DATE_MAX_DAYS - 50), ofArray(NUMBER_MAX - 5, DATE_MAX_DAYS - 25)));
  		toBePreserved.add(addChunk("advertiser-date", ofArray(NUMBER_MAX - 10, DATE_MAX_DAYS - 50), ofArray(NUMBER_MAX - 10, DATE_MAX_DAYS - 25)));
  
  		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(NUMBER_MIN, DATE_MIN_DAYS), ofArray(LOWER_NUMBER_BOUNDARY, DATE_MAX_DAYS - 50)));
  		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(LOWER_NUMBER_BOUNDARY - 10, DATE_MIN_DAYS + 50), ofArray(LOWER_NUMBER_BOUNDARY - 5, LOWER_DATE_BOUNDARY_DAYS + 10)));
  		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(LOWER_NUMBER_BOUNDARY - 10, DATE_MIN_DAYS + 50), ofArray(LOWER_NUMBER_BOUNDARY - 10, LOWER_DATE_BOUNDARY_DAYS + 10)));
  		toBeCleanedUp.add(addChunk("advertiser-date", ofArray(NUMBER_MIN + 10, DATE_MAX_DAYS - 50), ofArray(NUMBER_MIN + 10, DATE_MAX_DAYS - 25)));
  
  		doTest();
  	}
  
  	private void doTest() {
  		await(stateManager.push(diffs));
  
  		Set<Object> expectedChunks = new HashSet<>();
  		expectedChunks.addAll(toBePreserved);
  		expectedChunks.addAll(toBeCleanedUp);
  
  		assertEquals(expectedChunks, stateManager.query(logState -> logState.getDataState().getAllChunks()));
  
  		stateManager = stateManagerFactory.createUninitialized(cubeStructure, description);
  		stateManagerFactory.start(stateManager);
  
  		Set<Object> irrelevantChunks = stateManager.query(logState -> logState.getDataState().getIrrelevantChunks())
  			.values()
  			.stream()
  			.flatMap(Collection::stream)
  			.map(AggregationChunk::getChunkId)
  			.collect(toSet());
  		assertEquals(toBeCleanedUp, irrelevantChunks);
  	}
  
  	private long addChunk(String aggregationId, PrimaryKey minKey, PrimaryKey maxKey) {
  		long chunkId = ++this.chunkId;
  		diffs.add(LogDiff.forCurrentPosition(
  			CubeDiff.of(Map.of(
  				aggregationId, AggregationDiff.of(Set.of(
  					AggregationChunk.create(
  						chunkId,
  						cubeStructure.getAggregationStructure(aggregationId).getMeasures(),
  						minKey,
  						maxKey,
  						1))))
  			)));
  		return chunkId;
  	}
  
  	private CubeStructure.Builder builderOfBasicCubeStructure() {
  		return CubeStructure.builder()
  			.withDimension("date", ofLocalDate())
  			.withDimension("advertiser", ofInt())
  			.withDimension("campaign", ofInt())
  			.withDimension("banner", ofInt())
  			.withDimension("enum", ofEnum(MyEnum.class))
  			.withRelation("campaign", "advertiser")
  			.withRelation("banner", "campaign")
  			.withMeasure("impressions", sum(ofLong()))
  			.withMeasure("clicks", sum(ofLong()))
  			.withMeasure("conversions", sum(ofLong()))
  			.withMeasure("revenue", sum(ofDouble()));
  	}
  
  	public enum MyEnum {
  		ONE, TWO, THREE
  	}
  }
  