This is an automated email from the ASF dual-hosted git repository.
hongzhi-gao pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/develop by this push:
new 7798b1035 Optimize DataSetWithoutTimeGenerator with primitive long
heap/set (#906)
7798b1035 is described below
commit 7798b1035edcbcc16b7baa2287d8b487971d3e30
Author: Hongzhi Gao <[email protected]>
AuthorDate: Mon Aug 17 16:33:17 2026 +0800
Optimize DataSetWithoutTimeGenerator with primitive long heap/set (#906)
* 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.
* Fix spotless formatting for LongOpenHashSet and LongHeapPriorityQueue
* Address review: use Set.add return value and gate perf test
* Add fastutil license attribution for derived heap and hash set
* Remove fastutil attribution from NOTICE
---
LICENSE | 12 +
.../query/dataset/DataSetWithoutTimeGenerator.java | 23 +-
.../apache/tsfile/utils/LongHeapPriorityQueue.java | 159 +++++++++
.../org/apache/tsfile/utils/LongOpenHashSet.java | 325 +++++++++++++++++
.../dataset/DataSetWithoutTimeGeneratorTest.java | 155 ++++++++
.../LongOpenHashSetAndHeapPriorityQueueTest.java | 393 +++++++++++++++++++++
6 files changed, 1055 insertions(+), 12 deletions(-)
diff --git a/LICENSE b/LICENSE
index 1ab9c86ed..2b867a613 100644
--- a/LICENSE
+++ b/LICENSE
@@ -347,3 +347,15 @@ Project page: https://github.com/simd-everywhere/simde
License: https://github.com/simd-everywhere/simde/blob/master/COPYING (MIT)
--------------------------------------------------------------------------------
+
+The following files include code modified from fastutil project.
+
+./java/tsfile/src/main/java/org/apache/tsfile/utils/LongHeapPriorityQueue.java
+./java/tsfile/src/main/java/org/apache/tsfile/utils/LongOpenHashSet.java
+
+Copyright: (C) 2002-2022 Sebastiano Vigna
+Copyright: (C) 2003-2022 Paolo Boldi and Sebastiano Vigna
+Project page: https://github.com/vigna/fastutil
+License: https://github.com/vigna/fastutil/blob/master/LICENSE-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..10e4c933b 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);
@@ -139,14 +139,13 @@ public class DataSetWithoutTimeGenerator extends
QueryDataSet {
/** keep heap from storing duplicate time. */
private void timeHeapPut(long time) {
- if (!timeSet.contains(time)) {
- timeSet.add(time);
- timeHeap.add(time);
+ if (timeSet.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..629b44248
--- /dev/null
+++
b/java/tsfile/src/main/java/org/apache/tsfile/utils/LongHeapPriorityQueue.java
@@ -0,0 +1,159 @@
+/*
+ * 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..fb46d15f5
--- /dev/null
+++ b/java/tsfile/src/main/java/org/apache/tsfile/utils/LongOpenHashSet.java
@@ -0,0 +1,325 @@
+/*
+ * 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..413a3d5ec
--- /dev/null
+++
b/java/tsfile/src/test/java/org/apache/tsfile/utils/LongOpenHashSetAndHeapPriorityQueueTest.java
@@ -0,0 +1,393 @@
+/*
+ * 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.Assume;
+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 {
+
+ private static final String RUN_PERFORMANCE_TEST_PROPERTY =
"tsfile.runPerformanceTests";
+
+ /**
+ * 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. Gated like other TsFile perf
tests via {@code
+ * -Dtsfile.runPerformanceTests=true}.
+ */
+ @Test
+ public void testPrimitiveStructuresPerformance() {
+ Assume.assumeTrue(
+ "Set -Dtsfile.runPerformanceTests=true to run the performance test",
+ Boolean.getBoolean(RUN_PERFORMANCE_TEST_PROPERTY));
+
+ 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;
+ }
+}
