Add support for bulk deletes

CHANGELOG.md

@@ -25,6 +25,7 @@ This project adheres to [Semantic Versioning](http://semver.org/).
  - #2694, Make `db-root-spec` stable. - @steve-chavez
    + This can be used to override the OpenAPI spec with a custom database function
  - #1567, On bulk inserts, missing values can get the column DEFAULT by using the `Prefer: missing=default` header - @steve-chavez
+ - #2314, Allow bulk delete by using DELETE with an array body - @laurenceisla
 
 ### Fixed
 

src/PostgREST/ApiRequest.hs

@@ -144,8 +144,8 @@ userApiRequest conf req reqBody = do
   qPrms <- first QueryParamError $ QueryParams.parse (pathIsProc && act `elem` [ActionInvoke InvGet, ActionInvoke InvHead]) $ rawQueryString req
   (acceptMediaType, contentMediaType) <- getMediaTypes conf hdrs act pInfo
   (schema, negotiatedByProfile) <- getSchema conf hdrs method
-  (topLevelRange, ranges) <- getRanges method qPrms hdrs
-  (payload, columns) <- getPayload reqBody contentMediaType qPrms act pInfo
+  (topLevelRange, ranges) <- getRanges method qPrms hdrs isBulk
+  (payload, columns) <- getPayload reqBody contentMediaType qPrms act pInfo isBulk
   return $ ApiRequest {
     iAction = act
   , iTarget = if | pathIsProc    -> TargetProc (QualifiedIdentifier schema pathName) pathIsRootSpec
@@ -154,7 +154,7 @@ userApiRequest conf req reqBody = do
   , iRange = ranges
   , iTopLevelRange = topLevelRange
   , iPayload = payload
-  , iPreferences = Preferences.fromHeaders hdrs
+  , iPreferences = preferences
   , iQueryParams = qPrms
   , iColumns = columns
   , iHeaders = iHdrs
@@ -172,6 +172,8 @@ userApiRequest conf req reqBody = do
     lookupHeader    = flip lookup hdrs
     iHdrs = [ (CI.foldedCase k, v) | (k,v) <- hdrs, k /= hCookie]
     iCkies = maybe [] parseCookies $ lookupHeader "Cookie"
+    preferences = Preferences.fromHeaders hdrs
+    isBulk = Preferences.preferParameters preferences == Just Preferences.MultipleObjects
 
 getPathInfo :: AppConfig -> [Text] -> Either ApiRequestError PathInfo
 getPathInfo AppConfig{configOpenApiMode, configDbRootSpec} path =
@@ -233,9 +235,10 @@ getSchema AppConfig{configDbSchemas} hdrs method = do
     acceptProfile = T.decodeUtf8 <$> lookupHeader "Accept-Profile"
     lookupHeader    = flip lookup hdrs
 
-getRanges :: ByteString -> QueryParams -> RequestHeaders -> Either ApiRequestError (NonnegRange, HM.HashMap Text NonnegRange)
-getRanges method QueryParams{qsOrder,qsRanges} hdrs
+getRanges :: ByteString -> QueryParams -> RequestHeaders -> Bool -> Either ApiRequestError (NonnegRange, HM.HashMap Text NonnegRange)
+getRanges method QueryParams{qsOrder,qsRanges} hdrs isBulk
   | isInvalidRange = Left $ InvalidRange (if rangeIsEmpty headerRange then LowerGTUpper else NegativeLimit)
+  | method == "DELETE" && not (null qsRanges) && isBulk = Left BulkLimitNotAllowedError
   | method `elem` ["PATCH", "DELETE"] && not (null qsRanges) && null qsOrder = Left LimitNoOrderError
   | method == "PUT" && topLevelRange /= allRange = Left PutLimitNotAllowedError
   | otherwise = Right (topLevelRange, ranges)
@@ -252,8 +255,8 @@ getRanges method QueryParams{qsOrder,qsRanges} hdrs
     isInvalidRange = topLevelRange == emptyRange && not (hasLimitZero limitRange)
     topLevelRange = fromMaybe allRange $ HM.lookup "limit" ranges -- if no limit is specified, get all the request rows
 
-getPayload :: RequestBody -> MediaType -> QueryParams.QueryParams -> Action -> PathInfo -> Either ApiRequestError (Maybe Payload, S.Set FieldName)
-getPayload reqBody contentMediaType QueryParams{qsColumns} action PathInfo{pathIsProc}= do
+getPayload :: RequestBody -> MediaType -> QueryParams.QueryParams -> Action -> PathInfo -> Bool -> Either ApiRequestError (Maybe Payload, S.Set FieldName)
+getPayload reqBody contentMediaType QueryParams{qsColumns} action PathInfo{pathIsProc} isBulk = do
   checkedPayload <- if shouldParsePayload then payload else Right Nothing
   let cols = case (checkedPayload, columns) of
         (Just ProcessedJSON{payKeys}, _)       -> payKeys
@@ -291,11 +294,13 @@ getPayload reqBody contentMediaType QueryParams{qsColumns} action PathInfo{pathI
       (ActionInvoke InvPost, _)              -> True
       (ActionMutate MutationSingleUpsert, _) -> True
       (ActionMutate MutationUpdate, _)       -> True
+      (ActionMutate MutationDelete, _)       -> isBulk
       _                                      -> False
 
     columns = case action of
       ActionMutate MutationCreate -> qsColumns
       ActionMutate MutationUpdate -> qsColumns
+      ActionMutate MutationDelete -> if isBulk then qsColumns else Nothing
       ActionInvoke InvPost        -> qsColumns
       _                           -> Nothing
 

src/PostgREST/ApiRequest/Types.hs

@@ -66,24 +66,26 @@ data ApiRequestError
   = AmbiguousRelBetween Text Text [Relationship]
   | AmbiguousRpc [ProcDescription]
   | BinaryFieldError MediaType
-  | MediaTypeError [ByteString]
+  | BulkDelColsPkMatchingError
+  | BulkLimitNotAllowedError
+  | ColumnNotFound Text Text
   | InvalidBody ByteString
   | InvalidFilters
   | InvalidRange RangeError
   | InvalidRpcMethod ByteString
   | LimitNoOrderError
-  | NotFound
+  | MediaTypeError [ByteString]
   | NoRelBetween Text Text (Maybe Text) Text RelationshipsMap
   | NoRpc Text Text [Text] Bool MediaType Bool [QualifiedIdentifier] [ProcDescription]
   | NotEmbedded Text
+  | NotFound
   | PutLimitNotAllowedError
   | QueryParamError QPError
   | RelatedOrderNotToOne Text Text
   | SpreadNotToOne Text Text
   | UnacceptableFilter Text
   | UnacceptableSchema [Text]
   | UnsupportedMethod ByteString
-  | ColumnNotFound Text Text
 
 data QPError = QPError Text Text
 data RangeError

src/PostgREST/Error.hs

@@ -59,28 +59,30 @@ class (JSON.ToJSON a) => PgrstError a where
   errorResponseFor err = responseLBS (status err) (headers err) $ errorPayload err
 
 instance PgrstError ApiRequestError where
-  status AmbiguousRelBetween{}   = HTTP.status300
-  status AmbiguousRpc{}          = HTTP.status300
-  status BinaryFieldError{}      = HTTP.status406
-  status MediaTypeError{}        = HTTP.status415
-  status InvalidBody{}           = HTTP.status400
-  status InvalidFilters          = HTTP.status405
-  status InvalidRpcMethod{}      = HTTP.status405
-  status InvalidRange{}          = HTTP.status416
-  status NotFound                = HTTP.status404
-
-  status NoRelBetween{}          = HTTP.status400
-  status NoRpc{}                 = HTTP.status404
-  status NotEmbedded{}           = HTTP.status400
-  status PutLimitNotAllowedError = HTTP.status400
-  status QueryParamError{}       = HTTP.status400
-  status RelatedOrderNotToOne{}  = HTTP.status400
-  status SpreadNotToOne{}        = HTTP.status400
-  status UnacceptableFilter{}    = HTTP.status400
-  status UnacceptableSchema{}    = HTTP.status406
-  status UnsupportedMethod{}     = HTTP.status405
-  status LimitNoOrderError       = HTTP.status400
-  status ColumnNotFound{}        = HTTP.status400
+  status AmbiguousRelBetween{}      = HTTP.status300
+  status AmbiguousRpc{}             = HTTP.status300
+  status BinaryFieldError{}         = HTTP.status406
+  status BulkDelColsPkMatchingError = HTTP.status400
+  status BulkLimitNotAllowedError   = HTTP.status400
+  status MediaTypeError{}           = HTTP.status415
+  status InvalidBody{}              = HTTP.status400
+  status InvalidFilters             = HTTP.status405
+  status InvalidRpcMethod{}         = HTTP.status405
+  status InvalidRange{}             = HTTP.status416
+  status NotFound                   = HTTP.status404
+
+  status NoRelBetween{}             = HTTP.status400
+  status NoRpc{}                    = HTTP.status404
+  status NotEmbedded{}              = HTTP.status400
+  status PutLimitNotAllowedError    = HTTP.status400
+  status QueryParamError{}          = HTTP.status400
+  status RelatedOrderNotToOne{}     = HTTP.status400
+  status SpreadNotToOne{}           = HTTP.status400
+  status UnacceptableFilter{}       = HTTP.status400
+  status UnacceptableSchema{}       = HTTP.status406
+  status UnsupportedMethod{}        = HTTP.status405
+  status LimitNoOrderError          = HTTP.status400
+  status ColumnNotFound{}           = HTTP.status400
 
   headers _ = [MediaType.toContentType MTApplicationJSON]
 
@@ -172,6 +174,18 @@ instance JSON.ToJSON ApiRequestError where
     "details" .= ("Only is null or not is null filters are allowed on embedded resources":: Text),
     "hint"    .= JSON.Null]
 
+  toJSON BulkDelColsPkMatchingError = JSON.object [
+    "code"    .= ApiRequestErrorCode21,
+    "message" .= ("The payload or 'columns' query string parameter must include all primary key columns for bulk deletes" :: Text),
+    "details" .= JSON.Null,
+    "hint"    .= JSON.Null]
+
+  toJSON BulkLimitNotAllowedError = JSON.object [
+    "code"    .= ApiRequestErrorCode22,
+    "message" .= ("limit/offset query string parameters are not allowed for bulk deletes" :: Text),
+    "details" .= JSON.Null,
+    "hint"    .= JSON.Null]
+
   toJSON (NoRelBetween parent child embedHint schema allRels) = JSON.object [
     "code"    .= SchemaCacheErrorCode00,
     "message" .= ("Could not find a relationship between '" <> parent <> "' and '" <> child <> "' in the schema cache" :: Text),
@@ -598,6 +612,8 @@ data ErrorCode
   | ApiRequestErrorCode18
   | ApiRequestErrorCode19
   | ApiRequestErrorCode20
+  | ApiRequestErrorCode21
+  | ApiRequestErrorCode22
   -- Schema Cache errors
   | SchemaCacheErrorCode00
   | SchemaCacheErrorCode01
@@ -644,6 +660,8 @@ buildErrorCode code = "PGRST" <> case code of
   ApiRequestErrorCode18  -> "118"
   ApiRequestErrorCode19  -> "119"
   ApiRequestErrorCode20  -> "120"
+  ApiRequestErrorCode21  -> "121"
+  ApiRequestErrorCode22  -> "122"
 
   SchemaCacheErrorCode00 -> "200"
   SchemaCacheErrorCode01 -> "201"

src/PostgREST/Plan.hs

@@ -518,7 +518,11 @@ mutatePlan mutation qi ApiRequest{iPreferences=preferences, ..} sCache readReq =
           then mapRight (\typedColumns -> Insert qi typedColumns body (Just (MergeDuplicates, pkCols)) combinedLogic returnings mempty False) typedColumnsOrError
         else
           Left InvalidFilters
-    MutationDelete -> Right $ Delete qi combinedLogic iTopLevelRange rootOrder returnings
+    MutationDelete ->
+        if isBulk && not (S.fromList pkCols `S.isSubsetOf` iColumns)
+          then Left BulkDelColsPkMatchingError
+        else
+          mapRight (\typedColumns -> Delete qi typedColumns body combinedLogic iTopLevelRange rootOrder returnings pkCols isBulk) typedColumnsOrError
   where
     confCols = fromMaybe pkCols qsOnConflict
     QueryParams.QueryParams{..} = iQueryParams
@@ -534,6 +538,8 @@ mutatePlan mutation qi ApiRequest{iPreferences=preferences, ..} sCache readReq =
     tbl = HM.lookup qi $ dbTables sCache
     typedColumnsOrError = resolveOrError tbl `traverse` S.toList iColumns
     applyDefaults = preferences.preferMissing == Just ApplyDefaults
+--    isBulk = preferences.preferParameters == Just MultipleObjects
+    isBulk = not (null pkCols) && preferences.preferParameters == Just MultipleObjects
 
 resolveOrError :: Maybe Table -> FieldName -> Either ApiRequestError TypedField
 resolveOrError Nothing _ = Left NotFound

src/PostgREST/Plan/MutatePlan.hs

@@ -38,8 +38,12 @@ data MutatePlan
       }
   | Delete
       { in_       :: QualifiedIdentifier
+      , delCols   :: [TypedField]
+      , delBody   :: Maybe LBS.ByteString
       , where_    :: [LogicTree]
       , mutRange  :: NonnegRange
       , mutOrder  :: [OrderTerm]
       , returning :: [FieldName]
+      , delPkFlts :: [FieldName]
+      , isBulk    :: Bool
       }

src/PostgREST/Query/QueryBuilder.hs

@@ -83,7 +83,7 @@ getSelectsJoins rr@(Node ReadPlan{select, relName, relToParent=Just rel, relAggA
 mutatePlanToQuery :: MutatePlan -> SQL.Snippet
 mutatePlanToQuery (Insert mainQi iCols body onConflct putConditions returnings _ applyDefaults) =
   "INSERT INTO " <> SQL.sql (fromQi mainQi) <> SQL.sql (if null iCols then " " else "(" <> cols <> ") ") <>
-  fromJsonBodyF body iCols True False applyDefaults <>
+  fromJsonBodyF body iCols True False False applyDefaults <>
   -- Only used for PUT
   (if null putConditions then mempty else "WHERE " <> intercalateSnippet " AND " (pgFmtLogicTree (QualifiedIdentifier mempty "pgrst_body") <$> putConditions)) <>
   SQL.sql (BS.unwords [
@@ -114,13 +114,13 @@ mutatePlanToQuery (Update mainQi uCols body logicForest range ordts returnings a
 
   | range == allRange =
     "UPDATE " <> mainTbl <> " SET " <> SQL.sql nonRangeCols <> " " <>
-    fromJsonBodyF body uCols False False applyDefaults <>
+    fromJsonBodyF body uCols False False False applyDefaults <>
     whereLogic <> " " <>
     SQL.sql (returningF mainQi returnings)
 
   | otherwise =
     "WITH " <>
-    "pgrst_update_body AS (" <> fromJsonBodyF body uCols True True applyDefaults <> "), " <>
+    "pgrst_update_body AS (" <> fromJsonBodyF body uCols True False True applyDefaults <> "), " <>
     "pgrst_affected_rows AS (" <>
       "SELECT " <> SQL.sql rangeIdF <> " FROM " <> mainTbl <>
       whereLogic <> " " <>
@@ -140,9 +140,12 @@ mutatePlanToQuery (Update mainQi uCols body logicForest range ordts returnings a
     rangeCols = BS.intercalate ", " ((\col -> pgFmtIdent (tfName col) <> " = (SELECT " <> pgFmtIdent (tfName col) <> " FROM pgrst_update_body) ") <$> uCols)
     (whereRangeIdF, rangeIdF) = mutRangeF mainQi (fst . otTerm <$> ordts)
 
-mutatePlanToQuery (Delete mainQi logicForest range ordts returnings)
+mutatePlanToQuery (Delete mainQi dCols body logicForest range ordts returnings pkFlts isBulk)
   | range == allRange =
     "DELETE FROM " <> SQL.sql (fromQi mainQi) <> " " <>
+    (if isBulk
+      then fromJsonBodyF body dCols False True False False
+      else mempty) <>
     whereLogic <> " " <>
     SQL.sql (returningF mainQi returnings)
 
@@ -160,7 +163,11 @@ mutatePlanToQuery (Delete mainQi logicForest range ordts returnings)
     SQL.sql (returningF mainQi returnings)
 
   where
-    whereLogic = if null logicForest then mempty else " WHERE " <> intercalateSnippet " AND " (pgFmtLogicTree mainQi <$> logicForest)
+    whereLogic = pgFmtWhereF $ if isBulk then logicForestF <> pgrstDeleteBodyF else logicForestF
+    pgFmtWhereF flts = if null flts then mempty else " WHERE " <> intercalateSnippet " AND " flts
+    logicForestF = pgFmtLogicTree mainQi <$> logicForest
+    pgrstDeleteBodyF = pgFmtBodyFilter mainQi (QualifiedIdentifier mempty "pgrst_body") <$> pkFlts
+    pgFmtBodyFilter table cte f = SQL.sql (pgFmtColumn table f <> " = " <> pgFmtColumn cte f)
     (whereRangeIdF, rangeIdF) = mutRangeF mainQi (fst . otTerm <$> ordts)
 
 callPlanToQuery :: CallPlan -> SQL.Snippet
@@ -171,7 +178,7 @@ callPlanToQuery (FunctionCall qi params args returnsScalar multipleCall returnin
     fromCall = case params of
       OnePosParam prm -> "FROM " <> callIt (singleParameter args $ encodeUtf8 $ ppType prm)
       KeyParams []    -> "FROM " <> callIt mempty
-      KeyParams prms  -> fromJsonBodyF args ((\p -> TypedField (ppName p) (ppType p) Nothing) <$> prms) False (not multipleCall) False <> ", " <>
+      KeyParams prms  -> fromJsonBodyF args ((\p -> TypedField (ppName p) (ppType p) Nothing) <$> prms) False False (not multipleCall) False <> ", " <>
                          "LATERAL " <> callIt (fmtParams prms)
 
     callIt :: SQL.Snippet -> SQL.Snippet

src/PostgREST/Query/SqlFragment.hs

@@ -231,11 +231,12 @@ pgFmtSelectItem table (f@(fName, jp), Nothing, alias) = pgFmtField table f <> SQ
 pgFmtSelectItem table (f@(fName, jp), Just cast, alias) = "CAST (" <> pgFmtField table f <> " AS " <> SQL.sql (encodeUtf8 cast) <> " )" <> SQL.sql (pgFmtAs fName jp alias)
 
 -- TODO: At this stage there shouldn't be a Maybe since ApiRequest should ensure that an INSERT/UPDATE has a body
-fromJsonBodyF :: Maybe LBS.ByteString -> [TypedField] -> Bool -> Bool -> Bool -> SQL.Snippet
-fromJsonBodyF body fields includeSelect includeLimitOne includeDefaults =
+fromJsonBodyF :: Maybe LBS.ByteString -> [TypedField] -> Bool -> Bool -> Bool -> Bool -> SQL.Snippet
+fromJsonBodyF body fields includeSelect includeUsing includeLimitOne includeDefaults =
   SQL.sql
   (if includeSelect then "SELECT " <> parsedCols <> " " else mempty) <>
-  "FROM (SELECT " <> jsonPlaceHolder <> " AS json_data) pgrst_payload, " <>
+  (if includeUsing then "USING " else "FROM ") <>
+  "(SELECT " <> jsonPlaceHolder <> " AS json_data) pgrst_payload, " <>
   -- convert a json object into a json array, this way we can use json_to_recordset for all json payloads
   -- Otherwise we'd have to use json_to_record for json objects and json_to_recordset for json arrays
   -- We do this in SQL to avoid processing the JSON in application code