Skip to content

Commit 85fdf8c

Browse files
jt2594838jt2594838
committed
add cache last values for load
1 parent 8f6f578 commit 85fdf8c

8 files changed

Lines changed: 171 additions & 83 deletions

File tree

‎integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,4 +113,10 @@ public DataNodeConfig setLoadLastCacheStrategy(String strategyName) {
113113
setProperty("last_cache_operation_on_load", strategyName);
114114
return this;
115115
}
116+
117+
@Override
118+
public DataNodeConfig setCacheLastValuesForLoad(boolean cacheLastValuesForLoad) {
119+
setProperty("cache_last_values_for_load", String.valueOf(cacheLastValuesForLoad));
120+
return this;
121+
}
116122
}

‎integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,4 +73,9 @@ public DataNodeConfig setMqttPayloadFormatter(String mqttPayloadFormatter) {
7373
public DataNodeConfig setLoadLastCacheStrategy(String strategyName) {
7474
return this;
7575
}
76+
77+
@Override
78+
public DataNodeConfig setCacheLastValuesForLoad(boolean cacheLastValuesForLoad) {
79+
return this;
80+
}
7681
}

‎integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,4 +43,6 @@ DataNodeConfig setLoadTsFileAnalyzeSchemaMemorySizeInBytes(
4343
DataNodeConfig setMqttPayloadFormatter(String mqttPayloadFormatter);
4444

4545
DataNodeConfig setLoadLastCacheStrategy(String strategyName);
46+
47+
DataNodeConfig setCacheLastValuesForLoad(boolean cacheLastValuesForLoad);
4648
}

‎integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadLastCacheIT.java‎

Lines changed: 118 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -106,7 +106,8 @@ public void setUp() throws Exception {
106106
EnvFactory.getEnv()
107107
.getConfig()
108108
.getDataNodeConfig()
109-
.setLoadLastCacheStrategy(lastCacheLoadStrategy.name());
109+
.setLoadLastCacheStrategy(lastCacheLoadStrategy.name())
110+
.setCacheLastValuesForLoad(true);
110111
EnvFactory.getEnv().initClusterEnvironment();
111112
}
112113

@@ -360,11 +361,12 @@ private static class PerformanceSchemas {
360361
private final List<String> columnNames;
361362
private final List<TSDataType> dataTypes;
362363

363-
public PerformanceSchemas(String database, String tableName, int measurementNum) {
364+
public PerformanceSchemas(
365+
String database, String tableName, int measurementNum, int blobMeasurementNum) {
364366
this.database = database;
365-
List<ColumnSchema> columnSchemas = new ArrayList<>(measurementNum);
366-
columnNames = new ArrayList<>(measurementNum);
367-
dataTypes = new ArrayList<>(measurementNum);
367+
List<ColumnSchema> columnSchemas = new ArrayList<>(measurementNum + blobMeasurementNum);
368+
columnNames = new ArrayList<>(measurementNum + blobMeasurementNum);
369+
dataTypes = new ArrayList<>(measurementNum + blobMeasurementNum);
368370

369371
columnSchemas.add(new ColumnSchema("device_id", TSDataType.STRING, ColumnCategory.TAG));
370372
columnNames.add("device_id");
@@ -374,17 +376,25 @@ public PerformanceSchemas(String database, String tableName, int measurementNum)
374376
columnNames.add("s" + i);
375377
dataTypes.add(TSDataType.INT64);
376378
}
379+
for (int i = 0; i < blobMeasurementNum; i++) {
380+
columnSchemas.add(
381+
new ColumnSchema("s" + (measurementNum + i), TSDataType.BLOB, ColumnCategory.FIELD));
382+
columnNames.add("s" + (measurementNum + i));
383+
dataTypes.add(TSDataType.BLOB);
384+
}
377385
tableSchema = new TableSchema(tableName, columnSchemas);
378386
}
379387
}
380388

381389
private void generateAndLoadOne(
382390
int deviceCnt,
383391
int measurementCnt,
392+
int blobMeasurementCnt,
384393
int pointCnt,
385394
int offset,
386395
PerformanceSchemas schemas,
387-
int fileNum)
396+
int fileNum,
397+
Statement statement)
388398
throws Exception {
389399
File file = new File("target" + File.separator + fileNum + ".tsfile");
390400
try (ITsFileWriter tsFileWriter =
@@ -398,50 +408,64 @@ private void generateAndLoadOne(
398408
for (int k = 0; k < measurementCnt; k++) {
399409
tablet.addValue(rowIndex, k + 1, (long) j + offset);
400410
}
411+
for (int k = 0; k < blobMeasurementCnt; k++) {
412+
tablet.addValue(rowIndex, k + 1 + measurementCnt, String.valueOf(j + offset));
413+
}
401414
rowIndex++;
402415
}
403416
}
404417
tsFileWriter.write(tablet);
405418
}
406419

