Browse Source

Merge pull request #3040 from kmlebedev/ydb_fix

fixes from comments
pull/3043/head
Chris Lu 3 years ago
committed by GitHub
parent
commit
e41b11b004
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
  1. 15
      weed/filer/ydb/ydb_queries.go
  2. 54
      weed/filer/ydb/ydb_store.go
  3. 9
      weed/filer/ydb/ydb_store_kv.go
  4. 4
      weed/filer/ydb/ydb_types.go

15
weed/filer/ydb/ydb_queries.go

@ -6,7 +6,7 @@ package ydb
import asql "github.com/chrislusf/seaweedfs/weed/filer/abstract_sql" import asql "github.com/chrislusf/seaweedfs/weed/filer/abstract_sql"
const ( const (
insertQuery = `
upsertQuery = `
PRAGMA TablePathPrefix("%v"); PRAGMA TablePathPrefix("%v");
DECLARE $dir_hash AS int64; DECLARE $dir_hash AS int64;
DECLARE $directory AS Utf8; DECLARE $directory AS Utf8;
@ -19,19 +19,6 @@ const (
VALUES VALUES
($dir_hash, $name, $directory, $meta, $expire_at);` ($dir_hash, $name, $directory, $meta, $expire_at);`
updateQuery = `
PRAGMA TablePathPrefix("%v");
DECLARE $dir_hash AS int64;
DECLARE $directory AS Utf8;
DECLARE $name AS Utf8;
DECLARE $meta AS String;
DECLARE $expire_at AS Optional<uint32>;
REPLACE INTO ` + asql.DEFAULT_TABLE + `
(dir_hash, name, directory, meta, expire_at)
VALUES
($dir_hash, $name, $directory, $meta, $expire_at);`
deleteQuery = ` deleteQuery = `
PRAGMA TablePathPrefix("%v"); PRAGMA TablePathPrefix("%v");
DECLARE $dir_hash AS int64; DECLARE $dir_hash AS int64;

54
weed/filer/ydb/ydb_store.go

@ -28,6 +28,7 @@ import (
const ( const (
defaultDialTimeOut = 10 defaultDialTimeOut = 10
maxRowsInQuery = 1000 // Limit number of rows in query results https://cloud.yandex.com/en-ru/docs/ydb/concepts/limits-ydb
) )
var ( var (
@ -89,7 +90,7 @@ func (store *YdbStore) initialize(dirBuckets string, dsn string, tablePathPrefix
dsn = os.Getenv("YDB_CONNECTION_STRING") dsn = os.Getenv("YDB_CONNECTION_STRING")
} }
store.DB, err = ydb.Open(ctx, dsn, opts...) store.DB, err = ydb.Open(ctx, dsn, opts...)
if err != nil || store.DB == nil {
if err != nil {
if store.DB != nil { if store.DB != nil {
_ = store.DB.Close(ctx) _ = store.DB.Close(ctx)
store.DB = nil store.DB = nil
@ -139,7 +140,7 @@ func (store *YdbStore) doTxOrDB(ctx context.Context, query *string, params *tabl
return err return err
} }
func (store *YdbStore) insertOrUpdateEntry(ctx context.Context, entry *filer.Entry, isUpdate bool) (err error) {
func (store *YdbStore) insertOrUpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
dir, name := entry.FullPath.DirAndName() dir, name := entry.FullPath.DirAndName()
meta, err := entry.EncodeAttributesAndChunks() meta, err := entry.EncodeAttributesAndChunks()
if err != nil { if err != nil {
@ -151,21 +152,15 @@ func (store *YdbStore) insertOrUpdateEntry(ctx context.Context, entry *filer.Ent
} }
tablePathPrefix, shortDir := store.getPrefix(ctx, &dir) tablePathPrefix, shortDir := store.getPrefix(ctx, &dir)
fileMeta := FileMeta{util.HashStringToLong(dir), name, *shortDir, meta} fileMeta := FileMeta{util.HashStringToLong(dir), name, *shortDir, meta}
var query *string
if isUpdate {
query = withPragma(tablePathPrefix, updateQuery)
} else {
query = withPragma(tablePathPrefix, insertQuery)
}
return store.doTxOrDB(ctx, query, fileMeta.queryParameters(entry.TtlSec), rwTX, nil)
return store.doTxOrDB(ctx, withPragma(tablePathPrefix, upsertQuery), fileMeta.queryParameters(entry.TtlSec), rwTX, nil)
} }
func (store *YdbStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) { func (store *YdbStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
return store.insertOrUpdateEntry(ctx, entry, false)
return store.insertOrUpdateEntry(ctx, entry)
} }
func (store *YdbStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) { func (store *YdbStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
return store.insertOrUpdateEntry(ctx, entry, true)
return store.insertOrUpdateEntry(ctx, entry)
} }
func (store *YdbStore) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) { func (store *YdbStore) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) {
@ -179,7 +174,9 @@ func (store *YdbStore) FindEntry(ctx context.Context, fullpath util.FullPath) (e
table.ValueParam("$name", types.UTF8Value(name))) table.ValueParam("$name", types.UTF8Value(name)))
err = store.doTxOrDB(ctx, query, queryParams, roTX, func(res result.Result) error { err = store.doTxOrDB(ctx, query, queryParams, roTX, func(res result.Result) error {
for res.NextResultSet(ctx) {
if !res.NextResultSet(ctx) || !res.HasNextRow() {
return nil
}
for res.NextRow() { for res.NextRow() {
if err = res.ScanNamed(named.OptionalWithDefault("meta", &data)); err != nil { if err = res.ScanNamed(named.OptionalWithDefault("meta", &data)); err != nil {
return fmt.Errorf("scanNamed %s : %v", fullpath, err) return fmt.Errorf("scanNamed %s : %v", fullpath, err)
@ -187,7 +184,6 @@ func (store *YdbStore) FindEntry(ctx context.Context, fullpath util.FullPath) (e
entryFound = true entryFound = true
return nil return nil
} }
}
return res.Err() return res.Err()
}) })
if err != nil { if err != nil {
@ -230,7 +226,7 @@ func (store *YdbStore) DeleteFolderChildren(ctx context.Context, fullpath util.F
} }
func (store *YdbStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) { func (store *YdbStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", nil)
return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", eachEntryFunc)
} }
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) { 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) {
@ -242,17 +238,39 @@ func (store *YdbStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath
} else { } else {
query = withPragma(tablePathPrefix, listDirectoryQuery) query = withPragma(tablePathPrefix, listDirectoryQuery)
} }
truncated := true
eachEntryFuncIsNotBreake := true
shortLimit := limit
if limit > maxRowsInQuery {
shortLimit = maxRowsInQuery * 2
}
entryCount := int64(0)
for truncated && eachEntryFuncIsNotBreake {
if lastFileName != "" {
startFileName = lastFileName
if includeStartFile {
query = withPragma(tablePathPrefix, listDirectoryQuery)
}
}
restLimit := limit - entryCount
if maxRowsInQuery > restLimit {
shortLimit = restLimit
}
queryParams := table.NewQueryParameters( queryParams := table.NewQueryParameters(
table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(*shortDir))), table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(*shortDir))),
table.ValueParam("$directory", types.UTF8Value(*shortDir)), table.ValueParam("$directory", types.UTF8Value(*shortDir)),
table.ValueParam("$start_name", types.UTF8Value(startFileName)), table.ValueParam("$start_name", types.UTF8Value(startFileName)),
table.ValueParam("$prefix", types.UTF8Value(prefix+"%")), table.ValueParam("$prefix", types.UTF8Value(prefix+"%")),
table.ValueParam("$limit", types.Uint64Value(uint64(limit))),
table.ValueParam("$limit", types.Uint64Value(uint64(shortLimit))),
) )
err = store.doTxOrDB(ctx, query, queryParams, roTX, func(res result.Result) error { err = store.doTxOrDB(ctx, query, queryParams, roTX, func(res result.Result) error {
var name string var name string
var data []byte var data []byte
for res.NextResultSet(ctx) {
if !res.NextResultSet(ctx) || !res.HasNextRow() {
truncated = false
return nil
}
truncated = res.CurrentResultSet().Truncated()
for res.NextRow() { for res.NextRow() {
if err := res.ScanNamed( if err := res.ScanNamed(
named.OptionalWithDefault("name", &name), named.OptionalWithDefault("name", &name),
@ -267,12 +285,14 @@ func (store *YdbStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath
return fmt.Errorf("scan decode %s : %v", entry.FullPath, err) return fmt.Errorf("scan decode %s : %v", entry.FullPath, err)
} }
if !eachEntryFunc(entry) { if !eachEntryFunc(entry) {
eachEntryFuncIsNotBreake = false
break break
} }
}
entryCount += 1
} }
return res.Err() return res.Err()
}) })
}
if err != nil { if err != nil {
return lastFileName, err return lastFileName, err
} }

9
weed/filer/ydb/ydb_store_kv.go

@ -18,7 +18,7 @@ func (store *YdbStore) KvPut(ctx context.Context, key []byte, value []byte) (err
dirStr, dirHash, name := abstract_sql.GenDirAndName(key) dirStr, dirHash, name := abstract_sql.GenDirAndName(key)
fileMeta := FileMeta{dirHash, name, dirStr, value} fileMeta := FileMeta{dirHash, name, dirStr, value}
return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) { return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) {
_, _, err = s.Execute(ctx, rwTX, *withPragma(&store.tablePathPrefix, insertQuery),
_, _, err = s.Execute(ctx, rwTX, *withPragma(&store.tablePathPrefix, upsertQuery),
fileMeta.queryParameters(0)) fileMeta.queryParameters(0))
if err != nil { if err != nil {
return fmt.Errorf("kv put execute %s: %v", util.NewFullPath(dirStr, name).Name(), err) return fmt.Errorf("kv put execute %s: %v", util.NewFullPath(dirStr, name).Name(), err)
@ -39,7 +39,9 @@ func (store *YdbStore) KvGet(ctx context.Context, key []byte) (value []byte, err
return fmt.Errorf("kv get execute %s: %v", util.NewFullPath(dirStr, name).Name(), err) return fmt.Errorf("kv get execute %s: %v", util.NewFullPath(dirStr, name).Name(), err)
} }
defer func() { _ = res.Close() }() defer func() { _ = res.Close() }()
for res.NextResultSet(ctx) {
if !res.NextResultSet(ctx) || !res.HasNextRow() {
return nil
}
for res.NextRow() { for res.NextRow() {
if err := res.ScanNamed(named.OptionalWithDefault("meta", &value)); err != nil { if err := res.ScanNamed(named.OptionalWithDefault("meta", &value)); err != nil {
return fmt.Errorf("scanNamed %s : %v", util.NewFullPath(dirStr, name).Name(), err) return fmt.Errorf("scanNamed %s : %v", util.NewFullPath(dirStr, name).Name(), err)
@ -47,7 +49,6 @@ func (store *YdbStore) KvGet(ctx context.Context, key []byte) (value []byte, err
valueFound = true valueFound = true
return nil return nil
} }
}
return res.Err() return res.Err()
}) })
@ -61,7 +62,7 @@ func (store *YdbStore) KvGet(ctx context.Context, key []byte) (value []byte, err
func (store *YdbStore) KvDelete(ctx context.Context, key []byte) (err error) { func (store *YdbStore) KvDelete(ctx context.Context, key []byte) (err error) {
dirStr, dirHash, name := abstract_sql.GenDirAndName(key) dirStr, dirHash, name := abstract_sql.GenDirAndName(key)
return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) { return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) {
_, _, err = s.Execute(ctx, rwTX, *withPragma(&store.tablePathPrefix, insertQuery),
_, _, err = s.Execute(ctx, rwTX, *withPragma(&store.tablePathPrefix, deleteQuery),
table.NewQueryParameters( table.NewQueryParameters(
table.ValueParam("$dir_hash", types.Int64Value(dirHash)), table.ValueParam("$dir_hash", types.Int64Value(dirHash)),
table.ValueParam("$name", types.UTF8Value(name)))) table.ValueParam("$name", types.UTF8Value(name))))

4
weed/filer/ydb/ydb_types.go

@ -10,9 +10,6 @@ import (
"github.com/ydb-platform/ydb-go-sdk/v3/table/types" "github.com/ydb-platform/ydb-go-sdk/v3/table/types"
) )
//go:generate ydbgen
//ydb:gen
type FileMeta struct { type FileMeta struct {
DirHash int64 `ydb:"type:int64"` DirHash int64 `ydb:"type:int64"`
Name string `ydb:"type:utf8"` Name string `ydb:"type:utf8"`
@ -20,7 +17,6 @@ type FileMeta struct {
Meta []byte `ydb:"type:string"` Meta []byte `ydb:"type:string"`
} }
//ydb:gen scan,value
type FileMetas []FileMeta type FileMetas []FileMeta
func (fm *FileMeta) queryParameters(ttlSec int32) *table.QueryParameters { func (fm *FileMeta) queryParameters(ttlSec int32) *table.QueryParameters {

Loading…
Cancel
Save