From c588ac3bad683fbebd91053f70f180b1f10213ea Mon Sep 17 00:00:00 2001 From: awildturtok Date: Wed, 30 Sep 2026 12:23:03 +0200 Subject: [PATCH 1/9] Optimize preprocessed bucket construction --- .../conquery/models/preproc/Preprocessed.java | 39 ++++++++----------- 1 file changed, 16 insertions(+), 23 deletions(-) diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java index e32f51b359..f1ac631fda 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java @@ -6,12 +6,11 @@ import java.io.OutputStream; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collection; import java.util.HashMap; -import java.util.HashSet; import java.util.IntSummaryStatistics; import java.util.List; import java.util.Map; +import java.util.TreeMap; import java.util.stream.Collectors; import java.util.zip.GZIPOutputStream; @@ -24,7 +23,6 @@ import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.models.preproc.parser.specific.StringParser; import com.fasterxml.jackson.core.JsonGenerator; -import com.google.common.collect.Maps; import com.google.common.hash.Hashing; import it.unimi.dsi.fastutil.ints.IntArrayList; import it.unimi.dsi.fastutil.ints.IntList; @@ -100,11 +98,10 @@ public void write(File file, int buckets) throws IOException { log.debug("Writing Headers"); - //TODO this could actually be done at read-time, avoiding large allocations entirely. But in a different smaller PR. - final Map> bucket2Entity = entityStart.keySet().stream() - .collect(Collectors.groupingBy(id -> getEntityBucket(buckets, id))) - .entrySet().stream() - .collect(Collectors.toMap(Map.Entry::getKey, entry -> new HashSet<>(entry.getValue()))); + final Map> bucket2Entity = new TreeMap<>(); + for (String entity : entityStart.keySet()) { + bucket2Entity.computeIfAbsent(getEntityBucket(buckets, entity), ignored -> new ArrayList<>()).add(entity); + } final int hash = descriptor.calculateValidityHash(job.getCsvDirectory(), job.getTag()); @@ -185,7 +182,7 @@ private Map combineStores(Object2IntMap entityStart return columnStores; } - private static void writePreprocessed(File file, PreprocessedHeader header, Map globalStarts, Map globalLengths, Map data, Map> bucket2Entities) throws IOException { + private static void writePreprocessed(File file, PreprocessedHeader header, Object2IntMap globalStarts, Object2IntMap globalLengths, Map data, Map> bucket2Entities) throws IOException { final OutputStream out = new GZIPOutputStream(new FileOutputStream(file)); try (JsonGenerator generator = Jackson.BINARY_MAPPER.copy().enable(JsonGenerator.Feature.AUTO_CLOSE_TARGET).getFactory().createGenerator(out)) { @@ -195,20 +192,15 @@ private static void writePreprocessed(File file, PreprocessedHeader header, Map< log.debug("Writing data"); - for (Map.Entry> bucketIds : bucket2Entities.entrySet()) { - final Collection entities = bucketIds.getValue(); - - final Map starts = Maps.filterKeys(globalStarts, entities::contains); - final Map lengths = Maps.filterKeys(globalLengths, entities::contains); - - final PreprocessedData preprocessedData = selectBucket(bucketIds.getKey(), starts, lengths, data); + for (Map.Entry> bucketIds : bucket2Entities.entrySet()) { + final PreprocessedData preprocessedData = selectBucket(bucketIds.getKey(), bucketIds.getValue(), globalStarts, globalLengths, data); generator.writeObject(preprocessedData); } } } - private static PreprocessedData selectBucket(int bucket, Map localStarts, Map localLengths, Map stores) { + private static PreprocessedData selectBucket(int bucket, List entities, Object2IntMap globalStarts, Object2IntMap globalLengths, Map stores) { final IntList selectionStart = new IntArrayList(); @@ -222,11 +214,9 @@ private static PreprocessedData selectBucket(int bucket, Map lo int currentStart = 0; - for (Map.Entry entity2Start : localStarts.entrySet()) { - final String entity = entity2Start.getKey(); - final int start = entity2Start.getValue(); - - final int length = localLengths.get(entity); + for (String entity : entities) { + final int start = globalStarts.getInt(entity); + final int length = globalLengths.getInt(entity); selectionStart.add(start); @@ -238,13 +228,16 @@ private static PreprocessedData selectBucket(int bucket, Map lo currentStart += length; } + final int[] selectionStarts = selectionStart.toIntArray(); + final int[] selectionLengths = selectionLength.toIntArray(); + final Map selected = new HashMap<>(); for (Map.Entry entry : stores.entrySet()) { final String name = entry.getKey(); final ColumnStore store = entry.getValue(); - selected.put(name, store.select(selectionStart.toIntArray(), selectionLength.toIntArray())); + selected.put(name, store.select(selectionStarts, selectionLengths)); } return new PreprocessedData(bucket, entityStarts, entityEnds, selected); From 4b4290900b3de55793b67d3d33149d4ee5d08e01 Mon Sep 17 00:00:00 2001 From: awildturtok Date: Wed, 30 Sep 2026 13:57:44 +0200 Subject: [PATCH 2/9] Reduce preprocessing finalization heap usage --- .../conquery/models/preproc/PPColumn.java | 6 +- .../conquery/models/preproc/Preprocessed.java | 225 ++++++++++-------- .../models/preproc/parser/Parser.java | 22 +- .../parser/specific/BooleanParser.java | 7 +- .../specific/CompoundDateRangeParser.java | 7 +- .../preproc/parser/specific/DateParser.java | 7 +- .../parser/specific/DateRangeParser.java | 9 +- .../parser/specific/DecimalParser.java | 9 +- .../parser/specific/IntegerParser.java | 21 +- .../preproc/parser/specific/MoneyParser.java | 7 +- .../preproc/parser/specific/RealParser.java | 9 +- .../preproc/parser/specific/StringParser.java | 7 +- 12 files changed, 211 insertions(+), 125 deletions(-) diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/PPColumn.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/PPColumn.java index f3e00d8be3..a3e7f2fa52 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/PPColumn.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/PPColumn.java @@ -25,11 +25,15 @@ public class PPColumn { public ColumnStore findBestType() { log.info("Compute best Subtype for Column[{}] with {}", getName(), getParser()); - ColumnStore decision = parser.findBestType(); + ColumnStore decision = createStore(parser.getLines()); // this only creates the headers log.debug("\t{}: {} -> {}", getName(), getParser(), decision); return decision; } + + public ColumnStore createStore(int lines) { + return parser.findBestType(lines); + } } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java index f1ac631fda..e5d6138852 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java @@ -4,14 +4,14 @@ import java.io.FileOutputStream; import java.io.IOException; import java.io.OutputStream; +import java.util.ArrayDeque; import java.util.ArrayList; -import java.util.Arrays; +import java.util.Deque; import java.util.HashMap; import java.util.IntSummaryStatistics; import java.util.List; import java.util.Map; import java.util.TreeMap; -import java.util.stream.Collectors; import java.util.zip.GZIPOutputStream; import com.bakdata.conquery.io.jackson.Jackson; @@ -26,7 +26,6 @@ import com.google.common.hash.Hashing; import it.unimi.dsi.fastutil.ints.IntArrayList; import it.unimi.dsi.fastutil.ints.IntList; -import it.unimi.dsi.fastutil.ints.IntLists; import it.unimi.dsi.fastutil.objects.Object2IntAVLTreeMap; import it.unimi.dsi.fastutil.objects.Object2IntMap; import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap; @@ -52,10 +51,34 @@ public class Preprocessed { /** * Per row store, entity. */ - private final List rowEntities = new ArrayList<>(); - + private final ArrayList rowEntities = new ArrayList<>(); private long rows; + private boolean finalized; + + private static final class BucketData { + private final int bucketId; + private final Object2IntMap entityStarts; + private final Object2IntMap entityEnds; + private int[] sourceEvents; + private final Map stores; + + private BucketData(int bucketId, Object2IntMap entityStarts, Object2IntMap entityEnds, int[] sourceEvents, Map stores) { + this.bucketId = bucketId; + this.entityStarts = entityStarts; + this.entityEnds = entityEnds; + this.sourceEvents = sourceEvents; + this.stores = stores; + } + + private PreprocessedData toPreprocessedData() { + return new PreprocessedData(bucketId, entityStarts, entityEnds, stores); + } + + private void releaseSourceEvents() { + sourceEvents = null; + } + } public Preprocessed(ConqueryConfig config, PreprocessingJob preprocessingJob) throws IOException { job = preprocessingJob; @@ -82,24 +105,22 @@ public Preprocessed(ConqueryConfig config, PreprocessingJob preprocessingJob) th } - public void write(File file, int buckets) throws IOException { + public synchronized void write(File file, int buckets) throws IOException { + ensureNotFinalized(); + finalized = true; - final Object2IntMap entityStart = new Object2IntAVLTreeMap<>(); final Object2IntMap entityLength = new Object2IntAVLTreeMap<>(); - calculateEntitySpans(entityStart, entityLength); + calculateEntityLengths(entityLength); final IntSummaryStatistics statistics = entityLength.values().intStream().summaryStatistics(); log.info("Statistics = {}", statistics); - final Map columnStores = combineStores(entityStart); - - log.debug("Writing Headers"); final Map> bucket2Entity = new TreeMap<>(); - for (String entity : entityStart.keySet()) { + for (String entity : entityLength.keySet()) { bucket2Entity.computeIfAbsent(getEntityBucket(buckets, entity), ignored -> new ArrayList<>()).add(entity); } @@ -107,9 +128,15 @@ public void write(File file, int buckets) throws IOException { final int hash = descriptor.calculateValidityHash(job.getCsvDirectory(), job.getTag()); final PreprocessedHeader header = - new PreprocessedHeader(descriptor.getName(), descriptor.getTable(), rows, entityStart.size(), bucket2Entity.size(), columns, hash); + new PreprocessedHeader(descriptor.getName(), descriptor.getTable(), rows, entityLength.size(), bucket2Entity.size(), columns, hash); + + final Deque bucketData = createBucketData(bucket2Entity, entityLength); - writePreprocessed(file, header, entityStart, entityLength, columnStores, bucket2Entity); + entityLength.clear(); + bucket2Entity.clear(); + + populateBucketStores(bucketData); + writePreprocessed(file, header, bucketData); } public static int getEntityBucket(int buckets, String id) { @@ -117,72 +144,100 @@ public static int getEntityBucket(int buckets, String id) { } /** - * Calculate beginning and length of entities in output data. + * Calculate the number of events per entity. */ - private void calculateEntitySpans(Object2IntMap entityStart, Object2IntMap entityLength) { + private void calculateEntityLengths(Object2IntMap entityLength) { - // Count the number of events for the entity for (String entity : rowEntities) { final int curr = entityLength.getOrDefault(entity, 0); entityLength.put(entity, curr + 1); } - - // Lay out the entities in order, adding their length. - int outIndex = 0; - - for (Object2IntMap.Entry entry : entityLength.object2IntEntrySet()) { - entityStart.put(entry.getKey(), outIndex); - outIndex += entry.getIntValue(); - } } - /** - * Combine raw by-Entity data into column stores, appropriately formatted. - */ - @SuppressWarnings("rawtypes") - private Map combineStores(Object2IntMap entityStart) { - final Map columnStores = Arrays.stream(columns).parallel().collect(Collectors.toMap(PPColumn::getName, PPColumn::findBestType)); - - // This object can be huge! - final Map entityEvents = new HashMap<>(entityStart.size()); + private Deque createBucketData(Map> bucket2Entities, Object2IntMap entityLengths) { + final Map entityEvents = new HashMap<>(entityLengths.size()); for (int pos = 0, size = rowEntities.size(); pos < size; pos++) { final String entity = rowEntities.get(pos); entityEvents.computeIfAbsent(entity, (ignored) -> new IntArrayList()).add(pos); } + rowEntities.clear(); + rowEntities.trimToSize(); - for (int colIdx = 0; colIdx < columns.length; colIdx++) { - final PPColumn ppColumn = columns[colIdx]; - final ColumnValues columnValues = values[colIdx]; + final Deque buckets = new ArrayDeque<>(bucket2Entities.size()); - //No need to preprocess the column further more, if it does not contain values, likely backed by a compound ColumnStore - if (columnValues == null) { - continue; + for (Map.Entry> bucket : bucket2Entities.entrySet()) { + final Object2IntMap entityStarts = new Object2IntOpenHashMap<>(); + final Object2IntMap entityEnds = new Object2IntOpenHashMap<>(); + final IntList sourceEvents = new IntArrayList(); + + int currentStart = 0; + for (String entity : bucket.getValue()) { + final int length = entityLengths.getInt(entity); + final IntList events = entityEvents.remove(entity); + + if (events == null || events.size() != length) { + throw new IllegalStateException("Entity events are not aligned for " + entity); + } + + entityStarts.put(entity, currentStart); + entityEnds.put(entity, currentStart + length); + currentStart += length; + + for (int event : events) { + sourceEvents.add(event); + } } - final ColumnStore store = columnStores.get(ppColumn.getName()); - entityStart.object2IntEntrySet().forEach(entry -> { - final String entity = entry.getKey(); - int outIndex = entry.getIntValue(); + buckets.addLast(new BucketData(bucket.getKey(), entityStarts, entityEnds, sourceEvents.toIntArray(), new HashMap<>())); + } - final IntList events = entityEvents.getOrDefault(entity, IntLists.emptyList()); + if (!entityEvents.isEmpty()) { + throw new IllegalStateException("Not all entity events were assigned to a bucket"); + } - for (int inIndex : events) { - if (columnValues.isNull(inIndex)) { - store.setNull(outIndex); - } - else { - final Object raw = columnValues.get(inIndex); - ppColumn.getParser().setValue(store, outIndex, raw); + return buckets; + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + private void populateBucketStores(Iterable buckets) { + for (int columnIndex = 0; columnIndex < columns.length; columnIndex++) { + final PPColumn column = columns[columnIndex]; + final ColumnValues columnValues = values[columnIndex]; + + log.info("Compute best Subtype for Column[{}] with {}", column.getName(), column.getParser()); + + try { + for (BucketData bucket : buckets) { + final ColumnStore store = column.createStore(bucket.sourceEvents.length); + + if (columnValues != null) { + for (int localEvent = 0; localEvent < bucket.sourceEvents.length; localEvent++) { + final int sourceEvent = bucket.sourceEvents[localEvent]; + + if (columnValues.isNull(sourceEvent)) { + store.setNull(localEvent); + } + else { + column.getParser().setValue(store, localEvent, columnValues.get(sourceEvent)); + } + } } - outIndex++; + + bucket.stores.put(column.getName(), store); } - }); + } + finally { + values[columnIndex] = null; + } + } + + for (BucketData bucket : buckets) { + bucket.releaseSourceEvents(); } - return columnStores; } - private static void writePreprocessed(File file, PreprocessedHeader header, Object2IntMap globalStarts, Object2IntMap globalLengths, Map data, Map> bucket2Entities) throws IOException { + private static void writePreprocessed(File file, PreprocessedHeader header, Deque buckets) throws IOException { final OutputStream out = new GZIPOutputStream(new FileOutputStream(file)); try (JsonGenerator generator = Jackson.BINARY_MAPPER.copy().enable(JsonGenerator.Feature.AUTO_CLOSE_TARGET).getFactory().createGenerator(out)) { @@ -192,62 +247,20 @@ private static void writePreprocessed(File file, PreprocessedHeader header, Obje log.debug("Writing data"); - for (Map.Entry> bucketIds : bucket2Entities.entrySet()) { - final PreprocessedData preprocessedData = selectBucket(bucketIds.getKey(), bucketIds.getValue(), globalStarts, globalLengths, data); - - generator.writeObject(preprocessedData); + BucketData bucket; + while ((bucket = buckets.pollFirst()) != null) { + generator.writeObject(bucket.toPreprocessedData()); } } } - private static PreprocessedData selectBucket(int bucket, List entities, Object2IntMap globalStarts, Object2IntMap globalLengths, Map stores) { - - - final IntList selectionStart = new IntArrayList(); - final IntList selectionLength = new IntArrayList(); - - - // First entity of Bucket starts at 0, the following are appended. - final Object2IntMap entityStarts = new Object2IntOpenHashMap<>(); - final Object2IntMap entityEnds = new Object2IntOpenHashMap<>(); - - - int currentStart = 0; - - for (String entity : entities) { - final int start = globalStarts.getInt(entity); - final int length = globalLengths.getInt(entity); - - selectionStart.add(start); - - selectionLength.add(length); - - entityStarts.put(entity, currentStart); - entityEnds.put(entity, currentStart + length); - - currentStart += length; - } - - final int[] selectionStarts = selectionStart.toIntArray(); - final int[] selectionLengths = selectionLength.toIntArray(); - - final Map selected = new HashMap<>(); - - for (Map.Entry entry : stores.entrySet()) { - final String name = entry.getKey(); - final ColumnStore store = entry.getValue(); - - selected.put(name, store.select(selectionStarts, selectionLengths)); - } - - return new PreprocessedData(bucket, entityStarts, entityEnds, selected); - } - public synchronized String addPrimary(String primary) { + ensureNotFinalized(); return primaryColumn.addLine(primary); } public synchronized void addRow(String primaryId, PPColumn[] columns, Object[] outRow) { + ensureNotFinalized(); final int event = rowEntities.size(); rowEntities.add(primaryId); @@ -273,4 +286,10 @@ public synchronized void addRow(String primaryId, PPColumn[] columns, Object[] o //update stats rows++; } + + private void ensureNotFinalized() { + if (finalized) { + throw new IllegalStateException("Preprocessed data has already been finalized"); + } + } } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java index f9a4addd64..ec0bd296c3 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java @@ -52,11 +52,22 @@ public final MAJOR_JAVA_TYPE parse(String v) throws ParsingException { protected abstract MAJOR_JAVA_TYPE parseValue(@Nonnull String value) throws ParsingException; public final STORE_TYPE findBestType() { + return findBestType(getLines()); + } + + /** + * Select the optimal store based on all registered values, but allocate it for the supplied number of lines. + */ + public final STORE_TYPE findBestType(int storeLines) { + if (storeLines < 0) { + throw new IllegalArgumentException("Store lines must not be negative"); + } + if (isEmpty()) { return (STORE_TYPE) EmptyStore.INSTANCE; // This implements all root ColumnStores. } - return decideType(); + return decideType(storeLines); } public boolean isEmpty() { @@ -66,7 +77,14 @@ public boolean isEmpty() { /** * Analyze all values and select an optimal store. */ - protected abstract STORE_TYPE decideType(); + protected STORE_TYPE decideType() { + return decideType(getLines()); + } + + /** + * Select the store representation from the globally collected parser statistics and allocate it with the supplied length. + */ + protected abstract STORE_TYPE decideType(int storeLines); /** * Process a single parsed line. diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanParser.java index 048127c22d..d4e21f1b8f 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanParser.java @@ -34,7 +34,12 @@ public static Boolean parseBoolean(@NotNull String value) { @Override protected BooleanStore decideType() { - return BitSetStore.create(getLines()); + return decideType(getLines()); + } + + @Override + protected BooleanStore decideType(int storeLines) { + return BitSetStore.create(storeLines); } @Override diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/CompoundDateRangeParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/CompoundDateRangeParser.java index d02536b969..1ad2b98036 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/CompoundDateRangeParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/CompoundDateRangeParser.java @@ -52,7 +52,12 @@ protected Boolean parseValue(@NotNull String value) throws ParsingException { @Override protected DateRangeStore decideType() { - return new CompoundDateRangeStore(this.startColumn, this.endColumn, BitSetStore.create(getLines())); + return decideType(getLines()); + } + + @Override + protected DateRangeStore decideType(int storeLines) { + return new CompoundDateRangeStore(this.startColumn, this.endColumn, BitSetStore.create(storeLines)); } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateParser.java index 65d3cedacf..bdfe3e292e 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateParser.java @@ -52,7 +52,12 @@ public Integer addLine(Integer v) { @Override protected DateStore decideType() { - IntegerStore subDecision = subType.findBestType(); + return decideType(getLines()); + } + + @Override + protected DateStore decideType(int storeLines) { + IntegerStore subDecision = subType.findBestType(storeLines); return new IntegerDateStore(subDecision); } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateRangeParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateRangeParser.java index 3a065e1c12..38cd4c45bd 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateRangeParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateRangeParser.java @@ -56,6 +56,11 @@ protected void registerValue(CDateRange v) { @Override protected DateRangeStore decideType() { + return decideType(getLines()); + } + + @Override + protected DateRangeStore decideType(int storeLines) { // Quarters cannot encode open ranges. if (!anyOpen && onlyQuarters) { @@ -64,7 +69,7 @@ protected DateRangeStore decideType() { quarterParser.setMaxValue(maxValue); quarterParser.setMinValue(minValue); - return new QuarterDateRangeStore(quarterParser.findBestType()); + return new QuarterDateRangeStore(quarterParser.findBestType(storeLines)); } @@ -78,7 +83,7 @@ protected DateRangeStore decideType() { maxParser.setLines(getLines()); } - return new DirectDateRangeStore(minParser.findBestType(), maxParser.findBestType()); + return new DirectDateRangeStore(minParser.findBestType(storeLines), maxParser.findBestType(storeLines)); } @Override diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java index 3a345c256c..f1f0f34ae7 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java @@ -46,10 +46,15 @@ protected void registerValue(BigDecimal v) { @Override protected DecimalStore decideType() { + return decideType(getLines()); + } + + @Override + protected DecimalStore decideType(int storeLines) { BigInteger unscaled = ScaledDecimalStore.unscale(maxScale, maxAbs); if (unscaled.bitLength() > 63) { - return DecimalArrayStore.create(getLines()); + return DecimalArrayStore.create(storeLines); } IntegerParser sub = new IntegerParser(getConfig()); @@ -57,7 +62,7 @@ protected DecimalStore decideType() { sub.setMinValue(-unscaled.longValueExact()); sub.setLines(getLines()); sub.setNullLines(getNullLines()); - IntegerStore subDecision = sub.findBestType(); + IntegerStore subDecision = sub.findBestType(storeLines); return new ScaledDecimalStore(maxScale, subDecision); } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerParser.java index fb5e15b0a4..40f0cf9a5c 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerParser.java @@ -46,6 +46,11 @@ protected void registerValue(Long v) { @Override protected IntegerStore decideType() { + return decideType(getLines()); + } + + @Override + protected IntegerStore decideType(int storeLines) { if (minValue > maxValue) { throw new IllegalStateException(String.format("Min (%d) > Max(%d)", minValue, maxValue)); } @@ -57,7 +62,7 @@ protected IntegerStore decideType() { span = Math.subtractExact(maxValue, minValue); } catch (ArithmeticException exception) { - return LongArrayStore.create(getLines()); + return LongArrayStore.create(storeLines); } // Create minimally required store. @@ -66,33 +71,33 @@ protected IntegerStore decideType() { if (span + 1 <= (long) Byte.MAX_VALUE - (long) Byte.MIN_VALUE) { if (minValue >= Byte.MIN_VALUE && maxValue + 1 <= Byte.MAX_VALUE) { - return ByteArrayStore.create(getLines()); + return ByteArrayStore.create(storeLines); } - return new RebasingIntegerStore(minValue, Byte.MIN_VALUE, ByteArrayStore.create(getLines())); + return new RebasingIntegerStore(minValue, Byte.MIN_VALUE, ByteArrayStore.create(storeLines)); } if (span + 1 <= (long) Short.MAX_VALUE - (long) Short.MIN_VALUE) { if (minValue >= Short.MIN_VALUE && maxValue + 1 <= Short.MAX_VALUE) { - return ShortArrayStore.create(getLines()); + return ShortArrayStore.create(storeLines); } - return new RebasingIntegerStore(minValue, Short.MIN_VALUE, ShortArrayStore.create(getLines())); + return new RebasingIntegerStore(minValue, Short.MIN_VALUE, ShortArrayStore.create(storeLines)); } if (span + 1 <= (long) Integer.MAX_VALUE - (long) Integer.MIN_VALUE) { if (minValue >= Integer.MIN_VALUE && maxValue + 1 <= Integer.MAX_VALUE) { - return IntArrayStore.create(getLines()); + return IntArrayStore.create(storeLines); } - return new RebasingIntegerStore(minValue, Integer.MIN_VALUE, IntArrayStore.create(getLines())); + return new RebasingIntegerStore(minValue, Integer.MIN_VALUE, IntArrayStore.create(storeLines)); } if (maxValue == Long.MAX_VALUE) { throw new IllegalStateException("Exceeds capacity of LongStore."); // todo migrate to external null-store } - return LongArrayStore.create(getLines()); + return LongArrayStore.create(storeLines); } @Override diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/MoneyParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/MoneyParser.java index 31e6675686..acbf4527e3 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/MoneyParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/MoneyParser.java @@ -44,12 +44,17 @@ protected void registerValue(BigDecimal v) { @Override protected MoneyStore decideType() { + return decideType(getLines()); + } + + @Override + protected MoneyStore decideType(int storeLines) { IntegerParser subParser = new IntegerParser(getConfig()); subParser.registerValue(maxValue.movePointRight(defaultFractionDigits).longValue()); subParser.registerValue(minValue.movePointRight(defaultFractionDigits).longValue()); subParser.setLines(getLines()); subParser.setNullLines(getNullLines()); - IntegerStore subDecision = subParser.findBestType(); + IntegerStore subDecision = subParser.findBestType(storeLines); return new MoneyIntStore(subDecision, defaultFractionDigits); } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/RealParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/RealParser.java index 0e13d3dac9..60d1b65e6f 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/RealParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/RealParser.java @@ -45,13 +45,18 @@ protected void registerValue(Double v) { */ @Override protected RealStore decideType() { + return decideType(getLines()); + } + + @Override + protected RealStore decideType(int storeLines) { log.debug("Max ULP = {}", floatULP); if (floatULP < requiredPrecision) { - return FloatArrayStore.create(getLines()); + return FloatArrayStore.create(storeLines); } else { - return DoubleArrayStore.create(getLines()); + return DoubleArrayStore.create(storeLines); } } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/StringParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/StringParser.java index 7acf67e522..d3036fd707 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/StringParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/StringParser.java @@ -41,13 +41,18 @@ protected void registerValue(String v) { @Override protected StringStore decideType() { + return decideType(getLines()); + } + + @Override + protected StringStore decideType(int storeLines) { //check if a singleton type is enough if (getLines() == 0) { return EmptyStore.INSTANCE; } - return StringStoreString.create(getLines()); + return StringStoreString.create(storeLines); } From 54c82b2a8aff93074e7bea707a2464f64ec3a29d Mon Sep 17 00:00:00 2001 From: awildturtok Date: Wed, 30 Sep 2026 14:23:34 +0200 Subject: [PATCH 3/9] Reduce preprocessing loop overhead --- .../conquery/models/preproc/Preprocessor.java | 39 ++++++++++++------- 1 file changed, 25 insertions(+), 14 deletions(-) diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java index 3a26d72bbb..82ae2034eb 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java @@ -7,6 +7,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.Objects; import java.util.Optional; @@ -36,6 +37,7 @@ @Slf4j @UtilityClass public class Preprocessor { + private static final int PROGRESS_UPDATE_INTERVAL = 4096; /** * Create version of file-name with tag. @@ -109,6 +111,7 @@ public static void preprocess(PreprocessingJob preprocessingJob, ProgressBar tot try (CountingInputStream countingIn = new CountingInputStream(new FileInputStream(sourceFile))) { long progress = 0; + int rowsSinceProgressUpdate = 0; CSVConfig csvSettings = config.getCsv(); // Create CSV parser according to config, but overriding some behaviour. @@ -134,10 +137,16 @@ public static void preprocess(PreprocessingJob preprocessingJob, ProgressBar tot outputs.add(op.createForHeaders(headerMap, dateReader, config)); } + final Object[] outRow = new Object[outputs.size()]; String[] row; // Read all CSV lines, apply Output transformations and add the to preprocessed. while ((row = parser.parseNext()) != null) { + rowsSinceProgressUpdate++; + if (rowsSinceProgressUpdate == PROGRESS_UPDATE_INTERVAL) { + progress = reportProgress(totalProgress, countingIn, progress); + rowsSinceProgressUpdate = 0; + } // Check if row shall be evaluated // This is explicitly NOT in a try-catch block as scripts may not fail and we should not recover from faulty scripts. @@ -151,7 +160,7 @@ public static void preprocess(PreprocessingJob preprocessingJob, ProgressBar tot final String primary = result.addPrimary(primaryId); - final Object[] outRow = applyOutputs(outputs, columns, row, lineId); + applyOutputs(outputs, columns, row, lineId, outRow); result.addRow(primary, columns, outRow); @@ -186,13 +195,12 @@ else if (errors == config.getPreprocessor().getMaximumPrintedErrors()) { } } finally { - //report progress - totalProgress.addCurrentValue(countingIn.getCount() - progress); - progress = countingIn.getCount(); lineId++; } } + reportProgress(totalProgress, countingIn, progress); + } finally { if (parser != null) { @@ -226,12 +234,10 @@ else if (errors == config.getPreprocessor().getMaximumPrintedErrors()) { } /** - * Apply each output for a single row. Returning all resulting values. + * Apply each output for a single row into the reusable output buffer. */ - private static Object[] applyOutputs(List outputs, PPColumn[] columns, String[] row, long lineId) + private static void applyOutputs(List outputs, PPColumn[] columns, String[] row, long lineId, Object[] outRow) throws ParsingException, OutputDescription.OutputException { - Object[] outRow = new Object[outputs.size()]; - for (int index = 0; index < outputs.size(); index++) { final OutputDescription.Output out = outputs.get(index); @@ -239,18 +245,23 @@ private static Object[] applyOutputs(List outputs, PPC final Parser parser = columns[index].getParser(); final Object result = out.createOutput(row, parser, lineId); - - if (result == null) { - continue; - } - outRow[index] = result; } catch (Exception e) { + Arrays.fill(outRow, null); throw new OutputDescription.OutputException(out.getDescription(), e); } } - return outRow; + } + + private static long reportProgress(ProgressBar totalProgress, CountingInputStream countingIn, long previousProgress) { + final long currentProgress = countingIn.getCount(); + final long progress = currentProgress - previousProgress; + if (progress > 0) { + totalProgress.addCurrentValue(progress); + } + + return currentProgress; } /** From 1c1173853cc041a8c02af3b065afa621017766a8 Mon Sep 17 00:00:00 2001 From: awildturtok Date: Wed, 30 Sep 2026 14:25:12 +0200 Subject: [PATCH 4/9] Avoid preprocessing trace and intern overhead --- .../com/bakdata/conquery/models/preproc/GroovyPredicate.java | 4 +++- .../com/bakdata/conquery/models/preproc/Preprocessed.java | 4 +++- .../bakdata/conquery/models/preproc/outputs/CopyOutput.java | 4 +++- .../com/bakdata/conquery/models/preproc/parser/Parser.java | 4 +++- .../models/preproc/parser/specific/DecimalParser.java | 4 +++- .../conquery/models/preproc/parser/specific/StringParser.java | 2 +- 6 files changed, 16 insertions(+), 6 deletions(-) diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/GroovyPredicate.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/GroovyPredicate.java index 69083867f4..f18c39940e 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/GroovyPredicate.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/GroovyPredicate.java @@ -11,7 +11,9 @@ public abstract class GroovyPredicate extends Script { public boolean filterRow(String[] row){ setProperty("row", row); - log.trace("Row = {}, length = {}", row, row.length); + if (log.isTraceEnabled()) { + log.trace("Row = {}, length = {}", row, row.length); + } return run(); } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java index e5d6138852..1ac2bcf229 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java @@ -279,7 +279,9 @@ public synchronized void addRow(String primaryId, PPColumn[] columns, Object[] o throw new IllegalStateException("Columns are not aligned"); } - log.trace("Registering `{}` for Column[{}]", outRow[col], columns[col].getName()); + if (log.isTraceEnabled()) { + log.trace("Registering `{}` for Column[{}]", outRow[col], columns[col].getName()); + } columns[col].getParser().addLine(outRow[col]); } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java index 30011f9dbf..2b52655ac4 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java @@ -56,7 +56,9 @@ public Output createForHeaders(Object2IntArrayMap headers, DateReader da return new Output() { @Override protected Object parseLine(String[] row, Parser type, long sourceLine) throws ParsingException { - log.trace("Registering `{}` in line {} for Output[{}]", row[column], sourceLine, this.getDescription().getName()); + if (log.isTraceEnabled()) { + log.trace("Registering `{}` in line {} for Output[{}]", row[column], sourceLine, this.getDescription().getName()); + } if (row[column] == null) { return null; diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java index ec0bd296c3..11183b162a 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java @@ -93,7 +93,9 @@ protected STORE_TYPE decideType() { */ public MAJOR_JAVA_TYPE addLine(MAJOR_JAVA_TYPE v) { lines++; - log.trace("Registering `{}` in line {}", v, lines); + if (log.isTraceEnabled()) { + log.trace("Registering `{}` in line {}", v, lines); + } if (v == null) { nullLines++; diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java index f1f0f34ae7..7363415e54 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java @@ -33,7 +33,9 @@ protected BigDecimal parseValue(String value) throws ParsingException { @Override protected void registerValue(BigDecimal v) { - log.trace("Registering `{}`", v); + if (log.isTraceEnabled()) { + log.trace("Registering `{}`", v); + } BigDecimal abs = v.abs(); if (v.scale() > maxScale) { diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/StringParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/StringParser.java index d3036fd707..75a310ac00 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/StringParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/StringParser.java @@ -31,7 +31,7 @@ public StringParser(ConqueryConfig config) { @Override protected String parseValue(String value) throws ParsingException { - return value.intern(); + return value; } @Override From 6428feac320769cc2a304c6e708eed6ab80a559a Mon Sep 17 00:00:00 2001 From: awildturtok Date: Wed, 30 Sep 2026 14:29:00 +0200 Subject: [PATCH 5/9] Enforce thread-confined preprocessing writes --- .../conquery/models/preproc/Preprocessed.java | 43 ++++++++++++++----- .../conquery/models/preproc/Preprocessor.java | 2 +- .../models/preproc/parser/ColumnValues.java | 4 ++ 3 files changed, 37 insertions(+), 12 deletions(-) diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java index 1ac2bcf229..6dd99b24c3 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java @@ -29,10 +29,14 @@ import it.unimi.dsi.fastutil.objects.Object2IntAVLTreeMap; import it.unimi.dsi.fastutil.objects.Object2IntMap; import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap; -import lombok.Data; +import lombok.AccessLevel; +import lombok.Getter; import lombok.extern.slf4j.Slf4j; -@Data +/** + * Mutable, thread-confined accumulator for a single preprocessing job. + */ +@Getter @Slf4j public class Preprocessed { @@ -55,6 +59,8 @@ public class Preprocessed { private long rows; private boolean finalized; + @Getter(AccessLevel.NONE) + private final Thread ownerThread = Thread.currentThread(); private static final class BucketData { private final int bucketId; @@ -105,8 +111,8 @@ public Preprocessed(ConqueryConfig config, PreprocessingJob preprocessingJob) th } - public synchronized void write(File file, int buckets) throws IOException { - ensureNotFinalized(); + public void write(File file, int buckets) throws IOException { + ensureMutable(); finalized = true; final Object2IntMap entityLength = new Object2IntAVLTreeMap<>(); @@ -254,25 +260,34 @@ private static void writePreprocessed(File file, PreprocessedHeader header, Dequ } } - public synchronized String addPrimary(String primary) { - ensureNotFinalized(); + String addPrimary(String primary) { + ensureMutable(); return primaryColumn.addLine(primary); } - public synchronized void addRow(String primaryId, PPColumn[] columns, Object[] outRow) { - ensureNotFinalized(); + void addRow(String primaryId, Object[] outRow) { + ensureMutable(); + if (outRow.length != columns.length || values.length != columns.length) { + throw new IllegalArgumentException("Output row and columns are not aligned"); + } + final int event = rowEntities.size(); - rowEntities.add(primaryId); for (int col = 0; col < outRow.length; col++) { - if (values[col] == null && outRow[col] != null) { throw new IllegalStateException(String.format("Expecting %s to be NULL, because no ColumnValues could be generated by the associated parser", outRow[col])); } + if (values[col] != null && values[col].getSize() != event) { + throw new IllegalStateException("Columns are not aligned"); + } + } + + for (int col = 0; col < outRow.length; col++) { if (values[col] == null) { continue; } + final int idx = values[col].add(outRow[col]); if (event != idx) { @@ -285,11 +300,17 @@ public synchronized void addRow(String primaryId, PPColumn[] columns, Object[] o columns[col].getParser().addLine(outRow[col]); } + rowEntities.add(primaryId); + //update stats rows++; } - private void ensureNotFinalized() { + private void ensureMutable() { + if (Thread.currentThread() != ownerThread) { + throw new IllegalStateException("Preprocessed data must only be accessed by its owning thread"); + } + if (finalized) { throw new IllegalStateException("Preprocessed data has already been finalized"); } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java index 82ae2034eb..d1843e0e85 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java @@ -162,7 +162,7 @@ public static void preprocess(PreprocessingJob preprocessingJob, ProgressBar tot final String primary = result.addPrimary(primaryId); applyOutputs(outputs, columns, row, lineId, outRow); - result.addRow(primary, columns, outRow); + result.addRow(primary, outRow); } catch (OutputDescription.OutputException e) { diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/ColumnValues.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/ColumnValues.java index c2857cc171..45806887a2 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/ColumnValues.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/ColumnValues.java @@ -39,4 +39,8 @@ public final int add(T value) { protected abstract int size(); + public final int getSize() { + return size(); + } + } From 10d626c6aa24a66506f74f16071673668c9ef339 Mon Sep 17 00:00:00 2001 From: awildturtok Date: Wed, 30 Sep 2026 14:44:53 +0200 Subject: [PATCH 6/9] Stage primitive preprocessing values without boxing --- .../conquery/models/preproc/OutputRow.java | 111 ++++++++++++++++++ .../conquery/models/preproc/Preprocessed.java | 18 +-- .../conquery/models/preproc/Preprocessor.java | 10 +- .../models/preproc/outputs/CopyOutput.java | 15 ++- .../models/preproc/outputs/EpochOutput.java | 11 ++ .../models/preproc/outputs/LineOutput.java | 6 + .../preproc/outputs/OutputDescription.java | 16 +++ .../models/preproc/parser/ColumnValues.java | 14 +++ .../models/preproc/parser/Parser.java | 60 ++++++++-- .../parser/specific/BooleanColumnValues.java | 12 ++ .../parser/specific/BooleanParser.java | 26 ++++ .../preproc/parser/specific/DateParser.java | 29 +++++ .../parser/specific/DoubleColumnValues.java | 9 ++ .../parser/specific/IntegerColumnValues.java | 9 ++ .../parser/specific/IntegerParser.java | 39 ++++++ .../parser/specific/LongColumnValues.java | 9 ++ .../preproc/parser/specific/RealParser.java | 34 +++++- 17 files changed, 403 insertions(+), 25 deletions(-) create mode 100644 backend/src/main/java/com/bakdata/conquery/models/preproc/OutputRow.java diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/OutputRow.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/OutputRow.java new file mode 100644 index 0000000000..91783a91ed --- /dev/null +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/OutputRow.java @@ -0,0 +1,111 @@ +package com.bakdata.conquery.models.preproc; + +import java.util.Arrays; + +/** + * Reusable staging area for a parsed output row. Primitive representations avoid boxing values before a row is committed. + */ +public final class OutputRow { + + private static final byte EMPTY = 0; + private static final byte NULL = 1; + private static final byte OBJECT = 2; + private static final byte LONG = 3; + private static final byte DOUBLE = 4; + private static final byte BOOLEAN = 5; + + private final byte[] representations; + private final Object[] objects; + private final long[] longs; + private final double[] doubles; + private final boolean[] booleans; + + public OutputRow(int size) { + representations = new byte[size]; + objects = new Object[size]; + longs = new long[size]; + doubles = new double[size]; + booleans = new boolean[size]; + } + + public int size() { + return representations.length; + } + + public boolean isNull(int index) { + return representations[index] == NULL; + } + + public void setNull(int index) { + representations[index] = NULL; + objects[index] = null; + } + + public void setObject(int index, Object value) { + if (value == null) { + setNull(index); + return; + } + + representations[index] = OBJECT; + objects[index] = value; + } + + public void setLong(int index, long value) { + representations[index] = LONG; + longs[index] = value; + objects[index] = null; + } + + public void setDouble(int index, double value) { + representations[index] = DOUBLE; + doubles[index] = value; + objects[index] = null; + } + + public void setBoolean(int index, boolean value) { + representations[index] = BOOLEAN; + booleans[index] = value; + objects[index] = null; + } + + public Object getObject(int index) { + return switch (representations[index]) { + case NULL, EMPTY -> null; + case OBJECT -> objects[index]; + case LONG -> longs[index]; + case DOUBLE -> doubles[index]; + case BOOLEAN -> booleans[index]; + default -> throw new IllegalStateException("Unknown output representation"); + }; + } + + public long getLong(int index) { + if (representations[index] == LONG) { + return longs[index]; + } + + return ((Number) objects[index]).longValue(); + } + + public double getDouble(int index) { + if (representations[index] == DOUBLE) { + return doubles[index]; + } + + return ((Number) objects[index]).doubleValue(); + } + + public boolean getBoolean(int index) { + if (representations[index] == BOOLEAN) { + return booleans[index]; + } + + return (Boolean) objects[index]; + } + + public void clear() { + Arrays.fill(representations, EMPTY); + Arrays.fill(objects, null); + } +} diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java index 6dd99b24c3..40d49b51b8 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java @@ -265,17 +265,17 @@ String addPrimary(String primary) { return primaryColumn.addLine(primary); } - void addRow(String primaryId, Object[] outRow) { + void addRow(String primaryId, OutputRow outRow) { ensureMutable(); - if (outRow.length != columns.length || values.length != columns.length) { + if (outRow.size() != columns.length || values.length != columns.length) { throw new IllegalArgumentException("Output row and columns are not aligned"); } final int event = rowEntities.size(); - for (int col = 0; col < outRow.length; col++) { - if (values[col] == null && outRow[col] != null) { - throw new IllegalStateException(String.format("Expecting %s to be NULL, because no ColumnValues could be generated by the associated parser", outRow[col])); + for (int col = 0; col < outRow.size(); col++) { + if (values[col] == null && !outRow.isNull(col)) { + throw new IllegalStateException(String.format("Expecting %s to be NULL, because no ColumnValues could be generated by the associated parser", outRow.getObject(col))); } if (values[col] != null && values[col].getSize() != event) { @@ -283,21 +283,21 @@ void addRow(String primaryId, Object[] outRow) { } } - for (int col = 0; col < outRow.length; col++) { + for (int col = 0; col < outRow.size(); col++) { if (values[col] == null) { continue; } - final int idx = values[col].add(outRow[col]); + final int idx = values[col].add(outRow, col); if (event != idx) { throw new IllegalStateException("Columns are not aligned"); } if (log.isTraceEnabled()) { - log.trace("Registering `{}` for Column[{}]", outRow[col], columns[col].getName()); + log.trace("Registering `{}` for Column[{}]", outRow.getObject(col), columns[col].getName()); } - columns[col].getParser().addLine(outRow[col]); + columns[col].getParser().addLine(outRow, col); } rowEntities.add(primaryId); diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java index d1843e0e85..fead180c8c 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java @@ -7,7 +7,6 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; -import java.util.Arrays; import java.util.List; import java.util.Objects; import java.util.Optional; @@ -137,7 +136,7 @@ public static void preprocess(PreprocessingJob preprocessingJob, ProgressBar tot outputs.add(op.createForHeaders(headerMap, dateReader, config)); } - final Object[] outRow = new Object[outputs.size()]; + final OutputRow outRow = new OutputRow(outputs.size()); String[] row; // Read all CSV lines, apply Output transformations and add the to preprocessed. @@ -236,7 +235,7 @@ else if (errors == config.getPreprocessor().getMaximumPrintedErrors()) { /** * Apply each output for a single row into the reusable output buffer. */ - private static void applyOutputs(List outputs, PPColumn[] columns, String[] row, long lineId, Object[] outRow) + private static void applyOutputs(List outputs, PPColumn[] columns, String[] row, long lineId, OutputRow outRow) throws ParsingException, OutputDescription.OutputException { for (int index = 0; index < outputs.size(); index++) { final OutputDescription.Output out = outputs.get(index); @@ -244,11 +243,10 @@ private static void applyOutputs(List outputs, PPColum try { final Parser parser = columns[index].getParser(); - final Object result = out.createOutput(row, parser, lineId); - outRow[index] = result; + out.createOutput(row, parser, lineId, outRow, index); } catch (Exception e) { - Arrays.fill(outRow, null); + outRow.clear(); throw new OutputDescription.OutputException(out.getDescription(), e); } } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java index 2b52655ac4..ac9aa093c6 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java @@ -6,6 +6,7 @@ import com.bakdata.conquery.models.config.ConqueryConfig; import com.bakdata.conquery.models.events.MajorTypeId; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.util.DateReader; import it.unimi.dsi.fastutil.objects.Object2IntArrayMap; @@ -54,11 +55,15 @@ public Output createForHeaders(Object2IntArrayMap headers, DateReader da final int column = headers.getInt(inputColumn); return new Output() { - @Override - protected Object parseLine(String[] row, Parser type, long sourceLine) throws ParsingException { + private void trace(String[] row, long sourceLine) { if (log.isTraceEnabled()) { log.trace("Registering `{}` in line {} for Output[{}]", row[column], sourceLine, this.getDescription().getName()); } + } + + @Override + protected Object parseLine(String[] row, Parser type, long sourceLine) throws ParsingException { + trace(row, sourceLine); if (row[column] == null) { return null; @@ -66,6 +71,12 @@ protected Object parseLine(String[] row, Parser type, long sourceLine) throws Pa return type.parse(row[column]); } + + @Override + protected void parseLine(String[] row, Parser type, long sourceLine, OutputRow outputRow, int outputIndex) throws ParsingException { + trace(row, sourceLine); + type.parse(row[column], outputRow, outputIndex); + } }; } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/EpochOutput.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/EpochOutput.java index 510c878c36..30b7d4703d 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/EpochOutput.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/EpochOutput.java @@ -6,6 +6,7 @@ import com.bakdata.conquery.models.config.ConqueryConfig; import com.bakdata.conquery.models.events.MajorTypeId; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.models.preproc.parser.specific.DateParser; import com.bakdata.conquery.util.DateReader; @@ -48,6 +49,16 @@ protected Object parseLine(String[] row, Parser type, long sourceLine) throws Pa return Integer.parseInt(row[columnIndex]); } + + @Override + protected void parseLine(String[] row, Parser type, long sourceLine, OutputRow outputRow, int outputIndex) { + if (row[columnIndex] == null) { + outputRow.setNull(outputIndex); + return; + } + + outputRow.setLong(outputIndex, Integer.parseInt(row[columnIndex])); + } }; } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/LineOutput.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/LineOutput.java index d1e6961d60..65616db471 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/LineOutput.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/LineOutput.java @@ -4,6 +4,7 @@ import com.bakdata.conquery.models.config.ConqueryConfig; import com.bakdata.conquery.models.events.MajorTypeId; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.models.preproc.parser.specific.IntegerParser; import com.bakdata.conquery.util.DateReader; @@ -38,6 +39,11 @@ public Output createForHeaders(Object2IntArrayMap headers, DateReader da protected Object parseLine(String[] row, Parser type, long sourceLine) throws ParsingException { return sourceLine; } + + @Override + protected void parseLine(String[] row, Parser type, long sourceLine, OutputRow outputRow, int outputIndex) { + outputRow.setLong(outputIndex, sourceLine); + } }; } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/OutputDescription.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/OutputDescription.java index 6bbf80bb4e..0d6d817529 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/OutputDescription.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/OutputDescription.java @@ -10,6 +10,7 @@ import com.bakdata.conquery.models.events.MajorTypeId; import com.bakdata.conquery.models.exceptions.ParsingException; import com.bakdata.conquery.models.preproc.ColumnDescription; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.TableInputDescriptor; import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.util.DateReader; @@ -83,6 +84,10 @@ public OutputDescription getDescription() { */ protected abstract Object parseLine(String[] row, Parser type, long sourceLine) throws ParsingException; + protected void parseLine(String[] row, Parser type, long sourceLine, OutputRow outputRow, int outputIndex) throws ParsingException { + outputRow.setObject(outputIndex, parseLine(row, type, sourceLine)); + } + /** * Parse row and test for NULL values, throwing an exception when Required but missing. */ @@ -95,6 +100,17 @@ public Object createOutput(String[] row, Parser type, long sourceLine) throws Pa return parsed; } + + /** + * Parse into a reusable row buffer and test for required null values. + */ + public void createOutput(String[] row, Parser type, long sourceLine, OutputRow outputRow, int outputIndex) throws ParsingException { + parseLine(row, type, sourceLine, outputRow, outputIndex); + + if (OutputDescription.this.isRequired() && outputRow.isNull(outputIndex)) { + throw new IllegalArgumentException(String.format("Required Output[%s] produced NULL value at line %d", OutputDescription.this.getName(), sourceLine)); + } + } } @Data diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/ColumnValues.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/ColumnValues.java index 45806887a2..7363924b64 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/ColumnValues.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/ColumnValues.java @@ -2,6 +2,7 @@ import java.util.BitSet; +import com.bakdata.conquery.models.preproc.OutputRow; import lombok.AccessLevel; import lombok.RequiredArgsConstructor; @@ -35,6 +36,19 @@ public final int add(T value) { return event; } + @SuppressWarnings("unchecked") + public int add(OutputRow outputRow, int outputIndex) { + return add((T) outputRow.getObject(outputIndex)); + } + + protected final int prepareAdd(boolean isNull) { + final int event = size(); + if (isNull) { + nulls.set(event); + } + return event; + } + protected abstract void append(T obj); protected abstract int size(); diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java index 11183b162a..994c1b90be 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java @@ -6,6 +6,7 @@ import com.bakdata.conquery.models.events.EmptyStore; import com.bakdata.conquery.models.events.stores.root.ColumnStore; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.Setter; @@ -42,10 +43,18 @@ public final MAJOR_JAVA_TYPE parse(String v) throws ParsingException { return parseValue(v); } catch (Exception e) { - throw new ParsingException("Failed to parse '" + v + "' with " + this.getClass().getSimpleName(), e); + throw parsingException(v, e); } } + public void parse(String value, OutputRow outputRow, int outputIndex) throws ParsingException { + outputRow.setObject(outputIndex, parse(value)); + } + + protected final ParsingException parsingException(String value, Exception cause) { + return new ParsingException("Failed to parse '" + value + "' with " + this.getClass().getSimpleName(), cause); + } + /** * Read a raw CSV-value and return a parsed representation. */ @@ -92,20 +101,57 @@ protected STORE_TYPE decideType() { * @param v a parsed value. */ public MAJOR_JAVA_TYPE addLine(MAJOR_JAVA_TYPE v) { - lines++; - if (log.isTraceEnabled()) { - log.trace("Registering `{}` in line {}", v, lines); - } - if (v == null) { - nullLines++; + recordNullLine(); } else { + recordObjectLine(v); registerValue(v); } return v; } + @SuppressWarnings("unchecked") + public void addLine(OutputRow outputRow, int outputIndex) { + addLine((MAJOR_JAVA_TYPE) outputRow.getObject(outputIndex)); + } + + protected final void recordNullLine() { + lines++; + nullLines++; + if (log.isTraceEnabled()) { + log.trace("Registering `null` in line {}", lines); + } + } + + protected final void recordObjectLine(Object value) { + lines++; + if (log.isTraceEnabled()) { + log.trace("Registering `{}` in line {}", value, lines); + } + } + + protected final void recordLongLine(long value) { + lines++; + if (log.isTraceEnabled()) { + log.trace("Registering `{}` in line {}", value, lines); + } + } + + protected final void recordDoubleLine(double value) { + lines++; + if (log.isTraceEnabled()) { + log.trace("Registering `{}` in line {}", value, lines); + } + } + + protected final void recordBooleanLine(boolean value) { + lines++; + if (log.isTraceEnabled()) { + log.trace("Registering `{}` in line {}", value, lines); + } + } + /** * Register/Analyze an incoming value for {@link #decideType()}. */ diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanColumnValues.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanColumnValues.java index 745d9da523..fd1de2674c 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanColumnValues.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanColumnValues.java @@ -3,6 +3,7 @@ import java.util.BitSet; import java.util.Objects; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.ColumnValues; class BooleanColumnValues extends ColumnValues { @@ -36,4 +37,15 @@ protected void append(Boolean obj) { size++; } + @Override + public int add(OutputRow outputRow, int outputIndex) { + final boolean isNull = outputRow.isNull(outputIndex); + final int event = prepareAdd(isNull); + if (!isNull) { + values.set(size, outputRow.getBoolean(outputIndex)); + } + size++; + return event; + } + } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanParser.java index d4e21f1b8f..5595e22272 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BooleanParser.java @@ -6,6 +6,7 @@ import com.bakdata.conquery.models.events.stores.primitive.BitSetStore; import com.bakdata.conquery.models.events.stores.root.BooleanStore; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.ColumnValues; import com.bakdata.conquery.models.preproc.parser.Parser; import lombok.ToString; @@ -23,6 +24,31 @@ protected Boolean parseValue(@Nonnull String value) throws ParsingException { return parseBoolean(value); } + @Override + public void parse(String value, OutputRow outputRow, int outputIndex) throws ParsingException { + if (value == null) { + outputRow.setNull(outputIndex); + return; + } + + try { + outputRow.setBoolean(outputIndex, parseBoolean(value)); + } + catch (Exception e) { + throw parsingException(value, e); + } + } + + @Override + public void addLine(OutputRow outputRow, int outputIndex) { + if (outputRow.isNull(outputIndex)) { + recordNullLine(); + return; + } + + recordBooleanLine(outputRow.getBoolean(outputIndex)); + } + @NotNull public static Boolean parseBoolean(@NotNull String value) { return switch (value) { diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateParser.java index bdfe3e292e..deaa29415b 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DateParser.java @@ -8,6 +8,7 @@ import com.bakdata.conquery.models.events.stores.root.DateStore; import com.bakdata.conquery.models.events.stores.root.IntegerStore; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.ColumnValues; import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.util.DateReader; @@ -38,6 +39,21 @@ protected Integer parseValue(@Nonnull String value) throws ParsingException { return CDate.ofLocalDate(dateReader.parseToLocalDate(value)); } + @Override + public void parse(String value, OutputRow outputRow, int outputIndex) throws ParsingException { + if (value == null) { + outputRow.setNull(outputIndex); + return; + } + + try { + outputRow.setLong(outputIndex, CDate.ofLocalDate(dateReader.parseToLocalDate(value))); + } + catch (Exception e) { + throw parsingException(value, e); + } + } + @Override public Integer addLine(Integer v) { if(v == null){ @@ -50,6 +66,19 @@ public Integer addLine(Integer v) { return subType.addLine(v.longValue()).intValue(); } + @Override + public void addLine(OutputRow outputRow, int outputIndex) { + if (outputRow.isNull(outputIndex)) { + recordNullLine(); + subType.addNull(); + return; + } + + final long value = outputRow.getLong(outputIndex); + recordLongLine(value); + subType.addLong(value); + } + @Override protected DateStore decideType() { return decideType(getLines()); diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DoubleColumnValues.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DoubleColumnValues.java index ce9e8e98b7..2dd34c1513 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DoubleColumnValues.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DoubleColumnValues.java @@ -1,5 +1,6 @@ package com.bakdata.conquery.models.preproc.parser.specific; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.ColumnValues; import it.unimi.dsi.fastutil.doubles.DoubleBigArrayBigList; import it.unimi.dsi.fastutil.doubles.DoubleBigList; @@ -22,6 +23,14 @@ protected void append(Double obj) { values.add(obj.doubleValue()); } + @Override + public int add(OutputRow outputRow, int outputIndex) { + final boolean isNull = outputRow.isNull(outputIndex); + final int event = prepareAdd(isNull); + values.add(isNull ? Double.NaN : outputRow.getDouble(outputIndex)); + return event; + } + @Override protected int size() { return (int) values.size64(); diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerColumnValues.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerColumnValues.java index 9f5fed0508..6349b06a9d 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerColumnValues.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerColumnValues.java @@ -1,5 +1,6 @@ package com.bakdata.conquery.models.preproc.parser.specific; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.ColumnValues; import it.unimi.dsi.fastutil.ints.IntArrayList; @@ -20,6 +21,14 @@ protected void append(Integer obj) { values.add(obj.intValue()); } + @Override + public int add(OutputRow outputRow, int outputIndex) { + final boolean isNull = outputRow.isNull(outputIndex); + final int event = prepareAdd(isNull); + values.add(isNull ? 0 : (int) outputRow.getLong(outputIndex)); + return event; + } + @Override protected int size() { return (int) values.size(); diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerParser.java index 40f0cf9a5c..d3f46bc99d 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/IntegerParser.java @@ -8,6 +8,7 @@ import com.bakdata.conquery.models.events.stores.root.IntegerStore; import com.bakdata.conquery.models.events.stores.specific.RebasingIntegerStore; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.ColumnValues; import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.util.NumberParsing; @@ -33,8 +34,27 @@ protected Long parseValue(String value) throws ParsingException { return NumberParsing.parseLong(value); } + @Override + public void parse(String value, OutputRow outputRow, int outputIndex) throws ParsingException { + if (value == null) { + outputRow.setNull(outputIndex); + return; + } + + try { + outputRow.setLong(outputIndex, NumberParsing.parseLong(value)); + } + catch (Exception e) { + throw parsingException(value, e); + } + } + @Override protected void registerValue(Long v) { + registerValue(v.longValue()); + } + + private void registerValue(long v) { if (v > maxValue) { maxValue = v; } @@ -43,6 +63,25 @@ protected void registerValue(Long v) { } } + @Override + public void addLine(OutputRow outputRow, int outputIndex) { + if (outputRow.isNull(outputIndex)) { + recordNullLine(); + return; + } + + addLong(outputRow.getLong(outputIndex)); + } + + void addLong(long value) { + recordLongLine(value); + registerValue(value); + } + + void addNull() { + recordNullLine(); + } + @Override protected IntegerStore decideType() { diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/LongColumnValues.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/LongColumnValues.java index e579b44f70..358947a7fc 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/LongColumnValues.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/LongColumnValues.java @@ -1,5 +1,6 @@ package com.bakdata.conquery.models.preproc.parser.specific; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.ColumnValues; import it.unimi.dsi.fastutil.longs.LongArrayList; import it.unimi.dsi.fastutil.longs.LongList; @@ -21,6 +22,14 @@ protected void append(Long obj) { values.add(obj.longValue()); } + @Override + public int add(OutputRow outputRow, int outputIndex) { + final boolean isNull = outputRow.isNull(outputIndex); + final int event = prepareAdd(isNull); + values.add(isNull ? 0L : outputRow.getLong(outputIndex)); + return event; + } + @Override protected int size() { return values.size(); diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/RealParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/RealParser.java index 60d1b65e6f..00648b336e 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/RealParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/RealParser.java @@ -5,6 +5,7 @@ import com.bakdata.conquery.models.events.stores.primitive.FloatArrayStore; import com.bakdata.conquery.models.events.stores.root.RealStore; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.ColumnValues; import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.util.NumberParsing; @@ -30,6 +31,21 @@ protected Double parseValue(String value) throws ParsingException { return NumberParsing.parseDouble(value); } + @Override + public void parse(String value, OutputRow outputRow, int outputIndex) throws ParsingException { + if (value == null) { + outputRow.setNull(outputIndex); + return; + } + + try { + outputRow.setDouble(outputIndex, NumberParsing.parseDouble(value)); + } + catch (Exception e) { + throw parsingException(value, e); + } + } + /** * Collect ULP of all values * @@ -37,7 +53,23 @@ protected Double parseValue(String value) throws ParsingException { */ @Override protected void registerValue(Double v) { - floatULP = Math.max(floatULP, Math.ulp(v.floatValue())); + registerValue(v.doubleValue()); + } + + private void registerValue(double value) { + floatULP = Math.max(floatULP, Math.ulp((float) value)); + } + + @Override + public void addLine(OutputRow outputRow, int outputIndex) { + if (outputRow.isNull(outputIndex)) { + recordNullLine(); + return; + } + + final double value = outputRow.getDouble(outputIndex); + recordDoubleLine(value); + registerValue(value); } /** From bf20db9c2417292d2a4132e742eff452d89bca42 Mon Sep 17 00:00:00 2001 From: awildturtok Date: Wed, 30 Sep 2026 15:04:22 +0200 Subject: [PATCH 7/9] Remove preprocessing trace logging --- .../models/preproc/GroovyPredicate.java | 5 ----- .../conquery/models/preproc/Preprocessed.java | 3 --- .../models/preproc/outputs/CopyOutput.java | 11 ----------- .../conquery/models/preproc/parser/Parser.java | 17 ----------------- .../preproc/parser/specific/DecimalParser.java | 6 ------ 5 files changed, 42 deletions(-) diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/GroovyPredicate.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/GroovyPredicate.java index f18c39940e..44f2e9f853 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/GroovyPredicate.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/GroovyPredicate.java @@ -1,19 +1,14 @@ package com.bakdata.conquery.models.preproc; import groovy.lang.Script; -import lombok.extern.slf4j.Slf4j; /** * A condition that is a groovy script and thus able to represent everything. */ -@Slf4j public abstract class GroovyPredicate extends Script { public boolean filterRow(String[] row){ setProperty("row", row); - if (log.isTraceEnabled()) { - log.trace("Row = {}, length = {}", row, row.length); - } return run(); } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java index 40d49b51b8..d57e64599d 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java @@ -294,9 +294,6 @@ void addRow(String primaryId, OutputRow outRow) { throw new IllegalStateException("Columns are not aligned"); } - if (log.isTraceEnabled()) { - log.trace("Registering `{}` for Column[{}]", outRow.getObject(col), columns[col].getName()); - } columns[col].getParser().addLine(outRow, col); } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java index ac9aa093c6..4e8863c3ed 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CopyOutput.java @@ -13,7 +13,6 @@ import lombok.Data; import lombok.NoArgsConstructor; import lombok.ToString; -import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.builder.HashCodeBuilder; /** @@ -23,7 +22,6 @@ @NoArgsConstructor @ToString(of = {"inputColumn", "inputType"}) @CPSType(id = "COPY", base = OutputDescription.class) -@Slf4j public class CopyOutput extends OutputDescription { public CopyOutput(String name, String inputColumn, MajorTypeId typeId) { @@ -55,16 +53,8 @@ public Output createForHeaders(Object2IntArrayMap headers, DateReader da final int column = headers.getInt(inputColumn); return new Output() { - private void trace(String[] row, long sourceLine) { - if (log.isTraceEnabled()) { - log.trace("Registering `{}` in line {} for Output[{}]", row[column], sourceLine, this.getDescription().getName()); - } - } - @Override protected Object parseLine(String[] row, Parser type, long sourceLine) throws ParsingException { - trace(row, sourceLine); - if (row[column] == null) { return null; } @@ -74,7 +64,6 @@ protected Object parseLine(String[] row, Parser type, long sourceLine) throws Pa @Override protected void parseLine(String[] row, Parser type, long sourceLine, OutputRow outputRow, int outputIndex) throws ParsingException { - trace(row, sourceLine); type.parse(row[column], outputRow, outputIndex); } }; diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java index 994c1b90be..70a5548127 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/Parser.java @@ -11,7 +11,6 @@ import lombok.RequiredArgsConstructor; import lombok.Setter; import lombok.ToString; -import lombok.extern.slf4j.Slf4j; /** * Base class used for parsing values in Preprocessing. @@ -25,7 +24,6 @@ @Setter @RequiredArgsConstructor @ToString -@Slf4j public abstract class Parser { @ToString.Exclude @@ -119,37 +117,22 @@ public void addLine(OutputRow outputRow, int outputIndex) { protected final void recordNullLine() { lines++; nullLines++; - if (log.isTraceEnabled()) { - log.trace("Registering `null` in line {}", lines); - } } protected final void recordObjectLine(Object value) { lines++; - if (log.isTraceEnabled()) { - log.trace("Registering `{}` in line {}", value, lines); - } } protected final void recordLongLine(long value) { lines++; - if (log.isTraceEnabled()) { - log.trace("Registering `{}` in line {}", value, lines); - } } protected final void recordDoubleLine(double value) { lines++; - if (log.isTraceEnabled()) { - log.trace("Registering `{}` in line {}", value, lines); - } } protected final void recordBooleanLine(boolean value) { lines++; - if (log.isTraceEnabled()) { - log.trace("Registering `{}` in line {}", value, lines); - } } /** diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java index 7363415e54..9ab54d589c 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java @@ -13,10 +13,8 @@ import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.util.NumberParsing; import lombok.ToString; -import lombok.extern.slf4j.Slf4j; @ToString(callSuper = true) -@Slf4j public class DecimalParser extends Parser { private transient int maxScale = Integer.MIN_VALUE; @@ -33,10 +31,6 @@ protected BigDecimal parseValue(String value) throws ParsingException { @Override protected void registerValue(BigDecimal v) { - if (log.isTraceEnabled()) { - log.trace("Registering `{}`", v); - } - BigDecimal abs = v.abs(); if (v.scale() > maxScale) { maxScale = v.scale(); From c667002fc3d62b241603c5fc71240d6a75c15829 Mon Sep 17 00:00:00 2001 From: awildturtok Date: Wed, 30 Sep 2026 15:39:47 +0200 Subject: [PATCH 8/9] Make OutputRow handling strict --- .../conquery/models/preproc/OutputRow.java | 63 ++++++++++++++----- .../conquery/models/preproc/Preprocessed.java | 13 +--- .../outputs/CompoundDateRangeOutput.java | 26 ++++++-- .../specific/BigDecimalColumnValues.java | 17 +++++ .../specific/CompoundDateRangeParser.java | 11 ++++ .../parser/specific/DecimalParser.java | 30 ++++++++- .../preproc/parser/specific/MoneyParser.java | 32 +++++++++- 7 files changed, 159 insertions(+), 33 deletions(-) create mode 100644 backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BigDecimalColumnValues.java diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/OutputRow.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/OutputRow.java index 91783a91ed..400bc9072c 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/OutputRow.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/OutputRow.java @@ -1,5 +1,6 @@ package com.bakdata.conquery.models.preproc; +import java.math.BigDecimal; import java.util.Arrays; /** @@ -13,6 +14,7 @@ public final class OutputRow { private static final byte LONG = 3; private static final byte DOUBLE = 4; private static final byte BOOLEAN = 5; + private static final byte BIG_DECIMAL = 6; private final byte[] representations; private final Object[] objects; @@ -69,43 +71,76 @@ public void setBoolean(int index, boolean value) { objects[index] = null; } + public void setBigDecimal(int index, BigDecimal value) { + if (value == null) { + setNull(index); + return; + } + + representations[index] = BIG_DECIMAL; + objects[index] = value; + } + public Object getObject(int index) { return switch (representations[index]) { - case NULL, EMPTY -> null; + case NULL -> null; case OBJECT -> objects[index]; - case LONG -> longs[index]; - case DOUBLE -> doubles[index]; - case BOOLEAN -> booleans[index]; - default -> throw new IllegalStateException("Unknown output representation"); + default -> throw unexpectedRepresentation(index, OBJECT); }; } public long getLong(int index) { - if (representations[index] == LONG) { - return longs[index]; + if (representations[index] != LONG) { + throw unexpectedRepresentation(index, LONG); } - return ((Number) objects[index]).longValue(); + return longs[index]; } public double getDouble(int index) { - if (representations[index] == DOUBLE) { - return doubles[index]; + if (representations[index] != DOUBLE) { + throw unexpectedRepresentation(index, DOUBLE); } - return ((Number) objects[index]).doubleValue(); + return doubles[index]; } public boolean getBoolean(int index) { - if (representations[index] == BOOLEAN) { - return booleans[index]; + if (representations[index] != BOOLEAN) { + throw unexpectedRepresentation(index, BOOLEAN); } - return (Boolean) objects[index]; + return booleans[index]; + } + + public BigDecimal getBigDecimal(int index) { + if (representations[index] != BIG_DECIMAL) { + throw unexpectedRepresentation(index, BIG_DECIMAL); + } + + return (BigDecimal) objects[index]; } public void clear() { Arrays.fill(representations, EMPTY); Arrays.fill(objects, null); } + + private IllegalStateException unexpectedRepresentation(int index, byte expected) { + return new IllegalStateException("Expected %s at output index %d, but found %s" + .formatted(representationName(expected), index, representationName(representations[index]))); + } + + private static String representationName(byte representation) { + return switch (representation) { + case EMPTY -> "EMPTY"; + case NULL -> "NULL"; + case OBJECT -> "OBJECT"; + case LONG -> "LONG"; + case DOUBLE -> "DOUBLE"; + case BOOLEAN -> "BOOLEAN"; + case BIG_DECIMAL -> "BIG_DECIMAL"; + default -> "UNKNOWN(" + representation + ")"; + }; + } } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java index d57e64599d..de1a2e4334 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java @@ -11,6 +11,7 @@ import java.util.IntSummaryStatistics; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.TreeMap; import java.util.zip.GZIPOutputStream; @@ -106,7 +107,7 @@ public Preprocessed(ConqueryConfig config, PreprocessingJob preprocessingJob) th final Parser parser = input.getOutput()[index].createParser(config); columns[index].setParser(parser); - values[index] = parser.createColumnValues(); + values[index] = Objects.requireNonNull(parser.createColumnValues(), () -> "Parser did not create ColumnValues: " + parser.getClass().getName()); } } @@ -274,20 +275,12 @@ void addRow(String primaryId, OutputRow outRow) { final int event = rowEntities.size(); for (int col = 0; col < outRow.size(); col++) { - if (values[col] == null && !outRow.isNull(col)) { - throw new IllegalStateException(String.format("Expecting %s to be NULL, because no ColumnValues could be generated by the associated parser", outRow.getObject(col))); - } - - if (values[col] != null && values[col].getSize() != event) { + if (values[col].getSize() != event) { throw new IllegalStateException("Columns are not aligned"); } } for (int col = 0; col < outRow.size(); col++) { - if (values[col] == null) { - continue; - } - final int idx = values[col].add(outRow, col); if (event != idx) { diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CompoundDateRangeOutput.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CompoundDateRangeOutput.java index e28988a480..f625c45438 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CompoundDateRangeOutput.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/outputs/CompoundDateRangeOutput.java @@ -8,6 +8,7 @@ import com.bakdata.conquery.models.config.ConqueryConfig; import com.bakdata.conquery.models.events.MajorTypeId; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.models.preproc.parser.specific.CompoundDateRangeParser; import com.bakdata.conquery.models.preproc.parser.specific.DateParser; @@ -53,25 +54,38 @@ public Output createForHeaders(Object2IntArrayMap headers, DateReader da // This output only verifies that the parsed data is valid and present, it will not store the CDateRanges themselves // Obviously this mean doing the work twice, but it's still better than storing the data twice also. return new Output() { + private final OutputRow bounds = new OutputRow(2); + @Override protected Object parseLine(String[] row, Parser type, long sourceLine) throws ParsingException { + return parseRange(row, sourceLine); + } + + @Override + protected void parseLine(String[] row, Parser type, long sourceLine, OutputRow outputRow, int outputIndex) throws ParsingException { + outputRow.setBoolean(outputIndex, parseRange(row, sourceLine)); + } + + private boolean parseRange(String[] row, long sourceLine) throws ParsingException { + startReader.createOutput(row, dateParser, sourceLine, bounds, 0); + endReader.createOutput(row, dateParser, sourceLine, bounds, 1); - final Object start = startReader.createOutput(row, dateParser, sourceLine); - final Object end = endReader.createOutput(row, dateParser, sourceLine); + final boolean startNull = bounds.isNull(0); + final boolean endNull = bounds.isNull(1); - if (start == null && end == null) { + if (startNull && endNull) { return false; } - if (!allowOpen && (start == null || end == null)) { + if (!allowOpen && (startNull || endNull)) { throw new IllegalArgumentException("Open Ranges are not allowed."); } return // Since it's not possible that BOTH are null either of them being null already implies an open and therefore valid range. - (start == null || end == null) + (startNull || endNull) // row is included if start <= end - || (Integer) start <= (Integer) end; + || bounds.getLong(0) <= bounds.getLong(1); } }; } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BigDecimalColumnValues.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BigDecimalColumnValues.java new file mode 100644 index 0000000000..dd9ed8e199 --- /dev/null +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/BigDecimalColumnValues.java @@ -0,0 +1,17 @@ +package com.bakdata.conquery.models.preproc.parser.specific; + +import java.math.BigDecimal; + +import com.bakdata.conquery.models.preproc.OutputRow; + +class BigDecimalColumnValues extends ListColumnValues { + + @Override + public int add(OutputRow outputRow, int outputIndex) { + if (outputRow.isNull(outputIndex)) { + return add(null); + } + + return add(outputRow.getBigDecimal(outputIndex)); + } +} diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/CompoundDateRangeParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/CompoundDateRangeParser.java index 1ad2b98036..fd73ba79ff 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/CompoundDateRangeParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/CompoundDateRangeParser.java @@ -7,6 +7,7 @@ import com.bakdata.conquery.models.events.stores.root.DateRangeStore; import com.bakdata.conquery.models.events.stores.specific.CompoundDateRangeStore; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.ColumnValues; import com.bakdata.conquery.models.preproc.parser.Parser; import io.dropwizard.validation.ValidationMethod; @@ -50,6 +51,16 @@ protected Boolean parseValue(@NotNull String value) throws ParsingException { throw new IllegalStateException("this Parser cannot parse values."); } + @Override + public void addLine(OutputRow outputRow, int outputIndex) { + if (outputRow.isNull(outputIndex)) { + recordNullLine(); + return; + } + + recordBooleanLine(outputRow.getBoolean(outputIndex)); + } + @Override protected DateRangeStore decideType() { return decideType(getLines()); diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java index 9ab54d589c..1c5e9581cf 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/DecimalParser.java @@ -9,6 +9,7 @@ import com.bakdata.conquery.models.events.stores.root.IntegerStore; import com.bakdata.conquery.models.events.stores.specific.ScaledDecimalStore; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.ColumnValues; import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.util.NumberParsing; @@ -29,6 +30,33 @@ protected BigDecimal parseValue(String value) throws ParsingException { return NumberParsing.parseBig(value); } + @Override + public void parse(String value, OutputRow outputRow, int outputIndex) throws ParsingException { + if (value == null) { + outputRow.setNull(outputIndex); + return; + } + + try { + outputRow.setBigDecimal(outputIndex, NumberParsing.parseBig(value)); + } + catch (Exception e) { + throw parsingException(value, e); + } + } + + @Override + public void addLine(OutputRow outputRow, int outputIndex) { + if (outputRow.isNull(outputIndex)) { + recordNullLine(); + return; + } + + final BigDecimal value = outputRow.getBigDecimal(outputIndex); + recordObjectLine(value); + registerValue(value); + } + @Override protected void registerValue(BigDecimal v) { BigDecimal abs = v.abs(); @@ -70,7 +98,7 @@ public void setValue(DecimalStore store, int event, BigDecimal value) { @Override public ColumnValues createColumnValues() { - return new ListColumnValues(); + return new BigDecimalColumnValues(); } } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/MoneyParser.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/MoneyParser.java index acbf4527e3..240f89dd7d 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/MoneyParser.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/parser/specific/MoneyParser.java @@ -7,6 +7,7 @@ import com.bakdata.conquery.models.events.stores.root.MoneyStore; import com.bakdata.conquery.models.events.stores.specific.MoneyIntStore; import com.bakdata.conquery.models.exceptions.ParsingException; +import com.bakdata.conquery.models.preproc.OutputRow; import com.bakdata.conquery.models.preproc.parser.ColumnValues; import com.bakdata.conquery.models.preproc.parser.Parser; import com.bakdata.conquery.util.NumberParsing; @@ -29,6 +30,33 @@ protected BigDecimal parseValue(String value) throws ParsingException { return NumberParsing.parseMoney(value); } + @Override + public void parse(String value, OutputRow outputRow, int outputIndex) throws ParsingException { + if (value == null) { + outputRow.setNull(outputIndex); + return; + } + + try { + outputRow.setBigDecimal(outputIndex, NumberParsing.parseMoney(value)); + } + catch (Exception e) { + throw parsingException(value, e); + } + } + + @Override + public void addLine(OutputRow outputRow, int outputIndex) { + if (outputRow.isNull(outputIndex)) { + recordNullLine(); + return; + } + + final BigDecimal value = outputRow.getBigDecimal(outputIndex); + recordObjectLine(value); + registerValue(value); + } + @Override protected void registerValue(BigDecimal v) { if (maxValue == null){ @@ -65,8 +93,8 @@ public void setValue(MoneyStore store, int event, BigDecimal value) { } @Override - public ColumnValues createColumnValues() { - return new ListColumnValues(); + public ColumnValues createColumnValues() { + return new BigDecimalColumnValues(); } } From 2c4dc437f52a214871de0b98cbe4db0143f328c2 Mon Sep 17 00:00:00 2001 From: awildturtok Date: Wed, 21 Oct 2026 11:08:03 +0200 Subject: [PATCH 9/9] use Object2IntOpenHashMap directly to get access to faster operands --- .../conquery/models/preproc/Preprocessed.java | 8 +- .../conquery/models/preproc/Preprocessor.java | 409 +++++++++--------- 2 files changed, 201 insertions(+), 216 deletions(-) diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java index de1a2e4334..1f394ddcf4 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessed.java @@ -27,7 +27,6 @@ import com.google.common.hash.Hashing; import it.unimi.dsi.fastutil.ints.IntArrayList; import it.unimi.dsi.fastutil.ints.IntList; -import it.unimi.dsi.fastutil.objects.Object2IntAVLTreeMap; import it.unimi.dsi.fastutil.objects.Object2IntMap; import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap; import lombok.AccessLevel; @@ -116,7 +115,7 @@ public void write(File file, int buckets) throws IOException { ensureMutable(); finalized = true; - final Object2IntMap entityLength = new Object2IntAVLTreeMap<>(); + final Object2IntOpenHashMap entityLength = new Object2IntOpenHashMap<>(); calculateEntityLengths(entityLength); @@ -153,11 +152,10 @@ public static int getEntityBucket(int buckets, String id) { /** * Calculate the number of events per entity. */ - private void calculateEntityLengths(Object2IntMap entityLength) { + private void calculateEntityLengths(Object2IntOpenHashMap entityLength) { for (String entity : rowEntities) { - final int curr = entityLength.getOrDefault(entity, 0); - entityLength.put(entity, curr + 1); + entityLength.addTo(entity, 1); } } diff --git a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java index fead180c8c..25dd506dde 100644 --- a/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java +++ b/backend/src/main/java/com/bakdata/conquery/models/preproc/Preprocessor.java @@ -28,7 +28,7 @@ import com.google.common.io.CountingInputStream; import com.univocity.parsers.csv.CsvParser; import it.unimi.dsi.fastutil.objects.Object2IntArrayMap; -import it.unimi.dsi.fastutil.objects.Object2IntMap; +import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap; import lombok.experimental.UtilityClass; import lombok.extern.slf4j.Slf4j; import org.apache.commons.io.FileUtils; @@ -36,255 +36,242 @@ @Slf4j @UtilityClass public class Preprocessor { - private static final int PROGRESS_UPDATE_INTERVAL = 4096; + private static final int PROGRESS_UPDATE_INTERVAL = 4096; - /** - * Create version of file-name with tag. - */ - public static File getTaggedVersion(File file, String tag, String extension) { - if (Strings.isNullOrEmpty(tag)) { - return file; - } + /** + * Create version of file-name with tag. + */ + public static File getTaggedVersion(File file, String tag, String extension) { + if (Strings.isNullOrEmpty(tag)) { + return file; + } - return new File(file.getParentFile(), file.getName().replaceAll(Pattern.quote(extension) + "$", String.format(".%s%s", tag, extension))); - } + return new File(file.getParentFile(), file.getName().replaceAll(Pattern.quote(extension) + "$", String.format(".%s%s", tag, extension))); + } - /** - * Apply transformations in descriptor, then write them out to CQPP file for imports. - *

- * Reads CSV file, per row extracts the primary key, then applies other transformations on each row, then compresses the data with {@link ColumnStore}. - */ - public static void preprocess(PreprocessingJob preprocessingJob, ProgressBar totalProgress, ConqueryConfig config, int buckets) throws IOException { + /** + * Apply transformations in descriptor, then write them out to CQPP file for imports. + *

+ * Reads CSV file, per row extracts the primary key, then applies other transformations on each row, then compresses the data with {@link ColumnStore}. + */ + public static void preprocess(PreprocessingJob preprocessingJob, ProgressBar totalProgress, ConqueryConfig config, int buckets) throws IOException { - final File preprocessedFile = preprocessingJob.getPreprocessedFile(); - TableImportDescriptor descriptor = preprocessingJob.getDescriptor(); + final File preprocessedFile = preprocessingJob.getPreprocessedFile(); + TableImportDescriptor descriptor = preprocessingJob.getDescriptor(); - // Create temp file that will be moved when finished (we ensure the same file system, to avoid unnecessary copying) - File tmp = new File(preprocessedFile.getParentFile(), preprocessedFile.getName() + ".tmp"); + // Create temp file that will be moved when finished (we ensure the same file system, to avoid unnecessary copying) + File tmp = new File(preprocessedFile.getParentFile(), preprocessedFile.getName() + ".tmp"); - // Ensures deletion on failure - tmp.deleteOnExit(); + // Ensures deletion on failure + tmp.deleteOnExit(); - if (!Files.isWritable(tmp.getParentFile().toPath())) { - throw new IllegalArgumentException("No write permission in " + LogUtil.printPath(tmp.getParentFile())); - } + if (!Files.isWritable(tmp.getParentFile().toPath())) { + throw new IllegalArgumentException("No write permission in " + LogUtil.printPath(tmp.getParentFile())); + } - if (!Files.isWritable(preprocessedFile.toPath().getParent())) { - throw new IllegalArgumentException("No write permission in " + LogUtil.printPath(preprocessedFile - .toPath() - .getParent())); - } + if (!Files.isWritable(preprocessedFile.toPath().getParent())) { + throw new IllegalArgumentException("No write permission in " + LogUtil.printPath(preprocessedFile + .toPath() + .getParent())); + } - //delete target file if it exists - if (preprocessedFile.exists()) { - FileUtils.forceDelete(preprocessedFile); - } + //delete target file if it exists + if (preprocessedFile.exists()) { + FileUtils.forceDelete(preprocessedFile); + } - log.info("PREPROCESSING START in {}", preprocessingJob); + log.info("PREPROCESSING START in {}", preprocessingJob); - int errors = 0; + int errors = 0; - final Preprocessed result = new Preprocessed(config, preprocessingJob); + final Preprocessed result = new Preprocessed(config, preprocessingJob); - long lineId = 0; + long lineId = 0; - // Gather exception classes to get better overview of what kind of errors are happening. - Object2IntMap> exceptions = new Object2IntArrayMap<>(); - exceptions.defaultReturnValue(0); + // Gather exception classes to get better overview of what kind of errors are happening. + Object2IntOpenHashMap> exceptions = new Object2IntOpenHashMap<>(); + exceptions.defaultReturnValue(0); - for (int inputSource = 0; inputSource < descriptor.getInputs().length; inputSource++) { - final TableInputDescriptor input = descriptor.getInputs()[inputSource]; - final File sourceFile = resolveSourceFile(input.getSourceFile(), preprocessingJob.getCsvDirectory(), preprocessingJob.getTag()); + for (int inputSource = 0; inputSource < descriptor.getInputs().length; inputSource++) { + final TableInputDescriptor input = descriptor.getInputs()[inputSource]; + final File sourceFile = resolveSourceFile(input.getSourceFile(), preprocessingJob.getCsvDirectory(), preprocessingJob.getTag()); - final String name = String.format("%s:%s[%d/%s]", descriptor.toString(), descriptor.getTable(), inputSource, sourceFile.getName()); - ConqueryMDC.setLocation(name); + final String name = String.format("%s:%s[%d/%s]", descriptor.toString(), descriptor.getTable(), inputSource, sourceFile.getName()); + ConqueryMDC.setLocation(name); - if (!(sourceFile.exists() && sourceFile.canRead())) { - throw new FileNotFoundException(sourceFile.getAbsolutePath()); - } + if (!(sourceFile.exists() && sourceFile.canRead())) { + throw new FileNotFoundException(sourceFile.getAbsolutePath()); + } - CsvParser parser = null; + CsvParser parser = null; - try (CountingInputStream countingIn = new CountingInputStream(new FileInputStream(sourceFile))) { - long progress = 0; - int rowsSinceProgressUpdate = 0; + try (CountingInputStream countingIn = new CountingInputStream(new FileInputStream(sourceFile))) { + long progress = 0; + int rowsSinceProgressUpdate = 0; - CSVConfig csvSettings = config.getCsv(); - // Create CSV parser according to config, but overriding some behaviour. - parser = csvSettings.withParseHeaders(true).withSkipHeader(false).createParser(); + CSVConfig csvSettings = config.getCsv(); + // Create CSV parser according to config, but overriding some behaviour. + parser = csvSettings.withParseHeaders(true).withSkipHeader(false).createParser(); - parser.beginParsing(FileUtil.isGZipped(sourceFile) ? new GZIPInputStream(countingIn) : countingIn, csvSettings.getEncoding()); + parser.beginParsing(FileUtil.isGZipped(sourceFile) ? new GZIPInputStream(countingIn) : countingIn, csvSettings.getEncoding()); - final String[] headers = parser.getContext().parsedHeaders(); + final String[] headers = parser.getContext().parsedHeaders(); - final Object2IntArrayMap headerMap = TableInputDescriptor.buildHeaderMap(headers); + final Object2IntArrayMap headerMap = TableInputDescriptor.buildHeaderMap(headers); - // Compile filter. - final GroovyPredicate filter = input.createFilter(headers); + // Compile filter. + final GroovyPredicate filter = input.createFilter(headers); - DateReader dateReader = config.getLocale().getDateReader(); - final OutputDescription.Output primaryOut = input.getPrimary().createForHeaders(headerMap, dateReader, config); - final List outputs = new ArrayList<>(); - final PPColumn[] columns = result.getColumns(); + DateReader dateReader = config.getLocale().getDateReader(); + final OutputDescription.Output primaryOut = input.getPrimary().createForHeaders(headerMap, dateReader, config); + final List outputs = new ArrayList<>(); + final PPColumn[] columns = result.getColumns(); - // Instantiate Outputs based on descriptors (apply header positions) - for (OutputDescription op : input.getOutput()) { - outputs.add(op.createForHeaders(headerMap, dateReader, config)); - } + // Instantiate Outputs based on descriptors (apply header positions) + for (OutputDescription op : input.getOutput()) { + outputs.add(op.createForHeaders(headerMap, dateReader, config)); + } - final OutputRow outRow = new OutputRow(outputs.size()); - String[] row; + final OutputRow outRow = new OutputRow(outputs.size()); + String[] row; - // Read all CSV lines, apply Output transformations and add the to preprocessed. - while ((row = parser.parseNext()) != null) { - rowsSinceProgressUpdate++; - if (rowsSinceProgressUpdate == PROGRESS_UPDATE_INTERVAL) { - progress = reportProgress(totalProgress, countingIn, progress); - rowsSinceProgressUpdate = 0; - } + // Read all CSV lines, apply Output transformations and add the to preprocessed. + while ((row = parser.parseNext()) != null) { + rowsSinceProgressUpdate++; + if (rowsSinceProgressUpdate == PROGRESS_UPDATE_INTERVAL) { + progress = reportProgress(totalProgress, countingIn, progress); + rowsSinceProgressUpdate = 0; + } - // Check if row shall be evaluated - // This is explicitly NOT in a try-catch block as scripts may not fail and we should not recover from faulty scripts. - if (filter != null && !filter.filterRow(row)) { - continue; - } + // Check if row shall be evaluated + // This is explicitly NOT in a try-catch block as scripts may not fail and we should not recover from faulty scripts. + if (filter != null && !filter.filterRow(row)) { + continue; + } - try { - String primaryId = - (String) Objects.requireNonNull(primaryOut.createOutput(row, result.getPrimaryColumn(), lineId), "primaryId may not be null"); + try { + String primaryId = + (String) Objects.requireNonNull(primaryOut.createOutput(row, result.getPrimaryColumn(), lineId), "primaryId may not be null"); - final String primary = result.addPrimary(primaryId); - applyOutputs(outputs, columns, row, lineId, outRow); + final String primary = result.addPrimary(primaryId); + applyOutputs(outputs, columns, row, lineId, outRow); - result.addRow(primary, outRow); + result.addRow(primary, outRow); - } - catch (OutputDescription.OutputException e) { - exceptions.put(e.getCause().getClass(), exceptions.getInt(e.getCause().getClass()) + 1); + } catch (OutputDescription.OutputException e) { + exceptions.addTo(e.getCause().getClass(), 1); - errors++; + errors++; - if (log.isTraceEnabled() || errors < config.getPreprocessor().getMaximumPrintedErrors()) { - log.warn("Failed to parse `{}` from line: {} content: {}", e.getSource(), lineId, row, e.getCause()); - } - else if (errors == config.getPreprocessor().getMaximumPrintedErrors()) { - log.warn("More erroneous lines occurred. Only the first " - + config.getPreprocessor().getMaximumPrintedErrors() - + " were printed."); - } + if (log.isTraceEnabled() || errors < config.getPreprocessor().getMaximumPrintedErrors()) { + log.warn("Failed to parse `{}` from line: {} content: {}", e.getSource(), lineId, row, e.getCause()); + } else if (errors == config.getPreprocessor().getMaximumPrintedErrors()) { + log.warn("More erroneous lines occurred. Only the first {} were printed.", config.getPreprocessor().getMaximumPrintedErrors()); + } - } - catch (Exception e) { - exceptions.put(e.getClass(), exceptions.getInt(e.getClass()) + 1); - - errors++; - - if (log.isTraceEnabled() || errors < config.getPreprocessor().getMaximumPrintedErrors()) { - log.warn("Failed to parse line: {} content: {}", lineId, row, e); - } - else if (errors == config.getPreprocessor().getMaximumPrintedErrors()) { - log.warn("More erroneous lines occurred. Only the first " - + config.getPreprocessor().getMaximumPrintedErrors() - + " were printed."); - } - } - finally { - lineId++; - } - } - - reportProgress(totalProgress, countingIn, progress); - - } - finally { - if (parser != null) { - parser.stopParsing(); - } - } - } - - if (errors > 0) { - log.warn("File `{}` contained {} faulty lines of ~{} total.", preprocessingJob, errors, lineId); - } - - if (log.isWarnEnabled()) { - exceptions.forEach((clazz, count) -> log.warn("Got {} `{}`", count, clazz.getSimpleName())); - } - - result.write(tmp, buckets); - - if (errors > 0) { - log.warn("Had {}% faulty lines ({} of ~{} lines)", String.format("%.2f", 100d * (double) errors / (double) lineId), errors, lineId); - } - - if ((double) errors / (double) lineId > config.getPreprocessor().getFaultyLineThreshold()) { - throw new RuntimeException("Too many faulty lines."); - } - - - //if successful move the tmp file to the target location - FileUtils.moveFile(tmp, preprocessedFile); - log.info("PREPROCESSING DONE in {}", preprocessingJob); - } - - /** - * Apply each output for a single row into the reusable output buffer. - */ - private static void applyOutputs(List outputs, PPColumn[] columns, String[] row, long lineId, OutputRow outRow) - throws ParsingException, OutputDescription.OutputException { - for (int index = 0; index < outputs.size(); index++) { - final OutputDescription.Output out = outputs.get(index); - - try { - final Parser parser = columns[index].getParser(); - - out.createOutput(row, parser, lineId, outRow, index); - } - catch (Exception e) { - outRow.clear(); - throw new OutputDescription.OutputException(out.getDescription(), e); - } - } - } - - private static long reportProgress(ProgressBar totalProgress, CountingInputStream countingIn, long previousProgress) { - final long currentProgress = countingIn.getCount(); - final long progress = currentProgress - previousProgress; - if (progress > 0) { - totalProgress.addCurrentValue(progress); - } - - return currentProgress; - } - - /** - * Resolve a source file with tag appended if present, in csvDirectory. - */ - public static File resolveSourceFile(String fileName, Path csvDirectory, Optional tag) { - if (tag.isEmpty()) { - return csvDirectory.resolve(fileName).toFile(); - } - - String name = fileName; - final String suffix; - - if (name.endsWith(".csv.gz")) { - name = name.substring(0, name.length() - ".csv.gz".length()); - suffix = ".csv.gz"; - } - else if (name.endsWith(".csv")) { - name = name.substring(0, name.length() - ".csv".length()); - suffix = ".csv"; - } - else { - throw new IllegalArgumentException("Unknown suffix for file " + name); - } - - return csvDirectory.resolve(name + "." + tag.get() + suffix).toFile(); - } + } catch (Exception e) { + exceptions.put(e.getClass(), exceptions.getInt(e.getClass()) + 1); + + errors++; + + if (log.isTraceEnabled() || errors < config.getPreprocessor().getMaximumPrintedErrors()) { + log.warn("Failed to parse line: {} content: {}", lineId, row, e); + } else if (errors == config.getPreprocessor().getMaximumPrintedErrors()) { + log.warn("More erroneous lines occurred. Only the first {} were printed.", config.getPreprocessor().getMaximumPrintedErrors()); + } + } finally { + lineId++; + } + } + + reportProgress(totalProgress, countingIn, progress); + + } finally { + if (parser != null) { + parser.stopParsing(); + } + } + } + + if (errors > 0) { + log.warn("File `{}` contained {} faulty lines of ~{} total.", preprocessingJob, errors, lineId); + } + + if (log.isWarnEnabled()) { + exceptions.forEach((clazz, count) -> log.warn("Got {} `{}`", count, clazz.getSimpleName())); + } + + result.write(tmp, buckets); + + if (errors > 0) { + log.warn("Had {}% faulty lines ({} of ~{} lines)", String.format("%.2f", 100d * (double) errors / (double) lineId), errors, lineId); + } + + if ((double) errors / (double) lineId > config.getPreprocessor().getFaultyLineThreshold()) { + throw new RuntimeException("Too many faulty lines."); + } + + + //if successful move the tmp file to the target location + FileUtils.moveFile(tmp, preprocessedFile); + log.info("PREPROCESSING DONE in {}", preprocessingJob); + } + + /** + * Apply each output for a single row into the reusable output buffer. + */ + private static void applyOutputs(List outputs, PPColumn[] columns, String[] row, long lineId, OutputRow outRow) + throws ParsingException, OutputDescription.OutputException { + for (int index = 0; index < outputs.size(); index++) { + final OutputDescription.Output out = outputs.get(index); + + try { + final Parser parser = columns[index].getParser(); + + out.createOutput(row, parser, lineId, outRow, index); + } catch (Exception e) { + outRow.clear(); + throw new OutputDescription.OutputException(out.getDescription(), e); + } + } + } + + private static long reportProgress(ProgressBar totalProgress, CountingInputStream countingIn, long previousProgress) { + final long currentProgress = countingIn.getCount(); + final long progress = currentProgress - previousProgress; + if (progress > 0) { + totalProgress.addCurrentValue(progress); + } + + return currentProgress; + } + + /** + * Resolve a source file with tag appended if present, in csvDirectory. + */ + public static File resolveSourceFile(String fileName, Path csvDirectory, Optional tag) { + if (tag.isEmpty()) { + return csvDirectory.resolve(fileName).toFile(); + } + + String name = fileName; + final String suffix; + + if (name.endsWith(".csv.gz")) { + name = name.substring(0, name.length() - ".csv.gz".length()); + suffix = ".csv.gz"; + } else if (name.endsWith(".csv")) { + name = name.substring(0, name.length() - ".csv".length()); + suffix = ".csv"; + } else { + throw new IllegalArgumentException("Unknown suffix for file " + name); + } + + return csvDirectory.resolve(name + "." + tag.get() + suffix).toFile(); + } }