Extract BINARY column size schema details #480 (#481)

Co-authored-by: siddontang <[email protected]>

Author: ckolbitsch-work

schema/schema.go

@@ -7,6 +7,7 @@ package schema
 import (
 	"database/sql"
 	"fmt"
+	"strconv"
 	"strings"
 
 	"github.com/pingcap/errors"
@@ -23,7 +24,7 @@ const (
 	TYPE_FLOAT                // float, double
 	TYPE_ENUM                 // enum
 	TYPE_SET                  // set
-	TYPE_STRING               // other
+	TYPE_STRING               // char, varchar, etc.
 	TYPE_DATETIME             // datetime
 	TYPE_TIMESTAMP            // timestamp
 	TYPE_DATE                 // date
@@ -32,6 +33,7 @@ const (
 	TYPE_JSON                 // json
 	TYPE_DECIMAL              // decimal
 	TYPE_MEDIUM_INT
+	TYPE_BINARY               // binary, varbinary
 )
 
 type TableColumn struct {
@@ -43,6 +45,8 @@ type TableColumn struct {
 	IsUnsigned bool
 	EnumValues []string
 	SetValues  []string
+	FixedSize  uint
+	MaxSize    uint
 }
 
 type Index struct {
@@ -94,6 +98,14 @@ func (ta *Table) AddColumn(name string, columnType string, collation string, ext
 				")"),
 			"'", "", -1),
 			",")
+	} else if strings.HasPrefix(columnType, "binary") {
+		ta.Columns[index].Type = TYPE_BINARY
+		size := getSizeFromColumnType(columnType)
+		ta.Columns[index].MaxSize = size
+		ta.Columns[index].FixedSize = size
+	} else if strings.HasPrefix(columnType, "varbinary") {
+		ta.Columns[index].Type = TYPE_BINARY
+		ta.Columns[index].MaxSize = getSizeFromColumnType(columnType)
 	} else if strings.HasPrefix(columnType, "datetime") {
 		ta.Columns[index].Type = TYPE_DATETIME
 	} else if strings.HasPrefix(columnType, "timestamp") {
@@ -110,8 +122,14 @@ func (ta *Table) AddColumn(name string, columnType string, collation string, ext
 		ta.Columns[index].Type = TYPE_MEDIUM_INT
 	} else if strings.Contains(columnType, "int") || strings.HasPrefix(columnType, "year") {
 		ta.Columns[index].Type = TYPE_NUMBER
+	} else if strings.HasPrefix(columnType, "char") {
+		ta.Columns[index].Type = TYPE_STRING
+		size := getSizeFromColumnType(columnType)
+		ta.Columns[index].FixedSize = size
+		ta.Columns[index].MaxSize = size
 	} else {
 		ta.Columns[index].Type = TYPE_STRING
+		ta.Columns[index].MaxSize = getSizeFromColumnType(columnType)
 	}
 
 	if strings.Contains(columnType, "unsigned") || strings.Contains(columnType, "zerofill") {
@@ -124,6 +142,27 @@ func (ta *Table) AddColumn(name string, columnType string, collation string, ext
 	}
 }
 
+func getSizeFromColumnType(columnType string) uint {
+	startIndex := strings.Index(columnType, "(")
+	if startIndex < 0 {
+		return 0
+	}
+
+	// we are searching for the first () and there may not be any closing
+	// brackets before the opening, so no need search at the offset from the
+	// opening ones
+	endIndex := strings.Index(columnType, ")")
+	if startIndex < 0 || endIndex < 0 || startIndex > endIndex {
+		return 0
+	}
+
+	i, err := strconv.Atoi(columnType[startIndex+1:endIndex])
+	if err != nil || i < 0 {
+		return 0
+	}
+	return uint(i)
+}
+
 func (ta *Table) FindColumn(name string) int {
 	for i, col := range ta.Columns {
 		if col.Name == name {

schema/schema_test.go

@@ -62,6 +62,9 @@ func (s *schemaTestSuite) TestSchema(c *C) {
             zfint INT ZEROFILL,
             name_ucs VARCHAR(256) CHARACTER SET ucs2,
             name_utf8 VARCHAR(256) CHARACTER SET utf8,
+            name_char CHAR(10),
+            name_binary BINARY(11),
+            name_varbinary VARBINARY(12),
             PRIMARY KEY(id2, id),
             UNIQUE (id1),
             INDEX name_idx (name)
@@ -74,20 +77,34 @@ func (s *schemaTestSuite) TestSchema(c *C) {
 	ta, err := NewTable(s.conn, "test", "schema_test")
 	c.Assert(err, IsNil)
 
-	c.Assert(ta.Columns, HasLen, 12)
+	c.Assert(ta.Columns, HasLen, 15)
 	c.Assert(ta.Indexes, HasLen, 3)
 	c.Assert(ta.PKColumns, DeepEquals, []int{2, 0})
 	c.Assert(ta.Indexes[0].Columns, HasLen, 2)
 	c.Assert(ta.Indexes[0].Name, Equals, "PRIMARY")
 	c.Assert(ta.Indexes[2].Name, Equals, "name_idx")
+	c.Assert(ta.Columns[3].Type, Equals, TYPE_STRING)
+	c.Assert(ta.Columns[3].MaxSize, Equals, uint(256))
+	c.Assert(ta.Columns[3].FixedSize, Equals, uint(0))
 	c.Assert(ta.Columns[4].EnumValues, DeepEquals, []string{"appointing", "serving", "abnormal", "stop", "noaftermarket", "finish", "financial_audit"})
 	c.Assert(ta.Columns[5].SetValues, DeepEquals, []string{"a", "b", "c"})
 	c.Assert(ta.Columns[7].Type, Equals, TYPE_DECIMAL)
 	c.Assert(ta.Columns[0].IsUnsigned, IsFalse)
 	c.Assert(ta.Columns[8].IsUnsigned, IsTrue)
 	c.Assert(ta.Columns[9].IsUnsigned, IsTrue)
 	c.Assert(ta.Columns[10].Collation, Matches, "^ucs2.*")
+	c.Assert(ta.Columns[10].MaxSize, Equals, uint(256))
+	c.Assert(ta.Columns[10].FixedSize, Equals, uint(0))
 	c.Assert(ta.Columns[11].Collation, Matches, "^utf8.*")
+	c.Assert(ta.Columns[12].Type, Equals, TYPE_STRING)
+	c.Assert(ta.Columns[12].MaxSize, Equals, uint(10))
+	c.Assert(ta.Columns[12].FixedSize, Equals, uint(10))
+	c.Assert(ta.Columns[13].Type, Equals, TYPE_BINARY)
+	c.Assert(ta.Columns[13].MaxSize, Equals, uint(11))
+	c.Assert(ta.Columns[13].FixedSize, Equals, uint(11))
+	c.Assert(ta.Columns[14].Type, Equals, TYPE_BINARY)
+	c.Assert(ta.Columns[14].MaxSize, Equals, uint(12))
+	c.Assert(ta.Columns[14].FixedSize, Equals, uint(0))
 
 	taSqlDb, err := NewTableFromSqlDB(s.sqlDB, "test", "schema_test")
 	c.Assert(err, IsNil)