Skip to content

Commit

Permalink
Merge branch 'master' into pipe-fix-synonym
Browse files Browse the repository at this point in the history
  • Loading branch information
@VGalaxies
VGalaxies committed Jan 14, 2025
2 parents 674b97d + f129298 commit 3f20241
Show file tree
Hide file tree
Showing 145 changed files with 1,682 additions and 864 deletions.
2 changes: 2 additions & 0 deletions example/pipe-count-point-processor/src/main/java/org/apache/iotdb/CountPointProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.pipe.api.PipeProcessor;
import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.collector.EventCollector;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
Expand All @@ -37,6 +38,7 @@
import java.util.Collections;
import java.util.concurrent.atomic.AtomicLong;

@TreeModel
public class CountPointProcessor implements PipeProcessor {
private static final String AGGREGATE_SERIES_KEY = "aggregate-series";
private static final AtomicLong writePointCount = new AtomicLong(0);
Expand Down
67 changes: 5 additions & 62 deletions example/session/src/main/java/org/apache/iotdb/AlignedTimeseriesSessionExample.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
Expand Down Expand Up @@ -341,18 +340,14 @@ private static void insertTabletWithAlignedTimeseriesMethod2()
schemaList.add(new MeasurementSchema("s2", TSDataType.INT32));

Tablet tablet = new Tablet(ROOT_SG1_D1_VECTOR2, schemaList);
long[] timestamps = tablet.timestamps;
Object[] values = tablet.values;

for (long time = 100; time < 200; time++) {
int row = tablet.getRowSize();
tablet.addTimestamp(row, time);

long[] sensor1 = (long[]) values[0];
sensor1[row] = new SecureRandom().nextLong();
tablet.addValue(row, 0, new SecureRandom().nextLong());

int[] sensor2 = (int[]) values[1];
sensor2[row] = new SecureRandom().nextInt();
tablet.addValue(row, 1, new SecureRandom().nextInt());

if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
session.insertAlignedTablet(tablet, true);
Expand All @@ -378,32 +373,19 @@ private static void insertNullableTabletWithAlignedTimeseries()

Tablet tablet = new Tablet(ROOT_SG1_D1_VECTOR3, schemaList);

long[] timestamps = tablet.timestamps;
Object[] values = tablet.values;
// Use the bitMap to mark the null value point
BitMap[] bitMaps = new BitMap[values.length];
tablet.bitMaps = bitMaps;

bitMaps[1] = new BitMap(tablet.getMaxRowNumber());
for (long time = 200; time < 300; time++) {
int row = tablet.getRowSize();
tablet.addTimestamp(row, time);

long[] sensor1 = (long[]) values[0];
sensor1[row] = new SecureRandom().nextLong();

int[] sensor2 = (int[]) values[1];
sensor2[row] = new SecureRandom().nextInt();
tablet.addValue(row, 0, new SecureRandom().nextLong());

// mark this point as null value
if (time % 5 == 0) {
bitMaps[1].mark(row);
if (time % 5 != 0) {
tablet.addValue(row, 1, new SecureRandom().nextInt());
}

if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
session.insertAlignedTablet(tablet, true);
tablet.reset();
bitMaps[1].reset();
}
}

Expand Down Expand Up @@ -595,44 +577,5 @@ private static void insertTabletsWithAlignedTimeseries()
tablet2.reset();
tablet3.reset();
}

// Method 2 to add tablet data
long[] timestamps1 = tablet1.timestamps;
Object[] values1 = tablet1.values;
long[] timestamps2 = tablet2.timestamps;
Object[] values2 = tablet2.values;
long[] timestamps3 = tablet3.timestamps;
Object[] values3 = tablet3.values;

for (long time = 0; time < 100; time++) {
int row1 = tablet1.getRowSize();
int row2 = tablet2.getRowSize();
int row3 = tablet3.getRowSize();
tablet1.addTimestamp(row1, time);
tablet2.addTimestamp(row2, time);
tablet3.addTimestamp(row3, time);
for (int i = 0; i < 2; i++) {
long[] sensor1 = (long[]) values1[i];
sensor1[row1] = i;
long[] sensor2 = (long[]) values2[i];
sensor2[row2] = i;
long[] sensor3 = (long[]) values3[i];
sensor3[row3] = i;
}
if (tablet1.getRowSize() == tablet1.getMaxRowNumber()) {
session.insertAlignedTablets(tabletMap, true);

tablet1.reset();
tablet2.reset();
tablet3.reset();
}
}

if (tablet1.getRowSize() != 0) {
session.insertAlignedTablets(tabletMap, true);
tablet1.reset();
tablet2.reset();
tablet3.reset();
}
}
}
23 changes: 0 additions & 23 deletions example/session/src/main/java/org/apache/iotdb/SessionConcurrentExample.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -151,7 +151,6 @@ private static void insertTablet(Session session, String deviceId)

