directly stream from bucket

core/pkg/storage/azurestorage.go

@@ -257,6 +257,27 @@ func (b *AzureStorage) Read(name string) ([]byte, error) {
 	return downloadedData.Bytes(), nil
 }
 
+// ReadStream returns a streaming reader for the specified blob path.
+func (b *AzureStorage) ReadStream(path string) (io.ReadCloser, error) {
+	path = trimLeading(path)
+	ctx := context.Background()
+
+	log.Debugf("AzureStorage::ReadStream::HTTPS(%s)", path)
+
+	downloadResponse, err := b.containerClient.NewBlobClient(path).DownloadStream(ctx, nil)
+	if err != nil {
+		if b.IsObjNotFoundErr(err) {
+			return nil, DoesNotExistError
+		}
+		return nil, fmt.Errorf("AzureStorage: ReadStream: failed to download blob %q: %w", path, err)
+	}
+
+	retryReader := downloadResponse.NewRetryReader(ctx, &azblob.RetryReaderOptions{
+		MaxRetries: int32(b.config.ReaderConfig.MaxRetryRequests),
+	})
+	return retryReader, nil
+}
+
 // ReadToLocalFile streams the specified blob at path to destPath on the local file system.
 func (b *AzureStorage) ReadToLocalFile(path, destPath string) error {
 	path = trimLeading(path)

core/pkg/storage/bucketstorage_test.go

@@ -107,6 +107,20 @@ func TestBucketStorage_ReadToLocalFile(t *testing.T) {
 	TestStorageReadToLocalFile(t, store)
 }
 
+func TestBucketStorage_ReadStream(t *testing.T) {
+	configPath := os.Getenv("TEST_BUCKET_CONFIG")
+	if configPath == "" {
+		t.Skip("skipping integration test, set environment variable TEST_BUCKET_CONFIG")
+	}
+	store, err := createStorage(configPath)
+	if err != nil {
+		t.Errorf("failed to create storage: %s", err.Error())
+		return
+	}
+
+	TestStorageReadStream(t, store)
+}
+
 // We should be able to call validate function with and without write and delete check without any errors
 func TestBucketStorage_Validate(t *testing.T) {
 	configPath := os.Getenv("TEST_BUCKET_CONFIG")

core/pkg/storage/clusterstorage.go

@@ -280,6 +280,18 @@ func (c *ClusterStorage) Read(path string) ([]byte, error) {
 	return jsonResp.Data, nil
 }
 
+// ReadStream returns a reader for the specified object path.
+//
+// Note: ClusterStorage does not currently expose a remote streaming endpoint, so this
+// implementation materializes the response via Read and wraps it as an io.ReadCloser.
+func (c *ClusterStorage) ReadStream(path string) (io.ReadCloser, error) {
+	data, err := c.Read(path)
+	if err != nil {
+		return nil, err
+	}
+	return io.NopCloser(bytes.NewReader(data)), nil
+}
+
 // ReadToLocalFile downloads the specified object at path to destPath on the local file system.
 //
 // Note: ClusterStorage does not currently expose a streaming download endpoint, so this implementation

core/pkg/storage/clusterstorage_test.go

@@ -3,6 +3,7 @@ package storage
 import (
 	"crypto/tls"
 	"encoding/json"
+	"io"
 	"net/http/httptest"
 	"net/http"
 	"net/url"
@@ -119,3 +120,54 @@ func TestClusterStorage_ReadToLocalFile(t *testing.T) {
 		t.Fatalf("destination file contents mismatch: got %q want %q", string(data), string(expected))
 	}
 }
+
+func TestClusterStorage_ReadStream(t *testing.T) {
+	expected := []byte("cluster-storage-stream-contents")
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/clusterStorage/read" {
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+
+		resp := Response[[]byte]{
+			Code: 0,
+			Data: expected,
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(resp)
+	}))
+	defer srv.Close()
+
+	u, err := url.Parse(srv.URL)
+	if err != nil {
+		t.Fatalf("parsing test server URL: %s", err)
+	}
+
+	port, err := strconv.Atoi(u.Port())
+	if err != nil {
+		t.Fatalf("parsing test server port: %s", err)
+	}
+
+	cs := &ClusterStorage{
+		client: &http.Client{},
+		host:   u.Hostname(),
+		port:   port,
+	}
+
+	r, err := cs.ReadStream("some/path")
+	if err != nil {
+		t.Fatalf("ReadStream failed: %s", err)
+	}
+	defer r.Close()
+
+	data, err := io.ReadAll(r)
+	if err != nil {
+		t.Fatalf("reading stream failed: %s", err)
+	}
+
+	if string(data) != string(expected) {
+		t.Fatalf("stream contents mismatch: got %q want %q", string(data), string(expected))
+	}
+}

