scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

remove SQL artificial limit

Browse source
This commit is contained in:
Nia Weiss 2020-09-03 09:14:11 -04:00
parent d3485dbdb3
commit daa784319c
No known key found for this signature in database
GPG key ID: 895E83409BFDA1BB
2 changed files with 84 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
sql/extract.go
View file

@ -595,34 +595,37 @@ func extractComparison(expr *sqlparser.ComparisonExpr) (col *parseColumn, op str
return
}
func extractLimitOffset(stmt *sqlparser.Select) (uint, uint, error) {
func extractLimitOffset(stmt *sqlparser.Select) (uint, uint, bool, bool, error) {
if stmt.Limit == nil {
return 100, 0, nil
return 0, 0, false, false, nil
}
var offset, limit uint
var hasOffset, hasLimit bool
if offsetExpr, ok := stmt.Limit.Offset.(*sqlparser.SQLVal); ok {
val, err := extractVal(offsetExpr)
if err != nil {
return 0, 0, err
return 0, 0, false, false, err
}
if offsetVal, ok := val.(int); ok {
offset = uint(offsetVal)
} else {
return 0, 0, errors.New("offset must be an integer")
return 0, 0, false, false, errors.New("offset must be an integer")
}
hasOffset = true
}
if limitExpr, ok := stmt.Limit.Rowcount.(*sqlparser.SQLVal); ok {
val, err := extractVal(limitExpr)
if err != nil {
return 0, 0, err
return 0, 0, false, false, err
}
if limitVal, ok := val.(int); ok {
limit = uint(limitVal)
} else {
return 0, 0, errors.New("limit must be an integer")
return 0, 0, false, false, errors.New("limit must be an integer")
}
hasLimit = true
}
return limit, offset, nil
return limit, offset, hasLimit, hasOffset, nil
}
// extractOrderBy returns the order by fields and directions (asc/desc)

102
sql/select.go
View file

@ -241,7 +241,7 @@ func (h handlerSelectFieldsFromTableWhere) Apply(stmt *sqlparser.Select, qm Quer
}
}
limit, offset, err := extractLimitOffset(stmt)
limit, offset, hasLimit, hasOffset, err := extractLimitOffset(stmt)
if err != nil {
return nil, errors.Wrap(err, "extracting limit")
}
@ -253,8 +253,7 @@ func (h handlerSelectFieldsFromTableWhere) Apply(stmt *sqlparser.Select, qm Quer
mr := &MappingResult{
IndexName: indexName,
//FieldFilters: fields,
Header: selectFields,
Header: selectFields,
}
// assign headers
@ -269,12 +268,24 @@ func (h handlerSelectFieldsFromTableWhere) Apply(stmt *sqlparser.Select, qm Quer
})
// Apply the limit and offset after sorting.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
if hasOffset {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return OffsetRows(result, offset)
})
}
if hasLimit {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(result, limit)
})
}
} else {
// Apply the limit and offset inside the query.
whereQuery = Limit(whereQuery, limit, offset)
switch {
case hasLimit:
whereQuery = Limit(whereQuery, limit, offset)
case hasOffset:
whereQuery = Offset(whereQuery, offset)
}
}
if len(fields) > 0 && fields[0] == "_id" {
@ -308,7 +319,7 @@ func (h handlerSelectDistinctFromTable) Apply(stmt *sqlparser.Select, qm QueryMa
return nil, errors.New("distinct requires a valid field column")
}
limit, offset, err := extractLimitOffset(stmt)
limit, offset, hasLimit, hasOffset, err := extractLimitOffset(stmt)
if err != nil {
return nil, errors.Wrap(err, "extracting limit")
}
@ -353,10 +364,17 @@ func (h handlerSelectDistinctFromTable) Apply(stmt *sqlparser.Select, qm QueryMa
})
}
// Apply a limit and offset to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
// Apply the limit and offset after sorting.
if hasOffset {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return OffsetRows(result, offset)
})
}
if hasLimit {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(result, limit)
})
}
return mr, nil
}
@ -461,7 +479,7 @@ func (h handlerSelectFuncFromTableWhere) Apply(stmt *sqlparser.Select, qm QueryM
}
}
limit, offset, err := extractLimitOffset(stmt)
limit, offset, hasLimit, hasOffset, err := extractLimitOffset(stmt)
if err != nil {
return nil, errors.Wrap(err, "extracting limit offset")
}
@ -506,9 +524,16 @@ func (h handlerSelectFuncFromTableWhere) Apply(stmt *sqlparser.Select, qm QueryM
}
// Apply a limit and offset to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
if hasOffset {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return OffsetRows(result, offset)
})
}
if hasLimit {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(result, limit)
})
}
return mr, nil
}
@ -599,7 +624,7 @@ func (h handlerSelectGroupBy) Apply(stmt *sqlparser.Select, qm QueryMask, indexF
}
}
limit, offset, err := extractLimitOffset(stmt)
limit, offset, hasLimit, hasOffset, err := extractLimitOffset(stmt)
if err != nil {
return nil, errors.Wrap(err, "extracting limit offset")
}
@ -649,9 +674,16 @@ func (h handlerSelectGroupBy) Apply(stmt *sqlparser.Select, qm QueryMask, indexF
}
// Apply a limit and offset to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
if hasOffset {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return OffsetRows(result, offset)
})
}
if hasLimit {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(result, limit)
})
}
return mr, nil
}
@ -684,7 +716,7 @@ func (f handlerSelectIDCountFromTable) Apply(stmt *sqlparser.Select, qm QueryMas
}
}
limit, offset, err := extractLimitOffset(stmt)
limit, offset, hasLimit, hasOffset, err := extractLimitOffset(stmt)
if err != nil {
return nil, errors.Wrap(err, "extracting limit offset")
}
@ -719,9 +751,16 @@ func (f handlerSelectIDCountFromTable) Apply(stmt *sqlparser.Select, qm QueryMas
// supported something like this in Pilosa itself.
// Apply a limit and offset to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
if hasOffset {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return OffsetRows(result, offset)
})
}
if hasLimit {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(result, limit)
})
}
return mr, nil
}
@ -800,7 +839,7 @@ func (h handlerSelectJoin) Apply(stmt *sqlparser.Select, qm QueryMask, indexFunc
qo = rowQry
}
limit, offset, err := extractLimitOffset(stmt)
limit, offset, hasLimit, hasOffset, err := extractLimitOffset(stmt)
if err != nil {
return nil, errors.Wrap(err, "extracting limit")
}
@ -824,9 +863,16 @@ func (h handlerSelectJoin) Apply(stmt *sqlparser.Select, qm QueryMask, indexFunc
}
// Apply a limit and offset to the result.
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(OffsetRows(result, offset), limit)
})
if hasOffset {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return OffsetRows(result, offset)
})
}
if hasLimit {
mr.addReducer(func(result pproto.ToRowser) pproto.ToRowser {
return LimitRows(result, limit)
})
}
return mr, nil
}