You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

79 lines
2.7 KiB

3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
3 years ago
  1. //go:build ydb
  2. // +build ydb
  3. package ydb
  4. import (
  5. "context"
  6. "fmt"
  7. "github.com/ydb-platform/ydb-go-sdk/v3/table/options"
  8. "github.com/chrislusf/seaweedfs/weed/filer"
  9. "github.com/chrislusf/seaweedfs/weed/filer/abstract_sql"
  10. "github.com/chrislusf/seaweedfs/weed/util"
  11. "github.com/ydb-platform/ydb-go-sdk/v3/table"
  12. "github.com/ydb-platform/ydb-go-sdk/v3/table/result/named"
  13. "github.com/ydb-platform/ydb-go-sdk/v3/table/types"
  14. )
  15. func (store *YdbStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
  16. dirStr, dirHash, name := abstract_sql.GenDirAndName(key)
  17. fileMeta := FileMeta{dirHash, name, dirStr, value}
  18. return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) {
  19. _, _, err = s.Execute(ctx, rwTX, *withPragma(&store.tablePathPrefix, insertQuery),
  20. fileMeta.queryParameters(0),
  21. options.WithQueryCachePolicy(options.WithQueryCachePolicyKeepInCache()))
  22. if err != nil {
  23. return fmt.Errorf("kv put execute %s: %v", util.NewFullPath(dirStr, name).Name(), err)
  24. }
  25. return nil
  26. })
  27. }
  28. func (store *YdbStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
  29. dirStr, dirHash, name := abstract_sql.GenDirAndName(key)
  30. valueFound := false
  31. err = store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error {
  32. _, res, err := s.Execute(ctx, roTX, *withPragma(&store.tablePathPrefix, findQuery),
  33. table.NewQueryParameters(
  34. table.ValueParam("$dir_hash", types.Int64Value(dirHash)),
  35. table.ValueParam("$name", types.UTF8Value(name))),
  36. options.WithQueryCachePolicy(options.WithQueryCachePolicyKeepInCache()))
  37. if err != nil {
  38. return fmt.Errorf("kv get execute %s: %v", util.NewFullPath(dirStr, name).Name(), err)
  39. }
  40. defer func() { _ = res.Close() }()
  41. for res.NextResultSet(ctx) {
  42. for res.NextRow() {
  43. if err := res.ScanNamed(named.OptionalWithDefault("meta", &value)); err != nil {
  44. return fmt.Errorf("scanNamed %s : %v", util.NewFullPath(dirStr, name).Name(), err)
  45. }
  46. valueFound = true
  47. return nil
  48. }
  49. }
  50. return res.Err()
  51. })
  52. if !valueFound {
  53. return nil, filer.ErrKvNotFound
  54. }
  55. return value, nil
  56. }
  57. func (store *YdbStore) KvDelete(ctx context.Context, key []byte) (err error) {
  58. dirStr, dirHash, name := abstract_sql.GenDirAndName(key)
  59. return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) {
  60. _, _, err = s.Execute(ctx, rwTX, *withPragma(&store.tablePathPrefix, insertQuery),
  61. table.NewQueryParameters(
  62. table.ValueParam("$dir_hash", types.Int64Value(dirHash)),
  63. table.ValueParam("$name", types.UTF8Value(name))),
  64. options.WithQueryCachePolicy(options.WithQueryCachePolicyKeepInCache()))
  65. if err != nil {
  66. return fmt.Errorf("kv delete %s: %v", util.NewFullPath(dirStr, name).Name(), err)
  67. }
  68. return nil
  69. })
  70. }