feat: data representations allow custom parsing and formatting of API fields.

See PR #2523. Most notable code changes: - Load data representation casts into schema cache. - Data representations for reads, filters, inserts, updates, views, over joins. - `CoercibleField` represents name references in queries where coercion may be needed. - `ResolverContext` help facilitate field resolution during planning. - Planner 'resolves' names in the API query and pairs them with any implicit conversions to be used in the query builder stage. - Tests for all of the above. - More consistent naming (TypedX -> CoercibleX). New: unit tests for more data representation use cases; helpful as examples as well. New: update CHANGELOG with data representations feature description. Fixed failing idempotence test. New: replace date formatter test with one that does something. Fixup: inadvertent CHANGELOG change after rebase. Cleanup: `tfName` -> `cfName` and related. Document what IRType means. Formatting. New: use a subquery to interpret `IN` literals requiring data rep transformation. - With the previous method, very long queries such as `ANY (ARRAY[test.color('000100'), test.color('CAFE12'), test.color('01E240'), ...` could be generated. Consider the case where the parser function name is 45 characters and there's a hundred literals. That's 4.5kB of SQL just for the function name alone! - New version uses `unnest`: `ANY (SELECT test.color(unnest('{000100,CAFE12,01E240,...}'::text[]))` to produce a much shorter query. - This is likely to be more performant and either way much more readable and debuggable in the logs.
2023-06-29 15:01:58 -05:00
parent 078c6ec08c
commit 0a1564ba5a
17 changed files with 1045 additions and 163 deletions
@@ -25,9 +25,9 @@ module PostgREST.Plan
  , inspectPlanTxMode
  ) where

-
 import qualified Data.ByteString.Lazy          as LBS
 import qualified Data.HashMap.Strict           as HM
+import qualified Data.HashMap.Strict.InsOrd    as HMI
 import qualified Data.List                     as L
 import qualified Data.Set                      as S
 import qualified PostgREST.SchemaCache.Routine as Routine
@@ -36,34 +36,39 @@ import Data.Either.Combinators (mapLeft, mapRight)
 import Data.List               (delete)
 import Data.Tree               (Tree (..))

-import PostgREST.ApiRequest               (Action (..),
-                                           ApiRequest (..),
-                                           InvokeMethod (..),
-                                           Mutation (..),
-                                           Payload (..))
-import PostgREST.Config                   (AppConfig (..))
-import PostgREST.Error                    (Error (..))
-import PostgREST.MediaType                (MediaType (..))
-import PostgREST.Query.SqlFragment        (sourceCTEName)
-import PostgREST.RangeQuery               (NonnegRange, allRange,
-                                           convertToLimitZeroRange,
-                                           restrictRange)
-import PostgREST.SchemaCache              (SchemaCache (..))
-import PostgREST.SchemaCache.Identifiers  (FieldName,
-                                           QualifiedIdentifier (..),
-                                           Schema)
-import PostgREST.SchemaCache.Relationship (Cardinality (..),
-                                           Junction (..),
-                                           Relationship (..),
-                                           RelationshipsMap,
-                                           relIsToOne)
-import PostgREST.SchemaCache.Routine      (Routine (..), RoutineMap,
-                                           RoutineParam (..),
-                                           funcReturnsCompositeAlias,
-                                           funcReturnsScalar,
-                                           funcReturnsSetOfScalar)
-import PostgREST.SchemaCache.Table        (Table (tableName),
-                                           tablePKCols)
+import PostgREST.ApiRequest                  (Action (..),
+                                              ApiRequest (..),
+                                              InvokeMethod (..),
+                                              Mutation (..),
+                                              Payload (..))
+import PostgREST.Config                      (AppConfig (..))
+import PostgREST.Error                       (Error (..))
+import PostgREST.MediaType                   (MediaType (..))
+import PostgREST.Query.SqlFragment           (sourceCTEName)
+import PostgREST.RangeQuery                  (NonnegRange, allRange,
+                                              convertToLimitZeroRange,
+                                              restrictRange)
+import PostgREST.SchemaCache                 (SchemaCache (..))
+import PostgREST.SchemaCache.Identifiers     (FieldName,
+                                              QualifiedIdentifier (..),
+                                              Schema)
+import PostgREST.SchemaCache.Relationship    (Cardinality (..),
+                                              Junction (..),
+                                              Relationship (..),
+                                              RelationshipsMap,
+                                              relIsToOne)
+import PostgREST.SchemaCache.Representations (DataRepresentation (..),
+                                              RepresentationsMap)
+import PostgREST.SchemaCache.Routine         (Routine (..),
+                                              RoutineMap,
+                                              RoutineParam (..),
+                                              funcReturnsCompositeAlias,
+                                              funcReturnsScalar,
+                                              funcReturnsSetOfScalar)
+import PostgREST.SchemaCache.Table           (Column (..), Table (..),
+                                              TablesMap,
+                                              tableColumnsList,
+                                              tablePKCols)

 import PostgREST.ApiRequest.Preferences
 import PostgREST.ApiRequest.Types
@@ -197,26 +202,93 @@ findProc qi argumentsKeys paramsAsSingleObject allProcs contentMediaType isInvPo
 inspectPlanTxMode :: SQL.Mode
 inspectPlanTxMode = SQL.Read

+-- | During planning we need to resolve Field -> CoercibleField (finding the context specific target type and map function).
+-- | ResolverContext facilitates this without the need to pass around a laundry list of parameters.
+data ResolverContext = ResolverContext
+  { tables          :: TablesMap
+  , representations :: RepresentationsMap
+  , qi              :: QualifiedIdentifier  -- ^ The table we're currently attending; changes as we recurse into joins etc.
+  , outputType      :: Text                 -- ^ The output type for the response payload; e.g. "csv", "json", "binary".
+  }
+
+resolveColumnField :: Column -> CoercibleField
+resolveColumnField col = CoercibleField (colName col) mempty (colNominalType col) Nothing (colDefault col)
+
+resolveTableFieldName :: Table -> FieldName -> CoercibleField
+resolveTableFieldName table fieldName =
+  fromMaybe (unknownField fieldName []) $ HMI.lookup fieldName (tableColumns table) >>=
+    Just . resolveColumnField
+
+resolveTableField :: Table -> Field -> CoercibleField
+resolveTableField table (fieldName, []) = resolveTableFieldName table fieldName
+-- If the field is known and a JSON path is given, always assume the JSON type. But don't assume a type for entirely unknown fields.
+resolveTableField table (fieldName, jp) =
+  case resolveTableFieldName table fieldName of
+    cf@CoercibleField{cfIRType=""} -> cf{cfJsonPath=jp}
+    cf                             -> cf{cfJsonPath=jp, cfIRType="json"}
+
+-- | Resolve a type within the context based on the given field name and JSON path. Although there are situations where failure to resolve a field is considered an error (see `resolveOrError`), there are also situations where we allow it (RPC calls). If it should be an error and `resolveOrError` doesn't fit, ensure to check the `cfIRType` isn't empty.
+resolveTypeOrUnknown :: ResolverContext -> Field -> CoercibleField
+resolveTypeOrUnknown ResolverContext{..} field@(fn, jp) =
+  fromMaybe (unknownField fn jp) $ HM.lookup qi tables >>=
+    Just . flip resolveTableField field
+
+-- | Install any pre-defined data representation from source to target to coerce this reference.
+--
+-- Note that we change the IR type here. This might seem unintuitive. The short of it is that for a CoercibleField without a transformer, input type == output type. A transformer maps from a -> b, so by definition the input type will be a and the output type b after. And cfIRType is the *input* type.
+--
+-- It might feel odd that once a transformer is added we 'forget' the target type (because now a /= b). You might also note there's no obvious way to stack transforms (even if there was a stack, you erased what type you're working with so it's awkward). Alas as satisfying as it would be to engineer a layered mapping system with full type information, we just don't need it.
+withTransformer :: ResolverContext -> Text -> Text -> CoercibleField -> CoercibleField
+withTransformer ResolverContext{representations} sourceType targetType field =
+  fromMaybe field $ HM.lookup (sourceType, targetType) representations >>=
+    (\fieldRepresentation -> Just field{cfIRType=sourceType, cfTransform=Just (drFunction fieldRepresentation)})
+
+-- | Map the intermediate representation type to the output type, if available.
+withOutputFormat :: ResolverContext -> CoercibleField -> CoercibleField
+withOutputFormat ctx@ResolverContext{outputType} field@CoercibleField{cfIRType} = withTransformer ctx cfIRType outputType field
+
+-- | Map text into the intermediate representation type, if available.
+withTextParse :: ResolverContext -> CoercibleField -> CoercibleField
+withTextParse ctx field@CoercibleField{cfIRType} = withTransformer ctx "text" cfIRType field
+
+-- | Map json into the intermediate representation type, if available.
+withJsonParse :: ResolverContext -> CoercibleField -> CoercibleField
+withJsonParse ctx field@CoercibleField{cfIRType} = withTransformer ctx "json" cfIRType field
+
+-- | Map the intermediate representation type to the output type defined by the resolver context (normally json), if available.
+resolveOutputField :: ResolverContext -> Field -> CoercibleField
+resolveOutputField ctx field = withOutputFormat ctx $ resolveTypeOrUnknown ctx field
+
+-- | Map the query string format of a value (text) into the intermediate representation type, if available.
+resolveQueryInputField :: ResolverContext -> Field -> CoercibleField
+resolveQueryInputField ctx field = withTextParse ctx $ resolveTypeOrUnknown ctx field
+
 -- | Builds the ReadPlan tree on a number of stages.
 -- | Adds filters, order, limits on its respective nodes.
 -- | Adds joins conditions obtained from resource embedding.
 readPlan :: QualifiedIdentifier -> AppConfig -> SchemaCache -> ApiRequest -> Either Error ReadPlanTree
