[Disk Manager]: Snapshot: Check chunk contents for zeroes

cloud/disk_manager/internal/pkg/dataplane/common/chunk.go

@@ -1,11 +1,14 @@
 package common
 
 import (
+	"bytes"
 	"hash/crc32"
 )
 
 ////////////////////////////////////////////////////////////////////////////////
 
+var zeroes = make([]byte, 1024*1024)
+
 type Chunk struct {
 	ID          string
 	Index       uint32
@@ -18,3 +21,20 @@ type Chunk struct {
 func (chunk Chunk) Checksum() uint32 {
 	return crc32.ChecksumIEEE(chunk.Data)
 }
+
+func (chunk Chunk) CheckDataIsAllZeroes() bool {
+	for i := 0; i < len(chunk.Data); i += len(zeroes) {
+		endOffset := i + len(zeroes)
+
+		if endOffset > len(chunk.Data) {
+			endOffset = len(chunk.Data)
+		}
+
+		dataToCheck := chunk.Data[i:endOffset]
+
+		if !bytes.Equal(dataToCheck, zeroes[:len(dataToCheck)]) {
+			return false
+		}
+	}
+	return true
+}

cloud/disk_manager/internal/pkg/dataplane/common/chunk_test.go

@@ -0,0 +1,22 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+////////////////////////////////////////////////////////////////////////////////
+
+func TestChunkIsZero(t *testing.T) {
+	chunk := Chunk{
+		Data: []byte{1, 2, 3, 4, 5},
+	}
+	require.False(t, chunk.CheckDataIsAllZeroes())
+
+	chunk = Chunk{Data: []byte{0, 0, 0}}
+	require.True(t, chunk.CheckDataIsAllZeroes())
+
+	chunk = Chunk{Data: make([]byte, 1024*1024*5)}
+	require.True(t, chunk.CheckDataIsAllZeroes())
+}

cloud/disk_manager/internal/pkg/dataplane/common/transfer.go

@@ -342,6 +342,10 @@ func (t Transferer) goRead(
 						return
 					}
 
+					if !chunk.Zero {
+						chunk.Zero = chunk.CheckDataIsAllZeroes()
+					}
+
 					select {
 					case chunks <- chunk:
 					case <-ctx.Done():

cloud/disk_manager/internal/pkg/dataplane/common/ya.make

@@ -7,6 +7,7 @@ SRCS(
 
 GO_TEST_SRCS(
     transfer_test.go
+    chunk_test.go
 )
 
 END()

cloud/disk_manager/internal/pkg/dataplane/snapshot/storage/storage_ydb_test.go

@@ -1262,6 +1262,40 @@ func TestCompressionMetricsCollection(t *testing.T) {
 	}
 }
 
+func TestSnapshotReadWriteAllZeroedChunk(t *testing.T) {
+	for _, testCase := range testCases() {
+		t.Run(
+			testCase.name, func(t *testing.T) {
+				f := createFixture(t)
+				defer f.teardown()
+
+				chunk := common.Chunk{
+					Index: 0,
+					Data:  make([]byte, 1024*1024*4),
+				}
+				chunkID, err := f.storage.WriteChunk(
+					f.ctx,
+					"",
+					"test",
+					chunk,
+					testCase.useS3,
+				)
+				require.NoError(t, err)
+
+				readChunk := common.Chunk{
+					ID:         chunkID,
+					Data:       make([]byte, len(chunk.Data)),
+					StoredInS3: testCase.useS3,
+				}
+				err = f.storage.ReadChunk(f.ctx, &readChunk)
+				require.NoError(t, err)
+				require.Equal(t, chunk.Data, readChunk.Data)
+				require.True(t, readChunk.Zero)
+			},
+		)
+	}
+}
+
 func expectHistogramCalledOnce(registry *mocks.RegistryMock, name string, compressionTags map[string]string) *mock.Call {
 	return registry.GetHistogram(
 		name,