From 06a118826746c0a8b69bbf4069f2c9b9b4afa37d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 1 Sep 2020 22:48:36 -0700 Subject: [PATCH] mysql/postgres: support kv operations --- .../abstract_sql/abstract_sql_store_kv.go | 78 ++++++++++++++++++- 1 file changed, 75 insertions(+), 3 deletions(-) diff --git a/weed/filer/abstract_sql/abstract_sql_store_kv.go b/weed/filer/abstract_sql/abstract_sql_store_kv.go index cfffc8918..de7d7b7a5 100644 --- a/weed/filer/abstract_sql/abstract_sql_store_kv.go +++ b/weed/filer/abstract_sql/abstract_sql_store_kv.go @@ -2,17 +2,89 @@ package abstract_sql import ( "context" + "database/sql" + "fmt" "github.com/chrislusf/seaweedfs/weed/filer" + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/util" ) func (store *AbstractSqlStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { - return filer.ErrKvNotImplemented + + dirStr, dirHash, name := genDirAndName(key) + + res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlInsert, dirHash, name, dirStr, value) + if err != nil { + return fmt.Errorf("kv insert: %s", err) + } + + // TODO maybe it will throw error before coming here? + + affectedRows, err := res.RowsAffected() + if err == nil && affectedRows > 0 { + return nil + } + + // now the insert failed possibly due to duplication constraints + glog.V(1).Infof("kv insert falls back to update: %s", err) + + res, err = store.getTxOrDB(ctx).ExecContext(ctx, store.SqlUpdate, value, dirHash, name, dirStr) + if err != nil { + return fmt.Errorf("kv upsert: %s", err) + } + + _, err = res.RowsAffected() + if err != nil { + return fmt.Errorf("kv upsert no rows affected: %s", err) + } + return nil + } func (store *AbstractSqlStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { - return nil, filer.ErrKvNotImplemented + + dirStr, dirHash, name := genDirAndName(key) + row := store.getTxOrDB(ctx).QueryRowContext(ctx, store.SqlFind, dirHash, name, dirStr) + + err = row.Scan(&value) + + if err == sql.ErrNoRows { + return nil, filer.ErrKvNotFound + } + + if err != nil { + return nil, fmt.Errorf("kv get: %v", err) + } + + return } func (store *AbstractSqlStore) KvDelete(ctx context.Context, key []byte) (err error) { - return filer.ErrKvNotImplemented + + dirStr, dirHash, name := genDirAndName(key) + + res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlDelete, dirHash, name, dirStr) + if err != nil { + return fmt.Errorf("kv delete: %s", err) + } + + _, err = res.RowsAffected() + if err != nil { + return fmt.Errorf("kv delete %s but no rows affected: %s", err) + } + + return nil + } + +func genDirAndName(key []byte) (dirStr string, dirHash int64, name string) { + for len(key) < 8 { + key = append(key, 0) + } + + dirHash = int64(util.BytesToUint64(key[:8])) + dirStr = string(key[:8]) + name = string(key[8:]) + + return +} \ No newline at end of file