-readPlan qi@QualifiedIdentifier{..} AppConfig{configDbMaxRows} SchemaCache{dbRelationships} apiRequest  =
-  mapLeft ApiRequestError $
-  treeRestrictRange configDbMaxRows (iAction apiRequest) =<<
-  addNullEmbedFilters =<<
-  validateSpreadEmbeds =<<
-  addRelatedOrders =<<
-  addRels qiSchema (iAction apiRequest) dbRelationships Nothing =<<
-  addLogicTrees apiRequest =<<
-  addRanges apiRequest =<<
-  addOrders apiRequest =<<
-  addFilters apiRequest (initReadRequest qi $ QueryParams.qsSelect $ iQueryParams apiRequest)
+readPlan qi@QualifiedIdentifier{..} AppConfig{configDbMaxRows} SchemaCache{dbTables, dbRelationships, dbRepresentations} apiRequest  =
+  let
+    -- JSON output format hardcoded for now. In the future we might want to support other output mappings such as CSV.
+    ctx = ResolverContext dbTables dbRepresentations qi "json"
+  in
+    mapLeft ApiRequestError $
+    treeRestrictRange configDbMaxRows (iAction apiRequest) =<<
+    addNullEmbedFilters =<<
+    validateSpreadEmbeds =<<
+    addRelatedOrders =<<
+    addDataRepresentationAliases =<<
+    expandStarsForDataRepresentations ctx =<<
+    addRels qiSchema (iAction apiRequest) dbRelationships Nothing =<<
+    addLogicTrees ctx apiRequest =<<
+    addRanges apiRequest =<<
+    addOrders apiRequest =<<
+    addFilters ctx apiRequest (initReadRequest ctx $ QueryParams.qsSelect $ iQueryParams apiRequest)

 -- Build the initial read plan tree
-initReadRequest :: QualifiedIdentifier -> [Tree SelectItem] -> ReadPlanTree
-initReadRequest qi@QualifiedIdentifier{..} =
-  foldr (treeEntry rootDepth) $ Node defReadPlan{from=qi, relName=qiName, depth=rootDepth} []
+initReadRequest :: ResolverContext -> [Tree SelectItem] -> ReadPlanTree
+initReadRequest ctx@ResolverContext{qi=QualifiedIdentifier{..}} =
+  foldr (treeEntry rootDepth) $ Node defReadPlan{from=qi ctx, relName=qiName, depth=rootDepth} []
  where
    rootDepth = 0
    defReadPlan = ReadPlan [] (QualifiedIdentifier mempty mempty) Nothing [] [] allRange mempty Nothing [] Nothing mempty Nothing Nothing False rootDepth
@@ -235,7 +307,49 @@ initReadRequest qi@QualifiedIdentifier{..} =
            (Node defReadPlan{from=QualifiedIdentifier qiSchema selRelation, relName=selRelation, relHint=selHint, relJoinType=selJoinType, depth=nxtDepth, relIsSpread=True} [])
            fldForest:rForest
        SelectField{..} ->
-          Node q{select=(selField, selCast, selAlias):select q} rForest
+          Node q{select=(resolveOutputField ctx{qi=from q} selField, selCast, selAlias):select q} rForest
+
+-- | Preserve the original field name if data representation is used to coerce the value.
+addDataRepresentationAliases :: ReadPlanTree -> Either ApiRequestError ReadPlanTree
+addDataRepresentationAliases rPlanTree = Right $ fmap (\rPlan@ReadPlan{select=sel} -> rPlan{select=map aliasSelectItem sel}) rPlanTree
+  where
+    aliasSelectItem :: (CoercibleField, Maybe Cast, Maybe Alias) -> (CoercibleField, Maybe Cast, Maybe Alias)
+    -- If there already is an alias, don't overwrite it.
+    aliasSelectItem (fld@(CoercibleField{cfName=fieldName, cfTransform=(Just _)}), Nothing, Nothing) = (fld, Nothing, Just fieldName)
+    aliasSelectItem fld = fld
+
+knownColumnsInContext :: ResolverContext -> [Column]
+knownColumnsInContext ResolverContext{..} =
+  fromMaybe [] $ HM.lookup qi tables >>=
+  Just . tableColumnsList
+
+-- | Expand "select *" into explicit field names of the table, if necessary to apply data representations.
+expandStarsForDataRepresentations :: ResolverContext -> ReadPlanTree -> Either ApiRequestError ReadPlanTree
+expandStarsForDataRepresentations ctx@ResolverContext{qi} rPlanTree = Right $ fmap expandStars rPlanTree
+  where
+    expandStars :: ReadPlan -> ReadPlan
+    -- When the schema is "" and the table is the source CTE, we assume the true source table is given in the from
+    -- alias and belongs to the request schema. See the bit in `addRels` with `newFrom = ...`.
+    expandStars rPlan@ReadPlan{from=(QualifiedIdentifier "" "pgrst_source"), fromAlias=(Just tblAlias)} =
+      expandStarsForTable ctx{qi=qi{qiName=tblAlias}} rPlan
+    expandStars rPlan@ReadPlan{from=fromTable} =
+      expandStarsForTable ctx{qi=fromTable} rPlan
+
+expandStarsForTable :: ResolverContext -> ReadPlan -> ReadPlan
+expandStarsForTable ctx@ResolverContext{representations, outputType} rplan@ReadPlan{select=selectItems} =
+  -- If we have a '*' select AND the target table has at least one data representation, expand.
+  if ("*" `elem` map (\(field, _, _) -> cfName field) selectItems) && any hasOutputRep knownColumns
+    then rplan{select=concatMap (expandStarSelectItem knownColumns) selectItems}
+    else rplan
+  where
+    knownColumns = knownColumnsInContext ctx
+
+    hasOutputRep :: Column -> Bool
+    hasOutputRep col = HM.member (colNominalType col, outputType) representations
+
+    expandStarSelectItem :: [Column] -> (CoercibleField, Maybe Cast, Maybe Alias) -> [(CoercibleField, Maybe Cast, Maybe Alias)]
+    expandStarSelectItem columns (CoercibleField{cfName="*", cfJsonPath=[]}, b, c) = map (\col -> (withOutputFormat ctx $ resolveColumnField col, b, c)) columns
+    expandStarSelectItem _ selectItem = [selectItem]

 -- | Enforces the `max-rows` config on the result
 treeRestrictRange :: Maybe Integer -> Action -> ReadPlanTree -> Either ApiRequestError ReadPlanTree
