You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2026/04/08 04:06:03 UTC
(iotdb) branch rc/2.0.8 created (now 689d072419a)
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a change to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 689d072419a correct metadata "modified" flags and related logic (#17172)
This branch includes the following new commits:
new e30c8ba3176 Fix blob accumulator query can't skip read blobStatistics caused by isModified don't change to true in the metadata of chunk and pageReader (#17127)
new 8b954fb9ed7 Fix value and time chunk misalignment when rewriting with time deletion (#17142)
new 29dea12bf5a fix: Fix ShowTimeseries estimated offset without limit scenario (#17116)
new 87511974e9d fix (#17140)
new 016208a9228 fix the bug that insert null into the time column #17150
new 5e961585986 Fix occasional CI failures of IoTDBMQTTServiceJsonIT. (#17146)
new 5e48375f066 Load: Support same measurement name and type with different encoding and compression (#17152)
new e4c0582bb1a Pipe: Banned the illegal names in pipe and pipe plugins (#17145)
new f9807d9290d Pipe: allow exclusion-only pattern lists (#17154)
new 1527e5317a8 Fixed the NPE when validating legacy sink (#17153)
new 26a93d8e08a Fix: Add Mods data scanning check in scan parser for non-aligned data (#17162)
new b4072594031 fix: Fix missing encoding and compression information for Measurement (#17141)
new c35a003c452 Pipe: preserve exclusions for complex wildcards (#17170)
new e26d075d3e9 [AINode] Optimize the statuscode for remove AINode (#17176)
new 26537f849bd Pipe: Changed the separator check semantic for pipes / pipe plugins (#17175)
new 689d072419a correct metadata "modified" flags and related logic (#17172)
The 16 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 04/16: fix (#17140)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 87511974e9df2c9e5630a9cbef119bc8ad5c7061
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Tue Feb 3 17:23:22 2026 +0800
fix (#17140)
---
.../org/apache/iotdb/confignode/manager/ClusterQuotaManager.java | 5 +++--
.../org/apache/iotdb/confignode/persistence/quota/QuotaInfo.java | 4 +---
2 files changed, 4 insertions(+), 5 deletions(-)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterQuotaManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterQuotaManager.java
index b6bff890f36..68b7c1680a8 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterQuotaManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterQuotaManager.java
@@ -47,6 +47,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
@@ -111,8 +112,8 @@ public class ClusterQuotaManager {
/** If the new quota is smaller than the quota already used, the setting fails. */
private boolean checkSpaceQuota(TSetSpaceQuotaReq req) {
for (String database : req.getDatabase()) {
- if (quotaInfo.getSpaceQuotaLimit().containsKey(database)) {
- TSpaceQuota spaceQuota = quotaInfo.getSpaceQuotaUsage().get(database);
+ TSpaceQuota spaceQuota = quotaInfo.getSpaceQuotaUsage().get(database);
+ if (Objects.nonNull(spaceQuota)) {
if (req.getSpaceLimit().getDeviceNum() != IoTDBConstant.UNLIMITED_VALUE
&& req.getSpaceLimit().getDeviceNum() != IoTDBConstant.DEFAULT_VALUE
&& spaceQuota.getDeviceNum() > req.getSpaceLimit().getDeviceNum()) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/quota/QuotaInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/quota/QuotaInfo.java
index 59b484de0fc..93d2b272109 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/quota/QuotaInfo.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/quota/QuotaInfo.java
@@ -87,9 +87,7 @@ public class QuotaInfo implements SnapshotProcessor {
spaceQuota.setDiskSize(IoTDBConstant.DEFAULT_VALUE);
}
}
- if (!spaceQuotaUsage.containsKey(database)) {
- spaceQuotaUsage.put(database, new TSpaceQuota());
- }
+ spaceQuotaUsage.computeIfAbsent(database, k -> new TSpaceQuota());
spaceQuotaLimit.put(database, spaceQuota);
}
return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
(iotdb) 05/16: fix the bug that insert null into the time column #17150
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 016208a9228c1fbafd26c95bd716e673f837a364
Author: alpass163 <al...@163.com>
AuthorDate: Tue Feb 3 18:10:03 2026 +0800
fix the bug that insert null into the time column #17150
---
.../relational/it/db/it/IoTDBInsertTableIT.java | 36 ++++++++++++++++++++++
.../plan/relational/sql/parser/AstBuilder.java | 2 ++
2 files changed, 38 insertions(+)
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java
index b079d022e79..c223761cd2a 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java
@@ -354,6 +354,42 @@ public class IoTDBInsertTableIT {
}
}
+ @Test
+ public void testInsertNullIntoTime() {
+ try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("use \"test\"");
+ statement.execute(
+ "CREATE TABLE insert_null (region string tag, time_test timestamp time, s1 int64 field)");
+
+ try {
+ statement.execute("insert into insert_null values('SZ', null, 1)");
+ fail("expected exception");
+ } catch (SQLException e) {
+ assertEquals("701: Timestamp cannot be null", e.getMessage());
+ }
+
+ try {
+ statement.execute("insert into insert_null(region, time_test, s1) values('SZ', null, 1)");
+ fail("expected exception");
+ } catch (SQLException e) {
+ assertEquals("701: Timestamp cannot be null", e.getMessage());
+ }
+
+ try {
+ statement.execute(
+ "insert into insert_null(region, time_test, s1) values('SZ', 2, 2),('HK', null ,3)");
+ fail("expected exception");
+ } catch (SQLException e) {
+ assertEquals("701: Timestamp cannot be null", e.getMessage());
+ }
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
@Test
public void testInsertNaN() throws SQLException {
try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
index 1ad212cf560..83105468a02 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
@@ -992,6 +992,8 @@ public class AstBuilder extends RelationalSqlBaseVisitor<Node> {
Expression timeExpression = expressions.get(timeColumnIndex);
if (timeExpression instanceof LongLiteral) {
timestamp = ((LongLiteral) timeExpression).getParsedValue();
+ } else if (timeExpression instanceof NullLiteral) {
+ throw new SemanticException("Timestamp cannot be null");
} else {
timestamp =
parseDateTimeFormat(
(iotdb) 07/16: Load: Support same measurement name and type with different encoding and compression (#17152)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5e48375f0663f5184527de8ee8a563d302280b6b
Author: Zhenyu Luo <zh...@apache.org>
AuthorDate: Wed Feb 4 09:35:57 2026 +0800
Load: Support same measurement name and type with different encoding and compression (#17152)
* Support same measurement name and type with different encoding and compression
This PR allows measurements with the same name and type to have different
encoding and compression types during TsFile loading. Previously, duplicate
measurements would throw an exception. Now, if measurements have the same
name and type but different encoding/compression, they will be deduplicated
by keeping only one schema entry.
Changes:
- Add updateDevice2TimeSeries method in LoadTsFileTreeSchemaCache to support
updating the device-to-timeseries mapping
- Modify makeSureNoDuplicatedMeasurementsInDevices in TreeSchemaAutoCreatorAndVerifier
to allow same measurement with different encoding/compression by deduplicating
instead of throwing an exception
- Only throw exception when duplicate measurements have different data types
* fix
* fix
* add IT
* spotless
---
.../org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java | 53 ++++++++++++++++++++++
.../load/TreeSchemaAutoCreatorAndVerifier.java | 32 +++++++++++--
2 files changed, 81 insertions(+), 4 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java
index 928c7875b19..19a4214bed2 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java
@@ -1058,6 +1058,59 @@ public class IoTDBLoadTsFileIT {
}
}
+ @Test
+ public void testLoadWithSameMeasurementNameDifferentDevice() throws Exception {
+ final String device = "root.sg.test_0.device_1";
+ MeasurementSchema measurement =
+ new MeasurementSchema("temperature", TSDataType.DOUBLE, TSEncoding.GORILLA);
+
+ final long writtenPoint1;
+ try (final TsFileGenerator generator =
+ new TsFileGenerator(new File(tmpDir, "same-measurement-1.tsfile"))) {
+ generator.registerTimeseries(device, Collections.singletonList(measurement));
+ generator.generateData(device, 1000, PARTITION_INTERVAL, false);
+ writtenPoint1 = generator.getTotalNumber();
+ }
+
+ measurement = new MeasurementSchema("temperature", TSDataType.DOUBLE, TSEncoding.PLAIN);
+ final long writtenPoint2;
+ try (final TsFileGenerator generator =
+ new TsFileGenerator(new File(tmpDir, "same-measurement-2.tsfile"))) {
+ generator.registerTimeseries(device, Collections.singletonList(measurement));
+ generator.generateData(device, 2000, PARTITION_INTERVAL / 10000, false);
+ writtenPoint2 = generator.getTotalNumber();
+ }
+
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement()) {
+
+ statement.execute(String.format("load \"%s\" sglevel=2", tmpDir.getAbsolutePath()));
+
+ try (final ResultSet resultSet = statement.executeQuery("select count(**) from root.sg.**")) {
+ if (resultSet.next()) {
+ final long sg1Count = resultSet.getLong("count(root.sg.test_0.device_1.temperature)");
+ Assert.assertEquals(writtenPoint1 + writtenPoint2, sg1Count);
+ } else {
+ Assert.fail("This ResultSet is empty.");
+ }
+ }
+
+ try (final ResultSet resultSet = statement.executeQuery("show timeseries root.sg.**")) {
+ int count = 0;
+ Set<String> expectedPaths = new HashSet<>();
+ expectedPaths.add(device + "." + measurement.getMeasurementName());
+ while (resultSet.next()) {
+ String path = resultSet.getString(ColumnHeaderConstant.TIMESERIES);
+ Assert.assertTrue("Unexpected timeseries path: " + path, expectedPaths.contains(path));
+ expectedPaths.remove(path);
+ count++;
+ }
+ Assert.assertEquals(1, count);
+ Assert.assertTrue("Not all expected timeseries found", expectedPaths.isEmpty());
+ }
+ }
+ }
+
@Test
@Ignore("Load with conversion is currently banned")
public void testLoadWithConvertOnTypeMismatchForTreeModel() throws Exception {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java
index c174bd7942b..29d4f1be07b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/TreeSchemaAutoCreatorAndVerifier.java
@@ -252,17 +252,41 @@ public class TreeSchemaAutoCreatorAndVerifier {
}
private void makeSureNoDuplicatedMeasurementsInDevices() throws LoadAnalyzeException {
+ boolean hasDuplicates = false;
+ final Map<IDeviceID, Set<MeasurementSchema>> deduplicatedDevice2TimeSeries = new HashMap<>();
+
for (final Map.Entry<IDeviceID, Set<MeasurementSchema>> entry :
schemaCache.getDevice2TimeSeries().entrySet()) {
final IDeviceID device = entry.getKey();
final Map<String, MeasurementSchema> measurement2Schema = new HashMap<>();
+ boolean deviceHasDuplicates = false;
+
for (final MeasurementSchema timeseriesSchema : entry.getValue()) {
final String measurement = timeseriesSchema.getMeasurementName();
- if (measurement2Schema.containsKey(measurement)) {
- throw new LoadAnalyzeException(
- String.format("Duplicated measurements %s in device %s.", measurement, device));
+ final MeasurementSchema existingSchema = measurement2Schema.get(measurement);
+
+ if (existingSchema != null) {
+ if (existingSchema.getType() != timeseriesSchema.getType()) {
+ throw new LoadAnalyzeException(
+ String.format("Duplicated measurements %s in device %s.", measurement, device));
+ }
+ deviceHasDuplicates = true;
+ hasDuplicates = true;
+ } else {
+ measurement2Schema.put(measurement, timeseriesSchema);
}
- measurement2Schema.put(measurement, timeseriesSchema);
+ }
+
+ if (deviceHasDuplicates) {
+ deduplicatedDevice2TimeSeries.put(device, new HashSet<>(measurement2Schema.values()));
+ }
+ }
+
+ if (hasDuplicates) {
+ Map<IDeviceID, Set<MeasurementSchema>> device2TimeSeries = schemaCache.getDevice2TimeSeries();
+ for (final Map.Entry<IDeviceID, Set<MeasurementSchema>> entry :
+ deduplicatedDevice2TimeSeries.entrySet()) {
+ device2TimeSeries.put(entry.getKey(), new HashSet<>(entry.getValue()));
}
}
}
(iotdb) 14/16: [AINode] Optimize the statuscode for remove AINode (#17176)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e26d075d3e93dade6d8e82537b0cc149c6d5a032
Author: Yongzao <yo...@apache.org>
AuthorDate: Fri Feb 6 14:25:51 2026 +0800
[AINode] Optimize the statuscode for remove AINode (#17176)
---
.../iotdb/confignode/manager/node/NodeManager.java | 2 +-
.../config/executor/ClusterConfigTaskExecutor.java | 17 +++++++++--------
2 files changed, 10 insertions(+), 9 deletions(-)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index 0c1d409cc70..e3d775259d6 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -558,7 +558,7 @@ public class NodeManager {
public TSStatus removeAINode() {
// check if the node exists
if (nodeInfo.getRegisteredAINodes().isEmpty()) {
- return new TSStatus(TSStatusCode.REMOVE_AI_NODE_ERROR.getStatusCode())
+ return new TSStatus(TSStatusCode.NO_REGISTERED_AI_NODE_ERROR.getStatusCode())
.setMessage("Remove AINode failed because there is no AINode in the cluster.");
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 3627e3f82af..e635aa5c283 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -3548,15 +3548,16 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
LOGGER.info("Starting to remove AINode");
try (ConfigNodeClient configNodeClient =
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
- TShowClusterResp showClusterResp = configNodeClient.showCluster();
- if (showClusterResp.getAiNodeListSize() < 1) {
- LOGGER.error("Remove AINode failed because there is no AINode in the cluster.");
- future.setException(
- new IOException("Remove AINode failed because there is no AINode in the cluster."));
- return future;
- }
TSStatus status = configNodeClient.removeAINode(new TAINodeRemoveReq());
- if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ if (status.getCode() == TSStatusCode.NO_REGISTERED_AI_NODE_ERROR.getStatusCode()) {
+ LOGGER.warn("Remove AINode failed because there is no AINode in the cluster.");
+ future.set(
+ new ConfigTaskResult(
+ new TSStatus(TSStatusCode.NO_REGISTERED_AI_NODE_ERROR.getStatusCode())
+ .setMessage(
+ "Remove AINode failed because there is no AINode in the cluster.")));
+ return future;
+ } else if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
future.setException(new IOException("Remove AINode failed: " + status.getMessage()));
return future;
} else {
(iotdb) 10/16: Fixed the NPE when validating legacy sink (#17153)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1527e5317a80849e2526df449d0e4c3c2d314f89
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Wed Feb 4 14:14:58 2026 +0800
Fixed the NPE when validating legacy sink (#17153)
* npe-fix
* unb
---
.../treemodel/auto/basic/IoTDBPipeDataSinkIT.java | 34 +++++++---------------
.../sink/protocol/legacy/IoTDBLegacyPipeSink.java | 11 ++++---
2 files changed, 18 insertions(+), 27 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
index 0a927cab298..035d5052cb7 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeAutoBasic;
@@ -35,6 +36,8 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
+import java.sql.Connection;
+import java.sql.Statement;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -211,31 +214,16 @@ public class IoTDBPipeDataSinkIT extends AbstractPipeDualTreeModelAutoIT {
final String receiverIp = receiverDataNode.getIp();
final int receiverPort = receiverDataNode.getPort();
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe testPipe ('sink'='iotdb-legacy-pipe-sink', 'ip'='%s', 'port'='%s', 'version'='1.3')",
+ receiverIp, receiverPort));
+ }
+
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
- final Map<String, String> sourceAttributes = new HashMap<>();
- final Map<String, String> processorAttributes = new HashMap<>();
- final Map<String, String> sinkAttributes = new HashMap<>();
-
- sourceAttributes.put("source.realtime.mode", "log");
- sourceAttributes.put("user", "root");
-
- sinkAttributes.put("sink", "iotdb-legacy-pipe-sink");
- sinkAttributes.put("sink.batch.enable", "false");
- sinkAttributes.put("sink.ip", receiverIp);
- sinkAttributes.put("sink.port", Integer.toString(receiverPort));
-
- // This version does not matter since it's no longer checked by the legacy receiver
- sinkAttributes.put("sink.version", "1.3");
-
- final TSStatus status =
- client.createPipe(
- new TCreatePipeReq("testPipe", sinkAttributes)
- .setExtractorAttributes(sourceAttributes)
- .setProcessorAttributes(processorAttributes));
-
- Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
-
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe("testPipe").getCode());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
index eefa5e739d5..332ca6bab7c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
@@ -37,6 +37,7 @@ import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.sink.payload.legacy.TsFilePipeData;
import org.apache.iotdb.db.storageengine.StorageEngine;
+import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.pipe.api.PipeConnector;
import org.apache.iotdb.pipe.api.annotation.TreeModel;
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
@@ -110,7 +111,7 @@ public class IoTDBLegacyPipeSink implements PipeConnector {
private String syncConnectorVersion;
private String pipeName;
- private String databaseName;
+ private String databaseName = "";
private IoTDBSyncClient client;
@@ -203,10 +204,12 @@ public class IoTDBLegacyPipeSink implements PipeConnector {
trustStore = parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PATH_KEY);
trustStorePwd = parameters.getString(SINK_IOTDB_SSL_TRUST_STORE_PWD_KEY);
- databaseName =
+ final DataRegion dataRegion =
StorageEngine.getInstance()
- .getDataRegion(new DataRegionId(configuration.getRuntimeEnvironment().getRegionId()))
- .getDatabaseName();
+ .getDataRegion(new DataRegionId(configuration.getRuntimeEnvironment().getRegionId()));
+ if (Objects.nonNull(dataRegion)) {
+ databaseName = dataRegion.getDatabaseName();
+ }
}
@Override
(iotdb) 15/16: Pipe: Changed the separator check semantic for pipes / pipe plugins (#17175)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 26537f849bd78a50053341ff04747ab812f275f8
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Fri Feb 6 14:48:29 2026 +0800
Pipe: Changed the separator check semantic for pipes / pipe plugins (#17175)
* fix
* sep
---
.../iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSyntaxIT.java | 2 +-
.../src/main/java/org/apache/iotdb/commons/utils/FileUtils.java | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSyntaxIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSyntaxIT.java
index 844477b6590..6ae8d56bfaa 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSyntaxIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSyntaxIT.java
@@ -342,7 +342,7 @@ public class IoTDBPipeSyntaxIT extends AbstractPipeDualTreeModelAutoIT {
public void testDirectoryErrors() throws SQLException {
try (final Connection connection = senderEnv.getConnection();
final Statement statement = connection.createStatement()) {
- List<String> wrongDirs = Arrays.asList(".", "..", "./hackYou", ".\\hackYouTwice");
+ List<String> wrongDirs = Arrays.asList(".", "..", "/hackYou", "..\\hackYouTwice");
if (SystemUtils.IS_OS_WINDOWS) {
wrongDirs = new ArrayList<>(wrongDirs);
wrongDirs.add("BombWindows/:*?");
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
index 595cd5e831c..139d08ef99b 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
@@ -583,7 +583,7 @@ public class FileUtils {
}
public static String getIllegalError4Directory(final String path) {
- if (path.equals(".") || path.equals("..") || path.contains("./") || path.contains(".\\")) {
+ if (path.equals(".") || path.equals("..") || path.contains("/") || path.contains("\\")) {
return ILLEGAL_PATH_MESSAGE;
}
if (!WindowsOSUtils.isLegalPathSegment4Windows(path)) {
(iotdb) 03/16: fix: Fix ShowTimeseries estimated offset without limit scenario (#17116)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 29dea12bf5a163c2b02242522d5486580580fa50
Author: Zhenyu Luo <zh...@apache.org>
AuthorDate: Tue Feb 3 12:00:36 2026 +0800
fix: Fix ShowTimeseries estimated offset without limit scenario (#17116)
* fix: Fix ShowTimeseries estimated offset without limit scenario
When ShowTimeseries has offset but no limit, the limit calculation
was incorrectly adding offset to 0, causing incorrect behavior.
This commit fixes:
1. LogicalPlanVisitor: When limit is 0, keep it as 0 instead of
adding offset
2. SchemaReaderLimitOffsetWrapper: When limit is 0, allow all
results to pass through (limit == 0 means no limit)
* fix
* fix
* fix
* add it
---
.../org/apache/iotdb/db/it/IoTDBSimpleQueryIT.java | 244 +++++++++++++++++++++
.../plan/planner/LogicalPlanVisitor.java | 15 +-
.../impl/SchemaReaderLimitOffsetWrapper.java | 2 +-
3 files changed, 257 insertions(+), 4 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSimpleQueryIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSimpleQueryIT.java
index 9cdd1cc3ce9..1d5765deafd 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSimpleQueryIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSimpleQueryIT.java
@@ -698,6 +698,250 @@ public class IoTDBSimpleQueryIT {
}
}
+ @Test
+ public void testShowTimeseriesWithOffsetOnly() throws SQLException {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+
+ // Test Case 1: Single Region scenario (single database, single device)
+ // Create first database and write data to single device to ensure single Region
+ statement.execute("CREATE DATABASE root.db1");
+ statement.execute("INSERT INTO root.db1.d0(timestamp, s1) VALUES (1, 1)");
+ statement.execute("INSERT INTO root.db1.d0(timestamp, s2) VALUES (1, 2)");
+ statement.execute("INSERT INTO root.db1.d0(timestamp, s3) VALUES (1, 3)");
+ statement.execute("flush");
+
+ // Verify single Region scenario - offset only
+ List<String> sg1AllPaths = new ArrayList<>();
+ try (ResultSet resultSet = statement.executeQuery("show timeseries root.db1.**")) {
+ while (resultSet.next()) {
+ sg1AllPaths.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(3, sg1AllPaths.size());
+
+ int offset1 = 2;
+ List<String> sg1ResultPaths = new ArrayList<>();
+ try (ResultSet resultSet =
+ statement.executeQuery("show timeseries root.db1.** offset " + offset1)) {
+ while (resultSet.next()) {
+ sg1ResultPaths.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(sg1AllPaths.size() - offset1, sg1ResultPaths.size());
+ for (int i = 0; i < sg1ResultPaths.size(); i++) {
+ Assert.assertEquals(sg1AllPaths.get(offset1 + i), sg1ResultPaths.get(i));
+ }
+
+ // Test Case 2: Single Region - different path patterns with offset
+ List<String> d0Paths = new ArrayList<>();
+ try (ResultSet resultSet = statement.executeQuery("show timeseries root.db1.d0.**")) {
+ while (resultSet.next()) {
+ d0Paths.add(resultSet.getString(1));
+ }
+ }
+
+ int offset2 = 1;
+ List<String> d0ResultPaths = new ArrayList<>();
+ try (ResultSet resultSet =
+ statement.executeQuery("show timeseries root.db1.d0.** offset " + offset2)) {
+ while (resultSet.next()) {
+ d0ResultPaths.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(d0Paths.size() - offset2, d0ResultPaths.size());
+ for (int i = 0; i < d0ResultPaths.size(); i++) {
+ Assert.assertEquals(d0Paths.get(offset2 + i), d0ResultPaths.get(i));
+ }
+
+ // Test Case 3: Single Region - with time condition and offset
+ // Insert data with different timestamps
+ statement.execute("INSERT INTO root.db1.d0(timestamp, s4) VALUES (10, 10)");
+ statement.execute("INSERT INTO root.db1.d0(timestamp, s5) VALUES (20, 20)");
+ statement.execute("INSERT INTO root.db1.d0(timestamp, s6) VALUES (30, 30)");
+ statement.execute("flush");
+
+ // Get all timeseries with time condition
+ List<String> timeFilteredPaths = new ArrayList<>();
+ try (ResultSet resultSet =
+ statement.executeQuery("show timeseries root.db1.** where time > 5")) {
+ while (resultSet.next()) {
+ timeFilteredPaths.add(resultSet.getString(1));
+ }
+ }
+
+ // Test offset with time condition in single Region
+ int offset3 = 1;
+ List<String> timeFilteredResultPaths = new ArrayList<>();
+ try (ResultSet resultSet =
+ statement.executeQuery("show timeseries root.db1.** where time > 5 offset " + offset3)) {
+ while (resultSet.next()) {
+ timeFilteredResultPaths.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(timeFilteredPaths.size() - offset3, timeFilteredResultPaths.size());
+ for (int i = 0; i < timeFilteredResultPaths.size(); i++) {
+ Assert.assertEquals(timeFilteredPaths.get(offset3 + i), timeFilteredResultPaths.get(i));
+ }
+
+ // Test Case 4: Multiple Regions scenario (multiple databases)
+ // Create second database to test multi-Region scenario
+ statement.execute("CREATE DATABASE root.db2");
+ statement.execute("INSERT INTO root.db2.d0(timestamp, s1) VALUES (4, 7)");
+ statement.execute("INSERT INTO root.db2.d0(timestamp, s2) VALUES (4, 8)");
+ statement.execute("INSERT INTO root.db2.d1(timestamp, s1) VALUES (5, 9)");
+ statement.execute("flush");
+
+ // Test across multiple databases - offset only
+ List<String> allPaths = new ArrayList<>();
+ try (ResultSet resultSet = statement.executeQuery("show timeseries root.db*.**")) {
+ while (resultSet.next()) {
+ allPaths.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(9, allPaths.size());
+
+ int offset4 = 3;
+ List<String> resultPaths = new ArrayList<>();
+ try (ResultSet resultSet =
+ statement.executeQuery("show timeseries root.db*.** offset " + offset4)) {
+ while (resultSet.next()) {
+ resultPaths.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(allPaths.size() - offset4, resultPaths.size());
+ for (int i = 0; i < resultPaths.size(); i++) {
+ Assert.assertEquals(allPaths.get(offset4 + i), resultPaths.get(i));
+ }
+
+ // Test Case 5: Edge case - offset equals or exceeds total count
+ int largeOffset = allPaths.size() + 10;
+ int count = 0;
+ try (ResultSet resultSet = statement.executeQuery("show timeseries offset " + largeOffset)) {
+ while (resultSet.next()) {
+ count++;
+ }
+ }
+ Assert.assertEquals("Should return 0 results when offset exceeds total count", 0, count);
+ }
+ }
+
+ @Test
+ public void testShowDevicesWithOffsetOnly() throws SQLException {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+
+ statement.execute("CREATE DATABASE root.db1");
+ statement.execute("INSERT INTO root.db1.d0(timestamp, s1) VALUES (1, 1)");
+ statement.execute("INSERT INTO root.db1.d0(timestamp, s2) VALUES (1, 2)");
+ statement.execute("INSERT INTO root.db1.d1(timestamp, s3) VALUES (1, 3)");
+ statement.execute("flush");
+
+ List<String> sg1AllDevices = new ArrayList<>();
+ try (ResultSet resultSet = statement.executeQuery("show devices root.db1.**")) {
+ while (resultSet.next()) {
+ sg1AllDevices.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(2, sg1AllDevices.size());
+
+ int offset1 = 1;
+ List<String> sg1ResultDevices = new ArrayList<>();
+ try (ResultSet resultSet =
+ statement.executeQuery("show devices root.db1.** offset " + offset1)) {
+ while (resultSet.next()) {
+ sg1ResultDevices.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(sg1AllDevices.size() - offset1, sg1ResultDevices.size());
+ for (int i = 0; i < sg1ResultDevices.size(); i++) {
+ Assert.assertEquals(sg1AllDevices.get(offset1 + i), sg1ResultDevices.get(i));
+ }
+
+ List<String> d0Devices = new ArrayList<>();
+ try (ResultSet resultSet = statement.executeQuery("show devices root.db1.d0.**")) {
+ while (resultSet.next()) {
+ d0Devices.add(resultSet.getString(1));
+ }
+ }
+
+ int offset2 = 0;
+ List<String> d0ResultDevices = new ArrayList<>();
+ try (ResultSet resultSet =
+ statement.executeQuery("show devices root.db1.d0.** offset " + offset2)) {
+ while (resultSet.next()) {
+ d0ResultDevices.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(d0Devices.size() - offset2, d0ResultDevices.size());
+ for (int i = 0; i < d0ResultDevices.size(); i++) {
+ Assert.assertEquals(d0Devices.get(offset2 + i), d0ResultDevices.get(i));
+ }
+
+ statement.execute("INSERT INTO root.db1.d2(timestamp, s4) VALUES (10, 10)");
+ statement.execute("INSERT INTO root.db1.d3(timestamp, s5) VALUES (20, 20)");
+ statement.execute("INSERT INTO root.db1.d4(timestamp, s6) VALUES (30, 30)");
+ statement.execute("flush");
+
+ List<String> timeFilteredDevices = new ArrayList<>();
+ try (ResultSet resultSet =
+ statement.executeQuery("show devices root.db1.** where time > 5")) {
+ while (resultSet.next()) {
+ timeFilteredDevices.add(resultSet.getString(1));
+ }
+ }
+
+ int offset3 = 1;
+ List<String> timeFilteredResultDevices = new ArrayList<>();
+ try (ResultSet resultSet =
+ statement.executeQuery("show devices root.db1.** where time > 5 offset " + offset3)) {
+ while (resultSet.next()) {
+ timeFilteredResultDevices.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(timeFilteredDevices.size() - offset3, timeFilteredResultDevices.size());
+ for (int i = 0; i < timeFilteredResultDevices.size(); i++) {
+ Assert.assertEquals(timeFilteredDevices.get(offset3 + i), timeFilteredResultDevices.get(i));
+ }
+
+ statement.execute("CREATE DATABASE root.db2");
+ statement.execute("INSERT INTO root.db2.d0(timestamp, s1) VALUES (4, 7)");
+ statement.execute("INSERT INTO root.db2.d1(timestamp, s2) VALUES (4, 8)");
+ statement.execute("INSERT INTO root.db2.d2(timestamp, s1) VALUES (5, 9)");
+ statement.execute("flush");
+
+ List<String> allDevices = new ArrayList<>();
+ try (ResultSet resultSet = statement.executeQuery("show devices root.db*.**")) {
+ while (resultSet.next()) {
+ allDevices.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(8, allDevices.size());
+
+ int offset4 = 3;
+ List<String> resultDevices = new ArrayList<>();
+ try (ResultSet resultSet =
+ statement.executeQuery("show devices root.db*.** offset " + offset4)) {
+ while (resultSet.next()) {
+ resultDevices.add(resultSet.getString(1));
+ }
+ }
+ Assert.assertEquals(allDevices.size() - offset4, resultDevices.size());
+ for (int i = 0; i < resultDevices.size(); i++) {
+ Assert.assertEquals(allDevices.get(offset4 + i), resultDevices.get(i));
+ }
+
+ int largeOffset = allDevices.size() + 10;
+ int count = 0;
+ try (ResultSet resultSet = statement.executeQuery("show devices offset " + largeOffset)) {
+ while (resultSet.next()) {
+ count++;
+ }
+ }
+ Assert.assertEquals("Should return 0 results when offset exceeds total count", 0, count);
+ }
+ }
+
@Test
public void testShowDevicesWithLimitOffset() throws SQLException {
try (Connection connection = EnvFactory.getEnv().getConnection();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
index c03be7e1bcc..ea7be0c2c33 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java
@@ -579,7 +579,10 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
limit = 0;
offset = 0;
} else if (!canPushDownOffsetLimit) {
- limit = showTimeSeriesStatement.getLimit() + showTimeSeriesStatement.getOffset();
+ limit =
+ showTimeSeriesStatement.getLimit() != 0
+ ? showTimeSeriesStatement.getLimit() + showTimeSeriesStatement.getOffset()
+ : 0;
offset = 0;
}
planBuilder =
@@ -637,7 +640,10 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
long limit = showDevicesStatement.getLimit();
long offset = showDevicesStatement.getOffset();
if (!canPushDownOffsetLimit) {
- limit = showDevicesStatement.getLimit() + showDevicesStatement.getOffset();
+ limit =
+ showDevicesStatement.getLimit() != 0
+ ? showDevicesStatement.getLimit() + showDevicesStatement.getOffset()
+ : 0;
offset = 0;
}
@@ -1009,7 +1015,10 @@ public class LogicalPlanVisitor extends StatementVisitor<PlanNode, MPPQueryConte
long limit = showLogicalViewStatement.getLimit();
long offset = showLogicalViewStatement.getOffset();
if (!canPushDownOffsetLimit) {
- limit = showLogicalViewStatement.getLimit() + showLogicalViewStatement.getOffset();
+ limit =
+ showLogicalViewStatement.getLimit() != 0
+ ? showLogicalViewStatement.getLimit() + showLogicalViewStatement.getOffset()
+ : 0;
offset = 0;
}
planBuilder =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java
index 3d9d6abc54e..f35d047753f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/read/resp/reader/impl/SchemaReaderLimitOffsetWrapper.java
@@ -101,7 +101,7 @@ public class SchemaReaderLimitOffsetWrapper<T extends ISchemaInfo> implements IS
public boolean hasNext() {
try {
isBlocked().get();
- return schemaReader.hasNext() && count < limit;
+ return schemaReader.hasNext() && (limit == 0 || count < limit);
} catch (Exception e) {
throw new RuntimeException(e);
}
(iotdb) 11/16: Fix: Add Mods data scanning check in scan parser for non-aligned data (#17162)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 26a93d8e08a1cb666e4c6752d7ba5f008e615ad7
Author: Zhenyu Luo <zh...@apache.org>
AuthorDate: Thu Feb 5 09:34:41 2026 +0800
Fix: Add Mods data scanning check in scan parser for non-aligned data (#17162)
* Fix: Add Mods data scanning check in scan parser for non-aligned data
This commit fixes the issue where the scan parser was not checking
Mods (modifications) data when processing non-aligned time series data.
Previously, only aligned data (VECTOR type) was checked for deletions
via Mods, but non-aligned single measurement data was missing this check.
The fix adds a Mods deletion check in the putValueToColumns method
for non-aligned data paths, ensuring that deleted data points are
properly filtered out during scan operations. This maintains consistency
with the aligned data handling and prevents deleted data from being
processed.
* add IT
---
.../IoTDBPipeTsFileDecompositionWithModsIT.java | 316 +++++++++++++++++++++
.../scan/TsFileInsertionEventScanParser.java | 5 +
2 files changed, 321 insertions(+)
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeTsFileDecompositionWithModsIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeTsFileDecompositionWithModsIT.java
index 89f1c3fb440..87571e1eb71 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeTsFileDecompositionWithModsIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeTsFileDecompositionWithModsIT.java
@@ -657,4 +657,320 @@ public class IoTDBPipeTsFileDecompositionWithModsIT extends AbstractPipeDualTree
"COUNT(root.sg1.d1.s5),",
Collections.singleton("3591,"));
}
+
+ /**
+ * Test IoTDB pipe handling TsFile decomposition with Mods (modification operations) in tree model
+ * - Large scale single point deletion scenario (non-aligned timeseries)
+ *
+ * <p>Test scenario: 1. Create database root.sg1 with 1 device: d1 (non-aligned timeseries with 5
+ * sensors) 2. Insert 20000 data points for each sensor with different time ranges: - s1: time
+ * 1-20000 - s2: time 10001-30000 - s3: time 20001-40000 - s4: time 30001-50000 - s5: time
+ * 40001-60000 3. Execute FLUSH operation to persist data to TsFile 4. Execute 2000 single point
+ * DELETE operations, each deleting one time point from different sensors 5. Execute FLUSH
+ * operation again 6. Create pipe with mods enabled 7. Verify correctness of receiver data: - Each
+ * sensor should have 19800 remaining data points - Deleted points should not appear in receiver
+ *
+ * <p>Test purpose: Verify that IoTDB pipe can correctly handle large scale single point deletion
+ * operations in TsFile under tree model with non-aligned timeseries, ensuring the binary search
+ * optimization in ModsOperationUtil works correctly with many modification entries.
+ */
+ @Test
+ public void testTsFileDecompositionWithModsLargeScaleSinglePointDeletionNonAligned()
+ throws Exception {
+ TestUtils.executeNonQueryWithRetry(senderEnv, "CREATE DATABASE root.sg1");
+ TestUtils.executeNonQueryWithRetry(receiverEnv, "CREATE DATABASE root.sg1");
+
+ // Insert 20000 data points for s1 (time 1-20000)
+ String s1 = "INSERT INTO root.sg1.d1(time, s1) VALUES ";
+ StringBuilder insertBuilder1 = new StringBuilder(s1);
+ for (int i = 1; i <= 20000; i++) {
+ insertBuilder1.append("(").append(i).append(",").append(1.0f).append(")");
+ if (i % 50 != 0) {
+ insertBuilder1.append(",");
+ } else {
+ TestUtils.executeNonQueryWithRetry(senderEnv, insertBuilder1.toString());
+ insertBuilder1 = new StringBuilder(s1);
+ }
+ }
+ // Execute remaining data if any
+ if (insertBuilder1.length() > s1.length()) {
+ TestUtils.executeNonQueryWithRetry(senderEnv, insertBuilder1.toString());
+ }
+
+ // Insert 20000 data points for s2 (time 10001-30000)
+ String s2 = "INSERT INTO root.sg1.d1(time, s2) VALUES ";
+ StringBuilder insertBuilder2 = new StringBuilder(s2);
+ for (int i = 10001; i <= 30000; i++) {
+ insertBuilder2.append("(").append(i).append(",").append(2.0f).append(")");
+ if (i % 50 != 0) {
+ insertBuilder2.append(",");
+ } else {
+ TestUtils.executeNonQueryWithRetry(senderEnv, insertBuilder2.toString());
+ insertBuilder2 = new StringBuilder(s2);
+ }
+ }
+ // Execute remaining data if any
+ if (insertBuilder2.length() > s2.length()) {
+ TestUtils.executeNonQueryWithRetry(senderEnv, insertBuilder2.toString());
+ }
+
+ // Insert 20000 data points for s3 (time 20001-40000)
+ String s3 = "INSERT INTO root.sg1.d1(time, s3) VALUES ";
+ StringBuilder insertBuilder3 = new StringBuilder(s3);
+ for (int i = 20001; i <= 40000; i++) {
+ insertBuilder3.append("(").append(i).append(",").append(3.0f).append(")");
+ if (i % 50 != 0) {
+ insertBuilder3.append(",");
+ } else {
+ TestUtils.executeNonQueryWithRetry(senderEnv, insertBuilder3.toString());
+ insertBuilder3 = new StringBuilder(s3);
+ }
+ }
+ // Execute remaining data if any
+ if (insertBuilder3.length() > s3.length()) {
+ TestUtils.executeNonQueryWithRetry(senderEnv, insertBuilder3.toString());
+ }
+
+ // Insert 20000 data points for s4 (time 30001-50000)
+ String s4 = "INSERT INTO root.sg1.d1(time, s4) VALUES ";
+ StringBuilder insertBuilder4 = new StringBuilder(s4);
+ for (int i = 30001; i <= 50000; i++) {
+ insertBuilder4.append("(").append(i).append(",").append(4.0f).append(")");
+ if (i % 50 != 0) {
+ insertBuilder4.append(",");
+ } else {
+ TestUtils.executeNonQueryWithRetry(senderEnv, insertBuilder4.toString());
+ insertBuilder4 = new StringBuilder(s4);
+ }
+ }
+ // Execute remaining data if any
+ if (insertBuilder4.length() > s4.length()) {
+ TestUtils.executeNonQueryWithRetry(senderEnv, insertBuilder4.toString());
+ }
+
+ // Insert 20000 data points for s5 (time 40001-60000)
+ String s5 = "INSERT INTO root.sg1.d1(time, s5) VALUES ";
+ StringBuilder insertBuilder5 = new StringBuilder(s5);
+ for (int i = 40001; i <= 60000; i++) {
+ insertBuilder5.append("(").append(i).append(",").append(5.0f).append(")");
+ if (i % 50 != 0) {
+ insertBuilder5.append(",");
+ } else {
+ TestUtils.executeNonQueryWithRetry(senderEnv, insertBuilder5.toString());
+ insertBuilder5 = new StringBuilder(s5);
+ }
+ }
+ // Execute remaining data if any
+ if (insertBuilder5.length() > s5.length()) {
+ TestUtils.executeNonQueryWithRetry(senderEnv, insertBuilder5.toString());
+ }
+
+ TestUtils.executeNonQueryWithRetry(senderEnv, "FLUSH");
+
+ // Execute 2000 single point DELETE operations
+ // Delete 400 points from each sensor (distributed across different time ranges)
+ for (int i = 0; i < 400; i++) {
+ // Delete from s1: every 10th point starting from 10
+ TestUtils.executeNonQueryWithRetry(
+ senderEnv, "DELETE FROM root.sg1.d1.s1 WHERE time = " + (10 + i * 10));
+
+ // Delete from s2: every 10th point starting from 10010
+ TestUtils.executeNonQueryWithRetry(
+ senderEnv, "DELETE FROM root.sg1.d1.s2 WHERE time = " + (10010 + i * 10));
+
+ // Delete from s3: every 10th point starting from 20010
+ TestUtils.executeNonQueryWithRetry(
+ senderEnv, "DELETE FROM root.sg1.d1.s3 WHERE time = " + (20010 + i * 10));
+
+ // Delete from s4: every 10th point starting from 30010
+ TestUtils.executeNonQueryWithRetry(
+ senderEnv, "DELETE FROM root.sg1.d1.s4 WHERE time = " + (30010 + i * 10));
+
+ // Delete from s5: every 10th point starting from 40010
+ TestUtils.executeNonQueryWithRetry(
+ senderEnv, "DELETE FROM root.sg1.d1.s5 WHERE time = " + (40010 + i * 10));
+ }
+
+ TestUtils.executeNonQueryWithRetry(senderEnv, "FLUSH");
+
+ // Verify sender data integrity before creating pipe to avoid leader election issues
+ // This ensures all data is properly persisted and consistent on sender side
+ // before starting the pipe synchronization process
+ TestUtils.assertDataEventuallyOnEnv(
+ senderEnv,
+ "SELECT COUNT(**) FROM root.sg1.d1",
+ "COUNT(root.sg1.d1.s3),COUNT(root.sg1.d1.s4),COUNT(root.sg1.d1.s5),COUNT(root.sg1.d1.s1),COUNT(root.sg1.d1.s2),",
+ Collections.singleton("19600,19600,19600,19600,19600,"));
+
+ executeNonQueryWithRetry(
+ senderEnv,
+ String.format(
+ "CREATE PIPE test_pipe WITH SOURCE ('mods.enable'='true') WITH CONNECTOR('ip'='%s', 'port'='%s', 'format'='tablet')",
+ receiverEnv.getDataNodeWrapperList().get(0).getIp(),
+ receiverEnv.getDataNodeWrapperList().get(0).getPort()));
+
+ // Verify total count of all sensors using COUNT(*)
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(**) FROM root.sg1.d1",
+ "COUNT(root.sg1.d1.s3),COUNT(root.sg1.d1.s4),COUNT(root.sg1.d1.s5),COUNT(root.sg1.d1.s1),COUNT(root.sg1.d1.s2),",
+ Collections.singleton("19600,19600,19600,19600,19600,"));
+
+ // Verify individual sensor counts
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s1) FROM root.sg1.d1",
+ "COUNT(root.sg1.d1.s1),",
+ Collections.singleton("19600,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s2) FROM root.sg1.d1",
+ "COUNT(root.sg1.d1.s2),",
+ Collections.singleton("19600,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s3) FROM root.sg1.d1",
+ "COUNT(root.sg1.d1.s3),",
+ Collections.singleton("19600,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s4) FROM root.sg1.d1",
+ "COUNT(root.sg1.d1.s4),",
+ Collections.singleton("19600,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s5) FROM root.sg1.d1",
+ "COUNT(root.sg1.d1.s5),",
+ Collections.singleton("19600,"));
+
+ // Verify count of deleted time ranges using COUNT with WHERE clause
+ // These should return 0 since all points in these ranges were deleted
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s1) FROM root.sg1.d1 WHERE time >= 10 AND time <= 4000 AND time % 10 = 0",
+ "COUNT(root.sg1.d1.s1),",
+ Collections.singleton("0,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s2) FROM root.sg1.d1 WHERE time >= 10010 AND time <= 14000 AND time % 10 = 0",
+ "COUNT(root.sg1.d1.s2),",
+ Collections.singleton("0,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s3) FROM root.sg1.d1 WHERE time >= 20010 AND time <= 24000 AND time % 10 = 0",
+ "COUNT(root.sg1.d1.s3),",
+ Collections.singleton("0,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s4) FROM root.sg1.d1 WHERE time >= 30010 AND time <= 34000 AND time % 10 = 0",
+ "COUNT(root.sg1.d1.s4),",
+ Collections.singleton("0,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s5) FROM root.sg1.d1 WHERE time >= 40010 AND time <= 44000 AND time % 10 = 0",
+ "COUNT(root.sg1.d1.s5),",
+ Collections.singleton("0,"));
+
+ // Verify count of non-deleted time ranges using multiple range queries
+ // Check ranges before deletion area
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s1) FROM root.sg1.d1 WHERE time >= 1 AND time < 10",
+ "COUNT(root.sg1.d1.s1),",
+ Collections.singleton("9,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s2) FROM root.sg1.d1 WHERE time >= 10001 AND time < 10010",
+ "COUNT(root.sg1.d1.s2),",
+ Collections.singleton("9,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s3) FROM root.sg1.d1 WHERE time >= 20001 AND time < 20010",
+ "COUNT(root.sg1.d1.s3),",
+ Collections.singleton("9,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s4) FROM root.sg1.d1 WHERE time >= 30001 AND time < 30010",
+ "COUNT(root.sg1.d1.s4),",
+ Collections.singleton("9,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s5) FROM root.sg1.d1 WHERE time >= 40001 AND time < 40010",
+ "COUNT(root.sg1.d1.s5),",
+ Collections.singleton("9,"));
+
+ // Check ranges after deletion area
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s1) FROM root.sg1.d1 WHERE time > 4000 AND time <= 20000",
+ "COUNT(root.sg1.d1.s1),",
+ Collections.singleton("16000,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s2) FROM root.sg1.d1 WHERE time > 14000 AND time <= 30000",
+ "COUNT(root.sg1.d1.s2),",
+ Collections.singleton("16000,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s3) FROM root.sg1.d1 WHERE time > 24000 AND time <= 40000",
+ "COUNT(root.sg1.d1.s3),",
+ Collections.singleton("16000,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s4) FROM root.sg1.d1 WHERE time > 34000 AND time <= 50000",
+ "COUNT(root.sg1.d1.s4),",
+ Collections.singleton("16000,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s5) FROM root.sg1.d1 WHERE time > 44000 AND time <= 60000",
+ "COUNT(root.sg1.d1.s5),",
+ Collections.singleton("16000,"));
+
+ // Check non-deleted points within deletion range (every 10th point except the ones we deleted)
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s1) FROM root.sg1.d1 WHERE time >= 10 AND time <= 4000 AND time % 10 != 0",
+ "COUNT(root.sg1.d1.s1),",
+ Collections.singleton("3591,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s2) FROM root.sg1.d1 WHERE time >= 10010 AND time <= 14000 AND time % 10 != 0",
+ "COUNT(root.sg1.d1.s2),",
+ Collections.singleton("3591,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s3) FROM root.sg1.d1 WHERE time >= 20010 AND time <= 24000 AND time % 10 != 0",
+ "COUNT(root.sg1.d1.s3),",
+ Collections.singleton("3591,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s4) FROM root.sg1.d1 WHERE time >= 30010 AND time <= 34000 AND time % 10 != 0",
+ "COUNT(root.sg1.d1.s4),",
+ Collections.singleton("3591,"));
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT COUNT(s5) FROM root.sg1.d1 WHERE time >= 40010 AND time <= 44000 AND time % 10 != 0",
+ "COUNT(root.sg1.d1.s5),",
+ Collections.singleton("3591,"));
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
index 5c9985fe748..99d35f7eb98 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
@@ -426,6 +426,11 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
}
}
} else {
+ if (!modsInfos.isEmpty()
+ && ModsOperationUtil.isDelete(data.currentTime(), modsInfos.get(0))) {
+ return false;
+ }
+
isNeedFillTime = true;
switch (tablet.getSchemas().get(0).getType()) {
case BOOLEAN:
(iotdb) 09/16: Pipe: allow exclusion-only pattern lists (#17154)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f9807d9290d184d2606ec6591454da2ae7f54107
Author: VGalaxies <dy...@gmail.com>
AuthorDate: Wed Feb 4 12:16:12 2026 +0800
Pipe: allow exclusion-only pattern lists (#17154)
* fix(pipe): allow exclusion-only pattern lists
* improve IT
---
.../auto/basic/IoTDBTreePatternFormatIT.java | 56 ++++++++++++++++++++++
.../pipe/datastructure/pattern/TreePattern.java | 12 +++--
2 files changed, 65 insertions(+), 3 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBTreePatternFormatIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBTreePatternFormatIT.java
index 9079daeab71..00e3078a355 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBTreePatternFormatIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBTreePatternFormatIT.java
@@ -385,4 +385,60 @@ public class IoTDBTreePatternFormatIT extends AbstractPipeDualTreeModelAutoIT {
"Time,root.db.d1.t,root.db.d2.s,",
expectedResSet);
}
+
+ @Test
+ public void testIoTDBPatternWithExclusionOnlyHistoricalData() throws Exception {
+ final Map<String, String> sourceAttributes = new HashMap<>();
+ sourceAttributes.put("source.pattern.exclusion", "root.db.d1.s*, root.db.d3.*");
+ sourceAttributes.put("source.inclusion", "data.insert");
+ sourceAttributes.put("user", "root");
+
+ final List<String> insertQueries =
+ Arrays.asList(
+ "insert into root.db.d1(time, s, s1, t) values (1, 1, 1, 1)",
+ "insert into root.db.d2(time, s) values (2, 2)",
+ "insert into root.db.d3(time, s) values (3, 3)",
+ "insert into root.db1.d1(time, s) values (4, 4)");
+
+ final Set<String> expectedResSet = new HashSet<>();
+ expectedResSet.add("1,null,1.0,null,");
+ expectedResSet.add("2,null,null,2.0,");
+ expectedResSet.add("4,4.0,null,null,");
+
+ testPipeWithMultiplePatterns(
+ sourceAttributes,
+ insertQueries,
+ true, // isHistorical = true
+ "select * from root.db1.**,root.db.**",
+ "Time,root.db1.d1.s,root.db.d1.t,root.db.d2.s,",
+ expectedResSet);
+ }
+
+ @Test
+ public void testIoTDBPatternWithExclusionOnlyRealtimeData() throws Exception {
+ final Map<String, String> sourceAttributes = new HashMap<>();
+ sourceAttributes.put("source.pattern.exclusion", "root.db.d1.s*, root.db.d3.*");
+ sourceAttributes.put("source.inclusion", "data.insert");
+ sourceAttributes.put("user", "root");
+
+ final List<String> insertQueries =
+ Arrays.asList(
+ "insert into root.db.d1(time, s, s1, t) values (1, 1, 1, 1)",
+ "insert into root.db.d2(time, s) values (2, 2)",
+ "insert into root.db.d3(time, s) values (3, 3)",
+ "insert into root.db1.d1(time, s) values (4, 4)");
+
+ final Set<String> expectedResSet = new HashSet<>();
+ expectedResSet.add("1,null,1.0,null,");
+ expectedResSet.add("2,null,null,2.0,");
+ expectedResSet.add("4,4.0,null,null,");
+
+ testPipeWithMultiplePatterns(
+ sourceAttributes,
+ insertQueries,
+ false, // isHistorical = false
+ "select * from root.db1.**,root.db.**",
+ "Time,root.db1.d1.s,root.db.d1.t,root.db.d2.s,",
+ expectedResSet);
+ }
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java
index 0cea3b34e48..5d6c808a0bb 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java
@@ -156,10 +156,16 @@ public abstract class TreePattern {
final boolean hasPatternInclusionKey =
sourceParameters.hasAnyAttributes(
EXTRACTOR_PATTERN_INCLUSION_KEY, SOURCE_PATTERN_INCLUSION_KEY);
+ final boolean hasPatternExclusionKey =
+ sourceParameters.hasAnyAttributes(
+ EXTRACTOR_PATTERN_EXCLUSION_KEY, SOURCE_PATTERN_EXCLUSION_KEY);
final boolean hasLegacyPathKey =
sourceParameters.hasAnyAttributes(EXTRACTOR_PATH_KEY, SOURCE_PATH_KEY);
final boolean hasLegacyPatternKey =
sourceParameters.hasAnyAttributes(EXTRACTOR_PATTERN_KEY, SOURCE_PATTERN_KEY);
+ final boolean usePatternSyntax =
+ hasPatternInclusionKey
+ || (hasPatternExclusionKey && !hasLegacyPathKey && !hasLegacyPatternKey);
if (hasPatternInclusionKey && (hasLegacyPathKey || hasLegacyPatternKey)) {
final String msg =
@@ -172,7 +178,7 @@ public abstract class TreePattern {
// 1. Parse INCLUSION patterns into a list
List<TreePattern> inclusionPatterns =
- hasPatternInclusionKey
+ usePatternSyntax
? parseIoTDBPatternList(
sourceParameters.getStringByKeys(
EXTRACTOR_PATTERN_INCLUSION_KEY, SOURCE_PATTERN_INCLUSION_KEY),
@@ -198,7 +204,7 @@ public abstract class TreePattern {
}
// 2. Parse EXCLUSION patterns into a list
- if (hasPatternInclusionKey
+ if (usePatternSyntax
&& sourceParameters.hasAnyAttributes(
EXTRACTOR_PATH_EXCLUSION_KEY, SOURCE_PATH_EXCLUSION_KEY)) {
final String msg =
@@ -210,7 +216,7 @@ public abstract class TreePattern {
}
List<TreePattern> exclusionPatterns =
- hasPatternInclusionKey
+ usePatternSyntax
? parseIoTDBPatternList(
sourceParameters.getStringByKeys(
EXTRACTOR_PATTERN_EXCLUSION_KEY, SOURCE_PATTERN_EXCLUSION_KEY),
(iotdb) 02/16: Fix value and time chunk misalignment when rewriting with time deletion (#17142)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 8b954fb9ed747cfa9feccf03fe0c35a2c368a61f
Author: shuwenwei <55...@users.noreply.github.com>
AuthorDate: Tue Feb 3 11:35:16 2026 +0800
Fix value and time chunk misalignment when rewriting with time deletion (#17142)
---
.../compaction/CompactionDataTypeAlterTest.java | 55 ++++++++++++++++++++++
pom.xml | 2 +-
2 files changed, 56 insertions(+), 1 deletion(-)
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeAlterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeAlterTest.java
index 0d08ca7832c..23d509f3dfe 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeAlterTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionDataTypeAlterTest.java
@@ -19,12 +19,16 @@
package org.apache.iotdb.db.storageengine.dataregion.compaction;
+import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadChunkCompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadPointCompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask;
+import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
+import org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils;
@@ -319,4 +323,55 @@ public class CompactionDataTypeAlterTest extends AbstractCompactionTest {
resource2.serialize();
seqResources.add(resource2);
}
+
+ @Test
+ public void testAlterDataTypeWithAlignedSeriesWithTimeDeletion()
+ throws IOException, WriteProcessException, IllegalPathException {
+ List<IMeasurementSchema> measurementSchemas1 = new ArrayList<>();
+ measurementSchemas1.add(new MeasurementSchema("s1", TSDataType.INT32));
+ measurementSchemas1.add(new MeasurementSchema("s2", TSDataType.INT32));
+
+ TsFileResource resource1 = createEmptyFileAndResource(true);
+ resource1.setStatusForTest(TsFileResourceStatus.COMPACTING);
+ try (TsFileWriter writer = new TsFileWriter(resource1.getTsFile())) {
+ writer.registerAlignedTimeseries(new Path(device), measurementSchemas1);
+ for (int i = 0; i < 100; i++) {
+ TSRecord record = new TSRecord(device, i);
+ record.addTuple(new IntDataPoint("s1", 0));
+ record.addTuple(new IntDataPoint("s2", 1));
+ writer.writeRecord(record);
+ }
+ writer.flush();
+ }
+ resource1.updateStartTime(device, 1);
+ resource1.updateEndTime(device, 1);
+ resource1.serialize();
+ ModificationFile modFile = resource1.getExclusiveModFile();
+ modFile.write(new TreeDeletionEntry(new MeasurementPath(device + ".*"), 0, 40));
+ modFile.close();
+ seqResources.add(resource1);
+
+ List<IMeasurementSchema> measurementSchemas2 = new ArrayList<>();
+ measurementSchemas2.add(new MeasurementSchema("s1", TSDataType.DOUBLE));
+ measurementSchemas2.add(new MeasurementSchema("s2", TSDataType.DOUBLE));
+ TsFileResource resource2 = createEmptyFileAndResource(true);
+ resource2.setStatusForTest(TsFileResourceStatus.COMPACTING);
+ try (TsFileWriter writer = new TsFileWriter(resource2.getTsFile())) {
+ writer.registerAlignedTimeseries(new Path(device), measurementSchemas2);
+ TSRecord record = new TSRecord(device, 200);
+ record.addTuple(new DoubleDataPoint("s1", 2.0));
+ record.addTuple(new DoubleDataPoint("s2", 3.0));
+ writer.writeRecord(record);
+ writer.flush();
+ }
+ resource2.updateStartTime(device, 2);
+ resource2.updateEndTime(device, 2);
+ resource2.serialize();
+ seqResources.add(resource2);
+
+ InnerSpaceCompactionTask task =
+ new InnerSpaceCompactionTask(
+ 0, tsFileManager, seqResources, true, new ReadChunkCompactionPerformer(), 0);
+ Assert.assertTrue(task.start());
+ }
}
diff --git a/pom.xml b/pom.xml
index 3b2133c0334..240202f34c1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -173,7 +173,7 @@
<thrift.version>0.14.1</thrift.version>
<xz.version>1.9</xz.version>
<zstd-jni.version>1.5.6-3</zstd-jni.version>
- <tsfile.version>2.2.1-260130-SNAPSHOT</tsfile.version>
+ <tsfile.version>2.2.1-260203-SNAPSHOT</tsfile.version>
</properties>
<!--
if we claim dependencies in dependencyManagement, then we do not claim
(iotdb) 01/16: Fix blob accumulator query can't skip read blobStatistics caused by isModified don't change to true in the metadata of chunk and pageReader (#17127)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e30c8ba3176320372312e4032adb417fa4aa68af
Author: libo <ze...@163.com>
AuthorDate: Mon Feb 2 15:44:24 2026 +0800
Fix blob accumulator query can't skip read blobStatistics caused by isModified don't change to true in the metadata of chunk and pageReader (#17127)
* Resolve Blob type
* Fix
* Fixed an issue in IT where aggregation queries skipped over Blob and DATE types, leading to incomplete coverage in aggregation queries. Having the same DataType does not ensure that their statistics are identical, so it has been changed to check whether the statistics are the same. Additionally, Blob type has been added to the types that cannot use statistics for acceleration.
* Rename a variable.
* Rename the function.
---
.../db/it/schema/IoTDBAlterTimeSeriesTypeIT.java | 1399 +++++++++++++++-----
.../it/schema/IoTDBAlterColumnTypeIT.java | 805 +++++++----
.../execution/operator/source/FileLoaderUtils.java | 8 +-
.../execution/operator/source/SeriesScanUtil.java | 8 +-
.../org/apache/iotdb/db/utils/SchemaUtils.java | 56 +-
5 files changed, 1720 insertions(+), 556 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBAlterTimeSeriesTypeIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBAlterTimeSeriesTypeIT.java
index d199dded371..3effc3a8b83 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBAlterTimeSeriesTypeIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBAlterTimeSeriesTypeIT.java
@@ -93,10 +93,8 @@ public class IoTDBAlterTimeSeriesTypeIT {
private static final Logger log = LoggerFactory.getLogger(IoTDBAlterTimeSeriesTypeIT.class);
private static long timeout = -1;
private static final String database = "root.alter";
- public static final List<TSDataType> DATA_TYPE_LIST =
- Arrays.asList(TSDataType.STRING, TSDataType.TEXT, TSDataType.BOOLEAN);
- public static final List<TSDataType> UNSUPPORT_ACCUMULATOR_QUERY_DATA_TYPE_LIST =
- Collections.singletonList(TSDataType.BLOB);
+ public static final List<TSDataType> UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST =
+ Arrays.asList(TSDataType.BLOB, TSDataType.TEXT, TSDataType.BOOLEAN);
@BeforeClass
public static void setUp() throws Exception {
@@ -129,6 +127,8 @@ public class IoTDBAlterTimeSeriesTypeIT {
testAlignDeviceSequenceDataQuery(from, to);
testAlignDeviceUnSequenceDataQuery(from, to);
testAlignDeviceUnSequenceOverlappedDataQuery(from, to);
+
+ doWriteAndAlterWriteForAccumulatorQuery(from, to);
}
}
}
@@ -295,7 +295,8 @@ public class IoTDBAlterTimeSeriesTypeIT {
}
assertFalse(dataSet.hasNext());
- if (DATA_TYPE_LIST.contains(from) || DATA_TYPE_LIST.contains(to)) {
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(from)
+ || UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(to)) {
dataSet =
session.executeQueryStatement(
"select first_value(s1),last_value(s1) from "
@@ -352,6 +353,83 @@ public class IoTDBAlterTimeSeriesTypeIT {
}
}
+ private void doWriteAndAlterWriteForAccumulatorQuery(TSDataType from, TSDataType to)
+ throws IoTDBConnectionException, StatementExecutionException {
+ try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+ session.executeNonQueryStatement(
+ "SET CONFIGURATION \"enable_unseq_space_compaction\"='false'");
+ session.executeNonQueryStatement("SET CONFIGURATION \"enable_seq_space_compaction\"='false'");
+ if (from == TSDataType.DATE && !to.isCompatible(from)) {
+ throw new NotSupportedException("Not supported DATE type.");
+ }
+
+ // create a table with type of "from"
+ session.executeNonQueryStatement(
+ "CREATE TIMESERIES " + database + ".write_and_alter_column_type_write.s1 " + from);
+
+ // write a sequence tsfile point of "from"
+ Tablet tablet =
+ new Tablet(
+ database + ".write_and_alter_column_type_write",
+ Collections.singletonList("s1"),
+ Collections.singletonList(from),
+ Collections.singletonList(ColumnCategory.FIELD));
+ for (int i = 1; i <= 512; i++) {
+ tablet.addTimestamp(0, i);
+ tablet.addValue("s1", 0, genValue(from, i));
+ session.insertTablet(tablet);
+ tablet.reset();
+ }
+ session.executeNonQueryStatement("FLUSH");
+
+ // alter the type to "to"
+ boolean isCompatible = MetadataUtils.canAlter(from, to);
+ if (isCompatible) {
+ session.executeNonQueryStatement(
+ "ALTER TIMESERIES "
+ + database
+ + ".write_and_alter_column_type_write.s1 SET DATA TYPE "
+ + to);
+ } else {
+ try {
+ session.executeNonQueryStatement(
+ "ALTER TIMESERIES "
+ + database
+ + ".write_and_alter_column_type_write.s1 SET DATA TYPE "
+ + to);
+ } catch (StatementExecutionException e) {
+ assertEquals(
+ "701: New type " + to + " is not compatible with the existing one " + from,
+ e.getMessage());
+ }
+ }
+
+ // write a sequence tsfile point of "to"
+ tablet =
+ new Tablet(
+ database + ".write_and_alter_column_type_write",
+ Collections.singletonList("s1"),
+ Collections.singletonList(to),
+ Collections.singletonList(ColumnCategory.FIELD));
+ for (int i = 513; i <= 1024; i++) {
+ tablet.addTimestamp(0, i);
+ tablet.addValue("s1", 0, genValue(to, i));
+ session.insertTablet(tablet);
+ tablet.reset();
+ }
+ session.executeNonQueryStatement("FLUSH");
+
+ standardAccumulatorQueryByTimeOnlyInvolveOldDataTypeTest(session, from, to);
+ standardAccumulatorQueryByTimeOnlyInvolveNewDataTypeTest(session, from, to);
+ standardAccumulatorQueryByTimeInvolveAllDataTypeTest(session, from, to);
+ } finally {
+ try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
+ session.executeNonQueryStatement(
+ "DELETE TIMESERIES " + database + ".write_and_alter_column_type_write.s1");
+ }
+ }
+ }
+
@Test
public void testAlterWithoutWrite() throws IoTDBConnectionException, StatementExecutionException {
Set<TSDataType> typesToTest = new HashSet<>();
@@ -1129,31 +1207,22 @@ public class IoTDBAlterTimeSeriesTypeIT {
tablet.reset();
}
- if (DATA_TYPE_LIST.contains(from) || DATA_TYPE_LIST.contains(to)) {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ String columns;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(from)
+ || UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(to)) {
+ columns = "first_value(s1),last_value(s1)";
} else {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ columns = "min_value(s1),max_value(s1),first_value(s1),last_value(s1)";
+ }
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select " + columns + " from " + database + ".construct_and_alter_column_type");
+ RowRecord rec = dataSet.next();
+ while (rec != null) {
+ System.out.println(rec.getFields().toString());
+ rec = dataSet.next();
}
+ dataSet.close();
try {
standardSelectTest(session, from, to);
@@ -1275,31 +1344,22 @@ public class IoTDBAlterTimeSeriesTypeIT {
tablet.reset();
}
- if (DATA_TYPE_LIST.contains(from) || DATA_TYPE_LIST.contains(to)) {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ String columns;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(from)
+ || UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(to)) {
+ columns = "first_value(s1),last_value(s1)";
} else {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ columns = "min_value(s1),max_value(s1),first_value(s1),last_value(s1)";
}
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select " + columns + " from " + database + ".construct_and_alter_column_type");
+ RowRecord rec = dataSet.next();
+ while (rec != null) {
+ System.out.println(rec.getFields().toString());
+ rec = dataSet.next();
+ }
+ dataSet.close();
try {
standardSelectTest(session, from, to);
@@ -1414,31 +1474,22 @@ public class IoTDBAlterTimeSeriesTypeIT {
}
// session.executeNonQueryStatement("FLUSH");
- if (DATA_TYPE_LIST.contains(from) || DATA_TYPE_LIST.contains(to)) {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ String columns;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(from)
+ || UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(to)) {
+ columns = "first_value(s1),last_value(s1)";
} else {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ columns = "min_value(s1),max_value(s1),first_value(s1),last_value(s1)";
}
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select " + columns + " from " + database + ".construct_and_alter_column_type");
+ RowRecord rec = dataSet.next();
+ while (rec != null) {
+ System.out.println(rec.getFields().toString());
+ rec = dataSet.next();
+ }
+ dataSet.close();
try {
standardSelectTest(session, from, to);
@@ -1556,31 +1607,22 @@ public class IoTDBAlterTimeSeriesTypeIT {
}
// session.executeNonQueryStatement("FLUSH");
- if (DATA_TYPE_LIST.contains(from) || DATA_TYPE_LIST.contains(to)) {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ String columns;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(from)
+ || UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(to)) {
+ columns = "first_value(s1),last_value(s1)";
} else {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ columns = "min_value(s1),max_value(s1),first_value(s1),last_value(s1)";
}
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select " + columns + " from " + database + ".construct_and_alter_column_type");
+ RowRecord rec = dataSet.next();
+ while (rec != null) {
+ System.out.println(rec.getFields().toString());
+ rec = dataSet.next();
+ }
+ dataSet.close();
try {
standardSelectTest(session, from, to);
@@ -1695,31 +1737,22 @@ public class IoTDBAlterTimeSeriesTypeIT {
}
// session.executeNonQueryStatement("FLUSH");
- if (DATA_TYPE_LIST.contains(from) || DATA_TYPE_LIST.contains(to)) {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ String columns;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(from)
+ || UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(to)) {
+ columns = "first_value(s1),last_value(s1)";
} else {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ columns = "min_value(s1),max_value(s1),first_value(s1),last_value(s1)";
+ }
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select " + columns + " from " + database + ".construct_and_alter_column_type");
+ RowRecord rec = dataSet.next();
+ while (rec != null) {
+ System.out.println(rec.getFields().toString());
+ rec = dataSet.next();
}
+ dataSet.close();
try {
standardSelectTest(session, from, to);
@@ -1838,31 +1871,22 @@ public class IoTDBAlterTimeSeriesTypeIT {
}
// session.executeNonQueryStatement("FLUSH");
- if (DATA_TYPE_LIST.contains(from) || DATA_TYPE_LIST.contains(to)) {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ String columns;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(from)
+ || UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(to)) {
+ columns = "first_value(s1),last_value(s1)";
} else {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ columns = "min_value(s1),max_value(s1),first_value(s1),last_value(s1)";
+ }
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select " + columns + " from " + database + ".construct_and_alter_column_type");
+ RowRecord rec = dataSet.next();
+ while (rec != null) {
+ System.out.println(rec.getFields().toString());
+ rec = dataSet.next();
}
+ dataSet.close();
try {
standardSelectTest(session, from, to);
@@ -1984,7 +2008,8 @@ public class IoTDBAlterTimeSeriesTypeIT {
"select s1 from "
+ database
+ ".construct_and_alter_column_type where cast(s1 as TEXT) >= '1' and cast(s2 as TEXT) > '2' order by time");
- // "select s1 from construct_and_alter_column_type where cast(s1 as
+ // "select s1 from " + database + ".construct_and_alter_column_type where
+ // cast(s1 as
// TEXT) >= '1' and s2 > '2' order by time");
} else {
dataSet1 =
@@ -2350,77 +2375,163 @@ public class IoTDBAlterTimeSeriesTypeIT {
private static void standardAccumulatorQueryTest(ISession session, TSDataType newType)
throws StatementExecutionException, IoTDBConnectionException {
- if (newType == TSDataType.DATE) {
- throw new NotSupportedException("Not supported DATE type.");
+ RowRecord rec = null;
+ RowRecord unsupportMinMaxRec = null;
+ SessionDataSet dataSet = null;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ dataSet =
+ session.executeQueryStatement(
+ "select first_value(s1),last_value(s1) from "
+ + database
+ + ".construct_and_alter_column_type");
+ unsupportMinMaxRec = dataSet.next();
+ } else {
+ dataSet =
+ session.executeQueryStatement(
+ "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
+ + database
+ + ".construct_and_alter_column_type");
+ rec = dataSet.next();
}
- SessionDataSet dataSet;
- RowRecord rec;
- if (!UNSUPPORT_ACCUMULATOR_QUERY_DATA_TYPE_LIST.contains(newType)) {
- int[] expectedValue;
- int max = 4;
- if (DATA_TYPE_LIST.contains(newType)) {
- dataSet =
- session.executeQueryStatement(
- "select first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- rec = dataSet.next();
- expectedValue = new int[] {1, 1024};
- if (newType == TSDataType.STRING
- || newType == TSDataType.TEXT
- || newType == TSDataType.BLOB) {
- // expectedValue[1] = 999;
- } else if (newType == TSDataType.BOOLEAN) {
- expectedValue = new int[] {19700102, 19721021};
+ Object[] expectedValue;
+ switch (newType) {
+ case BOOLEAN:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Boolean[] {false, true};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ expectedValue[i], unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Boolean[] {false, true, false, true};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).getObjectValue(newType));
+ }
}
- max = 2;
- } else {
- dataSet =
- session.executeQueryStatement(
- "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- rec = dataSet.next();
- expectedValue = new int[] {1, 1024, 1, 1024};
- if (newType == TSDataType.STRING
- || newType == TSDataType.TEXT
- || newType == TSDataType.BLOB) {
- expectedValue[1] = 999;
- } else if (newType == TSDataType.BOOLEAN) {
- expectedValue = new int[] {19700102, 19721021, 19700102, 19721021};
+ break;
+ case DATE:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new String[] {"1970-01-02", "1971-05-28"};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ String.valueOf(expectedValue[i]), unsupportMinMaxRec.getFields().get(i).toString());
+ }
+ } else {
+ expectedValue = new String[] {"1970-01-02", "1971-05-28", "1970-01-02", "1971-05-28"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
+ }
}
- }
-
- if (newType != TSDataType.BOOLEAN) {
- for (int i = 0; i < max; i++) {
- if (newType == TSDataType.BLOB) {
- assertEquals(genValue(newType, expectedValue[i]), rec.getFields().get(i).getBinaryV());
- } else if (newType == TSDataType.DATE) {
- assertEquals(genValue(newType, expectedValue[i]), rec.getFields().get(i).getDateV());
- } else {
- log.info(
- "i is {}, expected value: {}, actual value: {}",
+ break;
+ case DOUBLE:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {1, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
i,
- genValue(newType, expectedValue[i]).toString(),
- rec.getFields().get(i).toString());
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).toString());
assertEquals(
- genValue(newType, expectedValue[i]).toString(), rec.getFields().get(i).toString());
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
}
}
-
- assertFalse(dataSet.hasNext());
-
- if (newType.isNumeric()) {
- dataSet =
- session.executeQueryStatement(
- "select avg(s1),sum(s1) from " + database + ".construct_and_alter_column_type");
- rec = dataSet.next();
- assertEquals(512.5, rec.getFields().get(0).getDoubleV(), 0.001);
- assertEquals(524800.0, rec.getFields().get(1).getDoubleV(), 0.001);
- assertFalse(dataSet.hasNext());
+ break;
+ case BLOB:
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {1, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
+ i,
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
}
- }
+ break;
+ case FLOAT:
+ case INT64:
+ case TIMESTAMP:
+ case INT32:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {1, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
+ i,
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i,
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ }
+ break;
+ default:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ }
+ break;
+ default:
+ break;
+ }
+
+ // test avg and sum
+ if (newType.isNumeric()) {
+ dataSet =
+ session.executeQueryStatement(
+ "select avg(s1),sum(s1) from " + database + ".construct_and_alter_column_type");
+ rec = dataSet.next();
+ assertEquals(512.5, rec.getFields().get(0).getDoubleV(), 0.001);
+ assertEquals(524800.0, rec.getFields().get(1).getDoubleV(), 0.001);
+ assertFalse(dataSet.hasNext());
}
// can use statistics information
@@ -2445,152 +2556,805 @@ public class IoTDBAlterTimeSeriesTypeIT {
private static void standardAccumulatorQueryTest(
ISession session, TSDataType from, TSDataType newType)
throws StatementExecutionException, IoTDBConnectionException {
- if (from == TSDataType.DATE) {
- throw new NotSupportedException("Not supported DATE type.");
+ if (!newType.isCompatible(from)) {
+ log.info(newType + " is Incompatible with " + from + ".");
}
- if (from == TSDataType.BLOB || newType == TSDataType.BLOB) {
- throw new NotSupportedException("Not supported BLOB type.");
- }
- // if (from == TSDataType.BOOLEAN
- // && (newType == TSDataType.STRING || newType == TSDataType.TEXT)) {
- if (from == TSDataType.BOOLEAN && DATA_TYPE_LIST.contains(newType)) {
- SessionDataSet dataSet =
+ RowRecord rec = null;
+ RowRecord unsupportMinMaxRec = null;
+ SessionDataSet dataSet = null;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ dataSet =
session.executeQueryStatement(
"select first_value(s1),last_value(s1) from "
+ database
+ ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- boolean[] expectedValue = {false, true};
- for (int i = 0; i < 2; i++) {
- if (newType == TSDataType.STRING || newType == TSDataType.TEXT) {
- assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
- }
- }
+ unsupportMinMaxRec = dataSet.next();
+ Assert.assertNotNull(unsupportMinMaxRec);
} else {
- SessionDataSet dataSet;
- int[] expectedValue;
- int max = 4;
- if (DATA_TYPE_LIST.contains(newType)) {
- dataSet =
- session.executeQueryStatement(
- "select first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- expectedValue = new int[] {1, 1024};
- max = 2;
- } else {
- dataSet =
- session.executeQueryStatement(
- "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- expectedValue = new int[] {1, 1024, 1, 1024};
- if (newType == TSDataType.BOOLEAN) {
- expectedValue = new int[] {19700102, 19721021, 19700102, 19721021};
+ dataSet =
+ session.executeQueryStatement(
+ "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
+ + database
+ + ".construct_and_alter_column_type");
+ rec = dataSet.next();
+ Assert.assertNotNull(rec);
+ }
+
+ Object[] expectedValue;
+ switch (from) {
+ case BOOLEAN:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Boolean[] {false, true};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Boolean[] {false, true, false, true};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
}
- }
- RowRecord rec = dataSet.next();
- if (newType != TSDataType.BOOLEAN) {
- for (int i = 0; i < max; i++) {
- if (newType == TSDataType.BLOB) {
- assertEquals(genValue(newType, expectedValue[i]), rec.getFields().get(i).getBinaryV());
- } else if (newType == TSDataType.DATE) {
- assertEquals(genValue(newType, expectedValue[i]), rec.getFields().get(i).getDateV());
- } else if (newType == TSDataType.STRING || newType == TSDataType.TEXT) {
- if (from == TSDataType.DATE) {
- log.info(
- "i is {}, expected value: {}, actual value: {}",
- i,
- new Binary(genValue(from, expectedValue[i]).toString(), StandardCharsets.UTF_8),
- rec.getFields().get(i).getBinaryV());
+ break;
+ case DATE:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new String[] {"1970-01-02", "1971-05-28"};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ String.valueOf(expectedValue[i]), unsupportMinMaxRec.getFields().get(i).toString());
+ }
+ } else {
+ expectedValue = new String[] {"1970-01-02", "1971-05-28", "1970-01-02", "1971-05-28"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
+ }
+ }
+ break;
+ case DOUBLE:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {1, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
+ i,
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i,
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ }
+ break;
+ case BLOB:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new String[] {"1", "1024"};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ }
+ } else {
+ expectedValue = new String[] {"1", "999", "1", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).toString());
+ }
+ }
+ break;
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {1, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ new Binary(expectedValue[i].toString(), StandardCharsets.UTF_8),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ new Binary(expectedValue[i].toString(), StandardCharsets.UTF_8),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ }
+ break;
+ case FLOAT:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new String[] {"1.0", "1024.0"};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
+ i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ }
+ } else {
+ expectedValue = new String[] {"1.0", "999.0", "1.0", "1024.0"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).toString());
+ }
+ }
+ break;
+ default:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
assertEquals(
- new Binary(genValue(from, expectedValue[i]).toString(), StandardCharsets.UTF_8),
- rec.getFields().get(i).getBinaryV());
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ }
+ break;
+ case INT64:
+ case TIMESTAMP:
+ case INT32:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {1, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
+ i,
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
} else {
- log.info(
- "i is {}, expected value: {}, actual value: {}",
- i,
- newType.castFromSingleValue(from, genValue(from, expectedValue[i])),
- rec.getFields().get(i).getBinaryV());
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i,
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ }
+ break;
+ default:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
assertEquals(
- newType.castFromSingleValue(from, genValue(from, expectedValue[i])),
- rec.getFields().get(i).getBinaryV());
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
}
- } else {
- log.info(
- "i is {}, expected value: {}, actual value: {}",
+ break;
+ }
+ break;
+ default:
+ break;
+ }
+
+ // test avg and sum
+ if (newType.isNumeric()) {
+ dataSet =
+ session.executeQueryStatement(
+ "select avg(s1),sum(s1) from " + database + ".construct_and_alter_column_type");
+ rec = dataSet.next();
+ assertEquals(512.5, rec.getFields().get(0).getDoubleV(), 0.001);
+ assertEquals(524800.0, rec.getFields().get(1).getDoubleV(), 0.001);
+ assertFalse(dataSet.hasNext());
+ }
+
+ // can get statistics information
+ dataSet =
+ session.executeQueryStatement(
+ "select count(*) from " + database + ".construct_and_alter_column_type where time > 0");
+ rec = dataSet.next();
+ assertEquals(1024, rec.getFields().get(0).getLongV());
+ assertFalse(dataSet.hasNext());
+
+ // can't use statistics information
+ dataSet =
+ session.executeQueryStatement(
+ "select count(*) from "
+ + database
+ + ".construct_and_alter_column_type where time > 10000");
+ rec = dataSet.next();
+ assertEquals(0, rec.getFields().get(0).getLongV());
+ assertFalse(dataSet.hasNext());
+ }
+
+ private static void standardAccumulatorQueryByTimeOnlyInvolveOldDataTypeTest(
+ ISession session, TSDataType from, TSDataType newType)
+ throws StatementExecutionException, IoTDBConnectionException {
+ if (!newType.isCompatible(from)) {
+ log.info(newType + " is Incompatible with " + from + ".");
+ }
+
+ RowRecord rec = null;
+ RowRecord unsupportMinMaxRec = null;
+ SessionDataSet dataSet = null;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ dataSet =
+ session.executeQueryStatement(
+ "select first_value(s1),last_value(s1) from "
+ + database
+ + ".write_and_alter_column_type_write where time <= 512");
+ unsupportMinMaxRec = dataSet.next();
+ Assert.assertNotNull(unsupportMinMaxRec);
+ } else {
+ dataSet =
+ session.executeQueryStatement(
+ "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
+ + database
+ + ".write_and_alter_column_type_write where time <= 512");
+ rec = dataSet.next();
+ Assert.assertNotNull(rec);
+ }
+
+ Object[] expectedValue;
+ switch (from) {
+ case BOOLEAN:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Boolean[] {false, true};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Boolean[] {false, true, false, true};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ }
+ break;
+ case DATE:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new String[] {"1970-01-02", "1971-05-28"};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ String.valueOf(expectedValue[i]), unsupportMinMaxRec.getFields().get(i).toString());
+ }
+ } else {
+ expectedValue = new String[] {"1970-01-02", "1971-05-28", "1970-01-02", "1971-05-28"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
+ }
+ }
+ break;
+ case DOUBLE:
+ case BLOB:
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {1, 512};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
+ i,
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1, 99, 1, 512};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
i,
- genValue(newType, expectedValue[i]).toString(),
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
rec.getFields().get(i).toString());
assertEquals(
- genValue(newType, expectedValue[i]).toString(), rec.getFields().get(i).toString());
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).getObjectValue(newType));
}
}
+ break;
+ case FLOAT:
+ case INT64:
+ case TIMESTAMP:
+ case INT32:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {1, 512};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
+ i,
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1, 99, 1, 512};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i,
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ }
+ break;
+ default:
+ expectedValue = new Integer[] {1, 512, 1, 512};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i,
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ }
+ break;
+ default:
+ break;
+ }
- assertFalse(dataSet.hasNext());
+ // test avg and sum
+ if (newType.isNumeric()) {
+ dataSet =
+ session.executeQueryStatement(
+ "select avg(s1),sum(s1) from "
+ + database
+ + ".write_and_alter_column_type_write where time <= 512");
+ rec = dataSet.next();
+ assertEquals(256.5, rec.getFields().get(0).getDoubleV(), 0.001);
+ assertEquals(131328.0, rec.getFields().get(1).getDoubleV(), 0.001);
+ assertFalse(dataSet.hasNext());
+ }
+
+ // can get statistics information
+ dataSet =
+ session.executeQueryStatement(
+ "select count(*) from "
+ + database
+ + ".write_and_alter_column_type_write where time <= 512");
+ rec = dataSet.next();
+ assertEquals(512, rec.getFields().get(0).getLongV());
+ assertFalse(dataSet.hasNext());
+ }
+
+ private static void standardAccumulatorQueryByTimeOnlyInvolveNewDataTypeTest(
+ ISession session, TSDataType from, TSDataType newType)
+ throws StatementExecutionException, IoTDBConnectionException {
+ if (!newType.isCompatible(from)) {
+ log.info(newType + " is Incompatible with " + from + ".");
+ }
+
+ RowRecord rec = null;
+ RowRecord unsupportMinMaxRec = null;
+ SessionDataSet dataSet = null;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ dataSet =
+ session.executeQueryStatement(
+ "select first_value(s1),last_value(s1) from "
+ + database
+ + ".write_and_alter_column_type_write where time > 512");
+ unsupportMinMaxRec = dataSet.next();
+ Assert.assertNotNull(unsupportMinMaxRec);
+ } else {
+ dataSet =
+ session.executeQueryStatement(
+ "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
+ + database
+ + ".write_and_alter_column_type_write where time > 512");
+ rec = dataSet.next();
+ Assert.assertNotNull(rec);
+ }
- if (newType.isNumeric()) {
- dataSet =
- session.executeQueryStatement(
- "select avg(s1),sum(s1) from " + database + ".construct_and_alter_column_type");
- rec = dataSet.next();
- assertEquals(512.5, rec.getFields().get(0).getDoubleV(), 0.001);
- assertEquals(524800.0, rec.getFields().get(1).getDoubleV(), 0.001);
- assertFalse(dataSet.hasNext());
+ Object[] expectedValue;
+ switch (from) {
+ case BOOLEAN:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {513, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1000, 999, 513, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
}
- }
+ break;
+ case DATE:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {513, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ String.valueOf(expectedValue[i]), unsupportMinMaxRec.getFields().get(i).toString());
+ }
+ } else {
+ expectedValue = new Integer[] {1000, 999, 513, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
+ }
+ }
+ break;
+ case DOUBLE:
+ case BLOB:
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {513, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
+ i,
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1000, 999, 513, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ }
+ break;
+ case FLOAT:
+ case INT64:
+ case TIMESTAMP:
+ case INT32:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {513, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
+ i,
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1000, 999, 513, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i,
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ }
+ break;
+ default:
+ expectedValue = new Integer[] {513, 1024, 513, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ }
+ break;
+ default:
+ break;
+ }
- // can use statistics information
+ // test avg and sum
+ if (newType.isNumeric()) {
dataSet =
session.executeQueryStatement(
- "select count(*) from "
+ "select avg(s1),sum(s1) from "
+ database
- + ".construct_and_alter_column_type where time > 0");
+ + ".write_and_alter_column_type_write where time > 512");
rec = dataSet.next();
- assertEquals(1024, rec.getFields().get(0).getLongV());
+ assertEquals(768.5, rec.getFields().get(0).getDoubleV(), 0.001);
+ assertEquals(393472.0, rec.getFields().get(1).getDoubleV(), 0.001);
assertFalse(dataSet.hasNext());
+ }
+
+ // can get statistics information
+ dataSet =
+ session.executeQueryStatement(
+ "select count(*) from "
+ + database
+ + ".write_and_alter_column_type_write where time > 512");
+ rec = dataSet.next();
+ assertEquals(512, rec.getFields().get(0).getLongV());
+ assertFalse(dataSet.hasNext());
+ }
+
+ private static void standardAccumulatorQueryByTimeInvolveAllDataTypeTest(
+ ISession session, TSDataType from, TSDataType newType)
+ throws StatementExecutionException, IoTDBConnectionException {
+ if (!newType.isCompatible(from)) {
+ log.info(newType + " is Incompatible with " + from + ".");
+ }
- // can't use statistics information
+ RowRecord rec = null;
+ RowRecord unsupportMinMaxRec = null;
+ SessionDataSet dataSet = null;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ dataSet =
+ session.executeQueryStatement(
+ "select first_value(s1),last_value(s1) from "
+ + database
+ + ".write_and_alter_column_type_write");
+ unsupportMinMaxRec = dataSet.next();
+ Assert.assertNotNull(unsupportMinMaxRec);
+ } else {
dataSet =
session.executeQueryStatement(
- "select count(*) from "
+ "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
+ database
- + ".construct_and_alter_column_type where time > 10000");
+ + ".write_and_alter_column_type_write");
rec = dataSet.next();
- assertEquals(0, rec.getFields().get(0).getLongV());
+ Assert.assertNotNull(rec);
+ }
+
+ Object[] expectedValue;
+ switch (from) {
+ case BOOLEAN:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new String[] {"false", "1024"};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new String[] {"1000", "true", "false", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ }
+ break;
+ case DATE:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new String[] {"1970-01-02", "1024"};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ String.valueOf(expectedValue[i]), unsupportMinMaxRec.getFields().get(i).toString());
+ }
+ } else {
+ expectedValue = new String[] {"1000", "999", "1970-01-02", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
+ }
+ }
+ break;
+ case DOUBLE:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new String[] {"1.0", "1024"};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ }
+ } else {
+ expectedValue = new String[] {"1.0", "999", "1.0", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).toString());
+ }
+ }
+ break;
+ case BLOB:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new String[] {"1", "1024"};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ }
+ } else {
+ expectedValue = new String[] {"1", "999", "1", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).toString());
+ }
+ }
+ break;
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {1, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ new Binary(expectedValue[i].toString(), StandardCharsets.UTF_8),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ new Binary(expectedValue[i].toString(), StandardCharsets.UTF_8),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ }
+ break;
+ case FLOAT:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new String[] {"1.0", "1024"};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
+ i, expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], unsupportMinMaxRec.getFields().get(i).toString());
+ }
+ } else {
+ expectedValue = new String[] {"1.0", "999", "1.0", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).toString());
+ }
+ }
+ break;
+ default:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ }
+ break;
+ case INT64:
+ case TIMESTAMP:
+ case INT32:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ expectedValue = new Integer[] {1, 1024};
+ for (int i = 0; i < 2; i++) {
+ assertExpectLog(
+ i,
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ unsupportMinMaxRec.getFields().get(i).getObjectValue(newType));
+ }
+ } else {
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i,
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ }
+ break;
+ default:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ }
+ break;
+ default:
+ break;
+ }
+
+ // test avg and sum
+ if (newType.isNumeric()) {
+ dataSet =
+ session.executeQueryStatement(
+ "select avg(s1),sum(s1) from " + database + ".write_and_alter_column_type_write");
+ rec = dataSet.next();
+ assertEquals(512.5, rec.getFields().get(0).getDoubleV(), 0.001);
+ assertEquals(524800.0, rec.getFields().get(1).getDoubleV(), 0.001);
assertFalse(dataSet.hasNext());
}
+
+ // can get statistics information
+ dataSet =
+ session.executeQueryStatement(
+ "select count(*) from " + database + ".write_and_alter_column_type_write");
+ rec = dataSet.next();
+ assertEquals(1024, rec.getFields().get(0).getLongV());
+ assertFalse(dataSet.hasNext());
}
private static void accumulatorQueryTestForDateType(ISession session, TSDataType newType)
throws StatementExecutionException, IoTDBConnectionException {
- if (newType != TSDataType.STRING && newType != TSDataType.TEXT) {
- return;
- }
-
log.info("Test the result that after transfered newType:");
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select first_value(s1),last_value(s1) from "
- + database
- + ".construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- int[] expectedValue = {19700102, 19721021};
- if (newType != TSDataType.BOOLEAN) {
+ SessionDataSet dataSet = null;
+ RowRecord rec = null;
+ if (UNSUPPORT_MIN_AND_MAX_QUERY_DATA_TYPE_LIST.contains(newType)) {
+ dataSet =
+ session.executeQueryStatement(
+ "select first_value(s1),last_value(s1) from "
+ + database
+ + ".construct_and_alter_column_type");
+ rec = dataSet.next();
+ Object[] expectedValue = new String[] {"1970-01-02", "1972-10-21"};
for (int i = 0; i < 2; i++) {
- if (newType == TSDataType.STRING || newType == TSDataType.TEXT) {
- log.info(
- "i is {}, expected value: {}, actual value: {}",
- i,
- TSDataType.getDateStringValue(expectedValue[i]),
- // rec.getFields().get(i).getBinaryV().toString());
- rec.getFields().get(i).getStringValue());
- assertEquals(
- TSDataType.getDateStringValue(expectedValue[i]),
- rec.getFields().get(i).getBinaryV().toString());
- }
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
+ }
+ } else {
+ dataSet =
+ session.executeQueryStatement(
+ "select min_value(s1),max_value(s1),first_value(s1),last_value(s1) from "
+ + database
+ + ".construct_and_alter_column_type");
+ rec = dataSet.next();
+ Object[] expectedValue =
+ new String[] {"1970-01-02", "1972-10-21", "1970-01-02", "1972-10-21"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
}
}
@@ -2613,6 +3377,17 @@ public class IoTDBAlterTimeSeriesTypeIT {
assertFalse(dataSet.hasNext());
}
+ private static void assertExpectLog(int i, Object expectedValue, Object actualValue) {
+ StackTraceElement[] stackTraceElements = Thread.currentThread().getStackTrace();
+ String callerMethodName = stackTraceElements[2].getMethodName();
+ log.info(
+ "[{}] i is {}, expected value: {}, actual value: {}",
+ callerMethodName,
+ i,
+ expectedValue,
+ actualValue);
+ }
+
private static void writeWithTablets(
long timestamp,
Object dataValue,
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBAlterColumnTypeIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBAlterColumnTypeIT.java
index 53f15a192e8..0903a0a3a0d 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBAlterColumnTypeIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBAlterColumnTypeIT.java
@@ -96,8 +96,6 @@ public class IoTDBAlterColumnTypeIT {
private static final Logger log = LoggerFactory.getLogger(IoTDBAlterColumnTypeIT.class);
private static long timeout = -1;
- public static final List<TSDataType> DATA_TYPE_LIST =
- Arrays.asList(TSDataType.BLOB, TSDataType.STRING, TSDataType.TEXT);
@BeforeClass
public static void setUp() throws Exception {
@@ -131,6 +129,8 @@ public class IoTDBAlterColumnTypeIT {
testAlignDeviceSequenceDataQuery(from, to);
testAlignDeviceUnSequenceDataQuery(from, to);
testAlignDeviceUnSequenceOverlappedDataQuery(from, to);
+
+ doWriteAndAlterWriteForAccumulatorQuery(from, to);
}
}
}
@@ -320,6 +320,75 @@ public class IoTDBAlterColumnTypeIT {
}
}
+ private void doWriteAndAlterWriteForAccumulatorQuery(TSDataType from, TSDataType to)
+ throws IoTDBConnectionException, StatementExecutionException {
+ try (ITableSession session = EnvFactory.getEnv().getTableSessionConnectionWithDB("test")) {
+ session.executeNonQueryStatement("SET CONFIGURATION enable_unseq_space_compaction='false'");
+ session.executeNonQueryStatement("SET CONFIGURATION enable_seq_space_compaction='false'");
+ if (from == TSDataType.DATE && !to.isCompatible(from)) {
+ throw new NotSupportedException("Not supported DATE type.");
+ }
+
+ // create a table with type of "from"
+ session.executeNonQueryStatement(
+ "CREATE TABLE write_and_alter_column_type_write (s1 " + from + ")");
+
+ // write a sequence tsfile point of "from"
+ Tablet tablet =
+ new Tablet(
+ "write_and_alter_column_type_write",
+ Collections.singletonList("s1"),
+ Collections.singletonList(from),
+ Collections.singletonList(ColumnCategory.FIELD));
+ for (int i = 1; i <= 512; i++) {
+ tablet.addTimestamp(0, i);
+ tablet.addValue("s1", 0, genValue(from, i));
+ session.insert(tablet);
+ tablet.reset();
+ }
+ session.executeNonQueryStatement("FLUSH");
+
+ // alter the type to "to"
+ boolean isCompatible = MetadataUtils.canAlter(from, to);
+ if (isCompatible) {
+ session.executeNonQueryStatement(
+ "ALTER TABLE write_and_alter_column_type_write ALTER COLUMN s1 SET DATA TYPE " + to);
+ } else {
+ try {
+ session.executeNonQueryStatement(
+ "ALTER TABLE write_and_alter_column_type_write ALTER COLUMN s1 SET DATA TYPE " + to);
+ } catch (StatementExecutionException e) {
+ assertEquals(
+ "701: New type " + to + " is not compatible with the existing one " + from,
+ e.getMessage());
+ }
+ }
+
+ // write a sequence tsfile point of "to"
+ tablet =
+ new Tablet(
+ "write_and_alter_column_type_write",
+ Collections.singletonList("s1"),
+ Collections.singletonList(to),
+ Collections.singletonList(ColumnCategory.FIELD));
+ for (int i = 513; i <= 1024; i++) {
+ tablet.addTimestamp(0, i);
+ tablet.addValue("s1", 0, genValue(to, i));
+ session.insert(tablet);
+ tablet.reset();
+ }
+ session.executeNonQueryStatement("FLUSH");
+
+ standardAccumulatorQueryByTimeOnlyInvolveOldDataTypeTest(session, from, to);
+ standardAccumulatorQueryByTimeOnlyInvolveNewDataTypeTest(session, from, to);
+ standardAccumulatorQueryByTimeInvolveAllDataTypeTest(session, from, to);
+ } finally {
+ try (ITableSession session = EnvFactory.getEnv().getTableSessionConnectionWithDB("test")) {
+ session.executeNonQueryStatement("DROP TABLE write_and_alter_column_type_write");
+ }
+ }
+ }
+
@Test
public void testAlterWithoutWrite() throws IoTDBConnectionException, StatementExecutionException {
Set<TSDataType> typesToTest = new HashSet<>();
@@ -1317,26 +1386,15 @@ public class IoTDBAlterColumnTypeIT {
tablet.reset();
}
- if (DATA_TYPE_LIST.contains(from) || DATA_TYPE_LIST.contains(to)) {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select first(s1),last(s1) from construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- rec = dataSet.next();
- }
- dataSet.close();
- } else {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select min(s1),max(s1),first(s1),last(s1) from construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select min(s1),max(s1),first(s1),last(s1) from construct_and_alter_column_type");
+ RowRecord rec = dataSet.next();
+ while (rec != null) {
+ System.out.println(rec.getFields().toString());
+ rec = dataSet.next();
}
+ dataSet.close();
try {
standardSelectTest(session, from, to);
@@ -1432,27 +1490,15 @@ public class IoTDBAlterColumnTypeIT {
}
// session.executeNonQueryStatement("FLUSH");
- if (DATA_TYPE_LIST.contains(from) || DATA_TYPE_LIST.contains(to)) {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select first(s1),last(s1) from construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
- } else {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select min(s1),max(s1),first(s1),last(s1) from construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select min(s1),max(s1),first(s1),last(s1) from construct_and_alter_column_type");
+ RowRecord rec = dataSet.next();
+ while (rec != null) {
+ System.out.println(rec.getFields().toString());
+ rec = dataSet.next();
}
+ dataSet.close();
try {
standardSelectTest(session, from, to);
@@ -1550,27 +1596,15 @@ public class IoTDBAlterColumnTypeIT {
}
// session.executeNonQueryStatement("FLUSH");
- if (DATA_TYPE_LIST.contains(from) || DATA_TYPE_LIST.contains(to)) {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select first(s1),last(s1) from construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
- } else {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select min(s1),max(s1),first(s1),last(s1) from construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- while (rec != null) {
- System.out.println(rec.getFields().toString());
- rec = dataSet.next();
- }
- dataSet.close();
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select min(s1),max(s1),first(s1),last(s1) from construct_and_alter_column_type");
+ RowRecord rec = dataSet.next();
+ while (rec != null) {
+ System.out.println(rec.getFields().toString());
+ rec = dataSet.next();
}
+ dataSet.close();
try {
standardSelectTest(session, from, to);
@@ -1964,74 +1998,217 @@ public class IoTDBAlterColumnTypeIT {
private static void standardAccumulatorQueryTest(ITableSession session, TSDataType newType)
throws StatementExecutionException, IoTDBConnectionException {
- if (newType == TSDataType.DATE) {
- throw new NotSupportedException("Not supported DATE type.");
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select min(s1),max(s1),first(s1),last(s1) from construct_and_alter_column_type");
+ RowRecord rec = dataSet.next();
+ Object[] expectedValue;
+ switch (newType) {
+ case BOOLEAN:
+ expectedValue = new Boolean[] {false, true, false, true};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ case DATE:
+ expectedValue = new String[] {"1970-01-02", "1971-05-28", "1970-01-02", "1971-05-28"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
+ }
+ break;
+ case DOUBLE:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ case BLOB:
+ case STRING:
+ case TEXT:
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ case FLOAT:
+ case INT64:
+ case TIMESTAMP:
+ case INT32:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ break;
+ default:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ break;
+ }
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ default:
+ break;
}
- SessionDataSet dataSet;
- RowRecord rec;
- int[] expectedValue;
- int max = 4;
- if (DATA_TYPE_LIST.contains(newType)) {
- dataSet =
- session.executeQueryStatement(
- "select first(s1),last(s1) from construct_and_alter_column_type");
- rec = dataSet.next();
- expectedValue = new int[] {1, 1024};
- if (newType == TSDataType.STRING
- || newType == TSDataType.TEXT
- || newType == TSDataType.BLOB) {
- // expectedValue[1] = 999;
- } else if (newType == TSDataType.BOOLEAN) {
- expectedValue = new int[] {19700102, 19721021};
- }
- max = 2;
- } else {
+ // test avg and sum
+ if (newType.isNumeric()) {
dataSet =
session.executeQueryStatement(
- "select min(s1),max(s1),first(s1),last(s1) from construct_and_alter_column_type");
+ "select avg(s1),sum(s1) from construct_and_alter_column_type");
rec = dataSet.next();
- expectedValue = new int[] {1, 1024, 1, 1024};
- if (newType == TSDataType.STRING
- || newType == TSDataType.TEXT
- || newType == TSDataType.BLOB) {
- expectedValue[1] = 999;
- } else if (newType == TSDataType.BOOLEAN) {
- expectedValue = new int[] {19700102, 19721021, 19700102, 19721021};
- }
+ assertEquals(512.5, rec.getFields().get(0).getDoubleV(), 0.001);
+ assertEquals(524800.0, rec.getFields().get(1).getDoubleV(), 0.001);
+ assertFalse(dataSet.hasNext());
}
- if (newType != TSDataType.BOOLEAN) {
- for (int i = 0; i < max; i++) {
- if (newType == TSDataType.BLOB) {
- assertEquals(genValue(newType, expectedValue[i]), rec.getFields().get(i).getBinaryV());
- } else if (newType == TSDataType.DATE) {
- assertEquals(genValue(newType, expectedValue[i]), rec.getFields().get(i).getDateV());
- } else {
- log.info(
- "i is {}, expected value: {}, actual value: {}",
+ // can use statistics information
+ dataSet =
+ session.executeQueryStatement(
+ "select count(*) from construct_and_alter_column_type where time > 0");
+ rec = dataSet.next();
+ assertEquals(1024, rec.getFields().get(0).getLongV());
+ assertFalse(dataSet.hasNext());
+
+ // can't use statistics information
+ dataSet =
+ session.executeQueryStatement(
+ "select count(*) from construct_and_alter_column_type where time > 10000");
+ rec = dataSet.next();
+ assertEquals(0, rec.getFields().get(0).getLongV());
+ assertFalse(dataSet.hasNext());
+ }
+
+ private static void standardAccumulatorQueryTest(
+ ITableSession session, TSDataType from, TSDataType newType)
+ throws StatementExecutionException, IoTDBConnectionException {
+ if (!newType.isCompatible(from)) {
+ log.info(newType + " is Incompatible with " + from + ".");
+ }
+
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select min(s1),max(s1),first(s1),last(s1) from construct_and_alter_column_type");
+ RowRecord rec = dataSet.next();
+ Object[] expectedValue;
+ switch (from) {
+ case BOOLEAN:
+ expectedValue = new Boolean[] {false, true, false, true};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ case DATE:
+ expectedValue = new String[] {"1970-01-02", "1971-05-28", "1970-01-02", "1971-05-28"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
+ }
+ break;
+ case DOUBLE:
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
i,
- genValue(newType, expectedValue[i]).toString(),
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
rec.getFields().get(i).toString());
assertEquals(
- genValue(newType, expectedValue[i]).toString(), rec.getFields().get(i).toString());
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).getObjectValue(newType));
}
- }
+ break;
+ case BLOB:
+ expectedValue = new String[] {"1", "999", "1", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).toString());
+ }
+ break;
+ case STRING:
+ case TEXT:
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ new Binary(expectedValue[i].toString(), StandardCharsets.UTF_8),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ case FLOAT:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ expectedValue = new String[] {"1.0", "999.0", "1.0", "1024.0"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).toString());
+ }
+ break;
+ default:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ }
+ break;
+ case INT64:
+ case TIMESTAMP:
+ case INT32:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ break;
+ default:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ break;
+ }
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ default:
+ break;
+ }
+ // test avg and sum
+ if (newType.isNumeric()) {
+ dataSet =
+ session.executeQueryStatement(
+ "select avg(s1),sum(s1) from construct_and_alter_column_type");
+ rec = dataSet.next();
+ assertEquals(512.5, rec.getFields().get(0).getDoubleV(), 0.001);
+ assertEquals(524800.0, rec.getFields().get(1).getDoubleV(), 0.001);
assertFalse(dataSet.hasNext());
-
- if (newType.isNumeric()) {
- dataSet =
- session.executeQueryStatement(
- "select avg(s1),sum(s1) from construct_and_alter_column_type");
- rec = dataSet.next();
- assertEquals(512.5, rec.getFields().get(0).getDoubleV(), 0.001);
- assertEquals(524800.0, rec.getFields().get(1).getDoubleV(), 0.001);
- assertFalse(dataSet.hasNext());
- }
}
- // can use statistics information
+ // can get statistics information
dataSet =
session.executeQueryStatement(
"select count(*) from construct_and_alter_column_type where time > 0");
@@ -2048,146 +2225,315 @@ public class IoTDBAlterColumnTypeIT {
assertFalse(dataSet.hasNext());
}
- private static void standardAccumulatorQueryTest(
+ private static void standardAccumulatorQueryByTimeOnlyInvolveOldDataTypeTest(
ITableSession session, TSDataType from, TSDataType newType)
throws StatementExecutionException, IoTDBConnectionException {
- if (from == TSDataType.DATE) {
- throw new NotSupportedException("Not supported DATE type.");
+ if (!newType.isCompatible(from)) {
+ log.info(newType + " is Incompatible with " + from + ".");
}
- if (from == TSDataType.BLOB || newType == TSDataType.BLOB) {
- throw new NotSupportedException("Not supported BLOB type.");
- }
- if (from == TSDataType.BOOLEAN && DATA_TYPE_LIST.contains(newType)) {
- SessionDataSet dataSet =
- session.executeQueryStatement(
- "select first(s1),last(s1) from construct_and_alter_column_type");
- RowRecord rec = dataSet.next();
- boolean[] expectedValue = {false, true};
- for (int i = 0; i < 2; i++) {
- if (newType == TSDataType.STRING || newType == TSDataType.TEXT) {
+
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select min(s1),max(s1),first(s1),last(s1) from write_and_alter_column_type_write where time <= 512");
+ RowRecord rec = dataSet.next();
+ Object[] expectedValue;
+ switch (from) {
+ case BOOLEAN:
+ expectedValue = new Boolean[] {false, true, false, true};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ case DATE:
+ expectedValue = new String[] {"1970-01-02", "1971-05-28", "1970-01-02", "1971-05-28"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
}
- }
- } else {
- SessionDataSet dataSet;
- int[] expectedValue;
- int max = 4;
- if (DATA_TYPE_LIST.contains(newType)) {
- dataSet =
- session.executeQueryStatement(
- "select first(s1),last(s1) from construct_and_alter_column_type");
- expectedValue = new int[] {1, 1024};
- max = 2;
- } else {
- dataSet =
- session.executeQueryStatement(
- "select min(s1),max(s1),first(s1),last(s1) from construct_and_alter_column_type");
- expectedValue = new int[] {1, 1024, 1, 1024};
- if (newType == TSDataType.STRING
- || newType == TSDataType.TEXT
- || newType == TSDataType.BLOB) {
- expectedValue[1] = 999;
- } else if (newType == TSDataType.BOOLEAN) {
- expectedValue = new int[] {19700102, 19721021, 19700102, 19721021};
+ break;
+ case DOUBLE:
+ case BLOB:
+ case STRING:
+ case TEXT:
+ expectedValue = new Integer[] {1, 99, 1, 512};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i,
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).getObjectValue(newType));
}
- }
-
- RowRecord rec = dataSet.next();
- if (newType != TSDataType.BOOLEAN) {
- for (int i = 0; i < max; i++) {
- if (newType == TSDataType.BLOB) {
- assertEquals(genValue(newType, expectedValue[i]), rec.getFields().get(i).getBinaryV());
- } else if (newType == TSDataType.DATE) {
- assertEquals(genValue(newType, expectedValue[i]), rec.getFields().get(i).getDateV());
- } else if (newType == TSDataType.STRING || newType == TSDataType.TEXT) {
- if (from == TSDataType.DATE) {
- log.info(
- "i is {}, expected value: {}, actual value: {}",
- i,
- new Binary(genValue(from, expectedValue[i]).toString(), StandardCharsets.UTF_8),
- rec.getFields().get(i).getBinaryV());
- assertEquals(
- new Binary(genValue(from, expectedValue[i]).toString(), StandardCharsets.UTF_8),
- rec.getFields().get(i).getBinaryV());
- } else {
- log.info(
- "i is {}, expected value: {}, actual value: {}",
- i,
- newType.castFromSingleValue(from, genValue(from, expectedValue[i])),
- rec.getFields().get(i).getBinaryV());
- assertEquals(
- newType.castFromSingleValue(from, genValue(from, expectedValue[i])),
- rec.getFields().get(i).getBinaryV());
- }
- } else {
- log.info(
- "i is {}, expected value: {}, actual value: {}",
- i,
- genValue(newType, expectedValue[i]).toString(),
- rec.getFields().get(i).toString());
- assertEquals(
- genValue(newType, expectedValue[i]).toString(), rec.getFields().get(i).toString());
- }
+ break;
+ case FLOAT:
+ case INT64:
+ case TIMESTAMP:
+ case INT32:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ expectedValue = new Integer[] {1, 99, 1, 512};
+ break;
+ default:
+ expectedValue = new Integer[] {1, 512, 1, 512};
+ break;
}
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i,
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, genValue(from, (int) expectedValue[i])),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ default:
+ break;
+ }
- assertFalse(dataSet.hasNext());
+ // test avg and sum
+ if (newType.isNumeric()) {
+ dataSet =
+ session.executeQueryStatement(
+ "select avg(s1),sum(s1) from write_and_alter_column_type_write where time <= 512");
+ rec = dataSet.next();
+ assertEquals(256.5, rec.getFields().get(0).getDoubleV(), 0.001);
+ assertEquals(131328.0, rec.getFields().get(1).getDoubleV(), 0.001);
+ assertFalse(dataSet.hasNext());
+ }
+
+ // can get statistics information
+ dataSet =
+ session.executeQueryStatement(
+ "select count(*) from write_and_alter_column_type_write where time <= 512");
+ rec = dataSet.next();
+ assertEquals(512, rec.getFields().get(0).getLongV());
+ assertFalse(dataSet.hasNext());
+ }
+
+ private static void standardAccumulatorQueryByTimeOnlyInvolveNewDataTypeTest(
+ ITableSession session, TSDataType from, TSDataType newType)
+ throws StatementExecutionException, IoTDBConnectionException {
+ if (!newType.isCompatible(from)) {
+ log.info(newType + " is Incompatible with " + from + ".");
+ }
- if (newType.isNumeric()) {
- dataSet =
- session.executeQueryStatement(
- "select avg(s1),sum(s1) from construct_and_alter_column_type");
- rec = dataSet.next();
- assertEquals(512.5, rec.getFields().get(0).getDoubleV(), 0.001);
- assertEquals(524800.0, rec.getFields().get(1).getDoubleV(), 0.001);
- assertFalse(dataSet.hasNext());
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select min(s1),max(s1),first(s1),last(s1) from write_and_alter_column_type_write where time > 512");
+ RowRecord rec = dataSet.next();
+ Object[] expectedValue;
+ switch (from) {
+ case BOOLEAN:
+ expectedValue = new Integer[] {1000, 999, 513, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
}
- }
+ break;
+ case DATE:
+ expectedValue = new Integer[] {1000, 999, 513, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
+ }
+ break;
+ case DOUBLE:
+ case BLOB:
+ case STRING:
+ case TEXT:
+ expectedValue = new Integer[] {1000, 999, 513, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ case FLOAT:
+ case INT64:
+ case TIMESTAMP:
+ case INT32:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ expectedValue = new Integer[] {1000, 999, 513, 1024};
+ break;
+ default:
+ expectedValue = new Integer[] {513, 1024, 513, 1024};
+ break;
+ }
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ default:
+ break;
+ }
- // can use statistics information
+ // test avg and sum
+ if (newType.isNumeric()) {
dataSet =
session.executeQueryStatement(
- "select count(*) from construct_and_alter_column_type where time > 0");
+ "select avg(s1),sum(s1) from write_and_alter_column_type_write where time > 512");
rec = dataSet.next();
- assertEquals(1024, rec.getFields().get(0).getLongV());
+ assertEquals(768.5, rec.getFields().get(0).getDoubleV(), 0.001);
+ assertEquals(393472.0, rec.getFields().get(1).getDoubleV(), 0.001);
assertFalse(dataSet.hasNext());
+ }
+
+ // can get statistics information
+ dataSet =
+ session.executeQueryStatement(
+ "select count(*) from write_and_alter_column_type_write where time > 512");
+ rec = dataSet.next();
+ assertEquals(512, rec.getFields().get(0).getLongV());
+ assertFalse(dataSet.hasNext());
+ }
+
+ private static void standardAccumulatorQueryByTimeInvolveAllDataTypeTest(
+ ITableSession session, TSDataType from, TSDataType newType)
+ throws StatementExecutionException, IoTDBConnectionException {
+ if (!newType.isCompatible(from)) {
+ log.info(newType + " is Incompatible with " + from + ".");
+ }
- // can't use statistics information
+ SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select min(s1),max(s1),first(s1),last(s1) from write_and_alter_column_type_write");
+ RowRecord rec = dataSet.next();
+ Object[] expectedValue;
+ switch (from) {
+ case BOOLEAN:
+ expectedValue = new String[] {"1000", "true", "false", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ newType.castFromSingleValue(from, expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ case DATE:
+ expectedValue = new String[] {"1000", "999", "1970-01-02", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
+ }
+ break;
+ case DOUBLE:
+ expectedValue = new String[] {"1.0", "999", "1.0", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).toString());
+ }
+ break;
+ case BLOB:
+ expectedValue = new String[] {"1", "999", "1", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).toString());
+ }
+ break;
+ case STRING:
+ case TEXT:
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(
+ new Binary(expectedValue[i].toString(), StandardCharsets.UTF_8),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ case FLOAT:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ expectedValue = new String[] {"1.0", "999", "1.0", "1024"};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(expectedValue[i], rec.getFields().get(i).toString());
+ }
+ break;
+ default:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ }
+ break;
+ case INT64:
+ case TIMESTAMP:
+ case INT32:
+ switch (newType) {
+ case STRING:
+ case TEXT:
+ expectedValue = new Integer[] {1, 999, 1, 1024};
+ break;
+ default:
+ expectedValue = new Integer[] {1, 1024, 1, 1024};
+ break;
+ }
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(
+ i, genValue(newType, (int) expectedValue[i]), rec.getFields().get(i).toString());
+ assertEquals(
+ genValue(newType, (int) expectedValue[i]),
+ rec.getFields().get(i).getObjectValue(newType));
+ }
+ break;
+ default:
+ break;
+ }
+
+ // test avg and sum
+ if (newType.isNumeric()) {
dataSet =
session.executeQueryStatement(
- "select count(*) from construct_and_alter_column_type where time > 10000");
+ "select avg(s1),sum(s1) from write_and_alter_column_type_write");
rec = dataSet.next();
- assertEquals(0, rec.getFields().get(0).getLongV());
+ assertEquals(512.5, rec.getFields().get(0).getDoubleV(), 0.001);
+ assertEquals(524800.0, rec.getFields().get(1).getDoubleV(), 0.001);
assertFalse(dataSet.hasNext());
}
+
+ // can get statistics information
+ dataSet =
+ session.executeQueryStatement("select count(*) from write_and_alter_column_type_write");
+ rec = dataSet.next();
+ assertEquals(1024, rec.getFields().get(0).getLongV());
+ assertFalse(dataSet.hasNext());
}
private static void accumulatorQueryTestForDateType(ITableSession session, TSDataType newType)
throws StatementExecutionException, IoTDBConnectionException {
- if (newType != TSDataType.STRING && newType != TSDataType.TEXT) {
- return;
- }
-
log.info("Test the result that after transfered newType:");
SessionDataSet dataSet =
session.executeQueryStatement(
- "select first(s1),last(s1) from construct_and_alter_column_type");
+ "select min(s1),max(s1),first(s1),last(s1) from construct_and_alter_column_type");
RowRecord rec = dataSet.next();
- int[] expectedValue = {19700102, 19721021};
- if (newType != TSDataType.BOOLEAN) {
- for (int i = 0; i < 2; i++) {
- if (newType == TSDataType.STRING || newType == TSDataType.TEXT) {
- log.info(
- "i is {}, expected value: {}, actual value: {}",
- i,
- TSDataType.getDateStringValue(expectedValue[i]),
- // rec.getFields().get(i).getBinaryV().toString());
- rec.getFields().get(i).getStringValue());
- assertEquals(
- TSDataType.getDateStringValue(expectedValue[i]),
- rec.getFields().get(i).getBinaryV().toString());
- }
- }
+ Object[] expectedValue = new String[] {"1970-01-02", "1972-10-21", "1970-01-02", "1972-10-21"};
+
+ for (int i = 0; i < 4; i++) {
+ assertExpectLog(i, expectedValue[i], rec.getFields().get(i).toString());
+ assertEquals(String.valueOf(expectedValue[i]), rec.getFields().get(i).toString());
}
// can use statistics information
@@ -2207,6 +2553,17 @@ public class IoTDBAlterColumnTypeIT {
assertFalse(dataSet.hasNext());
}
+ private static void assertExpectLog(int i, Object expectedValue, Object actualValue) {
+ StackTraceElement[] stackTraceElements = Thread.currentThread().getStackTrace();
+ String callerMethodName = stackTraceElements[2].getMethodName();
+ log.info(
+ "[{}] i is {}, expected value: {}, actual value: {}",
+ callerMethodName,
+ i,
+ expectedValue,
+ actualValue);
+ }
+
private static void writeWithTablets(
long timestamp,
Object dataValue,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java
index 70f3ce5e0e6..9b341dc9971 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java
@@ -472,10 +472,10 @@ public class FileLoaderUtils {
(AbstractAlignedChunkMetadata) chunkMetaData;
for (int i = 0; i < alignedChunkMetadata.getValueChunkMetadataList().size(); i++) {
if (alignedChunkMetadata.getValueChunkMetadataList().get(i) != null) {
- if (!SchemaUtils.isUsingSameColumn(
+ if (!SchemaUtils.isUsingSameStatistics(
alignedChunkMetadata.getValueChunkMetadataList().get(i).getDataType(),
targetDataTypeList.get(i))
- && targetDataTypeList.get(i).equals(TSDataType.STRING)) {
+ && !SchemaUtils.canUseStatisticsAfterAlter(targetDataTypeList.get(i))) {
isModified = true;
alignedChunkMetadata.getValueChunkMetadataList().get(i).setModified(true);
}
@@ -484,8 +484,8 @@ public class FileLoaderUtils {
IChunkLoader chunkLoader = alignedChunkMetadata.getChunkLoader();
chunkReader = chunkLoader.getChunkReader(alignedChunkMetadata, globalTimeFilter);
} else {
- if (!SchemaUtils.isUsingSameColumn(chunkMetaData.getDataType(), targetDataTypeList.get(0))
- && targetDataTypeList.get(0).equals(TSDataType.STRING)) {
+ if (!SchemaUtils.isUsingSameStatistics(chunkMetaData.getDataType(), targetDataTypeList.get(0))
+ && !SchemaUtils.canUseStatisticsAfterAlter(targetDataTypeList.get(0))) {
isModified = true;
chunkMetaData.setModified(true);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
index a44a43d0f17..1e1cffba5ff 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
@@ -511,18 +511,18 @@ public class SeriesScanUtil implements Accountable {
(AbstractAlignedChunkMetadata) chunkMetadata;
for (int i = 0; i < alignedChunkMetadata.getValueChunkMetadataList().size(); i++) {
if ((alignedChunkMetadata.getValueChunkMetadataList().get(i) != null)
- && !SchemaUtils.isUsingSameColumn(
+ && !SchemaUtils.isUsingSameStatistics(
alignedChunkMetadata.getValueChunkMetadataList().get(i).getDataType(),
getTsDataTypeList().get(i))
- && getTsDataTypeList().get(i).equals(TSDataType.STRING)) {
+ && !SchemaUtils.canUseStatisticsAfterAlter(getTsDataTypeList().get(i))) {
alignedChunkMetadata.getValueChunkMetadataList().get(i).setModified(true);
}
}
chunkMetadata = alignedChunkMetadata;
} else if (chunkMetadata instanceof ChunkMetadata) {
- if (!SchemaUtils.isUsingSameColumn(
+ if (!SchemaUtils.isUsingSameStatistics(
chunkMetadata.getDataType(), getTsDataTypeList().get(0))
- && getTsDataTypeList().get(0).equals(TSDataType.STRING)) {
+ && !SchemaUtils.canUseStatisticsAfterAlter(getTsDataTypeList().get(0))) {
chunkMetadata.setModified(true);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
index e1d1aa59849..6044b2331d1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
import org.apache.iotdb.db.utils.constant.SqlConstant;
+import com.google.common.collect.ImmutableSet;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata;
import org.apache.tsfile.file.metadata.AbstractAlignedTimeSeriesMetadata;
@@ -46,12 +47,16 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.addPartialSuffix;
public class SchemaUtils {
private static final Map<TSDataType, Class> dataTypeColumnClassMap;
+ private static final Map<TSDataType, Class> dataTypeColumnStatisticsClassMap;
+ private static final Set<TSDataType> canNotUseStatisticAfterAlterClassSet =
+ ImmutableSet.of(TSDataType.STRING, TSDataType.TEXT, TSDataType.BLOB);
public static final Logger logger = LoggerFactory.getLogger(SchemaUtils.class);
private static final Binary EMPTY_BINARY = new Binary("", StandardCharsets.UTF_8);
@@ -66,6 +71,13 @@ public class SchemaUtils {
dataTypeColumnClassMap.put(TSDataType.STRING, BinaryColumn.class);
dataTypeColumnClassMap.put(TSDataType.BLOB, BinaryColumn.class);
dataTypeColumnClassMap.put(TSDataType.TEXT, BinaryColumn.class);
+
+ dataTypeColumnStatisticsClassMap = new HashMap<>();
+ dataTypeColumnStatisticsClassMap.put(TSDataType.INT32, IntColumn.class);
+ dataTypeColumnStatisticsClassMap.put(TSDataType.DATE, IntColumn.class);
+
+ dataTypeColumnStatisticsClassMap.put(TSDataType.INT64, LongColumn.class);
+ dataTypeColumnStatisticsClassMap.put(TSDataType.TIMESTAMP, LongColumn.class);
}
private SchemaUtils() {}
@@ -294,17 +306,38 @@ public class SchemaUtils {
if (originalDataType == dataType) {
return true;
}
+ if (!dataTypeColumnClassMap.containsKey(originalDataType)
+ || !dataTypeColumnClassMap.containsKey(dataType)) {
+ return false;
+ }
return Objects.equals(
dataTypeColumnClassMap.get(originalDataType), dataTypeColumnClassMap.get(dataType));
}
+ public static boolean isUsingSameStatistics(TSDataType originalDataType, TSDataType dataType) {
+ if (originalDataType == dataType) {
+ return true;
+ }
+ if (!dataTypeColumnStatisticsClassMap.containsKey(originalDataType)
+ || !dataTypeColumnStatisticsClassMap.containsKey(dataType)) {
+ return false;
+ }
+ return Objects.equals(
+ dataTypeColumnStatisticsClassMap.get(originalDataType),
+ dataTypeColumnStatisticsClassMap.get(dataType));
+ }
+
+ public static boolean canUseStatisticsAfterAlter(TSDataType dataType) {
+ return !canNotUseStatisticAfterAlterClassSet.contains(dataType);
+ }
+
public static void changeMetadataModified(
TimeseriesMetadata timeseriesMetadata, TSDataType targetDataType) {
if (timeseriesMetadata == null) {
return;
}
- if (!SchemaUtils.isUsingSameColumn(timeseriesMetadata.getTsDataType(), targetDataType)
- && ((targetDataType == TSDataType.STRING) || (targetDataType == TSDataType.TEXT))) {
+ if (!SchemaUtils.isUsingSameStatistics(timeseriesMetadata.getTsDataType(), targetDataType)
+ && !SchemaUtils.canUseStatisticsAfterAlter(targetDataType)) {
timeseriesMetadata.setModified(true);
List<IChunkMetadata> chunkMetadataList = timeseriesMetadata.getChunkMetadataList();
if (chunkMetadataList != null) {
@@ -328,10 +361,9 @@ public class SchemaUtils {
for (TimeseriesMetadata timeseriesMetadata :
alignedTimeSeriesMetadata.getValueTimeseriesMetadataList()) {
if ((timeseriesMetadata != null)
- && !SchemaUtils.isUsingSameColumn(
+ && !SchemaUtils.isUsingSameStatistics(
timeseriesMetadata.getTsDataType(), targetDataTypeList.get(i))
- && ((targetDataTypeList.get(i) == TSDataType.STRING)
- || (targetDataTypeList.get(i) == TSDataType.TEXT))) {
+ && !SchemaUtils.canUseStatisticsAfterAlter(targetDataTypeList.get(i))) {
timeseriesMetadata.setModified(true);
alignedTimeSeriesMetadata.setModified(true);
List<IChunkMetadata> chunkMetadataList = timeseriesMetadata.getChunkMetadataList();
@@ -353,8 +385,8 @@ public class SchemaUtils {
return;
}
- if (!SchemaUtils.isUsingSameColumn(timeseriesMetadata.getTsDataType(), targetDataType)
- && ((targetDataType == TSDataType.STRING) || (targetDataType == TSDataType.TEXT))) {
+ if (!SchemaUtils.isUsingSameStatistics(timeseriesMetadata.getTsDataType(), targetDataType)
+ && !SchemaUtils.canUseStatisticsAfterAlter(targetDataType)) {
timeseriesMetadata.setModified(true);
List<IChunkMetadata> chunkMetadataList = timeseriesMetadata.getChunkMetadataList();
if (chunkMetadataList != null) {
@@ -372,8 +404,8 @@ public class SchemaUtils {
if (chunkMetadata == null) {
return;
}
- if (!SchemaUtils.isUsingSameColumn(sourceDataType, targetDataType)
- && ((targetDataType == TSDataType.STRING) || (targetDataType == TSDataType.TEXT))) {
+ if (!SchemaUtils.isUsingSameStatistics(sourceDataType, targetDataType)
+ && !SchemaUtils.canUseStatisticsAfterAlter(targetDataType)) {
chunkMetadata.setModified(true);
}
}
@@ -388,9 +420,9 @@ public class SchemaUtils {
int i = 0;
for (IChunkMetadata iChunkMetadata : chunkMetadata.getValueChunkMetadataList()) {
if ((iChunkMetadata != null)
- && !SchemaUtils.isUsingSameColumn(sourceDataType, targetDataTypeList.get(i))
- && ((targetDataTypeList.get(i) == TSDataType.STRING)
- || (targetDataTypeList.get(i) == TSDataType.TEXT))) {
+ && !SchemaUtils.isUsingSameStatistics(
+ iChunkMetadata.getDataType(), targetDataTypeList.get(i))
+ && !SchemaUtils.canUseStatisticsAfterAlter(targetDataTypeList.get(i))) {
iChunkMetadata.setModified(true);
chunkMetadata.setModified(true);
}
(iotdb) 13/16: Pipe: preserve exclusions for complex wildcards (#17170)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit c35a003c452e38cb57a04b409c74ec82158e417a
Author: VGalaxies <vg...@apache.org>
AuthorDate: Fri Feb 6 10:01:27 2026 +0800
Pipe: preserve exclusions for complex wildcards (#17170)
* fix(pipe): preserve exclusions for complex wildcards
* spotless
* perf(pipe): avoid extra traversal in pattern trie
---
.../auto/basic/IoTDBTreePatternFormatIT.java | 58 ++++++++++++++++++++++
.../db/pipe/pattern/TreePatternPruningTest.java | 20 ++++++++
.../pipe/datastructure/pattern/TreePattern.java | 28 ++++++++++-
3 files changed, 104 insertions(+), 2 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBTreePatternFormatIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBTreePatternFormatIT.java
index 00e3078a355..fc5d483ea5c 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBTreePatternFormatIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBTreePatternFormatIT.java
@@ -441,4 +441,62 @@ public class IoTDBTreePatternFormatIT extends AbstractPipeDualTreeModelAutoIT {
"Time,root.db1.d1.s,root.db.d1.t,root.db.d2.s,",
expectedResSet);
}
+
+ @Test
+ public void testIoTDBPatternWithInclusionAndExclusionHistoricalData() throws Exception {
+ final Map<String, String> sourceAttributes = new HashMap<>();
+ sourceAttributes.put(
+ "source.pattern.inclusion", "root.test.g_0.d_2*.**,root.test.g_0.d_20.s_0");
+ sourceAttributes.put("source.pattern.exclusion", "root.test.g_0.d_20.**");
+ sourceAttributes.put("source.inclusion", "data.insert");
+ sourceAttributes.put("user", "root");
+
+ final List<String> insertQueries =
+ Arrays.asList(
+ "insert into root.test.g_0.d_21(time, s_1) values (1, 1)",
+ "insert into root.test.g_0.d_22(time, s_1) values (2, 2)",
+ "insert into root.test.g_0.d_20(time, s_0, s_1) values (3, 3, 3)",
+ "insert into root.test.g_0.d_30(time, s_1) values (4, 4)");
+
+ final Set<String> expectedResSet = new HashSet<>();
+ expectedResSet.add("1,1.0,null,");
+ expectedResSet.add("2,null,2.0,");
+
+ testPipeWithMultiplePatterns(
+ sourceAttributes,
+ insertQueries,
+ true, // isHistorical = true
+ "select * from root.test.**",
+ "Time,root.test.g_0.d_21.s_1,root.test.g_0.d_22.s_1,",
+ expectedResSet);
+ }
+
+ @Test
+ public void testIoTDBPatternWithInclusionAndExclusionRealtimeData() throws Exception {
+ final Map<String, String> sourceAttributes = new HashMap<>();
+ sourceAttributes.put(
+ "source.pattern.inclusion", "root.test.g_0.d_2*.**,root.test.g_0.d_20.s_0");
+ sourceAttributes.put("source.pattern.exclusion", "root.test.g_0.d_20.**");
+ sourceAttributes.put("source.inclusion", "data.insert");
+ sourceAttributes.put("user", "root");
+
+ final List<String> insertQueries =
+ Arrays.asList(
+ "insert into root.test.g_0.d_21(time, s_1) values (1, 1)",
+ "insert into root.test.g_0.d_22(time, s_1) values (2, 2)",
+ "insert into root.test.g_0.d_20(time, s_0, s_1) values (3, 3, 3)",
+ "insert into root.test.g_0.d_30(time, s_1) values (4, 4)");
+
+ final Set<String> expectedResSet = new HashSet<>();
+ expectedResSet.add("1,1.0,null,");
+ expectedResSet.add("2,null,2.0,");
+
+ testPipeWithMultiplePatterns(
+ sourceAttributes,
+ insertQueries,
+ false, // isHistorical = false
+ "select * from root.test.**",
+ "Time,root.test.g_0.d_21.s_1,root.test.g_0.d_22.s_1,",
+ expectedResSet);
+ }
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/TreePatternPruningTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/TreePatternPruningTest.java
index d5a209e5d87..eaa7e178653 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/TreePatternPruningTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/TreePatternPruningTest.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.UnionIoTDBTreePattern;
+import org.apache.iotdb.commons.pipe.datastructure.pattern.WithExclusionIoTDBTreePattern;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
@@ -172,4 +173,23 @@ public class TreePatternPruningTest {
// Expected exception
}
}
+
+ @Test
+ public void testWithExclusionPreserved() {
+ final PipeParameters params =
+ new PipeParameters(
+ new HashMap<String, String>() {
+ {
+ put(
+ PipeSourceConstant.SOURCE_PATTERN_INCLUSION_KEY,
+ "root.test.g_0.d_2*.**,root.test.g_0.d_20.s_0");
+ put(PipeSourceConstant.SOURCE_PATTERN_EXCLUSION_KEY, "root.test.g_0.d_20.**");
+ }
+ });
+
+ final TreePattern result = TreePattern.parsePipePatternFromSourceParameters(params);
+ Assert.assertTrue(result instanceof WithExclusionIoTDBTreePattern);
+ Assert.assertEquals(
+ "INCLUSION(root.test.g_0.d_2*.**), EXCLUSION(root.test.g_0.d_20.**)", result.getPattern());
+ }
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java
index 5d6c808a0bb..399565ecaab 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.commons.pipe.datastructure.pattern;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.path.PathPatternUtil;
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.commons.utils.TestOnly;
@@ -873,6 +874,7 @@ public abstract class TreePattern {
/** A specialized Trie to efficiently check path coverage. */
private static class PatternTrie {
private final TrieNode root = new TrieNode();
+ private final List<PartialPath> complexPatterns = new ArrayList<>();
private static class TrieNode {
// Children nodes mapped by specific path segments (excluding *)
@@ -892,6 +894,12 @@ public abstract class TreePattern {
final String[] nodes = path.getNodes();
for (final String segment : nodes) {
+ if (PathPatternUtil.hasWildcard(segment)
+ && !IoTDBConstant.ONE_LEVEL_PATH_WILDCARD.equals(segment)
+ && !IoTDBConstant.MULTI_LEVEL_PATH_WILDCARD.equals(segment)) {
+ complexPatterns.add(path);
+ return;
+ }
// If we are at a node that is already a MultiLevelWildcard (**),
// everything below is already covered. We can stop adding.
if (node.isMultiLevelWildcard) {
@@ -924,7 +932,15 @@ public abstract class TreePattern {
/** Checks if the given path is covered by any existing pattern in the Trie. */
public boolean isCovered(final PartialPath path) {
- return checkCoverage(root, path.getNodes(), 0);
+ if (checkCoverage(root, path.getNodes(), 0)) {
+ return true;
+ }
+ for (final PartialPath complexPattern : complexPatterns) {
+ if (complexPattern.include(path)) {
+ return true;
+ }
+ }
+ return false;
}
private boolean checkCoverage(final TrieNode node, final String[] pathNodes, final int index) {
@@ -957,7 +973,15 @@ public abstract class TreePattern {
/** Checks if the given path overlaps with any pattern in the Trie. */
public boolean overlaps(final PartialPath path) {
- return checkOverlap(root, path.getNodes(), 0);
+ if (checkOverlap(root, path.getNodes(), 0)) {
+ return true;
+ }
+ for (final PartialPath complexPattern : complexPatterns) {
+ if (complexPattern.overlapWith(path)) {
+ return true;
+ }
+ }
+ return false;
}
private boolean checkOverlap(final TrieNode node, final String[] pathNodes, final int index) {
(iotdb) 16/16: correct metadata "modified" flags and related logic (#17172)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 689d072419a31d5ca2fc7c1a32db1f71eec316da
Author: shuwenwei <55...@users.noreply.github.com>
AuthorDate: Mon Feb 9 15:58:18 2026 +0800
correct metadata "modified" flags and related logic (#17172)
---
.../db/it/schema/IoTDBAlterTimeSeriesTypeIT.java | 6 +-
.../execution/operator/source/FileLoaderUtils.java | 80 ++++---------
.../execution/operator/source/SeriesScanUtil.java | 47 +-------
.../schemaregion/utils/ResourceByPathUtils.java | 6 +-
.../read/reader/chunk/MemAlignedChunkReader.java | 7 ++
.../read/reader/chunk/MemAlignedPageReader.java | 9 +-
.../read/reader/chunk/MemChunkReader.java | 7 ++
.../read/reader/chunk/MemPageReader.java | 9 +-
.../org/apache/iotdb/db/utils/SchemaUtils.java | 126 ++++++---------------
pom.xml | 2 +-
10 files changed, 89 insertions(+), 210 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBAlterTimeSeriesTypeIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBAlterTimeSeriesTypeIT.java
index 3effc3a8b83..e5fa5ca59a7 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBAlterTimeSeriesTypeIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBAlterTimeSeriesTypeIT.java
@@ -26,8 +26,8 @@ import org.apache.iotdb.isession.ISession;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
-import org.apache.iotdb.itbase.category.TableClusterIT;
-import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
@@ -87,7 +87,7 @@ import static org.junit.Assert.fail;
@SuppressWarnings("ResultOfMethodCallIgnored")
@RunWith(IoTDBTestRunner.class)
-@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
+@Category({LocalStandaloneIT.class, ClusterIT.class})
public class IoTDBAlterTimeSeriesTypeIT {
private static final Logger log = LoggerFactory.getLogger(IoTDBAlterTimeSeriesTypeIT.class);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java
index 9b341dc9971..3eff3b9a53f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java
@@ -40,7 +40,6 @@ import org.apache.iotdb.db.utils.ModificationUtils;
import org.apache.iotdb.db.utils.SchemaUtils;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata;
import org.apache.tsfile.file.metadata.AbstractAlignedTimeSeriesMetadata;
import org.apache.tsfile.file.metadata.AlignedTimeSeriesMetadata;
import org.apache.tsfile.file.metadata.IChunkMetadata;
@@ -113,13 +112,11 @@ public class FileLoaderUtils {
context.isDebug(),
context);
if (timeSeriesMetadata != null) {
- SchemaUtils.changeMetadataModified(timeSeriesMetadata, seriesPath.getSeriesType());
long t2 = System.nanoTime();
List<ModEntry> pathModifications =
context.getPathModifications(
resource, seriesPath.getDeviceId(), seriesPath.getMeasurement());
- timeSeriesMetadata.setModified(
- timeSeriesMetadata.isModified() || !pathModifications.isEmpty());
+ timeSeriesMetadata.setModified(!pathModifications.isEmpty());
timeSeriesMetadata.setChunkMetadataLoader(
new DiskChunkMetadataLoader(resource, context, globalTimeFilter, pathModifications));
int modificationCount = pathModifications.size();
@@ -141,13 +138,14 @@ public class FileLoaderUtils {
timeSeriesMetadata =
(TimeseriesMetadata) resource.getTimeSeriesMetadata(seriesPath, globalTimeFilter);
if (timeSeriesMetadata != null) {
- SchemaUtils.changeMetadataModified(timeSeriesMetadata, seriesPath.getSeriesType());
timeSeriesMetadata.setChunkMetadataLoader(
new MemChunkMetadataLoader(resource, seriesPath, context, globalTimeFilter));
}
}
if (timeSeriesMetadata != null) {
+ SchemaUtils.changeTimeseriesMetadataModified(
+ timeSeriesMetadata, seriesPath.getSeriesType());
if (timeSeriesMetadata.getStatistics().getStartTime()
> timeSeriesMetadata.getStatistics().getEndTime()) {
return null;
@@ -209,15 +207,12 @@ public class FileLoaderUtils {
alignedTimeSeriesMetadata =
loadAlignedTimeSeriesMetadataFromDisk(
resource, alignedPath, context, globalTimeFilter, ignoreAllNullRows);
- SchemaUtils.changeAlignedMetadataModified(alignedTimeSeriesMetadata, targetDataTypeList);
} else { // if the tsfile is unclosed, we just get it directly from TsFileResource
loadFromMem = true;
alignedTimeSeriesMetadata =
(AbstractAlignedTimeSeriesMetadata)
resource.getTimeSeriesMetadata(alignedPath, globalTimeFilter);
if (alignedTimeSeriesMetadata != null) {
- SchemaUtils.changeAlignedMetadataModified(alignedTimeSeriesMetadata, targetDataTypeList);
-
alignedTimeSeriesMetadata.setChunkMetadataLoader(
new MemAlignedChunkMetadataLoader(
resource, alignedPath, context, globalTimeFilter, ignoreAllNullRows));
@@ -226,6 +221,8 @@ public class FileLoaderUtils {
}
if (alignedTimeSeriesMetadata != null) {
+ SchemaUtils.changeAlignedTimeseriesMetadataModified(
+ alignedTimeSeriesMetadata, targetDataTypeList);
if (alignedTimeSeriesMetadata.getTimeseriesMetadata().getStatistics().getStartTime()
> alignedTimeSeriesMetadata.getTimeseriesMetadata().getStatistics().getEndTime()) {
return null;
@@ -320,25 +317,18 @@ public class FileLoaderUtils {
new ArrayList<>(valueMeasurementList.size());
// if all the queried aligned sensors does not exist, we will return null
boolean exist = false;
- int i = 0;
- for (String valueMeasurement : valueMeasurementList) {
+ for (String measurement : valueMeasurementList) {
TimeseriesMetadata valueColumn =
cache.get(
filePath,
- new TimeSeriesMetadataCacheKey(
- resource.getTsFileID(), deviceId, valueMeasurement),
+ new TimeSeriesMetadataCacheKey(resource.getTsFileID(), deviceId, measurement),
allSensors,
context.ignoreNotExistsDevice()
|| resource.getTimeIndexType() == ITimeIndex.FILE_TIME_INDEX_TYPE,
isDebug,
context);
exist = (exist || (valueColumn != null));
- if (valueColumn != null) {
- SchemaUtils.changeAlignedMetadataModified(
- valueColumn, alignedPath.getSchemaList().get(i).getType());
- }
valueTimeSeriesMetadataList.add(valueColumn);
- i++;
}
if (!ignoreAllNullRows || exist) {
alignedTimeSeriesMetadata =
@@ -458,55 +448,23 @@ public class FileLoaderUtils {
* IOException will be thrown
*/
public static List<IPageReader> loadPageReaderList(
- IChunkMetadata chunkMetaData,
- Filter globalTimeFilter,
- boolean isAligned,
- List<TSDataType> targetDataTypeList)
+ IChunkMetadata chunkMetaData, Filter globalTimeFilter, List<TSDataType> targetDataTypeList)
throws IOException {
checkArgument(chunkMetaData != null, "Can't init null chunkMeta");
+ IChunkLoader chunkLoader = chunkMetaData.getChunkLoader();
IChunkReader chunkReader;
- boolean isModified = false;
- if (isAligned) {
- AbstractAlignedChunkMetadata alignedChunkMetadata =
- (AbstractAlignedChunkMetadata) chunkMetaData;
- for (int i = 0; i < alignedChunkMetadata.getValueChunkMetadataList().size(); i++) {
- if (alignedChunkMetadata.getValueChunkMetadataList().get(i) != null) {
- if (!SchemaUtils.isUsingSameStatistics(
- alignedChunkMetadata.getValueChunkMetadataList().get(i).getDataType(),
- targetDataTypeList.get(i))
- && !SchemaUtils.canUseStatisticsAfterAlter(targetDataTypeList.get(i))) {
- isModified = true;
- alignedChunkMetadata.getValueChunkMetadataList().get(i).setModified(true);
- }
- }
- }
- IChunkLoader chunkLoader = alignedChunkMetadata.getChunkLoader();
- chunkReader = chunkLoader.getChunkReader(alignedChunkMetadata, globalTimeFilter);
- } else {
- if (!SchemaUtils.isUsingSameStatistics(chunkMetaData.getDataType(), targetDataTypeList.get(0))
- && !SchemaUtils.canUseStatisticsAfterAlter(targetDataTypeList.get(0))) {
- isModified = true;
- chunkMetaData.setModified(true);
- }
- IChunkLoader chunkLoader = chunkMetaData.getChunkLoader();
- try {
- chunkReader = chunkLoader.getChunkReader(chunkMetaData, globalTimeFilter);
- } catch (ChunkTypeInconsistentException e) {
- // if the chunk in tsfile is a value chunk of aligned series, we should skip all data of
- // this chunk.
- return Collections.emptyList();
- }
+ try {
+ chunkReader = chunkLoader.getChunkReader(chunkMetaData, globalTimeFilter);
+ } catch (ChunkTypeInconsistentException e) {
+ // if the chunk in tsfile is a value chunk of aligned series but registered series is
+ // non-aligned, we should skip all data of this chunk.
+ return Collections.emptyList();
}
-
- return isModified
- ? chunkReader.loadPageReaderList().stream()
- .peek(
- iPageReader -> {
- iPageReader.setModified(true);
- })
- .collect(Collectors.toList())
- : chunkReader.loadPageReaderList();
+ if (chunkMetaData.isDataTypeModifiedAndCannotUseStatistics()) {
+ chunkReader.markDataTypeModifiedAndCannotUseStatistics();
+ }
+ return chunkReader.loadPageReaderList();
}
/**
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
index 1e1cffba5ff..5e7111bf09c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
@@ -45,8 +45,6 @@ import org.apache.iotdb.db.utils.datastructure.MemPointIterator;
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata;
-import org.apache.tsfile.file.metadata.ChunkMetadata;
import org.apache.tsfile.file.metadata.IChunkMetadata;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.file.metadata.IMetadata;
@@ -453,15 +451,6 @@ public class SeriesScanUtil implements Accountable {
orderUtils.getOverlapCheckTime(firstChunkMetadata.getStatistics()));
unpackAllOverlappedTimeSeriesMetadataToCachedChunkMetadata(
orderUtils.getOverlapCheckTime(firstChunkMetadata.getStatistics()), false);
- if (isAligned) {
- SchemaUtils.changeAlignedMetadataModified(
- (AbstractAlignedChunkMetadata) firstChunkMetadata,
- firstChunkMetadata.getDataType(),
- getTsDataTypeList());
- } else {
- SchemaUtils.changeMetadataModified(
- firstChunkMetadata, firstChunkMetadata.getDataType(), dataType);
- }
if (firstChunkMetadata.equals(cachedChunkMetadata.peek())) {
firstChunkMetadata = cachedChunkMetadata.poll();
break;
@@ -489,45 +478,13 @@ public class SeriesScanUtil implements Accountable {
if (init && firstChunkMetadata == null && !cachedChunkMetadata.isEmpty()) {
firstChunkMetadata = cachedChunkMetadata.poll();
- if (isAligned) {
- SchemaUtils.changeAlignedMetadataModified(
- (AbstractAlignedChunkMetadata) firstChunkMetadata,
- firstChunkMetadata.getDataType(),
- getTsDataTypeList());
- } else {
- SchemaUtils.changeMetadataModified(
- firstChunkMetadata, firstChunkMetadata.getDataType(), dataType);
- }
}
}
protected void unpackOneTimeSeriesMetadata(ITimeSeriesMetadata timeSeriesMetadata) {
List<IChunkMetadata> chunkMetadataList =
FileLoaderUtils.loadChunkMetadataList(timeSeriesMetadata);
- chunkMetadataList.forEach(
- chunkMetadata -> {
- if (chunkMetadata instanceof AbstractAlignedChunkMetadata) {
- AbstractAlignedChunkMetadata alignedChunkMetadata =
- (AbstractAlignedChunkMetadata) chunkMetadata;
- for (int i = 0; i < alignedChunkMetadata.getValueChunkMetadataList().size(); i++) {
- if ((alignedChunkMetadata.getValueChunkMetadataList().get(i) != null)
- && !SchemaUtils.isUsingSameStatistics(
- alignedChunkMetadata.getValueChunkMetadataList().get(i).getDataType(),
- getTsDataTypeList().get(i))
- && !SchemaUtils.canUseStatisticsAfterAlter(getTsDataTypeList().get(i))) {
- alignedChunkMetadata.getValueChunkMetadataList().get(i).setModified(true);
- }
- }
- chunkMetadata = alignedChunkMetadata;
- } else if (chunkMetadata instanceof ChunkMetadata) {
- if (!SchemaUtils.isUsingSameStatistics(
- chunkMetadata.getDataType(), getTsDataTypeList().get(0))
- && !SchemaUtils.canUseStatisticsAfterAlter(getTsDataTypeList().get(0))) {
- chunkMetadata.setModified(true);
- }
- }
- chunkMetadata.setSeq(timeSeriesMetadata.isSeq());
- });
+ chunkMetadataList.forEach(chunkMetadata -> chunkMetadata.setSeq(timeSeriesMetadata.isSeq()));
cachedChunkMetadata.addAll(chunkMetadataList);
}
@@ -704,7 +661,7 @@ public class SeriesScanUtil implements Accountable {
}
List<IPageReader> pageReaderList =
FileLoaderUtils.loadPageReaderList(
- chunkMetaData, scanOptions.getGlobalTimeFilter(), isAligned, getTsDataTypeList());
+ chunkMetaData, scanOptions.getGlobalTimeFilter(), getTsDataTypeList());
// init TsBlockBuilder for each page reader
pageReaderList.forEach(p -> p.initTsBlockBuilder(getTsDataTypeList()));
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java
index 690cea9f03b..adb76d9bbdd 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java
@@ -364,7 +364,7 @@ class AlignedResourceByPathUtils extends ResourceByPathUtils {
TSDataType targetDataType = measurementMap.get(measurement);
if (valueChunkMetadata.getDataType() != targetDataType) {
SchemaUtils.rewriteAlignedChunkMetadataStatistics(alignedChunkMetadata, i, targetDataType);
- alignedChunkMetadata.setModified(true);
+ alignedChunkMetadata.setDataTypeModifiedAndCannotUseStatistics(true);
}
}
}
@@ -571,7 +571,7 @@ class MeasurementResourceByPathUtils extends ResourceByPathUtils {
return null;
}
chunkMetadataList.set(index, chunkMetadata);
- chunkMetadata.setModified(true);
+ chunkMetadata.setDataTypeModifiedAndCannotUseStatistics(true);
}
if (!useFakeStatistics) {
if (chunkMetadata != null && targetDataType.isCompatible(chunkMetadata.getDataType())) {
@@ -597,7 +597,7 @@ class MeasurementResourceByPathUtils extends ResourceByPathUtils {
return null;
}
memChunk.setChunkMetadata(rewritedChunkMetadata);
- memChunk.getChunkMetaData().setModified(true);
+ memChunk.getChunkMetaData().setDataTypeModifiedAndCannotUseStatistics(true);
}
if (useFakeStatistics) {
memChunk.initChunkMetaFromTVListsWithFakeStatistics();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedChunkReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedChunkReader.java
index 093ae9cc327..c85dcef9ace 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedChunkReader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedChunkReader.java
@@ -85,6 +85,13 @@ public class MemAlignedChunkReader implements IChunkReader {
// Do nothing because mem chunk reader will not open files
}
+ @Override
+ public void markDataTypeModifiedAndCannotUseStatistics() {
+ for (IPageReader iPageReader : pageReaderList) {
+ iPageReader.setModified(true);
+ }
+ }
+
@Override
public List<IPageReader> loadPageReaderList() {
return this.pageReaderList;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedPageReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedPageReader.java
index a299684a49c..758721f9a05 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedPageReader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemAlignedPageReader.java
@@ -60,6 +60,9 @@ public class MemAlignedPageReader implements IPageReader {
private Filter recordFilter;
private PaginationController paginationController = UNLIMITED_PAGINATION_CONTROLLER;
+ // data type is modified in query and statistics cannot be used
+ private boolean modified;
+
private TsBlockBuilder builder;
public MemAlignedPageReader(
@@ -214,11 +217,13 @@ public class MemAlignedPageReader implements IPageReader {
@Override
public boolean isModified() {
- return false;
+ return modified;
}
@Override
- public void setModified(boolean modified) {}
+ public void setModified(boolean modified) {
+ this.modified = modified;
+ }
@Override
public void initTsBlockBuilder(List<TSDataType> dataTypes) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkReader.java
index 4b21d87d114..8cee8b970c5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkReader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemChunkReader.java
@@ -129,6 +129,13 @@ public class MemChunkReader implements IChunkReader, IPointReader {
// Do nothing because mem chunk reader will not open files
}
+ @Override
+ public void markDataTypeModifiedAndCannotUseStatistics() {
+ for (IPageReader pageReader : this.pageReaderList) {
+ pageReader.setModified(true);
+ }
+ }
+
@Override
public List<IPageReader> loadPageReaderList() {
return this.pageReaderList;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemPageReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemPageReader.java
index e7427ef87d8..0f656a96f69 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemPageReader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/MemPageReader.java
@@ -57,6 +57,9 @@ public class MemPageReader implements IPageReader {
private PaginationController paginationController = UNLIMITED_PAGINATION_CONTROLLER;
+ // data type is modified in query and statistics cannot be used
+ private boolean modified;
+
public MemPageReader(
Supplier<TsBlock> tsBlockSupplier,
int pageIndex,
@@ -223,11 +226,13 @@ public class MemPageReader implements IPageReader {
@Override
public boolean isModified() {
- return false;
+ return modified;
}
@Override
- public void setModified(boolean modified) {}
+ public void setModified(boolean modified) {
+ this.modified = modified;
+ }
@Override
public void initTsBlockBuilder(List<TSDataType> dataTypes) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
index 6044b2331d1..281b1be2b35 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/SchemaUtils.java
@@ -40,6 +40,8 @@ import org.apache.tsfile.utils.Binary;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
@@ -314,120 +316,58 @@ public class SchemaUtils {
dataTypeColumnClassMap.get(originalDataType), dataTypeColumnClassMap.get(dataType));
}
- public static boolean isUsingSameStatistics(TSDataType originalDataType, TSDataType dataType) {
- if (originalDataType == dataType) {
- return true;
- }
- if (!dataTypeColumnStatisticsClassMap.containsKey(originalDataType)
- || !dataTypeColumnStatisticsClassMap.containsKey(dataType)) {
- return false;
- }
- return Objects.equals(
- dataTypeColumnStatisticsClassMap.get(originalDataType),
- dataTypeColumnStatisticsClassMap.get(dataType));
- }
-
- public static boolean canUseStatisticsAfterAlter(TSDataType dataType) {
- return !canNotUseStatisticAfterAlterClassSet.contains(dataType);
- }
-
- public static void changeMetadataModified(
- TimeseriesMetadata timeseriesMetadata, TSDataType targetDataType) {
+ public static void changeTimeseriesMetadataModified(
+ @Nullable TimeseriesMetadata timeseriesMetadata, TSDataType targetDataType) {
if (timeseriesMetadata == null) {
return;
}
- if (!SchemaUtils.isUsingSameStatistics(timeseriesMetadata.getTsDataType(), targetDataType)
- && !SchemaUtils.canUseStatisticsAfterAlter(targetDataType)) {
- timeseriesMetadata.setModified(true);
- List<IChunkMetadata> chunkMetadataList = timeseriesMetadata.getChunkMetadataList();
- if (chunkMetadataList != null) {
- for (IChunkMetadata chunkMetadata : chunkMetadataList) {
- if (chunkMetadata != null) {
- chunkMetadata.setModified(true);
- }
- }
- }
+ if (!SchemaUtils.canUseStatisticsAfterAlter(
+ timeseriesMetadata.getTsDataType(), targetDataType)) {
+ timeseriesMetadata.setDataTypeModifiedAndCannotUseStatistics(true);
}
}
- public static void changeAlignedMetadataModified(
- AbstractAlignedTimeSeriesMetadata alignedTimeSeriesMetadata,
+ public static void changeAlignedTimeseriesMetadataModified(
+ @Nullable AbstractAlignedTimeSeriesMetadata alignedTimeSeriesMetadata,
List<TSDataType> targetDataTypeList) {
if (alignedTimeSeriesMetadata == null) {
return;
}
-
- int i = 0;
- for (TimeseriesMetadata timeseriesMetadata :
- alignedTimeSeriesMetadata.getValueTimeseriesMetadataList()) {
- if ((timeseriesMetadata != null)
- && !SchemaUtils.isUsingSameStatistics(
- timeseriesMetadata.getTsDataType(), targetDataTypeList.get(i))
- && !SchemaUtils.canUseStatisticsAfterAlter(targetDataTypeList.get(i))) {
- timeseriesMetadata.setModified(true);
- alignedTimeSeriesMetadata.setModified(true);
- List<IChunkMetadata> chunkMetadataList = timeseriesMetadata.getChunkMetadataList();
- if (chunkMetadataList != null) {
- for (IChunkMetadata chunkMetadata : chunkMetadataList) {
- if (chunkMetadata != null) {
- chunkMetadata.setModified(true);
- }
- }
- }
+ for (int i = 0; i < alignedTimeSeriesMetadata.getValueTimeseriesMetadataList().size(); i++) {
+ TimeseriesMetadata valueTimeseriesMetadata =
+ alignedTimeSeriesMetadata.getValueTimeseriesMetadataList().get(i);
+ if (valueTimeseriesMetadata != null
+ && !SchemaUtils.canUseStatisticsAfterAlter(
+ valueTimeseriesMetadata.getTsDataType(), targetDataTypeList.get(i))) {
+ alignedTimeSeriesMetadata.setDataTypeModifiedAndCannotUseStatistics(true);
+ return;
}
- i++;
}
}
- public static void changeAlignedMetadataModified(
- TimeseriesMetadata timeseriesMetadata, TSDataType targetDataType) {
- if (timeseriesMetadata == null) {
- return;
- }
-
- if (!SchemaUtils.isUsingSameStatistics(timeseriesMetadata.getTsDataType(), targetDataType)
- && !SchemaUtils.canUseStatisticsAfterAlter(targetDataType)) {
- timeseriesMetadata.setModified(true);
- List<IChunkMetadata> chunkMetadataList = timeseriesMetadata.getChunkMetadataList();
- if (chunkMetadataList != null) {
- for (IChunkMetadata chunkMetadata : chunkMetadataList) {
- if (chunkMetadata != null) {
- chunkMetadata.setModified(true);
- }
- }
- }
+ public static boolean canUseStatisticsAfterAlter(
+ TSDataType originalDataType, TSDataType targetDataType) {
+ if (isUsingSameStatistics(originalDataType, targetDataType)) {
+ return true;
}
+ return canUseStatisticsAfterAlter(targetDataType);
}
- public static void changeMetadataModified(
- IChunkMetadata chunkMetadata, TSDataType sourceDataType, TSDataType targetDataType) {
- if (chunkMetadata == null) {
- return;
+ private static boolean isUsingSameStatistics(TSDataType originalDataType, TSDataType dataType) {
+ if (originalDataType == dataType) {
+ return true;
}
- if (!SchemaUtils.isUsingSameStatistics(sourceDataType, targetDataType)
- && !SchemaUtils.canUseStatisticsAfterAlter(targetDataType)) {
- chunkMetadata.setModified(true);
+ if (!dataTypeColumnStatisticsClassMap.containsKey(originalDataType)
+ || !dataTypeColumnStatisticsClassMap.containsKey(dataType)) {
+ return false;
}
+ return Objects.equals(
+ dataTypeColumnStatisticsClassMap.get(originalDataType),
+ dataTypeColumnStatisticsClassMap.get(dataType));
}
- public static void changeAlignedMetadataModified(
- AbstractAlignedChunkMetadata chunkMetadata,
- TSDataType sourceDataType,
- List<TSDataType> targetDataTypeList) {
- if (chunkMetadata == null) {
- return;
- }
- int i = 0;
- for (IChunkMetadata iChunkMetadata : chunkMetadata.getValueChunkMetadataList()) {
- if ((iChunkMetadata != null)
- && !SchemaUtils.isUsingSameStatistics(
- iChunkMetadata.getDataType(), targetDataTypeList.get(i))
- && !SchemaUtils.canUseStatisticsAfterAlter(targetDataTypeList.get(i))) {
- iChunkMetadata.setModified(true);
- chunkMetadata.setModified(true);
- }
- i++;
- }
+ private static boolean canUseStatisticsAfterAlter(TSDataType dataType) {
+ return !canNotUseStatisticAfterAlterClassSet.contains(dataType);
}
public static void rewriteAlignedChunkMetadataStatistics(
diff --git a/pom.xml b/pom.xml
index 240202f34c1..387ab73ece6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -173,7 +173,7 @@
<thrift.version>0.14.1</thrift.version>
<xz.version>1.9</xz.version>
<zstd-jni.version>1.5.6-3</zstd-jni.version>
- <tsfile.version>2.2.1-260203-SNAPSHOT</tsfile.version>
+ <tsfile.version>2.2.1-260206-SNAPSHOT</tsfile.version>
</properties>
<!--
if we claim dependencies in dependencyManagement, then we do not claim
(iotdb) 08/16: Pipe: Banned the illegal names in pipe and pipe plugins (#17145)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e4c0582bb1af97b87bd3800487bf1283abb147d0
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Wed Feb 4 09:38:44 2026 +0800
Pipe: Banned the illegal names in pipe and pipe plugins (#17145)
* fix
* fix
* fix
* sp-linux
---
.../treemodel/auto/basic/IoTDBPipeSyntaxIT.java | 64 ++++++++++++++++++++++
.../legacy/IoTDBLegacyPipeReceiverAgent.java | 5 +-
.../config/executor/ClusterConfigTaskExecutor.java | 34 +++++++++---
.../org/apache/iotdb/commons/utils/FileUtils.java | 12 ++++
4 files changed, 107 insertions(+), 8 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSyntaxIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSyntaxIT.java
index 30954d71e98..844477b6590 100644
--- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSyntaxIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeSyntaxIT.java
@@ -31,6 +31,7 @@ import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeAutoBasic;
import org.apache.iotdb.pipe.it.dual.treemodel.auto.AbstractPipeDualTreeModelAutoIT;
import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.tsfile.external.commons.lang3.SystemUtils;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -41,6 +42,7 @@ import java.io.File;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -336,6 +338,68 @@ public class IoTDBPipeSyntaxIT extends AbstractPipeDualTreeModelAutoIT {
}
}
+ @Test
+ public void testDirectoryErrors() throws SQLException {
+ try (final Connection connection = senderEnv.getConnection();
+ final Statement statement = connection.createStatement()) {
+ List<String> wrongDirs = Arrays.asList(".", "..", "./hackYou", ".\\hackYouTwice");
+ if (SystemUtils.IS_OS_WINDOWS) {
+ wrongDirs = new ArrayList<>(wrongDirs);
+ wrongDirs.add("BombWindows/:*?");
+ wrongDirs.add("AUX");
+ }
+ for (final String name : wrongDirs) {
+ testDirectoryError(name, statement);
+ }
+ }
+ }
+
+ private void testDirectoryError(final String wrongDir, final Statement statement) {
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+
+ final String receiverIp = receiverDataNode.getIp();
+ final int receiverPort = receiverDataNode.getPort();
+
+ try {
+ statement.execute(
+ String.format(
+ "create pipe `"
+ + wrongDir
+ + "` with source ()"
+ + " with processor ()"
+ + " with sink ("
+ + "'sink'='invalid-param',"
+ + "'sink.ip'='%s',"
+ + "'sink.port'='%s',"
+ + "'sink.batch.enable'='false')",
+ receiverIp,
+ receiverPort));
+ fail();
+ } catch (final Exception ignore) {
+ // Expected
+ }
+
+ try {
+ statement.execute(
+ String.format(
+ "create pipePlugin `"
+ + wrongDir
+ + "` as 'org.apache.iotdb.db.pipe.example.TestProcessor' USING URI '%s'",
+ new File(
+ System.getProperty("user.dir")
+ + File.separator
+ + "target"
+ + File.separator
+ + "test-classes"
+ + File.separator)
+ .toURI()
+ + "PipePlugin.jar"));
+ fail();
+ } catch (final SQLException e) {
+ Assert.assertTrue(e.getMessage().contains("1600: Failed to create pipe plugin"));
+ }
+ }
+
@Test
public void testBrackets() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
index 70922d4022b..2c45c42a455 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.commons.audit.UserEntity;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.sink.payload.legacy.PipeData;
@@ -53,6 +54,7 @@ import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.time.ZoneId;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
@@ -125,7 +127,8 @@ public class IoTDBLegacyPipeReceiverAgent {
}
private boolean validatePipeName(final TSyncIdentityInfo info) {
- return info.isSetPipeName() && !info.getPipeName().contains(File.separator);
+ return info.isSetPipeName()
+ && Objects.isNull(FileUtils.getIllegalError4Directory(info.getPipeName()));
}
private void createConnection(final SyncIdentityInfo identityInfo) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index b241709898b..3627e3f82af 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -91,6 +91,7 @@ import org.apache.iotdb.commons.trigger.service.TriggerExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFClassLoader;
import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
import org.apache.iotdb.commons.utils.CommonDateTimeUtils;
+import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.commons.utils.PathUtils;
import org.apache.iotdb.commons.utils.SerializeUtils;
import org.apache.iotdb.commons.utils.TimePartitionUtils;
@@ -988,6 +989,15 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
final String className = createPipePluginStatement.getClassName();
final String uriString = createPipePluginStatement.getUriString();
+ final String pathError = FileUtils.getIllegalError4Directory(pluginName);
+ if (Objects.nonNull(pathError)) {
+ future.setException(
+ new IoTDBException(
+ String.format("Failed to create pipe plugin %s. " + pathError, pluginName),
+ TSStatusCode.CREATE_PIPE_PLUGIN_ERROR.getStatusCode()));
+ return future;
+ }
+
if (uriString == null || uriString.isEmpty()) {
future.setException(
new IoTDBException(
@@ -2133,6 +2143,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
// Verify that Pipe is disabled if TSFile encryption is enabled
+ final String pipeName = createPipeStatement.getPipeName();
if (!Objects.equals(TSFileDescriptor.getInstance().getConfig().getEncryptType(), "UNENCRYPTED")
&& !Objects.equals(
TSFileDescriptor.getInstance().getConfig().getEncryptType(),
@@ -2141,18 +2152,27 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
new IoTDBException(
String.format(
"Failed to create Pipe %s because TSFile is configured with encryption, which prohibits the use of Pipe",
- createPipeStatement.getPipeName()),
+ pipeName),
TSStatusCode.PIPE_ERROR.getStatusCode()));
return future;
}
// Validate pipe name
- if (createPipeStatement.getPipeName().startsWith(PipeStaticMeta.SYSTEM_PIPE_PREFIX)) {
+ if (pipeName.startsWith(PipeStaticMeta.SYSTEM_PIPE_PREFIX)) {
future.setException(
new IoTDBException(
String.format(
"Failed to create pipe %s, pipe name starting with \"%s\" are not allowed to be created.",
- createPipeStatement.getPipeName(), PipeStaticMeta.SYSTEM_PIPE_PREFIX),
+ pipeName, PipeStaticMeta.SYSTEM_PIPE_PREFIX),
+ TSStatusCode.PIPE_ERROR.getStatusCode()));
+ return future;
+ }
+
+ final String pathError = FileUtils.getIllegalError4Directory(pipeName);
+ if (Objects.nonNull(pathError)) {
+ future.setException(
+ new IoTDBException(
+ String.format("Failed to create pipe %s, " + pathError, pipeName),
TSStatusCode.PIPE_ERROR.getStatusCode()));
return future;
}
@@ -2161,7 +2181,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
try {
PipeDataNodeAgent.plugin()
.validate(
- createPipeStatement.getPipeName(),
+ pipeName,
createPipeStatement.getSourceAttributes(),
createPipeStatement.getProcessorAttributes(),
createPipeStatement.getSinkAttributes());
@@ -2183,7 +2203,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
final TCreatePipeReq realtimeReq =
new TCreatePipeReq()
// Append suffix to the pipeline name for real-time data
- .setPipeName(createPipeStatement.getPipeName() + "_realtime")
+ .setPipeName(pipeName + "_realtime")
// NOTE: set if not exists always to true to handle partial failure
.setIfNotExistsCondition(true)
// Use extractor parameters for real-time data
@@ -2211,7 +2231,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
final TCreatePipeReq historyReq =
new TCreatePipeReq()
// Append suffix to the pipeline name for historical data
- .setPipeName(createPipeStatement.getPipeName() + "_history")
+ .setPipeName(pipeName + "_history")
.setIfNotExistsCondition(createPipeStatement.hasIfNotExistsCondition())
// Use source parameters for historical data
.setExtractorAttributes(
@@ -2254,7 +2274,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor {
CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
final TCreatePipeReq req =
new TCreatePipeReq()
- .setPipeName(createPipeStatement.getPipeName())
+ .setPipeName(pipeName)
.setIfNotExistsCondition(createPipeStatement.hasIfNotExistsCondition())
.setExtractorAttributes(createPipeStatement.getSourceAttributes())
.setProcessorAttributes(createPipeStatement.getProcessorAttributes())
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
index baed8bcd537..595cd5e831c 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
@@ -57,6 +57,8 @@ public class FileUtils {
"Renamed file {} to {} because it already exists in the target directory: {}";
private static final String COPY_FILE_MESSAGE =
"Copy file {} to {} because it already exists in the target directory: {}";
+ private static final String ILLEGAL_PATH_MESSAGE =
+ "The path cannot be '.', '..', './' or '.\\'. ";
private FileUtils() {}
@@ -579,4 +581,14 @@ public class FileUtils {
targetFile,
targetFile.getParentFile().getAbsolutePath());
}
+
+ public static String getIllegalError4Directory(final String path) {
+ if (path.equals(".") || path.equals("..") || path.contains("./") || path.contains(".\\")) {
+ return ILLEGAL_PATH_MESSAGE;
+ }
+ if (!WindowsOSUtils.isLegalPathSegment4Windows(path)) {
+ return WindowsOSUtils.OS_SEGMENT_ERROR;
+ }
+ return null;
+ }
}
(iotdb) 12/16: fix: Fix missing encoding and compression information for Measurement (#17141)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b4072594031c7039a585103f863195ad0194441c
Author: Zhenyu Luo <zh...@apache.org>
AuthorDate: Thu Feb 5 12:28:34 2026 +0800
fix: Fix missing encoding and compression information for Measurement (#17141)
---
.../tsfile/parser/scan/TsFileInsertionEventScanParser.java | 12 ++++++++++--
1 file changed, 10 insertions(+), 2 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
index 99d35f7eb98..5dc6228f1d1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java
@@ -512,7 +512,11 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
: new SinglePageWholeChunkReader(chunk);
currentIsAligned = false;
currentMeasurements.add(
- new MeasurementSchema(chunkHeader.getMeasurementID(), chunkHeader.getDataType()));
+ new MeasurementSchema(
+ chunkHeader.getMeasurementID(),
+ chunkHeader.getDataType(),
+ chunkHeader.getEncodingType(),
+ chunkHeader.getCompressionType()));
modsInfos.addAll(
ModsOperationUtil.initializeMeasurementMods(
currentDevice,
@@ -582,7 +586,11 @@ public class TsFileInsertionEventScanParser extends TsFileInsertionEventParser {
valueChunkSize += chunkHeader.getDataSize();
valueChunkList.add(chunk);
currentMeasurements.add(
- new MeasurementSchema(chunkHeader.getMeasurementID(), chunkHeader.getDataType()));
+ new MeasurementSchema(
+ chunkHeader.getMeasurementID(),
+ chunkHeader.getDataType(),
+ chunkHeader.getEncodingType(),
+ chunkHeader.getCompressionType()));
modsInfos.addAll(
ModsOperationUtil.initializeMeasurementMods(
currentDevice,
(iotdb) 06/16: Fix occasional CI failures of IoTDBMQTTServiceJsonIT. (#17146)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rc/2.0.8
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5e961585986711e44cae9cc364010f66d97a65ff
Author: wenyanshi-123 <34...@qq.com>
AuthorDate: Tue Feb 3 19:47:03 2026 +0800
Fix occasional CI failures of IoTDBMQTTServiceJsonIT. (#17146)
---
.../iotdb/db/it/mqtt/IoTDBMQTTServiceJsonIT.java | 132 +++++++++++++++++----
1 file changed, 110 insertions(+), 22 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/mqtt/IoTDBMQTTServiceJsonIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/mqtt/IoTDBMQTTServiceJsonIT.java
index b9fdb5e1980..91031c0202d 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/mqtt/IoTDBMQTTServiceJsonIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/mqtt/IoTDBMQTTServiceJsonIT.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.it.mqtt;
import org.apache.iotdb.isession.ISession;
+import org.apache.iotdb.isession.SessionConfig;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
@@ -58,8 +59,9 @@ public class IoTDBMQTTServiceJsonIT {
private BlockingConnection connection;
private static final String IP = System.getProperty("RemoteIp", "127.0.0.1");
- private static final String USER = System.getProperty("RemoteUser", "root");
- private static final String PASSWORD = System.getProperty("RemotePassword", "root");
+ private static final String USER = System.getProperty("RemoteUser", SessionConfig.DEFAULT_USER);
+ private static final String PASSWORD =
+ System.getProperty("RemotePassword", SessionConfig.DEFAULT_PASSWORD);
public static final String FORMATTER = "json";
@Before
@@ -85,7 +87,7 @@ public class IoTDBMQTTServiceJsonIT {
@After
public void tearDown() throws Exception {
try {
- if (connection != null) {
+ if (connection != null && connection.isConnected()) {
connection.disconnect();
}
} catch (IOException e) {
@@ -120,9 +122,13 @@ public class IoTDBMQTTServiceJsonIT {
}
RowRecord row = dataSet.next();
List<Field> fields = row.getFields();
- assertEquals(2, fields.size());
- assertEquals(1.5, fields.get(0).getDoubleV(), 0.001);
- assertEquals(2.5, fields.get(1).getDoubleV(), 0.001);
+ if (fields.size() != 2) {
+ return false;
+ }
+ if (Math.abs(fields.get(0).getDoubleV() - 1.5) > 0.001
+ || Math.abs(fields.get(1).getDoubleV() - 2.5) > 0.001) {
+ return false;
+ }
return true;
} catch (StatementExecutionException e) {
if (e.getMessage() != null && e.getMessage().contains("does not exist")) {
@@ -132,6 +138,14 @@ public class IoTDBMQTTServiceJsonIT {
}
}
});
+ try (final SessionDataSet dataSet =
+ session.executeQueryStatement("select s1, s2 from root.sg.d1 where time = 1")) {
+ assertTrue(dataSet.hasNext());
+ List<Field> fields = dataSet.next().getFields();
+ assertEquals(2, fields.size());
+ assertEquals(1.5, fields.get(0).getDoubleV(), 0.001);
+ assertEquals(2.5, fields.get(1).getDoubleV(), 0.001);
+ }
}
}
@@ -159,8 +173,11 @@ public class IoTDBMQTTServiceJsonIT {
return false;
}
RowRecord row = dataSet.next();
- // Should have 3 records
- assertEquals(3, row.getFields().get(0).getLongV());
+ long count = row.getFields().get(0).getLongV();
+ // Retry when data not visible yet (count 0) or incomplete
+ if (count != 3) {
+ return false;
+ }
return true;
} catch (StatementExecutionException e) {
if (e.getMessage() != null && e.getMessage().contains("does not exist")) {
@@ -170,6 +187,12 @@ public class IoTDBMQTTServiceJsonIT {
}
}
});
+ // Assert after await so failure message is clear on timeout
+ try (final SessionDataSet dataSet =
+ session.executeQueryStatement("select count(s1) from root.sg.d2")) {
+ assertTrue(dataSet.hasNext());
+ assertEquals(3, dataSet.next().getFields().get(0).getLongV());
+ }
}
}
@@ -203,7 +226,9 @@ public class IoTDBMQTTServiceJsonIT {
return false;
}
// sum should be 10 + 20 + 30 = 60
- assertEquals(60.0, sum, 0.001);
+ if (Math.abs(sum - 60.0) > 0.001) {
+ return false;
+ }
return true;
} catch (StatementExecutionException e) {
if (e.getMessage() != null && e.getMessage().contains("does not exist")) {
@@ -213,6 +238,17 @@ public class IoTDBMQTTServiceJsonIT {
}
}
});
+ try (final SessionDataSet dataSet =
+ session.executeQueryStatement("select s1 from root.sg.d3")) {
+ int count = 0;
+ double sum = 0;
+ while (dataSet.hasNext()) {
+ sum += dataSet.next().getFields().get(0).getDoubleV();
+ count++;
+ }
+ assertEquals(3, count);
+ assertEquals(60.0, sum, 0.001);
+ }
}
}
@@ -242,11 +278,15 @@ public class IoTDBMQTTServiceJsonIT {
return false;
}
List<Field> fields = dataSet.next().getFields();
- assertEquals(4, fields.size());
- assertEquals(100, fields.get(0).getIntV());
- assertEquals(3.14f, fields.get(1).getFloatV(), 0.01);
- assertTrue(fields.get(2).getBoolV());
- assertEquals("hello", fields.get(3).getStringValue());
+ if (fields.size() != 4) {
+ return false;
+ }
+ if (fields.get(0).getIntV() != 100
+ || Math.abs(fields.get(1).getFloatV() - 3.14f) > 0.01
+ || !fields.get(2).getBoolV()
+ || !"hello".equals(fields.get(3).getStringValue())) {
+ return false;
+ }
return true;
} catch (StatementExecutionException e) {
if (e.getMessage() != null && e.getMessage().contains("does not exist")) {
@@ -256,6 +296,17 @@ public class IoTDBMQTTServiceJsonIT {
}
}
});
+ try (final SessionDataSet dataSet =
+ session.executeQueryStatement(
+ "select intVal, floatVal, boolVal, textVal from root.sg.d4 where time = 1")) {
+ assertTrue(dataSet.hasNext());
+ List<Field> fields = dataSet.next().getFields();
+ assertEquals(4, fields.size());
+ assertEquals(100, fields.get(0).getIntV());
+ assertEquals(3.14f, fields.get(1).getFloatV(), 0.01);
+ assertTrue(fields.get(2).getBoolV());
+ assertEquals("hello", fields.get(3).getStringValue());
+ }
}
}
@@ -284,7 +335,9 @@ public class IoTDBMQTTServiceJsonIT {
if (!dataSet1.hasNext()) {
return false;
}
- assertEquals(25.5, dataSet1.next().getFields().get(0).getDoubleV(), 0.001);
+ if (Math.abs(dataSet1.next().getFields().get(0).getDoubleV() - 25.5) > 0.001) {
+ return false;
+ }
}
// Check device2
try (final SessionDataSet dataSet2 =
@@ -293,7 +346,9 @@ public class IoTDBMQTTServiceJsonIT {
if (!dataSet2.hasNext()) {
return false;
}
- assertEquals(26.5, dataSet2.next().getFields().get(0).getDoubleV(), 0.001);
+ if (Math.abs(dataSet2.next().getFields().get(0).getDoubleV() - 26.5) > 0.001) {
+ return false;
+ }
}
// Check device3
try (final SessionDataSet dataSet3 =
@@ -302,7 +357,9 @@ public class IoTDBMQTTServiceJsonIT {
if (!dataSet3.hasNext()) {
return false;
}
- assertEquals(27.5, dataSet3.next().getFields().get(0).getDoubleV(), 0.001);
+ if (Math.abs(dataSet3.next().getFields().get(0).getDoubleV() - 27.5) > 0.001) {
+ return false;
+ }
}
return true;
} catch (StatementExecutionException e) {
@@ -313,6 +370,21 @@ public class IoTDBMQTTServiceJsonIT {
}
}
});
+ try (final SessionDataSet dataSet1 =
+ session.executeQueryStatement("select temp from root.sg.device1 where time = 1")) {
+ assertTrue(dataSet1.hasNext());
+ assertEquals(25.5, dataSet1.next().getFields().get(0).getDoubleV(), 0.001);
+ }
+ try (final SessionDataSet dataSet2 =
+ session.executeQueryStatement("select temp from root.sg.device2 where time = 1")) {
+ assertTrue(dataSet2.hasNext());
+ assertEquals(26.5, dataSet2.next().getFields().get(0).getDoubleV(), 0.001);
+ }
+ try (final SessionDataSet dataSet3 =
+ session.executeQueryStatement("select temp from root.sg.device3 where time = 1")) {
+ assertTrue(dataSet3.hasNext());
+ assertEquals(27.5, dataSet3.next().getFields().get(0).getDoubleV(), 0.001);
+ }
}
}
@@ -340,13 +412,14 @@ public class IoTDBMQTTServiceJsonIT {
while (dataSet.hasNext()) {
RowRecord row = dataSet.next();
List<Field> fields = row.getFields();
- assertEquals(2, fields.size());
- // Temperature should be between 20 and 25
+ if (fields.size() != 2) {
+ return false;
+ }
double temp = fields.get(0).getDoubleV();
- assertTrue(temp >= 20.0 && temp <= 25.0);
- // Humidity should be between 60 and 65
double humidity = fields.get(1).getDoubleV();
- assertTrue(humidity >= 60.0 && humidity <= 65.0);
+ if (temp < 20.0 || temp > 25.0 || humidity < 60.0 || humidity > 65.0) {
+ return false;
+ }
count++;
}
return count == 5;
@@ -358,6 +431,21 @@ public class IoTDBMQTTServiceJsonIT {
}
}
});
+ try (final SessionDataSet dataSet =
+ session.executeQueryStatement("select temperature, humidity from root.sg.d5")) {
+ int count = 0;
+ while (dataSet.hasNext()) {
+ RowRecord row = dataSet.next();
+ List<Field> fields = row.getFields();
+ assertEquals(2, fields.size());
+ double temp = fields.get(0).getDoubleV();
+ double humidity = fields.get(1).getDoubleV();
+ assertTrue(temp >= 20.0 && temp <= 25.0);
+ assertTrue(humidity >= 60.0 && humidity <= 65.0);
+ count++;
+ }
+ assertEquals(5, count);
+ }
}
}
}