mirror of
https://github.com/chrislusf/seaweedfs
synced 2025-06-29 16:22:46 +02:00
abd5102819
* fix(ydb): table creation with params * fix(ydb): migrate to new query client & prevent creation table-bucket on get request * fix(ydb): use new query client with kv req * fix(ydb): use directory in every query * fix(ydb): del unused import * fix(ydb): tests & default const usage
82 lines
2.7 KiB
Go
82 lines
2.7 KiB
Go
//go:build ydb
|
|
// +build ydb
|
|
|
|
package ydb
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"github.com/seaweedfs/seaweedfs/weed/filer"
|
|
"github.com/seaweedfs/seaweedfs/weed/filer/abstract_sql"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
"github.com/ydb-platform/ydb-go-sdk/v3/query"
|
|
"github.com/ydb-platform/ydb-go-sdk/v3/table"
|
|
"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 store.DB.Query().Do(ctx, func(ctx context.Context, s query.Session) (err error) {
|
|
_, err = s.Query(ctx, *withPragma(&store.tablePathPrefix, upsertQuery),
|
|
query.WithParameters(fileMeta.queryParameters(0)), rwQC)
|
|
if err != nil {
|
|
return fmt.Errorf("kv put execute %s: %v", util.NewFullPath(dirStr, name).Name(), err)
|
|
}
|
|
return nil
|
|
}, query.WithIdempotent())
|
|
}
|
|
|
|
func (store *YdbStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
|
|
dirStr, dirHash, name := abstract_sql.GenDirAndName(key)
|
|
valueFound := false
|
|
err = store.DB.Query().Do(ctx, func(ctx context.Context, s query.Session) error {
|
|
res, err := s.Query(ctx, *withPragma(&store.tablePathPrefix, findQuery),
|
|
query.WithParameters(table.NewQueryParameters(
|
|
table.ValueParam("$dir_hash", types.Int64Value(dirHash)),
|
|
table.ValueParam("$directory", types.UTF8Value(dirStr)),
|
|
table.ValueParam("$name", types.UTF8Value(name)))), roQC)
|
|
if err != nil {
|
|
return fmt.Errorf("kv get execute %s: %v", util.NewFullPath(dirStr, name).Name(), err)
|
|
}
|
|
defer func() { _ = res.Close(ctx) }()
|
|
for rs, err := range res.ResultSets(ctx) {
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for row, err := range rs.Rows(ctx) {
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := row.Scan(&value); err != nil {
|
|
return fmt.Errorf("scan %s : %v", util.NewFullPath(dirStr, name).Name(), err)
|
|
}
|
|
valueFound = true
|
|
return nil
|
|
}
|
|
}
|
|
return nil
|
|
}, query.WithIdempotent())
|
|
|
|
if !valueFound {
|
|
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 store.DB.Query().Do(ctx, func(ctx context.Context, s query.Session) (err error) {
|
|
_, err = s.Query(ctx, *withPragma(&store.tablePathPrefix, deleteQuery),
|
|
query.WithParameters(table.NewQueryParameters(
|
|
table.ValueParam("$dir_hash", types.Int64Value(dirHash)),
|
|
table.ValueParam("$directory", types.UTF8Value(dirStr)),
|
|
table.ValueParam("$name", types.UTF8Value(name)))), rwQC)
|
|
if err != nil {
|
|
return fmt.Errorf("kv delete %s: %v", util.NewFullPath(dirStr, name).Name(), err)
|
|
}
|
|
return nil
|
|
}, query.WithIdempotent())
|
|
|
|
}
|