support for cluster can acquire multi be in same host and fixed some issues

fe/src/com/baidu/palo/alter/DecommissionBackendJob.java

@@ -43,12 +43,12 @@
 import com.baidu.palo.clone.Clone;
 import com.baidu.palo.clone.CloneJob.JobPriority;
 import com.baidu.palo.cluster.Cluster;
+import com.baidu.palo.persist.BackendIdsUpdateInfo;
 import com.baidu.palo.common.Config;
 import com.baidu.palo.common.DdlException;
 import com.baidu.palo.common.FeMetaVersion;
 import com.baidu.palo.common.MetaNotFoundException;
 import com.baidu.palo.common.io.Text;
-import com.baidu.palo.persist.ClusterInfo;
 import com.baidu.palo.system.Backend;
 import com.baidu.palo.system.Backend.BackendState;
 import com.baidu.palo.system.SystemInfoService;
@@ -263,6 +263,15 @@ public synchronized boolean sendTasks() {
                             continue;
                         }
 
+
+                        // exclude backend in same hosts with the other replica
+                        Set<String> hosts = Sets.newHashSet();
+                        for (Replica replica : tablet.getReplicas()) {
+                            if (replica.getBackendId() != backendId) {
+                                hosts.add(clusterInfo.getBackend(replica.getBackendId()).getHost());
+                            }   
+                        } 
+
                         // choose dest backend
                         long destBackendId = -1L;
                         int num = 0;
@@ -273,7 +282,7 @@ public synchronized boolean sendTasks() {
                                 return true;
                             }
 
