在建立和修改Schema时可以配置复制参数

lealone-cluster/src/main/java/org/lealone/cluster/db/Keyspace.java

@@ -1,31 +1,63 @@
+/*
+ * 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.lealone.cluster.db;
 
+import java.util.HashMap;
+
 import org.lealone.cluster.config.DatabaseDescriptor;
+import org.lealone.cluster.exceptions.ConfigurationException;
 import org.lealone.cluster.locator.AbstractReplicationStrategy;
-
-public class Keyspace {
-    private static Keyspace INSTANCE = new Keyspace();
-
-    public static Keyspace open(String keyspaceName) {
-        return INSTANCE;
-    }
-
-    public static Iterable<Keyspace> all() {
-        return null;
-    }
-
-    private final AbstractReplicationStrategy replicationStrategy;
-
-    public Keyspace() {
-        //TODO 按表或按数据库配置
-        replicationStrategy = DatabaseDescriptor.getDefaultReplicationStrategy();
+import org.lealone.cluster.service.StorageService;
+import org.lealone.dbobject.Schema;
+import org.lealone.dbobject.Schema.ReplicationPropertiesChangeListener;
+
+public class Keyspace implements ReplicationPropertiesChangeListener {
+    private static final Keyspace INSTANCE = new Keyspace();
+    private static final HashMap<Schema, AbstractReplicationStrategy> replicationStrategys = new HashMap<>();
+    private static final AbstractReplicationStrategy defaultReplicationStrategy = DatabaseDescriptor
+            .getDefaultReplicationStrategy();
+
+    public static AbstractReplicationStrategy getReplicationStrategy(Schema schema) {
+        if (schema.getReplicationProperties() == null)
+            return defaultReplicationStrategy;
+        AbstractReplicationStrategy replicationStrategy = replicationStrategys.get(schema);
+        if (replicationStrategy == null) {
+            HashMap<String, String> map = new HashMap<>(schema.getReplicationProperties());
+            String className = map.remove("class");
+            if (className == null) {
+                throw new ConfigurationException("Missing replication strategy class");
+            }
+
+            replicationStrategy = AbstractReplicationStrategy.createReplicationStrategy(schema.getFullName(),
+                    AbstractReplicationStrategy.getClass(className), StorageService.instance.getTokenMetaData(),
+                    DatabaseDescriptor.getEndpointSnitch(), map);
+            schema.setReplicationPropertiesChangeListener(INSTANCE);
+            replicationStrategys.put(schema, replicationStrategy);
+        }
+        return replicationStrategy;
     }
 
-    public AbstractReplicationStrategy getReplicationStrategy() {
-        return replicationStrategy;
+    private Keyspace() {
     }
 
-    public String getName() {
-        return null;
+    @Override
+    public void replicationPropertiesChanged(Schema schema) {
+        replicationStrategys.remove(schema);
+        getReplicationStrategy(schema);
     }
 }

lealone-cluster/src/main/java/org/lealone/cluster/router/P2PRouter.java

@@ -49,6 +49,7 @@
 import org.lealone.command.router.Router;
 import org.lealone.command.router.SerializedResult;
 import org.lealone.command.router.SortedResult;
+import org.lealone.dbobject.Schema;
 import org.lealone.dbobject.table.TableFilter;
 import org.lealone.message.DbException;
 import org.lealone.result.ResultInterface;
@@ -110,9 +111,7 @@ public int executeMerge(Merge merge) {
     private static int executeInsertOrMerge(InsertOrMerge iom) {
         final String localDataCenter = DatabaseDescriptor.getEndpointSnitch()
                 .getDatacenter(Utils.getBroadcastAddress());
-        String keyspaceName = iom.getTable().getSchema().getName();
-        //AbstractReplicationStrategy rs = Keyspace.open(keyspaceName).getReplicationStrategy();
-
+        Schema schema = iom.getTable().getSchema();
         List<Row> localRows = null;
         Map<InetAddress, List<Row>> localDataCenterRows = null;
         Map<InetAddress, List<Row>> remoteDataCenterRows = null;
@@ -124,9 +123,9 @@ private static int executeInsertOrMerge(InsertOrMerge iom) {
             if (partitionKey == null)
                 partitionKey = ValueUuid.getNewRandom();
             Token tk = StorageService.getPartitioner().getToken(ByteBuffer.wrap(partitionKey.getBytesNoCopy()));
-            List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(keyspaceName, tk);
+            List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(schema, tk);
             Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetaData().pendingEndpointsFor(
-                    tk, keyspaceName);
+                    tk, schema.getFullName());
 
             Iterable<InetAddress> targets = Iterables.concat(naturalEndpoints, pendingEndpoints);
             for (InetAddress destination : targets) {
@@ -350,11 +349,11 @@ private static List<InetAddress> getTargetEndpointsIfEqual(TableFilter tableFilt
         Value endPK = getPartitionKey(endRow);
 
         if (startPK != null && endPK != null && startPK == endPK) {
-            String keyspaceName = tableFilter.getTable().getSchema().getName();
+            Schema schema = tableFilter.getTable().getSchema();
             Token tk = StorageService.getPartitioner().getToken(ByteBuffer.wrap(startPK.getBytesNoCopy()));
-            List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(keyspaceName, tk);
+            List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(schema, tk);
             Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetaData().pendingEndpointsFor(
-                    tk, keyspaceName);
+                    tk, schema.getFullName());
 
             naturalEndpoints.addAll(pendingEndpoints);
             return naturalEndpoints;

lealone-cluster/src/main/java/org/lealone/cluster/service/StorageService.java

@@ -69,6 +69,7 @@
 import org.lealone.cluster.utils.Pair;
 import org.lealone.cluster.utils.Utils;
 import org.lealone.cluster.utils.WrappedRunnable;
+import org.lealone.dbobject.Schema;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -484,8 +485,8 @@ private void setTokens(Collection<Token> tokens) {
         setMode(Mode.NORMAL, false);
     }
 
-    public Collection<Range<Token>> getLocalRanges(String keyspaceName) {
-        return getRangesForEndpoint(keyspaceName, Utils.getBroadcastAddress());
+    public Collection<Range<Token>> getLocalRanges(Schema schema) {
+        return getRangesForEndpoint(schema, Utils.getBroadcastAddress());
     }
 
     public void register(IEndpointLifecycleSubscriber subscriber) {
@@ -1135,8 +1136,8 @@ private List<String> stringify(Iterable<InetAddress> endpoints) {
      * @param ep endpoint we are interested in.
      * @return ranges for the specified endpoint.
      */