@@ -390,8 +504,8 @@ findRel schema allRels origin target hint =
            )
      ) $ fromMaybe mempty $ HM.lookup (QualifiedIdentifier schema origin, schema) allRels

-addFilters :: ApiRequest -> ReadPlanTree -> Either ApiRequestError ReadPlanTree
-addFilters ApiRequest{..} rReq =
+addFilters :: ResolverContext -> ApiRequest -> ReadPlanTree -> Either ApiRequestError ReadPlanTree
+addFilters ctx ApiRequest{..} rReq =
  foldr addFilterToNode (Right rReq) flts
  where
    QueryParams.QueryParams{..} = iQueryParams
@@ -403,7 +517,7 @@ addFilters ApiRequest{..} rReq =

    addFilterToNode :: (EmbedPath, Filter) -> Either ApiRequestError ReadPlanTree ->  Either ApiRequestError ReadPlanTree
    addFilterToNode =
-      updateNode (\flt (Node q@ReadPlan{where_=lf} f) -> Node q{ReadPlan.where_=addFilterToLogicForest flt lf}  f)
+      updateNode (\flt (Node q@ReadPlan{from=fromTable, where_=lf} f) -> Node q{ReadPlan.where_=addFilterToLogicForest (resolveFilter ctx{qi=fromTable} flt) lf}  f)

 addOrders :: ApiRequest -> ReadPlanTree -> Either ApiRequestError ReadPlanTree
 addOrders ApiRequest{..} rReq =
@@ -447,15 +561,15 @@ addNullEmbedFilters (Node rp@ReadPlan{where_=oldLogic} forest) = do
  newLogic <- getFilters readPlans `traverse` oldLogic
  Node rp{ReadPlan.where_= newLogic} <$> (addNullEmbedFilters `traverse` forest)
  where
-    getFilters :: [ReadPlan] -> LogicTree -> Either ApiRequestError LogicTree
-    getFilters rPlans (Expr b lOp trees) = Expr b lOp <$> (getFilters rPlans `traverse` trees)
-    getFilters rPlans flt@(Stmnt (Filter (fld, []) opExpr)) =
+    getFilters :: [ReadPlan] -> CoercibleLogicTree -> Either ApiRequestError CoercibleLogicTree
+    getFilters rPlans (CoercibleExpr b lOp trees) = CoercibleExpr b lOp <$> (getFilters rPlans `traverse` trees)
+    getFilters rPlans flt@(CoercibleStmnt (CoercibleFilter (CoercibleField fld [] _ _ _) opExpr)) =
      let foundRP = find (\ReadPlan{relName, relAlias} -> fld == fromMaybe relName relAlias) rPlans in
      case (foundRP, opExpr) of
-        (Just ReadPlan{relAggAlias}, OpExpr b (Is TriNull)) -> Right $ Stmnt $ FilterNullEmbed b relAggAlias
+        (Just ReadPlan{relAggAlias}, OpExpr b (Is TriNull)) -> Right $ CoercibleStmnt $ CoercibleFilterNullEmbed b relAggAlias
        (Just ReadPlan{relName}, _)                         -> Left $ UnacceptableFilter relName
        _                                                   -> Right flt
-    getFilters _ flt@(Stmnt _)        = Right flt
+    getFilters _ flt@(CoercibleStmnt _)        = Right flt

 addRanges :: ApiRequest -> ReadPlanTree -> Either ApiRequestError ReadPlanTree
 addRanges ApiRequest{..} rReq =
@@ -469,14 +583,22 @@ addRanges ApiRequest{..} rReq =
    addRangeToNode :: (EmbedPath, NonnegRange) -> Either ApiRequestError ReadPlanTree -> Either ApiRequestError ReadPlanTree
    addRangeToNode = updateNode (\r (Node q f) -> Node q{range_=r} f)

-addLogicTrees :: ApiRequest -> ReadPlanTree -> Either ApiRequestError ReadPlanTree
-addLogicTrees ApiRequest{..} rReq =
+addLogicTrees :: ResolverContext -> ApiRequest -> ReadPlanTree -> Either ApiRequestError ReadPlanTree
+addLogicTrees ctx ApiRequest{..} rReq =
  foldr addLogicTreeToNode (Right rReq) qsLogic
  where
    QueryParams.QueryParams{..} = iQueryParams

    addLogicTreeToNode :: (EmbedPath, LogicTree) -> Either ApiRequestError ReadPlanTree -> Either ApiRequestError ReadPlanTree
-    addLogicTreeToNode = updateNode (\t (Node q@ReadPlan{where_=lf} f) -> Node q{ReadPlan.where_=t:lf} f)
+    addLogicTreeToNode = updateNode (\t (Node q@ReadPlan{from=fromTable, where_=lf} f) -> Node q{ReadPlan.where_=resolveLogicTree ctx{qi=fromTable} t:lf} f)
+
+resolveLogicTree :: ResolverContext -> LogicTree -> CoercibleLogicTree
+resolveLogicTree ctx (Stmnt flt) = CoercibleStmnt $ resolveFilter ctx flt
+resolveLogicTree ctx (Expr b op lts) = CoercibleExpr b op (map (resolveLogicTree ctx) lts)
+
+resolveFilter :: ResolverContext -> Filter -> CoercibleFilter
+resolveFilter ctx (Filter fld opExpr) = CoercibleFilter{field=resolveQueryInputField ctx fld, opExpr=opExpr}
+resolveFilter _ (FilterNullEmbed isNot fieldName) = CoercibleFilterNullEmbed isNot fieldName

 -- Validates that spread embeds are only done on to-one relationships
 validateSpreadEmbeds :: ReadPlanTree -> Either ApiRequestError ReadPlanTree
@@ -502,7 +624,7 @@ updateNode f (targetNodeName:remainingPath, a) (Right (Node rootNode forest)) =
    findNode = find (\(Node ReadPlan{relName, relAlias} _) -> relName == targetNodeName || relAlias == Just targetNodeName) forest

 mutatePlan :: Mutation -> QualifiedIdentifier -> ApiRequest -> SchemaCache -> ReadPlanTree -> Either Error MutatePlan
-mutatePlan mutation qi ApiRequest{iPreferences=Preferences{..}, ..} sCache readReq = mapLeft ApiRequestError $
+mutatePlan mutation qi ApiRequest{iPreferences=Preferences{..}, ..} SchemaCache{dbTables, dbRepresentations} readReq = mapLeft ApiRequestError $
  case mutation of
    MutationCreate ->
      mapRight (\typedColumns -> Insert qi typedColumns body ((,) <$> preferResolution <*> Just confCols) [] returnings pkCols applyDefaults) typedColumnsOrError
@@ -520,27 +642,28 @@ mutatePlan mutation qi ApiRequest{iPreferences=Preferences{..}, ..} sCache readR
          Left InvalidFilters
    MutationDelete -> Right $ Delete qi combinedLogic iTopLevelRange rootOrder returnings
  where
