vendor/github.com/aliyun/aliyun-oss-go-sdk/oss/select_csv_objcet_test.go

   1 package oss
   2
   3 import (
   4         "io/ioutil"
   5         "os"
   6         "strconv"
   7         "strings"
   8         "io"
   9
  10         . "gopkg.in/check.v1"
  11 )
  12
  13 type OssSelectCsvSuite struct {
  14         client *Client
  15         bucket *Bucket
  16 }
  17
  18 var _ = Suite(&OssSelectCsvSuite{})
  19
  20 func (s *OssSelectCsvSuite) SetUpSuite(c *C) {
  21         client, err := New(endpoint, accessID, accessKey)
  22         c.Assert(err, IsNil)
  23         s.client = client
  24         s.client.Config.LogLevel = Error // Debug
  25         // s.client.Config.Timeout = 5
  26         err = s.client.CreateBucket(bucketName)
  27         c.Assert(err, IsNil)
  28         bucket, err := s.client.Bucket(bucketName)
  29         c.Assert(err, IsNil)
  30         s.bucket = bucket
  31
  32         testLogger.Println("test select csv started")
  33 }
  34
  35 func (s *OssSelectCsvSuite) TearDownSuite(c *C) {
  36         // Delete objects
  37         marker := Marker("")
  38         for {
  39                 lor, err := s.bucket.ListObjects(marker)
  40                 c.Assert(err, IsNil)
  41                 for _, object := range lor.Objects {
  42                         err = s.bucket.DeleteObject(object.Key)
  43                         c.Assert(err, IsNil)
  44                 }
  45                 marker = Marker(lor.NextMarker)
  46                 if !lor.IsTruncated {
  47                         break
  48                 }
  49         }
  50
  51         err := s.client.DeleteBucket(bucketName)
  52         c.Assert(err, IsNil)
  53
  54         testLogger.Println("test select csv completed")
  55 }
  56
  57 func (s *OssSelectCsvSuite) SetUpTest(c *C) {
  58         testLogger.Println("test func", c.TestName(), "start")
  59 }
  60
  61 func (s *OssSelectCsvSuite) TearDownTest(c *C) {
  62         testLogger.Println("test func", c.TestName(), "succeed")
  63 }
  64
  65 // TestCreateSelectObjectMeta
  66 func (s *OssSelectCsvSuite) TestCreateSelectCsvObjectMeta(c *C) {
  67         key := "sample_data.csv"
  68         localCsvFile := "../sample/sample_data.csv"
  69         err := s.bucket.PutObjectFromFile(key, localCsvFile)
  70         c.Assert(err, IsNil)
  71         csvMeta := CsvMetaRequest{}
  72         var bo bool
  73         csvMeta.OverwriteIfExists = &bo
  74         res, err := s.bucket.CreateSelectCsvObjectMeta(key, csvMeta)
  75         c.Assert(err, IsNil)
  76         l, err := readCsvLine(localCsvFile)
  77         c.Assert(err, IsNil)
  78         c.Assert(res.RowsCount, Equals, int64(l))
  79
  80         bo = true
  81         csvMeta.OverwriteIfExists = &bo
  82         csvMeta.InputSerialization.CSV.RecordDelimiter = "\n"
  83         csvMeta.InputSerialization.CSV.FieldDelimiter = ","
  84         csvMeta.InputSerialization.CSV.QuoteCharacter = "\""
  85         res, err = s.bucket.CreateSelectCsvObjectMeta(key, csvMeta)
  86         c.Assert(err, IsNil)
  87         c.Assert(res.RowsCount, Equals, int64(l))
  88
  89         err = s.bucket.DeleteObject(key)
  90         c.Assert(err, IsNil)
  91 }
  92
  93 func (s *OssSelectCsvSuite) TestSelectCsvObjectIsEmpty(c *C) {
  94         key := "sample_data.csv"
  95         localCsvFile := "../sample/sample_data.csv"
  96         err := s.bucket.PutObjectFromFile(key, localCsvFile)
  97         c.Assert(err, IsNil)
  98         csvMeta := CsvMetaRequest{}
  99         _, err = s.bucket.CreateSelectCsvObjectMeta(key, csvMeta)
 100         c.Assert(err, IsNil)
 101         selReq := SelectRequest{}
 102         selReq.Expression = "select Year, StateAbbr, CityName, PopulationCount from ossobject where CityName != ''"
 103         selReq.InputSerializationSelect.CsvBodyInput.FileHeaderInfo = "Use"
 104
 105         body, err := s.bucket.SelectObject(key, selReq)
 106         c.Assert(err, IsNil)
 107         defer body.Close()
 108
 109         p := make([]byte, 512)
 110         n, err := body.Read(p)
 111         c.Assert(err, IsNil)
 112         c.Assert(n, Equals, 512)
 113         p1 := make([]byte, 3)
 114         _, err = body.Read(p1)
 115         c.Assert(err, IsNil)
 116         rets, err := ioutil.ReadAll(body)
 117         c.Assert(err, IsNil)
 118         str, err := readCsvIsEmpty(localCsvFile)
 119         c.Assert(err, IsNil)
 120         c.Assert(string(p)+string(p1)+string(rets), Equals, str)
 121
 122         err = s.bucket.DeleteObject(key)
 123         c.Assert(err, IsNil)
 124 }
 125
 126 func (s *OssSelectCsvSuite) TestSelectObjectIntoFile(c *C) {
 127         var bo bool = true
 128         key := "sample_data.csv"
 129         localCsvFile := "../sample/sample_data.csv"
 130         err := s.bucket.PutObjectFromFile(key, localCsvFile)
 131         c.Assert(err, IsNil)
 132         csvMeta := CsvMetaRequest{
 133                 InputSerialization: InputSerialization {
 134                         CSV: CSV {
 135                                 RecordDelimiter: "\n",
 136                                 FieldDelimiter: ",",
 137                                 QuoteCharacter: "\"",
 138                         },
 139                 },
 140                 OverwriteIfExists: &bo,
 141         }
 142         res, err := s.bucket.CreateSelectCsvObjectMeta(key, csvMeta)
 143         c.Assert(err, IsNil)
 144         l, err := readCsvLine(localCsvFile)
 145         c.Assert(err, IsNil)
 146         c.Assert(res.RowsCount, Equals, int64(l))
 147
 148         selReq := SelectRequest{
 149                 Expression:"select * from ossobject",
 150                 InputSerializationSelect: InputSerializationSelect {
 151                         CsvBodyInput :CSVSelectInput{
 152                                 FileHeaderInfo: "None",
 153                                 CommentCharacter: "#",
 154                                 RecordDelimiter: "\n",
 155                                 FieldDelimiter: ",",
 156                                 QuoteCharacter:"\"",
 157                                 Range:"",
 158                         },
 159                 },
 160         }
 161         outfile := "sample_data_out.csv"
 162         err = s.bucket.SelectObjectIntoFile(key, outfile, selReq)
 163         c.Assert(err, IsNil)
 164
 165         fd1, err := os.Open(outfile)
 166         c.Assert(err,IsNil)
 167         defer fd1.Close()
 168         fd2, err := os.Open(localCsvFile)
 169         c.Assert(err,IsNil)
 170         defer fd2.Close()
 171         str1, err := ioutil.ReadAll(fd1)
 172         c.Assert(err,IsNil)
 173         str2 ,err := ioutil.ReadAll(fd2)
 174         c.Assert(err,IsNil)
 175         c.Assert(string(str1), Equals, string(str2))
 176
 177         err = os.Remove(outfile)
 178         c.Assert(err, IsNil)
 179         err = s.bucket.DeleteObject(key)
 180         c.Assert(err, IsNil)
 181 }
 182
 183 func(s *OssSelectCsvSuite) TestSelectCsvObjectRange(c *C) {
 184         key := "sample_data.csv"
 185         localCsvFile := "../sample/sample_data.csv"
 186         err := s.bucket.PutObjectFromFile(key, localCsvFile)
 187         c.Assert(err, IsNil)
 188         csvMeta := CsvMetaRequest{}
 189         _,err = s.bucket.CreateSelectCsvObjectMeta(key, csvMeta)
 190         c.Assert(err, IsNil)
 191         selReq := SelectRequest{}
 192         selReq.Expression = "select Year,StateAbbr, CityName, Short_Question_Text from ossobject"
 193         selReq.InputSerializationSelect.CsvBodyInput.FileHeaderInfo = "Use"
 194         selReq.InputSerializationSelect.CsvBodyInput.Range = "0-2"
 195         body, err := s.bucket.SelectObject(key, selReq)
 196         c.Assert(err, IsNil)
 197         defer body.Close()
 198         rets, err := ioutil.ReadAll(body)
 199
 200         str,err := readCsvRange(localCsvFile, 0, 2)
 201         c.Assert(err, IsNil)
 202         c.Assert(string(rets), Equals, str)
 203
 204         err = s.bucket.DeleteObject(key)
 205         c.Assert(err, IsNil)
 206 }
 207
 208 func(s *OssSelectCsvSuite) TestSelectCsvObjectLike(c *C) {
 209         key := "sample_data.csv"
 210         localCsvFile := "../sample/sample_data.csv"
 211         err := s.bucket.PutObjectFromFile(key, localCsvFile)
 212         c.Assert(err, IsNil)
 213         selReq := SelectRequest{}
 214         selReq.Expression =  "select Year, StateAbbr, CityName, Short_Question_Text from ossobject where Measure like '%blood pressure%Years'"
 215         selReq.InputSerializationSelect.CsvBodyInput.FileHeaderInfo = "Use"
 216         ret,err := s.bucket.SelectObject(key, selReq)
 217         c.Assert(err, IsNil)
 218         defer ret.Close()
 219         ts, err := ioutil.ReadAll(ret)
 220         c.Assert(err, IsNil)
 221         str, err := readCsvLike(localCsvFile)
 222         c.Assert(err, IsNil)
 223         c.Assert(string(ts), Equals, str)
 224
 225         err = s.bucket.DeleteObject(key)
 226         c.Assert(err, IsNil)
 227 }
 228
 229 func(s *OssSelectCsvSuite) TestSelectCsvObjectIntAggregation(c *C) {
 230         key := "sample_data.csv"
 231         localCsvFile := "../sample/sample_data.csv"
 232         err := s.bucket.PutObjectFromFile(key, localCsvFile)
 233         c.Assert(err, IsNil)
 234         selReq := SelectRequest{}
 235         selReq.Expression =  `select avg(cast(year as int)), max(cast(year as int)), min(cast(year as int)) from ossobject where year = 2015`
 236         selReq.InputSerializationSelect.CsvBodyInput.FileHeaderInfo = "Use"
 237         ret,err := s.bucket.SelectObject(key, selReq)
 238         c.Assert(err, IsNil)
 239         defer ret.Close()
 240         ts, err := ioutil.ReadAll(ret)
 241         c.Assert(err, IsNil)
 242
 243         c.Assert(string(ts), Equals, "2015,2015,2015\n")
 244
 245         err = s.bucket.DeleteObject(key)
 246         c.Assert(err, IsNil)
 247 }
 248
 249 func(s *OssSelectCsvSuite) TestSelectCsvObjectFloatAggregation(c *C) {
 250         key := "sample_data.csv"
 251         localCsvFile := "../sample/sample_data.csv"
 252         err := s.bucket.PutObjectFromFile(key, localCsvFile)
 253         c.Assert(err, IsNil)
 254         selReq := SelectRequest{}
 255         selReq.Expression =  `select avg(cast(data_value as double)), max(cast(data_value as double)), sum(cast(data_value as double)) from ossobject`
 256         selReq.InputSerializationSelect.CsvBodyInput.FileHeaderInfo = "Use"
 257         ret,err := s.bucket.SelectObject(key, selReq)
 258         c.Assert(err, IsNil)
 259         defer ret.Close()
 260         ts, err := ioutil.ReadAll(ret)
 261         strR := string(ts)
 262         c.Assert(err, IsNil)
 263
 264         avg, max, sum , err := readCsvFloatAgg(localCsvFile)
 265         c.Assert(err, IsNil)
 266
 267         s1 := strconv.FormatFloat(avg, 'f', 5, 32) + ","
 268         s1 += strconv.FormatFloat(max, 'f', 5, 32) + ","
 269         s1 += strconv.FormatFloat(sum, 'f', 5, 32) + ","
 270         retS := ""
 271         for _, v := range strings.Split(strR[:len(strR)-1], ",") {
 272                 vv, err := strconv.ParseFloat(v, 64)
 273                 c.Assert(err, IsNil)
 274                 retS += strconv.FormatFloat(vv, 'f', 5, 32) + ","
 275         }
 276         c.Assert(s1, Equals, retS)
 277
 278         err = s.bucket.DeleteObject(key)
 279         c.Assert(err, IsNil)
 280 }
 281
 282 func(s *OssSelectCsvSuite) TestSelectCsvObjectConcat(c *C) {
 283         key := "sample_data.csv"
 284         localCsvFile := "../sample/sample_data.csv"
 285         err := s.bucket.PutObjectFromFile(key, localCsvFile)
 286         c.Assert(err, IsNil)
 287         selReq := SelectRequest{}
 288         selReq.Expression =  `select Year,StateAbbr, CityName, Short_Question_Text from ossobject where (data_value || data_value_unit) = '14.8%'`
 289         selReq.InputSerializationSelect.CsvBodyInput.FileHeaderInfo = "Use"
 290         ret,err := s.bucket.SelectObject(key, selReq)
 291         c.Assert(err, IsNil)
 292         defer ret.Close()
 293         ts, err := ioutil.ReadAll(ret)
 294         c.Assert(err, IsNil)
 295
 296         str, err := readCsvConcat(localCsvFile)
 297         c.Assert(err, IsNil)
 298         c.Assert(string(ts), Equals, str)
 299
 300         err = s.bucket.DeleteObject(key)
 301         c.Assert(err, IsNil)
 302 }
 303
 304 func (s *OssSelectCsvSuite) TestSelectCsvObjectComplicateConcat(c *C) {
 305         key := "sample_data.csv"
 306         localCsvFile := "../sample/sample_data.csv"
 307         err := s.bucket.PutObjectFromFile(key, localCsvFile)
 308         c.Assert(err, IsNil)
 309         selReq := SelectRequest{}
 310         selReq.Expression =  `
 311         select
 312                 Year,StateAbbr, CityName, Short_Question_Text, data_value,
 313                 data_value_unit, category, high_confidence_limit
 314         from
 315                 ossobject
 316         where
 317                 data_value > 14.8 and
 318                 data_value_unit = '%' or
 319                 Measure like '%18 Years' and
 320                 Category = 'Unhealthy Behaviors' or
 321                 high_confidence_limit > 70.0 `
 322
 323         selReq.InputSerializationSelect.CsvBodyInput.FileHeaderInfo = "Use"
 324         ret,err := s.bucket.SelectObject(key, selReq)
 325         c.Assert(err, IsNil)
 326         defer ret.Close()
 327         ts, err := ioutil.ReadAll(ret)
 328         c.Assert(err, IsNil)
 329
 330         str, err := readCsvComplicateCondition(localCsvFile)
 331         c.Assert(err, IsNil)
 332         c.Assert(string(ts), Equals, str)
 333
 334         err = s.bucket.DeleteObject(key)
 335         c.Assert(err, IsNil)
 336 }
 337
 338 func (s *OssSelectCsvSuite) TestSelectCsvObjectInvalidSql(c *C) {
 339         key := "sample_data.csv"
 340         localCsvFile := "../sample/sample_data.csv"
 341         err := s.bucket.PutObjectFromFile(key, localCsvFile)
 342         c.Assert(err, IsNil)
 343         selReq := SelectRequest{}
 344         selReq.Expression =  `select * from ossobject where avg(cast(year as int)) > 2016`
 345         selReq.InputSerializationSelect.CsvBodyInput.FileHeaderInfo = "Use"
 346         _, err = s.bucket.SelectObject(key, selReq)
 347         c.Assert(err, NotNil)
 348
 349         selReq.Expression =  ``
 350         _, err = s.bucket.SelectObject(key, selReq)
 351         c.Assert(err, NotNil)
 352
 353         selReq.Expression =  `select year || CityName from ossobject`
 354         _, err = s.bucket.SelectObject(key, selReq)
 355         c.Assert(err, NotNil)
 356
 357         selReq.Expression =  `select * from ossobject group by CityName`
 358         _, err = s.bucket.SelectObject(key, selReq)
 359         c.Assert(err, NotNil)
 360
 361         selReq.Expression =  `select * from ossobject order by _1`
 362         _, err = s.bucket.SelectObject(key, selReq)
 363         c.Assert(err, NotNil)
 364
 365         selReq.Expression =  `select * from ossobject oss join s3object s3 on oss.CityName = s3.CityName`
 366         _, err = s.bucket.SelectObject(key, selReq)
 367         c.Assert(err, NotNil)
 368
 369         selReq.Expression =  `select _1 from ossobject`
 370         ret, err := s.bucket.SelectObject(key, selReq)
 371         c.Assert(err, IsNil)
 372         defer ret.Close()
 373         _, err = ioutil.ReadAll(ret)
 374         c.Assert(err, IsNil)
 375
 376         err = s.bucket.DeleteObject(key)
 377         c.Assert(err, IsNil)
 378 }
 379
 380 func (s *OssSelectCsvSuite) TestSelectCsvObjectWithOutputDelimiters(c *C) {
 381         key := "sample_data.csv"
 382         content := "abc,def\n"
 383         err := s.bucket.PutObject(key, strings.NewReader(content))
 384         c.Assert(err, IsNil)
 385         selReq := SelectRequest{}
 386         selReq.Expression =  `select _1, _2 from ossobject `
 387         selReq.OutputSerializationSelect.CsvBodyOutput.RecordDelimiter = "\r\n"
 388         selReq.OutputSerializationSelect.CsvBodyOutput.FieldDelimiter = "|"
 389
 390         ret,err := s.bucket.SelectObject(key, selReq)
 391         c.Assert(err, IsNil)
 392         defer ret.Close()
 393         ts, err := ioutil.ReadAll(ret)
 394         c.Assert(err, IsNil)
 395         c.Assert(string(ts), Equals, "abc|def\r\n")
 396
 397         err = s.bucket.DeleteObject(key)
 398         c.Assert(err, IsNil)
 399 }
 400
 401 func (s *OssSelectCsvSuite) TestSelectCsvObjectWithCrc(c *C) {
 402         key := "sample_data.csv"
 403         content := "abc,def\n"
 404         err := s.bucket.PutObject(key, strings.NewReader(content))
 405         c.Assert(err, IsNil)
 406         selReq := SelectRequest{}
 407         selReq.Expression =  `select * from ossobject`
 408         bo := true
 409         selReq.OutputSerializationSelect.EnablePayloadCrc = &bo
 410
 411         ret,err := s.bucket.SelectObject(key, selReq)
 412         c.Assert(err, IsNil)
 413         defer ret.Close()
 414         ts, err := ioutil.ReadAll(ret)
 415         c.Assert(err, IsNil)
 416         c.Assert(string(ts), Equals, content)
 417
 418         err = s.bucket.DeleteObject(key)
 419         c.Assert(err, IsNil)
 420 }
 421
 422 func (s *OssSelectCsvSuite) TestSelectCsvObjectWithSkipPartialData(c *C) {
 423         key := "sample_data.csv"
 424         content := "abc,def\nefg\n"
 425         err := s.bucket.PutObject(key, strings.NewReader(content))
 426         c.Assert(err, IsNil)
 427         selReq := SelectRequest{}
 428         selReq.Expression =  `select _1, _2 from ossobject`
 429         bo := true
 430         selReq.SelectOptions.SkipPartialDataRecord = &bo
 431         ret,err := s.bucket.SelectObject(key, selReq)
 432         c.Assert(err, IsNil)
 433         defer ret.Close()
 434         ts, err := ioutil.ReadAll(ret)
 435         c.Assert(err, IsNil)
 436         c.Assert(string(ts), Equals, "abc,def\n")
 437
 438         err = s.bucket.DeleteObject(key)
 439         c.Assert(err, IsNil)
 440 }
 441
 442 func (s *OssSelectCsvSuite) TestSelectCsvObjectWithOutputRaw(c *C) {
 443         key := "sample_data.csv"
 444         content := "abc,def\n"
 445         err := s.bucket.PutObject(key, strings.NewReader(content))
 446         c.Assert(err, IsNil)
 447         selReq := SelectRequest{}
 448         selReq.Expression =  `select _1 from ossobject`
 449         bo := true
 450         selReq.OutputSerializationSelect.OutputRawData = &bo
 451
 452         ret,err := s.bucket.SelectObject(key, selReq)
 453         c.Assert(err, IsNil)
 454         defer ret.Close()
 455         ts, err := ioutil.ReadAll(ret)
 456         c.Assert(err, IsNil)
 457         c.Assert(string(ts), Equals, "abc\n")
 458
 459         err = s.bucket.DeleteObject(key)
 460         c.Assert(err, IsNil)
 461 }
 462
 463 func (s *OssSelectCsvSuite) TestSelectCsvObjectWithKeepColumns(c *C) {
 464         key := "sample_data.csv"
 465         content := "abc,def\n"
 466         err := s.bucket.PutObject(key, strings.NewReader(content))
 467         c.Assert(err, IsNil)
 468         selReq := SelectRequest{}
 469         selReq.Expression =  `select _1 from ossobject`
 470         bo := true
 471         selReq.OutputSerializationSelect.KeepAllColumns = &bo
 472
 473         ret,err := s.bucket.SelectObject(key, selReq)
 474         c.Assert(err, IsNil)
 475         defer ret.Close()
 476         ts, err := ioutil.ReadAll(ret)
 477         c.Assert(err, IsNil)
 478         c.Assert(string(ts), Equals, "abc,\n")
 479
 480         err = s.bucket.DeleteObject(key)
 481         c.Assert(err, IsNil)
 482 }
 483
 484 func (s *OssSelectCsvSuite) TestSelectCsvObjectWithOutputHeader(c *C) {
 485         key := "sample_data.csv"
 486         content := "name,job\nabc,def\n"
 487         err := s.bucket.PutObject(key, strings.NewReader(content))
 488         c.Assert(err, IsNil)
 489         selReq := SelectRequest{}
 490         selReq.Expression =  `select name from ossobject`
 491         bo := true
 492         selReq.OutputSerializationSelect.OutputHeader = &bo
 493         selReq.InputSerializationSelect.CsvBodyInput.FileHeaderInfo = "Use"
 494
 495         ret,err := s.bucket.SelectObject(key, selReq)
 496         c.Assert(err, IsNil)
 497         defer ret.Close()
 498         ts, err := ioutil.ReadAll(ret)
 499         c.Assert(err, IsNil)
 500         c.Assert(string(ts), Equals, "name\nabc\n")
 501
 502         err = s.bucket.DeleteObject(key)
 503         c.Assert(err, IsNil)
 504 }
 505
 506 func (s *OssSelectCsvSuite) TestSelectCsvObjectRead(c *C) {
 507         key := "sample_data.csv"
 508         content := "name,job\nabc,def\n"
 509         err := s.bucket.PutObject(key, strings.NewReader(content))
 510         c.Assert(err, IsNil)
 511         selReq := SelectRequest{}
 512         selReq.Expression =  `select name from ossobject`
 513         bo := true
 514         selReq.OutputSerializationSelect.OutputHeader = &bo
 515         selReq.InputSerializationSelect.CsvBodyInput.FileHeaderInfo = "Use"
 516         selReq.OutputSerializationSelect.EnablePayloadCrc = &bo
 517
 518         ret,err := s.bucket.SelectObject(key, selReq)
 519         c.Assert(err, IsNil)
 520         defer ret.Close()
 521
 522         // case 1: read length > data length
 523         p := make([]byte, 512)
 524         n, err := ret.Read(p[:20])
 525         if err != nil && err != io.EOF {
 526                 c.Assert(err, IsNil)
 527         }
 528         c.Assert(string(p[:n]), Equals, "name\nabc\n")
 529         ts, err := ioutil.ReadAll(ret)
 530         c.Assert(err, IsNil)
 531         c.Assert(string(ts), Equals, "")
 532
 533         // case 2: read length = data length
 534         ret,err = s.bucket.SelectObject(key, selReq)
 535         c.Assert(err, IsNil)
 536         defer ret.Close()
 537         n, err = ret.Read(p[:9])
 538         if err != nil && err != io.EOF {
 539                 c.Assert(err, IsNil)
 540         }
 541         c.Assert(string(p[:n]), Equals, "name\nabc\n")
 542         ts, err = ioutil.ReadAll(ret)
 543         c.Assert(err, IsNil)
 544         c.Assert(string(ts), Equals, "")
 545
 546         // case 3: read length > one frame length and read length < two frame, (this data = 2 * frame length)
 547         ret,err = s.bucket.SelectObject(key, selReq)
 548         c.Assert(err, IsNil)
 549         defer ret.Close()
 550         n, err = ret.Read(p[:7])
 551         if err != nil && err != io.EOF {
 552                 c.Assert(err, IsNil)
 553         }
 554         c.Assert(string(p[:n]), Equals, "name\nab")
 555         ts, err = ioutil.ReadAll(ret)
 556         c.Assert(err, IsNil)
 557         c.Assert(string(ts), Equals, "c\n")
 558
 559         // case 4: read length = a frame length (this data = 2 * frame length)
 560         ret,err = s.bucket.SelectObject(key, selReq)
 561         c.Assert(err, IsNil)
 562         defer ret.Close()
 563         n, err = ret.Read(p[:5])
 564         if err != nil && err != io.EOF {
 565                 c.Assert(err, IsNil)
 566         }
 567         c.Assert(string(p[:n]), Equals, "name\n")
 568         ts, err = ioutil.ReadAll(ret)
 569         c.Assert(err, IsNil)
 570         c.Assert(string(ts), Equals, "abc\n")
 571
 572         // case 5: read length < a frame length (this data = 2 * frame length)
 573         ret,err = s.bucket.SelectObject(key, selReq)
 574         c.Assert(err, IsNil)
 575         defer ret.Close()
 576         n, err = ret.Read(p[:3])
 577         if err != nil && err != io.EOF {
 578                 c.Assert(err, IsNil)
 579         }
 580         c.Assert(string(p[:n]), Equals, "nam")
 581         ts, err = ioutil.ReadAll(ret)
 582         c.Assert(err, IsNil)
 583         c.Assert(string(ts), Equals, "e\nabc\n")
 584
 585         err = s.bucket.DeleteObject(key)
 586         c.Assert(err, IsNil)
 587 }
 588
 589 // OssProgressListener is the progress listener
 590 type OssSelectProgressListener struct {
 591 }
 592 // ProgressChanged handles progress event
 593 func (listener *OssSelectProgressListener) ProgressChanged(event *ProgressEvent) {
 594         switch event.EventType {
 595         case TransferStartedEvent:
 596                 testLogger.Printf("Transfer Started.\n")
 597         case TransferDataEvent:
 598                 testLogger.Printf("Transfer Data, This time consumedBytes: %d \n", event.ConsumedBytes)
 599         case TransferCompletedEvent:
 600                 testLogger.Printf("Transfer Completed, This time consumedBytes: %d.\n", event.ConsumedBytes)
 601         case TransferFailedEvent:
 602                 testLogger.Printf("Transfer Failed, This time consumedBytes: %d.\n", event.ConsumedBytes)
 603         default:
 604         }
 605 }
 606
 607 func(s *OssSelectCsvSuite) TestSelectCsvObjectConcatProgress(c *C) {
 608         key := "sample_data.csv"
 609         localCsvFile := "../sample/sample_data.csv"
 610         err := s.bucket.PutObjectFromFile(key, localCsvFile)
 611         c.Assert(err, IsNil)
 612         selReq := SelectRequest{}
 613         selReq.Expression =  `select Year,StateAbbr, CityName, Short_Question_Text from ossobject where (data_value || data_value_unit) = '14.8%'`
 614         selReq.InputSerializationSelect.CsvBodyInput.FileHeaderInfo = "Use"
 615         ret,err := s.bucket.SelectObject(key, selReq, Progress(&OssSelectProgressListener{}))
 616         c.Assert(err, IsNil)
 617         defer ret.Close()
 618         ts, err := ioutil.ReadAll(ret)
 619         c.Assert(err, IsNil)
 620
 621         str, err := readCsvConcat(localCsvFile)
 622         c.Assert(err, IsNil)
 623         c.Assert(string(ts), Equals, str)
 624
 625         err = s.bucket.DeleteObject(key)
 626         c.Assert(err, IsNil)
 627 }