From bfe8680b131f1e94a1bc91426504578b02f266ea Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Fri, 4 Oct 2019 09:21:16 -0500
Subject: [PATCH] handle byte slice batch record IDs

also check for unsupported field types and error
---
 gpexp/importbatch.go | 24 ++++++++++++++++++++----
 1 file changed, 20 insertions(+), 4 deletions(-)

diff --git a/gpexp/importbatch.go b/gpexp/importbatch.go
index be74b59..adff9ee 100644
--- a/gpexp/importbatch.go
+++ b/gpexp/importbatch.go
@@ -130,6 +130,8 @@ func NewBatch(client *pilosa.Client, size int, index *pilosa.Index, fields []*pi
 			hasTime = typ == pilosa.FieldTypeTime || hasTime
 		case pilosa.FieldTypeInt:
 			values[field.Name()] = make([]int64, 0, size)
+		default:
+			return nil, errors.Errorf("field type %s is not currently supported through Batch", typ)
 		}
 	}
 	b := &Batch{
@@ -254,10 +256,7 @@ func (b *Batch) Add(rec Row) error {
 		return errors.Errorf("record needs to match up with batch fields, got %d fields and %d record", len(b.header), len(rec.Values))
 	}
 
-	switch rid := rec.ID.(type) {
-	case uint64:
-		b.ids = append(b.ids, rid)
-	case string:
+	handleStringID := func(rid string) error {
 		if colID, ok, err := b.transCache.GetCol(b.index.Name(), rid); err != nil {
 			return errors.Wrap(err, "translating column")
 		} else if ok {
@@ -271,6 +270,23 @@ func (b *Batch) Add(rec Row) error {
 			b.toTranslateID[rid] = ints
 			b.ids = append(b.ids, 0)
 		}
+		return nil
+	}
+	var err error
+
+	switch rid := rec.ID.(type) {
+	case uint64:
+		b.ids = append(b.ids, rid)
+	case string:
+		err := handleStringID(rid)
+		if err != nil {
+			return err
+		}
+	case []byte:
+		err = handleStringID(string(rid))
+		if err != nil {
+			return err
+		}
 	default: // TODO support nil ID as being auto-allocated.
 		return errors.Errorf("unsupported id type %T value %v", rid, rid)
 	}