Structured Message: DataLake Append (#43275)

* datalake append * null fix * fixes
Azure · Apr 17, 2024 · 69ff822 · 69ff822
1 parent d3a625d
commit 69ff822
Show file tree

Hide file tree

Showing 11 changed files with 94 additions and 32 deletions.
diff --git a/sdk/storage/Azure.Storage.Common/tests/Shared/TransferValidationTestBase.cs b/sdk/storage/Azure.Storage.Common/tests/Shared/TransferValidationTestBase.cs
@@ -8,6 +8,7 @@
 using System.Security.Cryptography;
 using System.Threading.Tasks;
 using Azure.Core;
+using Azure.Core.Diagnostics;
 using Azure.Core.Pipeline;
 using Azure.Core.TestFramework;
 using Azure.Storage.Shared;
@@ -232,7 +233,7 @@ void AssertChecksum(RequestHeaders headers, string headerName)
             };
         }
 
-#if BlobSDK
+#if BlobSDK || DataLakeSDK
         internal static Action<Request> GetRequestStructuredMessageAssertion(
             StructuredMessage.Flags flags,
             Func<Request, bool> isStructuredMessageExpected = default,
@@ -315,7 +316,7 @@ void AssertChecksum(ResponseHeaders headers, string headerName)
             };
         }
 
-#if BlobSDK
+#if BlobSDK || DataLakeSDK
         internal static Action<Response> GetResponseStructuredMessageAssertion(
             StructuredMessage.Flags flags,
             Func<Response, bool> isStructuredMessageExpected = default)
@@ -422,7 +423,7 @@ public virtual async Task UploadPartitionSuccessfulHashComputation(StorageChecks
             };
 
             // make pipeline assertion for checking checksum was present on upload
-#if BlobSDK
+#if BlobSDK || DataLakeSDK
             var assertion = algorithm.ResolveAuto() == StorageChecksumAlgorithm.StorageCrc64
                 ? GetRequestStructuredMessageAssertion(StructuredMessage.Flags.StorageCrc64, null, dataLength)
                 : GetRequestChecksumHeaderAssertion(algorithm);
@@ -531,9 +532,9 @@ public virtual async Task UploadPartitionTamperedStreamThrows(StorageChecksumAlg
                 // Act
                 streamTamperPolicy.TransformRequestBody = true;
                 AsyncTestDelegate operation = async () => await UploadPartitionAsync(client, stream, validationOptions);
-
+                using var listener = AzureEventSourceListener.CreateConsoleLogger();
                 // Assert
-#if BlobSDK
+#if BlobSDK || DataLakeSDK
                 AssertWriteChecksumMismatch(operation, algorithm,
                     expectStructuredMessage: algorithm.ResolveAuto() == StorageChecksumAlgorithm.StorageCrc64);
 #else
@@ -553,7 +554,7 @@ public virtual async Task UploadPartitionUsesDefaultClientValidationOptions(
             var data = GetRandomBuffer(dataLength);
 
             // make pipeline assertion for checking checksum was present on upload
-#if BlobSDK
+#if BlobSDK || DataLakeSDK
             var assertion = clientAlgorithm.ResolveAuto() == StorageChecksumAlgorithm.StorageCrc64
                 ? GetRequestStructuredMessageAssertion(StructuredMessage.Flags.StorageCrc64, null, dataLength)
                 : GetRequestChecksumHeaderAssertion(clientAlgorithm);
@@ -599,7 +600,7 @@ public virtual async Task UploadPartitionOverwritesDefaultClientValidationOption
             };
 
             // make pipeline assertion for checking checksum was present on upload
-#if BlobSDK
+#if BlobSDK || DataLakeSDK
             var assertion = overrideAlgorithm.ResolveAuto() == StorageChecksumAlgorithm.StorageCrc64
                 ? GetRequestStructuredMessageAssertion(StructuredMessage.Flags.StorageCrc64, null, dataLength)
                 : GetRequestChecksumHeaderAssertion(overrideAlgorithm);
@@ -1021,7 +1022,7 @@ public virtual async Task ParallelUploadOneShotSuccessfulHashComputation(Storage
             };
 
             // make pipeline assertion for checking checksum was present on upload
-#if BlobSDK
+#if BlobSDK || DataLakeSDK
             var assertion = algorithm.ResolveAuto() == StorageChecksumAlgorithm.StorageCrc64
                 ? GetRequestStructuredMessageAssertion(StructuredMessage.Flags.StorageCrc64, ParallelUploadIsChecksumExpected, dataLength)
                 : GetRequestChecksumHeaderAssertion(algorithm, isChecksumExpected: ParallelUploadIsChecksumExpected);
@@ -1115,7 +1116,7 @@ public virtual async Task ParallelUploadUsesDefaultClientValidationOptions(
                 };
 
             // make pipeline assertion for checking checksum was present on upload
-#if BlobSDK
+#if BlobSDK || DataLakeSDK
             var assertion = clientAlgorithm.ResolveAuto() == StorageChecksumAlgorithm.StorageCrc64 && !split
                 ? GetRequestStructuredMessageAssertion(StructuredMessage.Flags.StorageCrc64, ParallelUploadIsChecksumExpected, dataLength)
                 : GetRequestChecksumHeaderAssertion(clientAlgorithm, isChecksumExpected: ParallelUploadIsChecksumExpected);
@@ -1173,7 +1174,7 @@ public virtual async Task ParallelUploadOverwritesDefaultClientValidationOptions
               };
 
             // make pipeline assertion for checking checksum was present on upload
