[AzDatalake] Fix Path Renames with SAS

sdk/storage/azdatalake/directory/client.go

@@ -8,6 +8,7 @@ package directory
 
 import (
 	"context"
+	"errors"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
@@ -254,24 +255,45 @@ func (d *Client) GetProperties(ctx context.Context, options *GetPropertiesOption
 	return newResp, err
 }
 
-func (d *Client) renamePathInURL(newName string) (string, string, string) {
-	endpoint := d.DFSURL()
-	separator := "/"
-	// Find the index of the last occurrence of the separator
-	lastIndex := strings.LastIndex(endpoint, separator)
-	// Split the string based on the last occurrence of the separator
-	firstPart := endpoint[:lastIndex] // From the beginning of the string to the last occurrence of the separator
-	newBlobURL, newPathURL := shared.GetURLs(runtime.JoinPaths(firstPart, newName))
-	oldURL, _ := url.Parse(d.DFSURL())
-	return oldURL.Path, newPathURL, newBlobURL
-}
-
 // Rename renames a directory. The original directory will no longer exist and the client will be stale.
-func (d *Client) Rename(ctx context.Context, newName string, options *RenameOptions) (RenameResponse, error) {
-	oldURL, newPathURL, newBlobURL := d.renamePathInURL(newName)
-	lac, mac, smac, createOpts := path.FormatRenameOptions(options, oldURL)
+func (d *Client) Rename(ctx context.Context, destinationPath string, options *RenameOptions) (RenameResponse, error) {
 	var newBlobClient *blockblob.Client
-	var err error
+	destinationPath = strings.Trim(strings.TrimSpace(destinationPath), "/")
+	if len(destinationPath) == 0 {
+		return RenameResponse{}, errors.New("destination path must not be empty")
+	}
+	urlParts, err := sas.ParseURL(d.DFSURL())
+	if err != nil {
+		return RenameResponse{}, err
+	}
+
+	oldPath, _ := url.Parse(d.DFSURL())
+	srcParts := strings.Split(d.DFSURL(), "?")
+	newSrcPath := oldPath.Path
+	newSrcQuery := ""
+	if len(srcParts) == 2 {
+		newSrcQuery = srcParts[1]
+	}
+	if len(newSrcQuery) > 0 {
+		newSrcPath = newSrcPath + "?" + newSrcQuery
+	}
+
+	destParts := strings.Split(destinationPath, "?")
+	newDestPath := destParts[0]
+	newDestQuery := ""
+	if len(destParts) == 2 {
+		newDestQuery = destParts[1]
+	}
+
+	urlParts.PathName = newDestPath
+	newPathURL := urlParts.String()
+	// replace the query part if it is present in destination path
+	if len(newDestQuery) > 0 {
+		newPathURL = strings.Split(newPathURL, "?")[0] + "?" + newDestQuery
+	}
+	newBlobURL, _ := shared.GetURLs(newPathURL)
+	lac, mac, smac, createOpts := path.FormatRenameOptions(options, newSrcPath)
+
 	if d.identityCredential() != nil {
 		newBlobClient, err = blockblob.NewClient(newBlobURL, *d.identityCredential(), nil)
 	} else if d.sharedKey() != nil {
@@ -280,6 +302,7 @@ func (d *Client) Rename(ctx context.Context, newName string, options *RenameOpti
 	} else {
 		newBlobClient, err = blockblob.NewClientWithNoCredential(newBlobURL, nil)
 	}
+
 	if err != nil {
 		return RenameResponse{}, exported.ConvertToDFSError(err)
 	}

sdk/storage/azdatalake/directory/client_test.go

@@ -2166,6 +2166,47 @@ func (s *RecordedTestSuite) TestDirSetHTTPHeadersIfETagMatchFalse() {
 	testcommon.ValidateErrorCode(_require, err, datalakeerror.ConditionNotMet)
 }
 
+func (s *UnrecordedTestSuite) TestDirectoryRenameUsingSAS() {
+	_require := require.New(s.T())
+	testName := s.T().Name()
+
+	cred, err := testcommon.GetGenericSharedKeyCredential(testcommon.TestAccountDatalake)
+	_require.NoError(err)
+
+	filesystemName := testcommon.GenerateFileSystemName(testName)
+	fsClient, err := testcommon.GetFileSystemClient(filesystemName, s.T(), testcommon.TestAccountDatalake, nil)
+	_require.NoError(err)
+	defer testcommon.DeleteFileSystem(context.Background(), _require, fsClient)
+
+	_, err = fsClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	perms := sas.DirectoryPermissions{Read: true, Create: true, Write: true, Move: true, Delete: true, List: true}
+	sasQueryParams, err := sas.DatalakeSignatureValues{
+		Protocol:       sas.ProtocolHTTPS,                    // Users MUST use HTTPS (not HTTP)
+		ExpiryTime:     time.Now().UTC().Add(48 * time.Hour), // 48-hours before expiration
+		FileSystemName: filesystemName,
+		Permissions:    perms.String(),
+	}.SignWithSharedKey(cred)
+	_require.NoError(err)
+
+	sasToken := sasQueryParams.Encode()
+
+	srcDirClient, err := directory.NewClientWithNoCredential(fsClient.DFSURL()+"/dir1?"+sasToken, nil)
+	_require.NoError(err)
+
+	_, err = srcDirClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	destPathWithSAS := "dir2?" + sasToken
+	_, err = srcDirClient.Rename(context.Background(), destPathWithSAS, nil)
+	_require.NoError(err)
+
+	_, err = srcDirClient.GetProperties(context.Background(), nil)
+	_require.Error(err)
+	testcommon.ValidateErrorCode(_require, err, datalakeerror.PathNotFound)
+}
+
 func (s *RecordedTestSuite) TestDirRenameNoOptions() {
 	_require := require.New(s.T())
 	testName := s.T().Name()

sdk/storage/azdatalake/file/client.go

@@ -238,24 +238,45 @@ func (f *Client) GetProperties(ctx context.Context, options *GetPropertiesOption
 	return newResp, err
 }
 
-func (f *Client) renamePathInURL(newName string) (string, string, string) {
-	endpoint := f.DFSURL()
-	separator := "/"
-	// Find the index of the last occurrence of the separator
-	lastIndex := strings.LastIndex(endpoint, separator)
-	// Split the string based on the last occurrence of the separator
-	firstPart := endpoint[:lastIndex] // From the beginning of the string to the last occurrence of the separator
-	newBlobURL, newPathURL := shared.GetURLs(runtime.JoinPaths(firstPart, newName))
-	oldURL, _ := url.Parse(f.DFSURL())
-	return oldURL.Path, newPathURL, newBlobURL
-}
-
 // Rename renames a file. The original file will no longer exist and the client will be stale.
-func (f *Client) Rename(ctx context.Context, newName string, options *RenameOptions) (RenameResponse, error) {
-	oldPathWithoutURL, newPathURL, newBlobURL := f.renamePathInURL(newName)
-	lac, mac, smac, createOpts := path.FormatRenameOptions(options, oldPathWithoutURL)
+func (f *Client) Rename(ctx context.Context, destinationPath string, options *RenameOptions) (RenameResponse, error) {
 	var newBlobClient *blockblob.Client
-	var err error
+	destinationPath = strings.Trim(strings.TrimSpace(destinationPath), "/")
+	if len(destinationPath) == 0 {
+		return RenameResponse{}, errors.New("destination path must not be empty")
+	}
+	urlParts, err := sas.ParseURL(f.DFSURL())
+	if err != nil {
+		return RenameResponse{}, err
+	}
+
+	oldPath, _ := url.Parse(f.DFSURL())
+	srcParts := strings.Split(f.DFSURL(), "?")
+	newSrcPath := oldPath.Path
+	newSrcQuery := ""
+	if len(srcParts) == 2 {
+		newSrcQuery = srcParts[1]
+	}
+	if len(newSrcQuery) > 0 {
+		newSrcPath = newSrcPath + "?" + newSrcQuery
+	}
+
+	destParts := strings.Split(destinationPath, "?")
+	newDestPath := destParts[0]
+	newDestQuery := ""
+	if len(destParts) == 2 {
+		newDestQuery = destParts[1]
+	}
+
+	urlParts.PathName = newDestPath
+	newPathURL := urlParts.String()
+	// replace the query part if it is present in destination path
+	if len(newDestQuery) > 0 {
+		newPathURL = strings.Split(newPathURL, "?")[0] + "?" + newDestQuery
+	}
+	newBlobURL, _ := shared.GetURLs(newPathURL)
+	lac, mac, smac, createOpts := path.FormatRenameOptions(options, newSrcPath)
+
 	if f.identityCredential() != nil {
 		newBlobClient, err = blockblob.NewClient(newBlobURL, *f.identityCredential(), nil)
 	} else if f.sharedKey() != nil {
@@ -264,15 +285,18 @@ func (f *Client) Rename(ctx context.Context, newName string, options *RenameOpti
 	} else {
 		newBlobClient, err = blockblob.NewClientWithNoCredential(newBlobURL, nil)
 	}
+
 	if err != nil {
 		return RenameResponse{}, exported.ConvertToDFSError(err)
 	}
 	newFileClient := (*Client)(base.NewPathClient(newPathURL, newBlobURL, newBlobClient, f.generatedFileClientWithDFS().InternalClient().WithClientName(shared.FileClient), f.sharedKey(), f.identityCredential(), f.getClientOptions()))
 	resp, err := newFileClient.generatedFileClientWithDFS().Create(ctx, createOpts, nil, lac, mac, smac, nil)
+
 	//return RenameResponse{
 	//	Response:      resp,
 	//	NewFileClient: newFileClient,
 	//}, exported.ConvertToDFSError(err)
+
 	return path.FormatRenameResponse(&resp), exported.ConvertToDFSError(err)
 }
 

sdk/storage/azdatalake/file/client_test.go

@@ -1973,6 +1973,47 @@ func (s *RecordedTestSuite) TestFileSetHTTPHeadersIfETagMatchFalse() {
 	testcommon.ValidateErrorCode(_require, err, datalakeerror.ConditionNotMet)
 }
 
+func (s *UnrecordedTestSuite) TestFileRenameUsingSAS() {
+	_require := require.New(s.T())
+	testName := s.T().Name()
+
+	cred, err := testcommon.GetGenericSharedKeyCredential(testcommon.TestAccountDatalake)
+	_require.NoError(err)
+
+	filesystemName := testcommon.GenerateFileSystemName(testName)
+	fsClient, err := testcommon.GetFileSystemClient(filesystemName, s.T(), testcommon.TestAccountDatalake, nil)
+	_require.NoError(err)
+	defer testcommon.DeleteFileSystem(context.Background(), _require, fsClient)
+
+	_, err = fsClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	perms := sas.FilePermissions{Read: true, Create: true, Write: true, Move: true, Delete: true, List: true}
+	sasQueryParams, err := sas.DatalakeSignatureValues{
+		Protocol:       sas.ProtocolHTTPS,                    // Users MUST use HTTPS (not HTTP)
+		ExpiryTime:     time.Now().UTC().Add(48 * time.Hour), // 48-hours before expiration
+		FileSystemName: filesystemName,
+		Permissions:    perms.String(),
+	}.SignWithSharedKey(cred)
+	_require.NoError(err)
+
+	sasToken := sasQueryParams.Encode()
+
+	srcFileClient, err := file.NewClientWithNoCredential(fsClient.DFSURL()+"/file1?"+sasToken, nil)
+	_require.NoError(err)
+
+	_, err = srcFileClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	destPathWithSAS := "file2?" + sasToken
+	_, err = srcFileClient.Rename(context.Background(), destPathWithSAS, nil)
+	_require.NoError(err)
+
+	_, err = srcFileClient.GetProperties(context.Background(), nil)
+	_require.Error(err)
+	testcommon.ValidateErrorCode(_require, err, datalakeerror.PathNotFound)
+}
+
 func (s *RecordedTestSuite) TestRenameNoOptions() {
 	_require := require.New(s.T())
 	testName := s.T().Name()