func (s *S3File) openWrite() {
	pr, pw := io.Pipe()
	uploader := s3manager.NewUploaderWithClient(s.client, s.uploaderOptions...)
	s.lock.Lock()
	s.pipeReader = pr
	s.pipeWriter = pw
	s.writeOpened = true
	s.uploader = uploader
	s.lock.Unlock()
	uploadParams := &s3manager.UploadInput{
		Bucket: aws.String(s.BucketName),
		Key:    aws.String(s.Key),
		ACL:    aws.String(s.ACL),
		Body:   s.pipeReader,
	}
	go func(uploader *s3manager.Uploader, params *s3manager.UploadInput, done chan error) {
		defer close(done)
		_, err := uploader.UploadWithContext(s.ctx, params)
		if err != nil {
			s.lock.Lock()
			s.err = err
			s.lock.Unlock()
			if s.writeOpened {
				s.pipeWriter.CloseWithError(err)
			}
		}
		done <- err
	}(s.uploader, uploadParams, s.writeDone)
}
openWrite creates an S3 uploader that consumes the Reader end of an io.Pipe.
Calling Close signals write completion.
package s3
import (
	"context"
	"errors"
	"fmt"
	"io"
	"sync"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3iface"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
	"github.com/xitongsys/parquet-go/source"
)
type S3File struct {
	ctx    context.Context
	client s3iface.S3API
	offset int64
	whence int
	writeOpened     bool
	writeDone       chan error
	pipeReader      *io.PipeReader
	pipeWriter      *io.PipeWriter
	uploader        *s3manager.Uploader
	uploaderOptions []func(*s3manager.Uploader)
	readOpened bool
	fileSize   int64
	downloader *s3manager.Downloader
	lock       sync.RWMutex
	err        error
	BucketName string
	Key        string
	ACL        string
}
const (
	rangeHeader       = "bytes=%d-%d"
	rangeHeaderSuffix = "bytes=%d"
)
var (
	errWhence        = errors.New("Seek: invalid whence")
	errInvalidOffset = errors.New("Seek: invalid offset")
	errFailedUpload  = errors.New("Write: failed upload")
	activeS3Session  *session.Session
	sessLock         sync.Mutex
)
func SetActiveSession(sess *session.Session) {
	sessLock.Lock()
	activeS3Session = sess
	sessLock.Unlock()
}
func NewS3FileWriter(
	ctx context.Context,
	bucket string,
	key string,
	acl string,
	uploaderOptions []func(*s3manager.Uploader),
	cfgs ...*aws.Config,
) (source.ParquetFile, error) {
	if activeS3Session == nil {
		sessLock.Lock()
		if activeS3Session == nil {
			activeS3Session = session.Must(session.NewSession())
		}
		sessLock.Unlock()
	}
	return NewS3FileWriterWithClient(
		ctx, s3.New(activeS3Session, cfgs...), bucket, key, acl, uploaderOptions)
}
func NewS3FileWriterWithClient(
	ctx context.Context,
	s3Client s3iface.S3API,
	bucket string,
	key string,
	acl string,
	uploaderOptions []func(*s3manager.Uploader),
) (source.ParquetFile, error) {
	file := &S3File{
		ctx:             ctx,
		client:          s3Client,
		writeDone:       make(chan error),
		uploaderOptions: uploaderOptions,
		BucketName:      bucket,
		Key:             key,
		ACL:             acl,
	}
	return file.Create(key)
}
func NewS3FileReader(ctx context.Context, bucket string, key string, cfgs ...*aws.Config) (source.ParquetFile, error) {
	if activeS3Session == nil {
		sessLock.Lock()
		if activeS3Session == nil {
			activeS3Session = session.Must(session.NewSession())
		}
		sessLock.Unlock()
	}
	return NewS3FileReaderWithClient(ctx, s3.New(activeS3Session, cfgs...), bucket, key)
}
func NewS3FileReaderWithClient(ctx context.Context, s3Client s3iface.S3API, bucket string, key string) (source.ParquetFile, error) {
	s3Downloader := s3manager.NewDownloaderWithClient(s3Client)
	file := &S3File{
		ctx:        ctx,
		client:     s3Client,
		downloader: s3Downloader,
		BucketName: bucket,
		Key:        key,
	}
	return file.Open(key)
}
func (s *S3File) Seek(offset int64, whence int) (int64, error) {
	if whence < io.SeekStart || whence > io.SeekEnd {
		return 0, errWhence
	}
	if s.fileSize > 0 {
		switch whence {
		case io.SeekStart:
			if offset < 0 || offset > s.fileSize {
				return 0, errInvalidOffset
			}
		case io.SeekCurrent:
			offset += s.offset
			if offset < 0 || offset > s.fileSize {
				return 0, errInvalidOffset
			}
		case io.SeekEnd:
			if offset > -1 || -offset > s.fileSize {
				return 0, errInvalidOffset
			}
		}
	}
	s.offset = offset
	s.whence = whence
	return s.offset, nil
}
func (s *S3File) Read(p []byte) (n int, err error) {
	if s.fileSize > 0 && s.offset >= s.fileSize {
		return 0, io.EOF
	}
	numBytes := len(p)
	getObjRange := s.getBytesRange(numBytes)
	getObj := &s3.GetObjectInput{
		Bucket: aws.String(s.BucketName),
		Key:    aws.String(s.Key),
	}
	if len(getObjRange) > 0 {
		getObj.Range = aws.String(getObjRange)
	}
	wab := aws.NewWriteAtBuffer(p)
	bytesDownloaded, err := s.downloader.DownloadWithContext(s.ctx, wab, getObj)
	if err != nil {
		return 0, err
	}
	s.offset += bytesDownloaded
	if buf := wab.Bytes(); len(buf) > numBytes {
		copy(p, buf)
		bytesDownloaded = int64(len(p))
	}
	return int(bytesDownloaded), err
}
func (s *S3File) Write(p []byte) (n int, err error) {
	s.lock.RLock()
	writeOpened := s.writeOpened
	s.lock.RUnlock()
	if !writeOpened {
		s.openWrite()
	}
	s.lock.RLock()
	writeError := s.err
	s.lock.RUnlock()
	if writeError != nil {
		return 0, writeError
	}
	bytesWritten, writeError := s.pipeWriter.Write(p)
	if writeError != nil {
		s.lock.Lock()
		s.err = writeError
		s.lock.Unlock()
		s.pipeWriter.CloseWithError(err)
		return 0, writeError
	}
	return bytesWritten, nil
}
func (s *S3File) Close() error {
	var err error
	if s.pipeWriter != nil {
		if err = s.pipeWriter.Close(); err != nil {
			return err
		}
	}
	if s.writeDone != nil {
		err = <-s.writeDone
	}
	return err
}
func (s *S3File) Open(name string) (source.ParquetFile, error) {
	s.lock.RLock()
	readOpened := s.readOpened
	s.lock.RUnlock()
	if !readOpened {
		if err := s.openRead(); err != nil {
			return nil, err
		}
	}
	if len(name) == 0 {
		name = s.Key
	}
	pf := &S3File{
		ctx:        s.ctx,
		client:     s.client,
		downloader: s.downloader,
		BucketName: s.BucketName,
		Key:        name,
		readOpened: s.readOpened,
		fileSize:   s.fileSize,
		offset:     0,
	}
	return pf, nil
}
func (s *S3File) Create(key string) (source.ParquetFile, error) {
	pf := &S3File{
		ctx:             s.ctx,
		client:          s.client,
		uploaderOptions: s.uploaderOptions,
		BucketName:      s.BucketName,
		ACL:             s.ACL,
		Key:             key,
		writeDone:       make(chan error),
	}
	pf.openWrite()
	return pf, nil
}