HBASE-25766 Introduce RegionSplitRestriction that restricts the pattern of the split point

Signed-off-by: Duo Zhang <zhangduo@apache.org>
Signed-off-by: Michael Stack <stack@apache.org>

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/SplitTableRegionProcedure.java

@@ -62,6 +62,7 @@ import org.apache.hadoop.hbase.regionserver.HRegionFileSystem;
 import org.apache.hadoop.hbase.regionserver.HStore;
 import org.apache.hadoop.hbase.regionserver.HStoreFile;
 import org.apache.hadoop.hbase.regionserver.RegionSplitPolicy;
+import org.apache.hadoop.hbase.regionserver.RegionSplitRestriction;
 import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
@@ -110,6 +111,21 @@ public class SplitTableRegionProcedure
     // we fail-fast on construction. There it skips the split with just a warning.
     checkOnline(env, regionToSplit);
     this.bestSplitRow = splitRow;
+    TableDescriptor tableDescriptor = env.getMasterServices().getTableDescriptors()
+      .get(getTableName());
+    Configuration conf = env.getMasterConfiguration();
+    if (hasBestSplitRow()) {
+      // Apply the split restriction for the table to the user-specified split point
+      RegionSplitRestriction splitRestriction =
+        RegionSplitRestriction.create(tableDescriptor, conf);
+      byte[] restrictedSplitRow = splitRestriction.getRestrictedSplitPoint(bestSplitRow);
+      if (!Bytes.equals(bestSplitRow, restrictedSplitRow)) {
+        LOG.warn("The specified split point {} violates the split restriction of the table. "
+            + "Using {} as a split point.", Bytes.toStringBinary(bestSplitRow),
+          Bytes.toStringBinary(restrictedSplitRow));
+        bestSplitRow = restrictedSplitRow;
+      }
+    }
     checkSplittable(env, regionToSplit);
     final TableName table = regionToSplit.getTable();
     final long rid = getDaughterRegionIdTimestamp(regionToSplit);
@@ -125,15 +141,14 @@ public class SplitTableRegionProcedure
         .setSplit(false)
         .setRegionId(rid)
         .build();
-    TableDescriptor htd = env.getMasterServices().getTableDescriptors().get(getTableName());
+    if(tableDescriptor.getRegionSplitPolicyClassName() != null) {
-    if(htd.getRegionSplitPolicyClassName() != null) {
       // Since we don't have region reference here, creating the split policy instance without it.
       // This can be used to invoke methods which don't require Region reference. This instantiation
       // of a class on Master-side though it only makes sense on the RegionServer-side is
       // for Phoenix Local Indexing. Refer HBASE-12583 for more information.
       Class<? extends RegionSplitPolicy> clazz =
-          RegionSplitPolicy.getSplitPolicyClass(htd, env.getMasterConfiguration());
+        RegionSplitPolicy.getSplitPolicyClass(tableDescriptor, conf);
-      this.splitPolicy = ReflectionUtils.newInstance(clazz, env.getMasterConfiguration());
+      this.splitPolicy = ReflectionUtils.newInstance(clazz, conf);
     }
   }
 
@@ -219,7 +234,7 @@ public class SplitTableRegionProcedure
       throw e;
     }
 