-#if BlobSDK
+#if BlobSDK || DataLakeSDK
             var assertion = overrideAlgorithm.ResolveAuto() == StorageChecksumAlgorithm.StorageCrc64 && !split
                 ? GetRequestStructuredMessageAssertion(StructuredMessage.Flags.StorageCrc64, ParallelUploadIsChecksumExpected, dataLength)
                 : GetRequestChecksumHeaderAssertion(overrideAlgorithm, isChecksumExpected: ParallelUploadIsChecksumExpected);
@@ -1751,7 +1752,7 @@ public virtual async Task DownloadSuccessfulHashVerification(StorageChecksumAlgo
             Assert.IsTrue(dest.ToArray().SequenceEqual(data));
         }
 
-#if BlobSDK
+#if BlobSDK || DataLakeSDK
         [TestCase(StorageChecksumAlgorithm.StorageCrc64, Constants.StructuredMessage.MaxDownloadCrcWithHeader, false, false)]
         [TestCase(StorageChecksumAlgorithm.StorageCrc64, Constants.StructuredMessage.MaxDownloadCrcWithHeader-1, false, false)]
         [TestCase(StorageChecksumAlgorithm.StorageCrc64, Constants.StructuredMessage.MaxDownloadCrcWithHeader+1, true, false)]

diff --git a/sdk/storage/Azure.Storage.Files.DataLake/assets.json b/sdk/storage/Azure.Storage.Files.DataLake/assets.json
@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "net",
   "TagPrefix": "net/storage/Azure.Storage.Files.DataLake",
-  "Tag": "net/storage/Azure.Storage.Files.DataLake_2eae678081"
+  "Tag": "net/storage/Azure.Storage.Files.DataLake_ade992acfe"
 }
diff --git a/sdk/storage/Azure.Storage.Files.DataLake/src/Azure.Storage.Files.DataLake.csproj b/sdk/storage/Azure.Storage.Files.DataLake/src/Azure.Storage.Files.DataLake.csproj
@@ -81,6 +81,8 @@
     <Compile Include="$(AzureStorageSharedSources)StorageServerTimeoutPolicy.cs" LinkBase="Shared" />
     <Compile Include="$(AzureStorageSharedSources)StorageVersionExtensions.cs" LinkBase="Shared" />
     <Compile Include="$(AzureStorageSharedSources)StructuredMessage.cs" LinkBase="Shared" />
+    <Compile Include="$(AzureStorageSharedSources)StructuredMessageDecodingStream.cs" LinkBase="Shared" />
+    <Compile Include="$(AzureStorageSharedSources)StructuredMessageEncodingStream.cs" LinkBase="Shared" />
     <Compile Include="$(AzureStorageSharedSources)TransferValidationOptionsExtensions.cs" LinkBase="Shared" />
     <Compile Include="$(AzureStorageSharedSources)UriExtensions.cs" LinkBase="Shared" />
     <Compile Include="$(AzureStorageSharedSources)UriQueryParamsCollection.cs" LinkBase="Shared" />

