header1,header2,header3
1,2,3.4
a,b,c
"d","e","f"
true,false,true
2006-01-02 15:04:06,"2006-01-02 16:04:06",2006-01-02 17:04:06

{"name": "Smith",
"age": 16,
"weight": 65.5,
"org": null,
"projects":
    [
     {"project_name":"project1", "completed":false},
     {"project_name":"project2", "completed":true}
    ]
}

{"name": "Smith",
"age": 16,
"org": null,
"projects":
    [
     {"project_name":"project1", "completed":false},
     {"project_name":"project2", "completed":true}
    ]
}
{"name": "charles",
"age": 17,
"org": "baidu",
"weight": 65.5,
"projects":
    [
     {"project_name":"project3", "completed":false},
     {"project_name":"project4", "completed":true}
    ]
}

public void selectCsv(BosClient client, String bucketName, String csvObject) {
    System.out.println("------ select csv object ------");
    SelectObjectRequest request = new SelectObjectRequest(bucketName, csvObject)
        .withSelectType(Constants.SELECT_TYPE_CSV)
        .withExpression("select * from BosObject limit 3")
        .withExpressionType(SelectObjectRequest.ExpressionType.SQL)
        .withInputSerialization(new InputSerialization()
                .withCompressionType("NONE")
                .withFileHeaderInfo("NONE")
                .withRecordDelimiter("\r\n")
                .withFieldDelimiter(",")
                .withQuoteCharacter("\"")
                .withCommentCharacter("#"))
        .withOutputSerialization(new OutputSerialization()
                .withOutputHeader(false)
                .withQuoteFields("ALWAYS")
                .withRecordDelimiter("\n")
                .withFieldDelimiter(",")
                .withQuoteCharacter("\""))
        .withRequestProgress(false);
    SelectObjectResponse response = client.selectObject(request);

    //输出查询结果
    printRecords(reponse.getMessages()); 
}

public void selectJson(BosClient client, String bucketName, String jsonObject) {
    System.out.println("------ select json object ------");
    SelectObjectRequest request = new SelectObjectRequest(bucketName, jsonkey)
        .withSelectType(Constants.SELECT_TYPE_JSON)
        .withExpression("select * from BosObject where age > 20")
        .withInputSerialization(new InputSerialization()
                .withCompressionType("NONE")
                .withJsonType("LINES"))
        .withOutputSerialization(new OutputSerialization()
                .withRecordDelimiter("\n"))
        .withRequestProgress(false);
    SelectObjectResponse response = client.selectObject(request);

    //输出查询结果
    printRecords(reponse.getMessages()); 
}

public void selectParquet(BosClient client, String bucketName, String parquetObject) {
    System.out.println("------ select parquet object ------");
    SelectObjectRequest request = new SelectObjectRequest(bucketName, parquetObject)
        .withSelectType(Constants.SELECT_TYPE_PARQUET)
        .withExpression("select * from BosObject where age > 20")
        .withInputSerialization(new InputSerialization()
                .withCompressionType("NONE"))
        .withOutputSerialization(new OutputSerialization()
                .withRecordDelimiter("\n"))
        .withRequestProgress(false);
    SelectObjectResponse response = client.selectObject(request);

    //输出查询结果
    printRecords(reponse.getMessages());
}

public void printRecords(SelectObjectResponse.Messages messages) {
    if (messages == null) {
        return;
    }
    while (messages.hasNext()) {
        SelectObjectResponse.CommonMessage message = messages.next();
        if (message.Type.equals("Records")) {
            for (String record: message.getRecords()) {
                System.out.println(record);
            }
        }
    }
}

