Implemented the alter encoding compression function for tree model

integration-test/src/test/java/org/apache/iotdb/db/it/schema/IoTDBAlterEncodingCompressorIT.java

@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.it.schema;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.itbase.category.LocalStandaloneIT;
+import org.apache.iotdb.util.AbstractSchemaIT;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runners.Parameterized;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
+
+@Category({LocalStandaloneIT.class, ClusterIT.class})
+public class IoTDBAlterEncodingCompressorIT extends AbstractSchemaIT {
+
+  public IoTDBAlterEncodingCompressorIT(SchemaTestMode schemaTestMode) {
+    super(schemaTestMode);
+  }
+
+  @Parameterized.BeforeParam
+  public static void before() throws Exception {
+    setUpEnvironment();
+    EnvFactory.getEnv().initClusterEnvironment();
+  }
+
+  @Parameterized.AfterParam
+  public static void after() throws Exception {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+    tearDownEnvironment();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    clearSchema();
+  }
+
+  @Test
+  public void alterEncodingAndCompressorTest() throws Exception {
+    if (schemaTestMode.equals(SchemaTestMode.PBTree)) {
+      return;
+    }
+    try (final Connection connection = EnvFactory.getEnv().getConnection();
+        final Statement statement = connection.createStatement()) {
+      statement.execute("create timeSeries root.vehicle.wind.a int32");
+
+      try {
+        statement.execute("alter timeSeries root.nonExist.** set encoding=PLAIN");
+        fail();
+      } catch (final SQLException e) {
+        Assert.assertEquals(
+            "508: Timeseries [root.nonExist.**] does not exist or is represented by device template",
+            e.getMessage());
+      }
+
+      try {
+        statement.execute("alter timeSeries if exists root.nonExist.** set encoding=PLAIN");
+      } catch (final SQLException e) {
+        fail(
+            "Alter encoding & compressor shall not fail when timeSeries not exists if set if exists");
+      }
+
+      try {
+        statement.execute("alter timeSeries if exists root.vehicle.** set encoding=aaa");
+        fail();
+      } catch (final SQLException e) {
+        Assert.assertEquals("701: Unsupported encoding: AAA", e.getMessage());
+      }
+
+      try {
+        statement.execute("alter timeSeries if exists root.vehicle.** set compressor=aaa");
+        fail();
+      } catch (final SQLException e) {
+        Assert.assertEquals("701: Unsupported compressor: AAA", e.getMessage());
+      }
+
+      try {
+        statement.execute("alter timeSeries if exists root.vehicle.** set falseKey=aaa");
+        fail();
+      } catch (final SQLException e) {
+        Assert.assertEquals("701: property falsekey is unsupported yet.", e.getMessage());
+      }
+
+      try {
+        statement.execute("alter timeSeries if exists root.vehicle.** set encoding=DICTIONARY");
+        fail();
+      } catch (final SQLException e) {
+        Assert.assertTrue(e.getMessage().contains("encoding DICTIONARY does not support INT32"));
+      }
+
+      statement.execute("alter timeSeries root.** set encoding=Plain, compressor=LZMA2");
+
+      try (final ResultSet resultSet = statement.executeQuery("SHOW TIMESERIES")) {
+        while (resultSet.next()) {
+          assertEquals("PLAIN", resultSet.getString(5));
+          assertEquals("LZMA2", resultSet.getString(6));
+        }
+      }
+
+      statement.execute("create user IoTDBUser '!@#$!dfdfzvd343'");
+      statement.execute("grant write on root.vehicle.wind.a to user IoTDBUser");
+      statement.execute("create timeSeries root.vehicle.wind.b int32");
+    }
+
+    try (final Connection connection =
+            EnvFactory.getEnv().getConnection("IoTDBUser", "!@#$!dfdfzvd343");
+        final Statement statement = connection.createStatement()) {
+      try {
+        statement.execute("alter timeSeries root.vechile.** set encoding=PLAIN, compressor=LZMA2");
+        fail();
+      } catch (final SQLException e) {
+        Assert.assertEquals(
+            "803: No permissions for this operation, please add privilege WRITE_SCHEMA on [root.vechile.**]",
+            e.getMessage());
+      }
+
+      try {
+        statement.execute(
+            "alter timeSeries root.vechile.wind.a, root.__audit.** set encoding=PLAIN, compressor=LZMA2");
+        fail();
+      } catch (final SQLException e) {
+        Assert.assertEquals(
+            "803: 'AUDIT' permission is needed to alter the encoding and compressor of database root.__audit",
+            e.getMessage());
+      }
+
+      try {
+        statement.execute(
+            "alter timeSeries if permitted root.vehicle.**, root.__audit.** set encoding=GORILLA, compressor=GZIP");
+      } catch (final SQLException e) {
+        fail("Alter encoding & compressor shall not fail when no privileges if set if permitted");
+      }
+    }
+
+    try (final Connection connection = EnvFactory.getEnv().getConnection();
+        final Statement statement = connection.createStatement()) {
+      try (final ResultSet resultSet =
+          statement.executeQuery("SHOW TIMESERIES root.__audit.**._0.password")) {
+        while (resultSet.next()) {
+          assertEquals("PLAIN", resultSet.getString(5));
+          assertEquals("LZMA2", resultSet.getString(6));
+        }
+      }
+
+      try (final ResultSet resultSet =
+          statement.executeQuery("SHOW TIMESERIES root.vehicle.wind.b")) {
+        resultSet.next();
+        assertEquals("TS_2DIFF", resultSet.getString(5));
+        assertEquals("LZ4", resultSet.getString(6));
+      }
+
+      try (final ResultSet resultSet =
+          statement.executeQuery("SHOW TIMESERIES root.vehicle.wind.a")) {
+        resultSet.next();
+        assertEquals("GORILLA", resultSet.getString(5));
+        assertEquals("GZIP", resultSet.getString(6));
+      }
+    }
+  }
+}

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeInclusionIT.java

