Mirror
/
seaweedfs
mirror of https://github.com/seaweedfs/seaweedfs.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

add more input params

pull/1759/head
Chris Lu 4 years ago
parent
f13c082583
commit
90ce1eec19
2 changed files with 12 additions and 12 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 16
      weed/filer/abstract_sql/abstract_sql_store.go
  2. 8
      weed/filer/abstract_sql/abstract_sql_store_kv.go

16
weed/filer/abstract_sql/abstract_sql_store.go
View File

@ -52,7 +52,7 @@ func (store *AbstractSqlStore) RollbackTransaction(ctx context.Context) error {
return nil return nil
} }
func (store *AbstractSqlStore) getTxOrDB(ctx context.Context) TxOrDB {
func (store *AbstractSqlStore) getTxOrDB(ctx context.Context, fullpath util.FullPath, isForChildren bool) TxOrDB {
if tx, ok := ctx.Value("tx").(*sql.Tx); ok { if tx, ok := ctx.Value("tx").(*sql.Tx); ok {
return tx return tx
} }
@ -71,7 +71,7 @@ func (store *AbstractSqlStore) InsertEntry(ctx context.Context, entry *filer.Ent
meta = util.MaybeGzipData(meta) meta = util.MaybeGzipData(meta)
} }
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlInsert, util.HashStringToLong(dir), name, dir, meta)
res, err := store.getTxOrDB(ctx, entry.FullPath, false).ExecContext(ctx, store.SqlInsert, util.HashStringToLong(dir), name, dir, meta)
if err == nil { if err == nil {
return return
} }
@ -84,7 +84,7 @@ func (store *AbstractSqlStore) InsertEntry(ctx context.Context, entry *filer.Ent
// now the insert failed possibly due to duplication constraints // now the insert failed possibly due to duplication constraints
glog.V(1).Infof("insert %s falls back to update: %v", entry.FullPath, err) glog.V(1).Infof("insert %s falls back to update: %v", entry.FullPath, err)
res, err = store.getTxOrDB(ctx).ExecContext(ctx, store.SqlUpdate, meta, util.HashStringToLong(dir), name, dir)
res, err = store.getTxOrDB(ctx, entry.FullPath, false).ExecContext(ctx, store.SqlUpdate, meta, util.HashStringToLong(dir), name, dir)
if err != nil { if err != nil {
return fmt.Errorf("upsert %s: %s", entry.FullPath, err) return fmt.Errorf("upsert %s: %s", entry.FullPath, err)
} }
@ -105,7 +105,7 @@ func (store *AbstractSqlStore) UpdateEntry(ctx context.Context, entry *filer.Ent
return fmt.Errorf("encode %s: %s", entry.FullPath, err) return fmt.Errorf("encode %s: %s", entry.FullPath, err)
} }
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlUpdate, meta, util.HashStringToLong(dir), name, dir)
res, err := store.getTxOrDB(ctx, entry.FullPath, false).ExecContext(ctx, store.SqlUpdate, meta, util.HashStringToLong(dir), name, dir)
if err != nil { if err != nil {
return fmt.Errorf("update %s: %s", entry.FullPath, err) return fmt.Errorf("update %s: %s", entry.FullPath, err)
} }
@ -120,7 +120,7 @@ func (store *AbstractSqlStore) UpdateEntry(ctx context.Context, entry *filer.Ent
func (store *AbstractSqlStore) FindEntry(ctx context.Context, fullpath util.FullPath) (*filer.Entry, error) { func (store *AbstractSqlStore) FindEntry(ctx context.Context, fullpath util.FullPath) (*filer.Entry, error) {
dir, name := fullpath.DirAndName() dir, name := fullpath.DirAndName()
row := store.getTxOrDB(ctx).QueryRowContext(ctx, store.SqlFind, util.HashStringToLong(dir), name, dir)
row := store.getTxOrDB(ctx, fullpath, false).QueryRowContext(ctx, store.SqlFind, util.HashStringToLong(dir), name, dir)
var data []byte var data []byte
if err := row.Scan(&data); err != nil { if err := row.Scan(&data); err != nil {
@ -144,7 +144,7 @@ func (store *AbstractSqlStore) DeleteEntry(ctx context.Context, fullpath util.Fu
dir, name := fullpath.DirAndName() dir, name := fullpath.DirAndName()
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlDelete, util.HashStringToLong(dir), name, dir)
res, err := store.getTxOrDB(ctx, fullpath, false).ExecContext(ctx, store.SqlDelete, util.HashStringToLong(dir), name, dir)
if err != nil { if err != nil {
return fmt.Errorf("delete %s: %s", fullpath, err) return fmt.Errorf("delete %s: %s", fullpath, err)
} }
@ -159,7 +159,7 @@ func (store *AbstractSqlStore) DeleteEntry(ctx context.Context, fullpath util.Fu
func (store *AbstractSqlStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) error { func (store *AbstractSqlStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) error {
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlDeleteFolderChildren, util.HashStringToLong(string(fullpath)), fullpath)
res, err := store.getTxOrDB(ctx, fullpath, true).ExecContext(ctx, store.SqlDeleteFolderChildren, util.HashStringToLong(string(fullpath)), fullpath)
if err != nil { if err != nil {
return fmt.Errorf("deleteFolderChildren %s: %s", fullpath, err) return fmt.Errorf("deleteFolderChildren %s: %s", fullpath, err)
} }
@ -178,7 +178,7 @@ func (store *AbstractSqlStore) ListDirectoryPrefixedEntries(ctx context.Context,
sqlText = store.SqlListInclusive sqlText = store.SqlListInclusive
} }
rows, err := store.getTxOrDB(ctx).QueryContext(ctx, sqlText, util.HashStringToLong(string(dirPath)), startFileName, string(dirPath), prefix+"%", limit+1)
rows, err := store.getTxOrDB(ctx, dirPath, true).QueryContext(ctx, sqlText, util.HashStringToLong(string(dirPath)), startFileName, string(dirPath), prefix+"%", limit+1)
if err != nil { if err != nil {
return lastFileName, fmt.Errorf("list %s : %v", dirPath, err) return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
} }

8
weed/filer/abstract_sql/abstract_sql_store_kv.go
View File

@ -15,7 +15,7 @@ func (store *AbstractSqlStore) KvPut(ctx context.Context, key []byte, value []by
dirStr, dirHash, name := genDirAndName(key) dirStr, dirHash, name := genDirAndName(key)
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlInsert, dirHash, name, dirStr, value)
res, err := store.getTxOrDB(ctx, "", false).ExecContext(ctx, store.SqlInsert, dirHash, name, dirStr, value)
if err == nil { if err == nil {
return return
} }
@ -28,7 +28,7 @@ func (store *AbstractSqlStore) KvPut(ctx context.Context, key []byte, value []by
// now the insert failed possibly due to duplication constraints // now the insert failed possibly due to duplication constraints
glog.V(1).Infof("kv insert falls back to update: %s", err) glog.V(1).Infof("kv insert falls back to update: %s", err)
res, err = store.getTxOrDB(ctx).ExecContext(ctx, store.SqlUpdate, value, dirHash, name, dirStr)
res, err = store.getTxOrDB(ctx, "", false).ExecContext(ctx, store.SqlUpdate, value, dirHash, name, dirStr)
if err != nil { if err != nil {
return fmt.Errorf("kv upsert: %s", err) return fmt.Errorf("kv upsert: %s", err)
} }
@ -44,7 +44,7 @@ func (store *AbstractSqlStore) KvPut(ctx context.Context, key []byte, value []by
func (store *AbstractSqlStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { func (store *AbstractSqlStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) {
dirStr, dirHash, name := genDirAndName(key) dirStr, dirHash, name := genDirAndName(key)
row := store.getTxOrDB(ctx).QueryRowContext(ctx, store.SqlFind, dirHash, name, dirStr)
row := store.getTxOrDB(ctx, "", false).QueryRowContext(ctx, store.SqlFind, dirHash, name, dirStr)
err = row.Scan(&value) err = row.Scan(&value)
@ -63,7 +63,7 @@ func (store *AbstractSqlStore) KvDelete(ctx context.Context, key []byte) (err er
dirStr, dirHash, name := genDirAndName(key) dirStr, dirHash, name := genDirAndName(key)
res, err := store.getTxOrDB(ctx).ExecContext(ctx, store.SqlDelete, dirHash, name, dirStr)
res, err := store.getTxOrDB(ctx, "", false).ExecContext(ctx, store.SqlDelete, dirHash, name, dirStr)
if err != nil { if err != nil {
return fmt.Errorf("kv delete: %s", err) return fmt.Errorf("kv delete: %s", err)
} }

Write Preview
Loading…
Cancel
Save