This is an automated email from the ASF dual-hosted git repository.
qiaojialin 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 1ff6db9 [IOTDB-2524] Aligned Timeseries support tags and attributes
(#5191)
1ff6db9 is described below
commit 1ff6db96bf688aff6024ef676d5de2bcd03436c0
Author: 任宇华 <[email protected]>
AuthorDate: Wed Mar 16 14:09:25 2022 +0800
[IOTDB-2524] Aligned Timeseries support tags and attributes (#5191)
---
.../apache/iotdb/cluster/metadata/CMManager.java | 9 +-
docs/UserGuide/Operate-Metadata/Timeseries.md | 39 ++++++-
docs/zh/UserGuide/Operate-Metadata/Timeseries.md | 34 +++++-
.../iotdb/AlignedTimeseriesSessionExample.java | 9 +-
.../apache/iotdb/session/IoTDBSessionSimpleIT.java | 5 +-
.../org/apache/iotdb/db/metadata/MManager.java | 41 +++++--
.../org/apache/iotdb/db/metadata/mtree/MTree.java | 5 +-
.../sys/CreateAlignedTimeSeriesOperator.java | 55 ++++++++-
.../physical/sys/CreateAlignedTimeSeriesPlan.java | 125 +++++++++++++++++++--
.../apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java | 8 +-
.../db/service/thrift/impl/TSServiceImpl.java | 4 +-
.../iotdb/db/metadata/idtable/IDTableTest.java | 4 +
.../db/qp/physical/PhysicalPlanSerializeTest.java | 2 +
.../java/org/apache/iotdb/session/Session.java | 19 +++-
.../session/IoTDBSessionDisableMemControlIT.java | 9 +-
.../session/IoTDBSessionVectorABDeviceIT.java | 2 +-
.../IoTDBSessionVectorAggregationWithUnSeqIT.java | 2 +-
.../iotdb/session/IoTDBSessionVectorInsertIT.java | 9 +-
thrift/src/main/thrift/rpc.thrift | 2 +
19 files changed, 350 insertions(+), 33 deletions(-)
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java
b/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java
index 3402a3e..28e6819 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/metadata/CMManager.java
@@ -742,7 +742,14 @@ public class CMManager extends MManager {
CreateAlignedTimeSeriesPlan plan =
new CreateAlignedTimeSeriesPlan(
- insertPlan.getDevicePath(), measurements, dataTypes, encodings,
compressors, null);
+ insertPlan.getDevicePath(),
+ measurements,
+ dataTypes,
+ encodings,
+ compressors,
+ null,
+ null,
+ null);
TSStatus result;
try {
result = coordinator.processPartitionedPlan(plan);
diff --git a/docs/UserGuide/Operate-Metadata/Timeseries.md
b/docs/UserGuide/Operate-Metadata/Timeseries.md
index 7761abb..965e457 100644
--- a/docs/UserGuide/Operate-Metadata/Timeseries.md
+++ b/docs/UserGuide/Operate-Metadata/Timeseries.md
@@ -84,14 +84,15 @@ IoTDB> delete timeseries root.ln.wf02.*
There are four optional clauses added in SHOW TIMESERIES, return information
of time series
+
Timeseries information includes: timeseries path, alias of measurement,
storage group it belongs to, data type, encoding type, compression type, tags
and attributes.
-
+
Examples:
* SHOW TIMESERIES
presents all timeseries information in JSON form
-
+
* SHOW TIMESERIES <`PathPattern`>
returns all timeseries information matching the given <`PathPattern`>. SQL
statements are as follows:
@@ -142,6 +143,7 @@ show timeseries root.ln.** limit 10 offset 10
all the returned timeseries information should be sorted in descending order
of the last timestamp of timeseries
+
It is worth noting that when the queried path does not exist, the system will
return no timeseries.
@@ -179,7 +181,6 @@ It costs 0.004s
Then the Metadata Tree will be as below:
<center><img style="width:100%; max-width:600px; margin-left:auto;
margin-right:auto; display:block;"
src="https://user-images.githubusercontent.com/19167280/69792176-1718f400-1201-11ea-861a-1a83c07ca144.jpg";></center>
-
As can be seen, `root` is considered as `LEVEL=0`. So when you enter
statements such as:
```
@@ -304,3 +305,35 @@ It costs 0.004s
```
> Notice that, we only support one condition in the where clause. Either it's
> an equal filter or it is an `contains` filter. In both case, the property in
> the where condition must be a tag.
+
+create aligned timeseries
+
+```
+create aligned timeseries root.sg1.d1(s1 INT32 tags(tag1=v1, tag2=v2)
attributes(attr1=v1, attr2=v2), s2 DOUBLE tags(tag3=v3, tag4=v4)
attributes(attr3=v3, attr4=v4))
+```
+
+The execution result is as follows:
+
+```
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+| timeseries|alias|storage group|dataType|encoding|compression|
tags| attributes|
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+|root.sg1.d1.s1| null| root.sg1| INT32| RLE|
SNAPPY|{"tag1":"v1","tag2":"v2"}|{"attr2":"v2","attr1":"v1"}|
+|root.sg1.d1.s2| null| root.sg1| DOUBLE| GORILLA|
SNAPPY|{"tag4":"v4","tag3":"v3"}|{"attr4":"v4","attr3":"v3"}|
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+```
+
+Support query：
+
+```
+IoTDB> show storage group where tag1='v1'
+Msg: 401: Error occurred while parsing SQL to physical plan: line 1:19
mismatched input 'where' expecting {<EOF>, ';'}
+IoTDB> show timeseries where tag1='v1'
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+| timeseries|alias|storage group|dataType|encoding|compression|
tags| attributes|
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+|root.sg1.d1.s1| null| root.sg1| INT32| RLE|
SNAPPY|{"tag1":"v1","tag2":"v2"}|{"attr2":"v2","attr1":"v1"}|
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+```
+
+The above operations are supported for timeseries tag, attribute updates, etc.
\ No newline at end of file
diff --git a/docs/zh/UserGuide/Operate-Metadata/Timeseries.md
b/docs/zh/UserGuide/Operate-Metadata/Timeseries.md
index 8d2b9d9..e12c1e7 100644
--- a/docs/zh/UserGuide/Operate-Metadata/Timeseries.md
+++ b/docs/zh/UserGuide/Operate-Metadata/Timeseries.md
@@ -139,6 +139,7 @@ show timeseries root.ln.** limit 10 offset 10
表示查询出的时间序列需要按照最近插入时间戳降序排列
+
需要注意的是，当查询路径不存在时，系统会返回 0 条时间序列。
## 统计时间序列总数
@@ -242,7 +243,6 @@ create timeseries root.turbine.d1.s1(temprature) with
datatype=FLOAT, encoding=R
* 标签点属性更新
创建时间序列后，我们也可以对其原有的标签点属性进行更新，主要有以下六种更新方式：
-
* 重命名标签或属性
```
ALTER timeseries root.turbine.d1.s1 RENAME tag1 TO newTag1
@@ -304,3 +304,35 @@ It costs 0.004s
```
> 注意，现在我们只支持一个查询条件，要么是等值条件查询，要么是包含条件查询。当然 where 子句中涉及的必须是标签值，而不能是属性值。
+
+创建对齐时间序列
+
+```
+create aligned timeseries root.sg1.d1(s1 INT32 tags(tag1=v1, tag2=v2)
attributes(attr1=v1, attr2=v2), s2 DOUBLE tags(tag3=v3, tag4=v4)
attributes(attr3=v3, attr4=v4))
+```
+
+执行结果如下：
+
+```
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+| timeseries|alias|storage group|dataType|encoding|compression|
tags| attributes|
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+|root.sg1.d1.s1| null| root.sg1| INT32| RLE|
SNAPPY|{"tag1":"v1","tag2":"v2"}|{"attr2":"v2","attr1":"v1"}|
+|root.sg1.d1.s2| null| root.sg1| DOUBLE| GORILLA|
SNAPPY|{"tag4":"v4","tag3":"v3"}|{"attr4":"v4","attr3":"v3"}|
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+```
+
+支持查询：
+
+```
+IoTDB> show storage group where tag1='v1'
+Msg: 401: Error occurred while parsing SQL to physical plan: line 1:19
mismatched input 'where' expecting {<EOF>, ';'}
+IoTDB> show timeseries where tag1='v1'
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+| timeseries|alias|storage group|dataType|encoding|compression|
tags| attributes|
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+|root.sg1.d1.s1| null| root.sg1| INT32| RLE|
SNAPPY|{"tag1":"v1","tag2":"v2"}|{"attr2":"v2","attr1":"v1"}|
++--------------+-----+-------------+--------+--------+-----------+-------------------------+---------------------------+
+```
+
+上述对时间序列标签、属性的更新等操作都支持。
\ No newline at end of file
diff --git
a/example/session/src/main/java/org/apache/iotdb/AlignedTimeseriesSessionExample.java
b/example/session/src/main/java/org/apache/iotdb/AlignedTimeseriesSessionExample.java
index aa734bc..aa52ed2 100644
---
a/example/session/src/main/java/org/apache/iotdb/AlignedTimeseriesSessionExample.java
+++
b/example/session/src/main/java/org/apache/iotdb/AlignedTimeseriesSessionExample.java
@@ -271,7 +271,14 @@ public class AlignedTimeseriesSessionExample {
compressors.add(CompressionType.SNAPPY);
}
session.createAlignedTimeseries(
- ROOT_SG1_D1, multiMeasurementComponents, dataTypes, encodings,
compressors, null);
+ ROOT_SG1_D1,
+ multiMeasurementComponents,
+ dataTypes,
+ encodings,
+ compressors,
+ null,
+ null,
+ null);
}
// be sure template is coordinate with tablet
diff --git
a/integration/src/test/java/org/apache/iotdb/session/IoTDBSessionSimpleIT.java
b/integration/src/test/java/org/apache/iotdb/session/IoTDBSessionSimpleIT.java
index 7bc81c1..85c2b42 100644
---
a/integration/src/test/java/org/apache/iotdb/session/IoTDBSessionSimpleIT.java
+++
b/integration/src/test/java/org/apache/iotdb/session/IoTDBSessionSimpleIT.java
@@ -1063,7 +1063,10 @@ public class IoTDBSessionSimpleIT {
tsDataTypes,
tsEncodings,
compressionTypes,
- Arrays.asList("alias1", "alias2", "alias3"));
+ Arrays.asList("alias1", "alias2", "alias3"),
+ null,
+ null);
+
fail("Exception expected");
} catch (StatementExecutionException e) {
assertTrue(
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
index 0323d09..4130049 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
@@ -650,7 +650,7 @@ public class MManager {
throws MetadataException {
createAlignedTimeSeries(
new CreateAlignedTimeSeriesPlan(
- prefixPath, measurements, dataTypes, encodings, compressors,
null));
+ prefixPath, measurements, dataTypes, encodings, compressors, null,
null, null));
}
/**
@@ -669,6 +669,8 @@ public class MManager {
List<String> measurements = plan.getMeasurements();
List<TSDataType> dataTypes = plan.getDataTypes();
List<TSEncoding> encodings = plan.getEncodings();
+ List<Map<String, String>> tagsList = plan.getTagsList();
+ List<Map<String, String>> attributesList = plan.getAttributesList();
for (int i = 0; i < measurements.size(); i++) {
SchemaUtils.checkDataTypeWithEncoding(dataTypes.get(i),
encodings.get(i));
@@ -677,16 +679,26 @@ public class MManager {
ensureStorageGroup(prefixPath);
// create time series in MTree
- mtree.createAlignedTimeseries(
- prefixPath,
- measurements,
- plan.getDataTypes(),
- plan.getEncodings(),
- plan.getCompressors());
+ List<IMeasurementMNode> measurementMNodeList =
+ mtree.createAlignedTimeseries(
+ prefixPath,
+ measurements,
+ plan.getDataTypes(),
+ plan.getEncodings(),
+ plan.getCompressors());
// the cached mNode may be replaced by new entityMNode in mtree
mNodeCache.invalidate(prefixPath);
+ for (int i = 0; i < measurements.size(); i++) {
+ if (!plan.getTagOffsets().isEmpty() && isRecovering) {
+ tagManager.recoverIndex(plan.getTagOffsets().get(i),
measurementMNodeList.get(i));
+ } else if (plan.getTagsList() != null &&
!plan.getTagsList().isEmpty()) {
+ // tag key, tag value
+ tagManager.addIndex(plan.getTagsList().get(i),
measurementMNodeList.get(i));
+ }
+ }
+
// update statistics and schemaDataTypeNumMap
totalSeriesNumber.addAndGet(measurements.size());
if (totalSeriesNumber.get() * ESTIMATED_SERIES_SIZE >=
MTREE_SIZE_THRESHOLD) {
@@ -694,9 +706,24 @@ public class MManager {
allowToCreateNewSeries = false;
}
// write log
+ List<Long> tagOffsets = new ArrayList<>();
if (!isRecovering) {
+ if ((plan.getTagsList() != null && !plan.getTagsList().isEmpty())
+ || (plan.getAttributesList() != null &&
!plan.getAttributesList().isEmpty())) {
+ for (int i = 0; i < measurements.size(); i++) {
+ tagOffsets.add(tagManager.writeTagFile(tagsList.get(i),
attributesList.get(i)));
+ }
+ } else {
+ for (int i = 0; i < measurements.size(); i++) {
+ tagOffsets.add(Long.parseLong("-1"));
+ }
+ }
+ plan.setTagOffsets(tagOffsets);
logWriter.createAlignedTimeseries(plan);
}
+ for (int i = 0; i < measurements.size(); i++) {
+ measurementMNodeList.get(i).setOffset(plan.getTagOffsets().get(i));
+ }
} catch (IOException e) {
throw new MetadataException(e);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTree.java
b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTree.java
index 0d1b571..ff99fd4 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTree.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/mtree/MTree.java
@@ -436,13 +436,14 @@ public class MTree implements Serializable {
* @param encodings encodings list
* @param compressors compressor
*/
- public void createAlignedTimeseries(
+ public List<IMeasurementMNode> createAlignedTimeseries(
PartialPath devicePath,
List<String> measurements,
List<TSDataType> dataTypes,
List<TSEncoding> encodings,
List<CompressionType> compressors)
throws MetadataException {
+ List<IMeasurementMNode> measurementMNodeList = new ArrayList<>();
MetaFormatUtils.checkSchemaMeasurementNames(measurements);
Pair<IMNode, Template> pair = checkAndAutoCreateInternalPath(devicePath);
IMNode cur = pair.left;
@@ -484,8 +485,10 @@ public class MTree implements Serializable {
measurements.get(i), dataTypes.get(i), encodings.get(i),
compressors.get(i)),
null);
entityMNode.addChild(measurements.get(i), measurementMNode);
+ measurementMNodeList.add(measurementMNode);
}
}
+ return measurementMNodeList;
}
private Pair<IMNode, Template> checkAndAutoCreateInternalPath(PartialPath
devicePath)
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/CreateAlignedTimeSeriesOperator.java
b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/CreateAlignedTimeSeriesOperator.java
index bc0b077..65c14c2 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/CreateAlignedTimeSeriesOperator.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/logical/sys/CreateAlignedTimeSeriesOperator.java
@@ -32,6 +32,7 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
public class CreateAlignedTimeSeriesOperator extends Operator {
@@ -42,6 +43,9 @@ public class CreateAlignedTimeSeriesOperator extends Operator
{
private List<TSEncoding> encodings = new ArrayList<>();
private List<CompressionType> compressors = new ArrayList<>();
private List<String> aliasList = null;
+ private List<Map<String, String>> tagsList = new ArrayList<>();
+ private List<Map<String, String>> attributesList = new ArrayList<>();
+ private List<Long> tagOffsets = null;
public CreateAlignedTimeSeriesOperator(int tokenIntType) {
super(tokenIntType);
@@ -116,6 +120,48 @@ public class CreateAlignedTimeSeriesOperator extends
Operator {
this.aliasList.add(alias);
}
+ public List<Map<String, String>> getTagsList() {
+ return tagsList;
+ }
+
+ public void setTagsList(List<Map<String, String>> tagsList) {
+ this.tagsList = tagsList;
+ }
+
+ public void addTagsList(Map<String, String> tags) {
+ this.tagsList.add(tags);
+ }
+
+ public List<Map<String, String>> getAttributesList() {
+ return attributesList;
+ }
+
+ public void setAttributesList(List<Map<String, String>> attributesList) {
+ this.attributesList = attributesList;
+ }
+
+ public void addAttributesList(Map<String, String> attributes) {
+ this.attributesList.add(attributes);
+ }
+
+ public List<Long> getTagOffsets() {
+ if (tagOffsets == null) {
+ tagOffsets = new ArrayList<>();
+ for (int i = 0; i < measurements.size(); i++) {
+ tagOffsets.add(Long.parseLong("-1"));
+ }
+ }
+ return tagOffsets;
+ }
+
+ public void setTagOffsets(List<Long> tagOffsets) {
+ this.tagOffsets = tagOffsets;
+ }
+
+ public void addTagOffsets(Long tagsOffset) {
+ this.tagOffsets.add(tagsOffset);
+ }
+
@Override
public PhysicalPlan generatePhysicalPlan(PhysicalGenerator generator)
throws QueryProcessException {
@@ -126,6 +172,13 @@ public class CreateAlignedTimeSeriesOperator extends
Operator {
}
return new CreateAlignedTimeSeriesPlan(
- prefixPath, measurements, dataTypes, encodings, compressors,
aliasList);
+ prefixPath,
+ measurements,
+ dataTypes,
+ encodings,
+ compressors,
+ aliasList,
+ tagsList,
+ attributesList);
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateAlignedTimeSeriesPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateAlignedTimeSeriesPlan.java
index b02a98e..5e7636e 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateAlignedTimeSeriesPlan.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/physical/sys/CreateAlignedTimeSeriesPlan.java
@@ -36,6 +36,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
public class CreateAlignedTimeSeriesPlan extends PhysicalPlan {
@@ -48,6 +49,9 @@ public class CreateAlignedTimeSeriesPlan extends PhysicalPlan
{
private List<TSEncoding> encodings;
private List<CompressionType> compressors;
private List<String> aliasList;
+ private List<Map<String, String>> tagsList;
+ private List<Map<String, String>> attributesList;
+ private List<Long> tagOffsets = null;
public CreateAlignedTimeSeriesPlan() {
super(Operator.OperatorType.CREATE_ALIGNED_TIMESERIES);
@@ -60,7 +64,9 @@ public class CreateAlignedTimeSeriesPlan extends PhysicalPlan
{
List<TSDataType> dataTypes,
List<TSEncoding> encodings,
List<CompressionType> compressors,
- List<String> aliasList) {
+ List<String> aliasList,
+ List<Map<String, String>> tagsList,
+ List<Map<String, String>> attributesList) {
super(Operator.OperatorType.CREATE_ALIGNED_TIMESERIES);
this.prefixPath = prefixPath;
this.measurements = measurements;
@@ -68,6 +74,8 @@ public class CreateAlignedTimeSeriesPlan extends PhysicalPlan
{
this.encodings = encodings;
this.compressors = compressors;
this.aliasList = aliasList;
+ this.tagsList = tagsList;
+ this.attributesList = attributesList;
this.canBeSplit = false;
}
@@ -119,11 +127,41 @@ public class CreateAlignedTimeSeriesPlan extends
PhysicalPlan {
this.aliasList = aliasList;
}
+ public List<Map<String, String>> getTagsList() {
+ return tagsList;
+ }
+
+ public void setTagsList(List<Map<String, String>> tagsList) {
+ this.tagsList = tagsList;
+ }
+
+ public List<Map<String, String>> getAttributesList() {
+ return attributesList;
+ }
+
+ public void setAttributesList(List<Map<String, String>> attributesList) {
+ this.attributesList = attributesList;
+ }
+
+ public List<Long> getTagOffsets() {
+ if (tagOffsets == null) {
+ tagOffsets = new ArrayList<>();
+ for (int i = 0; i < measurements.size(); i++) {
+ tagOffsets.add(Long.parseLong("-1"));
+ }
+ }
+ return tagOffsets;
+ }
+
+ public void setTagOffsets(List<Long> tagOffsets) {
+ this.tagOffsets = tagOffsets;
+ }
+
@Override
public String toString() {
return String.format(
- "devicePath: %s, measurements: %s, dataTypes: %s, encodings: %s,
compressions: %s",
- prefixPath, measurements, dataTypes, encodings, compressors);
+ "devicePath: %s, measurements: %s, dataTypes: %s, encodings: %s,
compressions: %s, tagOffsets: %s",
+ prefixPath, measurements, dataTypes, encodings, compressors,
tagOffsets);
}
@Override
@@ -159,9 +197,18 @@ public class CreateAlignedTimeSeriesPlan extends
PhysicalPlan {
for (CompressionType compressor : compressors) {
stream.write(compressor.ordinal());
}
+ if (tagOffsets == null) {
+ tagOffsets = new ArrayList<>();
+ for (int i = 0; i < measurements.size(); i++) {
+ tagOffsets.add(Long.parseLong("-1"));
+ }
+ }
+ for (Long tagOffset : tagOffsets) {
+ stream.writeLong(tagOffset);
+ }
// alias
- if (aliasList != null) {
+ if (aliasList != null && !aliasList.isEmpty()) {
stream.write(1);
for (String alias : aliasList) {
ReadWriteIOUtils.write(alias, stream);
@@ -169,6 +216,27 @@ public class CreateAlignedTimeSeriesPlan extends
PhysicalPlan {
} else {
stream.write(0);
}
+
+ // tags
+ if (tagsList != null && !tagsList.isEmpty()) {
+ stream.write(1);
+ for (Map<String, String> tags : tagsList) {
+ ReadWriteIOUtils.write(tags, stream);
+ }
+ } else {
+ stream.write(0);
+ }
+
+ // attributes
+ if (attributesList != null && !attributesList.isEmpty()) {
+ stream.write(1);
+ for (Map<String, String> attributes : attributesList) {
+ ReadWriteIOUtils.write(attributes, stream);
+ }
+ } else {
+ stream.write(0);
+ }
+
stream.writeLong(index);
}
@@ -192,9 +260,12 @@ public class CreateAlignedTimeSeriesPlan extends
PhysicalPlan {
for (CompressionType compressor : compressors) {
buffer.put((byte) compressor.ordinal());
}
+ for (Long tagOffset : tagOffsets) {
+ buffer.putLong(tagOffset);
+ }
// alias
- if (aliasList != null) {
+ if (aliasList != null && !aliasList.isEmpty()) {
buffer.put((byte) 1);
for (String alias : aliasList) {
ReadWriteIOUtils.write(alias, buffer);
@@ -203,6 +274,26 @@ public class CreateAlignedTimeSeriesPlan extends
PhysicalPlan {
buffer.put((byte) 0);
}
+ // tags
+ if (tagsList != null && !tagsList.isEmpty()) {
+ buffer.put((byte) 1);
+ for (Map<String, String> tags : tagsList) {
+ ReadWriteIOUtils.write(tags, buffer);
+ }
+ } else {
+ buffer.put((byte) 0);
+ }
+
+ // attributes
+ if (attributesList != null && !attributesList.isEmpty()) {
+ buffer.put((byte) 1);
+ for (Map<String, String> attributes : attributesList) {
+ ReadWriteIOUtils.write(attributes, buffer);
+ }
+ } else {
+ buffer.put((byte) 0);
+ }
+
buffer.putLong(index);
}
@@ -230,6 +321,10 @@ public class CreateAlignedTimeSeriesPlan extends
PhysicalPlan {
for (int i = 0; i < size; i++) {
compressors.add(CompressionType.values()[buffer.get()]);
}
+ tagOffsets = new ArrayList<>();
+ for (int i = 0; i < size; i++) {
+ tagOffsets.add(buffer.getLong());
+ }
// alias
if (buffer.get() == 1) {
@@ -238,6 +333,21 @@ public class CreateAlignedTimeSeriesPlan extends
PhysicalPlan {
aliasList.add(ReadWriteIOUtils.readString(buffer));
}
}
+ // tags
+ if (buffer.get() == 1) {
+ tagsList = new ArrayList<>();
+ for (int i = 0; i < size; i++) {
+ tagsList.add(ReadWriteIOUtils.readMap(buffer));
+ }
+ }
+
+ // attributes
+ if (buffer.get() == 1) {
+ attributesList = new ArrayList<>();
+ for (int i = 0; i < size; i++) {
+ attributesList.add(ReadWriteIOUtils.readMap(buffer));
+ }
+ }
this.index = buffer.getLong();
}
@@ -256,11 +366,12 @@ public class CreateAlignedTimeSeriesPlan extends
PhysicalPlan {
&& Objects.equals(measurements, that.measurements)
&& Objects.equals(dataTypes, that.dataTypes)
&& Objects.equals(encodings, that.encodings)
- && Objects.equals(compressors, that.compressors);
+ && Objects.equals(compressors, that.compressors)
+ && Objects.equals(tagOffsets, that.tagOffsets);
}
@Override
public int hashCode() {
- return Objects.hash(prefixPath, measurements, dataTypes, encodings,
compressors);
+ return Objects.hash(prefixPath, measurements, dataTypes, encodings,
compressors, tagOffsets);
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
b/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
index 335d615..d369ca4 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/sql/IoTDBSqlVisitor.java
@@ -298,11 +298,11 @@ public class IoTDBSqlVisitor extends
IoTDBSqlParserBaseVisitor<Operator> {
}
if (ctx.tagClause() != null) {
- throw new SQLParserException("create aligned timeseries: tag is not
supported yet.");
+ parseTagClause(ctx.tagClause(), createAlignedTimeSeriesOperator);
}
if (ctx.attributeClause() != null) {
- throw new SQLParserException("create aligned timeseries: attribute is
not supported yet.");
+ parseAttributeClause(ctx.attributeClause(),
createAlignedTimeSeriesOperator);
}
}
@@ -2689,6 +2689,8 @@ public class IoTDBSqlVisitor extends
IoTDBSqlParserBaseVisitor<Operator> {
Map<String, String> tags = extractMap(ctx.propertyClause(),
ctx.propertyClause(0));
if (operator instanceof CreateTimeSeriesOperator) {
((CreateTimeSeriesOperator) operator).setTags(tags);
+ } else if (operator instanceof CreateAlignedTimeSeriesOperator) {
+ ((CreateAlignedTimeSeriesOperator) operator).addTagsList(tags);
} else if (operator instanceof AlterTimeSeriesOperator) {
((AlterTimeSeriesOperator) operator).setTagsMap(tags);
}
@@ -2698,6 +2700,8 @@ public class IoTDBSqlVisitor extends
IoTDBSqlParserBaseVisitor<Operator> {
Map<String, String> attributes = extractMap(ctx.propertyClause(),
ctx.propertyClause(0));
if (operator instanceof CreateTimeSeriesOperator) {
((CreateTimeSeriesOperator) operator).setAttributes(attributes);
+ } else if (operator instanceof CreateAlignedTimeSeriesOperator) {
+ ((CreateAlignedTimeSeriesOperator)
operator).addAttributesList(attributes);
} else if (operator instanceof AlterTimeSeriesOperator) {
((AlterTimeSeriesOperator) operator).setAttributesMap(attributes);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java
b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java
index 3f9dce6..3f4b37d 100644
---
a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java
+++
b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/TSServiceImpl.java
@@ -1762,7 +1762,9 @@ public class TSServiceImpl implements TSIService.Iface {
dataTypes,
encodings,
compressors,
- req.measurementAlias);
+ req.measurementAlias,
+ req.tagsList,
+ req.attributesList);
TSStatus status = serviceProvider.checkAuthority(plan,
req.getSessionId());
return status != null ? status : executeNonQueryPlan(plan);
} catch (IoTDBException e) {
diff --git
a/server/src/test/java/org/apache/iotdb/db/metadata/idtable/IDTableTest.java
b/server/src/test/java/org/apache/iotdb/db/metadata/idtable/IDTableTest.java
index 30b1428..2c17245 100644
--- a/server/src/test/java/org/apache/iotdb/db/metadata/idtable/IDTableTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/metadata/idtable/IDTableTest.java
@@ -109,6 +109,8 @@ public class IDTableTest {
Arrays.asList(
TSEncoding.valueOf("RLE"), TSEncoding.valueOf("RLE"),
TSEncoding.valueOf("RLE")),
Arrays.asList(compressionType, compressionType, compressionType),
+ null,
+ null,
null);
manager.createAlignedTimeSeries(plan);
@@ -188,6 +190,8 @@ public class IDTableTest {
Arrays.asList(
TSEncoding.valueOf("RLE"), TSEncoding.valueOf("RLE"),
TSEncoding.valueOf("RLE")),
Arrays.asList(compressionType, compressionType, compressionType),
+ null,
+ null,
null);
manager.createAlignedTimeSeries(plan);
diff --git
a/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanSerializeTest.java
b/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanSerializeTest.java
index 4244c1b..9cb9110 100644
---
a/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanSerializeTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/qp/physical/PhysicalPlanSerializeTest.java
@@ -199,6 +199,8 @@ public class PhysicalPlanSerializeTest {
Arrays.asList(TSDataType.DOUBLE, TSDataType.INT32),
Arrays.asList(TSEncoding.RLE, TSEncoding.RLE),
Arrays.asList(CompressionType.SNAPPY, CompressionType.SNAPPY),
+ null,
+ null,
null);
PhysicalPlan result =
testTwoSerializeMethodAndDeserialize(createAlignedTimeSeriesPlan);
diff --git a/session/src/main/java/org/apache/iotdb/session/Session.java
b/session/src/main/java/org/apache/iotdb/session/Session.java
index 9a31211..fc0fe71 100644
--- a/session/src/main/java/org/apache/iotdb/session/Session.java
+++ b/session/src/main/java/org/apache/iotdb/session/Session.java
@@ -506,11 +506,20 @@ public class Session {
List<TSDataType> dataTypes,
List<TSEncoding> encodings,
List<CompressionType> compressors,
- List<String> measurementAliasList)
+ List<String> measurementAliasList,
+ List<Map<String, String>> tagsList,
+ List<Map<String, String>> attributesList)
throws IoTDBConnectionException, StatementExecutionException {
TSCreateAlignedTimeseriesReq request =
getTSCreateAlignedTimeseriesReq(
- deviceId, measurements, dataTypes, encodings, compressors,
measurementAliasList);
+ deviceId,
+ measurements,
+ dataTypes,
+ encodings,
+ compressors,
+ measurementAliasList,
+ tagsList,
+ attributesList);
defaultSessionConnection.createAlignedTimeseries(request);
}
@@ -520,7 +529,9 @@ public class Session {
List<TSDataType> dataTypes,
List<TSEncoding> encodings,
List<CompressionType> compressors,
- List<String> measurementAliasList) {
+ List<String> measurementAliasList,
+ List<Map<String, String>> tagsList,
+ List<Map<String, String>> attributesList) {
TSCreateAlignedTimeseriesReq request = new TSCreateAlignedTimeseriesReq();
request.setPrefixPath(prefixPath);
request.setMeasurements(measurements);
@@ -529,6 +540,8 @@ public class Session {
request.setCompressors(
compressors.stream().map(CompressionType::ordinal).collect(Collectors.toList()));
request.setMeasurementAlias(measurementAliasList);
+ request.setTagsList(tagsList);
+ request.setAttributesList(attributesList);
return request;
}
diff --git
a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionDisableMemControlIT.java
b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionDisableMemControlIT.java
index 748790a..b8b13bb 100644
---
a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionDisableMemControlIT.java
+++
b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionDisableMemControlIT.java
@@ -151,7 +151,14 @@ public class IoTDBSessionDisableMemControlIT {
compressors.add(CompressionType.SNAPPY);
}
session.createAlignedTimeseries(
- "root.sg.d", multiMeasurementComponents, dataTypes, encodings,
compressors, null);
+ "root.sg.d",
+ multiMeasurementComponents,
+ dataTypes,
+ encodings,
+ compressors,
+ null,
+ null,
+ null);
List<MeasurementSchema> schemaList = new ArrayList<>();
schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
schemaList.add(new MeasurementSchema("s2", TSDataType.DOUBLE));
diff --git
a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorABDeviceIT.java
b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorABDeviceIT.java
index ec9cb41..72780e7 100644
---
a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorABDeviceIT.java
+++
b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorABDeviceIT.java
@@ -191,7 +191,7 @@ public class IoTDBSessionVectorABDeviceIT {
compressors.add(CompressionType.SNAPPY);
}
session.createAlignedTimeseries(
- ROOT_SG1_D1, measurements, dataTypes, encodings, compressors, null);
+ ROOT_SG1_D1, measurements, dataTypes, encodings, compressors, null,
null, null);
}
private static void prepareAlignedTimeSeriesData()
diff --git
a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorAggregationWithUnSeqIT.java
b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorAggregationWithUnSeqIT.java
index 5785310..98ee834 100644
---
a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorAggregationWithUnSeqIT.java
+++
b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorAggregationWithUnSeqIT.java
@@ -156,7 +156,7 @@ public class IoTDBSessionVectorAggregationWithUnSeqIT {
compressors.add(CompressionType.SNAPPY);
}
session.createAlignedTimeseries(
- ROOT_SG1_D1_VECTOR1, measurements, dataTypes, encodings, compressors,
null);
+ ROOT_SG1_D1_VECTOR1, measurements, dataTypes, encodings, compressors,
null, null, null);
}
private static void prepareAlignedTimeseriesDataWithUnSeq()
diff --git
a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorInsertIT.java
b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorInsertIT.java
index 54a8a40..c36fa1c 100644
---
a/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorInsertIT.java
+++
b/session/src/test/java/org/apache/iotdb/session/IoTDBSessionVectorInsertIT.java
@@ -462,7 +462,14 @@ public class IoTDBSessionVectorInsertIT {
compressors.add(CompressionType.SNAPPY);
}
session.createAlignedTimeseries(
- "root.sg.d", multiMeasurementComponents, dataTypes, encodings,
compressors, null);
+ "root.sg.d",
+ multiMeasurementComponents,
+ dataTypes,
+ encodings,
+ compressors,
+ null,
+ null,
+ null);
List<MeasurementSchema> schemaList = new ArrayList<>();
schemaList.add(new MeasurementSchema("s1", TSDataType.INT64));
schemaList.add(new MeasurementSchema("s2", TSDataType.DOUBLE));
diff --git a/thrift/src/main/thrift/rpc.thrift
b/thrift/src/main/thrift/rpc.thrift
index 78ccce6..ae0086e 100644
--- a/thrift/src/main/thrift/rpc.thrift
+++ b/thrift/src/main/thrift/rpc.thrift
@@ -314,6 +314,8 @@ struct TSCreateAlignedTimeseriesReq {
5: required list<i32> encodings
6: required list<i32> compressors
7: optional list<string> measurementAlias
+ 8: optional list<map<string, string>> tagsList
+ 9: optional list<map<string, string>> attributesList
}
struct TSRawDataQueryReq {
![https://user-images.githubusercontent.com/19167280/69792176-1718f400-1201-11ea-861a-1a83c07ca144.jpg"](https://user-images.githubusercontent.com/19167280/69792176-1718f400-1201-11ea-861a-1a83c07ca144.jpg"){kind=link}