closeBitrotWriters(writers)
		if err != nil {
			t.Fatalf("Test %d: failed to create random test data: %v", i, err)
		}

		readers := make([]io.ReaderAt, len(disks))
		for i, disk := range disks {
			shardFilesize := erasure.ShardFileSize(test.size)
			readers[i] = newBitrotReader(disk, nil, "testbucket", "testobject", shardFilesize, test.algorithm, bitrotWriterSum(writers[i]), erasure.ShardSize())
		}

// RLockedFile represents a read locked file, implements a special
// closer which only closes the associated *os.File when the ref count.
// has reached zero, i.e when all the readers have given up their locks.
type RLockedFile struct {
	*LockedFile
	mutex sync.Mutex
	refs  int // Holds read lock refs.
}

// IsClosed - Check if the rlocked file is already closed.

			readers[index] = newBitrotReader(disk, metaArr[index].Data, bucket, partPath, tillOffset,
				checksumInfo.Algorithm, checksumInfo.Hash, erasure.ShardSize())

			// Prefer local disks
			prefer[index] = disk.Hostname() == ""
		}

		written, err := erasure.Decode(ctx, writer, readers, partOffset, partLength, partSize, prefer)

	if algo == HighwayHash256S {
		return newStreamingBitrotReader(disk, data, bucket, filePath, tillOffset, algo, shardSize)
	}
	return newWholeBitrotReader(disk, bucket, filePath, algo, tillOffset, sum)
}

// Close all the readers.
func closeBitrotReaders(rs []io.ReaderAt) {
	for _, r := range rs {
		if r != nil {

		// not a storage error.
		return nil
	}
	readers := make([]*metacacheReader, len(disks))
	defer func() {
		for _, r := range readers {
			r.Close()
		}
	}()
	for i := range disks {
		r, w := io.Pipe()
		// Make sure we close the pipe so blocked writes doesn't stay around.
		defer r.CloseWithError(context.Canceled)

		readers[i] = newMetacacheReader(r)
		d := disks[i]

}

// Heal reads from readers, reconstruct shards and writes the data to the writers.
func (e Erasure) Heal(ctx context.Context, writers []io.Writer, readers []io.ReaderAt, totalLength int64, prefer []bool) (derr error) {
	if len(writers) != e.parityBlocks+e.dataBlocks {
		return errInvalidArgument
	}

	reader := newParallelReader(readers, e, 0, totalLength)
	if len(readers) == len(prefer) {

			continue
		}
		t.Run(fmt.Sprintf("%d-locks", locks), func(t *testing.T) {
			// Number of readers per lock...
			for _, readers := range []int{1, 10, 100} {
				if locks > 1000 && readers > 1 {
					continue
				}
				if testing.Short() && readers > 10 {
					continue
				}
				t.Run(fmt.Sprintf("%d-read", readers), func(t *testing.T) {
					l := newLocker()
					for i := 0; i < locks; i++ {
						var tmp [16]byte

			newSetCounts[ss] = struct{}{}
		}
	}

	setCounts = []uint64{}
	for setCount := range newSetCounts {
		setCounts = append(setCounts, setCount)
	}

	// Not necessarily needed but it ensures to the readers
	// eyes that we prefer a sorted setCount slice for the
	// subsequent function to figure out the right common
	// divisor, it avoids loops.
	sort.Slice(setCounts, func(i, j int) bool {

	return *resp, nil
}

// where we keep old *Readers
var readMsgpReaderPool = sync.Pool{New: func() interface{} { return &msgp.Reader{} }}

// mspNewReader returns a *Reader that reads from the provided reader.
// The reader will be buffered.
// Return with readMsgpReaderPoolPut when done.
func msgpNewReader(r io.Reader) *msgp.Reader {
	p := readMsgpReaderPool.Get().(*msgp.Reader)
	if p.R == nil {

	"github.com/minio/minio/internal/s3select/sql"
)

// Reader implements reading records from parquet input.
type Reader struct {
	io.Closer
	r *parquetgo.FileReader
}

// NewParquetReader creates a Reader2 from a io.ReadSeekCloser.
func NewParquetReader(rsc io.ReadSeekCloser, _ *ReaderArgs) (r *Reader, err error) {
	fr, err := parquetgo.NewFileReader(rsc)
	if err != nil {