+    ctx = ResolverContext dbTables dbRepresentations qi "json"
    confCols = fromMaybe pkCols qsOnConflict
    QueryParams.QueryParams{..} = iQueryParams
    returnings =
      if preferRepresentation == None
        then []
        else inferColsEmbedNeeds readReq pkCols
-    pkCols = maybe mempty tablePKCols $ HM.lookup qi $ dbTables sCache
-    logic = map snd qsLogic
+    tbl = HM.lookup qi dbTables
+    pkCols = maybe mempty tablePKCols tbl
+    logic = map (resolveLogicTree ctx . snd) qsLogic
    rootOrder = maybe [] snd $ find (\(x, _) -> null x) qsOrder
-    combinedLogic = foldr addFilterToLogicForest logic qsFiltersRoot
+    combinedLogic = foldr (addFilterToLogicForest . resolveFilter ctx) logic qsFiltersRoot
    body = payRaw <$> iPayload -- the body is assumed to be json at this stage(ApiRequest validates)
-    tbl = HM.lookup qi $ dbTables sCache
-    typedColumnsOrError = resolveOrError tbl `traverse` S.toList iColumns
    applyDefaults = preferMissing == Just ApplyDefaults
+    typedColumnsOrError = resolveOrError ctx tbl `traverse` S.toList iColumns

