From 18498172a8fede2b28258578fd7452fad26fa07b Mon Sep 17 00:00:00 2001 From: Steven Allen Date: Tue, 1 Oct 2019 10:28:22 -0700 Subject: [PATCH 1/6] feat(query): correctly handle limit/offset --- go.sum | 2 ++ s3.go | 18 +++++++++++++----- s3_test.go | 13 +------------ 3 files changed, 16 insertions(+), 17 deletions(-) diff --git a/go.sum b/go.sum index 4d8fd9d..8a472cc 100644 --- a/go.sum +++ b/go.sum @@ -51,6 +51,8 @@ github.com/aws/aws-sdk-go v1.35.30/go.mod h1:tlPOdRjfxPBpNIwqDj61rmsnA85v9jc0Ps9 github.com/benbjohnson/clock v1.0.2/go.mod h1:bGMdMPoPVvcYyt1gHDf4J2KE153Yf9BuiUKYMaxlTDM= github.com/benbjohnson/clock v1.0.3 h1:vkLuvpK4fmtSCuo60+yC63p7y0BmQ8gm5ZXGuBCJyXg= github.com/benbjohnson/clock v1.0.3/go.mod h1:bGMdMPoPVvcYyt1gHDf4J2KE153Yf9BuiUKYMaxlTDM= +github.com/aws/aws-sdk-go v1.28.13 h1:JyCQQ86yil3hg7MtWdNH8Pbcgx92qlUV2v22Km63Mf4= +github.com/aws/aws-sdk-go v1.28.13/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo= github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q= github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= diff --git a/s3.go b/s3.go index 34b2113..ec69424 100644 --- a/s3.go +++ b/s3.go @@ -176,15 +176,18 @@ func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { // S3 store a "/foo" key as "foo" so we need to trim the leading "/" q.Prefix = strings.TrimPrefix(q.Prefix, "/") - limit := q.Limit + q.Offset - if limit == 0 || limit > listMax { - limit = listMax + sent := 0 + queryLimit := func() int64 { + if max := q.Limit - sent; q.Limit <= 0 && max < listMax { + return int64(max) + } + return listMax } resp, err := s.S3.ListObjectsV2(&s3.ListObjectsV2Input{ Bucket: aws.String(s.Bucket), Prefix: aws.String(s.s3Path(q.Prefix)), - MaxKeys: aws.Int64(int64(limit)), + MaxKeys: aws.Int64(queryLimit()), }) if err != nil { return nil, err @@ -192,6 +195,10 @@ func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { index := q.Offset nextValue := func() (dsq.Result, bool) { + if q.Limit > 0 && sent >= q.Limit { + return dsq.Result{}, false + } + for index >= len(resp.Contents) { if !*resp.IsTruncated { return dsq.Result{}, false @@ -203,7 +210,7 @@ func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { Bucket: aws.String(s.Bucket), Prefix: aws.String(s.s3Path(q.Prefix)), Delimiter: aws.String("/"), - MaxKeys: aws.Int64(listMax), + MaxKeys: aws.Int64(queryLimit()), ContinuationToken: resp.NextContinuationToken, }) if err != nil { @@ -224,6 +231,7 @@ func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { } index++ + sent++ return dsq.Result{Entry: entry}, true } diff --git a/s3_test.go b/s3_test.go index 4f573e7..8e34c30 100644 --- a/s3_test.go +++ b/s3_test.go @@ -34,18 +34,7 @@ func TestSuiteLocalS3(t *testing.T) { t.Fatal(err) } - t.Run("basic operations", func(t *testing.T) { - dstest.SubtestBasicPutGet(t, s3ds) - }) - t.Run("not found operations", func(t *testing.T) { - dstest.SubtestNotFounds(t, s3ds) - }) - t.Run("many puts and gets, query", func(t *testing.T) { - dstest.SubtestManyKeysAndQuery(t, s3ds) - }) - t.Run("return sizes", func(t *testing.T) { - dstest.SubtestReturnSizes(t, s3ds) - }) + dstest.SubtestAll(t, s3ds) } func devMakeBucket(s3obj *s3.S3, bucketName string) error { From fcbd2ef19c5021a797ca682db6878d3bcfc43498 Mon Sep 17 00:00:00 2001 From: Steven Allen Date: Wed, 12 Feb 2020 13:45:30 -0800 Subject: [PATCH 2/6] fix: clean prefix --- s3.go | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/s3.go b/s3.go index ec69424..3ec8126 100644 --- a/s3.go +++ b/s3.go @@ -173,8 +173,9 @@ func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { return nil, fmt.Errorf("s3ds: filters or orders are not supported") } - // S3 store a "/foo" key as "foo" so we need to trim the leading "/" - q.Prefix = strings.TrimPrefix(q.Prefix, "/") + // Normalize the path and strip the leading / as S3 stores values + // without the leading /. + prefix := ds.NewKey(q.Prefix).String()[1:] sent := 0 queryLimit := func() int64 { @@ -186,7 +187,7 @@ func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { resp, err := s.S3.ListObjectsV2(&s3.ListObjectsV2Input{ Bucket: aws.String(s.Bucket), - Prefix: aws.String(s.s3Path(q.Prefix)), + Prefix: aws.String(s.s3Path(prefix)), MaxKeys: aws.Int64(queryLimit()), }) if err != nil { @@ -208,7 +209,7 @@ func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { resp, err = s.S3.ListObjectsV2(&s3.ListObjectsV2Input{ Bucket: aws.String(s.Bucket), - Prefix: aws.String(s.s3Path(q.Prefix)), + Prefix: aws.String(s.s3Path(prefix)), Delimiter: aws.String("/"), MaxKeys: aws.Int64(queryLimit()), ContinuationToken: resp.NextContinuationToken, From 2b4382d0ee1faad5c637acd7ba91b1cbc2d08c9a Mon Sep 17 00:00:00 2001 From: Steven Allen Date: Wed, 12 Feb 2020 14:15:57 -0800 Subject: [PATCH 3/6] fix: support order/filter --- s3.go | 40 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 38 insertions(+), 2 deletions(-) diff --git a/s3.go b/s3.go index 3ec8126..ee1b9bd 100644 --- a/s3.go +++ b/s3.go @@ -168,9 +168,45 @@ func (s *S3Bucket) Delete(k ds.Key) error { return err } +func querySupported(q dsq.Query) bool { + if len(q.Orders) > 0 { + switch q.Orders[0].(type) { + case dsq.OrderByKey, *dsq.OrderByKey: + // We order by key by default. + default: + return false + } + } + return len(q.Filters) == 0 +} + func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { - if q.Orders != nil || q.Filters != nil { - return nil, fmt.Errorf("s3ds: filters or orders are not supported") + // Handle ordering + if !querySupported(q) { + // OK, time to do this the naive way. + + // Skip the stuff we can't apply. + baseQuery := q + baseQuery.Filters = nil + baseQuery.Orders = nil + baseQuery.Limit = 0 // needs to apply after we order + baseQuery.Offset = 0 // ditto. + + // perform the base query. + res, err := s.Query(baseQuery) + if err != nil { + return nil, err + } + + // fix the query + res = dsq.ResultsReplaceQuery(res, q) + + // Remove the prefix, S3 has already handled it. + naiveQuery := q + naiveQuery.Prefix = "" + + // Apply the rest of the query + return dsq.NaiveQueryApply(naiveQuery, res), nil } // Normalize the path and strip the leading / as S3 stores values From cb1ea9fe6d5c5eddda30f8b28785669173b9654b Mon Sep 17 00:00:00 2001 From: Steven Allen Date: Wed, 12 Feb 2020 14:33:34 -0800 Subject: [PATCH 4/6] fix: set the delimiter for the initial query --- s3.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/s3.go b/s3.go index ee1b9bd..7efdfd3 100644 --- a/s3.go +++ b/s3.go @@ -222,9 +222,10 @@ func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { } resp, err := s.S3.ListObjectsV2(&s3.ListObjectsV2Input{ - Bucket: aws.String(s.Bucket), - Prefix: aws.String(s.s3Path(prefix)), - MaxKeys: aws.Int64(queryLimit()), + Bucket: aws.String(s.Bucket), + Prefix: aws.String(s.s3Path(prefix)), + Delimiter: aws.String("/"), + MaxKeys: aws.Int64(queryLimit()), }) if err != nil { return nil, err From b185709fa4477b2a1618f6345d708e22529fc87c Mon Sep 17 00:00:00 2001 From: Steven Allen Date: Wed, 12 Feb 2020 14:45:58 -0800 Subject: [PATCH 5/6] fix: concurrent modification and query --- s3.go | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/s3.go b/s3.go index 7efdfd3..6dc5e81 100644 --- a/s3.go +++ b/s3.go @@ -233,6 +233,7 @@ func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { index := q.Offset nextValue := func() (dsq.Result, bool) { + tryAgain: if q.Limit > 0 && sent >= q.Limit { return dsq.Result{}, false } @@ -262,8 +263,18 @@ func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { } if !q.KeysOnly { value, err := s.Get(ds.NewKey(entry.Key)) - if err != nil { - return dsq.Result{Error: err}, false + switch err { + case nil: + case ds.ErrNotFound: + // This just means the value got deleted in the + // mean-time. That's not an error. + // + // We could use a loop instead of a goto, but + // this is one of those rare cases where a goto + // is easier to understand. + goto tryAgain + default: + return dsq.Result{Entry: entry, Error: err}, false } entry.Value = value } From a92819084c0c76fde01cc6b5011eac87c40511b0 Mon Sep 17 00:00:00 2001 From: Steven Allen Date: Wed, 12 Feb 2020 15:37:09 -0800 Subject: [PATCH 6/6] fix: s3 query offset --- s3.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/s3.go b/s3.go index 6dc5e81..ecb338b 100644 --- a/s3.go +++ b/s3.go @@ -215,8 +215,8 @@ func (s *S3Bucket) Query(q dsq.Query) (dsq.Results, error) { sent := 0 queryLimit := func() int64 { - if max := q.Limit - sent; q.Limit <= 0 && max < listMax { - return int64(max) + if q.Limit > 0 && (q.Limit-sent) < listMax { + return int64(q.Limit - sent) } return listMax }