From 8f7fa8c2d8dcade2376fed03f4770302ebd971db Mon Sep 17 00:00:00 2001
From: siddontang <siddontang@gmail.com>
Date: Tue, 22 Sep 2015 14:39:30 +0800
Subject: [PATCH] table: add ColumnOffset to get column offset with name

---
 table/table.go              |  3 +++
 table/tables/tables.go      | 32 ++++++++++++++++++++++++++++++++
 table/tables/tables_test.go | 22 ++++++++++++++++++++++
 3 files changed, 57 insertions(+)

diff --git a/table/table.go b/table/table.go
index 7270c82f90..ad8ccd8e3a 100644
--- a/table/table.go
+++ b/table/table.go
@@ -107,6 +107,9 @@ type Table interface {
 	// LockRow locks a row.
 	// If update is true, set row lock key to current txn.
 	LockRow(ctx context.Context, h int64, update bool) error
+
+	// ColumnOffset gets the column offset in whole columns with column name.
+	ColumnOffset(name string) (int, error)
 }
 
 // TableFromMeta builds a table.Table from *model.TableInfo.
diff --git a/table/tables/tables.go b/table/tables/tables.go
index c714e5a60f..584c78ca98 100644
--- a/table/tables/tables.go
+++ b/table/tables/tables.go
@@ -558,6 +558,38 @@ func (t *Table) AllocAutoID() (int64, error) {
 	return t.alloc.Alloc(t.ID)
 }
 
+// ColumnOffset implements table.Table ColumnOffset interface.
+func (t *Table) ColumnOffset(name string) (int, error) {
+	seps := strings.Split(name, ".")
+	var (
+		table string
+		col   string
+	)
+
+	// Now we only support column or table.column qualified name.
+	// TODO: support db.table.column later.
+	if len(seps) == 1 {
+		col = name
+	} else if len(seps) == 2 {
+		table = seps[0]
+		col = seps[1]
+	} else {
+		return -1, errors.Errorf("invalid column format %s", name)
+	}
+
+	if len(table) > 0 && table != t.Name.O {
+		return -1, errors.Errorf("column table %s is not equal %s", table, t.Name.O)
+	}
+
+	for _, c := range t.Columns {
+		if strings.EqualFold(col, c.Name.L) {
+			return c.Offset, nil
+		}
+	}
+
+	return -1, errors.Errorf("unknown column %s", name)
+}
+
 func init() {
 	table.TableFromMeta = TableFromMeta
 }
diff --git a/table/tables/tables_test.go b/table/tables/tables_test.go
index 9d97daee03..cc837d2339 100644
--- a/table/tables/tables_test.go
+++ b/table/tables/tables_test.go
@@ -102,6 +102,28 @@ func (ts *testSuite) TestBasic(c *C) {
 	c.Assert(cnt, Equals, 0)
 	_, err = ts.se.Execute("drop table test.t")
 	c.Assert(err, IsNil)
+
+	offsetTbl := []struct {
+		Name   string
+		Offset int
+	}{
+		{"a", 0},
+		{"b", 1},
+		{"c", -1},
+		{"t.a", 0},
+		{"t.c", -1},
+		{"t1.c", -1},
+		{"test.test.t.c", -1},
+	}
+
+	for _, t := range offsetTbl {
+		index, err := tb.ColumnOffset(t.Name)
+		if t.Offset == -1 {
+			c.Assert(err, NotNil)
+		} else {
+			c.Assert(index, Equals, t.Offset)
+		}
+	}
 }
 
 func countEntriesWithPrefix(ctx context.Context, prefix string) (int, error) {