Fix versioning.

api/object-handlers.go

@@ -880,7 +880,7 @@ func (api ObjectAPIHandlers) AppendObjectHandler(w http.ResponseWriter, r *http.
 		return
 	}
 
-	if objInfo != nil && objInfo.Type != meta.ObjectTypeAppendable {
+	if objInfo != nil && (objInfo.Type != meta.ObjectTypeAppendable || objInfo.DeleteMarker) {
 		WriteErrorResponse(w, r, ErrObjectNotAppendable)
 		return
 	}

meta/client/client.go

@@ -17,7 +17,7 @@ type Client interface {
 	GetObject(bucketName, objectName, version string) (object *Object, err error)
 	GetAllObject(bucketName, objectName, version string) (object []*Object, err error)
 	PutObject(object *Object, tx interface{}) error
-	UpdateAppendObject(object *Object) error
+	UpdateAppendObject(object *Object, versionId string) error
 	UpdateObjectAttrs(object *Object) error
 	DeleteObject(object *Object, tx interface{}) error
 	UpdateObjectAcl(object *Object) error
@@ -27,7 +27,7 @@ type Client interface {
 	PutBucket(bucket *Bucket) error
 	CheckAndPutBucket(bucket *Bucket) (bool, error)
 	DeleteBucket(bucket *Bucket) error
-	ListObjects(bucketName, marker, verIdMarker, prefix, delimiter string, versioned bool, maxKeys int) (retObjects []*Object, prefixes []string, truncated bool, nextMarker, nextVerIdMarker string, err error)
+	ListObjects(bucketName, marker, verIdMarker, prefix, delimiter string, versioned bool, maxKeys int, withDeleteMarker bool) (retObjects []*Object, prefixes []string, truncated bool, nextMarker, nextVerIdMarker string, err error)
 	UpdateUsage(bucketName string, size int64, tx interface{}) error
 	UpdateUsages(usages map[string]int64, tx interface{}) error
 
@@ -37,10 +37,6 @@ type Client interface {
 	PutObjectPart(multipart *Multipart, part *Part, tx interface{}) (err error)
 	DeleteMultipart(multipart *Multipart, tx interface{}) (err error)
 	ListMultipartUploads(bucketName, keyMarker, uploadIdMarker, prefix, delimiter, encodingType string, maxUploads int) (uploads []datatype.Upload, prefixs []string, isTruncated bool, nextKeyMarker, nextUploadIdMarker string, err error)
-	//objmap
-	GetObjectMap(bucketName, objectName string) (objMap *ObjMap, err error)
-	PutObjectMap(objMap *ObjMap, tx interface{}) error
-	DeleteObjectMap(objMap *ObjMap, tx interface{}) error
 	//cluster
 	GetCluster(fsid, pool string) (cluster Cluster, err error)
 	//lc

meta/client/tidbclient/bucket.go

@@ -3,7 +3,6 @@ package tidbclient
 import (
 	"database/sql"
 	"encoding/json"
-	"strconv"
 	"strings"
 	"time"
 
@@ -153,23 +152,23 @@ func (t *TidbClient) CheckAndPutBucket(bucket *Bucket) (bool, error) {
 	return processed, err
 }
 
-func (t *TidbClient) ListObjects(bucketName, marker, verIdMarker, prefix, delimiter string, versioned bool, maxKeys int) (retObjects []*Object, prefixes []string, truncated bool, nextMarker, nextVerIdMarker string, err error) {
+func (t *TidbClient) ListObjects(bucketName, marker, verIdMarker, prefix, delimiter string, versioned bool, maxKeys int, withDeleteMarker bool) (retObjects []*Object, prefixes []string, truncated bool, nextMarker, nextVerIdMarker string, err error) {
 	const MaxObjectList = 10000
-	if versioned {
-		return
-	}
 	var count int
 	var exit bool
 	objectMap := make(map[string]struct{})
 	objectNum := make(map[string]int)
 	commonPrefixes := make(map[string]struct{})
 	omarker := marker
+
+	helper.Logger.Printf(20, bucketName, marker, verIdMarker, prefix, delimiter, versioned, maxKeys, withDeleteMarker)
+
 	for {
 		var loopcount int
 		var sqltext string
 		var rows *sql.Rows
 		args := make([]interface{}, 0)
-		sqltext = "select bucketname,name,version from objects where bucketName=?"
+		sqltext = "select bucketname,name,version,deletemarker from objects where bucketName=?"
 		args = append(args, bucketName)
 		if prefix != "" {
 			sqltext += " and name like ?"
@@ -205,11 +204,13 @@ func (t *TidbClient) ListObjects(bucketName, marker, verIdMarker, prefix, delimi
 			loopcount += 1
 			//fetch related date
 			var bucketname, name string
-			var version uint64
+			var version uint64 // Internal version, the same as in DB.
+			var deletemarker bool
 			err = rows.Scan(
 				&bucketname,
 				&name,
 				&version,
+				&deletemarker,
 			)
 			if err != nil {
 				return
@@ -222,15 +223,18 @@ func (t *TidbClient) ListObjects(bucketName, marker, verIdMarker, prefix, delimi
 			objectNum[name] += 1
 			marker = name
 
-			if _, ok := objectMap[name]; !ok {
-				objectMap[name] = struct{}{}
-			} else {
-				continue
+			//filte by objectname
+			if !versioned {
+				if _, ok := objectMap[name]; !ok {
+					objectMap[name] = struct{}{}
+				} else {
+					continue
+				}
 			}
 			//filte by deletemarker
-			/*if deletemarker {
+			if deletemarker && !withDeleteMarker {
 				continue
-			}*/
+			}
 			if name == omarker {
 				continue
 			}
@@ -258,10 +262,9 @@ func (t *TidbClient) ListObjects(bucketName, marker, verIdMarker, prefix, delimi
 				}
 			}
 			var o *Object
-			Strver := strconv.FormatUint(version, 10)
-			o, err = t.GetObject(bucketname, name, Strver)
+			o, err = t.GetObject(bucketname, name, ConvertRawVersionToS3Version(version))
 			if err != nil {
-				helper.Logger.Printf(2, "ListObjects: failed to GetObject(%s, %s, %s), err: %v", bucketname, name, Strver, err)
+				helper.Logger.Printf(2, "ListObjects: failed to GetObject(%s, %s, %s), err: %v", bucketname, name, err)
 				return
 			}
 			count += 1

meta/client/tidbclient/object.go

@@ -5,6 +5,7 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	. "github.com/journeymidnight/yig/error"
+	"github.com/journeymidnight/yig/helper"
 	. "github.com/journeymidnight/yig/meta/types"
 	"github.com/xxtea/xxtea-go/xxtea"
 	"math"
@@ -20,9 +21,16 @@ func (t *TidbClient) GetObject(bucketName, objectName, version string) (object *
 	if version == "" {
 		sqltext = "select * from objects where bucketname=? and name=? order by bucketname,name,version limit 1;"
 		row = t.Client.QueryRow(sqltext, bucketName, objectName)
+	} else if version == ObjectNullVersion {
+		sqltext = "select * from objects where bucketname=? and name=? and nullversion=1 limit 1;" // There should be only one NullVersion object.
+		row = t.Client.QueryRow(sqltext, bucketName, objectName)
 	} else {
-		sqltext = "select * from objects where bucketname=? and name=? and version=?;"
-		row = t.Client.QueryRow(sqltext, bucketName, objectName, version)
+		sqltext = "select * from objects where bucketname=? and name=? and version=? limit 1;"
+		internalVersion, err := ConvertS3VersionToRawVersion(version)
+		if err != nil {
+			return nil, ErrInternalError
+		}
+		row = t.Client.QueryRow(sqltext, bucketName, objectName, internalVersion)
 	}
 	object = &Object{}
 	err = row.Scan(
@@ -77,32 +85,46 @@ func (t *TidbClient) GetObject(bucketName, objectName, version string) (object *
 		}
 		object.PartsIndex = &SimpleIndex{Index: sortedPartNum}
 	}
-	var reversedTime uint64
-	timestamp := math.MaxUint64 - reversedTime
-	timeData := []byte(strconv.FormatUint(timestamp, 10))
-	object.VersionId = hex.EncodeToString(xxtea.Encrypt(timeData, XXTEA_KEY))
+	object.VersionId = ConvertRawVersionToS3Version(iversion)
+
+	helper.Logger.Println(20, "tidb client GetObject():", bucketName, objectName, version, iversion, object.VersionId, object.NullVersion, object.DeleteMarker)
+
 	return
 }
 
+func ConvertRawVersionToS3Version(rawVersion uint64) string {
+	return hex.EncodeToString(xxtea.Encrypt([]byte(strconv.FormatUint(rawVersion, 10)), XXTEA_KEY))
+}
+
+func ConvertS3VersionToRawVersion(s3Version string) (string, error) {
+	versionEncryped, err := hex.DecodeString(s3Version)
+	if err != nil {
+		helper.Logger.Printf(2, "Err in DecodeString()", s3Version)
+		return "", ErrInternalError
+	}
+
+	return string(xxtea.Decrypt(versionEncryped, XXTEA_KEY)), nil
+}
+
 func (t *TidbClient) GetAllObject(bucketName, objectName, version string) (object []*Object, err error) {
 	sqltext := "select version from objects where bucketname=? and name=?;"
-	var versions []string
+	var versions []uint64
 	rows, err := t.Client.Query(sqltext, bucketName, objectName)
 	if err != nil {
 		return
 	}
 	defer rows.Close()
 	for rows.Next() {
-		var sversion string
-		err = rows.Scan(&sversion)
+		var iversion uint64
+		err = rows.Scan(&iversion)
 		if err != nil {
 			return
 		}
-		versions = append(versions, sversion)
+		versions = append(versions, iversion)
 	}
 	for _, v := range versions {
 		var obj *Object
-		obj, err = t.GetObject(bucketName, objectName, v)
+		obj, err = t.GetObject(bucketName, objectName, ConvertRawVersionToS3Version(v))
 		if err != nil {
 			return
 		}
@@ -123,8 +145,12 @@ func (t *TidbClient) UpdateObjectAttrs(object *Object) error {
 	return err
 }
 
-func (t *TidbClient) UpdateAppendObject(o *Object) (err error) {
-	sql, args := o.GetAppendSql()
+func (t *TidbClient) UpdateAppendObject(o *Object, versionId string) (err error) {
+	rawVersionId, err := ConvertS3VersionToRawVersion(versionId)
+	if err != nil {
+		return err
+	}
+	sql, args := o.GetAppendSql(rawVersionId)
 	_, err = t.Client.Exec(sql, args...)
 	return err
 }
@@ -143,7 +169,9 @@ func (t *TidbClient) PutObject(object *Object, tx interface{}) (err error) {
 		}()
 	}
 	sqlTx, _ = tx.(*sql.Tx)
-	sql, args := object.GetCreateSql()
+
+	sql, args, iversion := object.GetCreateSql()
+	object.VersionId = ConvertRawVersionToS3Version(iversion)
 	_, err = sqlTx.Exec(sql, args...)
 	if object.Parts != nil {
 		v := math.MaxUint64 - uint64(object.LastModifiedTime.UnixNano())
@@ -178,6 +206,7 @@ func (t *TidbClient) DeleteObject(object *Object, tx interface{}) (err error) {
 	version := strconv.FormatUint(v, 10)
 	sqltext := "delete from objects where name=? and bucketname=? and version=?;"
 	_, err = sqlTx.Exec(sqltext, object.Name, object.BucketName, version)
+	helper.Logger.Println(20, sqltext, object.Name, object.BucketName, version, v)
 	if err != nil {
 		return err
 	}