Merge 6d59a974eb into 626954b76c

weed/filer/abstract_sql/abstract_sql_store.go

@@ -66,11 +66,27 @@ const (
 	DEFAULT_TABLE = "filemeta"
 )
 
-type TxOrDB interface {
+type TxOrDBWithoutClose interface {
 	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
 	QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row
 	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
 }
+type TxOrDB interface {
+	TxOrDBWithoutClose
+	Close() error
+}
+
+type TxOrDBWrapper struct {
+	TxOrDBWithoutClose
+	CloseFunc func() error
+}
+
+func (t *TxOrDBWrapper) Close() error {
+	if t.CloseFunc == nil {
+		return nil
+	}
+	return t.CloseFunc()
+}
 
 func (store *AbstractSqlStore) BeginTransaction(ctx context.Context) (context.Context, error) {
 	tx, err := store.DB.BeginTx(ctx, &sql.TxOptions{
@@ -102,9 +118,14 @@ func (store *AbstractSqlStore) getTxOrDB(ctx context.Context, fullpath util.Full
 	bucket = DEFAULT_TABLE
 
 	if tx, ok := ctx.Value("tx").(*sql.Tx); ok {
-		txOrDB = tx
+		txOrDB = &TxOrDBWrapper{tx, nil}
 	} else {
-		txOrDB = store.DB
+		conn, connErr := store.DB.Conn(context.Background())
+		if connErr != nil {
+			err = connErr
+			return
+		}
+		txOrDB = &TxOrDBWrapper{conn, conn.Close}
 	}
 
 	if !store.SupportBucketTable {
@@ -155,6 +176,7 @@ func (store *AbstractSqlStore) InsertEntry(ctx context.Context, entry *filer.Ent
 	if err != nil {
 		return fmt.Errorf("findDB %s : %v", entry.FullPath, err)
 	}
+	defer db.Close()
 
 	dir, name := shortPath.DirAndName()
 	meta, err := entry.EncodeAttributesAndChunks()
@@ -191,6 +213,7 @@ func (store *AbstractSqlStore) UpdateEntry(ctx context.Context, entry *filer.Ent
 	if err != nil {
 		return fmt.Errorf("findDB %s : %v", entry.FullPath, err)
 	}
+	defer db.Close()
 
 	dir, name := shortPath.DirAndName()
 	meta, err := entry.EncodeAttributesAndChunks()
@@ -216,6 +239,7 @@ func (store *AbstractSqlStore) FindEntry(ctx context.Context, fullpath util.Full
 	if err != nil {
 		return nil, fmt.Errorf("findDB %s : %v", fullpath, err)
 	}
+	defer db.Close()
 
 	dir, name := shortPath.DirAndName()
 	row := db.QueryRowContext(ctx, store.GetSqlFind(bucket), util.HashStringToLong(dir), name, dir)
@@ -244,6 +268,7 @@ func (store *AbstractSqlStore) DeleteEntry(ctx context.Context, fullpath util.Fu
 	if err != nil {
 		return fmt.Errorf("findDB %s : %v", fullpath, err)
 	}
+	defer db.Close()
 
 	dir, name := shortPath.DirAndName()
 
@@ -266,6 +291,7 @@ func (store *AbstractSqlStore) DeleteFolderChildren(ctx context.Context, fullpat
 	if err != nil {
 		return fmt.Errorf("findDB %s : %v", fullpath, err)
 	}
+	defer db.Close()
 
 	if isValidBucket(bucket) && shortPath == "/" {
 		if err = store.deleteTable(ctx, bucket); err == nil {
@@ -297,6 +323,7 @@ func (store *AbstractSqlStore) ListDirectoryPrefixedEntries(ctx context.Context,
 	if err != nil {
 		return lastFileName, fmt.Errorf("findDB %s : %v", dirPath, err)
 	}
+	defer db.Close()
 
 	sqlText := store.GetSqlListExclusive(bucket)
 	if includeStartFile {