func extractHeader(body io.Reader, myHeaders http.Header) error {
	for {
		headerTypeName, err := extractHeaderType(body)
		if err != nil {
			if err == io.EOF {
				break
			}
			return err
		}
		extractUint8(body)
		headerValueName, err := extractHeaderValue(body)
		if err != nil {
			return err
		}
		myHeaders.Set(headerTypeName, headerValueName)
	}
	return nil
}
extracts the relevant information from the Headers.
package minio
import (
	"bytes"
	"context"
	"encoding/binary"
	"encoding/xml"
	"errors"
	"fmt"
	"hash"
	"hash/crc32"
	"io"
	"net/http"
	"net/url"
	"strings"
	"github.com/minio/minio-go/v6/pkg/encrypt"
	"github.com/minio/minio-go/v6/pkg/s3utils"
)
type CSVFileHeaderInfo string
const (
	CSVFileHeaderInfoNone   CSVFileHeaderInfo = "NONE"
	CSVFileHeaderInfoIgnore                   = "IGNORE"
	CSVFileHeaderInfoUse                      = "USE"
)
type SelectCompressionType string
const (
	SelectCompressionNONE SelectCompressionType = "NONE"
	SelectCompressionGZIP                       = "GZIP"
	SelectCompressionBZIP                       = "BZIP2"
)
type CSVQuoteFields string
const (
	CSVQuoteFieldsAlways   CSVQuoteFields = "Always"
	CSVQuoteFieldsAsNeeded                = "AsNeeded"
)
type QueryExpressionType string
const (
	QueryExpressionTypeSQL QueryExpressionType = "SQL"
)
type JSONType string
const (
	JSONDocumentType JSONType = "DOCUMENT"
	JSONLinesType             = "LINES"
)
type ParquetInputOptions struct{}
type CSVInputOptions struct {
	FileHeaderInfo    CSVFileHeaderInfo
	fileHeaderInfoSet bool
	RecordDelimiter    string
	recordDelimiterSet bool
	FieldDelimiter    string
	fieldDelimiterSet bool
	QuoteCharacter    string
	quoteCharacterSet bool
	QuoteEscapeCharacter    string
	quoteEscapeCharacterSet bool
	Comments    string
	commentsSet bool
}
func (c *CSVInputOptions) SetFileHeaderInfo(val CSVFileHeaderInfo) {
	c.FileHeaderInfo = val
	c.fileHeaderInfoSet = true
}
func (c *CSVInputOptions) SetRecordDelimiter(val string) {
	c.RecordDelimiter = val
	c.recordDelimiterSet = true
}
func (c *CSVInputOptions) SetFieldDelimiter(val string) {
	c.FieldDelimiter = val
	c.fieldDelimiterSet = true
}
func (c *CSVInputOptions) SetQuoteCharacter(val string) {
	c.QuoteCharacter = val
	c.quoteCharacterSet = true
}
func (c *CSVInputOptions) SetQuoteEscapeCharacter(val string) {
	c.QuoteEscapeCharacter = val
	c.quoteEscapeCharacterSet = true
}
func (c *CSVInputOptions) SetComments(val string) {
	c.Comments = val
	c.commentsSet = true
}
func (c CSVInputOptions) MarshalXML(e *xml.Encoder, start xml.StartElement) error {
	if err := e.EncodeToken(start); err != nil {
		return err
	}
	if c.FileHeaderInfo != "" || c.fileHeaderInfoSet {
		if err := e.EncodeElement(c.FileHeaderInfo, xml.StartElement{Name: xml.Name{Local: "FileHeaderInfo"}}); err != nil {
			return err
		}
	}
	if c.RecordDelimiter != "" || c.recordDelimiterSet {
		if err := e.EncodeElement(c.RecordDelimiter, xml.StartElement{Name: xml.Name{Local: "RecordDelimiter"}}); err != nil {
			return err
		}
	}
	if c.FieldDelimiter != "" || c.fieldDelimiterSet {
		if err := e.EncodeElement(c.FieldDelimiter, xml.StartElement{Name: xml.Name{Local: "FieldDelimiter"}}); err != nil {
			return err
		}
	}
	if c.QuoteCharacter != "" || c.quoteCharacterSet {
		if err := e.EncodeElement(c.QuoteCharacter, xml.StartElement{Name: xml.Name{Local: "QuoteCharacter"}}); err != nil {
			return err
		}
	}
	if c.QuoteEscapeCharacter != "" || c.quoteEscapeCharacterSet {
		if err := e.EncodeElement(c.QuoteEscapeCharacter, xml.StartElement{Name: xml.Name{Local: "QuoteEscapeCharacter"}}); err != nil {
			return err
		}
	}
	if c.Comments != "" || c.commentsSet {
		if err := e.EncodeElement(c.Comments, xml.StartElement{Name: xml.Name{Local: "Comments"}}); err != nil {
			return err
		}
	}
	return e.EncodeToken(xml.EndElement{Name: start.Name})
}
type CSVOutputOptions struct {
	QuoteFields    CSVQuoteFields
	quoteFieldsSet bool
	RecordDelimiter    string
	recordDelimiterSet bool
	FieldDelimiter    string
	fieldDelimiterSet bool
	QuoteCharacter    string
	quoteCharacterSet bool
	QuoteEscapeCharacter    string
	quoteEscapeCharacterSet bool
}
func (c *CSVOutputOptions) SetQuoteFields(val CSVQuoteFields) {
	c.QuoteFields = val
	c.quoteFieldsSet = true
}
func (c *CSVOutputOptions) SetRecordDelimiter(val string) {
	c.RecordDelimiter = val
	c.recordDelimiterSet = true
}
func (c *CSVOutputOptions) SetFieldDelimiter(val string) {
	c.FieldDelimiter = val
	c.fieldDelimiterSet = true
}
func (c *CSVOutputOptions) SetQuoteCharacter(val string) {
	c.QuoteCharacter = val
	c.quoteCharacterSet = true
}
func (c *CSVOutputOptions) SetQuoteEscapeCharacter(val string) {
	c.QuoteEscapeCharacter = val
	c.quoteEscapeCharacterSet = true
}
func (c CSVOutputOptions) MarshalXML(e *xml.Encoder, start xml.StartElement) error {
	if err := e.EncodeToken(start); err != nil {
		return err
	}
	if c.QuoteFields != "" || c.quoteFieldsSet {
		if err := e.EncodeElement(c.QuoteFields, xml.StartElement{Name: xml.Name{Local: "QuoteFields"}}); err != nil {
			return err
		}
	}
	if c.RecordDelimiter != "" || c.recordDelimiterSet {
		if err := e.EncodeElement(c.RecordDelimiter, xml.StartElement{Name: xml.Name{Local: "RecordDelimiter"}}); err != nil {
			return err
		}
	}
	if c.FieldDelimiter != "" || c.fieldDelimiterSet {
		if err := e.EncodeElement(c.FieldDelimiter, xml.StartElement{Name: xml.Name{Local: "FieldDelimiter"}}); err != nil {
			return err
		}
	}
	if c.QuoteCharacter != "" || c.quoteCharacterSet {
		if err := e.EncodeElement(c.QuoteCharacter, xml.StartElement{Name: xml.Name{Local: "QuoteCharacter"}}); err != nil {
			return err
		}
	}
	if c.QuoteEscapeCharacter != "" || c.quoteEscapeCharacterSet {
		if err := e.EncodeElement(c.QuoteEscapeCharacter, xml.StartElement{Name: xml.Name{Local: "QuoteEscapeCharacter"}}); err != nil {
			return err
		}
	}
	return e.EncodeToken(xml.EndElement{Name: start.Name})
}
type JSONInputOptions struct {
	Type    JSONType
	typeSet bool
}
func (j *JSONInputOptions) SetType(typ JSONType) {
	j.Type = typ
	j.typeSet = true
}
func (j JSONInputOptions) MarshalXML(e *xml.Encoder, start xml.StartElement) error {
	if err := e.EncodeToken(start); err != nil {
		return err
	}
	if j.Type != "" || j.typeSet {
		if err := e.EncodeElement(j.Type, xml.StartElement{Name: xml.Name{Local: "Type"}}); err != nil {
			return err
		}
	}
	return e.EncodeToken(xml.EndElement{Name: start.Name})
}
type JSONOutputOptions struct {
	RecordDelimiter    string
	recordDelimiterSet bool
}
func (j *JSONOutputOptions) SetRecordDelimiter(val string) {
	j.RecordDelimiter = val
	j.recordDelimiterSet = true
}
func (j JSONOutputOptions) MarshalXML(e *xml.Encoder, start xml.StartElement) error {
	if err := e.EncodeToken(start); err != nil {
		return err
	}
	if j.RecordDelimiter != "" || j.recordDelimiterSet {
		if err := e.EncodeElement(j.RecordDelimiter, xml.StartElement{Name: xml.Name{Local: "RecordDelimiter"}}); err != nil {
			return err
		}
	}
	return e.EncodeToken(xml.EndElement{Name: start.Name})
}
type SelectObjectInputSerialization struct {
	CompressionType SelectCompressionType
	Parquet         *ParquetInputOptions `xml:"Parquet,omitempty"`
	CSV             *CSVInputOptions     `xml:"CSV,omitempty"`
	JSON            *JSONInputOptions    `xml:"JSON,omitempty"`
}
type SelectObjectOutputSerialization struct {
	CSV  *CSVOutputOptions  `xml:"CSV,omitempty"`
	JSON *JSONOutputOptions `xml:"JSON,omitempty"`
}
type SelectObjectOptions struct {
	XMLName              xml.Name           `xml:"SelectObjectContentRequest" json:"-"`
	ServerSideEncryption encrypt.ServerSide `xml:"-"`
	Expression           string
	ExpressionType       QueryExpressionType
	InputSerialization   SelectObjectInputSerialization
	OutputSerialization  SelectObjectOutputSerialization
	RequestProgress      struct {
		Enabled bool
	}
}
func (o SelectObjectOptions) Header() http.Header {
	headers := make(http.Header)
	if o.ServerSideEncryption != nil && o.ServerSideEncryption.Type() == encrypt.SSEC {
		o.ServerSideEncryption.Marshal(headers)
	}
	return headers
}
type SelectObjectType string
const (
	SelectObjectTypeCSV     SelectObjectType = "CSV"
	SelectObjectTypeJSON                     = "JSON"
	SelectObjectTypeParquet                  = "Parquet"
)
type preludeInfo struct {
	totalLen  uint32
	headerLen uint32
}
type SelectResults struct {
	pipeReader *io.PipeReader
	resp       *http.Response
	stats      *StatsMessage
	progress   *ProgressMessage
}
type ProgressMessage struct {
	XMLName xml.Name `xml:"Progress" json:"-"`
	StatsMessage
}
type StatsMessage struct {
	XMLName        xml.Name `xml:"Stats" json:"-"`
	BytesScanned   int64
	BytesProcessed int64
	BytesReturned  int64
}
type messageType string
const (
	errorMsg  messageType = "error"
	commonMsg             = "event"
)
type eventType string
const (
	endEvent      eventType = "End"
	recordsEvent            = "Records"
	progressEvent           = "Progress"
	statsEvent              = "Stats"
)
type contentType string
const (
	xmlContent contentType = "text/xml"
)
func (c Client) SelectObjectContent(ctx context.Context, bucketName, objectName string, opts SelectObjectOptions) (*SelectResults, error) {
	if err := s3utils.CheckValidBucketName(bucketName); err != nil {
		return nil, err
	}
	if err := s3utils.CheckValidObjectName(objectName); err != nil {
		return nil, err
	}
	selectReqBytes, err := xml.Marshal(opts)
	if err != nil {
		return nil, err
	}
	urlValues := make(url.Values)
	urlValues.Set("select", "")
	urlValues.Set("select-type", "2")
	resp, err := c.executeMethod(ctx, "POST", requestMetadata{
		bucketName:       bucketName,
		objectName:       objectName,
		queryValues:      urlValues,
		customHeader:     opts.Header(),
		contentMD5Base64: sumMD5Base64(selectReqBytes),
		contentSHA256Hex: sum256Hex(selectReqBytes),
		contentBody:      bytes.NewReader(selectReqBytes),
		contentLength:    int64(len(selectReqBytes)),
	})
	if err != nil {
		return nil, err
	}
	return NewSelectResults(resp, bucketName)
}
func NewSelectResults(resp *http.Response, bucketName string) (*SelectResults, error) {
	if resp.StatusCode != http.StatusOK {
		return nil, httpRespToErrorResponse(resp, bucketName, "")
	}
	pipeReader, pipeWriter := io.Pipe()
	streamer := &SelectResults{
		resp:       resp,
		stats:      &StatsMessage{},
		progress:   &ProgressMessage{},
		pipeReader: pipeReader,
	}
	streamer.start(pipeWriter)
	return streamer, nil
}
func (s *SelectResults) Close() error {
	defer closeResponse(s.resp)
	return s.pipeReader.Close()
}
func (s *SelectResults) Read(b []byte) (n int, err error) {
	return s.pipeReader.Read(b)
}
func (s *SelectResults) Stats() *StatsMessage {
	return s.stats
}
func (s *SelectResults) Progress() *ProgressMessage {
	return s.progress
}
func (s *SelectResults) start(pipeWriter *io.PipeWriter) {
	go func() {
		for {
			var prelude preludeInfo
			var headers = make(http.Header)
			var err error
			crc := crc32.New(crc32.IEEETable)
			crcReader := io.TeeReader(s.resp.Body, crc)
			prelude, err = processPrelude(crcReader, crc)
			if err != nil {
				pipeWriter.CloseWithError(err)
				closeResponse(s.resp)
				return
			}
			if prelude.headerLen > 0 {
				if err = extractHeader(io.LimitReader(crcReader, int64(prelude.headerLen)), headers); err != nil {
					pipeWriter.CloseWithError(err)
					closeResponse(s.resp)
					return
				}
			}
			payloadLen := prelude.PayloadLen()
			m := messageType(headers.Get("message-type"))
			switch m {
			case errorMsg:
				pipeWriter.CloseWithError(errors.New(headers.Get("error-code") + ":\"" + headers.Get("error-message") + "\""))
				closeResponse(s.resp)
				return
			case commonMsg:
				c := contentType(headers.Get("content-type"))
				e := eventType(headers.Get("event-type"))
				switch e {
				case endEvent:
					pipeWriter.Close()
					closeResponse(s.resp)
					return
				case recordsEvent:
					if _, err = io.Copy(pipeWriter, io.LimitReader(crcReader, payloadLen)); err != nil {
						pipeWriter.CloseWithError(err)
						closeResponse(s.resp)
						return
					}
				case progressEvent:
					switch c {
					case xmlContent:
						if err = xmlDecoder(io.LimitReader(crcReader, payloadLen), s.progress); err != nil {
							pipeWriter.CloseWithError(err)
							closeResponse(s.resp)
							return
						}
					default:
						pipeWriter.CloseWithError(fmt.Errorf("Unexpected content-type %s sent for event-type %s", c, progressEvent))
						closeResponse(s.resp)
						return
					}
				case statsEvent:
					switch c {
					case xmlContent:
						if err = xmlDecoder(io.LimitReader(crcReader, payloadLen), s.stats); err != nil {
							pipeWriter.CloseWithError(err)
							closeResponse(s.resp)
							return
						}
					default:
						pipeWriter.CloseWithError(fmt.Errorf("Unexpected content-type %s sent for event-type %s", c, statsEvent))
						closeResponse(s.resp)
						return
					}
				}
			}
			if err := checkCRC(s.resp.Body, crc.Sum32()); err != nil {
				pipeWriter.CloseWithError(err)
				closeResponse(s.resp)
				return
			}
		}
	}()
}
func (p preludeInfo) PayloadLen() int64 {
	return int64(p.totalLen - p.headerLen - 16)
}
func processPrelude(prelude io.Reader, crc hash.Hash32) (preludeInfo, error) {
	var err error
	var pInfo = preludeInfo{}
	pInfo.totalLen, err = extractUint32(prelude)
	if err != nil {
		return pInfo, err
	}
	pInfo.headerLen, err = extractUint32(prelude)
	if err != nil {
		return pInfo, err
	}
	preCRC := crc.Sum32()
	if err := checkCRC(prelude, preCRC); err != nil {
		return pInfo, err
	}
	return pInfo, nil
}