-resolveOrError :: Maybe Table -> FieldName -> Either ApiRequestError TypedField
-resolveOrError Nothing _ = Left NotFound
-resolveOrError (Just table) field =
-  case resolveTableField table field of
-    Nothing         -> Left $ ColumnNotFound (tableName table) field
-    Just typedField -> Right typedField
+resolveOrError :: ResolverContext -> Maybe Table -> FieldName -> Either ApiRequestError CoercibleField
+resolveOrError _ Nothing _ = Left NotFound
+resolveOrError ctx (Just table) field =
+  case resolveTableFieldName table field of
+    CoercibleField{cfIRType=""} -> Left $ ColumnNotFound (tableName table) field
+    cf                          -> Right $ withJsonParse ctx cf

 callPlan :: Routine -> ApiRequest -> S.Set FieldName -> LBS.ByteString -> ReadPlanTree -> CallPlan
 callPlan proc ApiRequest{iPreferences=Preferences{..}} paramKeys args readReq = FunctionCall {
@@ -569,7 +692,7 @@ inferColsEmbedNeeds (Node ReadPlan{select} forest) pkCols
  | "*" `elem` fldNames = ["*"]
  | otherwise           = returnings
  where
-    fldNames = (\((fld, _), _, _) -> fld) <$> select
+    fldNames = cfName . (\(f, _, _) -> f) <$> select
    -- Without fkCols, when a mutatePlan to
    -- /projects?select=name,clients(name) occurs, the RETURNING SQL part would
    -- be `RETURNING name`(see QueryBuilder).  This would make the embedding
@@ -608,8 +731,8 @@ inferColsEmbedNeeds (Node ReadPlan{select} forest) pkCols

 -- Traditional filters(e.g. id=eq.1) are added as root nodes of the LogicTree
 -- they are later concatenated with AND in the QueryBuilder
-addFilterToLogicForest :: Filter -> [LogicTree] -> [LogicTree]
-addFilterToLogicForest flt lf = Stmnt flt : lf
+addFilterToLogicForest :: CoercibleFilter -> [CoercibleLogicTree] -> [CoercibleLogicTree]
+addFilterToLogicForest flt lf = CoercibleStmnt flt : lf

 -- | If raw(binary) output is requested, check that MediaType is one of the
 -- admitted rawMediaTypes and that`?select=...` contains only one field other
@@ -638,6 +761,6 @@ binaryField AppConfig{configRawMediaTypes} acceptMediaType proc rpTree
      _                        -> False

    fstFieldName :: ReadPlanTree -> Maybe FieldName
-    fstFieldName (Node ReadPlan{select=(("*", []), _, _):_} [])  = Nothing
-    fstFieldName (Node ReadPlan{select=[((fld, []), _, _)]} []) = Just fld
+    fstFieldName (Node ReadPlan{select=(CoercibleField{cfName="*", cfJsonPath=[]}, _, _):_} [])  = Nothing
+    fstFieldName (Node ReadPlan{select=[(CoercibleField{cfName=fld, cfJsonPath=[]}, _, _)]} [])  = Just fld
    fstFieldName _                                               = Nothing
@@ -6,8 +6,9 @@ where
 import qualified Data.ByteString.Lazy as LBS

 import PostgREST.ApiRequest.Preferences  (PreferResolution)
-import PostgREST.ApiRequest.Types        (LogicTree, OrderTerm)
-import PostgREST.Plan.Types              (TypedField)
+import PostgREST.ApiRequest.Types        (OrderTerm)
+import PostgREST.Plan.Types              (CoercibleField,
+                                          CoercibleLogicTree)
 import PostgREST.RangeQuery              (NonnegRange)
 import PostgREST.SchemaCache.Identifiers (FieldName,
                                          QualifiedIdentifier)
@@ -18,19 +19,19 @@ import Protolude
 data MutatePlan
  = Insert
      { in_        :: QualifiedIdentifier
-      , insCols    :: [TypedField]
+      , insCols    :: [CoercibleField]
      , insBody    :: Maybe LBS.ByteString
      , onConflict :: Maybe (PreferResolution, [FieldName])
-      , where_     :: [LogicTree]
+      , where_     :: [CoercibleLogicTree]
      , returning  :: [FieldName]
      , insPkCols  :: [FieldName]
      , applyDefs  :: Bool
      }
  | Update
      { in_       :: QualifiedIdentifier
-      , updCols   :: [TypedField]
+      , updCols   :: [CoercibleField]
      , updBody   :: Maybe LBS.ByteString
-      , where_    :: [LogicTree]
+      , where_    :: [CoercibleLogicTree]
      , mutRange  :: NonnegRange
      , mutOrder  :: [OrderTerm]
      , returning :: [FieldName]
@@ -38,7 +39,7 @@ data MutatePlan
      }
  | Delete
      { in_       :: QualifiedIdentifier
-      , where_    :: [LogicTree]
+      , where_    :: [CoercibleLogicTree]
      , mutRange  :: NonnegRange
      , mutOrder  :: [OrderTerm]
      , returning :: [FieldName]
@@ -6,9 +6,11 @@ module PostgREST.Plan.ReadPlan

 import Data.Tree (Tree (..))

-import PostgREST.ApiRequest.Types         (Alias, Cast, Depth, Field,
-                                           Hint, JoinType, LogicTree,
-                                           NodeName, OrderTerm)
+import PostgREST.ApiRequest.Types         (Alias, Cast, Depth, Hint,
+                                           JoinType, NodeName,
+                                           OrderTerm)
+import PostgREST.Plan.Types               (CoercibleField (..),
+                                           CoercibleLogicTree)
 import PostgREST.RangeQuery               (NonnegRange)
 import PostgREST.SchemaCache.Identifiers  (FieldName,
                                           QualifiedIdentifier)
@@ -26,10 +28,10 @@ data JoinCondition =
  deriving (Eq)

 data ReadPlan = ReadPlan
-  { select       :: [(Field, Maybe Cast, Maybe Alias)]
+  { select       :: [(CoercibleField, Maybe Cast, Maybe Alias)]
  , from         :: QualifiedIdentifier
  , fromAlias    :: Maybe Alias
-  , where_       :: [LogicTree]
+  , where_       :: [CoercibleLogicTree]
  , order        :: [OrderTerm]
  , range_       :: NonnegRange
  , relName      :: NodeName
@@ -1,24 +1,50 @@
 module PostgREST.Plan.Types
-  ( TypedField(..)
-  , resolveTableField
+  ( CoercibleField(..)
+  , unknownField
+  , CoercibleLogicTree(..)
+  , CoercibleFilter(..)
+  , TransformerProc
  ) where

-import qualified Data.HashMap.Strict.InsOrd as HMI
+import PostgREST.ApiRequest.Types (JsonPath, LogicOperator, OpExpr)

 import PostgREST.SchemaCache.Identifiers (FieldName)
-import PostgREST.SchemaCache.Table       (Column (..), Table (..))

 import Protolude

-- | A TypedField is a field with sufficient information to be read from JSON with `json_to_recordset`.
-data TypedField = TypedField
-   { tfName    :: FieldName
-   , tfIRType  :: Text -- ^ The initial type of the field, before any casting.
-   , tfDefault :: Maybe Text
-   } deriving (Eq)
+type TransformerProc = Text

-resolveTableField :: Table -> FieldName -> Maybe TypedField
-resolveTableField table fieldName =
-  case HMI.lookup fieldName (tableColumns table) of
-    Just column -> Just $ TypedField (colName column) (colNominalType column) (colDefault column)
-    Nothing     -> Nothing
+-- | A CoercibleField pairs the name of a query element with any type coercion information we need for some specific use case.
+-- |
+-- | As suggested by the name, it's often a reference to a field in a table but really it can be any nameable element (function parameter, calculation with an alias, etc) with a knowable type.
+-- |
+-- | In the simplest case, it allows us to parse JSON payloads with `json_to_recordset`, for which we need to know both the name and the type of each thing we'd like to extract. At a higher level, CoercibleField generalises to reflect that any value we work with in a query may need type specific handling.
+-- |
+-- | CoercibleField is the foundation for the Data Representations feature. This feature allow user-definable mappings between database types so that the same data can be presented or interpreted in various ways as needed. Sometimes the way Postgres coerces data implicitly isn't right for the job. Different mappings might be appropriate for different situations: parsing a filter from a query string requires one function (text -> field type) while parsing a payload from JSON takes another (json -> field type). And the reverse, outputting a field as JSON, requires yet a third (field type -> json). CoercibleField is that "job specific" reference to an element paired with the type we desire for that particular purpose and the function we'll use to get there, if any.
+-- |
+-- | In the planning phase, we "resolve" generic named elements into these specialised CoercibleFields. Again this is context specific: two different CoercibleFields both representing the exact same table column in the database, even in the same query, might have two different target types and mapping functions. For example, one might represent a column in a filter, and another the very same column in an output role to be sent in the response body.
+-- |
+-- | The type value is allowed to be the empty string. The analog here is soft type checking in programming languages: sometimes we don't need a variable to have a specified type and things will work anyhow. So the empty type variant is valid when we don't know and *don't need to know* about the specific type in some context. Note that this variation should not be used if it guarantees failure: in that case you should instead raise an error at the planning stage and bail out. For example, we can't parse JSON with `json_to_recordset` without knowing the types of each recipient field, and so error out. Using the empty string for the type would be incorrect and futile. On the other hand we use the empty type for RPC calls since type resolution isn't implemented for RPC, but it's fine because the query still works with Postgres' implicit coercion. In the future, hopefully we will support data representations across the board and then the empty type may be permanently retired.
+data CoercibleField = CoercibleField
+  { cfName      :: FieldName
+  , cfJsonPath  :: JsonPath
+  , cfIRType    :: Text                  -- ^ The native Postgres type of the field, the intermediate (IR) type before mapping.
+  , cfTransform :: Maybe TransformerProc -- ^ The optional mapping from irType -> targetType.
+  , cfDefault   :: Maybe Text
+  } deriving Eq
+
+unknownField :: FieldName -> JsonPath -> CoercibleField
+unknownField name path = CoercibleField name path "" Nothing Nothing
+
+-- | Like an API request LogicTree, but with coercible field information.
+data CoercibleLogicTree
+  = CoercibleExpr Bool LogicOperator [CoercibleLogicTree]
+  | CoercibleStmnt CoercibleFilter
+  deriving (Eq)
+
+data CoercibleFilter = CoercibleFilter
+  { field  :: CoercibleField
+  , opExpr :: OpExpr
+  }
+  | CoercibleFilterNullEmbed Bool FieldName
+  deriving (Eq)
@@ -55,7 +55,7 @@ readPlanToQuery (Node ReadPlan{select,from=mainQi,fromAlias,where_=logicForest,o
  where
    fromFrag = fromF relToParent mainQi fromAlias
    qi = getQualifiedIdentifier relToParent mainQi fromAlias
-    defSelect = [(("*", []), Nothing, Nothing)] -- gets all the columns in case of an empty select, ignoring/obtaining these columns is done at the aggregation stage
+    defSelect = [(unknownField "*" [], Nothing, Nothing)] -- gets all the columns in case of an empty select, ignoring/obtaining these columns is done at the aggregation stage
    (selects, joins) = foldr getSelectsJoins ([],[]) forest

 getSelectsJoins :: ReadPlanTree -> ([SQL.Snippet], [SQL.Snippet]) -> ([SQL.Snippet], [SQL.Snippet])
@@ -98,11 +98,11 @@ mutatePlanToQuery (Insert mainQi iCols body onConflct putConditions returnings _
      MergeDuplicates  ->
        if null iCols
           then "DO NOTHING"
-           else "DO UPDATE SET " <> intercalateSnippet ", " ((pgFmtIdent . tfName) <> const " = EXCLUDED." <> (pgFmtIdent . tfName) <$> iCols)
+           else "DO UPDATE SET " <> intercalateSnippet ", " ((pgFmtIdent . cfName) <> const " = EXCLUDED." <> (pgFmtIdent . cfName) <$> iCols)
    ) onConflct <> " " <>
  returningF mainQi returnings
  where
-    cols = intercalateSnippet ", " $ pgFmtIdent . tfName <$> iCols
+    cols = intercalateSnippet ", " $ pgFmtIdent . cfName <$> iCols

 -- An update without a limit is always filtered with a WHERE
 mutatePlanToQuery (Update mainQi uCols body logicForest range ordts returnings applyDefaults)
@@ -136,8 +136,8 @@ mutatePlanToQuery (Update mainQi uCols body logicForest range ordts returnings a
    whereLogic = if null logicForest then mempty else " WHERE " <> intercalateSnippet " AND " (pgFmtLogicTree mainQi <$> logicForest)
    mainTbl = fromQi mainQi
    emptyBodyReturnedColumns = if null returnings then "NULL" else intercalateSnippet ", " (pgFmtColumn (QualifiedIdentifier mempty $ qiName mainQi) <$> returnings)
-    nonRangeCols = intercalateSnippet ", " (pgFmtIdent . tfName <> const " = " <> pgFmtColumn (QualifiedIdentifier mempty "pgrst_body") . tfName <$> uCols)
-    rangeCols = intercalateSnippet ", " ((\col -> pgFmtIdent (tfName col) <> " = (SELECT " <> pgFmtIdent (tfName col) <> " FROM pgrst_update_body) ") <$> uCols)
+    nonRangeCols = intercalateSnippet ", " (pgFmtIdent . cfName <> const " = " <> pgFmtColumn (QualifiedIdentifier mempty "pgrst_body") . cfName <$> uCols)
+    rangeCols = intercalateSnippet ", " ((\col -> pgFmtIdent (cfName col) <> " = (SELECT " <> pgFmtIdent (cfName col) <> " FROM pgrst_update_body) ") <$> uCols)
    (whereRangeIdF, rangeIdF) = mutRangeF mainQi (fst . otTerm <$> ordts)

 mutatePlanToQuery (Delete mainQi logicForest range ordts returnings)
@@ -171,7 +171,7 @@ callPlanToQuery (FunctionCall qi params args returnsScalar returnsSetOfScalar re
    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 True False <> ", " <>
+      KeyParams prms  -> fromJsonBodyF args ((\p -> CoercibleField (ppName p) mempty (ppType p) Nothing Nothing) <$> prms) False True False <> ", " <>
                         "LATERAL " <> callIt (fmtParams prms)

    callIt :: SQL.Snippet -> SQL.Snippet
@@ -55,14 +55,13 @@ import Control.Arrow ((***))
 import Data.Foldable                 (foldr1)
 import Text.InterpolatedString.Perl6 (qc)

-import PostgREST.ApiRequest.Types        (Alias, Cast, Field,
-                                          Filter (..),
+import PostgREST.ApiRequest.Types        (Alias, Cast,
                                          FtsOperator (..),
                                          JsonOperand (..),
                                          JsonOperation (..),
                                          JsonPath,
                                          LogicOperator (..),
-                                          LogicTree (..), OpExpr (..),
+                                          OpExpr (..),
                                          OpQuantifier (..),
                                          Operation (..),
                                          OrderDirection (..),
@@ -74,7 +73,10 @@ import PostgREST.ApiRequest.Types        (Alias, Cast, Field,
 import PostgREST.MediaType               (MTPlanFormat (..),
                                          MTPlanOption (..))
 import PostgREST.Plan.ReadPlan           (JoinCondition (..))
-import PostgREST.Plan.Types              (TypedField (..))
+import PostgREST.Plan.Types              (CoercibleField (..),
+                                          CoercibleFilter (..),
+                                          CoercibleLogicTree (..),
+                                          unknownField)
 import PostgREST.RangeQuery              (NonnegRange, allRange,
                                          rangeLimit, rangeOffset)
 import PostgREST.SchemaCache.Identifiers (FieldName,
@@ -235,23 +237,36 @@ pgFmtColumn :: QualifiedIdentifier -> Text -> SQL.Snippet
 pgFmtColumn table "*" = fromQi table <> ".*"
 pgFmtColumn table c   = fromQi table <> "." <> pgFmtIdent c

-pgFmtField :: QualifiedIdentifier -> Field -> SQL.Snippet
-pgFmtField table (c, []) = pgFmtColumn table c
+pgFmtCallUnary :: Text -> SQL.Snippet -> SQL.Snippet
+pgFmtCallUnary f x = SQL.sql (encodeUtf8 f) <> "(" <> x <> ")"
+
+pgFmtField :: QualifiedIdentifier -> CoercibleField -> SQL.Snippet
+pgFmtField table CoercibleField{cfName=fn, cfJsonPath=[]} = pgFmtColumn table fn
 -- Using to_jsonb instead of to_json to avoid missing operator errors when filtering:
 -- "operator does not exist: json = unknown"
-pgFmtField table (c, jp) = "to_jsonb(" <> pgFmtColumn table c <> ")" <> pgFmtJsonPath jp
+pgFmtField table CoercibleField{cfName=fn, cfJsonPath=jp} = "to_jsonb(" <> pgFmtColumn table fn <> ")" <> pgFmtJsonPath jp

-pgFmtSelectItem :: QualifiedIdentifier -> (Field, Maybe Cast, Maybe Alias) -> SQL.Snippet
-pgFmtSelectItem table (f@(fName, jp), Nothing, alias) = pgFmtField table f <> pgFmtAs fName jp alias
+-- Select the value of a named element from a table, applying its optional coercion mapping if any.
+pgFmtTableCoerce :: QualifiedIdentifier -> CoercibleField -> SQL.Snippet
+pgFmtTableCoerce table fld@(CoercibleField{cfTransform=(Just formatterProc)}) = pgFmtCallUnary formatterProc (pgFmtField table fld)
+pgFmtTableCoerce table f = pgFmtField table f
+
+-- | Like the previous but now we just have a name so no namespace or JSON paths.
+pgFmtCoerceNamed :: CoercibleField -> SQL.Snippet
+pgFmtCoerceNamed CoercibleField{cfName=fn, cfTransform=(Just formatterProc)} = pgFmtCallUnary formatterProc (pgFmtIdent fn) <> " AS " <> pgFmtIdent fn
+pgFmtCoerceNamed CoercibleField{cfName=fn} = pgFmtIdent fn
+
+pgFmtSelectItem :: QualifiedIdentifier -> (CoercibleField, Maybe Cast, Maybe Alias) -> SQL.Snippet
+pgFmtSelectItem table (fld, Nothing, alias) = pgFmtTableCoerce table fld <> pgFmtAs (cfName fld) (cfJsonPath fld) alias
 -- Ideally we'd quote the cast with "pgFmtIdent cast". However, that would invalidate common casts such as "int", "bigint", etc.
 -- Try doing: `select 1::"bigint"` - it'll err, using "int8" will work though. There's some parser magic that pg does that's invalidated when quoting.
 -- Not quoting should be fine, we validate the input on Parsers.
-pgFmtSelectItem table (f@(fName, jp), Just cast, alias) = "CAST (" <> pgFmtField table f <> " AS " <> SQL.sql (encodeUtf8 cast) <> " )" <> pgFmtAs fName jp alias
+pgFmtSelectItem table (fld, Just cast, alias) = "CAST (" <> pgFmtTableCoerce table fld <> " AS " <> SQL.sql (encodeUtf8 cast) <> " )" <> pgFmtAs (cfName fld) (cfJsonPath fld) 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 :: Maybe LBS.ByteString -> [CoercibleField] -> Bool -> Bool -> Bool -> SQL.Snippet
 fromJsonBodyF body fields includeSelect includeLimitOne includeDefaults =
-  (if includeSelect then "SELECT " <> parsedCols <> " " else mempty) <>
+  (if includeSelect then "SELECT " <> namedCols <> " " else mempty) <>
  "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
@@ -260,7 +275,7 @@ fromJsonBodyF body fields includeSelect includeLimitOne includeDefaults =
  (if includeDefaults
    then "LATERAL (SELECT jsonb_agg(jsonb_build_object(" <> defsJsonb <> ") || elem) AS val from jsonb_array_elements(pgrst_uniform_json.val) elem) pgrst_json_defs, "
    else mempty) <>
-  "LATERAL (SELECT * FROM " <>
+  "LATERAL (SELECT " <> parsedCols <> " FROM " <>
    (if null fields
      -- When we are inserting no columns (e.g. using default values), we can't use our ordinary `json_to_recordset`
      -- because it can't extract records with no columns (there's no valid syntax for the `AS (colName colType,...)`
@@ -270,12 +285,13 @@ fromJsonBodyF body fields includeSelect includeLimitOne includeDefaults =
    ) <>
  ") pgrst_body "
  where
-    parsedCols = intercalateSnippet ", " $ fromQi  . QualifiedIdentifier "pgrst_body" . tfName <$> fields
-    typedCols = intercalateSnippet ", " $ pgFmtIdent . tfName <> const " " <> SQL.sql . encodeUtf8 . tfIRType <$> fields
+    namedCols = intercalateSnippet ", " $ fromQi  . QualifiedIdentifier "pgrst_body" . cfName <$> fields
+    parsedCols = intercalateSnippet ", " $ pgFmtCoerceNamed <$> fields
+    typedCols = intercalateSnippet ", " $ pgFmtIdent . cfName <> const " " <> SQL.sql . encodeUtf8 . cfIRType <$> fields
    defsJsonb = SQL.sql $ BS.intercalate "," fieldsWDefaults
    fieldsWDefaults = mapMaybe (\case
-        TypedField{tfName=nam, tfDefault=Just def} -> Just $ encodeUtf8 (pgFmtLit nam <> ", " <> def)
-        TypedField{tfDefault=Nothing} -> Nothing
+        CoercibleField{cfName=nam, cfDefault=Just def} -> Just $ encodeUtf8 (pgFmtLit nam <> ", " <> def)
+        CoercibleField{cfDefault=Nothing} -> Nothing
      ) fields
    (finalBodyF, jsonTypeofF, jsonBuildArrayF, jsonArrayElementsF, jsonToRecordsetF) =
      if includeDefaults
@@ -291,8 +307,8 @@ pgFmtOrderTerm qi ot =
    maybe mempty nullOrder $ otNullOrder ot])
  where
    fmtOTerm = \case
-      OrderTerm{otTerm}                        -> pgFmtField qi otTerm
-      OrderRelationTerm{otRelation, otRelTerm} -> pgFmtField (QualifiedIdentifier mempty otRelation) otRelTerm
+      OrderTerm{otTerm=(fn, jp)}                        -> pgFmtField qi (unknownField fn jp)
+      OrderRelationTerm{otRelation, otRelTerm=(fn, jp)} -> pgFmtField (QualifiedIdentifier mempty otRelation) (unknownField fn jp)

    direction OrderAsc  = "ASC"
    direction OrderDesc = "DESC"
@@ -300,17 +316,31 @@ pgFmtOrderTerm qi ot =
    nullOrder OrderNullsFirst = "NULLS FIRST"
    nullOrder OrderNullsLast  = "NULLS LAST"

+-- | Interpret a literal in the way the planner indicated through the CoercibleField.
+pgFmtUnknownLiteralForField :: SQL.Snippet -> CoercibleField -> SQL.Snippet
+pgFmtUnknownLiteralForField value CoercibleField{cfTransform=(Just parserProc)} = pgFmtCallUnary parserProc value
+-- But when no transform is requested, we just use the literal as-is.
+pgFmtUnknownLiteralForField value _ = value

-pgFmtFilter :: QualifiedIdentifier -> Filter -> SQL.Snippet
-pgFmtFilter _ (FilterNullEmbed hasNot fld) = pgFmtIdent fld <> " IS " <> (if hasNot then "NOT" else mempty) <> " NULL"
-pgFmtFilter _ (Filter _ (NoOpExpr _)) = mempty -- TODO unreachable because NoOpExpr is filtered on QueryParams
-pgFmtFilter table (Filter fld (OpExpr hasNot oper)) = notOp <> " " <> pgFmtField table fld <> case oper of
-   Op op val -> " " <> simpleOperator op <> " " <> unknownLiteral val
+-- | Array version of the above, used by ANY().
+pgFmtArrayLiteralForField :: [Text] -> CoercibleField -> SQL.Snippet
+-- When a transformation is requested, we need to apply the transformation to each element of the array. This could be done by just making a query with `parser(value)` for each value, but may lead to huge query lengths. Imagine `data_representations.color_from_text('...'::text)` for repeated for a hundred values. Instead we use `unnest()` to unpack a standard array literal and then apply the transformation to each element, like a map.
+-- Note the literals will be treated as text since in every case when we use ANY() the parameters are textual (coming from a query string). We want to rely on the `text->domain` parser to do the right thing.
+pgFmtArrayLiteralForField values CoercibleField{cfTransform=(Just parserProc)} = SQL.sql "(SELECT " <> pgFmtCallUnary parserProc (SQL.sql "unnest(" <> unknownLiteral (pgBuildArrayLiteral values) <> "::text[])") <> ")"
+-- When no transformation is requested, we don't need a subquery.
+pgFmtArrayLiteralForField values _ = unknownLiteral (pgBuildArrayLiteral values)
+
+
+pgFmtFilter :: QualifiedIdentifier -> CoercibleFilter -> SQL.Snippet
+pgFmtFilter _ (CoercibleFilterNullEmbed hasNot fld) = pgFmtIdent fld <> " IS " <> (if hasNot then "NOT" else mempty) <> " NULL"
+pgFmtFilter _ (CoercibleFilter _ (NoOpExpr _)) = mempty -- TODO unreachable because NoOpExpr is filtered on QueryParams
+pgFmtFilter table (CoercibleFilter fld (OpExpr hasNot oper)) = notOp <> " " <> pgFmtField table fld <> case oper of
+   Op op val  -> " " <> simpleOperator op <> " " <> pgFmtUnknownLiteralForField (unknownLiteral val) fld

   OpQuant op quant val -> " " <> quantOperator op <> " " <> case op of
     OpLike  -> fmtQuant quant $ unknownLiteral (T.map star val)
     OpILike -> fmtQuant quant $ unknownLiteral (T.map star val)
-     _       -> fmtQuant quant $ unknownLiteral val
+     _       -> fmtQuant quant $ pgFmtUnknownLiteralForField (unknownLiteral val) fld

   -- IS cannot be prepared. `PREPARE boolplan AS SELECT * FROM projects where id IS $1` will give a syntax error.
   -- The above can be fixed by using `PREPARE boolplan AS SELECT * FROM projects where id IS NOT DISTINCT FROM $1;`
@@ -329,7 +359,7 @@ pgFmtFilter table (Filter fld (OpExpr hasNot oper)) = notOp <> " " <> pgFmtField
   -- + Can invalidate prepared statements: multiple parameters on an IN($1, $2, $3) will lead to using different prepared statements and not take advantage of caching.
   In vals -> " " <> case vals of
      [""] -> "= ANY('{}') "
-      _    -> "= ANY (" <> unknownLiteral (pgBuildArrayLiteral vals) <> ") "
+      _    -> "= ANY (" <> pgFmtArrayLiteralForField vals fld <> ") "

   Fts op lang val -> " " <> ftsOperator op <> "(" <> ftsLang lang <> unknownLiteral val <> ") "
 where
@@ -345,14 +375,14 @@ pgFmtJoinCondition :: JoinCondition -> SQL.Snippet
 pgFmtJoinCondition (JoinCondition (qi1, col1) (qi2, col2)) =
  pgFmtColumn qi1 col1 <> " = " <> pgFmtColumn qi2 col2

-pgFmtLogicTree :: QualifiedIdentifier -> LogicTree -> SQL.Snippet
-pgFmtLogicTree qi (Expr hasNot op forest) = SQL.sql notOp <> " (" <> intercalateSnippet (opSql op) (pgFmtLogicTree qi <$> forest) <> ")"
+pgFmtLogicTree :: QualifiedIdentifier -> CoercibleLogicTree -> SQL.Snippet
+pgFmtLogicTree qi (CoercibleExpr hasNot op forest) = SQL.sql notOp <> " (" <> intercalateSnippet (opSql op) (pgFmtLogicTree qi <$> forest) <> ")"
  where
    notOp =  if hasNot then "NOT" else mempty

    opSql And = " AND "
    opSql Or  = " OR "
-pgFmtLogicTree qi (Stmnt flt) = pgFmtFilter qi flt
+pgFmtLogicTree qi (CoercibleStmnt flt) = pgFmtFilter qi flt

 pgFmtJsonPath :: JsonPath -> SQL.Snippet
 pgFmtJsonPath = \case
@@ -40,32 +40,38 @@ import qualified Hasql.Transaction          as SQL
 import Contravariant.Extras          (contrazip2)
 import Text.InterpolatedString.Perl6 (q)

-import PostgREST.Config                   (AppConfig (..))
-import PostgREST.Config.Database          (pgVersionStatement,
-                                           toIsolationLevel)
-import PostgREST.Config.PgVersion         (PgVersion, pgVersion100,
-                                           pgVersion110, pgVersion120)
-import PostgREST.SchemaCache.Identifiers  (AccessSet, FieldName,
-                                           QualifiedIdentifier (..),
-                                           Schema)
-import PostgREST.SchemaCache.Relationship (Cardinality (..),
-                                           Junction (..),
-                                           Relationship (..),
-                                           RelationshipsMap)
-import PostgREST.SchemaCache.Routine      (FuncVolatility (..),
-                                           PgType (..), RetType (..),
-                                           Routine (..), RoutineMap,
-                                           RoutineParam (..))
-import PostgREST.SchemaCache.Table        (Column (..), ColumnMap,
-                                           Table (..), TablesMap)
+import PostgREST.Config                      (AppConfig (..))
+import PostgREST.Config.Database             (pgVersionStatement,
+                                              toIsolationLevel)
+import PostgREST.Config.PgVersion            (PgVersion, pgVersion100,
+                                              pgVersion110,
+                                              pgVersion120)
+import PostgREST.SchemaCache.Identifiers     (AccessSet, FieldName,
+                                              QualifiedIdentifier (..),
+                                              Schema)
+import PostgREST.SchemaCache.Relationship    (Cardinality (..),
+                                              Junction (..),
+                                              Relationship (..),
+                                              RelationshipsMap)
+import PostgREST.SchemaCache.Representations (DataRepresentation (..),
+                                              RepresentationsMap)
+import PostgREST.SchemaCache.Routine         (FuncVolatility (..),
+                                              PgType (..),
+                                              RetType (..),
+                                              Routine (..),
+                                              RoutineMap,
+                                              RoutineParam (..))
+import PostgREST.SchemaCache.Table           (Column (..), ColumnMap,
+                                              Table (..), TablesMap)

 import Protolude


 data SchemaCache = SchemaCache
-  { dbTables        :: TablesMap
-  , dbRelationships :: RelationshipsMap
-  , dbRoutines      :: RoutineMap
+  { dbTables          :: TablesMap
+  , dbRelationships   :: RelationshipsMap
+  , dbRoutines        :: RoutineMap
+  , dbRepresentations :: RepresentationsMap
  }
  deriving (Generic, JSON.ToJSON)

@@ -116,6 +122,7 @@ querySchemaCache AppConfig{..} = do
  m2oRels <- SQL.statement mempty $ allM2OandO2ORels pgVer prepared
  funcs   <- SQL.statement schemas $ allFunctions pgVer prepared
  cRels   <- SQL.statement mempty $ allComputedRels prepared
+  reps    <- SQL.statement schemas $ dataRepresentations prepared
  _       <-
    let sleepCall = SQL.Statement "select pg_sleep($1)" (param HE.int4) HD.noResult prepared in
    whenJust configInternalSCSleep (`SQL.statement` sleepCall) -- only used for testing
@@ -127,6 +134,7 @@ querySchemaCache AppConfig{..} = do
      dbTables = tabsWViewsPks
    , dbRelationships = getOverrideRelationshipsMap rels cRels
    , dbRoutines = funcs
+    , dbRepresentations = reps
    }
  where
    schemas = toList configDbSchemas
@@ -156,10 +164,11 @@ getOverrideRelationshipsMap rels cRels =
 removeInternal :: [Schema] -> SchemaCache -> SchemaCache
 removeInternal schemas dbStruct =
  SchemaCache {
-      dbTables        = HM.filterWithKey (\(QualifiedIdentifier sch _) _ -> sch `elem` schemas) $ dbTables dbStruct
-    , dbRelationships = filter (\r -> qiSchema (relForeignTable r) `elem` schemas && not (hasInternalJunction r)) <$>
-                        HM.filterWithKey (\(QualifiedIdentifier sch _, _) _ -> sch `elem` schemas ) (dbRelationships dbStruct)
-    , dbRoutines      = dbRoutines dbStruct -- procs are only obtained from the exposed schemas, no need to filter them.
+      dbTables          = HM.filterWithKey (\(QualifiedIdentifier sch _) _ -> sch `elem` schemas) $ dbTables dbStruct
+    , dbRelationships   = filter (\r -> qiSchema (relForeignTable r) `elem` schemas && not (hasInternalJunction r)) <$>
+                          HM.filterWithKey (\(QualifiedIdentifier sch _, _) _ -> sch `elem` schemas ) (dbRelationships dbStruct)
+    , dbRoutines        = dbRoutines dbStruct -- procs are only obtained from the exposed schemas, no need to filter them.
+    , dbRepresentations = dbRepresentations dbStruct -- no need to filter, not directly exposed through the API
    }
  where
    hasInternalJunction ComputedRelationship{} = False
@@ -280,6 +289,42 @@ decodeFuncs =
                      | v == 's' = Stable
                      | otherwise = Volatile -- only 'v' can happen here

+decodeRepresentations :: HD.Result RepresentationsMap
+decodeRepresentations =
+  HM.fromList . map (\rep@DataRepresentation{drSourceType, drTargetType} -> ((drSourceType, drTargetType), rep)) <$> HD.rowList row
+  where
+    row = DataRepresentation
+      <$> column HD.text
+      <*> column HD.text
+      <*> column HD.text
+
+-- Selects all potential data representation transformations. To qualify the cast must be
+-- 1. to or from a domain
+-- 2. implicit
+-- For the time being it must also be to/from JSON or text, although one can imagine a future where we support special
+-- cases like CSV specific representations.
+dataRepresentations :: Bool -> SQL.Statement [Schema] RepresentationsMap
+dataRepresentations = SQL.Statement sql (arrayParam HE.text) decodeRepresentations
+  where
+    sql = [q|
+    SELECT
+      c.castsource::regtype::text,
+      c.casttarget::regtype::text,
+      c.castfunc::regproc::text
+    FROM
+      pg_catalog.pg_cast c
+    JOIN pg_catalog.pg_type src_t
+      ON c.castsource::oid = src_t.oid
+    JOIN pg_catalog.pg_type dst_t
+      ON c.casttarget::oid = dst_t.oid
+    WHERE
+      c.castcontext = 'i'
+      AND c.castmethod = 'f'
+      AND has_function_privilege(c.castfunc, 'execute')
+      AND ((src_t.typtype = 'd' AND c.casttarget IN ('json'::regtype::oid , 'text'::regtype::oid))
+       OR (dst_t.typtype = 'd' AND c.castsource IN ('json'::regtype::oid , 'text'::regtype::oid)))
+    |]
+
 allFunctions :: PgVersion -> Bool -> SQL.Statement [Schema] RoutineMap
 allFunctions pgVer = SQL.Statement sql (arrayParam HE.text) decodeFuncs
  where
@@ -0,0 +1,29 @@
+{-# LANGUAGE DeriveAnyClass #-}
+{-# LANGUAGE DeriveGeneric  #-}
+
+module PostgREST.SchemaCache.Representations
+  ( DataRepresentation(..)
+  , RepresentationsMap
+  ) where
+
+import qualified Data.Aeson          as JSON
+import qualified Data.HashMap.Strict as HM
+
+
+import Protolude
+
+-- | Data representations allow user customisation of how to present and receive data through APIs, per field.
+-- This structure is used for the library of available transforms. It answers questions like:
+-- - What function, if any, should be used to present a certain field that's been selected for API output?
+-- - How do we parse incoming data for a certain field type when inserting or updating?
+-- - And similarly, how do we parse textual data in a query string to be used as a filter?
+--
+-- Support for outputting special formats like CSV and binary data would fit into the same system.
+data DataRepresentation = DataRepresentation
+  { drSourceType :: Text
+  , drTargetType :: Text
+  , drFunction   :: Text
+  } deriving (Eq, Show, Generic, JSON.ToJSON, JSON.FromJSON)
+
+-- The representation map maps from (source type, target type) to a DR.
+type RepresentationsMap = HM.HashMap (Text, Text) DataRepresentation