abstract_sql_store.go 10.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360
  1. package abstract_sql
  2. import (
  3. "context"
  4. "database/sql"
  5. "fmt"
  6. "github.com/seaweedfs/seaweedfs/weed/filer"
  7. "github.com/seaweedfs/seaweedfs/weed/glog"
  8. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  9. "github.com/seaweedfs/seaweedfs/weed/util"
  10. "strings"
  11. "sync"
  12. )
  13. type SqlGenerator interface {
  14. GetSqlInsert(tableName string) string
  15. GetSqlUpdate(tableName string) string
  16. GetSqlFind(tableName string) string
  17. GetSqlDelete(tableName string) string
  18. GetSqlDeleteFolderChildren(tableName string) string
  19. GetSqlListExclusive(tableName string) string
  20. GetSqlListInclusive(tableName string) string
  21. GetSqlCreateTable(tableName string) string
  22. GetSqlDropTable(tableName string) string
  23. }
  24. type AbstractSqlStore struct {
  25. SqlGenerator
  26. DB *sql.DB
  27. SupportBucketTable bool
  28. dbs map[string]bool
  29. dbsLock sync.Mutex
  30. }
  31. var _ filer.BucketAware = (*AbstractSqlStore)(nil)
  32. func (store *AbstractSqlStore) CanDropWholeBucket() bool {
  33. return store.SupportBucketTable
  34. }
  35. func (store *AbstractSqlStore) OnBucketCreation(bucket string) {
  36. store.dbsLock.Lock()
  37. defer store.dbsLock.Unlock()
  38. store.CreateTable(context.Background(), bucket)
  39. if store.dbs == nil {
  40. return
  41. }
  42. store.dbs[bucket] = true
  43. }
  44. func (store *AbstractSqlStore) OnBucketDeletion(bucket string) {
  45. store.dbsLock.Lock()
  46. defer store.dbsLock.Unlock()
  47. store.deleteTable(context.Background(), bucket)
  48. if store.dbs == nil {
  49. return
  50. }
  51. delete(store.dbs, bucket)
  52. }
  53. const (
  54. DEFAULT_TABLE = "filemeta"
  55. )
  56. type TxOrDB interface {
  57. ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
  58. QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row
  59. QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
  60. }
  61. func (store *AbstractSqlStore) BeginTransaction(ctx context.Context) (context.Context, error) {
  62. tx, err := store.DB.BeginTx(ctx, &sql.TxOptions{
  63. Isolation: sql.LevelReadCommitted,
  64. ReadOnly: false,
  65. })
  66. if err != nil {
  67. return ctx, err
  68. }
  69. return context.WithValue(ctx, "tx", tx), nil
  70. }
  71. func (store *AbstractSqlStore) CommitTransaction(ctx context.Context) error {
  72. if tx, ok := ctx.Value("tx").(*sql.Tx); ok {
  73. return tx.Commit()
  74. }
  75. return nil
  76. }
  77. func (store *AbstractSqlStore) RollbackTransaction(ctx context.Context) error {
  78. if tx, ok := ctx.Value("tx").(*sql.Tx); ok {
  79. return tx.Rollback()
  80. }
  81. return nil
  82. }
  83. func (store *AbstractSqlStore) getTxOrDB(ctx context.Context, fullpath util.FullPath, isForChildren bool) (txOrDB TxOrDB, bucket string, shortPath util.FullPath, err error) {
  84. shortPath = fullpath
  85. bucket = DEFAULT_TABLE
  86. if tx, ok := ctx.Value("tx").(*sql.Tx); ok {
  87. txOrDB = tx
  88. } else {
  89. txOrDB = store.DB
  90. }
  91. if !store.SupportBucketTable {
  92. return
  93. }
  94. if !strings.HasPrefix(string(fullpath), "/buckets/") {
  95. return
  96. }
  97. // detect bucket
  98. bucketAndObjectKey := string(fullpath)[len("/buckets/"):]
  99. t := strings.Index(bucketAndObjectKey, "/")
  100. if t < 0 && !isForChildren {
  101. return
  102. }
  103. bucket = bucketAndObjectKey
  104. shortPath = "/"
  105. if t > 0 {
  106. bucket = bucketAndObjectKey[:t]
  107. shortPath = util.FullPath(bucketAndObjectKey[t:])
  108. }
  109. if isValidBucket(bucket) {
  110. store.dbsLock.Lock()
  111. defer store.dbsLock.Unlock()
  112. if store.dbs == nil {
  113. store.dbs = make(map[string]bool)
  114. }
  115. if _, found := store.dbs[bucket]; !found {
  116. if err = store.CreateTable(ctx, bucket); err == nil {
  117. store.dbs[bucket] = true
  118. }
  119. }
  120. }
  121. return
  122. }
  123. func (store *AbstractSqlStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
  124. db, bucket, shortPath, err := store.getTxOrDB(ctx, entry.FullPath, false)
  125. if err != nil {
  126. return fmt.Errorf("findDB %s : %v", entry.FullPath, err)
  127. }
  128. dir, name := shortPath.DirAndName()
  129. meta, err := entry.EncodeAttributesAndChunks()
  130. if err != nil {
  131. return fmt.Errorf("encode %s: %s", entry.FullPath, err)
  132. }
  133. if len(entry.GetChunks()) > filer.CountEntryChunksForGzip {
  134. meta = util.MaybeGzipData(meta)
  135. }
  136. sqlInsert := "insert"
  137. res, err := db.ExecContext(ctx, store.GetSqlInsert(bucket), util.HashStringToLong(dir), name, dir, meta)
  138. if err != nil && strings.Contains(strings.ToLower(err.Error()), "duplicate entry") {
  139. // now the insert failed possibly due to duplication constraints
  140. sqlInsert = "falls back to update"
  141. glog.V(1).Infof("insert %s %s: %v", entry.FullPath, sqlInsert, err)
  142. res, err = db.ExecContext(ctx, store.GetSqlUpdate(bucket), meta, util.HashStringToLong(dir), name, dir)
  143. }
  144. if err != nil {
  145. return fmt.Errorf("%s %s: %s", sqlInsert, entry.FullPath, err)
  146. }
  147. _, err = res.RowsAffected()
  148. if err != nil {
  149. return fmt.Errorf("%s %s but no rows affected: %s", sqlInsert, entry.FullPath, err)
  150. }
  151. return nil
  152. }
  153. func (store *AbstractSqlStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
  154. db, bucket, shortPath, err := store.getTxOrDB(ctx, entry.FullPath, false)
  155. if err != nil {
  156. return fmt.Errorf("findDB %s : %v", entry.FullPath, err)
  157. }
  158. dir, name := shortPath.DirAndName()
  159. meta, err := entry.EncodeAttributesAndChunks()
  160. if err != nil {
  161. return fmt.Errorf("encode %s: %s", entry.FullPath, err)
  162. }
  163. res, err := db.ExecContext(ctx, store.GetSqlUpdate(bucket), meta, util.HashStringToLong(dir), name, dir)
  164. if err != nil {
  165. return fmt.Errorf("update %s: %s", entry.FullPath, err)
  166. }
  167. _, err = res.RowsAffected()
  168. if err != nil {
  169. return fmt.Errorf("update %s but no rows affected: %s", entry.FullPath, err)
  170. }
  171. return nil
  172. }
  173. func (store *AbstractSqlStore) FindEntry(ctx context.Context, fullpath util.FullPath) (*filer.Entry, error) {
  174. db, bucket, shortPath, err := store.getTxOrDB(ctx, fullpath, false)
  175. if err != nil {
  176. return nil, fmt.Errorf("findDB %s : %v", fullpath, err)
  177. }
  178. dir, name := shortPath.DirAndName()
  179. row := db.QueryRowContext(ctx, store.GetSqlFind(bucket), util.HashStringToLong(dir), name, dir)
  180. var data []byte
  181. if err := row.Scan(&data); err != nil {
  182. if err == sql.ErrNoRows {
  183. return nil, filer_pb.ErrNotFound
  184. }
  185. return nil, fmt.Errorf("find %s: %v", fullpath, err)
  186. }
  187. entry := &filer.Entry{
  188. FullPath: fullpath,
  189. }
  190. if err := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); err != nil {
  191. return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
  192. }
  193. return entry, nil
  194. }
  195. func (store *AbstractSqlStore) DeleteEntry(ctx context.Context, fullpath util.FullPath) error {
  196. db, bucket, shortPath, err := store.getTxOrDB(ctx, fullpath, false)
  197. if err != nil {
  198. return fmt.Errorf("findDB %s : %v", fullpath, err)
  199. }
  200. dir, name := shortPath.DirAndName()
  201. res, err := db.ExecContext(ctx, store.GetSqlDelete(bucket), util.HashStringToLong(dir), name, dir)
  202. if err != nil {
  203. return fmt.Errorf("delete %s: %s", fullpath, err)
  204. }
  205. _, err = res.RowsAffected()
  206. if err != nil {
  207. return fmt.Errorf("delete %s but no rows affected: %s", fullpath, err)
  208. }
  209. return nil
  210. }
  211. func (store *AbstractSqlStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) error {
  212. db, bucket, shortPath, err := store.getTxOrDB(ctx, fullpath, true)
  213. if err != nil {
  214. return fmt.Errorf("findDB %s : %v", fullpath, err)
  215. }
  216. if isValidBucket(bucket) && shortPath == "/" {
  217. if err = store.deleteTable(ctx, bucket); err == nil {
  218. store.dbsLock.Lock()
  219. delete(store.dbs, bucket)
  220. store.dbsLock.Unlock()
  221. return nil
  222. } else {
  223. return err
  224. }
  225. }
  226. glog.V(4).Infof("delete %s SQL %s %d", string(shortPath), store.GetSqlDeleteFolderChildren(bucket), util.HashStringToLong(string(shortPath)))
  227. res, err := db.ExecContext(ctx, store.GetSqlDeleteFolderChildren(bucket), util.HashStringToLong(string(shortPath)), string(shortPath))
  228. if err != nil {
  229. return fmt.Errorf("deleteFolderChildren %s: %s", fullpath, err)
  230. }
  231. _, err = res.RowsAffected()
  232. if err != nil {
  233. return fmt.Errorf("deleteFolderChildren %s but no rows affected: %s", fullpath, err)
  234. }
  235. return nil
  236. }
  237. func (store *AbstractSqlStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  238. db, bucket, shortPath, err := store.getTxOrDB(ctx, dirPath, true)
  239. if err != nil {
  240. return lastFileName, fmt.Errorf("findDB %s : %v", dirPath, err)
  241. }
  242. sqlText := store.GetSqlListExclusive(bucket)
  243. if includeStartFile {
  244. sqlText = store.GetSqlListInclusive(bucket)
  245. }
  246. rows, err := db.QueryContext(ctx, sqlText, util.HashStringToLong(string(shortPath)), startFileName, string(shortPath), prefix+"%", limit+1)
  247. if err != nil {
  248. return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
  249. }
  250. defer rows.Close()
  251. for rows.Next() {
  252. var name string
  253. var data []byte
  254. if err = rows.Scan(&name, &data); err != nil {
  255. glog.V(0).Infof("scan %s : %v", dirPath, err)
  256. return lastFileName, fmt.Errorf("scan %s: %v", dirPath, err)
  257. }
  258. lastFileName = name
  259. entry := &filer.Entry{
  260. FullPath: util.NewFullPath(string(dirPath), name),
  261. }
  262. if err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); err != nil {
  263. glog.V(0).Infof("scan decode %s : %v", entry.FullPath, err)
  264. return lastFileName, fmt.Errorf("scan decode %s : %v", entry.FullPath, err)
  265. }
  266. if !eachEntryFunc(entry) {
  267. break
  268. }
  269. }
  270. return lastFileName, nil
  271. }
  272. func (store *AbstractSqlStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  273. return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", nil)
  274. }
  275. func (store *AbstractSqlStore) Shutdown() {
  276. store.DB.Close()
  277. }
  278. func isValidBucket(bucket string) bool {
  279. return bucket != DEFAULT_TABLE && bucket != ""
  280. }
  281. func (store *AbstractSqlStore) CreateTable(ctx context.Context, bucket string) error {
  282. if !store.SupportBucketTable {
  283. return nil
  284. }
  285. _, err := store.DB.ExecContext(ctx, store.SqlGenerator.GetSqlCreateTable(bucket))
  286. return err
  287. }
  288. func (store *AbstractSqlStore) deleteTable(ctx context.Context, bucket string) error {
  289. if !store.SupportBucketTable {
  290. return nil
  291. }
  292. _, err := store.DB.ExecContext(ctx, store.SqlGenerator.GetSqlDropTable(bucket))
  293. return err
  294. }