*: fix table namespace classifier.

table/codec.go

@@ -82,9 +82,13 @@ func decodeCmpUintToInt(u uint64) int64 {
 	return int64(u ^ signMask)
 }
 
-// IsPureTableID return true iff b is consist of tablePrefix and 8-byte tableID
+// IsPureTableID returns true if b is consist of tablePrefix and 8-byte tableID
 func IsPureTableID(b []byte) bool {
-	return len(b) == len(tablePrefix)+8
+	_, key, err := decodeBytes(b)
+	if err != nil {
+		return false
+	}
+	return len(key) == len(tablePrefix)+8
 }
 
 func decodeBytes(b []byte) ([]byte, []byte, error) {

table/codec_test.go

@@ -0,0 +1,81 @@
+// Copyright 2017 PingCAP, Inc.
+//
+// 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package table
+
+import (
+	"testing"
+
+	. "github.com/pingcap/check"
+)
+
+func TestTable(t *testing.T) {
+	TestingT(t)
+}
+
+var pads = make([]byte, encGroupSize)
+
+var _ = Suite(&testCodecSuite{})
+
+type testCodecSuite struct{}
+
+func encodeBytes(data []byte) Key {
+	// Allocate more space to avoid unnecessary slice growing.
+	// Assume that the byte slice size is about `(len(data) / encGroupSize + 1) * (encGroupSize + 1)` bytes,
+	// that is `(len(data) / 8 + 1) * 9` in our implement.
+	dLen := len(data)
+	result := make([]byte, 0, (dLen/encGroupSize+1)*(encGroupSize+1))
+	for idx := 0; idx <= dLen; idx += encGroupSize {
+		remain := dLen - idx
+		padCount := 0
+		if remain >= encGroupSize {
+			result = append(result, data[idx:idx+encGroupSize]...)
+		} else {
+			padCount = encGroupSize - remain
+			result = append(result, data[idx:]...)
+			result = append(result, pads[:padCount]...)
+		}
+
+		marker := encMarker - byte(padCount)
+		result = append(result, marker)
+	}
+	return result
+}
+
+func (s *testCodecSuite) TestDecodeBytes(c *C) {
+	key := "abcdefghijklmnopqrstuvwxyz"
+	for i := 0; i < len(key); i++ {
+		_, k, err := decodeBytes(encodeBytes([]byte(key[:i])))
+		c.Assert(err, IsNil)
+		c.Assert(string(k), Equals, key[:i])
+	}
+}
+
+func (s *testCodecSuite) TestTableID(c *C) {
+	key := encodeBytes([]byte("t\x80\x00\x00\x00\x00\x00\x00\xff"))
+	c.Assert(DefaultIDDecoder.DecodeTableID(key), Equals, int64(0xff))
+	c.Assert(IsPureTableID(key), IsTrue)
+
+	key = encodeBytes([]byte("t\x80\x00\x00\x00\x00\x00\x00\xff_i\x01\x02"))
+	c.Assert(DefaultIDDecoder.DecodeTableID(key), Equals, int64(0xff))
+	c.Assert(IsPureTableID(key), IsFalse)
+
+	key = []byte("t\x80\x00\x00\x00\x00\x00\x00\xff")
+	c.Assert(DefaultIDDecoder.DecodeTableID(key), Equals, int64(0))
+
+	key = encodeBytes([]byte("T\x00\x00\x00\x00\x00\x00\x00\xff"))
+	c.Assert(DefaultIDDecoder.DecodeTableID(key), Equals, int64(0))
+
+	key = encodeBytes([]byte("t\x80\x00\x00\x00\x00\x00\xff"))
+	c.Assert(DefaultIDDecoder.DecodeTableID(key), Equals, int64(0))
+}

table/namespace_classifier.go

@@ -123,6 +123,7 @@ func (c *tableNamespaceClassifier) GetAllNamespaces() []string {
 	for name := range c.nsInfo.namespaces {
 		nsList = append(nsList, name)
 	}
+	nsList = append(nsList, namespace.DefaultNamespace)
 	return nsList
 }
 
@@ -213,6 +214,10 @@ func (c *tableNamespaceClassifier) CreateNamespace(name string) error {
 		return errors.New("Name should be 0-9, a-z or A-Z")
 	}
 
+	if name == namespace.DefaultNamespace {
+		return errors.Errorf("%s is reserved as default namespace", name)
+	}
+
 	if n := c.nsInfo.getNamespaceByName(name); n != nil {
 		return errors.New("Duplicate namespace Name")
 	}

table/namespace_classifier_test.go

@@ -14,9 +14,8 @@
 package table
 
 import (
-	"sort"
-
 	"bytes"
+	"sort"
 
 	. "github.com/pingcap/check"
 	"github.com/pingcap/kvproto/pkg/metapb"
@@ -40,7 +39,7 @@ const (
 	globalStoreID = 987
 )
 
-var tableStartKey = []byte{'t', 0, 0, 0, '1', 0, 0, 0, 0}
+var tableStartKey = encodeBytes([]byte{'t', 0, 0, 0, '1', 0, 0, 0, 0})
 
 func (d mockTableIDDecoderForTarget) DecodeTableID(key Key) int64 {
 	return targetTableID
@@ -105,7 +104,7 @@ func (s *testTableNamespaceSuite) TestTableNameSpaceGetAllNamespace(c *C) {
 	classifier := s.newClassifier(c, mockTableIDDecoderForTarget{})
 	ns := classifier.GetAllNamespaces()
 	sort.Strings(ns)
-	c.Assert(ns, DeepEquals, []string{"test1", "test2"})
+	c.Assert(ns, DeepEquals, []string{"global", "test1", "test2"})
 }
 
 func (s *testTableNamespaceSuite) TestTableNameSpaceGetStoreNamespace(c *C) {