[feat](Nereids) add restore state check of partition after rewrite

fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceClient.java

@@ -79,7 +79,7 @@ private long connectionAgeExpiredAt() {
         if (!isMetaServiceEndpointList && connectionAgeBase > 1) {
             long base = TimeUnit.MINUTES.toMillis(connectionAgeBase);
             long now = System.currentTimeMillis();
-            long rand = random.nextLong(base);
+            long rand = 0;
             return now + base + rand;
         }
         return Long.MAX_VALUE;

fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java

@@ -45,6 +45,7 @@
 import org.apache.doris.nereids.rules.rewrite.CheckMatchExpression;
 import org.apache.doris.nereids.rules.rewrite.CheckMultiDistinct;
 import org.apache.doris.nereids.rules.rewrite.CheckPrivileges;
+import org.apache.doris.nereids.rules.rewrite.CheckRestorePartition;
 import org.apache.doris.nereids.rules.rewrite.ClearContextStatus;
 import org.apache.doris.nereids.rules.rewrite.CollectCteConsumerOutput;
 import org.apache.doris.nereids.rules.rewrite.CollectFilterAboveConsumer;
@@ -483,6 +484,7 @@ public class Rewriter extends AbstractBatchJobExecutor {
                                 new ExpressionRewrite(CheckLegalityAfterRewrite.INSTANCE),
                                 new CheckMatchExpression(),
                                 new CheckMultiDistinct(),
+                                new CheckRestorePartition(),
                                 new CheckAfterRewrite()
                         )
                 ),

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java

@@ -93,6 +93,7 @@ public enum RuleType {
     // check analysis rule
     CHECK_AGGREGATE_ANALYSIS(RuleTypeClass.CHECK),
     CHECK_ANALYSIS(RuleTypeClass.CHECK),
+    CHECK_RESTORE_PARTITION(RuleTypeClass.CHECK),
     CHECK_OBJECT_TYPE_ANALYSIS(RuleTypeClass.CHECK),
     CHECK_DATA_TYPES(RuleTypeClass.CHECK),
 

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/CheckRestorePartition.java

@@ -0,0 +1,47 @@
+// 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.doris.nereids.rules.rewrite;
+
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.rules.Rule;
+import org.apache.doris.nereids.rules.RuleType;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+
+/**
+ * If there are restore partition after rewrite throw an exception because it is not readable or writable
+ */
+public class CheckRestorePartition extends OneRewriteRuleFactory {
+    @Override
+    public Rule build() {
+        return logicalOlapScan().then(this::checkRestorePartition).toRule(RuleType.CHECK_RESTORE_PARTITION);
+    }
+
+    private LogicalOlapScan checkRestorePartition(LogicalOlapScan scan) {
+        if (scan.getSelectedPartitionIds() != null) {
+            for (long id : scan.getSelectedPartitionIds()) {
+                Partition partition = scan.getTable().getPartition(id);
+                if (partition.getState() == Partition.PartitionState.RESTORE) {
+                    throw new AnalysisException("Partition state is not NORMAL: "
+                            + partition.getName() + ":" + "RESTORING");
+                }
+            }
+        }
+        return scan;
+    }
+}

fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/CheckRestorePartitionTest.java

@@ -0,0 +1,68 @@
+// 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.doris.nereids.rules.rewrite;
+
+import org.apache.doris.catalog.Partition;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.util.MemoPatternMatchSupported;
+import org.apache.doris.nereids.util.MemoTestUtils;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class CheckRestorePartitionTest extends TestWithFeService implements MemoPatternMatchSupported {
+
+    @Override
+    protected void runBeforeAll() throws Exception {
+        createDatabase("test_restore_partition");
+        useDatabase("test_restore_partition");
+
+        createTable("create table test_restore(id int, part int) "
+                + "partition by range(part) ("
+                + "  partition p1 values[('1'), ('2')),"
+                + "  partition p2 values[('2'), ('3'))"
+                + ") "
+                + "distributed by hash(id) "
+                + "properties ('replication_num'='1')");
+
+        FeConstants.runningUnitTest = true;
+    }
+
+    @Test
+    void testOlapScanPartitionWithSingleColumnCase() {
+        String sql = "select * from test_restore";
+        Plan plan = PlanChecker.from(createCascadesContext(sql))
+                .analyze(sql)
+                .getPlan();
+
+        LogicalOlapScan scan = (LogicalOlapScan) plan.child(0).child(0);
+        for (Partition partition : scan.getTable().getPartitions()) {
+            partition.setState(Partition.PartitionState.RESTORE);
+        }
+        AnalysisException exception = Assertions.assertThrows(AnalysisException.class,
+                () -> PlanChecker.from(MemoTestUtils.createConnectContext(), scan)
+                .applyBottomUp(new CheckRestorePartition()));
+        Assertions.assertTrue(exception.getMessage().contains("Partition state is not NORMAL:"));
+    }
+
+}