core/pkg/storage/filestorage.go

@@ -118,6 +118,21 @@ func (fs *FileStorage) Read(path string) ([]byte, error) {
 	return b, nil
 }
 
+// ReadStream returns a streaming reader for the specified file path.
+func (fs *FileStorage) ReadStream(path string) (io.ReadCloser, error) {
+	f := filepath.Join(fs.baseDir, path)
+
+	file, err := os.Open(f)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return nil, DoesNotExistError
+		}
+		return nil, fmt.Errorf("opening %s: %w", f, err)
+	}
+
+	return file, nil
+}
+
 // ReadToLocalFile streams the specified file at path to destPath on the local file system.
 //
 // For FileStorage, this is implemented as a local file copy.

core/pkg/storage/filestorage_test.go

@@ -41,3 +41,9 @@ func TestFileStorage_ReadToLocalFile(t *testing.T) {
 	store := NewFileStorage(storeBaseDir)
 	TestStorageReadToLocalFile(t, store)
 }
+
+func TestFileStorage_ReadStream(t *testing.T) {
+	storeBaseDir := t.TempDir()
+	store := NewFileStorage(storeBaseDir)
+	TestStorageReadStream(t, store)
+}

core/pkg/storage/gcsstorage.go

@@ -140,6 +140,7 @@ func (gs *GCSStorage) Read(name string) ([]byte, error) {
 		}
 		return nil, err
 	}