diff --git a/sdk/storage/Azure.Storage.Files.DataLake/src/DataLakeFileClient.cs b/sdk/storage/Azure.Storage.Files.DataLake/src/DataLakeFileClient.cs
@@ -16,6 +16,7 @@
 using Azure.Storage.Blobs.Models;
 using Azure.Storage.Files.DataLake.Models;
 using Azure.Storage.Sas;
+using Azure.Storage.Shared;
 using Metadata = System.Collections.Generic.IDictionary<string, string>;
 
 namespace Azure.Storage.Files.DataLake
@@ -2331,13 +2332,36 @@ internal virtual async Task<Response> AppendInternal(
             using (ClientConfiguration.Pipeline.BeginLoggingScope(nameof(DataLakeFileClient)))
             {
                 // compute hash BEFORE attaching progress handler
-                ContentHasher.GetHashResult hashResult = await ContentHasher.GetHashOrDefaultInternal(
-                    content,
-                    validationOptions,
-                    async,
-                    cancellationToken).ConfigureAwait(false);
+                ContentHasher.GetHashResult hashResult = null;
+                long contentLength = (content?.Length - content?.Position) ?? 0;
+                long? structuredContentLength = default;
+                string structuredBodyType = null;
+                if (content != null &&
+                    validationOptions != null &&
+                    validationOptions.ChecksumAlgorithm.ResolveAuto() == StorageChecksumAlgorithm.StorageCrc64 &&
+                    validationOptions.PrecalculatedChecksum.IsEmpty)
+                {
+                    // report progress in terms of caller bytes, not encoded bytes
+                    structuredContentLength = contentLength;
+                    structuredBodyType = Constants.StructuredMessage.CrcStructuredMessage;
+                    content = content.WithNoDispose().WithProgress(progressHandler);
+                    content = new StructuredMessageEncodingStream(
+                        content,
+                        Constants.StructuredMessage.DefaultSegmentContentLength,
+                        StructuredMessage.Flags.StorageCrc64);
+                    contentLength = content.Length - content.Position;
+                }
+                else
+                {
+                    // compute hash BEFORE attaching progress handler
+                    hashResult = await ContentHasher.GetHashOrDefaultInternal(
+                        content,
+                        validationOptions,
+                        async,
+                        cancellationToken).ConfigureAwait(false);
+                    content = content?.WithNoDispose().WithProgress(progressHandler);
+                }
 
-                content = content?.WithNoDispose().WithProgress(progressHandler);
                 ClientConfiguration.Pipeline.LogMethodEnter(
                     nameof(DataLakeFileClient),
                     message:
@@ -2372,6 +2396,8 @@ internal virtual async Task<Response> AppendInternal(
                             encryptionKey: ClientConfiguration.CustomerProvidedKey?.EncryptionKey,
                             encryptionKeySha256: ClientConfiguration.CustomerProvidedKey?.EncryptionKeyHash,
                             encryptionAlgorithm: ClientConfiguration.CustomerProvidedKey?.EncryptionAlgorithm == null ? null : EncryptionAlgorithmTypeInternal.AES256,
+                            structuredBodyType: structuredBodyType,
+                            structuredContentLength: structuredContentLength,
                             leaseId: leaseId,
                             leaseAction: leaseAction,
                             leaseDuration: leaseDurationLong,
@@ -2391,6 +2417,8 @@ internal virtual async Task<Response> AppendInternal(
                             encryptionKey: ClientConfiguration.CustomerProvidedKey?.EncryptionKey,
                             encryptionKeySha256: ClientConfiguration.CustomerProvidedKey?.EncryptionKeyHash,
                             encryptionAlgorithm: ClientConfiguration.CustomerProvidedKey?.EncryptionAlgorithm == null ? null : EncryptionAlgorithmTypeInternal.AES256,
+                            structuredBodyType: structuredBodyType,
+                            structuredContentLength: structuredContentLength,
                             leaseId: leaseId,
                             leaseAction: leaseAction,
                             leaseDuration: leaseDurationLong,

diff --git a/sdk/storage/Azure.Storage.Files.DataLake/src/Generated/FileSystemRestClient.cs b/sdk/storage/Azure.Storage.Files.DataLake/src/Generated/FileSystemRestClient.cs
diff --git a/sdk/storage/Azure.Storage.Files.DataLake/src/Generated/PathAppendDataHeaders.cs b/sdk/storage/Azure.Storage.Files.DataLake/src/Generated/PathAppendDataHeaders.cs