Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ public void testFailedToCreateTimeseriesSDTProperties() {
} catch (Exception e) {
assertEquals(
TSStatusCode.ILLEGAL_PARAMETER.getStatusCode()
+ ": SDT compression deviation cannot be negative. Failed to create timeseries for path root.sg1.d0.s1",
+ ": SDT property compdev must be non-negative, but was [-2]. Failed to create timeseries for path root.sg1.d0.s1",
e.getMessage());
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,4 +26,30 @@ private DataNodeSchemaMessages() {}
public static final String
EXCEPTION_CANNOT_RECOVER_DUPLICATED_SCHEMAREGION_ARG_FOUND_IN_DATABASES_ARG_AND_ARG_D9F60E05 =
"Cannot recover duplicated SchemaRegion [%s] found in databases [%s] and [%s].";

public static final String FAILED_TO_RECOVER_TAG_INDEX =
"Failed to recover tag index for measurement [{}] in database [{}], SchemaRegion [{}], tag offset [{}].";
public static final String SCHEMA_REGION_FAILED_TO_RECOVER =
"SchemaRegion [%d] in StorageGroup [%s] failed to recover.";
public static final String SDT_COMPRESSION_DEVIATION_REQUIRED =
"SDT compression requires the compdev property";
public static final String SDT_COMPRESSION_DEVIATION_NEGATIVE =
"SDT property compdev must be non-negative, but was [%s]";
public static final String SDT_COMPRESSION_DEVIATION_FORMAT_ERROR =
"SDT property compdev must be a number, but was [%s]";
public static final String SDT_COMPRESSION_MAX_GREATER_THAN_MIN =
"SDT property compmaxtime [%s] must be greater than compmintime [%s]";
public static final String SDT_COMPRESSION_TIME_NEGATIVE =
"SDT property %s must be non-negative, but was [%s]";
public static final String SDT_COMPRESSION_TIME_FORMAT_ERROR =
"SDT property %s must be a 64-bit integer, but was [%s]";
public static final String SDT_ENABLED_NO_COMPRESSION_TIME =
"Timeseries [{}] uses SDT without property [{}]; using default [{}].";
public static final String CREATE_TEMPLATE_ERROR =
"Failed to create schema template [%s] on ConfigNode: %s";
public static final String GET_ALL_TEMPLATE_ERROR =
"Failed to get schema templates from ConfigNode: %s";
public static final String GET_TEMPLATE_INFO_ERROR =
"Failed to get schema template [%s] from ConfigNode: %s";
public static final String ILLEGAL_PATH_LOG = "illegal path {}";
}
Original file line number Diff line number Diff line change
Expand Up @@ -26,4 +26,23 @@ private DataNodeSchemaMessages() {}
public static final String
EXCEPTION_CANNOT_RECOVER_DUPLICATED_SCHEMAREGION_ARG_FOUND_IN_DATABASES_ARG_AND_ARG_D9F60E05 =
"无法恢复重复的 SchemaRegion [%s],其同时存在于数据库 [%s] 和 [%s] 中。";

