This is an automated email from the ASF dual-hosted git repository. hongzhi-gao pushed a commit to branch optimize/dataset-without-time-generator-primitive-heap in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 3b9cc9f90d0b6431e56a94dbb93767ace5a838d6 Author: 761417898 <[email protected]> AuthorDate: Mon Aug 17 11:12:47 2026 +0800 Optimize DataSetWithoutTimeGenerator with primitive long heap/set Replace PriorityQueue<Long>/HashSet<Long> with fastutil-derived LongHeapPriorityQueue/LongOpenHashSet to avoid boxing on the multi-way merge hot path. Update NOTICE and add regression tests. --- NOTICE | 7 + .../query/dataset/DataSetWithoutTimeGenerator.java | 20 +- .../apache/tsfile/utils/LongHeapPriorityQueue.java | 158 +++++++++ .../org/apache/tsfile/utils/LongOpenHashSet.java | 324 +++++++++++++++++ .../dataset/DataSetWithoutTimeGeneratorTest.java | 155 +++++++++ .../LongOpenHashSetAndHeapPriorityQueueTest.java | 385 +++++++++++++++++++++ 6 files changed, 1039 insertions(+), 10 deletions(-) diff --git a/NOTICE b/NOTICE index 0810a509f..4dd5905aa 100644 --- a/NOTICE +++ b/NOTICE @@ -20,3 +20,10 @@ Copyright 2001-2019 The Apache Software Foundation This product includes software developed at The Apache Software Foundation (http://www.apache.org/). + +============================================================================ + +This product includes code derived from fastutil (http://fastutil.di.unimi.it/): + Copyright (C) 2002-2022 Sebastiano Vigna + Copyright (C) 2003-2022 Paolo Boldi and Sebastiano Vigna + Licensed under the Apache License, Version 2.0 diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/DataSetWithoutTimeGenerator.java b/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/DataSetWithoutTimeGenerator.java index 4e3b870ca..37efe00af 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/DataSetWithoutTimeGenerator.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/DataSetWithoutTimeGenerator.java @@ -26,14 +26,13 @@ import org.apache.tsfile.read.common.Field; import org.apache.tsfile.read.common.Path; import org.apache.tsfile.read.common.RowRecord; import org.apache.tsfile.read.reader.series.AbstractFileSeriesReader; +import org.apache.tsfile.utils.LongHeapPriorityQueue; +import org.apache.tsfile.utils.LongOpenHashSet; import org.apache.tsfile.write.UnSupportedDataTypeException; import java.io.IOException; import java.util.ArrayList; -import java.util.HashSet; import java.util.List; -import java.util.PriorityQueue; -import java.util.Set; /** multi-way merging data set, no need to use TimeGenerator. */ public class DataSetWithoutTimeGenerator extends QueryDataSet { @@ -45,9 +44,9 @@ public class DataSetWithoutTimeGenerator extends QueryDataSet { private List<Boolean> hasDataRemaining; /** heap only need to store time. */ - private PriorityQueue<Long> timeHeap; + private LongHeapPriorityQueue timeHeap; - private Set<Long> timeSet; + private LongOpenHashSet timeSet; /** * constructor of DataSetWithoutTimeGenerator. @@ -68,8 +67,9 @@ public class DataSetWithoutTimeGenerator extends QueryDataSet { private void initHeap() throws IOException { hasDataRemaining = new ArrayList<>(); batchDataList = new ArrayList<>(); - timeHeap = new PriorityQueue<>(); - timeSet = new HashSet<>(); + int seriesCount = Math.max(paths.size(), 1); + timeHeap = new LongHeapPriorityQueue(seriesCount); + timeSet = new LongOpenHashSet(seriesCount); for (int i = 0; i < paths.size(); i++) { AbstractFileSeriesReader reader = readers.get(i); @@ -141,12 +141,12 @@ public class DataSetWithoutTimeGenerator extends QueryDataSet { private void timeHeapPut(long time) { if (!timeSet.contains(time)) { timeSet.add(time); - timeHeap.add(time); + timeHeap.enqueue(time); } } - private Long timeHeapGet() { - Long t = timeHeap.poll(); + private long timeHeapGet() { + long t = timeHeap.dequeueLong(); timeSet.remove(t); return t; } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/utils/LongHeapPriorityQueue.java b/java/tsfile/src/main/java/org/apache/tsfile/utils/LongHeapPriorityQueue.java new file mode 100644 index 000000000..c8476d7f0 --- /dev/null +++ b/java/tsfile/src/main/java/org/apache/tsfile/utils/LongHeapPriorityQueue.java @@ -0,0 +1,158 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.tsfile.utils; + +import java.util.Arrays; +import java.util.NoSuchElementException; + +/** + * A type-specific heap-based priority queue for {@code long} values (natural order). + * + * <p>Copied and trimmed from fastutil 8.5.8 (Apache License 2.0, + * http://fastutil.di.unimi.it/): + * + * <ul> + * <li>{@code it.unimi.dsi.fastutil.longs.LongHeapPriorityQueue} + * <li>{@code it.unimi.dsi.fastutil.longs.LongHeaps} ({@code upHeap}/{@code downHeap} natural-order + * branches inlined below) + * </ul> + * + * <p>Comparator / serialization / collection constructors were dropped. Array growth uses {@link + * Arrays#copyOf} instead of fastutil {@code LongArrays#grow}. + * + * <p>Copyright (C) 2003-2022 Paolo Boldi and Sebastiano Vigna + */ +public class LongHeapPriorityQueue { + + private static final long[] EMPTY = new long[0]; + + /** The heap array. Copied from fastutil {@code LongHeapPriorityQueue#heap}. */ + private long[] heap = EMPTY; + + /** The number of elements in this queue. Copied from fastutil {@code LongHeapPriorityQueue#size}. */ + private int size; + + /** Copied from fastutil {@code LongHeapPriorityQueue(int)} (natural order only). */ + public LongHeapPriorityQueue(int capacity) { + if (capacity > 0) { + this.heap = new long[capacity]; + } + } + + /** Copied from fastutil {@code LongHeapPriorityQueue()}. */ + public LongHeapPriorityQueue() { + this(0); + } + + /** Copied from fastutil {@code LongHeapPriorityQueue#enqueue(long)}. */ + public void enqueue(long x) { + if (size == heap.length) { + heap = grow(heap, size + 1); + } + heap[size++] = x; + upHeap(heap, size, size - 1); + } + + /** Copied from fastutil {@code LongHeapPriorityQueue#dequeueLong()}. */ + public long dequeueLong() { + if (size == 0) { + throw new NoSuchElementException(); + } + final long result = heap[0]; + heap[0] = heap[--size]; + if (size != 0) { + downHeap(heap, size, 0); + } + return result; + } + + /** Copied from fastutil {@code LongHeapPriorityQueue#firstLong()}. */ + public long firstLong() { + if (size == 0) { + throw new NoSuchElementException(); + } + return heap[0]; + } + + /** Copied from fastutil {@code LongHeapPriorityQueue#size()}. */ + public int size() { + return size; + } + + public boolean isEmpty() { + return size == 0; + } + + /** Copied from fastutil {@code LongHeapPriorityQueue#clear()}. */ + public void clear() { + size = 0; + } + + /** Local substitute for fastutil {@code LongArrays#grow(long[], int)}. */ + private static long[] grow(final long[] array, final int length) { + int newLength = (int) Math.min(Math.max(2L * array.length, length), Integer.MAX_VALUE - 8); + if (newLength < length) { + newLength = length; + } + return Arrays.copyOf(array, newLength); + } + + /** + * Copied from fastutil {@code LongHeaps#downHeap(long[], int, int, LongComparator)} natural-order + * branch ({@code c == null}). + */ + private static int downHeap(final long[] heap, final int size, int i) { + final long e = heap[i]; + int child; + while ((child = (i << 1) + 1) < size) { + long t = heap[child]; + final int right = child + 1; + if (right < size && heap[right] < t) { + t = heap[child = right]; + } + if (e <= t) { + break; + } + heap[i] = t; + i = child; + } + heap[i] = e; + return i; + } + + /** + * Copied from fastutil {@code LongHeaps#upHeap(long[], int, int, LongComparator)} natural-order + * branch ({@code c == null}). + */ + private static int upHeap(final long[] heap, final int size, int i) { + final long e = heap[i]; + while (i != 0) { + final int parent = (i - 1) >>> 1; + final long t = heap[parent]; + if (t <= e) { + break; + } + heap[i] = t; + i = parent; + } + heap[i] = e; + return i; + } +} diff --git a/java/tsfile/src/main/java/org/apache/tsfile/utils/LongOpenHashSet.java b/java/tsfile/src/main/java/org/apache/tsfile/utils/LongOpenHashSet.java new file mode 100644 index 000000000..1f1a8997f --- /dev/null +++ b/java/tsfile/src/main/java/org/apache/tsfile/utils/LongOpenHashSet.java @@ -0,0 +1,324 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.tsfile.utils; + +import java.util.Arrays; + +/** + * A type-specific open-addressing hash set for {@code long} values. + * + * <p>Copied and trimmed from fastutil 8.5.8 (Apache License 2.0, + * http://fastutil.di.unimi.it/): + * + * <ul> + * <li>{@code it.unimi.dsi.fastutil.longs.LongOpenHashSet} + * <li>{@code it.unimi.dsi.fastutil.HashCommon} (inlined helpers: {@code mix}, {@code arraySize}, + * {@code maxFill}, {@code nextPowerOfTwo}) + * </ul> + * + * <p>Only {@code add}/{@code contains}/{@code remove}/size ops are kept. Key {@code 0L} uses a + * dedicated null-key slot, matching fastutil. + * + * <p>Copyright (C) 2002-2022 Sebastiano Vigna + */ +public class LongOpenHashSet { + + /** Copied from fastutil {@code Hash#DEFAULT_INITIAL_SIZE}. */ + private static final int DEFAULT_INITIAL_SIZE = 16; + + /** Copied from fastutil {@code Hash#DEFAULT_LOAD_FACTOR}. */ + private static final float DEFAULT_LOAD_FACTOR = .75f; + + /** Copied from fastutil {@code HashCommon#LONG_PHI}. */ + private static final long LONG_PHI = 0x9E3779B97F4A7C15L; + + /** The array of keys. Copied from fastutil {@code LongOpenHashSet#key}. */ + private long[] key; + + /** The mask for wrapping a position counter. Copied from fastutil {@code LongOpenHashSet#mask}. */ + private int mask; + + /** + * Whether this set contains the null key ({@code 0L}). Copied from fastutil {@code + * LongOpenHashSet#containsNull}. + */ + private boolean containsNull; + + /** + * The current table size. Note that an additional element is allocated for storing the null key. + * Copied from fastutil {@code LongOpenHashSet#n}. + */ + private int n; + + /** Threshold after which we rehash. Copied from fastutil {@code LongOpenHashSet#maxFill}. */ + private int maxFill; + + /** + * We never resize below this threshold, which is the construction-time {@code n}. Copied from + * fastutil {@code LongOpenHashSet#minN}. + */ + private final int minN; + + /** + * Number of entries in the set (including the null key, if present). Copied from fastutil {@code + * LongOpenHashSet#size}. + */ + private int size; + + /** The acceptable load factor. Copied from fastutil {@code LongOpenHashSet#f}. */ + private final float f; + + /** Copied from fastutil {@code LongOpenHashSet(int, float)}. */ + public LongOpenHashSet(final int expected, final float f) { + if (f <= 0 || f >= 1) { + throw new IllegalArgumentException("Load factor must be greater than 0 and smaller than 1"); + } + if (expected < 0) { + throw new IllegalArgumentException("The expected number of elements must be nonnegative"); + } + this.f = f; + minN = n = arraySize(expected, f); + mask = n - 1; + maxFill = maxFill(n, f); + key = new long[n + 1]; + } + + /** Copied from fastutil {@code LongOpenHashSet(int)}. */ + public LongOpenHashSet(final int expected) { + this(expected, DEFAULT_LOAD_FACTOR); + } + + /** Copied from fastutil {@code LongOpenHashSet()}. */ + public LongOpenHashSet() { + this(DEFAULT_INITIAL_SIZE, DEFAULT_LOAD_FACTOR); + } + + /** Copied from fastutil {@code LongOpenHashSet#add(long)}. */ + public boolean add(final long k) { + int pos; + if (k == 0L) { + if (containsNull) { + return false; + } + containsNull = true; + } else { + long curr; + final long[] key = this.key; + if (!((curr = key[pos = (int) mix(k) & mask]) == 0L)) { + if (curr == k) { + return false; + } + while (!((curr = key[pos = (pos + 1) & mask]) == 0L)) { + if (curr == k) { + return false; + } + } + } + key[pos] = k; + } + if (size++ >= maxFill) { + rehash(arraySize(size + 1, f)); + } + return true; + } + + /** Copied from fastutil {@code LongOpenHashSet#contains(long)}. */ + public boolean contains(final long k) { + if (k == 0L) { + return containsNull; + } + long curr; + final long[] key = this.key; + int pos; + if ((curr = key[pos = (int) mix(k) & mask]) == 0L) { + return false; + } + if (k == curr) { + return true; + } + while (true) { + if ((curr = key[pos = (pos + 1) & mask]) == 0L) { + return false; + } + if (k == curr) { + return true; + } + } + } + + /** Copied from fastutil {@code LongOpenHashSet#remove(long)}. */ + public boolean remove(final long k) { + if (k == 0L) { + if (containsNull) { + return removeNullEntry(); + } + return false; + } + long curr; + final long[] key = this.key; + int pos; + if ((curr = key[pos = (int) mix(k) & mask]) == 0L) { + return false; + } + if (k == curr) { + return removeEntry(pos); + } + while (true) { + if ((curr = key[pos = (pos + 1) & mask]) == 0L) { + return false; + } + if (k == curr) { + return removeEntry(pos); + } + } + } + + /** Copied from fastutil {@code LongOpenHashSet#clear()}. */ + public void clear() { + if (size == 0) { + return; + } + size = 0; + containsNull = false; + Arrays.fill(key, 0L); + } + + public int size() { + return size; + } + + public boolean isEmpty() { + return size == 0; + } + + /** Copied from fastutil {@code LongOpenHashSet#realSize()}. */ + private int realSize() { + return containsNull ? size - 1 : size; + } + + /** Copied from fastutil {@code LongOpenHashSet#removeEntry(int)}. */ + private boolean removeEntry(final int pos) { + size--; + shiftKeys(pos); + if (n > minN && size < maxFill / 4 && n > DEFAULT_INITIAL_SIZE) { + rehash(n / 2); + } + return true; + } + + /** Copied from fastutil {@code LongOpenHashSet#removeNullEntry()}. */ + private boolean removeNullEntry() { + containsNull = false; + key[n] = 0L; + size--; + if (n > minN && size < maxFill / 4 && n > DEFAULT_INITIAL_SIZE) { + rehash(n / 2); + } + return true; + } + + /** + * Shifts left entries with the specified hash code, starting at the specified position, and + * empties the resulting free entry. + * + * <p>Copied from fastutil {@code LongOpenHashSet#shiftKeys(int)}. + */ + private void shiftKeys(int pos) { + int last; + int slot; + long curr; + final long[] key = this.key; + for (; ; ) { + pos = ((last = pos) + 1) & mask; + for (; ; ) { + if ((curr = key[pos]) == 0L) { + key[last] = 0L; + return; + } + slot = (int) mix(curr) & mask; + if (last <= pos ? last >= slot || slot > pos : last >= slot && slot > pos) { + break; + } + pos = (pos + 1) & mask; + } + key[last] = curr; + } + } + + /** Copied from fastutil {@code LongOpenHashSet#rehash(int)}. */ + private void rehash(final int newN) { + final long[] key = this.key; + final int mask = newN - 1; + final long[] newKey = new long[newN + 1]; + int i = n; + int pos; + for (int j = realSize(); j-- != 0; ) { + while (key[--i] == 0L) { + // skip empty slots + } + if (!(newKey[pos = (int) mix(key[i]) & mask] == 0L)) { + while (!(newKey[pos = (pos + 1) & mask] == 0L)) { + // probe + } + } + newKey[pos] = key[i]; + } + n = newN; + this.mask = mask; + maxFill = maxFill(n, f); + this.key = newKey; + } + + /** Copied from fastutil {@code HashCommon#mix(long)}. */ + private static long mix(final long x) { + long h = x * LONG_PHI; + h ^= h >>> 32; + return h ^ (h >>> 16); + } + + /** Copied from fastutil {@code HashCommon#arraySize(int, float)}. */ + private static int arraySize(final int expected, final float f) { + final long s = Math.max(2, nextPowerOfTwo((long) Math.ceil(expected / f))); + if (s > (1 << 30)) { + throw new IllegalArgumentException( + "Too large (" + expected + " expected elements with load factor " + f + ")"); + } + return (int) s; + } + + /** Copied from fastutil {@code HashCommon#maxFill(int, float)}. */ + private static int maxFill(final int n, final float f) { + return Math.min((int) Math.ceil(n * f), n - 1); + } + + /** Copied from fastutil {@code HashCommon#nextPowerOfTwo(long)}. */ + private static long nextPowerOfTwo(long x) { + if (x == 0) { + return 1; + } + x--; + x |= x >> 1; + x |= x >> 2; + x |= x >> 4; + x |= x >> 8; + x |= x >> 16; + return (x | x >> 32) + 1; + } +} diff --git a/java/tsfile/src/test/java/org/apache/tsfile/read/query/dataset/DataSetWithoutTimeGeneratorTest.java b/java/tsfile/src/test/java/org/apache/tsfile/read/query/dataset/DataSetWithoutTimeGeneratorTest.java new file mode 100644 index 000000000..d027f75bd --- /dev/null +++ b/java/tsfile/src/test/java/org/apache/tsfile/read/query/dataset/DataSetWithoutTimeGeneratorTest.java @@ -0,0 +1,155 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.tsfile.read.query.dataset; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IChunkMetadata; +import org.apache.tsfile.read.common.BatchData; +import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.read.common.RowRecord; +import org.apache.tsfile.read.reader.series.AbstractFileSeriesReader; + +import org.junit.Test; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +public class DataSetWithoutTimeGeneratorTest { + + @Test + public void testMultiWayMergeWithSparseTimestamps() throws IOException { + BatchData series0 = batchOf(TSDataType.INT64, new long[] {1, 3, 5}, new long[] {10, 30, 50}); + BatchData series1 = batchOf(TSDataType.INT64, new long[] {2, 3, 4}, new long[] {20, 31, 40}); + + List<Path> paths = + Arrays.asList(new Path("root.d1", "s0", true), new Path("root.d1", "s1", true)); + List<TSDataType> types = Arrays.asList(TSDataType.INT64, TSDataType.INT64); + List<AbstractFileSeriesReader> readers = + Arrays.asList(new FakeSeriesReader(series0), new FakeSeriesReader(series1)); + + DataSetWithoutTimeGenerator dataSet = new DataSetWithoutTimeGenerator(paths, types, readers); + + assertTrue(dataSet.hasNext()); + RowRecord row1 = dataSet.next(); + assertEquals(1L, row1.getTimestamp()); + assertEquals(10L, row1.getFields().get(0).getLongV()); + assertNull(row1.getFields().get(1)); + + RowRecord row2 = dataSet.next(); + assertEquals(2L, row2.getTimestamp()); + assertNull(row2.getFields().get(0)); + assertEquals(20L, row2.getFields().get(1).getLongV()); + + RowRecord row3 = dataSet.next(); + assertEquals(3L, row3.getTimestamp()); + assertEquals(30L, row3.getFields().get(0).getLongV()); + assertEquals(31L, row3.getFields().get(1).getLongV()); + + RowRecord row4 = dataSet.next(); + assertEquals(4L, row4.getTimestamp()); + assertNull(row4.getFields().get(0)); + assertEquals(40L, row4.getFields().get(1).getLongV()); + + RowRecord row5 = dataSet.next(); + assertEquals(5L, row5.getTimestamp()); + assertEquals(50L, row5.getFields().get(0).getLongV()); + assertNull(row5.getFields().get(1)); + + assertFalse(dataSet.hasNext()); + } + + @Test + public void testMultiWayMergeAcrossBatches() throws IOException { + BatchData batch1 = batchOf(TSDataType.INT32, new long[] {1, 2}, new int[] {1, 2}); + BatchData batch2 = batchOf(TSDataType.INT32, new long[] {3}, new int[] {3}); + + List<Path> paths = Collections.singletonList(new Path("root.d1", "s0", true)); + List<TSDataType> types = Collections.singletonList(TSDataType.INT32); + List<AbstractFileSeriesReader> readers = + Collections.singletonList(new FakeSeriesReader(batch1, batch2)); + + DataSetWithoutTimeGenerator dataSet = new DataSetWithoutTimeGenerator(paths, types, readers); + List<Long> times = new ArrayList<>(); + while (dataSet.hasNext()) { + times.add(dataSet.next().getTimestamp()); + } + assertEquals(Arrays.asList(1L, 2L, 3L), times); + } + + private static BatchData batchOf(TSDataType type, long[] times, long[] values) { + BatchData batchData = new BatchData(type); + for (int i = 0; i < times.length; i++) { + batchData.putLong(times[i], values[i]); + } + return batchData; + } + + private static BatchData batchOf(TSDataType type, long[] times, int[] values) { + BatchData batchData = new BatchData(type); + for (int i = 0; i < times.length; i++) { + batchData.putInt(times[i], values[i]); + } + return batchData; + } + + private static class FakeSeriesReader extends AbstractFileSeriesReader { + + private final List<BatchData> batches; + private int index; + + FakeSeriesReader(BatchData... batches) { + super(null, Collections.emptyList(), null); + this.batches = Arrays.asList(batches); + } + + @Override + public boolean hasNextBatch() { + return index < batches.size(); + } + + @Override + public BatchData nextBatch() { + return batches.get(index++); + } + + @Override + protected void initChunkReader(IChunkMetadata chunkMetaData) { + // unused in fake reader + } + + @Override + protected boolean chunkCanSkip(IChunkMetadata chunkMetaData) { + return false; + } + + @Override + public void close() { + // no-op + } + } +} diff --git a/java/tsfile/src/test/java/org/apache/tsfile/utils/LongOpenHashSetAndHeapPriorityQueueTest.java b/java/tsfile/src/test/java/org/apache/tsfile/utils/LongOpenHashSetAndHeapPriorityQueueTest.java new file mode 100644 index 000000000..9f7785f98 --- /dev/null +++ b/java/tsfile/src/test/java/org/apache/tsfile/utils/LongOpenHashSetAndHeapPriorityQueueTest.java @@ -0,0 +1,385 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.tsfile.utils; + +import org.junit.Test; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.PriorityQueue; +import java.util.Random; +import java.util.Set; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +/** + * Correctness and smoke-performance tests for the fastutil-derived {@link LongHeapPriorityQueue} + * and {@link LongOpenHashSet}. + * + * <p>Coverage split (no overlap by design): + * + * <ul> + * <li>{@link #testLongHeapPriorityQueueOrder()} — heap only + * <li>{@link #testLongOpenHashSetBasicOps()} — set API / {@code 0L} / ctor validation + * <li>{@link #testLongOpenHashSetMatchesHashSet()} — random parity vs {@link HashSet} + * <li>{@link #testLongOpenHashSetHeavyChurn()} — sustained add/remove (rehash / {@code + * shiftKeys}) + * <li>{@link #testHeapAndSetDedupPatternMatchesPriorityQueue()} — combined dedup-merge pattern + * <li>{@link #testPrimitiveStructuresPerformance()} — micro-benchmark only + * </ul> + */ +public class LongOpenHashSetAndHeapPriorityQueueTest { + + /** + * Heap-only: empty-queue errors, natural order (incl. duplicates / extremes), {@code clear} and + * growth — step-by-step parity with {@link PriorityQueue}{@code <Long>}. + */ + @Test + public void testLongHeapPriorityQueueOrder() { + LongHeapPriorityQueue empty = new LongHeapPriorityQueue(); + assertTrue(empty.isEmpty()); + assertEquals(0, empty.size()); + try { + empty.firstLong(); + fail("expected NoSuchElementException on firstLong()"); + } catch (NoSuchElementException expected) { + // ok + } + try { + empty.dequeueLong(); + fail("expected NoSuchElementException on dequeueLong()"); + } catch (NoSuchElementException expected) { + // ok + } + + LongHeapPriorityQueue heap = new LongHeapPriorityQueue(); + PriorityQueue<Long> boxed = new PriorityQueue<>(); + long[] values = + new long[] {5L, 1L, 3L, 2L, 4L, 1L, 1L, Long.MIN_VALUE, Long.MAX_VALUE, 0L, -1L}; + for (long v : values) { + heap.enqueue(v); + boxed.add(v); + assertEquals(boxed.size(), heap.size()); + assertEquals(boxed.peek().longValue(), heap.firstLong()); + } + while (!boxed.isEmpty()) { + assertEquals(boxed.poll().longValue(), heap.dequeueLong()); + assertEquals(boxed.size(), heap.size()); + if (!boxed.isEmpty()) { + assertEquals(boxed.peek().longValue(), heap.firstLong()); + } + } + + // clear + grow from zero-capacity + heap.enqueue(9L); + heap.clear(); + assertTrue(heap.isEmpty()); + boxed.clear(); + for (int i = 0; i < 64; i++) { + long v = 1000L - i; + heap.enqueue(v); + boxed.add(v); + } + List<Long> fromPrimitive = new ArrayList<>(); + List<Long> fromBoxed = new ArrayList<>(); + while (!heap.isEmpty()) { + fromPrimitive.add(heap.dequeueLong()); + fromBoxed.add(boxed.poll()); + } + assertEquals(fromBoxed, fromPrimitive); + } + + /** + * Set-only deterministic API smoke: empty ops, duplicate add, null-key {@code 0L}, extremes, + * {@code clear}, and illegal constructor arguments. Broader random parity lives in {@link + * #testLongOpenHashSetMatchesHashSet()}. + */ + @Test + public void testLongOpenHashSetBasicOps() { + LongOpenHashSet set = new LongOpenHashSet(8); + Set<Long> boxed = new HashSet<>(); + + assertTrue(set.isEmpty()); + assertFalse(set.contains(1L)); + assertEquals(boxed.remove(1L), set.remove(1L)); + + assertEquals(boxed.add(10L), set.add(10L)); + assertEquals(boxed.add(10L), set.add(10L)); + assertTrue(set.contains(10L)); + assertFalse(set.contains(11L)); + + // fastutil null-key slot + assertEquals(boxed.add(0L), set.add(0L)); + assertEquals(boxed.remove(0L), set.remove(0L)); + assertEquals(boxed.add(0L), set.add(0L)); + assertEquals(boxed.add(0L), set.add(0L)); + + for (long edge : new long[] {Long.MIN_VALUE, Long.MAX_VALUE, -1L}) { + assertEquals(boxed.add(edge), set.add(edge)); + assertEquals(boxed.contains(edge), set.contains(edge)); + } + assertEquals(boxed.size(), set.size()); + + set.clear(); + boxed.clear(); + assertTrue(set.isEmpty()); + assertFalse(set.contains(0L)); + assertEquals(boxed.add(0L), set.add(0L)); + assertEquals(1, set.size()); + + try { + new LongOpenHashSet(-1); + fail("expected IllegalArgumentException for negative expected size"); + } catch (IllegalArgumentException expected) { + // ok + } + try { + new LongOpenHashSet(8, 0f); + fail("expected IllegalArgumentException for invalid load factor"); + } catch (IllegalArgumentException expected) { + // ok + } + try { + new LongOpenHashSet(8, 1f); + fail("expected IllegalArgumentException for invalid load factor"); + } catch (IllegalArgumentException expected) { + // ok + } + } + + /** + * Set-only randomized parity with {@link HashSet}{@code <Long>}: mixed add/contains/remove, + * boundary keys, and a mid-run {@code clear}. Does not stress sliding-window rehash (see {@link + * #testLongOpenHashSetHeavyChurn()}). + */ + @Test + public void testLongOpenHashSetMatchesHashSet() { + LongOpenHashSet primitive = new LongOpenHashSet(16); + Set<Long> boxed = new HashSet<>(); + Random random = new Random(42); + long[] edges = + new long[] {0L, -1L, Long.MIN_VALUE, Long.MAX_VALUE, Integer.MAX_VALUE, Integer.MIN_VALUE}; + + for (int i = 0; i < 20_000; i++) { + long value; + int mode = i % 10; + if (mode == 0) { + value = edges[random.nextInt(edges.length)]; + } else { + value = random.nextLong(); + } + + int op = i % 5; + if (op <= 1) { + assertEquals(boxed.add(value), primitive.add(value)); + } else if (op <= 3) { + assertEquals(boxed.contains(value), primitive.contains(value)); + } else { + assertEquals(boxed.remove(value), primitive.remove(value)); + } + assertEquals(boxed.size(), primitive.size()); + + if (i == 10_000) { + primitive.clear(); + boxed.clear(); + assertEquals(0, primitive.size()); + } + } + + for (Long v : boxed) { + assertTrue(primitive.contains(v)); + } + } + + /** + * Set-only sliding-window churn (fixed live size) to exercise {@code shiftKeys} / shrink-rehash, + * including periodic {@code 0L}. Compared against {@link HashSet} for each op. + */ + @Test + public void testLongOpenHashSetHeavyChurn() { + LongOpenHashSet set = new LongOpenHashSet(8); + Set<Long> boxed = new HashSet<>(); + final int window = 16; + for (int i = 0; i < 100_000; i++) { + long addVal = (i % 32 == 0) ? 0L : i; + assertEquals(boxed.add(addVal), set.add(addVal)); + if (i >= window) { + long removeVal = ((i - window) % 32 == 0) ? 0L : (i - window); + assertEquals(boxed.remove(removeVal), set.remove(removeVal)); + } + assertEquals(boxed.size(), set.size()); + } + for (Long v : new ArrayList<>(boxed)) { + assertTrue(set.remove(v)); + } + assertTrue(set.isEmpty()); + } + + /** + * Combined heap+set dedup pattern used by {@code DataSetWithoutTimeGenerator#timeHeapPut/Get}, + * mirrored against {@link PriorityQueue}+{@link HashSet}. Covers aligned duplicate puts and a few + * representative time bases (0 / normal / near {@link Long#MAX_VALUE}). + */ + @Test + public void testHeapAndSetDedupPatternMatchesPriorityQueue() { + final int series = 64; + final int rows = 30_000; + long[] bases = new long[] {0L, 1_700_000_000_000L, Long.MAX_VALUE - series - 10}; + for (long baseTime : bases) { + runDedupMergeParity(series, rows, baseTime); + } + } + + /** + * Micro-benchmark only (not a correctness test): boxed vs primitive under the same merge pattern. + * Soft guard against catastrophic regressions. + */ + @Test + public void testPrimitiveStructuresPerformance() { + final int series = 50; + final int rows = 100_000; + final long baseTime = 1_700_000_000_000L; + + runBoxed(series, 10_000, baseTime); + runPrimitive(series, 10_000, baseTime); + + long boxedNs = runBoxed(series, rows, baseTime); + long primitiveNs = runPrimitive(series, rows, baseTime); + + System.out.printf( + "DataSetWithoutTimeGenerator heap/set microbench: boxed=%d ms, primitive=%d ms, speedup=%.2fx%n", + boxedNs / 1_000_000L, primitiveNs / 1_000_000L, (double) boxedNs / (double) primitiveNs); + + assertTrue( + "primitive path unexpectedly much slower than boxed: boxed=" + + boxedNs + + "ns primitive=" + + primitiveNs + + "ns", + primitiveNs < boxedNs * 2); + } + + private static void runDedupMergeParity(int series, int rows, long baseTime) { + LongHeapPriorityQueue primitiveHeap = new LongHeapPriorityQueue(series); + LongOpenHashSet primitiveSet = new LongOpenHashSet(series); + PriorityQueue<Long> boxedHeap = new PriorityQueue<>(); + Set<Long> boxedSet = new HashSet<>(); + + Random random = new Random(7 ^ baseTime); + long[] heads = new long[series]; + for (int i = 0; i < series; i++) { + heads[i] = baseTime + i; + put(primitiveHeap, primitiveSet, heads[i]); + putBoxed(boxedHeap, boxedSet, heads[i]); + } + + for (int row = 0; row < rows; row++) { + assertEquals(boxedHeap.peek().longValue(), primitiveHeap.firstLong()); + long min = poll(primitiveHeap, primitiveSet); + assertEquals(min, pollBoxed(boxedHeap, boxedSet)); + assertEquals(boxedSet.size(), primitiveSet.size()); + + for (int s = 0; s < series; s++) { + if (heads[s] == min) { + // step==0 produces duplicate put (dedup path) + heads[s] += (row % 5 == 0) ? 0 : 1 + random.nextInt(3); + put(primitiveHeap, primitiveSet, heads[s]); + putBoxed(boxedHeap, boxedSet, heads[s]); + } + } + assertEquals(boxedHeap.size(), primitiveHeap.size()); + } + } + + private static long runBoxed(int series, int rows, long baseTime) { + PriorityQueue<Long> heap = new PriorityQueue<>(series); + Set<Long> set = new HashSet<>(series * 2); + long[] heads = new long[series]; + for (int i = 0; i < series; i++) { + heads[i] = baseTime + i; + putBoxed(heap, set, heads[i]); + } + + long start = System.nanoTime(); + for (int row = 0; row < rows; row++) { + long min = pollBoxed(heap, set); + for (int s = 0; s < series; s++) { + if (heads[s] == min) { + heads[s] += 1; + putBoxed(heap, set, heads[s]); + } + } + } + return System.nanoTime() - start; + } + + private static long runPrimitive(int series, int rows, long baseTime) { + LongHeapPriorityQueue heap = new LongHeapPriorityQueue(series); + LongOpenHashSet set = new LongOpenHashSet(series); + long[] heads = new long[series]; + for (int i = 0; i < series; i++) { + heads[i] = baseTime + i; + put(heap, set, heads[i]); + } + + long start = System.nanoTime(); + for (int row = 0; row < rows; row++) { + long min = poll(heap, set); + for (int s = 0; s < series; s++) { + if (heads[s] == min) { + heads[s] += 1; + put(heap, set, heads[s]); + } + } + } + return System.nanoTime() - start; + } + + private static void put(LongHeapPriorityQueue heap, LongOpenHashSet set, long time) { + if (!set.contains(time)) { + set.add(time); + heap.enqueue(time); + } + } + + private static long poll(LongHeapPriorityQueue heap, LongOpenHashSet set) { + long t = heap.dequeueLong(); + set.remove(t); + return t; + } + + private static void putBoxed(PriorityQueue<Long> heap, Set<Long> set, long time) { + if (!set.contains(time)) { + set.add(time); + heap.add(time); + } + } + + private static long pollBoxed(PriorityQueue<Long> heap, Set<Long> set) { + Long t = heap.poll(); + set.remove(t); + return t; + } +}
