Merge EventFilterCte into PreprocessingCte

backend/src/main/java/com/bakdata/conquery/sql/conversion/cqelement/concept/CQConceptConverter.java

@@ -54,7 +54,7 @@ public class CQConceptConverter implements NodeConverter<CQConcept> {
 	public CQConceptConverter() {
 		this.connectorCTEs = List.of(
 				new PreprocessingCte(),
-				new EventFilterCte(),
+				// new EventFilterCte(),
 				new AggregationSelectCte(),
 				new JoinBranchesCte(),
 				new AggregationFilterCte()

backend/src/main/java/com/bakdata/conquery/sql/conversion/cqelement/concept/ConceptCteStep.java

@@ -13,8 +13,7 @@ public enum ConceptCteStep implements CteStep {
 
 	// connector
 	PREPROCESSING("preprocessing", null),
-	EVENT_FILTER("event_filter", PREPROCESSING),
-	AGGREGATION_SELECT("group_select", EVENT_FILTER),
+	AGGREGATION_SELECT("group_select", PREPROCESSING),
 	JOIN_BRANCHES("join_branches", AGGREGATION_SELECT),
 	AGGREGATION_FILTER("group_filter", JOIN_BRANCHES),
 
@@ -27,7 +26,6 @@ public enum ConceptCteStep implements CteStep {
 
 	public static final Set<CteStep> MANDATORY_STEPS = Set.of(
 			PREPROCESSING,
-			EVENT_FILTER,
 			AGGREGATION_SELECT,
 			JOIN_BRANCHES,
 			AGGREGATION_FILTER

backend/src/main/java/com/bakdata/conquery/sql/conversion/cqelement/concept/PreprocessingCte.java

@@ -3,14 +3,17 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 import com.bakdata.conquery.sql.conversion.dialect.SqlFunctionProvider;
+import com.bakdata.conquery.sql.conversion.model.ColumnDateRange;
 import com.bakdata.conquery.sql.conversion.model.QueryStep;
 import com.bakdata.conquery.sql.conversion.model.Selects;
 import com.bakdata.conquery.sql.conversion.model.SqlIdColumns;
 import com.bakdata.conquery.sql.conversion.model.filter.SqlFilters;
 import com.bakdata.conquery.sql.conversion.model.filter.WhereCondition;
 import com.bakdata.conquery.sql.conversion.model.select.SqlSelect;
+import com.google.common.base.Preconditions;
 import org.jooq.Condition;
 import org.jooq.Record;
 import org.jooq.Table;
@@ -19,6 +22,35 @@
 
 class PreprocessingCte extends ConnectorCte {
 
+	public static List<Condition> collectEventFilterConditions(CQTableContext tableContext) {
+
+		List<Condition> eventFilterConditions = tableContext.getSqlFilters().stream()
+															.flatMap(conceptFilter -> conceptFilter.getWhereClauses().getEventFilters().stream())
+															.map(WhereCondition::condition)
+															.toList();
+
+		if (tableContext.getConversionContext().isWithStratification()) {
+			return addStratificationCondition(eventFilterConditions, tableContext);
+		}
+		return eventFilterConditions;
+	}
+
+	public static List<Condition> addStratificationCondition(List<Condition> eventFilterConditions, CQTableContext tableContext) {
+		Selects previousSelects = tableContext.getPrevious().getQualifiedSelects();
+		Preconditions.checkArgument(
+				previousSelects.getStratificationDate().isPresent() && previousSelects.getValidityDate().isPresent(),
+				"Can't apply stratification for table %s".formatted(tableContext.getConnectorTables().getRootTable())
+		);
+
+		// we filter every entry where stratification date range and validity date range do not overlap
+		SqlFunctionProvider functionProvider = tableContext.getFunctionProvider();
+		ColumnDateRange stratificationDate = previousSelects.getStratificationDate().get();
+		ColumnDateRange validityDate = previousSelects.getValidityDate().get();
+		Condition stratificationCondition = functionProvider.dateRestriction(stratificationDate, validityDate);
+
+		return Stream.concat(Stream.of(stratificationCondition), eventFilterConditions.stream()).toList();
+	}
+
 	@Override
 	public ConceptCteStep cteStep() {
 		return ConceptCteStep.PREPROCESSING;
@@ -36,6 +68,7 @@ public QueryStep.QueryStepBuilder convertStep(CQTableContext tableContext) {
 											  .validityDate(Optional.of(tableContext.getValidityDate()))
 											  .sqlSelects(forPreprocessing)
 											  .build();
+
 		// all where clauses that don't require any preprocessing (connector/child conditions)
 		List<Condition> conditions = new ArrayList<>();
 
@@ -45,15 +78,20 @@ public QueryStep.QueryStepBuilder convertStep(CQTableContext tableContext) {
 			}
 		}
 
+		conditions.addAll(collectEventFilterConditions(tableContext));
 
-		QueryStep.QueryStepBuilder builder = QueryStep.builder()
-													  .selects(preprocessingSelects)
-													  .conditions(conditions);
+		if (tableContext.getIds().getSecondaryId().isPresent()) {
+			conditions.add(tableContext.getIds().getSecondaryId().get().isNotNull());
+		}
 
 		if (tableContext.getConversionContext().isWithStratification()) {
 			return joinWithStratificationTable(forPreprocessing, conditions, tableContext);
 		}
 
+		QueryStep.QueryStepBuilder builder = QueryStep.builder()
+				.selects(preprocessingSelects)
+				.conditions(conditions);
+
 		TableLike<Record> rootTable = QueryStep.toTableLike(tableContext.getConnectorTables().getPredecessor(ConceptCteStep.PREPROCESSING));
 		return builder.fromTable(rootTable);
 

backend/src/main/java/com/bakdata/conquery/sql/conversion/cqelement/concept/TablePath.java

@@ -91,7 +91,7 @@ private static TablePathInfo collectConnectorTables(CQConcept cqConcept, CQTable
 
 		// interval packing requiredw
 		tableInfo.setContainsIntervalPacking(true);
-		tableInfo.addMappings(IntervalPackingCteStep.getMappings(EVENT_FILTER, context.getDialectBundle()));
+		tableInfo.addMappings(IntervalPackingCteStep.getMappings(PREPROCESSING, context.getDialectBundle()));
 
 		// validity date propagation not necessary
 		if (!cqConcept.isAggregateEventDates()) {

backend/src/main/java/com/bakdata/conquery/sql/conversion/dialect/clickhouse/ClickhouseDistinctSelectConverter.java

@@ -24,7 +24,7 @@ public ConnectorSqlSelects connectorSelect(DistinctSelect distinctSelect, Select
 		SingleColumnSqlSelect preprocessingSelect =
 				MappableSingleColumnSelect.getSubstringSelect(distinctSelect.getColumn().get(), distinctSelect.getSubstringRange(), selectContext, alias);
 
-		String eventFilterTable = selectContext.getTables().cteName(ConceptCteStep.EVENT_FILTER);
+		String eventFilterTable = selectContext.getTables().cteName(ConceptCteStep.PREPROCESSING);
 		SingleColumnSqlSelect qualified = preprocessingSelect.qualify(eventFilterTable);
 
 		FieldWrapper<?> grouped =

backend/src/main/java/com/bakdata/conquery/sql/conversion/forms/README.md

@@ -375,7 +375,7 @@ from "vers_stamm"
               on "full_stratification"."primary_id" = "vers_stamm"."pid"
 ```
 
-2. In the `EVENT_FILTER` step, we filter all entries where the stratification range and the subjects validity date do
+2. In the `PREPROCESSING` step, we filter all entries where the stratification range and the subjects validity date do
    not overlap. This is important because we only want to compute aggregations for those ranges that satisfy this
    condition.
 

backend/src/main/java/com/bakdata/conquery/sql/conversion/model/aggregator/CountQuartersSqlAggregator.java

@@ -40,7 +40,7 @@ private static CommonAggregationSelect<Integer> createSingleDateColumnAggregatio
 
 		ExtractingSqlSelect<Date> rootSelect = new ExtractingSqlSelect<>(tables.getRootTable(), countColumn.getName(), Date.class);
 
-		Field<Date> qualifiedRootSelect = rootSelect.qualify(tables.cteName(ConceptCteStep.EVENT_FILTER)).select();
+		Field<Date> qualifiedRootSelect = rootSelect.qualify(tables.cteName(ConceptCteStep.PREPROCESSING)).select();
 		FieldWrapper<Integer> countQuartersAggregation =
 				new FieldWrapper<>(DSL.nullif(DSL.countDistinct(functionProvider.yearQuarter(qualifiedRootSelect)), 0).as(alias), countColumn.getName());
 
@@ -73,7 +73,7 @@ private static CommonAggregationSelect<BigDecimal> sumQuarterCount(
 		Field<Integer> quarterCount = calcQuarterCount(quarterStart, nextQuarterStart, alias, functionProvider);
 		FieldWrapper<Integer> quarterCountWrapper = new FieldWrapper<>(quarterCount);
 
-		Field<Integer> qualifiedQuarterCount = quarterCountWrapper.qualify(tables.cteName(ConceptCteStep.EVENT_FILTER)).select();
+		Field<Integer> qualifiedQuarterCount = quarterCountWrapper.qualify(tables.cteName(ConceptCteStep.PREPROCESSING)).select();
 		FieldWrapper<BigDecimal> quarterCountAggregation = new FieldWrapper<>(DSL.nullif(DSL.sum(qualifiedQuarterCount), BigDecimal.ZERO).as(alias));
 
 		return CommonAggregationSelect.<BigDecimal>builder().rootSelect(quarterCountWrapper).groupBy(quarterCountAggregation).build();

backend/src/main/java/com/bakdata/conquery/sql/conversion/model/aggregator/DateDistanceSqlAggregator.java

@@ -16,12 +16,12 @@
 import com.bakdata.conquery.sql.conversion.model.select.*;
 import org.jooq.Condition;
 import org.jooq.Field;
-import org.jooq.impl.DSL;
 
 import java.sql.Date;
 import java.time.LocalDate;
 import java.time.temporal.ChronoUnit;
 
+import static org.jooq.impl.DSL.*;
 import static org.jooq.impl.DSL.field;
 
 public class DateDistanceSqlAggregator implements SelectConverter<DateDistanceSelect>, FilterConverter<DateDistanceFilter, Range.LongRange> {
@@ -33,16 +33,18 @@ public ConnectorSqlSelects connectorSelect(DateDistanceSelect select, SelectCont
 		String alias = selectContext.getNameGenerator().selectName(select);
 		ConnectorSqlTables tables = selectContext.getTables();
 
-		FieldWrapper<Integer> dateDistanceSelect = createDateDistanceSelect(column, alias, select.getTimeUnit(), tables, selectContext.getConversionContext());
+		Field<Integer> dateDistanceSelect = createDateDistanceSelect(column, select.getTimeUnit(), tables, selectContext.getConversionContext())
+				.as(alias);
+		FieldWrapper<Integer> dateDistanceWrapper = new FieldWrapper<>(dateDistanceSelect);
 
-		Field<Integer> qualifiedDateDistance = dateDistanceSelect.qualify(tables.getPredecessor(ConceptCteStep.AGGREGATION_SELECT)).select();
-		FieldWrapper<Integer> minDateDistance = new FieldWrapper<>(DSL.min(qualifiedDateDistance).as(alias));
+		Field<Integer> qualifiedDateDistance = dateDistanceWrapper.qualify(tables.getPredecessor(ConceptCteStep.AGGREGATION_SELECT)).select();
+		FieldWrapper<Integer> minDateDistance = new FieldWrapper<>(min(qualifiedDateDistance).as(alias));
 
 		String finalPredecessor = tables.getPredecessor(ConceptCteStep.AGGREGATION_FILTER);
 		ExtractingSqlSelect<Integer> finalSelect = minDateDistance.qualify(finalPredecessor);
 
 		return ConnectorSqlSelects.builder()
-								  .preprocessingSelect(dateDistanceSelect)
+								  .preprocessingSelect(dateDistanceWrapper)
 								  .aggregationSelect(minDateDistance)
 								  .finalSelect(finalSelect)
 								  .build();
@@ -52,48 +54,36 @@ public ConnectorSqlSelects connectorSelect(DateDistanceSelect select, SelectCont
 	public SqlFilters convertToSqlFilter(DateDistanceFilter filter, FilterContext<Range.LongRange> filterContext) {
 
 		Column column = filter.getColumn().resolve();
-		String alias = filterContext.getNameGenerator().selectName(filter);
-		ConnectorSqlTables tables = filterContext.getTables();
+		Field<Integer> dateDistance = createDateDistanceSelect(column , filter.getTimeUnit(), filterContext.getTables(), filterContext.getConversionContext());
 
-		FieldWrapper<Integer> dateDistanceSelect = createDateDistanceSelect(column, alias, filter.getTimeUnit(), tables, filterContext.getConversionContext());
-		ConnectorSqlSelects selects = ConnectorSqlSelects.builder().preprocessingSelect(dateDistanceSelect).build();
-
-		String eventFilterCteName = tables.getPredecessor(ConceptCteStep.EVENT_FILTER);
-		Field<Integer> qualifiedDateDistanceSelect = dateDistanceSelect.qualify(eventFilterCteName).select();
-		WhereCondition dateDistanceCondition = new DateDistanceCondition(qualifiedDateDistanceSelect, filterContext.getValue());
+		WhereCondition dateDistanceCondition = new DateDistanceCondition(dateDistance, filterContext.getValue());
 
 		WhereClauses whereClauses = WhereClauses.builder().eventFilter(dateDistanceCondition).build();
 
-		return new SqlFilters(selects, whereClauses);
+		return new SqlFilters(ConnectorSqlSelects.none(), whereClauses);
 	}
 
 	@Override
 	public Condition convertForTableExport(DateDistanceFilter filter, FilterContext<Range.LongRange> filterContext) {
 
 		Column column = filter.getColumn().resolve();
-		String tableName = column.getTable().getName();
-		String columnName = column.getName();
-
-		Field<Date> startDateField = DSL.field(DSL.name(tableName, columnName), Date.class);
-		Field<Date> endDate = getEndDate(filterContext.getConversionContext());
+		Field<Integer> dateDistance = createDateDistanceSelect(column , filter.getTimeUnit(), filterContext.getTables(), filterContext.getConversionContext());
 
-		Field<Integer> dateDistance = filterContext.getFunctionProvider().dateDistance(filter.getTimeUnit(), startDateField, endDate);
 		return new DateDistanceCondition(dateDistance, filterContext.getValue()).condition();
 	}
 
-	private FieldWrapper<Integer> createDateDistanceSelect(
+	private Field<Integer> createDateDistanceSelect(
 			Column column,
-			String alias,
 			ChronoUnit timeUnit,
 			SqlTables tables,
 			ConversionContext conversionContext
 	) {
-        Field<Date> startDate = field(DSL.name(tables.getRootTable(), column.getName()), Date.class);
+        Field<Date> startDate = field(name(tables.getRootTable(), column.getName()), Date.class);
 
         Field<Date> endDate = getEndDate(conversionContext);
 
 		SqlFunctionProvider functionProvider = conversionContext.getFunctionProvider();
-		return new FieldWrapper<>(functionProvider.dateDistance(timeUnit, startDate, endDate).as(alias));
+		return functionProvider.dateDistance(timeUnit, startDate, endDate);
 	}
 
 	private Field<Date> getEndDate(ConversionContext conversionContext) {
@@ -105,7 +95,7 @@ private Field<Date> getEndDate(ConversionContext conversionContext) {
 			ColumnDateRange stratificationDate = conversionContext.getStratificationTable().getQualifiedSelects().getStratificationDate().get();
 			ColumnDateRange dualColumn = functionProvider.toDualColumn(stratificationDate);
 			// end date is allways treated exclusive, so we get the actual end date when subtracting 1 day
-			return functionProvider.addDays(dualColumn.getEnd(), DSL.inline(-1));
+			return functionProvider.addDays(dualColumn.getEnd(), inline(-1));
 		}
 
 		LocalDate endDate;