public static final String FAILED_TO_RECOVER_TAG_INDEX =
"恢复测点 [{}] 的 tag 索引失败,数据库为 [{}],SchemaRegion 为 [{}],tag 偏移量为 [{}]。";
public static final String SCHEMA_REGION_FAILED_TO_RECOVER =
"SchemaRegion [%d] 在 StorageGroup [%s] 中恢复失败。";
public static final String SDT_COMPRESSION_DEVIATION_REQUIRED = "SDT 压缩必须设置 compdev 属性";
public static final String SDT_COMPRESSION_DEVIATION_NEGATIVE = "SDT 属性 compdev 必须为非负数,实际值为 [%s]";
public static final String SDT_COMPRESSION_DEVIATION_FORMAT_ERROR =
"SDT 属性 compdev 必须为数字,实际值为 [%s]";
public static final String SDT_COMPRESSION_MAX_GREATER_THAN_MIN =
"SDT 属性 compmaxtime [%s] 必须大于 compmintime [%s]";
public static final String SDT_COMPRESSION_TIME_NEGATIVE = "SDT 属性 %s 必须为非负数,实际值为 [%s]";
public static final String SDT_COMPRESSION_TIME_FORMAT_ERROR = "SDT 属性 %s 必须为 64 位整数,实际值为 [%s]";
public static final String SDT_ENABLED_NO_COMPRESSION_TIME =
"时间序列 [{}] 已启用 SDT 但未设置属性 [{}],使用默认值 [{}]。";
public static final String CREATE_TEMPLATE_ERROR = "在 ConfigNode 上创建元数据模板 [%s] 失败:%s";
public static final String GET_ALL_TEMPLATE_ERROR = "从 ConfigNode 获取元数据模板列表失败:%s";
public static final String GET_TEMPLATE_INFO_ERROR = "从 ConfigNode 获取元数据模板 [%s] 失败:%s";
public static final String ILLEGAL_PATH_LOG = "非法路径 {}";
}
Original file line number Diff line number Diff line change
Expand Up @@ -342,11 +342,12 @@ private Callable<ISchemaRegion> recoverSchemaRegionTask(
timeRecord);
return schemaRegion;
} catch (MetadataException e) {
logger.error(
throw new RuntimeException(
String.format(
"SchemaRegion [%d] in StorageGroup [%s] failed to recover.",
schemaRegionId.getId(), storageGroup.getFullPath()));
throw new RuntimeException(e);
DataNodeSchemaMessages.SCHEMA_REGION_FAILED_TO_RECOVER,
schemaRegionId.getId(),
storageGroup.getFullPath()),
e);
}
};
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import org.apache.iotdb.db.exception.metadata.SchemaDirCreationFailureException;
import org.apache.iotdb.db.exception.metadata.SchemaQuotaExceededException;
import org.apache.iotdb.db.exception.metadata.SeriesOverflowException;
import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;
import org.apache.iotdb.db.queryengine.common.schematree.ClusterSchemaTree;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metedata.write.CreateTimeSeriesNode;
import org.apache.iotdb.db.schemaengine.metric.ISchemaRegionMetric;
Expand Down Expand Up @@ -109,8 +110,6 @@
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.apache.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;