@@ -79,17 +79,18 @@ public void testPureSchemaInclusion() throws Exception {
           Arrays.asList(
               // TODO: add database creation after the database auto creating on receiver can be
               // banned
-              "create timeseries root.ln.wf01.wt01.status with datatype=BOOLEAN,encoding=PLAIN",
-              "ALTER timeseries root.ln.wf01.wt01.status ADD TAGS tag3=v3",
-              "ALTER timeseries root.ln.wf01.wt01.status ADD ATTRIBUTES attr4=v4"),
+              "create timeSeries root.ln.wf01.wt01.status with datatype=BOOLEAN,encoding=PLAIN",
+              "ALTER timeSeries root.ln.wf01.wt01.status ADD TAGS tag3=v3",
+              "ALTER timeSeries root.ln.wf01.wt01.status ADD ATTRIBUTES attr4=v4",
+              "ALTER timeSeries root.** set compressor=ZSTD"),
           null);
 
       TestUtils.assertDataEventuallyOnEnv(
           receiverEnv,
           "show timeseries root.ln.**",
           "Timeseries,Alias,Database,DataType,Encoding,Compression,Tags,Attributes,Deadband,DeadbandParameters,ViewType,",
           Collections.singleton(
-              "root.ln.wf01.wt01.status,null,root.ln,BOOLEAN,PLAIN,LZ4,{\"tag3\":\"v3\"},{\"attr4\":\"v4\"},null,null,BASE,"));
+              "root.ln.wf01.wt01.status,null,root.ln,BOOLEAN,PLAIN,ZSTD,{\"tag3\":\"v3\"},{\"attr4\":\"v4\"},null,null,BASE,"));
 
       TestUtils.executeNonQueries(
           senderEnv,

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/IoTDBSqlParser.g4

@@ -39,7 +39,7 @@ ddlStatement
     // Database
     : createDatabase | dropDatabase | dropPartition | alterDatabase | showDatabases | countDatabases
     // Timeseries & Path
-    | createTimeseries | dropTimeseries | alterTimeseries
+    | createTimeseries | dropTimeseries | alterTimeseries | alterEncodingCompressor
     | showDevices | showTimeseries | showChildPaths | showChildNodes | countDevices | countTimeseries | countNodes
     // Device Template
     | createSchemaTemplate | createTimeseriesUsingSchemaTemplate | dropSchemaTemplate | dropTimeseriesOfSchemaTemplate
@@ -177,6 +177,10 @@ alterClause
     | UPSERT aliasClause? tagClause? attributeClause?
     ;
 
+alterEncodingCompressor
+    : ALTER TIMESERIES (IF EXISTS)? (IF PERMITTED)? prefixPath (COMMA prefixPath)* SET attributePair (COMMA attributePair)*
+    ;
+
 aliasClause
     : ALIAS operator_eq alias
     ;

iotdb-core/antlr/src/main/antlr4/org/apache/iotdb/db/qp/sql/SqlLexer.g4

@@ -610,6 +610,10 @@ PATHS
     : P A T H S
     ;
 
+PERMITTED
+    : P E R M I T T E D
+    ;
+
 PIPE
     : P I P E
     ;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java

@@ -90,6 +90,8 @@ public enum CnToDnAsyncRequestType {
   DELETE_DATA_FOR_DELETE_SCHEMA,
   DELETE_TIMESERIES,
 
+  ALTER_ENCODING_COMPRESSOR,
+
   CONSTRUCT_SCHEMA_BLACK_LIST_WITH_TEMPLATE,
   ROLLBACK_SCHEMA_BLACK_LIST_WITH_TEMPLATE,
   DEACTIVATE_TEMPLATE,

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java

@@ -48,6 +48,7 @@
 import org.apache.iotdb.confignode.client.async.handlers.rpc.subscription.ConsumerGroupPushMetaRPCHandler;
 import org.apache.iotdb.confignode.client.async.handlers.rpc.subscription.TopicPushMetaRPCHandler;
 import org.apache.iotdb.mpp.rpc.thrift.TActiveTriggerInstanceReq;
+import org.apache.iotdb.mpp.rpc.thrift.TAlterEncodingCompressorReq;
 import org.apache.iotdb.mpp.rpc.thrift.TAlterViewReq;
 import org.apache.iotdb.mpp.rpc.thrift.TCheckSchemaRegionUsingTemplateReq;
 import org.apache.iotdb.mpp.rpc.thrift.TCheckTimeSeriesExistenceReq;
@@ -309,6 +310,11 @@ protected void initActionMapBuilder() {
         CnToDnAsyncRequestType.DELETE_TIMESERIES,
         (req, client, handler) ->
             client.deleteTimeSeries((TDeleteTimeSeriesReq) req, (SchemaUpdateRPCHandler) handler));
+    actionMapBuilder.put(
+        CnToDnAsyncRequestType.ALTER_ENCODING_COMPRESSOR,
+        (req, client, handler) ->
+            client.alterEncodingCompressor(
+                (TAlterEncodingCompressorReq) req, (SchemaUpdateRPCHandler) handler));
     actionMapBuilder.put(
         CnToDnAsyncRequestType.CONSTRUCT_SCHEMA_BLACK_LIST_WITH_TEMPLATE,
         (req, client, handler) ->

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeAsyncRequestRPCHandler.java

@@ -83,6 +83,7 @@ public static DataNodeAsyncRequestRPCHandler<?> buildHandler(
       case ROLLBACK_SCHEMA_BLACK_LIST:
       case DELETE_DATA_FOR_DELETE_SCHEMA:
       case DELETE_TIMESERIES:
+      case ALTER_ENCODING_COMPRESSOR:
       case CONSTRUCT_SCHEMA_BLACK_LIST_WITH_TEMPLATE:
       case ROLLBACK_SCHEMA_BLACK_LIST_WITH_TEMPLATE:
       case DEACTIVATE_TEMPLATE:

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java

@@ -62,6 +62,7 @@
 import org.apache.iotdb.confignode.consensus.request.write.partition.CreateSchemaPartitionPlan;
 import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan;
 import org.apache.iotdb.confignode.consensus.request.write.partition.UpdateRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeAlterEncodingCompressorPlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeCreateTableOrViewPlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeactivateTemplatePlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteDevicesPlan;
@@ -552,6 +553,9 @@ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOExcept
         case PipeDeleteDevices:
           plan = new PipeDeleteDevicesPlan();
           break;
+        case PipeAlterEncodingCompressor:
+          plan = new PipeAlterEncodingCompressorPlan();
+          break;
         case UpdateTriggersOnTransferNodes:
           plan = new UpdateTriggersOnTransferNodesPlan();
           break;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java

@@ -307,6 +307,7 @@ public enum ConfigPhysicalPlanType {
   PipeSetTTL((short) 1705),
   PipeCreateTableOrView((short) 1706),
   PipeDeleteDevices((short) 1707),
+  PipeAlterEncodingCompressor((short) 1708),
 
   /** Subscription */
   CreateTopic((short) 1800),

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanVisitor.java

@@ -24,6 +24,7 @@
 import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan;
 import org.apache.iotdb.confignode.consensus.request.write.database.DeleteDatabasePlan;
 import org.apache.iotdb.confignode.consensus.request.write.database.SetTTLPlan;
+import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeAlterEncodingCompressorPlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeCreateTableOrViewPlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeactivateTemplatePlan;
 import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteDevicesPlan;
@@ -193,6 +194,8 @@ public R process(final ConfigPhysicalPlan plan, final C context) {
         return visitRenameTable((RenameTablePlan) plan, context);
       case RenameView:
         return visitRenameView((RenameViewPlan) plan, context);
+      case PipeAlterEncodingCompressor:
+        return visitPipeAlterEncodingCompressor((PipeAlterEncodingCompressorPlan) plan, context);
       default:
         return visitPlan(plan, context);
     }
@@ -504,4 +507,9 @@ public R visitRenameTable(final RenameTablePlan renameTablePlan, final C context
   public R visitRenameView(final RenameViewPlan renameViewPlan, final C context) {
     return visitRenameTable(renameViewPlan, context);
   }
+
+  public R visitPipeAlterEncodingCompressor(
+      final PipeAlterEncodingCompressorPlan pipeAlterEncodingCompressorPlan, final C context) {
+    return visitPlan(pipeAlterEncodingCompressorPlan, context);
+  }
 }