407-
try (final Connection connection = EnvFactory.getEnv().getTableConnection();
408-
final Statement statement = connection.createStatement()) {
409-
statement.execute("USE " + schemas.database);
410-
statement.execute(
411-
String.format(
412-
"load '%s' with ('database-name'='%s')", file.getAbsolutePath(), schemas.database));
413-
}
420+
statement.execute(
421+
String.format(
422+
"load '%s' with ('database-name'='%s')", file.getAbsolutePath(), schemas.database));
423+
414424
file.delete();
415425
}
416426

417427
private void generateAndLoadAll(
418-
int deviceCnt, int measurementCnt, int pointCnt, PerformanceSchemas schemas, int fileNum)
428+
int deviceCnt,
429+
int measurementCnt,
430+
int blobMeasurementCnt,
431+
int pointCnt,
432+
PerformanceSchemas schemas,
433+
int fileNum)
419434
throws Exception {
420-
for (int i = 0; i < fileNum; i++) {
421-
generateAndLoadOne(deviceCnt, measurementCnt, pointCnt, pointCnt * i, schemas, fileNum);
422-
}
423-
}
424-
425-
private long queryLastOnce(int deviceNum, int measurementNum, PerformanceSchemas schemas)
426-
throws SQLException {
427435
try (final Connection connection = EnvFactory.getEnv().getTableConnection();
428436
final Statement statement = connection.createStatement()) {
429437
statement.execute("USE " + schemas.database);
430438

431-
try (final ResultSet resultSet =
432-
statement.executeQuery(
433-
String.format(
434-
"select last(%s) from %s where device_id='%s'",
435-
"s" + measurementNum, schemas.tableSchema.getTableName(), "d" + deviceNum))) {
436-
if (resultSet.next()) {
437-
return resultSet.getLong("_col0");
438-
} else {
439-
return -1;
440-
}
441-
} catch (SQLException e) {
442-
if (!e.getMessage().contains("does not exist")) {
443-
throw e;
444-
}
439+
for (int i = 0; i < fileNum; i++) {
440+
generateAndLoadOne(
441+
deviceCnt,
442+
measurementCnt,
443+
blobMeasurementCnt,
444+
pointCnt,
445+
pointCnt * i,
446+
schemas,
447+
fileNum,
448+
statement);
449+
}
450+
}
451+
}
452+
453+
private long queryLastOnce(
454+
int deviceNum, int measurementNum, PerformanceSchemas schemas, Statement statement)
455+
throws SQLException {
456+
try (final ResultSet resultSet =
457+
statement.executeQuery(
458+
String.format(
459+
"select last(time),last_by(%s, time) from %s where device_id='%s'",
460+
"s" + measurementNum, schemas.tableSchema.getTableName(), "d" + deviceNum))) {
461+
if (resultSet.next()) {
462+
return resultSet.getLong("_col0");
463+
} else {
464+
return -1;
465+
}
466+
} catch (SQLException e) {
467+
if (!e.getMessage().contains("does not exist")) {
468+
throw e;
445469
}
446470
}
447471
return -1;
@@ -460,43 +484,53 @@ private void queryAll(
460484
long totalStart = System.currentTimeMillis();
461485
List<Long> timeConsumptions = new ArrayList<>();
462486

463-
while (true) {
464-
int deviceNum = random.nextInt(deviceCnt);
465-
int measurementNum = random.nextInt(measurementCnt);
466-
rateLimiter.acquire();
467-
long start = System.currentTimeMillis();
468-
long result = queryLastOnce(deviceNum, measurementNum, schemas);
469-
long timeConsumption = System.currentTimeMillis() - start;
470-
if (result == -1) {
471-
try {
472-
Thread.sleep(1000);
473-
continue;
474-
} catch (InterruptedException e) {
475-
Thread.currentThread().interrupt();
487+
try (final Connection connection = EnvFactory.getEnv().getTableConnection();
488+
final Statement statement = connection.createStatement()) {
489+
statement.execute("USE " + schemas.database);
490+
491+
while (true) {
492+
int deviceNum = random.nextInt(deviceCnt);
493+
int measurementNum = random.nextInt(measurementCnt);
494+
rateLimiter.acquire();
495+
long start = System.currentTimeMillis();
496+
long result = queryLastOnce(deviceNum, measurementNum, schemas, statement);
497+
long timeConsumption = System.currentTimeMillis() - start;
498+
if (result == -1) {
499+
try {
500+
Thread.sleep(1000);
501+
continue;
502+
} catch (InterruptedException e) {
503+
Thread.currentThread().interrupt();
504+
}
505+
}
506+
System.out.printf(
507+
"%s: d%d.s%d %s %s%n", new Date(), deviceNum, measurementNum, result, timeConsumption);
508+
timeConsumptions.add(timeConsumption);
509+
if (result == (long) pointCnt * fileCnt - 1) {
510+
break;
476511
}
477-
}
478-
System.out.printf("%s: %s %s%n", new Date(), result, timeConsumption);
479-
timeConsumptions.add(timeConsumption);
480-
if (result == (long) pointCnt * fileCnt - 1) {
481-
break;
482512
}
483513
}
514+
484515
System.out.printf(
485-
"Synchronization ends after %dms%n, query latency avg %f",
516+
"Synchronization ends after %dms, query latency avg %fms %n",
486517
System.currentTimeMillis() - totalStart,
487518
timeConsumptions.stream().mapToLong(i -> i).average().orElse(0.0));
488519
}
489520

490521
// @Ignore("Performance")
491522
@Test
492523
public void testTableLoadPerformance() throws Exception {
493-
int deviceCnt = 1000;
524+
int deviceCnt = 100;
494525
int measurementCnt = 100;
526+
int blobMeasurementCnt = 10;
495527
int pointCnt = 100;
496-
int fileCnt = 100;
497-
int queryPerSec = 10;
528+
int fileCnt = 1000;
529+
int queryPerSec = 1000;
530+
int queryThreadsNum = 10;
498531

499-
PerformanceSchemas schemas = new PerformanceSchemas("test", "test_table", measurementCnt);
532+
PerformanceSchemas schemas =
533+
new PerformanceSchemas("test", "test_table", measurementCnt, blobMeasurementCnt);
500534

501535
try (final Connection connection = EnvFactory.getEnv().getTableConnection();
502536
final Statement statement = connection.createStatement()) {
@@ -507,28 +541,41 @@ public void testTableLoadPerformance() throws Exception {
507541
new Thread(
508542
() -> {
509543
try {
510-
generateAndLoadAll(deviceCnt, measurementCnt, pointCnt, schemas, fileCnt);
511-
} catch (Exception e) {
544+
generateAndLoadAll(
545+
deviceCnt, measurementCnt, blobMeasurementCnt, pointCnt, schemas, fileCnt);
546+
} catch (Throwable e) {
512547
e.printStackTrace();
513548
}
514549
});
515550

516551
RateLimiter rateLimiter = RateLimiter.create(queryPerSec);
517-
Thread queryThread =
518-
new Thread(
519-
() -> {
520-
try {
521-
queryAll(deviceCnt, measurementCnt, pointCnt, fileCnt, schemas, rateLimiter);
522-
} catch (SQLException e) {
523-
e.printStackTrace();
524-
}
525-
});
552+
List<Thread> queryThreads = new ArrayList<>(queryThreadsNum);
553+
for (int i = 0; i < queryThreadsNum; i++) {
554+
Thread queryThread =
555+
new Thread(
556+
() -> {
557+
try {
558+
queryAll(
559+
deviceCnt,
560+
measurementCnt + blobMeasurementCnt,
561+
pointCnt,
562+
fileCnt,
563+
schemas,
564+
rateLimiter);
565+
} catch (Throwable e) {
566+
e.printStackTrace();
567+
}
568+
});
569+
queryThreads.add(queryThread);
570+
}
526571

527572
loadThread.start();
528-
queryThread.start();
573+
queryThreads.forEach(Thread::start);
529574

530575
loadThread.join();
531-
queryThread.join();
576+
for (Thread queryThread : queryThreads) {
577+
queryThread.join();
578+
}
532579
}
533580

534581
private static class SchemaConfig {

‎iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -237,7 +237,7 @@ public TSStatus write(ConsensusGroupId groupId, IConsensusRequest request)
237237
} else if (!impl.isActive()) {
238238
String message =
239239
String.format(
240-
"Peer is inactive and not ready to write request, %s, DataNode Id: %s",
240+
"Peer is inactive and not ready to write request, %s, DatsaNode Id: %s",
241241
groupId.toString(), impl.getThisNode().getNodeId());
242242
return RpcUtils.getStatus(TSStatusCode.WRITE_PROCESS_REJECT, message);
243243
} else {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1151,6 +1151,11 @@ public class IoTDBConfig {
11511151

11521152
private LastCacheLoadStrategy lastCacheLoadStrategy = LastCacheLoadStrategy.UPDATE;
11531153

1154+
/**
1155+
* Whether to cache last values when constructing TsFileResource during LOAD. When set to true,
1156+
* blob series will be forcibly ignored even if lastCacheLoadStrategy =
1157+
* LastCacheLoadStrategy.UPDATE.
1158+
*/
11541159
private boolean cacheLastValuesForLoad = true;
11551160

11561161
IoTDBConfig() {}

0 commit comments

Comments
 (0)