/**
* This class takes the responsibility of serialization of all the metadata info of one certain
* schema region and persistent it into files. This class contains the interfaces to modify the
Expand Down Expand Up @@ -508,9 +507,12 @@ public void loadSnapshot(File latestSnapshotRootDir) {
tagManager.recoverIndex(measurementMNode.getOffset(), measurementMNode);
} catch (IOException e) {
logger.error(
"Failed to recover tagIndex for {} in schemaRegion {}.",
storageGroupFullPath + PATH_SEPARATOR + measurementMNode.getFullPath(),
schemaRegionId);
DataNodeSchemaMessages.FAILED_TO_RECOVER_TAG_INDEX,
measurementMNode.getName(),
storageGroupFullPath,
schemaRegionId,
measurementMNode.getOffset(),
e);
}
},
deviceMNode -> {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.exception.metadata.AliasAlreadyExistException;
import org.apache.iotdb.db.exception.metadata.IllegalParameterOfPathException;
import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -145,25 +146,36 @@ public static void checkTimeseriesProps(String path, Map<String, String> props)
private static void checkSDTFormat(String path, Map<String, String> props)
throws IllegalParameterOfPathException {
if (!props.containsKey(SDT_COMP_DEV)) {
throw new IllegalParameterOfPathException("SDT compression deviation is required", path);
throw new IllegalParameterOfPathException(
DataNodeSchemaMessages.SDT_COMPRESSION_DEVIATION_REQUIRED, path);
}

try {
double d = Double.parseDouble(props.get(SDT_COMP_DEV));
if (d < 0) {
throw new IllegalParameterOfPathException(
"SDT compression deviation cannot be negative", path);
String.format(
DataNodeSchemaMessages.SDT_COMPRESSION_DEVIATION_NEGATIVE, props.get(SDT_COMP_DEV)),
path);
}
} catch (NumberFormatException e) {
throw new IllegalParameterOfPathException("SDT compression deviation formatting error", path);
throw new IllegalParameterOfPathException(
String.format(
DataNodeSchemaMessages.SDT_COMPRESSION_DEVIATION_FORMAT_ERROR,
props.get(SDT_COMP_DEV)),
path);
}

long compMinTime = sdtCompressionTimeFormat(SDT_COMP_MIN_TIME, props, path);
long compMaxTime = sdtCompressionTimeFormat(SDT_COMP_MAX_TIME, props, path);

if (compMaxTime <= compMinTime) {
throw new IllegalParameterOfPathException(
"SDT compression maximum time needs to be greater than compression minimum time", path);
String.format(
DataNodeSchemaMessages.SDT_COMPRESSION_MAX_GREATER_THAN_MIN,
compMaxTime,
compMinTime),
path);
}
}

Expand All @@ -172,20 +184,23 @@ private static long sdtCompressionTimeFormat(
throws IllegalParameterOfPathException {
boolean isCompMaxTime = compTime.equals(SDT_COMP_MAX_TIME);
long time = isCompMaxTime ? Long.MAX_VALUE : 0;
String s = isCompMaxTime ? "maximum" : "minimum";
if (props.containsKey(compTime)) {
String value = props.get(compTime);
try {
time = Long.parseLong(props.get(compTime));
if (time < 0) {
throw new IllegalParameterOfPathException(
String.format("SDT compression %s time cannot be negative", s), path);
}
} catch (IllegalParameterOfPathException e) {
time = Long.parseLong(value);
} catch (NumberFormatException e) {
throw new IllegalParameterOfPathException(
String.format(
DataNodeSchemaMessages.SDT_COMPRESSION_TIME_FORMAT_ERROR, compTime, value),
path);
}
if (time < 0) {
throw new IllegalParameterOfPathException(
String.format("SDT compression %s time formatting error", s), path);
String.format(DataNodeSchemaMessages.SDT_COMPRESSION_TIME_NEGATIVE, compTime, value),
path);
}
} else {
logger.info("{} enabled SDT but did not set compression {} time", path, s);
logger.debug(DataNodeSchemaMessages.SDT_ENABLED_NO_COMPRESSION_TIME, path, compTime, time);
}
return time;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesResp;
import org.apache.iotdb.confignode.rpc.thrift.TGetTemplateResp;
import org.apache.iotdb.confignode.rpc.thrift.TSetSchemaTemplateReq;
import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;
import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
Expand Down Expand Up @@ -130,9 +131,11 @@ public TSStatus createSchemaTemplate(CreateSchemaTemplateStatement statement) {
e,
TSStatusCode.CREATE_TEMPLATE_ERROR.getStatusCode()));
} catch (ClientManagerException | TException e) {
throw new RuntimeException(
new IoTDBException(
"create template error.", e, TSStatusCode.CREATE_TEMPLATE_ERROR.getStatusCode()));
throw wrapTemplateRpcException(
String.format(
DataNodeSchemaMessages.CREATE_TEMPLATE_ERROR, statement.getName(), e.getMessage()),
e,
TSStatusCode.CREATE_TEMPLATE_ERROR);
}
}

Expand Down Expand Up @@ -175,9 +178,10 @@ public List<Template> getAllTemplates() {
throw new IoTDBRuntimeException(resp.getStatus());
}
} catch (ClientManagerException | TException e) {
throw new RuntimeException(
new IoTDBException(
"get all template error.", TSStatusCode.UNDEFINED_TEMPLATE.getStatusCode()));
throw wrapTemplateRpcException(
String.format(DataNodeSchemaMessages.GET_ALL_TEMPLATE_ERROR, e.getMessage()),
e,
TSStatusCode.UNDEFINED_TEMPLATE);
}
return templatesList;
}
Expand All @@ -196,12 +200,20 @@ public Template getTemplate(String name) throws IoTDBException {
throw new IoTDBException(resp.status);
}
} catch (ClientManagerException | TException e) {
throw new RuntimeException(
new IoTDBException(
"get template info error.", TSStatusCode.UNDEFINED_TEMPLATE.getStatusCode()));
throw wrapTemplateRpcException(
String.format(DataNodeSchemaMessages.GET_TEMPLATE_INFO_ERROR, name, e.getMessage()),
e,
TSStatusCode.UNDEFINED_TEMPLATE);
}
}

private static RuntimeException wrapTemplateRpcException(
String message, Exception cause, TSStatusCode statusCode) {
IoTDBException exception = new IoTDBException(message, statusCode.getStatusCode());
exception.addSuppressed(cause);
return new RuntimeException(exception);
}

@Override
public void setSchemaTemplate(String queryId, String name, PartialPath path) {
try (ConfigNodeClient configNodeClient =
Expand Down Expand Up @@ -256,7 +268,7 @@ public List<PartialPath> getPathsSetTemplate(String name, PathPatternTree scope)
try {
listPath.add(new PartialPath(item));
} catch (IllegalPathException e) {
e.printStackTrace();
LOGGER.error(DataNodeSchemaMessages.ILLEGAL_PATH_LOG, item, e);
}
});
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
/*
* 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.iotdb.db.metadata;

import org.apache.iotdb.db.exception.metadata.IllegalParameterOfPathException;
import org.apache.iotdb.db.schemaengine.schemaregion.utils.MetaFormatUtils;
import org.apache.iotdb.rpc.TSStatusCode;

import org.junit.Test;

import java.util.HashMap;
import java.util.Map;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;

public class MetaFormatUtilsTest {

private static final String PATH = "root.sg.d.s";

@Test
public void testNonNumericSdtProperties() {
for (String property : new String[] {"compdev", "compmintime", "compmaxtime"}) {
assertInvalidProperty(property, "invalid");
}
}

@Test
public void testNegativeSdtProperties() {
for (String property : new String[] {"compdev", "compmintime", "compmaxtime"}) {
assertInvalidProperty(property, "-1");
}
}

@Test
public void testSdtTimeOverflow() {
for (String property : new String[] {"compmintime", "compmaxtime"}) {
assertInvalidProperty(property, "9223372036854775808");
assertInvalidProperty(property, "-9223372036854775809");
}
}

@Test
public void testInvalidSdtTimeRange() {
Map<String, String> props = sdtProps();
props.put("compmintime", "20");
props.put("compmaxtime", "10");
IllegalParameterOfPathException exception =
assertThrows(
IllegalParameterOfPathException.class,
() -> MetaFormatUtils.checkTimeseriesProps(PATH, props));

assertUserParameterError(exception);
assertTrue(exception.getMessage().contains("compmintime [20]"));
assertTrue(exception.getMessage().contains("compmaxtime [10]"));
}

@Test
public void testValidSdtTimeBoundsAndDefaults() throws IllegalParameterOfPathException {
Map<String, String> props = sdtProps();
MetaFormatUtils.checkTimeseriesProps(PATH, props);
props.put("compmintime", "0");
MetaFormatUtils.checkTimeseriesProps(PATH, props);
props.put("compmaxtime", Long.toString(Long.MAX_VALUE));
MetaFormatUtils.checkTimeseriesProps(PATH, props);
}

private static void assertInvalidProperty(String property, String value) {
Map<String, String> props = sdtProps();
props.put(property, value);
IllegalParameterOfPathException exception =
assertThrows(
IllegalParameterOfPathException.class,
() -> MetaFormatUtils.checkTimeseriesProps(PATH, props));

assertUserParameterError(exception);
assertTrue(exception.getMessage().contains(property));
assertTrue(exception.getMessage().contains("[" + value + "]"));
}

private static void assertUserParameterError(IllegalParameterOfPathException exception) {
assertEquals(TSStatusCode.ILLEGAL_PARAMETER.getStatusCode(), exception.getErrorCode());
assertTrue(exception.isUserException());
assertTrue(exception.getMessage().contains(PATH));
}

private static Map<String, String> sdtProps() {
Map<String, String> props = new HashMap<>();
props.put("loss", "sdt");
props.put("compdev", "0.01");
return props;
}
}
Loading
Loading