You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by ho...@apache.org on 2026/08/17 03:13:02 UTC
(tsfile) branch optimize/dataset-without-time-generator-primitive-heap created (now 3b9cc9f90)
This is an automated email from the ASF dual-hosted git repository.
hongzhi-gao pushed a change to branch optimize/dataset-without-time-generator-primitive-heap
in repository https://gitbox.apache.org/repos/asf/tsfile.git
at 3b9cc9f90 Optimize DataSetWithoutTimeGenerator with primitive long heap/set
This branch includes the following new commits:
new 3b9cc9f90 Optimize DataSetWithoutTimeGenerator with primitive long heap/set
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(tsfile) 01/01: Optimize DataSetWithoutTimeGenerator with primitive long heap/set
Posted by ho...@apache.org.
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 <76...@qq.com>
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;
+ }
+}