Tablet tablet = new Tablet(deviceId, schemaList, 100);

// Method 1 to add tablet data
long timestamp = System.currentTimeMillis();
for (long row = 0; row < 100; row++) {
int rowIndex = tablet.getRowSize();
Expand All @@ -171,27 +170,5 @@ private static void insertTablet(Session session, String deviceId)
session.insertTablet(tablet);
tablet.reset();
}

// Method 2 to add tablet data
long[] timestamps = tablet.timestamps;
Object[] values = tablet.values;

for (long time = 0; time < 100; time++) {
int row = tablet.getRowSize();
tablet.addTimestamp(row, time);
for (int i = 0; i < 3; i++) {
long[] sensor = (long[]) values[i];
sensor[row] = i;
}
if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
session.insertTablet(tablet, true);
tablet.reset();
}
}

if (tablet.getRowSize() != 0) {
session.insertTablet(tablet);
tablet.reset();
}
}
}
107 changes: 2 additions & 105 deletions example/session/src/main/java/org/apache/iotdb/SessionExample.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
Expand Down Expand Up @@ -401,7 +400,6 @@ private static void insertTablet() throws IoTDBConnectionException, StatementExe

Tablet tablet = new Tablet(ROOT_SG1_D1, schemaList, 100);

// Method 1 to add tablet data
long timestamp = System.currentTimeMillis();

for (long row = 0; row < 100; row++) {
Expand All @@ -422,28 +420,6 @@ private static void insertTablet() throws IoTDBConnectionException, StatementExe
session.insertTablet(tablet);
tablet.reset();
}

// Method 2 to add tablet data
long[] timestamps = tablet.timestamps;
Object[] values = tablet.values;

for (long time = 0; time < 100; time++) {
int row = tablet.getRowSize();
tablet.addTimestamp(row, time);
for (int i = 0; i < 3; i++) {
long[] sensor = (long[]) values[i];
sensor[row] = i;
}
if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
session.insertTablet(tablet, true);
tablet.reset();
}
}

if (tablet.getRowSize() != 0) {
session.insertTablet(tablet);
tablet.reset();
}
}

private static void insertTabletWithNullValues()
Expand All @@ -465,11 +441,7 @@ private static void insertTabletWithNullValues()

Tablet tablet = new Tablet(ROOT_SG1_D1, schemaList, 100);

// Method 1 to add tablet data
insertTablet1(schemaList, tablet);

// Method 2 to add tablet data
insertTablet2(schemaList, tablet);
}

private static void insertTablet1(List<IMeasurementSchema> schemaList, Tablet tablet)
Expand All @@ -482,11 +454,9 @@ private static void insertTablet1(List<IMeasurementSchema> schemaList, Tablet ta
tablet.addTimestamp(rowIndex, timestamp);
for (int s = 0; s < 3; s++) {
long value = random.nextLong();
// mark null value
if (row % 3 == s) {
tablet.bitMaps[s].mark((int) row);
if (row % 3 != s) {
tablet.addValue(schemaList.get(s).getMeasurementName(), rowIndex, value);
}
tablet.addValue(schemaList.get(s).getMeasurementName(), rowIndex, value);
}
if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
session.insertTablet(tablet, true);
Expand All @@ -501,39 +471,6 @@ private static void insertTablet1(List<IMeasurementSchema> schemaList, Tablet ta
}
}

private static void insertTablet2(List<IMeasurementSchema> schemaList, Tablet tablet)
throws IoTDBConnectionException, StatementExecutionException {
long[] timestamps = tablet.timestamps;
Object[] values = tablet.values;
BitMap[] bitMaps = new BitMap[schemaList.size()];
for (int s = 0; s < 3; s++) {
bitMaps[s] = new BitMap(tablet.getMaxRowNumber());
}
tablet.bitMaps = bitMaps;

for (long time = 0; time < 100; time++) {
int row = tablet.getRowSize();
tablet.addTimestamp(row, time);
for (int i = 0; i < 3; i++) {
long[] sensor = (long[]) values[i];
// mark null value
if (row % 3 == i) {
bitMaps[i].mark(row);
}
sensor[row] = i;
}
if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
session.insertTablet(tablet, true);
tablet.reset();
}
}

if (tablet.getRowSize() != 0) {
session.insertTablet(tablet);
tablet.reset();
}
}

