fix: recover import reader from premature EOF (#49648)

## Summary

Fixes #49645

Bulk import can hit `unexpected EOF` when a remote object stream stays
idle during function execution and the object-store reader returns a
premature EOF. This PR makes sequential import readers recover by
reopening the object reader at the last successful offset when EOF
happens before the known object size.

## Changes

- Add reopen-and-retry support to `RetryableReader` for retryable read
errors and premature EOF.
- Use `ChunkManager.Size` as a fallback object-size source for readers
whose `Size()` is lazy or unreliable before reading.
- Return unexpected EOF, not clean EOF, when premature EOF retries are
exhausted.
- Enable the reopen path for JSON, CSV, and Numpy import readers.
- Keep Parquet unchanged because Arrow uses `ReadAt`/`Seek`, which this
sequential wrapper does not cover.

## Test Plan

- [x] `make static-check`
- [x] `git diff --check`
- [x] `go test ./internal/util/importutilv2/common
./internal/util/importutilv2/json ./internal/util/importutilv2/csv
./internal/util/importutilv2/numpy ./internal/util/importutilv2/parquet
-count=1 -ldflags="-r ${RPATH}"`

---------

Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>

internal/storage/gcp_native_object_storage.go

@@ -58,7 +58,11 @@ func (gcs *GcpNativeObjectStorage) GetObject(ctx context.Context, bucketName, ob
 	if offset == 0 && size == 0 {
 		reader, err = obj.NewReader(ctx)
 	} else {
-		reader, err = obj.NewRangeReader(ctx, offset, size)
+		length := size
+		if offset > 0 && size == 0 {
+			length = -1
+		}
+		reader, err = obj.NewRangeReader(ctx, offset, length)
 	}
 
 	if err != nil {

internal/storage/local_chunk_manager.go

@@ -80,6 +80,23 @@ func (lcm *LocalChunkManager) Reader(ctx context.Context, filePath string) (File
 	}, nil
 }
 
+func (lcm *LocalChunkManager) ReaderAtOffset(ctx context.Context, filePath string, offset int64) (FileReader, error) {
+	if offset < 0 {
+		return nil, io.EOF
+	}
+	reader, err := lcm.Reader(ctx, filePath)
+	if err != nil {
+		return nil, err
+	}
+	if offset > 0 {
+		if _, err = reader.Seek(offset, io.SeekStart); err != nil {
+			_ = reader.Close()
+			return nil, err
+		}
+	}
+	return reader, nil
+}
+
 // Write writes the data to local storage.
 func (lcm *LocalChunkManager) Write(ctx context.Context, filePath string, content []byte) error {
 	dir := path.Dir(filePath)

internal/storage/minio_object_storage.go

@@ -57,7 +57,11 @@ func newMinioObjectStorageWithConfig(ctx context.Context, c *objectstorage.Confi
 func (minioObjectStorage *MinioObjectStorage) GetObject(ctx context.Context, bucketName, objectName string, offset int64, size int64) (FileReader, error) {
 	opts := minio.GetObjectOptions{}
 	if offset > 0 {
-		err := opts.SetRange(offset, offset+size-1)
+		end := int64(0)
+		if size > 0 {
+			end = offset + size - 1
+		}
+		err := opts.SetRange(offset, end)
 		if err != nil {
 			log.Warn("failed to set range", zap.String("bucket", bucketName), zap.String("path", objectName), zap.Error(err))
 			return nil, mapObjectStorageError(objectName, err)

internal/storage/remote_chunk_manager.go

@@ -144,6 +144,19 @@ func (mcm *RemoteChunkManager) Reader(ctx context.Context, filePath string) (Fil
 	return reader, nil
 }
 
+func (mcm *RemoteChunkManager) ReaderAtOffset(ctx context.Context, filePath string, offset int64) (FileReader, error) {
+	if offset < 0 {
+		return nil, io.EOF
+	}
+
+	reader, err := mcm.getObject(ctx, mcm.bucketName, filePath, offset, int64(0))
+	if err != nil {
+		log.Warn("failed to get object", zap.String("bucket", mcm.bucketName), zap.String("path", filePath), zap.Int64("offset", offset), zap.Error(err))
+		return nil, err
+	}
+	return reader, nil
+}
+
 func (mcm *RemoteChunkManager) Size(ctx context.Context, filePath string) (int64, error) {
 	var objectInfo int64
 	var err error

internal/util/importutilv2/common/mock_reader.go

@@ -56,6 +56,48 @@ func CustomMockReader(reader io.Reader) storage.FileReader {
 	}
 }
 
+type prematureEOFReader struct {
+	*strings.Reader
+	limit        int
+	read         int
+	reportedSize int64
+}
+
+// NewPrematureEOFReader creates a test reader that returns EOF before the reported object size is reached.
+func NewPrematureEOFReader(content string, limit int) storage.FileReader {
+	return NewPrematureEOFReaderWithSize(content, limit, int64(len(content)))
+}
+
+// NewPrematureEOFReaderWithSize creates a test reader that reports a custom object size.
+func NewPrematureEOFReaderWithSize(content string, limit int, reportedSize int64) storage.FileReader {
+	reader := strings.NewReader(content)
+	return &prematureEOFReader{
+		Reader:       reader,
+		limit:        limit,
+		reportedSize: reportedSize,
+	}
+}
+
+func (r *prematureEOFReader) Read(p []byte) (int, error) {
+	if r.read >= r.limit {
+		return 0, io.EOF
+	}
+	if remaining := r.limit - r.read; len(p) > remaining {
+		p = p[:remaining]
+	}
+	n, err := r.Reader.Read(p)
+	r.read += n
+	return n, err
+}
+
+func (r *prematureEOFReader) Close() error {
+	return nil
+}
+
+func (r *prematureEOFReader) Size() (int64, error) {
+	return r.reportedSize, nil
+}
+
 func newErrorMockReader(content string, err error, errCount int) storage.FileReader {
 	reader := strings.NewReader(content)
 	return &mockFileReader{

internal/util/importutilv2/common/retryable_reader.go

@@ -37,35 +37,149 @@ type RetryableReader interface {
 	Read(p []byte) (n int, err error)
 }
 
+type (
+	ReopenReaderFunc func(context.Context, string, int64) (storage.FileReader, error)
+	ReaderSizeFunc   func(context.Context, string) (int64, error)
+)
+
+type offsetReader interface {
+	ReaderAtOffset(context.Context, string, int64) (storage.FileReader, error)
+}
+
+// NewChunkManagerReopenReaderFunc creates a reopen function that resumes reading at the given offset.
+func NewChunkManagerReopenReaderFunc(cm storage.ChunkManager) ReopenReaderFunc {
+	return func(ctx context.Context, path string, offset int64) (storage.FileReader, error) {
+		if reader, ok := cm.(offsetReader); ok {
+			return reader.ReaderAtOffset(ctx, path, offset)
+		}
+		reader, err := cm.Reader(ctx, path)
+		if err != nil {
+			return nil, err
+		}
+		if offset > 0 {
+			if _, err = reader.Seek(offset, io.SeekStart); err != nil {
+				_ = reader.Close()
+				return nil, err
+			}
+		}
+		return reader, nil
+	}
+}
+
 // retryableReader is the implementation of RetryableReader.
 type retryableReader struct {
 	storage.FileReader
 	ctx           context.Context
 	path          string
 	retryAttempts uint
+	reopen        ReopenReaderFunc
+	sizeFunc      ReaderSizeFunc
+	offset        int64
+	size          int64
 }
 
 // NewRetryableReader creates a new RetryableReader.
 func NewRetryableReader(ctx context.Context, path string, reader storage.FileReader) RetryableReader {
+	return newRetryableReader(ctx, path, reader, nil, nil)
+}
+
+func NewRetryableReaderWithReopen(ctx context.Context, path string, reader storage.FileReader, reopen ReopenReaderFunc, sizeFunc ReaderSizeFunc) RetryableReader {
+	return newRetryableReader(ctx, path, reader, reopen, sizeFunc)
+}
+
+func newRetryableReader(ctx context.Context, path string, reader storage.FileReader, reopen ReopenReaderFunc, sizeFunc ReaderSizeFunc) RetryableReader {
+	size := int64(-1)
+	if reader != nil {
+		var err error
+		size, err = reader.Size()
+		if err != nil {
+			size = -1
+		}
+	}
 	return &retryableReader{
 		FileReader:    reader,
 		ctx:           ctx,
 		path:          path,
 		retryAttempts: paramtable.Get().CommonCfg.StorageReadRetryAttempts.GetAsUint(),
+		reopen:        reopen,
+		sizeFunc:      sizeFunc,
+		size:          size,
 	}
 }
 
+func (r *retryableReader) objectSize() (int64, bool) {
+	if r.size > 0 {
+		return r.size, true
+	}
+	if r.sizeFunc == nil {
+		return r.size, r.size >= 0
+	}
+	size, err := r.sizeFunc(r.ctx, r.path)
+	if err != nil {
+		log.Ctx(r.ctx).Warn("retryable reader failed to get object size",
+			zap.String("path", r.path),
+			zap.Error(err),
+		)
+		return 0, false
+	}
+	r.size = size
+	r.sizeFunc = nil
+	return size, true
+}
+
+func (r *retryableReader) reopenAtOffset() error {
+	if r.reopen == nil {
+		return nil
+	}
+	if r.FileReader != nil {
+		_ = r.Close()
+		r.FileReader = nil
+	}
+	reader, err := r.reopen(r.ctx, r.path, r.offset)
+	if err != nil {
+		return storage.ToMilvusIoError(r.path, err)
+	}
+	r.FileReader = reader
+	return nil
+}
+
 // Read reads from the underlying FileReader and retries on errors.
 func (r *retryableReader) Read(p []byte) (int, error) {
 	var n int
 	var err error
 	err = retry.Handle(r.ctx, func() (bool, error) {
+		if r.FileReader == nil {
+			if reopenErr := r.reopenAtOffset(); reopenErr != nil {
+				return !merr.IsNonRetryableErr(reopenErr), reopenErr
+			}
+			if r.FileReader == nil {
+				err = storage.ToMilvusIoError(r.path, io.ErrClosedPipe)
+				return false, err
+			}
+		}
 		n, err = r.FileReader.Read(p)
+		if n > 0 {
+			r.offset += int64(n)
+			// Preserve io.Reader semantics: return buffered bytes first.
+			// Any accompanying error is handled by a following Read, which may reopen.
+			return false, nil
+		}
 		if err == nil {
 			return false, nil
 		}
-		// EOF is not an error - don't retry
 		if errors.Is(err, io.EOF) {
+			if size, ok := r.objectSize(); ok && r.offset < size && r.reopen != nil {
+				err = storage.ToMilvusIoError(r.path, io.ErrUnexpectedEOF)
+				log.Ctx(r.ctx).Warn("retryable reader got premature EOF",
+					zap.String("path", r.path),
+					zap.Int64("offset", r.offset),
+					zap.Int64("size", size),
+				)
+				if reopenErr := r.reopenAtOffset(); reopenErr != nil {
+					return !merr.IsNonRetryableErr(reopenErr), reopenErr
+				}
+				return true, err
+			}
 			return false, err
 		}
 		// Context canceled or deadline exceeded - don't retry
@@ -81,6 +195,9 @@ func (r *retryableReader) Read(p []byte) (int, error) {
 		if merr.IsNonRetryableErr(err) {
 			return false, err
 		}
+		if reopenErr := r.reopenAtOffset(); reopenErr != nil {
+			return !merr.IsNonRetryableErr(reopenErr), reopenErr
+		}
 		// Retry everything else (network errors, timeouts, 500s, etc.)
 		return true, err
 	}, retry.Attempts(r.retryAttempts))

internal/util/importutilv2/common/retryable_reader_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/minio/minio-go/v7"
 	"github.com/stretchr/testify/assert"
 
+	"github.com/milvus-io/milvus/internal/storage"
 	"github.com/milvus-io/milvus/pkg/v3/util/merr"
 	"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
 )
@@ -121,6 +122,196 @@ func (m *customMockReader) ReadAt(p []byte, off int64) (int, error) {
 	return 0, nil
 }
 
+type closeAwareReader struct {
+	storage.FileReader
+	closed         bool
+	readAfterClose int
+}
+
+func (r *closeAwareReader) Read(p []byte) (int, error) {
+	if r.closed {
+		r.readAfterClose++
+		return 0, errors.New("read after close")
+	}
+	return r.FileReader.Read(p)
+}
+
+func (r *closeAwareReader) Close() error {
+	r.closed = true
+	return r.FileReader.Close()
+}
+
+func newSizeFunc(size int64) ReaderSizeFunc {
+	return func(context.Context, string) (int64, error) {
+		return size, nil
+	}
+}
+
+func newReopenFunc(content string, openCount *int, offsets *[]int64) ReopenReaderFunc {
+	return func(_ context.Context, _ string, offset int64) (storage.FileReader, error) {
+		if openCount != nil {
+			*openCount = *openCount + 1
+		}
+		if offsets != nil {
+			*offsets = append(*offsets, offset)
+		}
+		if offset < 0 || offset > int64(len(content)) {
+			return nil, io.EOF
+		}
+		return NewMockReader(content[offset:]), nil
+	}
+}
+
+func TestRetryableReader_ReopenOnPrematureEOF(t *testing.T) {
+	ctx := context.Background()
+	path := "/test/path"
+	content := "hello world"
+	openCount := 0
+	offsets := make([]int64, 0, 1)
+	reopen := newReopenFunc(content, &openCount, &offsets)
+
+	reader := NewRetryableReaderWithReopen(ctx, path, NewPrematureEOFReader(content, 5), reopen, newSizeFunc(int64(len(content))))
+	buf := make([]byte, len(content))
+	n, err := io.ReadFull(reader, buf)
+
+	assert.NoError(t, err)
+	assert.Equal(t, len(content), n)
+	assert.Equal(t, content, string(buf))
+	assert.Equal(t, 1, openCount)
+	assert.Equal(t, []int64{5}, offsets)
+}
+
+func TestRetryableReader_FinalEOFIsNotRetried(t *testing.T) {
+	ctx := context.Background()
+	path := "/test/path"
+	content := "done"
+	openCount := 0
+	reopen := newReopenFunc(content, &openCount, nil)
+
+	reader := NewRetryableReaderWithReopen(ctx, path, NewMockReader(content), reopen, newSizeFunc(int64(len(content))))
+	buf := make([]byte, len(content))
+	n, err := io.ReadFull(reader, buf)
+	assert.NoError(t, err)
+	assert.Equal(t, len(content), n)
+
+	n, err = reader.Read(buf)
+	assert.ErrorIs(t, err, io.EOF)
+	assert.Equal(t, 0, n)
+	assert.Equal(t, 0, openCount)
+}
+
+func TestRetryableReader_ReopenOnRetryableError(t *testing.T) {
+	ctx := context.Background()
+	path := "/test/path"
+	content := "data after retry"
+	openCount := 0
+	reopen := newReopenFunc(content, &openCount, nil)
+
+	reader := NewRetryableReaderWithReopen(ctx, path, newErrorMockReader(content, errors.New("network timeout"), 1), reopen, newSizeFunc(int64(len(content))))
+	buf := make([]byte, len(content))
+	n, err := reader.Read(buf)
+
+	assert.NoError(t, err)
+	assert.Equal(t, len(content), n)
+	assert.Equal(t, content, string(buf))
+	assert.Equal(t, 1, openCount)
+}
+
+func TestRetryableReader_ExhaustedPrematureEOFReturnsUnexpectedEOF(t *testing.T) {
+	ctx := context.Background()
+	path := "/test/path"
+	content := "hello world"
+	reopen := func(_ context.Context, _ string, offset int64) (storage.FileReader, error) {
+		if offset < 0 || offset > int64(len(content)) {
+			return nil, io.EOF
+		}
+		return NewPrematureEOFReader(content[offset:], 0), nil
+	}
+
+	reader := &retryableReader{
+		FileReader:    NewPrematureEOFReader(content, 5),
+		ctx:           ctx,
+		path:          path,
+		retryAttempts: 2,
+		reopen:        reopen,
+		sizeFunc:      newSizeFunc(int64(len(content))),
+		size:          -1,
+	}
+	buf := make([]byte, len(content))
+	n, err := io.ReadFull(reader, buf)
+
+	assert.ErrorIs(t, err, merr.ErrIoUnexpectEOF)
+	assert.False(t, errors.Is(err, io.EOF))
+	assert.Equal(t, 5, n)
+}
+
+func TestRetryableReader_UsesSizeFuncForPrematureEOF(t *testing.T) {
+	ctx := context.Background()
+	path := "/test/path"
+	content := "hello world"
+	openCount := 0
+	reopen := newReopenFunc(content, &openCount, nil)
+
+	reader := NewRetryableReaderWithReopen(ctx, path, NewPrematureEOFReaderWithSize(content, 5, 0), reopen, newSizeFunc(int64(len(content))))
+	buf := make([]byte, len(content))
+	n, err := io.ReadFull(reader, buf)
+
+	assert.NoError(t, err)
+	assert.Equal(t, len(content), n)
+	assert.Equal(t, content, string(buf))
+	assert.Equal(t, 1, openCount)
+}
+
+func TestRetryableReader_UsesSizeFuncForImmediateEOF(t *testing.T) {
+	ctx := context.Background()
+	path := "/test/path"
+	content := "hello world"
+	openCount := 0
+	reopen := newReopenFunc(content, &openCount, nil)
+
+	reader := NewRetryableReaderWithReopen(ctx, path, NewPrematureEOFReaderWithSize(content, 0, 0), reopen, newSizeFunc(int64(len(content))))
+	buf := make([]byte, len(content))
+	n, err := io.ReadFull(reader, buf)
+
+	assert.NoError(t, err)
+	assert.Equal(t, len(content), n)
+	assert.Equal(t, content, string(buf))
+	assert.Equal(t, 1, openCount)
+}
+
+func TestRetryableReader_DoesNotReadClosedReaderAfterFailedReopen(t *testing.T) {
+	ctx := context.Background()
+	path := "/test/path"
+	content := "hello world"
+	initialReader := &closeAwareReader{FileReader: NewPrematureEOFReader(content, 0)}
+	openCount := 0
+	reopen := func(_ context.Context, _ string, offset int64) (storage.FileReader, error) {
+		openCount++
+		if openCount == 1 {
+			return nil, errors.New("temporary reopen failure")
+		}
+		return NewMockReader(content[offset:]), nil
+	}
+
+	reader := &retryableReader{
+		FileReader:    initialReader,
+		ctx:           ctx,
+		path:          path,
+		retryAttempts: 3,
+		reopen:        reopen,
+		sizeFunc:      newSizeFunc(int64(len(content))),
+		size:          -1,
+	}
+	buf := make([]byte, len(content))
+	n, err := io.ReadFull(reader, buf)
+
+	assert.NoError(t, err)
+	assert.Equal(t, len(content), n)
+	assert.Equal(t, content, string(buf))
+	assert.Equal(t, 2, openCount)
+	assert.Equal(t, 0, initialReader.readAfterClose)
+}
+
 func TestRetryableReader_DenylistRetry_NonRetryableErrors(t *testing.T) {
 	ctx := context.Background()
 	nonRetryableErrors := []error{

internal/util/importutilv2/csv/reader.go

@@ -54,7 +54,7 @@ func NewReader(ctx context.Context, cm storage.ChunkManager, schema *schemapb.Co
 	if err != nil {
 		return nil, merr.WrapErrImportFailed(fmt.Sprintf("read csv file failed, path=%s, err=%s", path, err.Error()))
 	}
-	retryableReader := common.NewRetryableReader(ctx, path, cmReader)
+	retryableReader := common.NewRetryableReaderWithReopen(ctx, path, cmReader, common.NewChunkManagerReopenReaderFunc(cm), cm.Size)
 	count, err := common.EstimateReadCountPerBatch(bufferSize, schema)
 	if err != nil {
 		return nil, err

internal/util/importutilv2/csv/reader_test.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	"math/rand"
 	"os"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/mock"
@@ -246,6 +247,7 @@ func (suite *ReaderSuite) TestError() {
 				return r, nil
 			}
 		})
+		cm.EXPECT().Size(mock.Anything, "dummy path").Return(int64(len(content)), nil).Maybe()
 
 		reader, err := NewReader(context.Background(), cm, schema, "dummy path", bufferSize, ',', "")
 		suite.Error(err)
@@ -346,6 +348,51 @@ func (suite *ReaderSuite) TestReadLoop() {
 	suite.Nil(data)
 }
 
+func (suite *ReaderSuite) TestReadRecoversFromPrematureEOF() {
+	schema := &schemapb.CollectionSchema{
+		Fields: []*schemapb.FieldSchema{
+			{
+				FieldID:      100,
+				Name:         "pk",
+				IsPrimaryKey: true,
+				DataType:     schemapb.DataType_Int64,
+			},
+			{
+				FieldID:  101,
+				Name:     "int32",
+				DataType: schemapb.DataType_Int32,
+			},
+		},
+	}
+	var contentBuilder strings.Builder
+	contentBuilder.WriteString("pk,int32\n")
+	for i := 1; i <= 100; i++ {
+		fmt.Fprintf(&contentBuilder, "%d,%d\n", i, i*10)
+	}
+	content := contentBuilder.String()
+	openCount := 0
+
+	cm := mocks.NewChunkManager(suite.T())
+	cm.EXPECT().Reader(mock.Anything, "dummy path").RunAndReturn(func(ctx context.Context, path string) (storage.FileReader, error) {
+		openCount++
+		if openCount == 1 {
+			return importcommon.NewPrematureEOFReader(content, 18), nil
+		}
+		return importcommon.NewMockReader(content), nil
+	})
+
+	reader, err := NewReader(context.Background(), cm, schema, "dummy path", 1024, ',', "")
+	suite.NoError(err)
+	defer reader.Close()
+
+	data, err := reader.Read()
+	suite.NoError(err)
+	suite.Equal(100, data.GetRowNum())
+	suite.Equal(int64(1), data.Data[100].GetRow(0))
+	suite.Equal(int32(1000), data.Data[101].GetRow(99))
+	suite.GreaterOrEqual(openCount, 2)
+}
+
 func TestCsvReader(t *testing.T) {
 	suite.Run(t, new(ReaderSuite))
 }

internal/util/importutilv2/json/reader.go

@@ -60,7 +60,7 @@ func newReader(ctx context.Context, cm storage.ChunkManager, schema *schemapb.Co
 	if err != nil {
 		return nil, merr.WrapErrImportFailed(fmt.Sprintf("read json file failed, path=%s, err=%s", path, err.Error()))
 	}
-	retryableReader := common.NewRetryableReader(ctx, path, r)
+	retryableReader := common.NewRetryableReaderWithReopen(ctx, path, r, common.NewChunkManagerReopenReaderFunc(cm), cm.Size)
 	count, err := common.EstimateReadCountPerBatch(bufferSize, schema)
 	if err != nil {
 		return nil, err

internal/util/importutilv2/json/reader_test.go

@@ -339,6 +339,7 @@ func (suite *ReaderSuite) TestDecodeError() {
 			r := importcommon.NewMockReader(jsonContent)
 			return r, nil
 		})
+		cm.EXPECT().Size(mock.Anything, "mockPath").Return(int64(len(jsonContent)), nil).Maybe()
 		var reader *reader
 		var err error
 		if isLinesFormat {
@@ -454,6 +455,39 @@ func (suite *ReaderSuite) TestReadCount() {
 	suite.Equal(20, data.GetRowNum())
 }
 
+func (suite *ReaderSuite) TestReadRecoversFromPrematureEOF() {
+	schema := &schemapb.CollectionSchema{
+		Fields: []*schemapb.FieldSchema{
+			{
+				FieldID:      100,
+				Name:         "pk",
+				IsPrimaryKey: true,
+				DataType:     schemapb.DataType_Int64,
+			},
+		},
+	}
+	jsonContent := `[{"pk":1},{"pk":2},{"pk":3}]`
+	openCount := 0
+
+	cm := mocks.NewChunkManager(suite.T())
+	cm.EXPECT().Reader(mock.Anything, "mockPath").RunAndReturn(func(ctx context.Context, path string) (storage.FileReader, error) {
+		openCount++
+		if openCount == 1 {
+			return importcommon.NewPrematureEOFReader(jsonContent, 10), nil
+		}
+		return importcommon.NewMockReader(jsonContent), nil
+	})
+
+	reader, err := NewReader(context.Background(), cm, schema, "mockPath", math.MaxInt)
+	suite.NoError(err)
+	defer reader.Close()
+
+	data, err := reader.Read()
+	suite.NoError(err)
+	suite.Equal(3, data.GetRowNum())
+	suite.GreaterOrEqual(openCount, 2)
+}
+
 func TestJsonReader(t *testing.T) {
 	suite.Run(t, new(ReaderSuite))
 }

internal/util/importutilv2/numpy/reader_test.go

@@ -447,6 +447,73 @@ func (suite *ReaderSuite) TestVector() {
 	}
 }
 
+func (suite *ReaderSuite) TestReadRecoversFromPrematureEOF() {
+	schema := &schemapb.CollectionSchema{
+		Fields: []*schemapb.FieldSchema{
+			{
+				FieldID:      100,
+				Name:         "pk",
+				IsPrimaryKey: true,
+				DataType:     schemapb.DataType_Int64,
+			},
+			{
+				FieldID:    101,
+				Name:       "varchar",
+				DataType:   schemapb.DataType_VarChar,
+				TypeParams: []*commonpb.KeyValuePair{{Key: common.MaxLengthKey, Value: "128"}},
+			},
+			{
+				FieldID:    102,
+				Name:       "vec",
+				DataType:   schemapb.DataType_FloatVector,
+				TypeParams: []*commonpb.KeyValuePair{{Key: common.DimKey, Value: fmt.Sprintf("%d", dim)}},
+			},
+		},
+	}
+	insertData, err := testutil.CreateInsertData(schema, 3)
+	suite.NoError(err)
+	files := map[int64]string{
+		100: "pk.npy",
+		101: "varchar.npy",
+		102: "vec.npy",
+	}
+	contents := make(map[string]string)
+	for _, field := range schema.GetFields() {
+		npyReader, err := createReader(insertData.Data[field.GetFieldID()], field.GetDataType())
+		suite.NoError(err)
+		contentBytes, err := io.ReadAll(npyReader)
+		suite.NoError(err)
+		contents[files[field.GetFieldID()]] = string(contentBytes)
+	}
+	openCount := make(map[string]int)
+
+	cm := mocks.NewChunkManager(suite.T())
+	for _, path := range files {
+		path := path
+		cm.EXPECT().Reader(mock.Anything, path).RunAndReturn(func(ctx context.Context, path string) (storage.FileReader, error) {
+			openCount[path]++
+			if openCount[path] == 1 {
+				return importcommon.NewPrematureEOFReader(contents[path], len(contents[path])-2), nil
+			}
+			return importcommon.NewMockReader(contents[path]), nil
+		})
+	}
+
+	reader, err := NewReader(context.Background(), cm, schema, lo.Values(files), math.MaxInt)
+	suite.NoError(err)
+	defer reader.Close()
+
+	data, err := reader.Read()
+	suite.NoError(err)
+	suite.Equal(3, data.GetRowNum())
+	for _, field := range schema.GetFields() {
+		for i := 0; i < data.GetRowNum(); i++ {
+			suite.Equal(insertData.Data[field.GetFieldID()].GetRow(i), data.Data[field.GetFieldID()].GetRow(i))
+		}
+		suite.GreaterOrEqual(openCount[files[field.GetFieldID()]], 2)
+	}
+}
+
 func TestNumpyCreateReaders(t *testing.T) {
 	ctx := context.Background()
 	cm := mocks.NewChunkManager(t)

internal/util/importutilv2/numpy/util.go

@@ -97,7 +97,7 @@ func CreateReaders(ctx context.Context, cm storage.ChunkManager, schema *schemap
 			return nil, merr.WrapErrImportFailed(
 				fmt.Sprintf("failed to read the file '%s', error: %s", path, err.Error()))
 		}
-		retryableReader := common.NewRetryableReader(ctx, path, reader)
+		retryableReader := common.NewRetryableReaderWithReopen(ctx, path, reader, common.NewChunkManagerReopenReaderFunc(cm), cm.Size)
 		readers[field.GetFieldID()] = retryableReader
 		readFields[field.GetName()] = field.GetFieldID()
 	}