-                            if (tablet.getReplicaByBackendId(destBackendIds.get(0)) != null) {
+                            if (hosts.contains(clusterInfo.getBackend(destBackendIds.get(0)).getHost())) {
                                 // replica can not in same backend
                                 continue;
                             }
@@ -469,23 +478,19 @@ public synchronized int tryFinishJob() {
                 // Shrinking capacity in cluser
                 if (decomissionType == DecomissionType.ClusterDecomission) {
                     for (String clusterName : clusterBackendsMap.keySet()) {
-                        final Map<Long, Backend> map = clusterBackendsMap.get(clusterName);
+                        final Map<Long, Backend> idToBackend = clusterBackendsMap.get(clusterName);
                         final Cluster cluster = Catalog.getInstance().getCluster(clusterName);
-                        final List<Long> removeIds = Lists.newArrayList();
-                        for (long id : map.keySet()) {
-                            final Backend backend = map.get(id);
-                            backend.setOwnerClusterName("");
+                        List<Long> backendList = Lists.newArrayList();
+                        for (long id : idToBackend.keySet()) {
+                            final Backend backend = idToBackend.get(id);
+                            backend.clearClusterName();
                             backend.setBackendState(BackendState.free);
                             backend.setDecommissioned(false);
+                            backendList.add(id);
                             cluster.removeBackend(id);
-                            Catalog.getInstance().getEditLog().logBackendStateChange(backend);
-                            removeIds.add(id);
                         }
-                        cluster.removeBackends(removeIds);
-                        ClusterInfo info = new ClusterInfo();
-                        info.setClusterName(cluster.getName());
-                        info.setBackendIdList(cluster.getBackendIdList());
-                        Catalog.getInstance().getEditLog().logUpdateCluster(info);
+                        BackendIdsUpdateInfo updateInfo = new BackendIdsUpdateInfo(backendList);
+                        Catalog.getInstance().getEditLog().logUpdateClusterAndBackendState(updateInfo);
                     }
                 }
             }

fe/src/com/baidu/palo/analysis/AlterClusterStmt.java

@@ -56,7 +56,7 @@ public void analyze(Analyzer analyzer) throws AnalysisException, InternalExcepti
             ErrorReport.reportAnalysisException(ErrorCode.ERR_CLUSTER_NO_PARAMETER);
         }
 
-        if (instanceNum < 0) {
+        if (instanceNum <= 0) {
             ErrorReport.reportAnalysisException(ErrorCode.ERR_CLUSTER_CREATE_ISTANCE_NUM_ERROR);
         }
     }

fe/src/com/baidu/palo/analysis/AlterSystemStmt.java

@@ -56,7 +56,7 @@ public void analyze(Analyzer analyzer) throws AnalysisException, InternalExcepti
     @Override
     public String toSql() {
         StringBuilder sb = new StringBuilder();
-        sb.append("ALTER CLUSTER ").append(alterClause.toSql());
+        sb.append("ALTER SYSTEM ").append(alterClause.toSql());
         return sb.toString();
     }
 

fe/src/com/baidu/palo/analysis/CancelAlterSystemStmt.java

@@ -25,8 +25,8 @@
 
 public class CancelAlterSystemStmt extends CancelStmt {
 
-    private List<String> hostPorts;
-    protected List<Pair<String, Integer>> hostPortPairs;
+    protected List<String> hostPorts;
+    private List<Pair<String, Integer>> hostPortPairs;
 
     public CancelAlterSystemStmt(List<String> hostPorts) {
         this.hostPorts = hostPorts;

fe/src/com/baidu/palo/analysis/ShowBackendsStmt.java

@@ -0,0 +1,40 @@
+// Copyright (c) 2017, Baidu.com, Inc. All Rights Reserved
+
+// Licensed 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 com.baidu.palo.analysis;
+
+import com.baidu.palo.catalog.Column;
+import com.baidu.palo.catalog.ColumnType;
+import com.baidu.palo.common.proc.BackendsProcDir;
+import com.baidu.palo.qe.ShowResultSetMetaData;
+
+public class ShowBackendsStmt extends ShowStmt {
+
+    public ShowBackendsStmt() {  
+    }
+
+    @Override
+    public ShowResultSetMetaData getMetaData() {
+         ShowResultSetMetaData.Builder builder = ShowResultSetMetaData.builder();
+         for (String title : BackendsProcDir.TITLE_NAMES) {
+             if (title.equals("HostName") || title.equals("HeartbeatPort") 
+                     || title.equals("BePort") || title.equals("HttpPort")) {
+                 continue;
+             }
+            builder.addColumn(new Column(title, ColumnType.createVarchar(30)));
+        }
+        return builder.build();
+    }
+}

fe/src/com/baidu/palo/analysis/ShowProcesslistStmt.java

@@ -28,6 +28,7 @@ public class ShowProcesslistStmt extends ShowStmt {
                     .addColumn(new Column("Id", ColumnType.createType(PrimitiveType.BIGINT)))
                     .addColumn(new Column("User", ColumnType.createVarchar(16)))
                     .addColumn(new Column("Host", ColumnType.createVarchar(16)))
+                    .addColumn(new Column("Cluster", ColumnType.createVarchar(16)))
                     .addColumn(new Column("Db", ColumnType.createVarchar(16)))
                     .addColumn(new Column("Command", ColumnType.createVarchar(16)))
                     .addColumn(new Column("Time", ColumnType.createType(PrimitiveType.INT)))

fe/src/com/baidu/palo/catalog/Catalog.java

@@ -130,6 +130,7 @@
 import com.baidu.palo.persist.Storage;
 import com.baidu.palo.persist.StorageInfo;
 import com.baidu.palo.persist.TableInfo;
+import com.baidu.palo.persist.BackendIdsUpdateInfo;
 import com.baidu.palo.qe.ConnectContext;
 import com.baidu.palo.qe.JournalObservable;
 import com.baidu.palo.qe.SessionVariable;
@@ -138,6 +139,7 @@
 import com.baidu.palo.system.Backend;
 import com.baidu.palo.system.Frontend;
 import com.baidu.palo.system.SystemInfoService;
+import com.baidu.palo.system.Backend.BackendState;
 import com.baidu.palo.task.AgentBatchTask;
 import com.baidu.palo.task.AgentTask;
 import com.baidu.palo.task.AgentTaskExecutor;
@@ -180,6 +182,7 @@
 import java.io.IOException;
 import java.net.HttpURLConnection;
 import java.net.InetAddress;
+import java.util.Iterator;
 import java.net.URL;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
@@ -3406,7 +3409,7 @@ private void createTablets(String clusterName, MaterializedIndex index, ReplicaS
                 List<Long> chosenBackendIds = Catalog.getCurrentSystemInfo().seqChooseBackendIds(replicationNum, true,
                         true, clusterName);
                 if (chosenBackendIds == null) {
-                    throw new DdlException("Failed to find enough alive backends. need: " + replicationNum);
+                    throw new DdlException("Failed to find enough host in all backends. need: " + replicationNum);
                 }
                 Preconditions.checkState(chosenBackendIds.size() == replicationNum);
                 for (long backendId : chosenBackendIds) {
@@ -4389,8 +4392,8 @@ public List<List<String>> showWhiteList(String user) {
      */
     public void createCluster(CreateClusterStmt stmt) throws DdlException {
         final String clusterName = stmt.getClusterName();
+        writeLock();
         try {
-            writeLock();
             if (nameToCluster.containsKey(clusterName)) {
                 ErrorReport.reportDdlException(ErrorCode.ERR_CLUSTER_HAS_EXIST, clusterName);
             } else {
@@ -4428,7 +4431,13 @@ private void unprotectCreateCluster(Cluster cluster) {
                 return;
             }
         }
-
+        final Iterator<Long> iterator = cluster.getBackendIdList().iterator();
+        while (iterator.hasNext()) {
+            final Long id = iterator.next();
+            final Backend backend = systemInfo.getBackend(id);
+            backend.setOwnerClusterName(cluster.getName());
+            backend.setBackendState(BackendState.using);
+        }
         idToCluster.put(cluster.getId(), cluster);
         nameToCluster.put(cluster.getName(), cluster);
         final InfoSchemaDb db = new InfoSchemaDb(cluster.getName());
@@ -4473,8 +4482,8 @@ public void replayCreateCluster(Cluster cluster) {
      * @throws DdlException
      */
     public void dropCluster(DropClusterStmt stmt) throws DdlException {
+        writeLock();
         try {
-            writeLock();
             final Cluster cluster = nameToCluster.get(stmt.getClusterName());
             final String clusterName = stmt.getClusterName();
             if (cluster == null) {
@@ -4537,9 +4546,8 @@ public void replayUpdateCluster(ClusterInfo info) {
     public void processModityCluster(AlterClusterStmt stmt) throws DdlException {
         final String clusterName = stmt.getAlterClusterName();
         final int newInstanceNum = stmt.getInstanceNum();
-
+        writeLock();
         try {
-            writeLock();
             if (!nameToCluster.containsKey(clusterName)) {
                 ErrorReport.reportDdlException(ErrorCode.ERR_CLUSTER_NO_EXISTS, clusterName);
             }
@@ -4682,8 +4690,8 @@ public void migrateDb(MigrateDbStmt stmt) throws DdlException {
         final String srcDbName = stmt.getSrcDb();
         final String desDbName = stmt.getDesDb();
 
+        writeLock();
         try {
-            writeLock();
             if (!nameToCluster.containsKey(srcClusterName)) {
                 ErrorReport.reportDdlException(ErrorCode.ERR_CLUSTER_SRC_CLUSTER_NO_EXIT, srcClusterName);
             }
@@ -4826,9 +4834,8 @@ public void linkDb(LinkDbStmt stmt) throws DdlException {
         final String desClusterName = stmt.getDesCluster();
         final String srcDbName = stmt.getSrcDb();
         final String desDbName = stmt.getDesDb();
-
+        writeLock();
         try {
-            writeLock();
             if (!nameToCluster.containsKey(srcClusterName)) {
                 ErrorReport.reportDdlException(ErrorCode.ERR_CLUSTER_SRC_CLUSTER_NO_EXIT, srcClusterName);
             }
@@ -5072,4 +5079,21 @@ public long loadBrokers(DataInputStream dis, long checksum) throws IOException,
         }
         return checksum;
     }
+
+    public void replayUpdateClusterAndBackends(BackendIdsUpdateInfo info) {
+        for (long id : info.getBackendList()) {
+            final Backend backend = systemInfo.getBackend(id);
+            writeLock();
+            try {
+                final Cluster cluster = nameToCluster.get(backend.getOwnerClusterName());
+                cluster.removeBackend(id);
+            } finally {
+                writeUnlock();  
+            }
+            backend.setDecommissioned(false);
+            backend.clearClusterName();
+            backend.setBackendState(BackendState.free);            
+        }
+    }
+
 }