diff options
| author | Chris Lu <chris.lu@gmail.com> | 2020-09-02 18:39:24 -0700 |
|---|---|---|
| committer | Chris Lu <chris.lu@gmail.com> | 2020-09-02 18:39:24 -0700 |
| commit | 0d99a5da3caa5a35ccc6048e6ec995487379e61e (patch) | |
| tree | 4ece9108e61db3bf72bf0ff6863c7139a9b9658b | |
| parent | 645a4af3db6fa97440fcaca632bb1ddd939f8268 (diff) | |
| download | seaweedfs-0d99a5da3caa5a35ccc6048e6ec995487379e61e.tar.xz seaweedfs-0d99a5da3caa5a35ccc6048e6ec995487379e61e.zip | |
c*: support kv operations
| -rw-r--r-- | weed/filer/cassandra/cassandra_store_kv.go | 51 |
1 files changed, 47 insertions, 4 deletions
diff --git a/weed/filer/cassandra/cassandra_store_kv.go b/weed/filer/cassandra/cassandra_store_kv.go index f7668746f..b6238cf0e 100644 --- a/weed/filer/cassandra/cassandra_store_kv.go +++ b/weed/filer/cassandra/cassandra_store_kv.go @@ -2,17 +2,60 @@ package cassandra import ( "context" + "fmt" "github.com/chrislusf/seaweedfs/weed/filer" + "github.com/gocql/gocql" ) func (store *CassandraStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { - return filer.ErrKvNotImplemented + dir, name := genDirAndName(key) + + if err := store.session.Query( + "INSERT INTO filemeta (directory,name,meta) VALUES(?,?,?) USING TTL ? ", + dir, name, value, 0).Exec(); err != nil { + return fmt.Errorf("kv insert: %s", err) + } + + return nil } -func (store *CassandraStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { - return nil, filer.ErrKvNotImplemented +func (store *CassandraStore) KvGet(ctx context.Context, key []byte) (data []byte, err error) { + dir, name := genDirAndName(key) + + if err := store.session.Query( + "SELECT meta FROM filemeta WHERE directory=? AND name=?", + dir, name).Consistency(gocql.One).Scan(&data); err != nil { + if err != gocql.ErrNotFound { + return nil, filer.ErrKvNotFound + } + } + + if len(data) == 0 { + return nil, filer.ErrKvNotFound + } + + return data, nil } func (store *CassandraStore) KvDelete(ctx context.Context, key []byte) (err error) { - return filer.ErrKvNotImplemented + dir, name := genDirAndName(key) + + if err := store.session.Query( + "DELETE FROM filemeta WHERE directory=? AND name=?", + dir, name).Exec(); err != nil { + return fmt.Errorf("kv delete: %v", err) + } + + return nil +} + +func genDirAndName(key []byte) (dir string, name string) { + for len(key) < 8 { + key = append(key, 0) + } + + dir = string(key[:8]) + name = string(key[8:]) + + return } |
