You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

358 lines
9.6 KiB

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