This is an automated email from the ASF dual-hosted git repository. justinchen pushed a commit to branch c-ger-p in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 177787a269ba67b413f4f780297ce9b69a79e339 Author: Caideyipi <[email protected]> AuthorDate: Wed Apr 15 17:14:26 2026 +0800 Enhance the last query permission && Fixed the rollback version of alter view / table plans && Deleted the unnecessary mods in Tree view deletion (#17465) * fix * fix * fix * fix * IT1 * change * right-1 * fixation * fix * partial * 13 * limit * fix * fix * mod * remove-usls * fix * fix * fix --- .../org/apache/iotdb/db/it/IoTDBRestServiceIT.java | 160 +++++++++++++ .../org/apache/iotdb/db/it/utils/TestUtils.java | 26 ++ .../iotdb/session/it/IoTDBSessionQueryIT.java | 77 ++++++ .../schema/table/RenameTableColumnProcedure.java | 247 +++++++++++++++++++ .../schema/table/SetTablePropertiesProcedure.java | 264 +++++++++++++++++++++ .../protocol/rest/v2/impl/RestApiServiceImpl.java | 31 ++- .../protocol/thrift/impl/ClientRPCServiceImpl.java | 84 ++++--- .../plan/parser/StatementGenerator.java | 2 +- .../schemaengine/schemaregion/ISchemaRegion.java | 2 +- .../schemaregion/impl/SchemaRegionMemoryImpl.java | 4 +- .../schemaregion/impl/SchemaRegionPBTreeImpl.java | 2 +- .../mtree/impl/mem/MTreeBelowSGMemoryImpl.java | 5 +- 12 files changed, 865 insertions(+), 39 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestServiceIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestServiceIT.java index ad1b93ad2d8..e8eca4937d5 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestServiceIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestServiceIT.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.it; import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant; +import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.it.env.EnvFactory; import org.apache.iotdb.it.env.cluster.env.SimpleEnv; import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; @@ -57,6 +58,7 @@ import java.sql.SQLException; import java.sql.Statement; import java.util.ArrayList; import java.util.Base64; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -513,6 +515,7 @@ public class IoTDBRestServiceIT { selectLast(httpClient); queryV2(httpClient); + selectFastLast(httpClient); queryGroupByLevelV2(httpClient); queryRowLimitV2(httpClient); queryShowChildPathsV2(httpClient); @@ -923,6 +926,71 @@ public class IoTDBRestServiceIT { } } + @Test + public void queryFastLastWithWrongAuthorization() { + CloseableHttpResponse response = null; + + TestUtils.executeNonQuery("create user abcd 'strongPassword@1234'"); + try { + final CloseableHttpClient httpClient = HttpClientBuilder.create().build(); + final HttpPost httpPost = new HttpPost("http://127.0.0.1:"; + port + "/rest/v2/fastLastQuery"); + httpPost.addHeader("Content-type", "application/json; charset=utf-8"); + httpPost.setHeader("Accept", "application/json"); + final String authorization = getAuthorization("abcd", "strongPassword@1234"); + httpPost.setHeader("Authorization", authorization); + final String sql = "{\"prefix_paths\":[\"root\",\"sg25\"]}"; + httpPost.setEntity(new StringEntity(sql, Charset.defaultCharset())); + for (int i = 0; i < 30; i++) { + try { + response = httpClient.execute(httpPost); + break; + } catch (Exception e) { + if (i == 29) { + throw e; + } + try { + Thread.sleep(1000); + } catch (InterruptedException ex) { + throw new RuntimeException(ex); + } + } + } + + Assert.assertEquals(200, response.getStatusLine().getStatusCode()); + String message = EntityUtils.toString(response.getEntity(), "utf-8"); + ObjectMapper mapper = new ObjectMapper(); + Map map = mapper.readValue(message, Map.class); + List<Long> timestampsResult = (List<Long>) map.get("timestamps"); + List<Long> expressionsResult = (List<Long>) map.get("expressions"); + List<List<Object>> valuesResult = (List<List<Object>>) map.get("values"); + Assert.assertTrue(map.size() > 0); + List<Object> expressions = + new ArrayList<Object>() { + { + add("Timeseries"); + add("Value"); + add("DataType"); + } + }; + + Assert.assertEquals(expressions, expressionsResult); + Assert.assertEquals(Collections.emptyList(), timestampsResult); + Assert.assertEquals(Collections.emptyList(), valuesResult); + } catch (IOException e) { + e.printStackTrace(); + fail(e.getMessage()); + } finally { + try { + if (response != null) { + response.close(); + } + } catch (IOException e) { + e.printStackTrace(); + fail(e.getMessage()); + } + } + } + public void query(CloseableHttpClient httpClient) { CloseableHttpResponse response = null; try { @@ -1677,6 +1745,98 @@ public class IoTDBRestServiceIT { } } + public void selectFastLast(CloseableHttpClient httpClient) { + // Only used in 1D scenarios + if (EnvFactory.getEnv().getDataNodeWrapperList().size() > 1) { + return; + } + CloseableHttpResponse response = null; + try { + HttpPost httpPost = getHttpPost("http://127.0.0.1:"; + port + "/rest/v2/fastLastQuery"); + String sql = "{\"prefix_paths\":[\"root\",\"sg25\"]}"; + httpPost.setEntity(new StringEntity(sql, Charset.defaultCharset())); + response = httpClient.execute(httpPost); + HttpEntity responseEntity = response.getEntity(); + String message = EntityUtils.toString(responseEntity, "utf-8"); + ObjectMapper mapper = new ObjectMapper(); + Map map = mapper.readValue(message, Map.class); + List<Long> timestampsResult = (List<Long>) map.get("timestamps"); + List<Long> expressionsResult = (List<Long>) map.get("expressions"); + List<List<Object>> valuesResult = (List<List<Object>>) map.get("values"); + Assert.assertTrue(map.size() > 0); + List<Object> expressions = + new ArrayList<Object>() { + { + add("Timeseries"); + add("Value"); + add("DataType"); + } + }; + List<Object> timestamps = + new ArrayList<Object>() { + { + add(1635232153960l); + add(1635232153960l); + add(1635232153960l); + add(1635232143960l); + add(1635232153960l); + add(1635232153960l); + } + }; + List<Object> values1 = + new ArrayList<Object>() { + { + add("root.sg25.s3"); + add("root.sg25.s4"); + add("root.sg25.s5"); + add("root.sg25.s6"); + add("root.sg25.s7"); + add("root.sg25.s8"); + } + }; + List<Object> values2 = + new ArrayList<Object>() { + { + add(""); + add("2"); + add("1635000012345556"); + add("1.41"); + add("false"); + add("3.5555"); + } + }; + List<Object> values3 = + new ArrayList<Object>() { + { + add("TEXT"); + add("INT32"); + add("INT64"); + add("FLOAT"); + add("BOOLEAN"); + add("DOUBLE"); + } + }; + + Assert.assertEquals(expressions, expressionsResult); + Assert.assertEquals(timestamps, timestampsResult); + Assert.assertEquals(values1, valuesResult.get(0)); + Assert.assertEquals(values2, valuesResult.get(1)); + Assert.assertEquals(values3, valuesResult.get(2)); + } catch (IOException e) { + e.printStackTrace(); + fail(e.getMessage()); + } finally { + try { + if (response != null) { + response.close(); + } + } catch (IOException e) { + e.printStackTrace(); + fail(e.getMessage()); + } + } + } + public void queryGroupByLevelV2(CloseableHttpClient httpClient) { CloseableHttpResponse response = null; try { diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java index abf59367082..f45209c7888 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java @@ -684,6 +684,32 @@ public class TestUtils { } } + public static void assertResultSetEqual( + SessionDataSet actualResultSet, + List<String> expectedColumnNames, + Set<String> expectedRetSet, + boolean ignoreTimeStamp) { + final Set<String> copiedSet = new HashSet<>(expectedRetSet); + try { + List<String> actualColumnNames = actualResultSet.getColumnNames(); + if (ignoreTimeStamp) { + assertEquals(expectedColumnNames, actualColumnNames); + } else { + assertEquals(TIMESTAMP_STR, actualColumnNames.get(0)); + assertEquals(expectedColumnNames, actualColumnNames.subList(1, actualColumnNames.size())); + } + + while (actualResultSet.hasNext()) { + RowRecord rowRecord = actualResultSet.next(); + assertTrue(copiedSet.remove(rowRecord.toString().replace('\t', ','))); + } + assertEquals(0, copiedSet.size()); + } catch (IoTDBConnectionException | StatementExecutionException e) { + e.printStackTrace(); + fail(e.getMessage()); + } + } + public static void createUser(String userName, String password) { createUser(EnvFactory.getEnv(), userName, password); } diff --git a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionQueryIT.java b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionQueryIT.java index ba5c22ab996..8ba5fcc6170 100644 --- a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionQueryIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionQueryIT.java @@ -22,6 +22,7 @@ package org.apache.iotdb.session.it; import org.apache.iotdb.common.rpc.thrift.TAggregationType; import org.apache.iotdb.commons.conf.IoTDBConstant; import org.apache.iotdb.db.it.utils.AlignedWriteUtil; +import org.apache.iotdb.db.it.utils.TestUtils; import org.apache.iotdb.isession.ISession; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.it.env.EnvFactory; @@ -29,6 +30,7 @@ import org.apache.iotdb.it.framework.IoTDBTestRunner; 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.RedirectException; import org.apache.iotdb.rpc.StatementExecutionException; import org.junit.AfterClass; @@ -244,6 +246,81 @@ public class IoTDBSessionQueryIT { } } + @Test + public void lastQueryWithPrefixTest() throws IoTDBConnectionException { + // Only used in 1D scenarios + if (EnvFactory.getEnv().getDataNodeWrapperList().size() > 1) { + return; + } + final Set<String> retArray = + new HashSet<>( + Arrays.asList( + "30,root.sg1.d1.s3,30,INT64", + "30,root.sg1.d1.s4,false,BOOLEAN", + "40,root.sg1.d1.s5,aligned_test40,TEXT", + "23,root.sg1.d1.s1,230000.0,FLOAT", + "40,root.sg1.d1.s2,40,INT32")); + + try (final ISession session = EnvFactory.getEnv().getSessionConnection()) { + // Push last cache first + try (final SessionDataSet resultSet = + session.executeFastLastDataQueryForOnePrefixPath(Arrays.asList("root", "sg1", "d1"))) { + assertResultSetEqual(resultSet, lastQueryColumnNames, retArray, true); + } + + try (final SessionDataSet resultSet = + session.executeFastLastDataQueryForOnePrefixPath(Arrays.asList("root", "sg1", "d1"))) { + assertResultSetEqual(resultSet, lastQueryColumnNames, retArray, true); + } + } catch (StatementExecutionException | RedirectException e) { + e.printStackTrace(); + fail(e.getMessage()); + } + } + + @Test + public void lastQueryWithoutPermissionTest() throws IoTDBConnectionException { + // Only used in 1D scenarios + if (EnvFactory.getEnv().getDataNodeWrapperList().size() > 1) { + return; + } + final String[] retArray = new String[] {}; + final Set<String> retArray2 = + new HashSet<>( + Arrays.asList( + "30,root.sg1.d1.s3,30,INT64", + "30,root.sg1.d1.s4,false,BOOLEAN", + "40,root.sg1.d1.s5,aligned_test40,TEXT", + "23,root.sg1.d1.s1,230000.0,FLOAT", + "40,root.sg1.d1.s2,40,INT32")); + TestUtils.executeNonQuery(EnvFactory.getEnv(), "create user abcd 'veryComplexPassword@123'"); + + try (final ISession session = + EnvFactory.getEnv().getSessionConnection("abcd", "veryComplexPassword@123"); + final ISession rootSession = EnvFactory.getEnv().getSessionConnection()) { + // Push last cache first + try (final SessionDataSet resultSet = + rootSession.executeFastLastDataQueryForOnePrefixPath( + Arrays.asList("root", "sg1", "d1"))) { + assertResultSetEqual(resultSet, lastQueryColumnNames, retArray2, true); + } + + try (final SessionDataSet resultSet = + session.executeLastDataQueryForOneDevice( + "root.sg1", "root.sg1.d1", Arrays.asList("notExist", "s1"), true)) { + assertResultSetEqual(resultSet, lastQueryColumnNames, retArray, true); + } + + try (final SessionDataSet resultSet = + session.executeFastLastDataQueryForOnePrefixPath(Arrays.asList("root", "sg1", "d1"))) { + assertResultSetEqual(resultSet, lastQueryColumnNames, retArray, true); + } + } catch (StatementExecutionException | RedirectException e) { + e.printStackTrace(); + fail(e.getMessage()); + } + } + // ------------------------------ Aggregation Query ------------------------------ @Test public void aggregationQueryTest() { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java new file mode 100644 index 00000000000..a2a6c72577c --- /dev/null +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/RenameTableColumnProcedure.java @@ -0,0 +1,247 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.procedure.impl.schema.table; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.exception.IoTDBException; +import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.confignode.consensus.request.write.table.RenameTableColumnPlan; +import org.apache.iotdb.confignode.consensus.request.write.table.view.RenameViewColumnPlan; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.exception.ProcedureException; +import org.apache.iotdb.confignode.procedure.impl.schema.table.view.RenameViewColumnProcedure; +import org.apache.iotdb.confignode.procedure.state.schema.RenameTableColumnState; +import org.apache.iotdb.confignode.procedure.store.ProcedureType; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.tsfile.utils.Pair; +import org.apache.tsfile.utils.ReadWriteIOUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Objects; + +public class RenameTableColumnProcedure + extends AbstractAlterOrDropTableProcedure<RenameTableColumnState> { + private static final Logger LOGGER = LoggerFactory.getLogger(RenameTableColumnProcedure.class); + + private String oldName; + private String newName; + + public RenameTableColumnProcedure(final boolean isGeneratedByPipe) { + super(isGeneratedByPipe); + } + + public RenameTableColumnProcedure( + final String database, + final String tableName, + final String queryId, + final String oldName, + final String newName, + final boolean isGeneratedByPipe) { + super(database, tableName, queryId, isGeneratedByPipe); + this.oldName = oldName; + this.newName = newName; + } + + @Override + protected Flow executeFromState( + final ConfigNodeProcedureEnv env, final RenameTableColumnState state) + throws InterruptedException { + final long startTime = System.currentTimeMillis(); + try { + switch (state) { + case COLUMN_CHECK: + LOGGER.info("Column check for table {}.{} when renaming column", database, tableName); + columnCheck(env); + break; + case PRE_RELEASE: + LOGGER.info("Pre release info of table {}.{} when renaming column", database, tableName); + preRelease(env); + break; + case RENAME_COLUMN: + LOGGER.info("Rename column to table {}.{} on config node", database, tableName); + renameColumn(env); + break; + case COMMIT_RELEASE: + LOGGER.info( + "Commit release info of table {}.{} when renaming column", database, tableName); + commitRelease(env); + return Flow.NO_MORE_STATE; + default: + setFailure(new ProcedureException("Unrecognized RenameTableColumnState " + state)); + return Flow.NO_MORE_STATE; + } + return Flow.HAS_MORE_STATE; + } finally { + LOGGER.info( + "RenameTableColumn-{}.{}-{} costs {}ms", + database, + tableName, + state, + (System.currentTimeMillis() - startTime)); + } + } + + private void columnCheck(final ConfigNodeProcedureEnv env) { + try { + final Pair<TSStatus, TsTable> result = + env.getConfigManager() + .getClusterSchemaManager() + .tableColumnCheckForColumnRenaming( + database, tableName, oldName, newName, this instanceof RenameViewColumnProcedure); + final TSStatus status = result.getLeft(); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + setFailure(new ProcedureException(new IoTDBException(status))); + return; + } + table = result.getRight(); + setNextState(RenameTableColumnState.PRE_RELEASE); + } catch (final MetadataException e) { + setFailure(new ProcedureException(e)); + } + } + + @Override + protected void preRelease(final ConfigNodeProcedureEnv env) { + super.preRelease(env); + setNextState(RenameTableColumnState.RENAME_COLUMN); + } + + private void renameColumn(final ConfigNodeProcedureEnv env) { + final TSStatus status = + env.getConfigManager() + .getClusterSchemaManager() + .executePlan( + this instanceof RenameViewColumnProcedure + ? new RenameViewColumnPlan(database, tableName, oldName, newName) + : new RenameTableColumnPlan(database, tableName, oldName, newName), + isGeneratedByPipe); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + setFailure(new ProcedureException(new IoTDBException(status))); + } else { + setNextState(RenameTableColumnState.COMMIT_RELEASE); + } + } + + @Override + protected void rollbackState(final ConfigNodeProcedureEnv env, final RenameTableColumnState state) + throws IOException, InterruptedException, ProcedureException { + final long startTime = System.currentTimeMillis(); + try { + switch (state) { + case RENAME_COLUMN: + LOGGER.info( + "Start rollback Renaming column to table {}.{} on configNode", + database, + table.getTableName()); + rollbackRenameColumn(env); + break; + case PRE_RELEASE: + LOGGER.info( + "Start rollback pre release info of table {}.{}", database, table.getTableName()); + rollbackPreRelease(env); + break; + } + } finally { + LOGGER.info( + "Rollback RenameTableColumn-{} costs {}ms.", + state, + (System.currentTimeMillis() - startTime)); + } + } + + private void rollbackRenameColumn(final ConfigNodeProcedureEnv env) { + if (table == null) { + return; + } + final TSStatus status = + env.getConfigManager() + .getClusterSchemaManager() + .executePlan( + this instanceof RenameViewColumnProcedure + ? new RenameViewColumnPlan(database, tableName, newName, oldName) + : new RenameTableColumnPlan(database, tableName, newName, oldName), + isGeneratedByPipe); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + setFailure(new ProcedureException(new IoTDBException(status))); + } + } + + @Override + protected RenameTableColumnState getState(final int stateId) { + return RenameTableColumnState.values()[stateId]; + } + + @Override + protected int getStateId(final RenameTableColumnState state) { + return state.ordinal(); + } + + @Override + protected RenameTableColumnState getInitialState() { + return RenameTableColumnState.COLUMN_CHECK; + } + + @Override + protected String getActionMessage() { + return "rename table column"; + } + + @Override + public void serialize(final DataOutputStream stream) throws IOException { + stream.writeShort( + isGeneratedByPipe + ? ProcedureType.PIPE_ENRICHED_RENAME_TABLE_COLUMN_PROCEDURE.getTypeCode() + : ProcedureType.RENAME_TABLE_COLUMN_PROCEDURE.getTypeCode()); + innerSerialize(stream); + } + + protected void innerSerialize(final DataOutputStream stream) throws IOException { + super.serialize(stream); + + ReadWriteIOUtils.write(oldName, stream); + ReadWriteIOUtils.write(newName, stream); + } + + @Override + public void deserialize(final ByteBuffer byteBuffer) { + super.deserialize(byteBuffer); + + this.oldName = ReadWriteIOUtils.readString(byteBuffer); + this.newName = ReadWriteIOUtils.readString(byteBuffer); + } + + @Override + public boolean equals(final Object o) { + return super.equals(o) + && Objects.equals(oldName, ((RenameTableColumnProcedure) o).oldName) + && Objects.equals(newName, ((RenameTableColumnProcedure) o).newName); + } + + @Override + public int hashCode() { + return Objects.hash(super.hashCode(), oldName, newName); + } +} diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/SetTablePropertiesProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/SetTablePropertiesProcedure.java new file mode 100644 index 00000000000..138ae9c9b50 --- /dev/null +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/table/SetTablePropertiesProcedure.java @@ -0,0 +1,264 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.procedure.impl.schema.table; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.exception.IoTDBException; +import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.schema.table.TsTable; +import org.apache.iotdb.confignode.consensus.request.write.table.SetTablePropertiesPlan; +import org.apache.iotdb.confignode.consensus.request.write.table.view.SetViewPropertiesPlan; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.exception.ProcedureException; +import org.apache.iotdb.confignode.procedure.impl.schema.table.view.SetViewPropertiesProcedure; +import org.apache.iotdb.confignode.procedure.state.schema.SetTablePropertiesState; +import org.apache.iotdb.confignode.procedure.store.ProcedureType; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.tsfile.utils.Pair; +import org.apache.tsfile.utils.ReadWriteIOUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; + +import static org.apache.iotdb.confignode.procedure.state.schema.SetTablePropertiesState.COMMIT_RELEASE; +import static org.apache.iotdb.confignode.procedure.state.schema.SetTablePropertiesState.PRE_RELEASE; +import static org.apache.iotdb.confignode.procedure.state.schema.SetTablePropertiesState.SET_PROPERTIES; +import static org.apache.iotdb.confignode.procedure.state.schema.SetTablePropertiesState.VALIDATE_TABLE; + +public class SetTablePropertiesProcedure + extends AbstractAlterOrDropTableProcedure<SetTablePropertiesState> { + + private static final Logger LOGGER = LoggerFactory.getLogger(SetTablePropertiesProcedure.class); + + private Map<String, String> originalProperties = new HashMap<>(); + private Map<String, String> updatedProperties; + + public SetTablePropertiesProcedure(final boolean isGeneratedByPipe) { + super(isGeneratedByPipe); + } + + public SetTablePropertiesProcedure( + final String database, + final String tableName, + final String queryId, + final Map<String, String> properties, + final boolean isGeneratedByPipe) { + super(database, tableName, queryId, isGeneratedByPipe); + this.updatedProperties = properties; + } + + @Override + protected Flow executeFromState( + final ConfigNodeProcedureEnv env, final SetTablePropertiesState state) + throws InterruptedException { + final long startTime = System.currentTimeMillis(); + try { + switch (state) { + case VALIDATE_TABLE: + validateTable(env); + LOGGER.info( + "Validate table for table {}.{} when setting properties", database, tableName); + if (!isFailed() && Objects.isNull(table)) { + LOGGER.info( + "The updated table has the same properties with the original one. Skip the procedure."); + return Flow.NO_MORE_STATE; + } + break; + case PRE_RELEASE: + preRelease(env); + LOGGER.info( + "Pre release info for table {}.{} when setting properties", database, tableName); + break; + case SET_PROPERTIES: + setProperties(env); + LOGGER.info("Set properties to table {}.{}", database, tableName); + break; + case COMMIT_RELEASE: + commitRelease(env); + LOGGER.info( + "Commit release info of table {}.{} when setting properties", database, tableName); + return Flow.NO_MORE_STATE; + default: + setFailure(new ProcedureException("Unrecognized AddTableColumnState " + state)); + return Flow.NO_MORE_STATE; + } + return Flow.HAS_MORE_STATE; + } finally { + LOGGER.info( + "SetTableProperties-{}.{}-{} costs {}ms", + database, + tableName, + state, + (System.currentTimeMillis() - startTime)); + } + } + + private void validateTable(final ConfigNodeProcedureEnv env) { + try { + final Pair<TSStatus, TsTable> result = + env.getConfigManager() + .getClusterSchemaManager() + .updateTableProperties( + database, + tableName, + originalProperties, + updatedProperties, + this instanceof SetViewPropertiesProcedure); + final TSStatus status = result.getLeft(); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + setFailure(new ProcedureException(new IoTDBException(status))); + return; + } + table = result.getRight(); + setNextState(PRE_RELEASE); + } catch (final MetadataException e) { + setFailure(new ProcedureException(e)); + } + } + + @Override + protected void preRelease(final ConfigNodeProcedureEnv env) { + super.preRelease(env); + setNextState(SET_PROPERTIES); + } + + private void setProperties(final ConfigNodeProcedureEnv env) { + final TSStatus status = + env.getConfigManager() + .getClusterSchemaManager() + .executePlan( + this instanceof SetViewPropertiesProcedure + ? new SetViewPropertiesPlan(database, tableName, updatedProperties) + : new SetTablePropertiesPlan(database, tableName, updatedProperties), + isGeneratedByPipe); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + setFailure(new ProcedureException(new IoTDBException(status))); + } else { + setNextState(COMMIT_RELEASE); + } + } + + @Override + protected String getActionMessage() { + return "set table properties"; + } + + @Override + protected void rollbackState( + final ConfigNodeProcedureEnv env, final SetTablePropertiesState state) + throws IOException, InterruptedException, ProcedureException { + final long startTime = System.currentTimeMillis(); + try { + switch (state) { + case PRE_RELEASE: + LOGGER.info( + "Start rollback pre release info for table {}.{} when setting properties", + database, + table.getTableName()); + rollbackPreRelease(env); + break; + case SET_PROPERTIES: + LOGGER.info( + "Start rollback set properties to table {}.{}", database, table.getTableName()); + rollbackSetProperties(env); + break; + } + } finally { + LOGGER.info( + "Rollback SetTableProperties-{} costs {}ms.", + state, + (System.currentTimeMillis() - startTime)); + } + } + + private void rollbackSetProperties(final ConfigNodeProcedureEnv env) { + if (table == null) { + return; + } + final TSStatus status = + env.getConfigManager() + .getClusterSchemaManager() + .executePlan( + this instanceof SetViewPropertiesProcedure + ? new SetViewPropertiesPlan(database, tableName, originalProperties) + : new SetTablePropertiesPlan(database, tableName, originalProperties), + isGeneratedByPipe); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + setFailure(new ProcedureException(new IoTDBException(status))); + } + } + + @Override + protected SetTablePropertiesState getState(final int stateId) { + return SetTablePropertiesState.values()[stateId]; + } + + @Override + protected int getStateId(final SetTablePropertiesState state) { + return state.ordinal(); + } + + @Override + protected SetTablePropertiesState getInitialState() { + return VALIDATE_TABLE; + } + + @Override + public void serialize(final DataOutputStream stream) throws IOException { + stream.writeShort( + isGeneratedByPipe + ? ProcedureType.PIPE_ENRICHED_SET_TABLE_PROPERTIES_PROCEDURE.getTypeCode() + : ProcedureType.SET_TABLE_PROPERTIES_PROCEDURE.getTypeCode()); + innerSerialize(stream); + } + + protected void innerSerialize(final DataOutputStream stream) throws IOException { + super.serialize(stream); + + ReadWriteIOUtils.write(originalProperties, stream); + ReadWriteIOUtils.write(updatedProperties, stream); + } + + @Override + public void deserialize(final ByteBuffer byteBuffer) { + super.deserialize(byteBuffer); + + this.originalProperties = ReadWriteIOUtils.readMap(byteBuffer); + this.updatedProperties = ReadWriteIOUtils.readMap(byteBuffer); + } + + @Override + public boolean equals(final Object o) { + return super.equals(o) + && Objects.equals(updatedProperties, ((SetTablePropertiesProcedure) o).updatedProperties); + } + + @Override + public int hashCode() { + return Objects.hash(super.hashCode(), updatedProperties); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java index 05fc4b80084..23aca924b7c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java @@ -55,6 +55,7 @@ import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator; import org.apache.iotdb.db.queryengine.plan.statement.Statement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement; import org.apache.iotdb.db.schemaengine.SchemaEngine; import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion; import org.apache.iotdb.db.utils.CommonUtils; @@ -102,7 +103,7 @@ public class RestApiServiceImpl extends RestApiService { public Response executeFastLastQueryStatement( PrefixPathList prefixPathList, SecurityContext securityContext) { Long queryId = null; - Statement statement = null; + QueryStatement statement = null; boolean finish = false; long startTime = System.nanoTime(); @@ -111,7 +112,23 @@ public class RestApiServiceImpl extends RestApiService { PartialPath prefixPath = new PartialPath(prefixPathList.getPrefixPaths().toArray(new String[0])); +<<<<<<< HEAD:iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java final Map<PartialPath, Map<String, TimeValuePair>> resultMap = new HashMap<>(); +======= + final Map<TableId, Map<IDeviceID, Map<String, Pair<TSDataType, TimeValuePair>>>> resultMap = + new HashMap<>(); + + // Check permission, the cost is rather low because the req only contains one prefix path + final IClientSession clientSession = SESSION_MANAGER.getCurrSession(); + final TSLastDataQueryReq tsLastDataQueryReq = + FastLastHandler.createTSLastDataQueryReq(clientSession, prefixPathList); + statement = StatementGenerator.createStatement(tsLastDataQueryReq); + + final Response response = authorizationHandler.checkAuthority(securityContext, statement); + if (response != null) { + return response; + } +>>>>>>> bb0f13b78bb (Enhance the last query permission && Fixed the rollback version of alter view / table plans && Deleted the unnecessary mods in Tree view deletion (#17465)):external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java final String prefixString = prefixPath.toString(); for (ISchemaRegion region : SchemaEngine.getInstance().getAllSchemaRegions()) { @@ -119,22 +136,28 @@ public class RestApiServiceImpl extends RestApiService { && !region.getDatabaseFullPath().startsWith(prefixString)) { continue; } +<<<<<<< HEAD:iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java region.fillLastQueryMap(prefixPath, resultMap); +======= + region.fillLastQueryMap(prefixPath, resultMap, statement.getAuthorityScope()); +>>>>>>> bb0f13b78bb (Enhance the last query permission && Fixed the rollback version of alter view / table plans && Deleted the unnecessary mods in Tree view deletion (#17465)):external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java } + // Check cache first +<<<<<<< HEAD:iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java if (!DataNodeSchemaCache.getInstance().getDeviceSchemaCache().getLastCache(resultMap)) { IClientSession clientSession = SESSION_MANAGER.getCurrSessionAndUpdateIdleTime(); TSLastDataQueryReq tsLastDataQueryReq = FastLastHandler.createTSLastDataQueryReq(clientSession, prefixPathList); statement = StatementGenerator.createStatement(tsLastDataQueryReq); +======= + if (!TableDeviceSchemaCache.getInstance().getLastCache(resultMap)) { +>>>>>>> bb0f13b78bb (Enhance the last query permission && Fixed the rollback version of alter view / table plans && Deleted the unnecessary mods in Tree view deletion (#17465)):external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java if (ExecuteStatementHandler.validateStatement(statement)) { return FastLastHandler.buildErrorResponse(TSStatusCode.EXECUTE_STATEMENT_ERROR); } - Optional.ofNullable(authorizationHandler.checkAuthority(securityContext, statement)) - .ifPresent(Response.class::cast); - queryId = SESSION_MANAGER.requestQueryId(); SessionInfo sessionInfo = SESSION_MANAGER.getSessionInfo(clientSession); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java index 8a498fce2e3..fef00a36a41 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java @@ -37,6 +37,7 @@ import org.apache.iotdb.commons.partition.DataPartitionQueryParam; import org.apache.iotdb.commons.path.AlignedPath; import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.path.PathPatternTree; import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.db.audit.AuditLogger; import org.apache.iotdb.db.auth.AuthorityChecker; @@ -87,6 +88,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsOfOneDeviceStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.QueryStatement; import org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateAlignedTimeSeriesStatement; import org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateMultiTimeSeriesStatement; import org.apache.iotdb.db.queryengine.plan.statement.metadata.CreateTimeSeriesStatement; @@ -819,7 +821,7 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { try { final long queryId = SESSION_MANAGER.requestQueryId(clientSession, req.statementId); - // 1. Map<Device, String[] measurements> ISchemaFetcher.getAllSensors(prefix) ~= 50ms + // 1.1 Map<Device, String[] measurements> ISchemaFetcher.getAllSensors(prefix) ~= 50ms final PartialPath prefixPath = new PartialPath(req.getPrefixes().toArray(new String[0])); if (prefixPath.hasWildcard()) { @@ -832,13 +834,26 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { final Map<PartialPath, Map<String, TimeValuePair>> resultMap = new HashMap<>(); int sensorNum = 0; + // 1.2 Check permission, the cost is rather low because the req only contains one prefix path + final QueryStatement s = StatementGenerator.createStatement(convert(req)); + final TSStatus status = + AuthorityChecker.checkAuthority( + s, + new TreeAccessCheckContext( + clientSession.getUserId(), + clientSession.getUsername(), + clientSession.getClientAddress())); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + return RpcUtils.getTSExecuteStatementResp(status); + } + final String prefixString = prefixPath.toString(); for (final ISchemaRegion region : SchemaEngine.getInstance().getAllSchemaRegions()) { if (!prefixString.startsWith(region.getDatabaseFullPath()) && !region.getDatabaseFullPath().startsWith(prefixString)) { continue; } - sensorNum += region.fillLastQueryMap(prefixPath, resultMap); + sensorNum += region.fillLastQueryMap(prefixPath, resultMap, s.getAuthorityScope()); } // 2.DATA_NODE_SCHEMA_CACHE.getLastCache() @@ -914,6 +929,20 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { long startTime = System.nanoTime(); Throwable t = null; try { + // Place the permission check first + final QueryStatement s = StatementGenerator.createStatement(convert(req)); + // permission check + final TSStatus status = + AuthorityChecker.checkAuthority( + s, + new TreeAccessCheckContext( + clientSession.getUserId(), + clientSession.getUsername(), + clientSession.getClientAddress())); + if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + return RpcUtils.getTSExecuteStatementResp(status); + } + String db; String deviceId; PartialPath devicePath; @@ -941,7 +970,7 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { // no valid DataRegion if (regionReplicaSets.isEmpty() || regionReplicaSets.size() == 1 && NOT_ASSIGNED == regionReplicaSets.get(0)) { - TSExecuteStatementResp resp = + final TSExecuteStatementResp resp = createResponse(DatasetHeaderFactory.getLastQueryHeader(), queryId); resp.setStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "")); resp.setQueryResult(Collections.emptyList()); @@ -951,7 +980,7 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { return resp; } - TEndPoint lastRegionLeader = + final TEndPoint lastRegionLeader = regionReplicaSets .get(regionReplicaSets.size() - 1) .dataNodeLocations @@ -962,33 +991,42 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { // read directly from cache if (isSameNode(lastRegionLeader)) { // the device's all dataRegions' leader are on current node, can use null entry in cache - boolean canUseNullEntry = + final boolean canUseNullEntry = regionReplicaSets.stream() .limit(regionReplicaSets.size() - 1L) .allMatch( regionReplicaSet -> isSameNode( regionReplicaSet.dataNodeLocations.get(0).mPPDataExchangeEndPoint)); - int sensorNum = req.sensors.size(); - TsBlockBuilder builder = LastQueryUtil.createTsBlockBuilder(sensorNum); + final int sensorNum = req.sensors.size(); + final TsBlockBuilder builder = LastQueryUtil.createTsBlockBuilder(sensorNum); boolean allCached = true; - for (String sensor : req.sensors) { - PartialPath fullPath; + + PathPatternTree queryTree = new PathPatternTree(); + for (final String sensor : req.sensors) { + final MeasurementPath fullPath; if (req.isLegalPathNodes()) { - fullPath = devicePath.concatNode(sensor); + fullPath = devicePath.concatAsMeasurementPath(sensor); } else { - fullPath = devicePath.concatNode((new PartialPath(sensor)).getFullPath()); + fullPath = devicePath.concatAsMeasurementPath((new PartialPath(sensor)).getFullPath()); } - TimeValuePair timeValuePair = DATA_NODE_SCHEMA_CACHE.getLastCache(fullPath); - if (timeValuePair == null) { - allCached = false; - break; - } else if (timeValuePair.getValue() == null) { - // there is no data for this sensor - if (!canUseNullEntry) { + queryTree.appendPathPattern(fullPath); + } + queryTree.constructTree(); + queryTree = s.getAuthorityScope().intersectWithFullPathPrefixTree(queryTree); + + if (!queryTree.isEmpty()) { + for (final MeasurementPath fullPath : queryTree.getAllPathPatterns(true)) { + final TimeValuePair timeValuePair = DATA_NODE_SCHEMA_CACHE.getLastCache(fullPath); + if (timeValuePair == null) { allCached = false; break; - } + } else if (timeValuePair.getValue() == null) { + // there is no data for this sensor + if (!canUseNullEntry) { + allCached = false; + break; + } } else { // we don't consider TTL LastQueryUtil.appendLastValue( @@ -1016,14 +1054,6 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { } } - // cache miss - Statement s = StatementGenerator.createStatement(convert(req)); - // permission check - TSStatus status = AuthorityChecker.checkAuthority(s, clientSession); - if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - return RpcUtils.getTSExecuteStatementResp(status); - } - quota = DataNodeThrottleQuotaManager.getInstance() .checkQuota(SESSION_MANAGER.getCurrSession().getUsername(), s); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGenerator.java index b220309df58..9f149abd531 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGenerator.java @@ -170,7 +170,7 @@ public class StatementGenerator { return queryStatement; } - public static Statement createStatement(TSLastDataQueryReq lastDataQueryReq) + public static QueryStatement createStatement(TSLastDataQueryReq lastDataQueryReq) throws IllegalPathException { final long startTime = System.nanoTime(); // construct query statement diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java index df69657f1d1..f085900a153 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/ISchemaRegion.java @@ -317,7 +317,7 @@ public interface ISchemaRegion { throws MetadataException; int fillLastQueryMap( - final PartialPath pattern, final Map<PartialPath, Map<String, TimeValuePair>> mapToFill) + final PartialPath pattern, final Map<PartialPath, Map<String, TimeValuePair>> mapToFill, final PathPatternTree scope) throws MetadataException; // endregion diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java index 7f8074c61ff..a964b7de4af 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java @@ -1335,9 +1335,9 @@ public class SchemaRegionMemoryImpl implements ISchemaRegion { @Override public int fillLastQueryMap( - final PartialPath pattern, final Map<PartialPath, Map<String, TimeValuePair>> mapToFill) + final PartialPath pattern, final Map<PartialPath, Map<String, TimeValuePair>> mapToFill, final PathPatternTree scope) throws MetadataException { - return mtree.fillLastQueryMap(pattern, mapToFill); + return mTree.fillLastQueryMap(pattern, mapToFill, scope); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java index fd8d7c5f1b5..d4e773dd593 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java @@ -1440,7 +1440,7 @@ public class SchemaRegionPBTreeImpl implements ISchemaRegion { @Override public int fillLastQueryMap( - final PartialPath pattern, final Map<PartialPath, Map<String, TimeValuePair>> mapToFill) { + final PartialPath pattern, final Map<PartialPath, Map<String, TimeValuePair>> mapToFill, final PathPatternTree scope) { throw new UnsupportedOperationException("Not implemented"); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java index 63d1e4c097f..e55b79f4645 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/mem/MTreeBelowSGMemoryImpl.java @@ -1085,12 +1085,11 @@ public class MTreeBelowSGMemoryImpl { } public int fillLastQueryMap( - final PartialPath prefixPath, final Map<PartialPath, Map<String, TimeValuePair>> mapToFill) + final PartialPath prefixPath, final Map<PartialPath, Map<String, TimeValuePair>> mapToFill, final PathPatternTree scope) throws MetadataException { final int[] sensorNum = {0}; try (final EntityUpdater<IMemMNode> updater = - new EntityUpdater<IMemMNode>( - rootNode, prefixPath, store, true, SchemaConstant.ALL_MATCH_SCOPE) { + new EntityUpdater<IMemMNode>(rootNode, prefixPath, store, true, scope) { @Override protected void updateEntity(final IDeviceMNode<IMemMNode> node) {
