From 1866e3802bb16dd582cf5a95bc41bb13e808a0ba Mon Sep 17 00:00:00 2001
From: Jonathan Ellis <jbellis@apache.org>
Date: Fri, 27 Mar 2009 02:45:18 +0000
Subject: [PATCH] allow user-specified Partitioners

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@759004 13f79535-47bb-0310-9956-ffa450edef68
---
 conf/storage-conf.xml                               |  8 +++++++-
 .../apache/cassandra/config/DatabaseDescriptor.java |  8 ++++----
 .../apache/cassandra/service/StorageService.java    | 13 +++++++------
 3 files changed, 18 insertions(+), 11 deletions(-)

diff --git a/conf/storage-conf.xml b/conf/storage-conf.xml
index c38812e360..45c0867168 100644
--- a/conf/storage-conf.xml
+++ b/conf/storage-conf.xml
@@ -1,6 +1,12 @@
 <Storage>
    <ClusterName>Test Cluster</ClusterName>
-   <HashingStrategy>RANDOM</HashingStrategy>
+   <!-- any IPartitioner may be used, including your own
+        as long as it is on the classpath.  Out of the box,
+        Cassandra provides
+        org.apache.cassandra.dht.RandomPartitioner and
+        org.apache.cassandra.dht.OrderPreservingPartitioner.
+        Range queries require using OrderPreservingPartitioner or a subclass. -->
+   <Partitioner>org.apache.cassandra.dht.RandomPartitioner</Partitioner>
    <RackAware>false</RackAware>
    <MulticastChannel>230.0.0.1</MulticastChannel>
    <ReplicationFactor>1</ReplicationFactor>
diff --git a/src/org/apache/cassandra/config/DatabaseDescriptor.java b/src/org/apache/cassandra/config/DatabaseDescriptor.java
index f906b3eb85..b7f1235c04 100644
--- a/src/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -89,7 +89,7 @@ public class DatabaseDescriptor
     */
     private static Map<String, Map<String, CFMetaData>> tableToCFMetaDataMap_;
     /* Hashing strategy Random or OPHF */
-    private static String hashingStrategy_ = DatabaseDescriptor.random_;
+    private static String partitionerClass_;
     /* if the size of columns or super-columns are more than this, indexing will kick in */
     private static int columnIndexSizeInKB_;
     /* Size of touch key cache */
@@ -138,7 +138,7 @@ public class DatabaseDescriptor
             zkAddress_ = xmlUtils.getNodeValue("/Storage/ZookeeperAddress");
 
             /* Hashing strategy */
-            hashingStrategy_ = xmlUtils.getNodeValue("/Storage/HashingStrategy");
+            partitionerClass_ = xmlUtils.getNodeValue("/Storage/Partitioner");
             /* Callout location */
             calloutLocation_ = xmlUtils.getNodeValue("/Storage/CalloutLocation");
 
@@ -469,9 +469,9 @@ public class DatabaseDescriptor
 
 
     
-    public static String getHashingStrategy()
+    public static String getPartitionerClass()
     {
-        return hashingStrategy_;
+        return partitionerClass_;
     }
     
     public static String getZkAddress()
diff --git a/src/org/apache/cassandra/service/StorageService.java b/src/org/apache/cassandra/service/StorageService.java
index 95de5c6a27..4b9dffd6f0 100644
--- a/src/org/apache/cassandra/service/StorageService.java
+++ b/src/org/apache/cassandra/service/StorageService.java
@@ -21,6 +21,7 @@ package org.apache.cassandra.service;
 import java.io.File;
 import java.io.IOException;
 import java.lang.management.ManagementFactory;
+import java.lang.reflect.InvocationTargetException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -434,14 +435,14 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
 
     static
     {
-        String hashingStrategy = DatabaseDescriptor.getHashingStrategy();
-        if (DatabaseDescriptor.ophf_.equalsIgnoreCase(hashingStrategy))
+        try
         {
-            partitioner_ = new OrderPreservingPartitioner();
-        }        
-        else
+            Class cls = Class.forName(DatabaseDescriptor.getPartitionerClass());
+            partitioner_ = (IPartitioner) cls.getConstructor().newInstance();
+        }
+        catch (Exception e)
         {
-            partitioner_ = new RandomPartitioner();
+            throw new RuntimeException(e);
         }
     }