aboutsummaryrefslogtreecommitdiff
path: root/weed
diff options
context:
space:
mode:
Diffstat (limited to 'weed')
-rw-r--r--weed/command/imports.go1
-rw-r--r--weed/command/scaffold/filer.toml11
-rw-r--r--weed/filer/ydb/readme.md16
-rw-r--r--weed/filer/ydb/ydb_store.go263
-rw-r--r--weed/filer/ydb/ydb_store_kv.go100
-rw-r--r--weed/filer/ydb/ydb_types.go13
-rw-r--r--weed/filer/ydb/ydb_types_ydbgen.go194
-rw-r--r--weed/server/filer_server.go1
8 files changed, 225 insertions, 374 deletions
diff --git a/weed/command/imports.go b/weed/command/imports.go
index 5b3195907..04079b162 100644
--- a/weed/command/imports.go
+++ b/weed/command/imports.go
@@ -32,4 +32,5 @@ import (
_ "github.com/chrislusf/seaweedfs/weed/filer/redis2"
_ "github.com/chrislusf/seaweedfs/weed/filer/redis3"
_ "github.com/chrislusf/seaweedfs/weed/filer/sqlite"
+ _ "github.com/chrislusf/seaweedfs/weed/filer/ydb"
)
diff --git a/weed/command/scaffold/filer.toml b/weed/command/scaffold/filer.toml
index 0a505bbdc..3a920d80b 100644
--- a/weed/command/scaffold/filer.toml
+++ b/weed/command/scaffold/filer.toml
@@ -295,6 +295,17 @@ password=""
# skip tls cert validation
insecure_skip_verify = true
+[ydb]
+enabled = false
+useBucketPrefix=true
+dsn="grpc://localhost:2136?database=/local"
+prefix="en"
+poolSizeLimit=50
+# Authenticate produced with one of next environment variables:
+# YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS=<path/to/sa_key_file> — used service account key file by path
+# YDB_ANONYMOUS_CREDENTIALS="1" — used for authenticate with anonymous access. Anonymous access needs for connect to testing YDB installation
+# YDB_METADATA_CREDENTIALS="1" — used metadata service for authenticate to YDB from yandex cloud virtual machine or from yandex function
+# YDB_ACCESS_TOKEN_CREDENTIALS=<access_token> — used for authenticate to YDB with short-life access token. For example, access token may be IAM token
##########################
##########################
diff --git a/weed/filer/ydb/readme.md b/weed/filer/ydb/readme.md
index 2221e13b5..fd6c5c0a2 100644
--- a/weed/filer/ydb/readme.md
+++ b/weed/filer/ydb/readme.md
@@ -1,6 +1,7 @@
## YDB
database: https://github.com/ydb-platform/ydb
+
go driver: https://github.com/ydb-platform/ydb-go-sdk
options:
@@ -8,15 +9,18 @@ options:
```
[ydb]
enabled=true
+dsn=grpcs://ydb-ru.yandex.net:2135/?database=/ru/home/username/db
prefix="seaweedfs"
useBucketPrefix=true
-coonectionUrl=grpcs://ydb-ru.yandex.net:2135/?database=/ru/home/username/db
+poolSizeLimit=50
```
-get ydb types
-```
-ydbgen -dir weed/filer/ydb
-```
+Authenticate produced with one of next environment variables:
+ * `YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS=<path/to/sa_key_file>` — used service account key file by path
+ * `YDB_ANONYMOUS_CREDENTIALS="1"` — used for authenticate with anonymous access. Anonymous access needs for connect to testing YDB installation
+ * `YDB_METADATA_CREDENTIALS="1"` — used metadata service for authenticate to YDB from yandex cloud virtual machine or from yandex function
+ * `YDB_ACCESS_TOKEN_CREDENTIALS=<access_token>` — used for authenticate to YDB with short-life access token. For example, access token may be IAM token
+ * `YDB_CONNECTION_STRING="grpcs://endpoint/?database=database"`
-i test using this dev database:
+ * i test using this dev database:
`make dev_ydb`
diff --git a/weed/filer/ydb/ydb_store.go b/weed/filer/ydb/ydb_store.go
index d82fc4da7..4f3477471 100644
--- a/weed/filer/ydb/ydb_store.go
+++ b/weed/filer/ydb/ydb_store.go
@@ -7,9 +7,12 @@ import (
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/util"
- "github.com/yandex-cloud/ydb-go-sdk/v2"
- "github.com/yandex-cloud/ydb-go-sdk/v2/connect"
- "github.com/yandex-cloud/ydb-go-sdk/v2/table"
+ environ "github.com/ydb-platform/ydb-go-sdk-auth-environ"
+ ydb "github.com/ydb-platform/ydb-go-sdk/v3"
+ "github.com/ydb-platform/ydb-go-sdk/v3/sugar"
+ "github.com/ydb-platform/ydb-go-sdk/v3/table"
+ "github.com/ydb-platform/ydb-go-sdk/v3/table/result/named"
+ "github.com/ydb-platform/ydb-go-sdk/v3/table/types"
"path"
"strings"
"time"
@@ -24,15 +27,12 @@ var (
table.BeginTx(table.WithOnlineReadOnly()),
table.CommitTx(),
)
- rwTX = table.TxControl(
- table.BeginTx(table.WithSerializableReadWrite()),
- table.CommitTx(),
- )
+ rwTX = table.DefaultTxControl()
)
type YdbStore struct {
SupportBucketTable bool
- DB *connect.Connection
+ DB ydb.Connection
dirBuckets string
tablePathPrefix string
}
@@ -48,34 +48,43 @@ func (store *YdbStore) GetName() string {
func (store *YdbStore) Initialize(configuration util.Configuration, prefix string) (err error) {
return store.initialize(
configuration.GetString("filer.options.buckets_folder"),
- configuration.GetString(prefix+"coonectionUrl"),
+ configuration.GetString(prefix+"dsn"),
configuration.GetString(prefix+"tablePathPrefix"),
configuration.GetBool(prefix+"useBucketPrefix"),
configuration.GetInt(prefix+"connectionTimeOut"),
+ configuration.GetInt(prefix+"poolSizeLimit"),
)
}
-func (store *YdbStore) initialize(dirBuckets string, sqlUrl string, tablePathPrefix string, useBucketPrefix bool, connectionTimeOut int) (err error) {
+func (store *YdbStore) initialize(dirBuckets string, dsn string, tablePathPrefix string, useBucketPrefix bool, connectionTimeOut int, poolSizeLimit int) (err error) {
store.dirBuckets = dirBuckets
store.tablePathPrefix = tablePathPrefix
store.SupportBucketTable = useBucketPrefix
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
if connectionTimeOut == 0 {
connectionTimeOut = defaultConnectionTimeOut
}
- var cancel context.CancelFunc
- connCtx, cancel := context.WithTimeout(context.Background(), time.Duration(connectionTimeOut)*time.Second)
- defer cancel()
- connParams := connect.MustConnectionString(sqlUrl)
- store.DB, err = connect.New(connCtx, connParams)
+ opts := []ydb.Option{
+ environ.WithEnvironCredentials(ctx),
+ ydb.WithDialTimeout(time.Duration(connectionTimeOut) * time.Second),
+ }
+ if poolSizeLimit > 0 {
+ opts = append(opts, ydb.WithSessionPoolSizeLimit(poolSizeLimit))
+ }
+ store.DB, err = ydb.Open(ctx, dsn, opts...)
if err != nil {
- store.DB.Close()
+ _ = store.DB.Close(ctx)
store.DB = nil
- return fmt.Errorf("can not connect to %s error:%v", sqlUrl, err)
+ return fmt.Errorf("can not connect to %s error:%v", dsn, err)
}
- defer store.DB.Close()
-
- if err = store.DB.EnsurePathExists(connCtx, connParams.Database()); err != nil {
- return fmt.Errorf("connect to %s error:%v", sqlUrl, err)
+ defer func() { _ = store.DB.Close(ctx) }()
+ store.tablePathPrefix = path.Join(store.DB.Name(), tablePathPrefix)
+ if err = sugar.RemoveRecursive(ctx, store.DB, store.tablePathPrefix); err != nil {
+ return fmt.Errorf("RemoveRecursive %s : %v", store.tablePathPrefix, err)
+ }
+ if err = sugar.MakeRecursive(ctx, store.DB, store.tablePathPrefix); err != nil {
+ return fmt.Errorf("MakeRecursive %s : %v", store.tablePathPrefix, err)
}
return nil
}
@@ -92,16 +101,14 @@ func (store *YdbStore) insertOrUpdateEntry(ctx context.Context, entry *filer.Ent
}
fileMeta := FileMeta{util.HashStringToLong(dir), name, dir, meta}
- return table.Retry(ctx, store.DB.Table().Pool(),
- table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) {
- stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), query))
- if err != nil {
- return err
- }
- _, _, err = stmt.Execute(ctx, rwTX, fileMeta.QueryParameters())
- return err
- }),
- )
+ return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) {
+ stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), query))
+ if err != nil {
+ return fmt.Errorf("Prepare %s : %v", dir, err)
+ }
+ _, _, err = stmt.Execute(ctx, rwTX, fileMeta.QueryParameters())
+ return err
+ })
}
func (store *YdbStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
@@ -114,68 +121,71 @@ func (store *YdbStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err
func (store *YdbStore) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) {
dir, name := fullpath.DirAndName()
- var res *table.Result
- err = table.Retry(ctx, store.DB.Table().Pool(),
- table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) {
- stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), findQuery))
- if err != nil {
- return err
+ var data []byte
+ entryFound := false
+ err = store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error {
+ stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), findQuery))
+ if err != nil {
+ return fmt.Errorf("Prepare %s : %v", entry.FullPath, err)
+ }
+ _, res, err := stmt.Execute(ctx, roTX, table.NewQueryParameters(
+ table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(dir))),
+ table.ValueParam("$name", types.UTF8Value(name))))
+ if err != nil {
+ return fmt.Errorf("Execute %s : %v", entry.FullPath, err)
+ }
+ defer func() {
+ _ = res.Close()
+ }()
+ for res.NextRow() {
+ if err := res.ScanNamed(named.Required("meta", &data)); err != nil {
+ return fmt.Errorf("scanNamed %s : %v", entry.FullPath, err)
}
- _, res, err = stmt.Execute(ctx, roTX, table.NewQueryParameters(
- table.ValueParam("$dir_hash", ydb.Int64Value(util.HashStringToLong(dir))),
- table.ValueParam("$name", ydb.UTF8Value(name))))
- return err
- }),
- )
+ entryFound = true
+ return nil
+ }
+ return res.Err()
+ })
if err != nil {
return nil, err
}
- defer res.Close()
-
- for res.NextResultSet(ctx) {
- for res.NextRow() {
- res.SeekItem("meta")
- entry.FullPath = fullpath
- if err := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(res.String())); err != nil {
- return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
- }
- return entry, nil
- }
+ if !entryFound {
+ return nil, filer_pb.ErrNotFound
+ }
+ entry.FullPath = fullpath
+ if err := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); err != nil {
+ return nil, fmt.Errorf("decode %s : %v", entry.FullPath, err)
}
- return nil, filer_pb.ErrNotFound
+ return entry, nil
}
func (store *YdbStore) DeleteEntry(ctx context.Context, fullpath util.FullPath) (err error) {
dir, name := fullpath.DirAndName()
- return table.Retry(ctx, store.DB.Table().Pool(),
- table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) {
- stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), deleteQuery))
- if err != nil {
- return err
- }
- _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters(
- table.ValueParam("$dir_hash", ydb.Int64Value(util.HashStringToLong(dir))),
- table.ValueParam("$name", ydb.UTF8Value(name))))
- return err
- }),
- )
+ return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) {
+ stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), deleteQuery))
+ if err != nil {
+ return fmt.Errorf("Prepare %s : %v", dir, err)
+ }
+ _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters(
+ table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(dir))),
+ table.ValueParam("$name", types.UTF8Value(name))))
+ return err
+ })
}
func (store *YdbStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) {
dir, _ := fullpath.DirAndName()
- return table.Retry(ctx, store.DB.Table().Pool(),
- table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) {
- stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), deleteFolderChildrenQuery))
- if err != nil {
- return err
- }
- _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters(
- table.ValueParam("$dir_hash", ydb.Int64Value(util.HashStringToLong(dir))),
- table.ValueParam("$directory", ydb.UTF8Value(dir))))
- return err
- }),
- )
+ return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) {
+ stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), deleteFolderChildrenQuery))
+ if err != nil {
+ return fmt.Errorf("Prepare %s : %v", dir, err)
+ }
+ _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters(
+ table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(dir))),
+ table.ValueParam("$directory", types.UTF8Value(dir))))
+ return err
+ })
}
func (store *YdbStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
@@ -184,62 +194,60 @@ func (store *YdbStore) ListDirectoryEntries(ctx context.Context, dirPath util.Fu
func (store *YdbStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
dir := string(dirPath)
- var res *table.Result
startFileCompOp := ">"
if includeStartFile {
startFileCompOp = ">="
}
- err = table.Retry(ctx, store.DB.Table().Pool(),
- table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) {
- stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), fmt.Sprintf(ListDirectoryQuery, startFileCompOp)))
- if err != nil {
- return err
+ err = store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error {
+ stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), fmt.Sprintf(ListDirectoryQuery, startFileCompOp)))
+ if err != nil {
+ return fmt.Errorf("Prepare %s : %v", dir, err)
+ }
+ _, res, err := stmt.Execute(ctx, roTX, table.NewQueryParameters(
+ table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(dir))),
+ table.ValueParam("$directory", types.UTF8Value(dir)),
+ table.ValueParam("$start_name", types.UTF8Value(startFileName)),
+ table.ValueParam("$prefix", types.UTF8Value(prefix)),
+ table.ValueParam("$limit", types.Int64Value(limit)),
+ ))
+ if err != nil {
+ return fmt.Errorf("Execute %s : %v", dir, err)
+ }
+ defer func() {
+ _ = res.Close()
+ }()
+ for res.NextResultSet(ctx) {
+ for res.NextRow() {
+ var name string
+ var data []byte
+ if err := res.ScanNamed(
+ named.Required("name", &name),
+ named.Required("meta", &data)); err != nil {
+ return fmt.Errorf("scanNamed %s : %v", dir, err)
+ }
+ lastFileName = name
+ entry := &filer.Entry{
+ FullPath: util.NewFullPath(dir, name),
+ }
+ if err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); err != nil {
+ glog.V(0).Infof("scan decode %s : %v", entry.FullPath, err)
+ return fmt.Errorf("scan decode %s : %v", entry.FullPath, err)
+ }
+ if !eachEntryFunc(entry) {
+ break
+ }
}
- _, res, err = stmt.Execute(ctx, roTX, table.NewQueryParameters(
- table.ValueParam("$dir_hash", ydb.Int64Value(util.HashStringToLong(dir))),
- table.ValueParam("$directory", ydb.UTF8Value(dir)),
- table.ValueParam("$start_name", ydb.UTF8Value(startFileName)),
- table.ValueParam("$prefix", ydb.UTF8Value(prefix)),
- table.ValueParam("$limit", ydb.Int64Value(limit)),
- ))
- return err
- }),
- )
+ }
+ return res.Err()
+ })
if err != nil {
return lastFileName, err
}
- defer res.Close()
-
- for res.NextSet() {
- for res.NextRow() {
- res.SeekItem("name")
- name := res.UTF8()
- res.SeekItem("meta")
- data := res.String()
- if res.Err() != nil {
- glog.V(0).Infof("scan %s : %v", dirPath, err)
- return lastFileName, fmt.Errorf("scan %s: %v", dirPath, err)
- }
- lastFileName = name
-
- entry := &filer.Entry{
- FullPath: util.NewFullPath(dir, name),
- }
- if err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); err != nil {
- glog.V(0).Infof("scan decode %s : %v", entry.FullPath, err)
- return lastFileName, fmt.Errorf("scan decode %s : %v", entry.FullPath, err)
- }
-
- if !eachEntryFunc(entry) {
- break
- }
- }
- }
return lastFileName, nil
}
func (store *YdbStore) BeginTransaction(ctx context.Context) (context.Context, error) {
- session, err := store.DB.Table().Pool().Create(ctx)
+ session, err := store.DB.Table().CreateSession(ctx)
if err != nil {
return ctx, err
}
@@ -251,21 +259,22 @@ func (store *YdbStore) BeginTransaction(ctx context.Context) (context.Context, e
}
func (store *YdbStore) CommitTransaction(ctx context.Context) error {
- if tx, ok := ctx.Value("tx").(*table.Transaction); ok {
- return tx.Commit(ctx)
+ if tx, ok := ctx.Value("tx").(table.Transaction); ok {
+ _, err := tx.CommitTx(ctx)
+ return err
}
return nil
}
func (store *YdbStore) RollbackTransaction(ctx context.Context) error {
- if tx, ok := ctx.Value("tx").(*table.Transaction); ok {
+ if tx, ok := ctx.Value("tx").(table.Transaction); ok {
return tx.Rollback(ctx)
}
return nil
}
func (store *YdbStore) Shutdown() {
- store.DB.Close()
+ _ = store.DB.Close(context.Background())
}
func (store *YdbStore) getPrefix(dir string) string {
diff --git a/weed/filer/ydb/ydb_store_kv.go b/weed/filer/ydb/ydb_store_kv.go
index 3473d756d..b2a7af74f 100644
--- a/weed/filer/ydb/ydb_store_kv.go
+++ b/weed/filer/ydb/ydb_store_kv.go
@@ -5,68 +5,74 @@ import (
"fmt"
"github.com/chrislusf/seaweedfs/weed/filer"
"github.com/chrislusf/seaweedfs/weed/filer/abstract_sql"
- "github.com/yandex-cloud/ydb-go-sdk/v2"
- "github.com/yandex-cloud/ydb-go-sdk/v2/table"
+ "github.com/chrislusf/seaweedfs/weed/util"
+ "github.com/ydb-platform/ydb-go-sdk/v3/table"
+ "github.com/ydb-platform/ydb-go-sdk/v3/table/result/named"
+ "github.com/ydb-platform/ydb-go-sdk/v3/table/types"
)
func (store *YdbStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) {
dirStr, dirHash, name := abstract_sql.GenDirAndName(key)
fileMeta := FileMeta{dirHash, name, dirStr, value}
- return table.Retry(ctx, store.DB.Table().Pool(),
- table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) {
- stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), insertQuery))
- if err != nil {
- return fmt.Errorf("kv put: %v", err)
- }
- _, _, err = stmt.Execute(ctx, rwTX, fileMeta.QueryParameters())
- return fmt.Errorf("kv put: %v", err)
- }),
- )
+ return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) {
+ stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), insertQuery))
+ if err != nil {
+ return fmt.Errorf("Prepare %s: %v", util.NewFullPath(dirStr, name), err)
+ }
+ _, _, err = stmt.Execute(ctx, rwTX, fileMeta.QueryParameters())
+ if err != nil {
+ return fmt.Errorf("kv put %s: %v", util.NewFullPath(dirStr, name), err)
+ }
+ return nil
+ })
}
func (store *YdbStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
dirStr, dirHash, name := abstract_sql.GenDirAndName(key)
- var res *table.Result
- err = table.Retry(ctx, store.DB.Table().Pool(),
- table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) {
- stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), findQuery))
- if err != nil {
- return err
- }
- _, res, err = stmt.Execute(ctx, roTX, table.NewQueryParameters(
- table.ValueParam("$dir_hash", ydb.Int64Value(dirHash)),
- table.ValueParam("$name", ydb.UTF8Value(name))))
- return err
- }),
- )
- if err != nil {
- return nil, fmt.Errorf("kv get: %v", err)
- }
- defer res.Close()
-
- for res.NextResultSet(ctx) {
+ valueFound := false
+ err = store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error {
+ stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), findQuery))
+ if err != nil {
+ return fmt.Errorf("Prepare %s: %v", util.NewFullPath(dirStr, name), err)
+ }
+ _, res, err := stmt.Execute(ctx, roTX, table.NewQueryParameters(
+ table.ValueParam("$dir_hash", types.Int64Value(dirHash)),
+ table.ValueParam("$name", types.UTF8Value(name))))
+ if err != nil {
+ return fmt.Errorf("kv get %s: %v", util.NewFullPath(dirStr, name), err)
+ }
+ defer func() { _ = res.Close() }()
for res.NextRow() {
- if err = res.Scan(&value); err != nil {
- return nil, fmt.Errorf("kv get: %v", err)
+ if err := res.ScanNamed(named.Required("meta", &value)); err != nil {
+ return fmt.Errorf("scanNamed %s : %v", util.NewFullPath(dirStr, name), err)
}
- return
+ valueFound = true
+ return nil
}
+ return res.Err()
+ })
+
+ if !valueFound {
+ return nil, filer.ErrKvNotFound
}
- return nil, filer.ErrKvNotFound
+
+ return value, nil
}
func (store *YdbStore) KvDelete(ctx context.Context, key []byte) (err error) {
dirStr, dirHash, name := abstract_sql.GenDirAndName(key)
- return table.Retry(ctx, store.DB.Table().Pool(),
- table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) {
- stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), deleteQuery))
- if err != nil {
- return fmt.Errorf("kv delete: %s", err)
- }
- _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters(
- table.ValueParam("$dir_hash", ydb.Int64Value(dirHash)),
- table.ValueParam("$name", ydb.UTF8Value(name))))
- return fmt.Errorf("kv delete: %s", err)
- }),
- )
+ return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) {
+ stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), insertQuery))
+ if err != nil {
+ return fmt.Errorf("Prepare %s: %v", util.NewFullPath(dirStr, name), err)
+ }
+ _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters(
+ table.ValueParam("$dir_hash", types.Int64Value(dirHash)),
+ table.ValueParam("$name", types.UTF8Value(name))))
+ if err != nil {
+ return fmt.Errorf("kv delete %s: %v", util.NewFullPath(dirStr, name), err)
+ }
+ return nil
+ })
+
}
diff --git a/weed/filer/ydb/ydb_types.go b/weed/filer/ydb/ydb_types.go
index 5bc0d2afd..fb1df3eee 100644
--- a/weed/filer/ydb/ydb_types.go
+++ b/weed/filer/ydb/ydb_types.go
@@ -1,5 +1,10 @@
package ydb
+import (
+ "github.com/ydb-platform/ydb-go-sdk/v3/table"
+ "github.com/ydb-platform/ydb-go-sdk/v3/table/types"
+)
+
//go:generate ydbgen
//ydb:gen
@@ -12,3 +17,11 @@ type FileMeta struct {
//ydb:gen scan,value
type FileMetas []FileMeta
+
+func (fm *FileMeta) QueryParameters() *table.QueryParameters {
+ return table.NewQueryParameters(
+ table.ValueParam("$dir_hash", types.Int64Value(fm.DirHash)),
+ table.ValueParam("$name", types.UTF8Value(fm.Name)),
+ table.ValueParam("$directory", types.UTF8Value(fm.Directory)),
+ table.ValueParam("$meta", types.StringValue(fm.Meta)))
+}
diff --git a/weed/filer/ydb/ydb_types_ydbgen.go b/weed/filer/ydb/ydb_types_ydbgen.go
deleted file mode 100644
index 8bd1906f9..000000000
--- a/weed/filer/ydb/ydb_types_ydbgen.go
+++ /dev/null
@@ -1,194 +0,0 @@
-// Code generated by ydbgen; DO NOT EDIT.
-
-package ydb
-
-import (
- "strconv"
-
- "github.com/yandex-cloud/ydb-go-sdk/v2"
- "github.com/yandex-cloud/ydb-go-sdk/v2/table"
-)
-
-var (
- _ = strconv.Itoa
- _ = ydb.StringValue
- _ = table.NewQueryParameters
-)
-
-func (f *FileMeta) Scan(res *table.Result) (err error) {
- res.SeekItem("dir_hash")
- f.DirHash = res.Int64()
-
- res.SeekItem("name")
- f.Name = res.UTF8()
-
- res.SeekItem("directory")
- f.Directory = res.UTF8()
-
- return res.Err()
-}
-
-func (f *FileMeta) QueryParameters() *table.QueryParameters {
- var v0 ydb.Value
- {
- vp0 := ydb.Int64Value(f.DirHash)
- v0 = vp0
- }
- var v1 ydb.Value
- {
- vp0 := ydb.UTF8Value(f.Name)
- v1 = vp0
- }
- var v2 ydb.Value
- {
- vp0 := ydb.UTF8Value(f.Directory)
- v2 = vp0
- }
- return table.NewQueryParameters(
- table.ValueParam("$dir_hash", v0),
- table.ValueParam("$name", v1),
- table.ValueParam("$directory", v2),
- )
-}
-
-func (f *FileMeta) StructValue() ydb.Value {
- var v0 ydb.Value
- {
- var v1 ydb.Value
- {
- vp0 := ydb.Int64Value(f.DirHash)
- v1 = vp0
- }
- var v2 ydb.Value
- {
- vp0 := ydb.UTF8Value(f.Name)
- v2 = vp0
- }
- var v3 ydb.Value
- {
- vp0 := ydb.UTF8Value(f.Directory)
- v3 = vp0
- }
- v0 = ydb.StructValue(
- ydb.StructFieldValue("dir_hash", v1),
- ydb.StructFieldValue("name", v2),
- ydb.StructFieldValue("directory", v3),
- )
- }
- return v0
-}
-
-func (f *FileMeta) StructType() ydb.Type {
- var t0 ydb.Type
- {
- fs0 := make([]ydb.StructOption, 3)
- var t1 ydb.Type
- {
- tp0 := ydb.TypeInt64
- t1 = tp0
- }
- fs0[0] = ydb.StructField("dir_hash", t1)
- var t2 ydb.Type
- {
- tp0 := ydb.TypeUTF8
- t2 = tp0
- }
- fs0[1] = ydb.StructField("name", t2)
- var t3 ydb.Type
- {
- tp0 := ydb.TypeUTF8
- t3 = tp0
- }
- fs0[2] = ydb.StructField("directory", t3)
- t0 = ydb.Struct(fs0...)
- }
- return t0
-}
-
-func (fs *FileMetas) Scan(res *table.Result) (err error) {
- for res.NextRow() {
- var x0 FileMeta
- res.SeekItem("dir_hash")
- x0.DirHash = res.Int64()
-
- res.SeekItem("name")
- x0.Name = res.UTF8()
-
- res.SeekItem("directory")
- x0.Directory = res.UTF8()
-
- if res.Err() == nil {
- *fs = append(*fs, x0)
- }
- }
- return res.Err()
-}
-
-func (fs FileMetas) ListValue() ydb.Value {
- var list0 ydb.Value
- vs0 := make([]ydb.Value, len(fs))
- for i0, item0 := range fs {
- var v0 ydb.Value
- {
- var v1 ydb.Value
- {
- var v2 ydb.Value
- {
- vp0 := ydb.Int64Value(item0.DirHash)
- v2 = vp0
- }
- var v3 ydb.Value
- {
- vp0 := ydb.UTF8Value(item0.Name)
- v3 = vp0
- }
- var v4 ydb.Value
- {
- vp0 := ydb.UTF8Value(item0.Directory)
- v4 = vp0
- }
- v1 = ydb.StructValue(
- ydb.StructFieldValue("dir_hash", v2),
- ydb.StructFieldValue("name", v3),
- ydb.StructFieldValue("directory", v4),
- )
- }
- v0 = v1
- }
- vs0[i0] = v0
- }
- if len(vs0) == 0 {
- var t1 ydb.Type
- {
- var t2 ydb.Type
- {
- fs0 := make([]ydb.StructOption, 3)
- var t3 ydb.Type
- {
- tp0 := ydb.TypeInt64
- t3 = tp0
- }
- fs0[0] = ydb.StructField("dir_hash", t3)
- var t4 ydb.Type
- {
- tp0 := ydb.TypeUTF8
- t4 = tp0
- }
- fs0[1] = ydb.StructField("name", t4)
- var t5 ydb.Type
- {
- tp0 := ydb.TypeUTF8
- t5 = tp0
- }
- fs0[2] = ydb.StructField("directory", t5)
- t2 = ydb.Struct(fs0...)
- }
- t1 = t2
- }
- t0 := ydb.List(t1)
- list0 = ydb.ZeroValue(t0)
- } else {
- list0 = ydb.ListValue(vs0...)
- }
- return list0
-}
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index 8779e9ac0..928aad253 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -38,6 +38,7 @@ import (
_ "github.com/chrislusf/seaweedfs/weed/filer/redis2"
_ "github.com/chrislusf/seaweedfs/weed/filer/redis3"
_ "github.com/chrislusf/seaweedfs/weed/filer/sqlite"
+ _ "github.com/chrislusf/seaweedfs/weed/filer/ydb"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/notification"
_ "github.com/chrislusf/seaweedfs/weed/notification/aws_sqs"
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-24 14:54:10 +0000