This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 59c46cf4f86 Pipe: Fixed the bug that sorter may use wrong value when
there are duplicated timestamps & Refactor (#15488)
59c46cf4f86 is described below
commit 59c46cf4f86d6ee25a231b22bce42b1a3d0e8b2d
Author: Caideyipi <[email protected]>
AuthorDate: Thu May 15 16:47:04 2025 +0800
Pipe: Fixed the bug that sorter may use wrong value when there are
duplicated timestamps & Refactor (#15488)
---
.../request/PipeTransferTabletRawReqV2.java | 2 +-
.../connector/protocol/opcua/OpcUaNameSpace.java | 2 +-
.../sorter/PipeTableModelTabletEventSorter.java | 142 +++++---------------
.../util/sorter/PipeTabletEventSorter.java | 146 +++++++++++++++------
.../sorter/PipeTreeModelTabletEventSorter.java | 60 +++------
.../pipe/connector/PipeTabletEventSorterTest.java | 72 +++++-----
6 files changed, 192 insertions(+), 232 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferTabletRawReqV2.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferTabletRawReqV2.java
index fd435ded99c..d0608cb7042 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferTabletRawReqV2.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/payload/evolvable/request/PipeTransferTabletRawReqV2.java
@@ -61,7 +61,7 @@ public class PipeTransferTabletRawReqV2 extends
PipeTransferTabletRawReq {
if (Objects.isNull(dataBaseName)) {
new
PipeTreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
} else {
- new
PipeTableModelTabletEventSorter(tablet).sortAndDeduplicateByTimestampIfNecessary();
+ new PipeTableModelTabletEventSorter(tablet).sortByTimestampIfNecessary();
}
try {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaNameSpace.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaNameSpace.java
index ac689ad8114..3af05080d80 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaNameSpace.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaNameSpace.java
@@ -138,7 +138,7 @@ public class OpcUaNameSpace extends
ManagedNamespaceWithLifecycle {
transferTabletRowForClientServerModel(
tablet.getDeviceId().split("\\."), newSchemas, timestamps, values);
} else {
- new
PipeTableModelTabletEventSorter(tablet).sortAndDeduplicateByTimestampIfNecessary();
+ new PipeTableModelTabletEventSorter(tablet).sortByTimestampIfNecessary();
final List<Integer> columnIndexes = new ArrayList<>();
for (int i = 0; i < schemas.size(); ++i) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTableModelTabletEventSorter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTableModelTabletEventSorter.java
index 4eb8672fe81..83f118c971e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTableModelTabletEventSorter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTableModelTabletEventSorter.java
@@ -23,30 +23,20 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.write.record.Tablet;
-import org.apache.tsfile.write.schema.IMeasurementSchema;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
-public class PipeTableModelTabletEventSorter {
-
- private final Tablet tablet;
-
- private Integer[] index;
- private boolean isUnSorted = false;
- private boolean hasDuplicates = false;
- private int deduplicatedSize;
+public class PipeTableModelTabletEventSorter extends PipeTabletEventSorter {
private int initIndexSize;
public PipeTableModelTabletEventSorter(final Tablet tablet) {
- this.tablet = tablet;
- deduplicatedSize = tablet == null ? 0 : tablet.getRowSize();
+ super(tablet);
+ deDuplicatedSize = tablet == null ? 0 : tablet.getRowSize();
}
/**
@@ -72,24 +62,24 @@ public class PipeTableModelTabletEventSorter {
final int deviceComparison = deviceID.compareTo(lastDevice);
if (deviceComparison == 0) {
if (previousTimestamp == currentTimestamp) {
- hasDuplicates = true;
+ isDeDuplicated = false;
continue;
}
if (previousTimestamp > currentTimestamp) {
- isUnSorted = true;
+ isSorted = false;
}
previousTimestamp = currentTimestamp;
continue;
}
if (deviceComparison < 0) {
- isUnSorted = true;
+ isSorted = false;
}
final List<Pair<Integer, Integer>> list =
deviceIDToIndexMap.computeIfAbsent(lastDevice, k -> new
ArrayList<>());
if (!list.isEmpty()) {
- isUnSorted = true;
+ isSorted = false;
}
list.add(new Pair<>(lasIndex, i));
lastDevice = deviceID;
@@ -100,85 +90,69 @@ public class PipeTableModelTabletEventSorter {
final List<Pair<Integer, Integer>> list =
deviceIDToIndexMap.computeIfAbsent(lastDevice, k -> new ArrayList<>());
if (!list.isEmpty()) {
- isUnSorted = true;
+ isSorted = false;
}
list.add(new Pair<>(lasIndex, tablet.getRowSize()));
- if (!isUnSorted && !hasDuplicates) {
+ if (isSorted && isDeDuplicated) {
return;
}
initIndexSize = 0;
- deduplicatedSize = 0;
+ deDuplicatedSize = 0;
index = new Integer[tablet.getRowSize()];
+ deDuplicatedIndex = new int[tablet.getRowSize()];
deviceIDToIndexMap.entrySet().stream()
.sorted(Map.Entry.comparingByKey())
.forEach(
entry -> {
- final int start = initIndexSize;
int i = initIndexSize;
for (Pair<Integer, Integer> pair : entry.getValue()) {
for (int j = pair.left; j < pair.right; j++) {
index[i++] = j;
}
}
- if (isUnSorted) {
- sortTimestamps(start, i);
- deduplicateTimestamps(start, i);
+ if (!isSorted) {
+ sortTimestamps(initIndexSize, i);
+ deDuplicateTimestamps(initIndexSize, i);
initIndexSize = i;
return;
}
- if (hasDuplicates) {
- deduplicateTimestamps(start, i);
+ if (!isDeDuplicated) {
+ deDuplicateTimestamps(initIndexSize, i);
}
initIndexSize = i;
});
- sortAndDeduplicateValuesAndBitMaps();
+ sortAndDeduplicateValuesAndBitMapsWithTimestamp();
}
- private void sortAndDeduplicateValuesAndBitMaps() {
- int columnIndex = 0;
+ private void sortAndDeduplicateValuesAndBitMapsWithTimestamp() {
tablet.setTimestamps(
(long[])
- PipeTabletEventSorter.reorderValueList(
- deduplicatedSize, tablet.getTimestamps(),
TSDataType.TIMESTAMP, index));
- for (int i = 0, size = tablet.getSchemas().size(); i < size; i++) {
- final IMeasurementSchema schema = tablet.getSchemas().get(i);
- if (schema != null) {
- tablet.getValues()[columnIndex] =
- PipeTabletEventSorter.reorderValueList(
- deduplicatedSize, tablet.getValues()[columnIndex],
schema.getType(), index);
- if (tablet.getBitMaps() != null && tablet.getBitMaps()[columnIndex] !=
null) {
- tablet.getBitMaps()[columnIndex] =
- PipeTabletEventSorter.reorderBitMap(
- deduplicatedSize, tablet.getBitMaps()[columnIndex], index);
- }
- columnIndex++;
- }
- }
-
- tablet.setRowSize(deduplicatedSize);
+ reorderValueListAndBitMap(tablet.getTimestamps(),
TSDataType.TIMESTAMP, null, null));
+ sortAndMayDeduplicateValuesAndBitMaps();
+ tablet.setRowSize(deDuplicatedSize);
}
private void sortTimestamps(final int startIndex, final int endIndex) {
Arrays.sort(this.index, startIndex, endIndex,
Comparator.comparingLong(tablet::getTimestamp));
}
- private void deduplicateTimestamps(final int startIndex, final int endIndex)
{
+ private void deDuplicateTimestamps(final int startIndex, final int endIndex)
{
final long[] timestamps = tablet.getTimestamps();
long lastTime = timestamps[index[startIndex]];
- index[deduplicatedSize++] = index[startIndex];
for (int i = startIndex + 1; i < endIndex; i++) {
if (lastTime != (lastTime = timestamps[index[i]])) {
- index[deduplicatedSize++] = index[i];
+ deDuplicatedIndex[deDuplicatedSize++] = i - 1;
}
}
+ deDuplicatedIndex[deDuplicatedSize++] = endIndex - 1;
}
- /** Sort by time only, and remove only rows with the same DeviceID and time.
*/
- public void sortAndDeduplicateByTimestampIfNecessary() {
+ /** Sort by time only. */
+ public void sortByTimestampIfNecessary() {
if (tablet == null || tablet.getRowSize() == 0) {
return;
}
@@ -189,15 +163,12 @@ public class PipeTableModelTabletEventSorter {
final long previousTimestamp = timestamps[i - 1];
if (currentTimestamp < previousTimestamp) {
- isUnSorted = true;
+ isSorted = false;
break;
}
- if (currentTimestamp == previousTimestamp) {
- hasDuplicates = true;
- }
}
- if (!isUnSorted && !hasDuplicates) {
+ if (isSorted) {
return;
}
@@ -206,68 +177,15 @@ public class PipeTableModelTabletEventSorter {
index[i] = i;
}
- if (isUnSorted) {
+ if (!isSorted) {
sortTimestamps();
-
- // Do deduplicate anyway.
- // isDeduplicated may be false positive when isUnSorted is true.
- deduplicateTimestamps();
- hasDuplicates = false;
}
- if (hasDuplicates) {
- deduplicateTimestamps();
- }
-
- sortAndDeduplicateValuesAndBitMapsIgnoreTimestamp();
+ sortAndMayDeduplicateValuesAndBitMaps();
}
private void sortTimestamps() {
Arrays.sort(this.index, Comparator.comparingLong(tablet::getTimestamp));
Arrays.sort(tablet.getTimestamps(), 0, tablet.getRowSize());
}
-
- private void deduplicateTimestamps() {
- deduplicatedSize = 1;
- final long[] timestamps = tablet.getTimestamps();
- long lastTime = timestamps[0];
- IDeviceID deviceID = tablet.getDeviceID(index[0]);
- final Set<IDeviceID> deviceIDSet = new HashSet<>();
- deviceIDSet.add(deviceID);
- for (int i = 1, size = tablet.getRowSize(); i < size; i++) {
- deviceID = tablet.getDeviceID(index[i]);
- if ((lastTime == (lastTime = timestamps[i]))) {
- if (!deviceIDSet.contains(deviceID)) {
- timestamps[deduplicatedSize] = lastTime;
- index[deduplicatedSize++] = index[i];
- deviceIDSet.add(deviceID);
- }
- } else {
- timestamps[deduplicatedSize] = lastTime;
- index[deduplicatedSize++] = index[i];
- deviceIDSet.clear();
- deviceIDSet.add(deviceID);
- }
- }
- }
-
- private void sortAndDeduplicateValuesAndBitMapsIgnoreTimestamp() {
- int columnIndex = 0;
- for (int i = 0, size = tablet.getSchemas().size(); i < size; i++) {
- final IMeasurementSchema schema = tablet.getSchemas().get(i);
- if (schema != null) {
- tablet.getValues()[columnIndex] =
- PipeTabletEventSorter.reorderValueList(
- deduplicatedSize, tablet.getValues()[columnIndex],
schema.getType(), index);
- if (tablet.getBitMaps() != null && tablet.getBitMaps()[columnIndex] !=
null) {
- tablet.getBitMaps()[columnIndex] =
- PipeTabletEventSorter.reorderBitMap(
- deduplicatedSize, tablet.getBitMaps()[columnIndex], index);
- }
- columnIndex++;
- }
- }
-
- tablet.setRowSize(deduplicatedSize);
- }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTabletEventSorter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTabletEventSorter.java
index f3756b2a46c..3ea470de38a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTabletEventSorter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTabletEventSorter.java
@@ -23,83 +23,153 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.write.UnSupportedDataTypeException;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
import java.time.LocalDate;
public class PipeTabletEventSorter {
- public static Object reorderValueList(
- final int deduplicatedSize,
+ protected final Tablet tablet;
+
+ protected Integer[] index;
+ protected boolean isSorted = true;
+ protected boolean isDeDuplicated = true;
+ protected int[] deDuplicatedIndex;
+ protected int deDuplicatedSize;
+
+ public PipeTabletEventSorter(final Tablet tablet) {
+ this.tablet = tablet;
+ }
+
+ // Input:
+ // Col: [1, null, 3, 6, null]
+ // Timestamp: [2, 1, 1, 1, 1]
+ // Intermediate:
+ // Index: [1, 2, 3, 4, 0]
+ // SortedTimestamp: [1, 2]
+ // DeduplicateIndex: [3, 4]
+ // Output:
+ // (Used index: [2(3), 4(0)])
+ // Col: [6, 1]
+ protected void sortAndMayDeduplicateValuesAndBitMaps() {
+ int columnIndex = 0;
+ for (int i = 0, size = tablet.getSchemas().size(); i < size; i++) {
+ final IMeasurementSchema schema = tablet.getSchemas().get(i);
+ if (schema != null) {
+ BitMap deDuplicatedBitMap = null;
+ BitMap originalBitMap = null;
+ if (tablet.getBitMaps() != null && tablet.getBitMaps()[columnIndex] !=
null) {
+ originalBitMap = tablet.getBitMaps()[columnIndex];
+ deDuplicatedBitMap = new BitMap(originalBitMap.getSize());
+ }
+
+ tablet.getValues()[columnIndex] =
+ reorderValueListAndBitMap(
+ tablet.getValues()[columnIndex],
+ schema.getType(),
+ originalBitMap,
+ deDuplicatedBitMap);
+
+ if (tablet.getBitMaps() != null && tablet.getBitMaps()[columnIndex] !=
null) {
+ tablet.getBitMaps()[columnIndex] = deDuplicatedBitMap;
+ }
+ columnIndex++;
+ }
+ }
+ }
+
+ protected Object reorderValueListAndBitMap(
final Object valueList,
final TSDataType dataType,
- final Integer[] index) {
+ final BitMap originalBitMap,
+ final BitMap deDuplicatedBitMap) {
switch (dataType) {
case BOOLEAN:
final boolean[] boolValues = (boolean[]) valueList;
- final boolean[] deduplicatedBoolValues = new
boolean[boolValues.length];
- for (int i = 0; i < deduplicatedSize; i++) {
- deduplicatedBoolValues[i] = boolValues[index[i]];
+ final boolean[] deDuplicatedBoolValues = new
boolean[boolValues.length];
+ for (int i = 0; i < deDuplicatedSize; i++) {
+ deDuplicatedBoolValues[i] =
+ boolValues[getLastNonnullIndex(i, originalBitMap,
deDuplicatedBitMap)];
}
- return deduplicatedBoolValues;
+ return deDuplicatedBoolValues;
case INT32:
final int[] intValues = (int[]) valueList;
- final int[] deduplicatedIntValues = new int[intValues.length];
- for (int i = 0; i < deduplicatedSize; i++) {
- deduplicatedIntValues[i] = intValues[index[i]];
+ final int[] deDuplicatedIntValues = new int[intValues.length];
+ for (int i = 0; i < deDuplicatedSize; i++) {
+ deDuplicatedIntValues[i] =
+ intValues[getLastNonnullIndex(i, originalBitMap,
deDuplicatedBitMap)];
}
- return deduplicatedIntValues;
+ return deDuplicatedIntValues;
case DATE:
final LocalDate[] dateValues = (LocalDate[]) valueList;
- final LocalDate[] deduplicatedDateValues = new
LocalDate[dateValues.length];
- for (int i = 0; i < deduplicatedSize; i++) {
- deduplicatedDateValues[i] = dateValues[index[i]];
+ final LocalDate[] deDuplicatedDateValues = new
LocalDate[dateValues.length];
+ for (int i = 0; i < deDuplicatedSize; i++) {
+ deDuplicatedDateValues[i] =
+ dateValues[getLastNonnullIndex(i, originalBitMap,
deDuplicatedBitMap)];
}
- return deduplicatedDateValues;
+ return deDuplicatedDateValues;
case INT64:
case TIMESTAMP:
final long[] longValues = (long[]) valueList;
- final long[] deduplicatedLongValues = new long[longValues.length];
- for (int i = 0; i < deduplicatedSize; i++) {
- deduplicatedLongValues[i] = longValues[index[i]];
+ final long[] deDuplicatedLongValues = new long[longValues.length];
+ for (int i = 0; i < deDuplicatedSize; i++) {
+ deDuplicatedLongValues[i] =
+ longValues[getLastNonnullIndex(i, originalBitMap,
deDuplicatedBitMap)];
}
- return deduplicatedLongValues;
+ return deDuplicatedLongValues;
case FLOAT:
final float[] floatValues = (float[]) valueList;
- final float[] deduplicatedFloatValues = new float[floatValues.length];
- for (int i = 0; i < deduplicatedSize; i++) {
- deduplicatedFloatValues[i] = floatValues[index[i]];
+ final float[] deDuplicatedFloatValues = new float[floatValues.length];
+ for (int i = 0; i < deDuplicatedSize; i++) {
+ deDuplicatedFloatValues[i] =
+ floatValues[getLastNonnullIndex(i, originalBitMap,
deDuplicatedBitMap)];
}
- return deduplicatedFloatValues;
+ return deDuplicatedFloatValues;
case DOUBLE:
final double[] doubleValues = (double[]) valueList;
- final double[] deduplicatedDoubleValues = new
double[doubleValues.length];
- for (int i = 0; i < deduplicatedSize; i++) {
- deduplicatedDoubleValues[i] = doubleValues[index[i]];
+ final double[] deDuplicatedDoubleValues = new
double[doubleValues.length];
+ for (int i = 0; i < deDuplicatedSize; i++) {
+ deDuplicatedDoubleValues[i] =
+ doubleValues[getLastNonnullIndex(i, originalBitMap,
deDuplicatedBitMap)];
}
- return deduplicatedDoubleValues;
+ return deDuplicatedDoubleValues;
case TEXT:
case BLOB:
case STRING:
final Binary[] binaryValues = (Binary[]) valueList;
- final Binary[] deduplicatedBinaryValues = new
Binary[binaryValues.length];
- for (int i = 0; i < deduplicatedSize; i++) {
- deduplicatedBinaryValues[i] = binaryValues[index[i]];
+ final Binary[] deDuplicatedBinaryValues = new
Binary[binaryValues.length];
+ for (int i = 0; i < deDuplicatedSize; i++) {
+ deDuplicatedBinaryValues[i] =
+ binaryValues[getLastNonnullIndex(i, originalBitMap,
deDuplicatedBitMap)];
}
- return deduplicatedBinaryValues;
+ return deDuplicatedBinaryValues;
default:
throw new UnSupportedDataTypeException(
String.format("Data type %s is not supported.", dataType));
}
}
- public static BitMap reorderBitMap(
- final int deduplicatedSize, final BitMap bitMap, final Integer[] index) {
- final BitMap deduplicatedBitMap = new BitMap(bitMap.getSize());
- for (int i = 0; i < deduplicatedSize; i++) {
- if (bitMap.isMarked(index[i])) {
- deduplicatedBitMap.mark(i);
+ private int getLastNonnullIndex(
+ final int i, final BitMap originalBitMap, final BitMap
deDuplicatedBitMap) {
+ if (deDuplicatedIndex == null) {
+ if (originalBitMap.isMarked(index[i])) {
+ deDuplicatedBitMap.mark(i);
+ }
+ return index[i];
+ }
+ if (originalBitMap == null) {
+ return index[deDuplicatedIndex[i]];
+ }
+ int lastNonnullIndex = deDuplicatedIndex[i];
+ int lastIndex = i > 0 ? deDuplicatedIndex[i - 1] : -1;
+ while (originalBitMap.isMarked(index[lastNonnullIndex])) {
+ --lastNonnullIndex;
+ if (lastNonnullIndex == lastIndex) {
+ deDuplicatedBitMap.mark(i);
+ return index[lastNonnullIndex + 1];
}
}
- return deduplicatedBitMap;
+ return index[lastNonnullIndex];
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTreeModelTabletEventSorter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTreeModelTabletEventSorter.java
index cbde0583639..a1b1923543b 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTreeModelTabletEventSorter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/util/sorter/PipeTreeModelTabletEventSorter.java
@@ -20,24 +20,15 @@
package org.apache.iotdb.db.pipe.connector.util.sorter;
import org.apache.tsfile.write.record.Tablet;
-import org.apache.tsfile.write.schema.IMeasurementSchema;
import java.util.Arrays;
import java.util.Comparator;
-public class PipeTreeModelTabletEventSorter {
-
- private final Tablet tablet;
-
- private boolean isSorted = true;
- private boolean isDeduplicated = true;
-
- private Integer[] index;
- private int deduplicatedSize;
+public class PipeTreeModelTabletEventSorter extends PipeTabletEventSorter {
public PipeTreeModelTabletEventSorter(final Tablet tablet) {
- this.tablet = tablet;
- deduplicatedSize = tablet == null ? 0 : tablet.getRowSize();
+ super(tablet);
+ deDuplicatedSize = tablet == null ? 0 : tablet.getRowSize();
}
public void deduplicateAndSortTimestampsIfNecessary() {
@@ -55,15 +46,16 @@ public class PipeTreeModelTabletEventSorter {
break;
}
if (currentTimestamp == previousTimestamp) {
- isDeduplicated = false;
+ isDeDuplicated = false;
}
}
- if (isSorted && isDeduplicated) {
+ if (isSorted && isDeDuplicated) {
return;
}
index = new Integer[tablet.getRowSize()];
+ deDuplicatedIndex = new int[tablet.getRowSize()];
for (int i = 0, size = tablet.getRowSize(); i < size; i++) {
index[i] = i;
}
@@ -71,53 +63,39 @@ public class PipeTreeModelTabletEventSorter {
if (!isSorted) {
sortTimestamps();
- // Do deduplicate anyway.
- // isDeduplicated may be false positive when isSorted is false.
+ // Do deDuplicated anyway.
+ // isDeDuplicated may be false positive when isSorted is false.
deduplicateTimestamps();
- isDeduplicated = true;
+ isDeDuplicated = true;
}
- if (!isDeduplicated) {
+ if (!isDeDuplicated) {
deduplicateTimestamps();
}
- sortAndDeduplicateValuesAndBitMaps();
+ sortAndMayDeduplicateValuesAndBitMaps();
}
private void sortTimestamps() {
+ // Index is sorted stably because it is Integer[]
Arrays.sort(index, Comparator.comparingLong(tablet::getTimestamp));
Arrays.sort(tablet.getTimestamps(), 0, tablet.getRowSize());
}
private void deduplicateTimestamps() {
- deduplicatedSize = 1;
+ deDuplicatedSize = 0;
long[] timestamps = tablet.getTimestamps();
for (int i = 1, size = tablet.getRowSize(); i < size; i++) {
if (timestamps[i] != timestamps[i - 1]) {
- index[deduplicatedSize] = index[i];
- timestamps[deduplicatedSize] = timestamps[i];
+ deDuplicatedIndex[deDuplicatedSize] = i - 1;
+ timestamps[deDuplicatedSize] = timestamps[i - 1];
- ++deduplicatedSize;
+ ++deDuplicatedSize;
}
}
- tablet.setRowSize(deduplicatedSize);
- }
- private void sortAndDeduplicateValuesAndBitMaps() {
- int columnIndex = 0;
- for (int i = 0, size = tablet.getSchemas().size(); i < size; i++) {
- final IMeasurementSchema schema = tablet.getSchemas().get(i);
- if (schema != null) {
- tablet.getValues()[columnIndex] =
- PipeTabletEventSorter.reorderValueList(
- deduplicatedSize, tablet.getValues()[columnIndex],
schema.getType(), index);
- if (tablet.getBitMaps() != null && tablet.getBitMaps()[columnIndex] !=
null) {
- tablet.getBitMaps()[columnIndex] =
- PipeTabletEventSorter.reorderBitMap(
- deduplicatedSize, tablet.getBitMaps()[columnIndex], index);
- }
- columnIndex++;
- }
- }
+ deDuplicatedIndex[deDuplicatedSize] = tablet.getRowSize() - 1;
+ timestamps[deDuplicatedSize] = timestamps[tablet.getRowSize() - 1];
+ tablet.setRowSize(++deDuplicatedSize);
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeTabletEventSorterTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeTabletEventSorterTest.java
index 92a4d33fcef..2745cf5d3de 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeTabletEventSorterTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/connector/PipeTabletEventSorterTest.java
@@ -103,23 +103,26 @@ public class PipeTabletEventSorterTest {
@Test
public void testTreeModelDeduplicate() {
- List<IMeasurementSchema> schemaList = new ArrayList<>();
+ final List<IMeasurementSchema> schemaList = new ArrayList<>();
schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
schemaList.add(new MeasurementSchema("s2", TSDataType.INT64));
schemaList.add(new MeasurementSchema("s3", TSDataType.INT64));
- Tablet tablet = new Tablet("root.sg.device", schemaList, 10);
+ final Tablet tablet = new Tablet("root.sg.device", schemaList, 10);
- long timestamp = 300;
+ final long timestamp = 300;
for (long i = 0; i < 10; i++) {
- int rowIndex = tablet.getRowSize();
+ final int rowIndex = tablet.getRowSize();
tablet.addTimestamp(rowIndex, timestamp);
for (int s = 0; s < 3; s++) {
- tablet.addValue(schemaList.get(s).getMeasurementName(), rowIndex,
timestamp);
+ tablet.addValue(
+ schemaList.get(s).getMeasurementName(),
+ rowIndex,
+ (i + s) % 3 != 0 ? timestamp + i : null);
}
}
- Set<Integer> indices = new HashSet<>();
+ final Set<Integer> indices = new HashSet<>();
for (int i = 0; i < 10; i++) {
indices.add((int) tablet.getTimestamp(i));
}
@@ -133,16 +136,9 @@ public class PipeTabletEventSorterTest {
Assert.assertEquals(indices.size(), tablet.getRowSize());
final long[] timestamps = Arrays.copyOfRange(tablet.getTimestamps(), 0,
tablet.getRowSize());
- for (int i = 0; i < 3; ++i) {
- Assert.assertArrayEquals(
- timestamps, Arrays.copyOfRange((long[]) tablet.getValues()[0], 0,
tablet.getRowSize()));
- }
-
- for (int i = 1; i < tablet.getRowSize(); ++i) {
- Assert.assertTrue(timestamps[i] > timestamps[i - 1]);
- for (int j = 0; j < 3; ++j) {
- Assert.assertTrue((long) tablet.getValue(i, j) > (long)
tablet.getValue(i - 1, j));
- }
+ Assert.assertEquals(timestamps[0] + 8, ((long[])
tablet.getValues()[0])[0]);
+ for (int i = 1; i < 3; ++i) {
+ Assert.assertEquals(timestamps[0] + 9, ((long[])
tablet.getValues()[i])[0]);
}
}
@@ -163,7 +159,7 @@ public class PipeTabletEventSorterTest {
}
rowIndex = tablet.getRowSize();
- tablet.addTimestamp(rowIndex, (long) rowIndex);
+ tablet.addTimestamp(rowIndex, rowIndex);
for (int s = 0; s < 3; s++) {
tablet.addValue(schemaList.get(s).getMeasurementName(), rowIndex,
(long) rowIndex);
}
@@ -230,19 +226,9 @@ public class PipeTabletEventSorterTest {
doTableModelTest(false, true);
}
- @Test
- public void testTableModelDeduplicateAndSort1() {
- doTableModelTest1(true, true);
- }
-
- @Test
- public void testTableModelDeduplicate1() {
- doTableModelTest1(true, false);
- }
-
@Test
public void testTableModelSort1() {
- doTableModelTest1(false, true);
+ doTableModelTest1();
}
public void doTableModelTest(final boolean hasDuplicates, final boolean
isUnSorted) {
@@ -268,9 +254,9 @@ public class PipeTabletEventSorterTest {
}
}
- public void doTableModelTest1(final boolean hasDuplicates, final boolean
isUnSorted) {
- final Tablet tablet = generateTablet("test", 10, hasDuplicates,
isUnSorted);
- new
PipeTableModelTabletEventSorter(tablet).sortAndDeduplicateByTimestampIfNecessary();
+ public void doTableModelTest1() {
+ final Tablet tablet = generateTablet("test", 10, false, true);
+ new PipeTableModelTabletEventSorter(tablet).sortByTimestampIfNecessary();
long[] timestamps = tablet.getTimestamps();
for (int i = 1; i < tablet.getRowSize(); i++) {
long time = timestamps[i];
@@ -343,16 +329,24 @@ public class PipeTabletEventSorterTest {
tablet.addTimestamp(rowIndex, value);
tablet.addValue(
"s0", rowIndex, new
Binary(String.valueOf(row).getBytes(StandardCharsets.UTF_8)));
- tablet.addValue("s1", rowIndex, value);
- tablet.addValue("s2", rowIndex, (value * 1.0f));
+ tablet.addValue("s1", rowIndex, hasDuplicates && j == 0 ? null :
value);
+ tablet.addValue("s2", rowIndex, hasDuplicates && j == 0 ? null :
(value * 1.0f));
tablet.addValue(
- "s3", rowIndex, new
Binary(String.valueOf(value).getBytes(StandardCharsets.UTF_8)));
- tablet.addValue("s4", rowIndex, value);
- tablet.addValue("s5", rowIndex, (int) value);
- tablet.addValue("s6", rowIndex, value * 0.1);
- tablet.addValue("s7", rowIndex, getDate((int) value));
+ "s3",
+ rowIndex,
+ hasDuplicates && j == 0
+ ? null
+ : new
Binary(String.valueOf(value).getBytes(StandardCharsets.UTF_8)));
+ tablet.addValue("s4", rowIndex, hasDuplicates && j == 0 ? null :
value);
+ tablet.addValue("s5", rowIndex, hasDuplicates && j == 0 ? null :
(int) value);
+ tablet.addValue("s6", rowIndex, hasDuplicates && j == 0 ? null :
value * 0.1);
+ tablet.addValue("s7", rowIndex, hasDuplicates && j == 0 ? null :
getDate((int) value));
tablet.addValue(
- "s8", rowIndex, new
Binary(String.valueOf(value).getBytes(StandardCharsets.UTF_8)));
+ "s8",
+ rowIndex,
+ hasDuplicates && j == 0
+ ? null
+ : new
Binary(String.valueOf(value).getBytes(StandardCharsets.UTF_8)));
rowIndex++;
tablet.setRowSize(rowIndex);
if (!hasDuplicates) {