+	defer reader.Close()
 
 	data, err := io.ReadAll(reader)
 	if err != nil {
@@ -149,6 +150,26 @@ func (gs *GCSStorage) Read(name string) ([]byte, error) {
 	return data, nil
 }
 
+// ReadStream returns a streaming reader for the specified object path.
+func (gs *GCSStorage) ReadStream(path string) (io.ReadCloser, error) {
+	path = trimLeading(path)
+	log.Debugf("GCSStorage::ReadStream::HTTPS(%s)", path)
+
+	ctx := context.Background()
+	reader, err := gs.bucket.Object(path).NewReader(ctx)
+	if err != nil {
+		if err == gcs.ErrObjectNotExist {
+			return nil, DoesNotExistError
+		}
+		if e, ok := err.(*googleapi.Error); ok && e.Code == http.StatusNotFound {
+			return nil, DoesNotExistError
+		}
+		return nil, err
+	}
+
+	return reader, nil
+}
+
 // ReadToLocalFile streams the specified object at path to destPath on the local file system.
 func (gs *GCSStorage) ReadToLocalFile(path, destPath string) error {
 	path = trimLeading(path)

core/pkg/storage/memorystorage.go

@@ -1,7 +1,9 @@
 package storage
 
 import (
+	"bytes"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"sync"
@@ -73,6 +75,21 @@ func (ms *MemoryStorage) Read(path string) ([]byte, error) {
 	return nil, DoesNotExistError
 }
 
+// ReadStream returns a streaming reader for the specified in-memory object.
+func (ms *MemoryStorage) ReadStream(path string) (io.ReadCloser, error) {
+	ms.lock.Lock()
+	defer ms.lock.Unlock()
+
+	path = filepath.Clean(path)
+
+	if file, ok := ms.directPaths[path]; ok {
+		data := append([]byte(nil), file.Contents...)
+		return io.NopCloser(bytes.NewReader(data)), nil
+	}
+
+	return nil, DoesNotExistError
+}
+
 // ReadToLocalFile writes the specified object at path to destPath on the local file system.
 func (ms *MemoryStorage) ReadToLocalFile(path, destPath string) error {
 	ms.lock.Lock()

core/pkg/storage/memorystorage_test.go

@@ -385,3 +385,8 @@ func TestMemoryStorage_ReadToLocalFile(t *testing.T) {
 	store := NewMemoryStorage()
 	TestStorageReadToLocalFile(t, store)
 }
+
+func TestMemoryStorage_ReadStream(t *testing.T) {
+	store := NewMemoryStorage()
+	TestStorageReadStream(t, store)
+}

core/pkg/storage/prefixedbucketstorage.go

@@ -6,6 +6,7 @@ package storage
 
 import (
 	"fmt"
+	"io"
 	"strings"
 
 	"github.com/pkg/errors"
@@ -81,6 +82,11 @@ func (pbs *PrefixedBucketStorage) Read(name string) ([]byte, error) {
 	return pbs.storage.Read(conditionalPrefix(pbs.prefix, name))
 }
 
+// ReadStream returns a streaming reader for the given object name.
+func (pbs *PrefixedBucketStorage) ReadStream(name string) (io.ReadCloser, error) {
+	return pbs.storage.ReadStream(conditionalPrefix(pbs.prefix, name))
+}
+
 // ReadToLocalFile streams the specified object at path to destPath on the local file system.
 func (pbs *PrefixedBucketStorage) ReadToLocalFile(path, destPath string) error {
 	return pbs.storage.ReadToLocalFile(conditionalPrefix(pbs.prefix, path), destPath)

core/pkg/storage/prefixedbucketstorage_test.go

@@ -12,3 +12,13 @@ func TestPrefixedBucketStorage_ReadToLocalFile(t *testing.T) {
 	TestStorageReadToLocalFile(t, store)
 }
 
+func TestPrefixedBucketStorage_ReadStream(t *testing.T) {
+	base := NewMemoryStorage()
+	store, err := NewPrefixedBucketStorage(base, "myprefix")
+	if err != nil {
+		t.Fatalf("failed to create prefixed storage: %s", err)
+	}
+
+	TestStorageReadStream(t, store)
+}
+

core/pkg/storage/s3storage.go

@@ -62,6 +62,8 @@ var defaultS3Config = S3Config{
 	PartSize: 1024 * 1024 * 64, // 64MB.
 }
 
+const defaultS3ReadChunkSize int64 = 8 * 1024 * 1024
+
 // Config stores the configuration for s3 bucket.
 type S3Config struct {
 	Bucket             string            `yaml:"bucket"`
@@ -299,6 +301,32 @@ func (s3 *S3Storage) Read(name string) ([]byte, error) {
 
 }
 
+// ReadStream returns an io.ReadCloser that incrementally streams an object from S3
+// by issuing byte-range requests under the hood.
+func (s3 *S3Storage) ReadStream(path string) (io.ReadCloser, error) {
+	path = trimLeading(path)
+
+	log.Debugf("S3Storage::ReadStream::%s(%s)", s3.protocol(), path)
+	ctx := context.Background()
+
+	sse, err := s3.getServerSideEncryption(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	objInfo, err := s3.client.StatObject(ctx, s3.name, path, minio.StatObjectOptions{ServerSideEncryption: sse})
+	if err != nil {
+		if s3.isDoesNotExist(err) || s3.isObjNotFound(err) {
+			return nil, DoesNotExistError
+		}
+		return nil, errors.Wrap(err, "StatObject from S3 failed")
+	}
+
+	return newS3ChunkReader(objInfo.Size, defaultS3ReadChunkSize, func(off, length int64) ([]byte, error) {
+		return s3.getRange(ctx, path, off, length)
+	}), nil
+}
+
 // ReadToLocalFile streams the specified object at path to destPath on the local file system.
 func (s3 *S3Storage) ReadToLocalFile(path, destPath string) error {
 	path = trimLeading(path)
@@ -580,15 +608,10 @@ func (s3 *S3Storage) getRange(ctx context.Context, name string, off, length int6
 	}
 
 	opts := &minio.GetObjectOptions{ServerSideEncryption: sse}
-	if length != -1 {
-		if err := opts.SetRange(off, off+length-1); err != nil {
-			return nil, err
-		}
-	} else if off > 0 {
-		if err := opts.SetRange(off, 0); err != nil {
-			return nil, err
-		}
+	if err := setGetObjectRange(opts, off, length); err != nil {
+		return nil, err
 	}
+
 	r, err := s3.client.GetObject(ctx, s3.name, name, *opts)
 	if err != nil {
 		if s3.isObjNotFound(err) {
@@ -596,21 +619,132 @@ func (s3 *S3Storage) getRange(ctx context.Context, name string, off, length int6
 		}
 		return nil, err
 	}
-	defer r.Close()
-
 	// NotFoundObject error is revealed only after first Read. This does the initial GetRequest. Prefetch this here
 	// for convenience.
 	if _, err := r.Read(nil); err != nil {
 		if s3.isObjNotFound(err) {
+			_ = r.Close()
 			return nil, DoesNotExistError
 		}
 
+		_ = r.Close()
 		return nil, errors.Wrap(err, "Read from S3 failed")
 	}
 
+	defer r.Close()
 	return io.ReadAll(r)
 }
 
+func setGetObjectRange(opts *minio.GetObjectOptions, off, length int64) error {
+	if off < 0 {
+		return errors.New("range offset must be >= 0")
+	}
+	if length < -1 || length == 0 {
+		return errors.New("range length must be -1 or > 0")
+	}
+
+	if length > 0 {
+		return opts.SetRange(off, off+length-1)
+	}
+	if off > 0 {
+		return opts.SetRange(off, 0)
+	}
+
+	return nil
+}
+
+type s3ChunkReader struct {
+	size      int64
+	chunkSize int64
+	pos       int64
+	chunkOff  int64
+	chunk     []byte
+	closed    bool
+	fetch     func(off, length int64) ([]byte, error)
+}
+
+func newS3ChunkReader(size, chunkSize int64, fetch func(off, length int64) ([]byte, error)) io.ReadCloser {
+	if chunkSize <= 0 {
+		chunkSize = defaultS3ReadChunkSize
+	}
+	return &s3ChunkReader{
+		size:      size,
+		chunkSize: chunkSize,
+		chunkOff:  -1,
+		fetch:     fetch,
+	}
+}
+
+func (r *s3ChunkReader) Read(p []byte) (int, error) {
+	if r.closed {
+		return 0, errors.New("s3 chunk reader is closed")
+	}
+	if len(p) == 0 {
+		return 0, nil
+	}
+	if r.pos >= r.size {
+		return 0, io.EOF
+	}
+
+	n := 0
+	for n < len(p) && r.pos < r.size {
+		if !r.hasChunkForPos(r.pos) {
+			if err := r.loadChunk(); err != nil {
+				if err == io.EOF && n > 0 {
+					return n, nil
+				}
+				return n, err
+			}
+		}
+
+		chunkIdx := int(r.pos - r.chunkOff)
+		wrote := copy(p[n:], r.chunk[chunkIdx:])
+		n += wrote
+		r.pos += int64(wrote)
+	}
+
+	if r.pos >= r.size {
+		return n, io.EOF
+	}
+	return n, nil
+}
+
+func (r *s3ChunkReader) Close() error {
+	r.closed = true
+	r.chunk = nil
+	return nil
+}
+
+func (r *s3ChunkReader) hasChunkForPos(pos int64) bool {
+	if len(r.chunk) == 0 || r.chunkOff < 0 {
+		return false
+	}
+	return pos >= r.chunkOff && pos < r.chunkOff+int64(len(r.chunk))
+}
+
+func (r *s3ChunkReader) loadChunk() error {
+	if r.pos >= r.size {
+		return io.EOF
+	}
+
+	length := r.chunkSize
+	if remaining := r.size - r.pos; remaining < length {
+		length = remaining
+	}
+
+	chunk, err := r.fetch(r.pos, length)
+	if err != nil {
+		return err
+	}
+	if len(chunk) == 0 {
+		return io.EOF
+	}
+
+	r.chunk = chunk
+	r.chunkOff = r.pos
+	return nil
+}
+
 // awsAuth retrieves credentials from the aws-sdk-go.
 type awsAuth struct {
 	Region string

core/pkg/storage/s3storage_test.go

@@ -1,7 +1,13 @@
 package storage
 
 import (
+	"bytes"
+	"errors"
+	"io"
+	"reflect"
 	"testing"
+
+	"github.com/minio/minio-go/v7"
 )
 
 // TestS3Storage_protocol tests the protocol() method returns correct values based on insecure flag
@@ -35,3 +41,123 @@ func TestS3Storage_protocol(t *testing.T) {
 		})
 	}
 }
+
+func TestSetGetObjectRange(t *testing.T) {
+	tests := []struct {
+		name      string
+		off       int64
+		length    int64
+		expectErr bool
+	}{
+		{
+			name:      "full object range",
+			off:       0,
+			length:    -1,
+			expectErr: false,
+		},
+		{
+			name:      "offset to EOF range",
+			off:       100,
+			length:    -1,
+			expectErr: false,
+		},
+		{
+			name:      "bounded range",
+			off:       128,
+			length:    4096,
+			expectErr: false,
+		},
+		{
+			name:      "negative offset rejected",
+			off:       -1,
+			length:    -1,
+			expectErr: true,
+		},
+		{
+			name:      "zero length rejected",
+			off:       0,
+			length:    0,
+			expectErr: true,
+		},
+		{
+			name:      "invalid negative length rejected",
+			off:       0,
+			length:    -2,
+			expectErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			opts := &minio.GetObjectOptions{}
+			err := setGetObjectRange(opts, tt.off, tt.length)
+			if tt.expectErr && err == nil {
+				t.Fatalf("expected error, got nil")
+			}
+			if !tt.expectErr && err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+		})
+	}
+}
+
+func TestS3ChunkReader_ReadUsesRanges(t *testing.T) {
+	data := []byte("abcdefghijklmnopqrstuvwxyz")
+	var calls [][2]int64
+
+	reader := newS3ChunkReader(int64(len(data)), 8, func(off, length int64) ([]byte, error) {
+		calls = append(calls, [2]int64{off, length})
+		end := off + length
+		if end > int64(len(data)) {
+			end = int64(len(data))
+		}
+		return data[off:end], nil
+	})
+	defer reader.Close()
+
+	got, err := io.ReadAll(reader)
+	if err != nil {
+		t.Fatalf("reading chunked reader failed: %v", err)
+	}
+	if !bytes.Equal(got, data) {
+		t.Fatalf("data mismatch: got=%q want=%q", string(got), string(data))
+	}
+
+	wantCalls := [][2]int64{
+		{0, 8},
+		{8, 8},
+		{16, 8},
+		{24, 2},
+	}
+	if !reflect.DeepEqual(calls, wantCalls) {
+		t.Fatalf("range calls mismatch: got=%v want=%v", calls, wantCalls)
+	}
+}
+
+func TestS3ChunkReader_Close(t *testing.T) {
+	reader := newS3ChunkReader(10, 4, func(off, length int64) ([]byte, error) {
+		return []byte("xxxx"), nil
+	})
+	if err := reader.Close(); err != nil {
+		t.Fatalf("close failed: %v", err)
+	}
+
+	p := make([]byte, 4)
+	_, err := reader.Read(p)
+	if err == nil {
+		t.Fatal("expected read error after close")
+	}
+}
+
+func TestS3ChunkReader_PropagatesFetchError(t *testing.T) {
+	reader := newS3ChunkReader(10, 4, func(off, length int64) ([]byte, error) {
+		return nil, errors.New("fetch failed")
+	})
+	defer reader.Close()
+
+	p := make([]byte, 4)
+	_, err := reader.Read(p)
+	if err == nil || err.Error() != "fetch failed" {
+		t.Fatalf("expected fetch error, got: %v", err)
+	}
+}

core/pkg/storage/storage.go

@@ -1,6 +1,7 @@
 package storage
 
 import (
+	"io"
 	"os"
 	"strings"
 	"time"
@@ -38,6 +39,10 @@ type Storage interface {
 	// read the contents.
 	Read(path string) ([]byte, error)
 
+	// ReadStream returns an io.ReadCloser for the specified path. Implementations should
+	// stream incrementally when possible to avoid loading entire objects into memory.
+	ReadStream(path string) (io.ReadCloser, error)
+
 	// ReadToLocalFile writes the specified file at path to destPath on the local file system.
 	// Implementations may stream data to minimize RAM usage, but some backends may still buffer
 	// data in memory depending on their capabilities. It is up to the caller to clean up the

core/pkg/storage/test.go

@@ -2,6 +2,7 @@ package storage
 
 import (
 	"fmt"
+	"io"
 	"os"
 	"path"
 	"path/filepath"
@@ -499,3 +500,76 @@ func TestStorageReadToLocalFile(t *testing.T, store Storage) {
 		})
 	}
 }
+
+func TestStorageReadStream(t *testing.T, store Storage) {
+	testName := "read_stream"
+
+	fileNames := []string{
+		"/file0.json",
+	}
+
+	err := createFiles(fileNames, testName, store)
+	if err != nil {
+		t.Fatalf("failed to create files: %s", err)
+	}
+
+	defer func() {
+		err = cleanupFiles(fileNames, testName, store)
+		if err != nil {
+			t.Fatalf("failed to clean up files: %s", err)
+		}
+	}()
+
+	testCases := map[string]struct {
+		path      string
+		expectErr bool
+	}{
+		"file exists": {
+			path:      path.Join(testpath, testName, "file0.json"),
+			expectErr: false,
+		},
+		"file does not exist": {
+			path:      path.Join(testpath, testName, "file1.json"),
+			expectErr: true,
+		},
+		"dir does not exist": {
+			path:      path.Join(testpath, testName, "dir0/file.json"),
+			expectErr: true,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			r, err := store.ReadStream(tc.path)
+			if tc.expectErr {
+				if err == nil {
+					if r != nil {
+						_ = r.Close()
+					}
+					t.Fatalf("expected error was not thrown")
+				}
+				return
+			}
+
+			if err != nil {
+				t.Fatalf("unexpected error: %s", err.Error())
+			}
+			defer r.Close()
+
+			b, err := io.ReadAll(r)
+			if err != nil {
+				t.Fatalf("reading stream: %s", err)
+			}
+
+			var content testFileContent
+			err = json.Unmarshal(b, &content)
+			if err != nil {
+				t.Fatalf("could not unmarshal file content: %s", err)
+			}
+
+			if content != tfc {
+				t.Fatalf("file content did not match written value")
+			}
+		})
+	}
+}