Optimize DataSetWithoutTimeGenerator with primitive long heap/set

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
+
+--------------------------------------------------------------------------------

java/tsfile/src/main/java/org/apache/tsfile/read/query/dataset/DataSetWithoutTimeGenerator.java

@@ -26,14 +26,13 @@
 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 DataSetWithoutTimeGenerator(
   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 RowRecord nextWithoutConstraint() throws IOException {
 
   /** 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;
   }

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;
+  }
+}