-    Collection<Range<Token>> getRangesForEndpoint(String keyspaceName, InetAddress ep) {
-        return Keyspace.open(keyspaceName).getReplicationStrategy().getAddressRanges().get(ep);
+    Collection<Range<Token>> getRangesForEndpoint(Schema schema, InetAddress ep) {
+        return Keyspace.getReplicationStrategy(schema).getAddressRanges().get(ep);
     }
 
     /**
@@ -1163,36 +1164,36 @@ public List<Range<Token>> getAllRanges(List<Token> sortedTokens) {
         return ranges;
     }
 
-    public List<InetAddress> getNaturalEndpoints(String keyspaceName, ByteBuffer key) {
-        return getNaturalEndpoints(keyspaceName, getPartitioner().getToken(key));
+    public List<InetAddress> getNaturalEndpoints(Schema schema, ByteBuffer key) {
+        return getNaturalEndpoints(schema, getPartitioner().getToken(key));
     }
 
     /**
      * This method returns the N endpoints that are responsible for storing the
      * specified key i.e for replication.
      *
-     * @param keyspaceName keyspace name also known as keyspace
+     * @param schema the schema
      * @param pos position for which we need to find the endpoint
      * @return the endpoint responsible for this token
      */
-    public List<InetAddress> getNaturalEndpoints(String keyspaceName, RingPosition<?> pos) {
-        return Keyspace.open(keyspaceName).getReplicationStrategy().getNaturalEndpoints(pos);
+    public List<InetAddress> getNaturalEndpoints(Schema schema, RingPosition<?> pos) {
+        return Keyspace.getReplicationStrategy(schema).getNaturalEndpoints(pos);
     }
 
     /**
      * This method attempts to return N endpoints that are responsible for storing the
      * specified key i.e for replication.
      *
-     * @param keyspace keyspace name also known as keyspace
+     * @param schema the schema
      * @param key key for which we need to find the endpoint
      * @return the endpoint responsible for this key
      */
-    public List<InetAddress> getLiveNaturalEndpoints(Keyspace keyspace, ByteBuffer key) {
-        return getLiveNaturalEndpoints(keyspace, getPartitioner().decorateKey(key));
+    public List<InetAddress> getLiveNaturalEndpoints(Schema schema, ByteBuffer key) {
+        return getLiveNaturalEndpoints(schema, getPartitioner().decorateKey(key));
     }
 
-    public List<InetAddress> getLiveNaturalEndpoints(Keyspace keyspace, RingPosition<?> pos) {
-        List<InetAddress> endpoints = keyspace.getReplicationStrategy().getNaturalEndpoints(pos);
+    public List<InetAddress> getLiveNaturalEndpoints(Schema schema, RingPosition<?> pos) {
+        List<InetAddress> endpoints = Keyspace.getReplicationStrategy(schema).getNaturalEndpoints(pos);
         List<InetAddress> liveEps = new ArrayList<>(endpoints.size());
 
         for (InetAddress endpoint : endpoints) {

lealone-common/src/main/java/org/lealone/command/CommandInterface.java

@@ -448,6 +448,11 @@ public interface CommandInterface {
      */
     int SHUTDOWN_DEFRAG = 84;
 
+    /**
+     * The type of a ALTER SCHEMA WTIH REPLICATION statement.
+     */
+    int ALTER_SCHEMA_WTIH_REPLICATION = 85;
+
     /**
      * Get command type.
      *

lealone-sql/src/main/java/org/lealone/command/Parser.java

@@ -11,12 +11,15 @@
 import java.nio.charset.Charset;
 import java.text.Collator;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 
 import org.lealone.api.ErrorCode;
 import org.lealone.api.Trigger;
 import org.lealone.command.ddl.AlterIndexRename;
 import org.lealone.command.ddl.AlterSchemaRename;
+import org.lealone.command.ddl.AlterSchemaWithReplication;
 import org.lealone.command.ddl.AlterSequence;
 import org.lealone.command.ddl.AlterTableAddConstraint;
 import org.lealone.command.ddl.AlterTableAlterColumn;
@@ -4093,9 +4096,31 @@ private CreateSchema parseCreateSchema() {
         } else {
             command.setAuthorization(session.getUser().getName());
         }
+
+        if (readIf("WITH")) {
+            read("REPLICATION");
+            read("=");
+            Map<String, String> replicationProperties = parseMap();
+            command.setReplicationProperties(replicationProperties);
+            checkReplicationProperties(replicationProperties);
+        }
+
         return command;
     }
 
+    private Map<String, String> parseMap() {
+        Map<String, String> map = new HashMap<>();
+        read("(");
+        while (!readIf(")")) {
+            String k = readString();
+            read(":");
+            String v = readString();
+            readIf(",");
+            map.put(k, v);
+        }
+        return map;
+    }
+
     private CreateSequence parseCreateSequence() {
         boolean ifNotExists = readIfNoExists();
         String sequenceName = readIdentifierWithSchema();
@@ -4431,8 +4456,27 @@ private AlterView parseAlterView() {
         return command;
     }
 
-    private AlterSchemaRename parseAlterSchema() {
+    private DefineCommand parseAlterSchema() {
         String schemaName = readIdentifierWithSchema();
+        if (readIf("WITH")) {
+            read("REPLICATION");
+            read("=");
+            AlterSchemaWithReplication command = new AlterSchemaWithReplication(session);
+            command.setSchema(getSchema(schemaName));
+            Map<String, String> replicationProperties = parseMap();
+            command.setReplicationProperties(replicationProperties);
+            checkReplicationProperties(replicationProperties);
+            return command;
+        }
+        return parseAlterSchemaRename(schemaName);
+    }
+
+    private void checkReplicationProperties(Map<String, String> replicationProperties) {
+        if (replicationProperties != null && !replicationProperties.containsKey("class"))
+            throw DbException.get(ErrorCode.SYNTAX_ERROR_1, sqlCommand + ", missing replication strategy class");
+    }
+
+    private AlterSchemaRename parseAlterSchemaRename(String schemaName) {
         Schema old = getSchema();
         AlterSchemaRename command = new AlterSchemaRename(session);
         command.setOldSchema(getSchema(schemaName));

lealone-sql/src/main/java/org/lealone/command/ddl/AlterSchemaRename.java

@@ -37,6 +37,7 @@ public void setNewName(String name) {
         newSchemaName = name;
     }
 
+    @Override
     public int update() {
         session.commit(true);
         Database db = session.getDatabase();
@@ -55,6 +56,7 @@ public int update() {
         return 0;
     }
 
+    @Override
     public int getType() {
         return CommandInterface.ALTER_SCHEMA_RENAME;
     }