-    if (bestSplitRow == null || bestSplitRow.length == 0) {
+    if (!hasBestSplitRow()) {
       throw new DoNotRetryIOException("Region not splittable because bestSplitPoint = null, " +
         "maybe table is too small for auto split. For force split, try specifying split row");
     }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DelimitedKeyPrefixRegionSplitPolicy.java

@@ -37,7 +37,11 @@ import org.apache.hadoop.hbase.util.Bytes;
  * <code>userid_eventtype_eventid</code>, and use prefix delimiter _, this split policy
  * ensures that all rows starting with the same userid, belongs to the same region.
  * @see KeyPrefixRegionSplitPolicy
+ *
+ * @deprecated since 2.5.0 and will be removed in 4.0.0. Use {@link RegionSplitRestriction},
+ *   instead.
  */
+@Deprecated
 @InterfaceAudience.Private
 public class DelimitedKeyPrefixRegionSplitPolicy extends IncreasingToUpperBoundRegionSplitPolicy {
 

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/DelimitedKeyPrefixRegionSplitRestriction.java

@@ -0,0 +1,84 @@
+/*
+ * 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.hadoop.hbase.regionserver;
+
+import java.io.IOException;
+import java.util.Arrays;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A {@link RegionSplitRestriction} implementation that groups rows by a prefix of the row-key with
+ * a delimiter. Only the first delimiter for the row key will define the prefix of the row key that
+ * is used for grouping.
+ * <p>
+ * This ensures that a region is not split "inside" a prefix of a row key.
+ * I.e. rows can be co-located in a region by their prefix.
+ *
+ * As an example, if you have row keys delimited with <code>_</code>, like
+ * <code>userid_eventtype_eventid</code>, and use prefix delimiter _, this split policy ensures
+ * that all rows starting with the same userid, belongs to the same region.
+ */
+@InterfaceAudience.Private
+public class DelimitedKeyPrefixRegionSplitRestriction extends RegionSplitRestriction {
+  private static final Logger LOG =
+    LoggerFactory.getLogger(DelimitedKeyPrefixRegionSplitRestriction.class);
+
+  public static final String DELIMITER_KEY =
+    "hbase.regionserver.region.split_restriction.delimiter";
+
+  private byte[] delimiter = null;
+
+  @Override
+  public void initialize(TableDescriptor tableDescriptor, Configuration conf) throws IOException {
+    String delimiterString = tableDescriptor.getValue(DELIMITER_KEY);
+    if (delimiterString == null || delimiterString.length() == 0) {
+      delimiterString = conf.get(DELIMITER_KEY);
+      if (delimiterString == null || delimiterString.length() == 0) {
+        LOG.error("{} not specified for table {}. "
+          + "Using the default RegionSplitRestriction", DELIMITER_KEY,
+          tableDescriptor.getTableName());
+        return;
+      }
+    }
+    delimiter = Bytes.toBytes(delimiterString);
+  }
+
+  @Override
+  public byte[] getRestrictedSplitPoint(byte[] splitPoint) {
+    if (delimiter != null) {
+      // find the first occurrence of delimiter in split point
+      int index = org.apache.hbase.thirdparty.com.google.common.primitives.Bytes.indexOf(
+        splitPoint, delimiter);
+      if (index < 0) {
+        LOG.warn("Delimiter {} not found for split key {}", Bytes.toString(delimiter),
+          Bytes.toStringBinary(splitPoint));
+        return splitPoint;
+      }
+
+      // group split keys by a prefix
+      return Arrays.copyOf(splitPoint, Math.min(index, splitPoint.length));
+    } else {
+      return splitPoint;
+    }
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java

@@ -697,6 +697,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
 
   private TableDescriptor htableDescriptor = null;
   private RegionSplitPolicy splitPolicy;
+  private RegionSplitRestriction splitRestriction;
   private FlushPolicy flushPolicy;
 
   private final MetricsRegion metricsRegion;
@@ -1042,6 +1043,9 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
     // Initialize split policy
     this.splitPolicy = RegionSplitPolicy.create(this, conf);
 
+    // Initialize split restriction
+    splitRestriction = RegionSplitRestriction.create(getTableDescriptor(), conf);
+
     // Initialize flush policy
     this.flushPolicy = FlushPolicyFactory.create(this, conf);
 
@@ -8090,6 +8094,9 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
     }
 
     byte[] ret = splitPolicy.getSplitPoint();
+    if (ret != null && ret.length > 0) {
+      ret = splitRestriction.getRestrictedSplitPoint(ret);
+    }
 
     if (ret != null) {
       try {

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyPrefixRegionSplitPolicy.java

@@ -29,7 +29,11 @@ import org.slf4j.LoggerFactory;
  *
  * This ensures that a region is not split "inside" a prefix of a row key.
  * I.e. rows can be co-located in a region by their prefix.
+ *
+ * @deprecated since 2.5.0 and will be removed in 4.0.0. Use {@link RegionSplitRestriction},
+ *   instead.
  */
+@Deprecated
 @InterfaceAudience.Private
 public class KeyPrefixRegionSplitPolicy extends IncreasingToUpperBoundRegionSplitPolicy {
   private static final Logger LOG = LoggerFactory

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/KeyPrefixRegionSplitRestriction.java

@@ -0,0 +1,76 @@
+/*
+ * 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.hadoop.hbase.regionserver;
+
+import java.io.IOException;
+import java.util.Arrays;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A {@link RegionSplitRestriction} implementation that groups rows by a prefix of the row-key.
+ * <p>
+ * This ensures that a region is not split "inside" a prefix of a row key.
+ * I.e. rows can be co-located in a region by their prefix.
+ */
+@InterfaceAudience.Private
+public class KeyPrefixRegionSplitRestriction extends RegionSplitRestriction {
+  private static final Logger LOG =
+    LoggerFactory.getLogger(KeyPrefixRegionSplitRestriction.class);
+
+  public static final String PREFIX_LENGTH_KEY =
+    "hbase.regionserver.region.split_restriction.prefix_length";
+
+  private int prefixLength;
+
+  @Override
+  public void initialize(TableDescriptor tableDescriptor, Configuration conf) throws IOException {
+    String prefixLengthString = tableDescriptor.getValue(PREFIX_LENGTH_KEY);
+    if (prefixLengthString == null) {
+      prefixLengthString = conf.get(PREFIX_LENGTH_KEY);
+      if (prefixLengthString == null) {
+        LOG.error("{} not specified for table {}. "
+          + "Using the default RegionSplitRestriction", PREFIX_LENGTH_KEY,
+          tableDescriptor.getTableName());
+        return;
+      }
+    }
+    try {
+      prefixLength = Integer.parseInt(prefixLengthString);
+    } catch (NumberFormatException ignored) {
+    }
+    if (prefixLength <= 0) {
+      LOG.error("Invalid value for {} for table {}:{}. "
+        + "Using the default RegionSplitRestriction", PREFIX_LENGTH_KEY,
+        tableDescriptor.getTableName(), prefixLengthString);
+    }
+  }
+
+  @Override
+  public byte[] getRestrictedSplitPoint(byte[] splitPoint) {
+    if (prefixLength > 0) {
+      // group split keys by a prefix
+      return Arrays.copyOf(splitPoint, Math.min(prefixLength, splitPoint.length));
+    } else {
+      return splitPoint;
+    }
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/NoRegionSplitRestriction.java

@@ -0,0 +1,40 @@
+/*
+ * 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.hadoop.hbase.regionserver;
+
+import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * A {@link RegionSplitRestriction} implementation that does nothing.
+ */
+@InterfaceAudience.Private
+public class NoRegionSplitRestriction extends RegionSplitRestriction {
+
+  @Override
+  public void initialize(TableDescriptor tableDescriptor, Configuration conf) throws IOException {
+  }
+
+  @Override
+  public byte[] getRestrictedSplitPoint(byte[] splitPoint) {
+    // Do nothing
+    return splitPoint;
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionSplitRestriction.java

@@ -0,0 +1,129 @@
+/*
+ * 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.hadoop.hbase.regionserver;
+
+import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A split restriction that restricts the pattern of the split point.
+ * <p>
+ * The difference between {@link RegionSplitPolicy} and RegionSplitRestriction is that
+ * RegionSplitRestriction defines how to split while {@link RegionSplitPolicy} defines when we need
+ * to split.
+ * <p>
+ * We can specify a split restriction, "KeyPrefix" or "DelimitedKeyPrefix", to a table with the
+ * "hbase.regionserver.region.split_restriction.type" property. The "KeyPrefix" split restriction
+ * groups rows by a prefix of the row-key. And the "DelimitedKeyPrefix" split restriction groups
+ * rows by a prefix of the row-key with a delimiter.
+ *
+ * For example:
+ * <pre>
+ * <code>
+ * # Create a table with a "KeyPrefix" split restriction, where the prefix length is 2 bytes
+ * hbase> create 'tbl1', 'fam',
+ *   {CONFIGURATION => {'hbase.regionserver.region.split_restriction.type' => 'KeyPrefix',
+ *                      'hbase.regionserver.region.split_restriction.prefix_length' => '2'}}
+ *
+ * # Create a table with a "DelimitedKeyPrefix" split restriction, where the delimiter is a comma
+ * hbase> create 'tbl2', 'fam',
+ *   {CONFIGURATION => {'hbase.regionserver.region.split_restriction.type' => 'DelimitedKeyPrefix',
+ *                      'hbase.regionserver.region.split_restriction.delimiter' => ','}}
+ * </code>
+ * </pre>
+ *
+ * Instead of specifying a split restriction to a table directly, we can also set the properties
+ * in hbase-site.xml. In this case, the specified split restriction is applied for all the tables.
+ * <p>
+ * Note that the split restriction is also applied to a user-specified split point so that we don't
+ * allow users to break the restriction.
+ *
+ * @see NoRegionSplitRestriction
+ * @see KeyPrefixRegionSplitRestriction
+ * @see DelimitedKeyPrefixRegionSplitRestriction
+ */
+@InterfaceAudience.Private
+public abstract class RegionSplitRestriction {
+  private static final Logger LOG = LoggerFactory.getLogger(RegionSplitRestriction.class);
+
+  public static final String RESTRICTION_TYPE_KEY =
+    "hbase.regionserver.region.split_restriction.type";
+
+  public static final String RESTRICTION_TYPE_NONE = "None";
+  public static final String RESTRICTION_TYPE_KEY_PREFIX = "KeyPrefix";
+  public static final String RESTRICTION_TYPE_DELIMITED_KEY_PREFIX = "DelimitedKeyPrefix";
+
+  /**
+   * Create the RegionSplitRestriction configured for the given table.
+   *
+   * @param tableDescriptor the table descriptor
+   * @param conf the configuration
+   * @return a RegionSplitRestriction instance
+   * @throws IOException if an error occurs
+   */
+  public static RegionSplitRestriction create(TableDescriptor tableDescriptor,
+    Configuration conf) throws IOException {
+    String type = tableDescriptor.getValue(RESTRICTION_TYPE_KEY);
+    if (type == null) {
+      type = conf.get(RESTRICTION_TYPE_KEY, RESTRICTION_TYPE_NONE);
+    }
+
+    RegionSplitRestriction ret;
+    switch (type) {
+      case RESTRICTION_TYPE_NONE:
+        ret = new NoRegionSplitRestriction();
+        break;
+      case RESTRICTION_TYPE_KEY_PREFIX:
+        ret = new KeyPrefixRegionSplitRestriction();
+        break;
+      case RESTRICTION_TYPE_DELIMITED_KEY_PREFIX:
+        ret = new DelimitedKeyPrefixRegionSplitRestriction();
+        break;
+      default:
+        LOG.warn("Invalid RegionSplitRestriction type specified: {}. "
+          + "Using the default RegionSplitRestriction", type);
+        ret = new NoRegionSplitRestriction();
+        break;
+    }
+    ret.initialize(tableDescriptor, conf);
+    return ret;
+  }
+
+  /**
+   * Initialize the RegionSplitRestriction instance
+   *
+   * @param tableDescriptor the table descriptor
+   * @param conf the configuration
+   * @throws IOException if an error occurs
+   */
+  public abstract void initialize(TableDescriptor tableDescriptor, Configuration conf)
+    throws IOException;
+
+  /**
+   * Returns a restricted split point.
+   *
+   * @param splitPoint the split point determined by {@link RegionSplitPolicy} or specified by a
+   *   user manually
+   * @return the restricted split point
+   */
+  public abstract byte[] getRestrictedSplitPoint(byte[] splitPoint);
+}

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionSplitRestriction.java

@@ -0,0 +1,184 @@
+/*
+ * 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.hadoop.hbase.regionserver;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.client.TableDescriptor;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+@Category({ RegionServerTests.class, SmallTests.class })
+public class TestRegionSplitRestriction {
+
+  @ClassRule
+  public static final HBaseClassTestRule CLASS_RULE =
+    HBaseClassTestRule.forClass(TestRegionSplitRestriction.class);
+
+  Configuration conf;
+  @Mock TableDescriptor tableDescriptor;
+
+  @Before
+  public void setup() {
+    MockitoAnnotations.initMocks(this);
+
+    conf = new Configuration();
+  }
+
+  @Test
+  public void testWhenTableDescriptorReturnsNoneType() throws IOException {
+    when(tableDescriptor.getValue(RegionSplitRestriction.RESTRICTION_TYPE_KEY))
+      .thenReturn(RegionSplitRestriction.RESTRICTION_TYPE_NONE);
+
+    RegionSplitRestriction splitRestriction =
+      RegionSplitRestriction.create(tableDescriptor, conf);
+    assertTrue(splitRestriction instanceof NoRegionSplitRestriction);
+  }
+
+  @Test
+  public void testWhenTableDescriptorReturnsKeyPrefixType() throws IOException {
+    when(tableDescriptor.getValue(RegionSplitRestriction.RESTRICTION_TYPE_KEY))
+      .thenReturn(RegionSplitRestriction.RESTRICTION_TYPE_KEY_PREFIX);
+
+    RegionSplitRestriction splitRestriction =
+      RegionSplitRestriction.create(tableDescriptor, conf);
+    assertTrue(splitRestriction instanceof KeyPrefixRegionSplitRestriction);
+  }
+
+  @Test
+  public void testWhenTableDescriptorReturnsDelimitedKeyPrefixType() throws IOException {
+    when(tableDescriptor.getValue(RegionSplitRestriction.RESTRICTION_TYPE_KEY))
+      .thenReturn(RegionSplitRestriction.RESTRICTION_TYPE_DELIMITED_KEY_PREFIX);
+
+    RegionSplitRestriction splitRestriction =
+      RegionSplitRestriction.create(tableDescriptor, conf);
+    assertTrue(splitRestriction instanceof DelimitedKeyPrefixRegionSplitRestriction);
+  }
+
+  @Test
+  public void testWhenConfigurationReturnsNoneType() throws IOException {
+    conf.set(RegionSplitRestriction.RESTRICTION_TYPE_KEY,
+      RegionSplitRestriction.RESTRICTION_TYPE_NONE);
+
+    RegionSplitRestriction splitRestriction =
+      RegionSplitRestriction.create(tableDescriptor, conf);
+    assertTrue(splitRestriction instanceof NoRegionSplitRestriction);
+  }
+
+  @Test
+  public void testWhenConfigurationReturnsKeyPrefixType() throws IOException {
+    conf.set(RegionSplitRestriction.RESTRICTION_TYPE_KEY,
+      RegionSplitRestriction.RESTRICTION_TYPE_KEY_PREFIX);
+
+    RegionSplitRestriction splitRestriction =
+      RegionSplitRestriction.create(tableDescriptor, conf);
+    assertTrue(splitRestriction instanceof KeyPrefixRegionSplitRestriction);
+  }
+
+  @Test
+  public void testWhenConfigurationReturnsDelimitedKeyPrefixType() throws IOException {
+    conf.set(RegionSplitRestriction.RESTRICTION_TYPE_KEY,
+      RegionSplitRestriction.RESTRICTION_TYPE_DELIMITED_KEY_PREFIX);
+
+    RegionSplitRestriction splitRestriction =
+      RegionSplitRestriction.create(tableDescriptor, conf);
+    assertTrue(splitRestriction instanceof DelimitedKeyPrefixRegionSplitRestriction);
+  }
+
+  @Test
+  public void testWhenTableDescriptorAndConfigurationReturnNull() throws IOException {
+    RegionSplitRestriction splitRestriction =
+      RegionSplitRestriction.create(tableDescriptor, conf);
+    assertTrue(splitRestriction instanceof NoRegionSplitRestriction);
+  }
+
+  @Test
+  public void testWhenTableDescriptorReturnsInvalidType() throws IOException {
+    when(tableDescriptor.getValue(RegionSplitRestriction.RESTRICTION_TYPE_KEY))
+      .thenReturn("Invalid");
+
+    RegionSplitRestriction splitRestriction =
+      RegionSplitRestriction.create(tableDescriptor, conf);
+    assertTrue(splitRestriction instanceof NoRegionSplitRestriction);
+  }
+
+  @Test
+  public void testNoneRegionSplitRestriction() throws IOException {
+    when(tableDescriptor.getValue(RegionSplitRestriction.RESTRICTION_TYPE_KEY))
+      .thenReturn(RegionSplitRestriction.RESTRICTION_TYPE_NONE);
+
+    NoRegionSplitRestriction noRegionSplitRestriction =
+      (NoRegionSplitRestriction) RegionSplitRestriction.create(tableDescriptor, conf);
+
+    byte[] restrictedSplit =
+      noRegionSplitRestriction.getRestrictedSplitPoint(Bytes.toBytes("abcd"));
+    assertEquals("abcd", Bytes.toString(restrictedSplit));
+  }
+
+  @Test
+  public void testKeyPrefixRegionSplitRestriction() throws IOException {
+    when(tableDescriptor.getValue(RegionSplitRestriction.RESTRICTION_TYPE_KEY))
+      .thenReturn(RegionSplitRestriction.RESTRICTION_TYPE_KEY_PREFIX);
+    when(tableDescriptor.getValue(KeyPrefixRegionSplitRestriction.PREFIX_LENGTH_KEY))
+      .thenReturn("2");
+
+    KeyPrefixRegionSplitRestriction keyPrefixRegionSplitRestriction =
+      (KeyPrefixRegionSplitRestriction) RegionSplitRestriction.create(
+        tableDescriptor, conf);
+
+    byte[] restrictedSplit =
+      keyPrefixRegionSplitRestriction.getRestrictedSplitPoint(Bytes.toBytes("abcd"));
+    assertEquals("ab", Bytes.toString(restrictedSplit));
+
+    restrictedSplit =
+      keyPrefixRegionSplitRestriction.getRestrictedSplitPoint(Bytes.toBytes("a"));
+    assertEquals("a", Bytes.toString(restrictedSplit));
+  }
+
+  @Test
+  public void testDelimitedKeyPrefixRegionSplitRestriction() throws IOException {
+    when(tableDescriptor.getValue(RegionSplitRestriction.RESTRICTION_TYPE_KEY))
+      .thenReturn(RegionSplitRestriction.RESTRICTION_TYPE_DELIMITED_KEY_PREFIX);
+    when(tableDescriptor.getValue(DelimitedKeyPrefixRegionSplitRestriction.DELIMITER_KEY))
+      .thenReturn(",");
+
+    DelimitedKeyPrefixRegionSplitRestriction delimitedKeyPrefixRegionSplitRestriction =
+      (DelimitedKeyPrefixRegionSplitRestriction) RegionSplitRestriction.create(
+        tableDescriptor, conf);
+
+    byte[] restrictedSplit = delimitedKeyPrefixRegionSplitRestriction
+      .getRestrictedSplitPoint(Bytes.toBytes("ab,cd"));
+    assertEquals("ab", Bytes.toString(restrictedSplit));
+
+    restrictedSplit = delimitedKeyPrefixRegionSplitRestriction
+      .getRestrictedSplitPoint(Bytes.toBytes("ijk"));
+    assertEquals("ijk", Bytes.toString(restrictedSplit));
+  }
+}