From f750f625b712169785f30ff96b6ab5c17ed1cd6d Mon Sep 17 00:00:00 2001
From: Thibault Deutsch <thibault.deutsch@gmail.com>
Date: Thu, 21 Sep 2023 11:23:48 +0100
Subject: [PATCH] hrpc: allow table creation with table attributes (#234)

Closes #161
---
 hrpc/create.go | 28 ++++++++++++++++++++++-----
 table_test.go  | 52 ++++++++++++++++++++++++++++++++++++++++++++++++--
 2 files changed, 73 insertions(+), 7 deletions(-)

diff --git a/hrpc/create.go b/hrpc/create.go
index c69c244e..6754df08 100644
--- a/hrpc/create.go
+++ b/hrpc/create.go
@@ -16,11 +16,12 @@ import (
 type CreateTable struct {
 	base
 
-	families  map[string]map[string]string
-	splitKeys [][]byte
+	attributes map[string]string
+	families   map[string]map[string]string
+	splitKeys  [][]byte
 }
 
-var defaultAttributes = map[string]string{
+var defaultFamiliesAttributes = map[string]string{
 	"BLOOMFILTER":         "ROW",
 	"VERSIONS":            "3",
 	"IN_MEMORY":           "false",
@@ -52,8 +53,8 @@ func NewCreateTable(ctx context.Context, table []byte,
 		option(ct)
 	}
 	for family, attrs := range families {
-		ct.families[family] = make(map[string]string, len(defaultAttributes))
-		for k, dv := range defaultAttributes {
+		ct.families[family] = make(map[string]string, len(defaultFamiliesAttributes))
+		for k, dv := range defaultFamiliesAttributes {
 			if v, ok := attrs[k]; ok {
 				ct.families[family][k] = v
 			} else {
@@ -71,6 +72,13 @@ func SplitKeys(sk [][]byte) func(*CreateTable) {
 	}
 }
 
+// TableAttributes will return an option that will set attributes on the created table
+func TableAttributes(attrs map[string]string) func(*CreateTable) {
+	return func(ct *CreateTable) {
+		ct.attributes = attrs
+	}
+}
+
 // Name returns the name of this RPC call.
 func (ct *CreateTable) Name() string {
 	return "CreateTable"
@@ -83,6 +91,14 @@ func (ct *CreateTable) Description() string {
 
 // ToProto converts the RPC into a protobuf message
 func (ct *CreateTable) ToProto() proto.Message {
+	pbAttributes := make([]*pb.BytesBytesPair, 0, len(ct.attributes))
+	for k, v := range ct.attributes {
+		pbAttributes = append(pbAttributes, &pb.BytesBytesPair{
+			First:  []byte(k),
+			Second: []byte(v),
+		})
+	}
+
 	pbFamilies := make([]*pb.ColumnFamilySchema, 0, len(ct.families))
 	for family, attrs := range ct.families {
 		f := &pb.ColumnFamilySchema{
@@ -97,6 +113,7 @@ func (ct *CreateTable) ToProto() proto.Message {
 		}
 		pbFamilies = append(pbFamilies, f)
 	}
+
 	return &pb.CreateTableRequest{
 		TableSchema: &pb.TableSchema{
 			TableName: &pb.TableName{
@@ -104,6 +121,7 @@ func (ct *CreateTable) ToProto() proto.Message {
 				Namespace: []byte("default"),
 				Qualifier: ct.table,
 			},
+			Attributes:     pbAttributes,
 			ColumnFamilies: pbFamilies,
 		},
 		SplitKeys: ct.splitKeys,
diff --git a/table_test.go b/table_test.go
index 5ee2cae1..c58b7e93 100644
--- a/table_test.go
+++ b/table_test.go
@@ -30,8 +30,10 @@ var infoFamily = map[string][]string{
 }
 
 var cFamilies = map[string]map[string]string{
-	"cf":  nil,
-	"cf2": nil,
+	"cf": nil,
+	"cf2": {
+		"MIN_VERSIONS": "1",
+	},
 }
 
 func TestCreateTable(t *testing.T) {
@@ -114,6 +116,52 @@ func TestCreatePresplitTable(t *testing.T) {
 	}
 }
 
+func TestCreateTableWithAttributes(t *testing.T) {
+	testTableName := t.Name() + "_" + getTimestampString()
+	t.Log("testTableName=" + testTableName)
+
+	attrs := map[string]string{
+		"NORMALIZATION_ENABLED": "TRUE",
+	}
+
+	ac := gohbase.NewAdminClient(*host)
+	crt := hrpc.NewCreateTable(
+		context.Background(),
+		[]byte(testTableName),
+		cFamilies,
+		hrpc.TableAttributes(attrs),
+	)
+
+	if err := ac.CreateTable(crt); err != nil {
+		t.Errorf("CreateTable returned an error: %v", err)
+	}
+
+	// check in hbase:meta if there's a region for the table
+	c := gohbase.NewClient(*host)
+	metaKey := testTableName + ","
+	keyFilter := filter.NewPrefixFilter([]byte(metaKey))
+	scan, err := hrpc.NewScanStr(context.Background(), metaTableName, hrpc.Filters(keyFilter))
+	if err != nil {
+		t.Fatalf("Failed to create Scan request: %s", err)
+	}
+
+	var rsp []*hrpc.Result
+	scanner := c.Scan(scan)
+	for {
+		res, err := scanner.Next()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			t.Fatal(err)
+		}
+		rsp = append(rsp, res)
+	}
+	if len(rsp) != 1 {
+		t.Errorf("Meta returned %d rows for prefix '%s' , want 1", len(rsp), metaKey)
+	}
+}
+
 func TestDisableDeleteTable(t *testing.T) {
 	testTableName := "test1_" + getTimestampString()
 	t.Log("testTableName=" + testTableName)