package main
import (
	"bufio"
	"encoding/binary"
	"fmt"
	"io"
	"strings"
)
import (
	"github.com/baidubce/bce-sdk-go/services/bos"
	"github.com/baidubce/bce-sdk-go/services/bos/api"
)
func main() {
	selectBosObject()
}
func selectBosObject() {
	// 初始化BosClient
	AK, SK := "ak", "sk"
	ENDPOINT := "bj.bcebos.com"
	bosClient, _ := bos.NewClient(AK, SK, ENDPOINT)
	// 先确保bucket，object已经存在，object满足csv/json文件格式要求
	bucket := "select-bucket"
	csvObject := "test.csv"
	fmt.Println("------ select csv object -------")
	csvArgs := &api.SelectObjectArgs{
		SelectType: "csv",
		SelectRequest: &api.SelectObjectRequest{
			Expression:     "c2VsZWN0ICogZnJvbSBCb3NPYmplY3Qgd2hlcmUgY2FzdChfMSBBUyBpbnQpICogY2FzdChfMiBBUyBpbnQpID4gY2FzdChfMyBBUyBmbG9hdCkgKyAx",
			ExpressionType: "SQL",
			InputSerialization: &api.SelectObjectInput{
				CompressionType: "NONE",
				CsvParams: map[string]string{
					"fileHeaderInfo":   "IGNORE",
					"recordDelimiter":  "Cg==",
					"fieldDelimiter":   "LA==",
					"quoteCharacter":   "Ig==",
					"commentCharacter": "Iw==",
				},
			},
			OutputSerialization: &api.SelectObjectOutput{
				OutputHeader: false,
				CsvParams: map[string]string{
					"quoteFields":     "ALWAYS",
					"recordDelimiter": "Cg==",
					"fieldDelimiter":  "LA==",
					"quoteCharacter":  "Ig==",
				},
			},
			RequestProgress: &api.SelectObjectProgress{
				Enabled: true,
			},
		},
	}
	csvRes, err := bosClient.SelectObject(bucket, csvObject, csvArgs)
	if err != nil {
		fmt.Println(err)
		return
	}
	parseMessages(csvRes)
	fmt.Println("------ select json object -------")
	jsonObject := "test.json"
	jsonArgs := &api.SelectObjectArgs{
		SelectType: "json",
		SelectRequest: &api.SelectObjectRequest{
			Expression:     "c2VsZWN0ICogZnJvbSBCb3NPYmplY3QucHJvamVjdHNbKl0ucHJvamVjdF9uYW1l",
			ExpressionType: "SQL",
			InputSerialization: &api.SelectObjectInput{
				CompressionType: "NONE",
				JsonParams: map[string]string{
					"type": "LINES",
				},
			},
			OutputSerialization: &api.SelectObjectOutput{
				JsonParams: map[string]string{
					"recordDelimiter": "Cg==",
				},
			},
			RequestProgress: &api.SelectObjectProgress{
				Enabled: true,
			},
		},
	}
	jsonRes, err := bosClient.SelectObject(bucket, jsonObject, jsonArgs)
	if err != nil {
		fmt.Println(err)
		return
	}
	parseMessages(jsonRes)
}
// 解析所有headers保存到map中
func parseHeaders(headers []byte) map[string]string {
	hm := make(map[string]string)
	index := 0
	for index < len(headers) {
		// headers key length
		keyLen := int(headers[index])
		index += 1
		// headers key
		key := headers[index : index+keyLen]
		index += keyLen

		// headers value length
		valLenByte := headers[index : index+2]
		valLen := int(binary.BigEndian.Uint16(valLenByte))
		index += 2
		// headers value
		val := headers[index : index+valLen]
		index += valLen
		hm[string(key)] = string(val)
	}
	return hm
}

func parseMessages(res *api.SelectObjectResult) {
	defer res.Body.Close()
	reader := bufio.NewReader(res.Body)
	for {
		// total length in prelude, 4 bytes
		p := make([]byte, 4)
		l, err := io.ReadFull(reader, p)
		if err != nil || l < 4 {
			fmt.Printf("read total length err: %+v, len: %d\n", err, l)
			break
		}
		totalLen := binary.BigEndian.Uint32(p)
		// headers length in prelude, 4 bytes
		l, err = io.ReadFull(reader, p)
		if err != nil || l < 4 {
			fmt.Printf("read headers length err: %+v, len: %d\n", err, l)
			break
		}
		headersLen := binary.BigEndian.Uint32(p)
		// headers part
		headers := make([]byte, headersLen)
		l, err = io.ReadFull(reader, headers)
		if err != nil || uint32(l) < headersLen {
			fmt.Printf("read headers data err: %+v, len: %d\n", err, l)
			break
		}
		// 获取header长度，并解析headers内容，判断具体的msg类型；end msg则结束读取，
		// cont msg则调用回调函数输出进度信息，record msg则输出记录信息
		headersMap := parseHeaders(headers)
		if headersMap["message-type"] == "Records" {
			// payload part
			payloadLen := totalLen - headersLen - 12
			payload := make([]byte, payloadLen)
			if _, err := io.ReadFull(reader, payload); err != nil {
				fmt.Printf("read payload data err: %+v\n", err)
			}
			// 设置你使用的OutputSerialization字段中的换行符做分行处理
			rs := strings.Split(string(payload), "\n")
			_, err = io.ReadFull(reader, p)
			crc := binary.BigEndian.Uint32(p)
			recordsMsg := &api.RecordsMessage{
				CommonMessage: api.CommonMessage{
					Prelude: api.Prelude{
						TotalLen:   totalLen,
						HeadersLen: headersLen,
					},
					Headers: headersMap,
					Crc32:   crc,
				},
				Records: rs,
			}
			fmt.Printf("RecordsMessage: %+v\n", recordsMsg)
			continue
		}
		if headersMap["message-type"] == "Cont" {
			// payload part, progress
			bs := make([]byte, 8)
			_, err = io.ReadFull(reader, bs)
			bytesScanned := binary.BigEndian.Uint64(bs)

			br := make([]byte, 8)
			_, err = io.ReadFull(reader, br)
			bytesReturned := binary.BigEndian.Uint64(br)

			_, err = io.ReadFull(reader, p)
			crc := binary.BigEndian.Uint32(p)

			contMsg := &api.ContinuationMessage{
				CommonMessage: api.CommonMessage{
					Prelude: api.Prelude{
						TotalLen:   totalLen,
						HeadersLen: headersLen,
					},
					Headers: headersMap,
					Crc32:   crc,
				},
				BytesScanned:  bytesScanned,
				BytesReturned: bytesReturned,
			}
			fmt.Printf("ContinuationMessage: %+v\n", contMsg)
			continue
		}
		if headersMap["message-type"] == "End" {
			_, err = io.ReadFull(reader, p)
			crc := binary.BigEndian.Uint32(p)

			endMsg := &api.EndMessage{
				CommonMessage: api.CommonMessage{
					Prelude: api.Prelude{
						TotalLen:   totalLen,
						HeadersLen: headersLen,
					},
					Headers: headersMap,
					Crc32:   crc,
				},
			}
			fmt.Printf("EndMessage: %+v\n", endMsg)
			break
		}
	}
}