aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorChris Lu <chris.lu@gmail.com>2020-09-02 18:39:24 -0700
committerChris Lu <chris.lu@gmail.com>2020-09-02 18:39:24 -0700
commit0d99a5da3caa5a35ccc6048e6ec995487379e61e (patch)
tree4ece9108e61db3bf72bf0ff6863c7139a9b9658b
parent645a4af3db6fa97440fcaca632bb1ddd939f8268 (diff)
downloadseaweedfs-0d99a5da3caa5a35ccc6048e6ec995487379e61e.tar.xz
seaweedfs-0d99a5da3caa5a35ccc6048e6ec995487379e61e.zip
c*: support kv operations
-rw-r--r--weed/filer/cassandra/cassandra_store_kv.go51
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
}