private static void insertTablets() throws IoTDBConnectionException, StatementExecutionException {
// The schema of measurements of one device
// only measurementId and data type in MeasurementSchema take effects in Tablet
Expand All @@ -551,7 +488,6 @@ private static void insertTablets() throws IoTDBConnectionException, StatementEx
tabletMap.put("root.sg1.d2", tablet2);
tabletMap.put("root.sg1.d3", tablet3);

// Method 1 to add tablet data
long timestamp = System.currentTimeMillis();
for (long row = 0; row < 100; row++) {
int row1 = tablet1.getRowSize();
Expand Down Expand Up @@ -581,45 +517,6 @@ private static void insertTablets() throws IoTDBConnectionException, StatementEx
tablet2.reset();
tablet3.reset();
}

// Method 2 to add tablet data
long[] timestamps1 = tablet1.timestamps;
Object[] values1 = tablet1.values;
long[] timestamps2 = tablet2.timestamps;
Object[] values2 = tablet2.values;
long[] timestamps3 = tablet3.timestamps;
Object[] values3 = tablet3.values;

for (long time = 0; time < 100; time++) {
int row1 = tablet1.getRowSize();
int row2 = tablet2.getRowSize();
int row3 = tablet3.getRowSize();
timestamps1[row1] = time;
timestamps2[row2] = time;
timestamps3[row3] = time;
for (int i = 0; i < 3; i++) {
long[] sensor1 = (long[]) values1[i];
sensor1[row1] = i;
long[] sensor2 = (long[]) values2[i];
sensor2[row2] = i;
long[] sensor3 = (long[]) values3[i];
sensor3[row3] = i;
}
if (tablet1.getRowSize() == tablet1.getMaxRowNumber()) {
session.insertTablets(tabletMap, true);

tablet1.reset();
tablet2.reset();
tablet3.reset();
}
}

if (tablet1.getRowSize() != 0) {
session.insertTablets(tabletMap, true);
tablet1.reset();
tablet2.reset();
tablet3.reset();
}
}

/**
Expand Down
12 changes: 6 additions & 6 deletions example/trigger/src/main/java/org/apache/iotdb/trigger/LoggerTrigger.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,17 +39,17 @@ public boolean fire(Tablet tablet) throws Exception {
List<IMeasurementSchema> measurementSchemaList = tablet.getSchemas();
for (int i = 0, n = measurementSchemaList.size(); i < n; i++) {
if (measurementSchemaList.get(i).getType().equals(TSDataType.DOUBLE)) {
logDouble((double[]) tablet.values[i]);
logDouble((double[]) tablet.getValues()[i]);
} else if (measurementSchemaList.get(i).getType().equals(TSDataType.FLOAT)) {
logFloat((float[]) tablet.values[i]);
logFloat((float[]) tablet.getValues()[i]);
} else if (measurementSchemaList.get(i).getType().equals(TSDataType.INT64)) {
logLong((long[]) tablet.values[i]);
logLong((long[]) tablet.getValues()[i]);
} else if (measurementSchemaList.get(i).getType().equals(TSDataType.INT32)) {
logInt((int[]) tablet.values[i]);
logInt((int[]) tablet.getValues()[i]);
} else if (measurementSchemaList.get(i).getType().equals(TSDataType.TEXT)) {
logText((Binary[]) tablet.values[i]);
logText((Binary[]) tablet.getValues()[i]);
} else if (measurementSchemaList.get(i).getType().equals(TSDataType.BOOLEAN)) {
logBoolean((boolean[]) tablet.values[i]);
logBoolean((boolean[]) tablet.getValues()[i]);
}
}
return true;
Expand Down
4 changes: 2 additions & 2 deletions example/trigger/src/main/java/org/apache/iotdb/trigger/StatisticsUpdaterTrigger.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -81,12 +81,12 @@ public void onCreate(TriggerAttributes attributes) throws Exception {
@Override
public boolean fire(Tablet tablet) throws Exception {
ensureSession();
if (tablet.bitMaps == null) {
if (tablet.getBitMaps() == null) {
cnt.addAndGet((long) tablet.getRowSize() * tablet.getSchemas().size());
return true;
}
for (int column = 0; column < tablet.getSchemas().size(); column++) {
BitMap bitMap = tablet.bitMaps[column];
BitMap bitMap = tablet.getBitMaps()[column];
if (bitMap == null) {
cnt.addAndGet(tablet.getRowSize());
} else {
Expand Down
Loading

0 comments on commit 3f20241

Please sign in to comment.