2022-04-22 16:38:35 +03:00
|
|
|
-- | This module defines the top-level translation functions pertaining to
|
|
|
|
-- streaming selects into Postgres AST.
|
2022-04-22 20:18:20 +03:00
|
|
|
--
|
|
|
|
-- Streaming subscriptions are subscriptions based on a user-provided cursor
|
|
|
|
-- column. Unlike live queries, streaming subscriptions can be used to only get
|
|
|
|
-- the part that has changed in the query's response, although this will be
|
|
|
|
-- dependent on the user's choice of the cursor column. The streaming starts
|
|
|
|
-- from the initial value provided by the user.
|
2022-04-22 16:38:35 +03:00
|
|
|
module Hasura.Backends.Postgres.Translate.Select.Streaming
|
|
|
|
( mkStreamSQLSelect,
|
|
|
|
selectStreamQuerySQL,
|
|
|
|
)
|
|
|
|
where
|
|
|
|
|
|
|
|
import Control.Monad.Writer.Strict (runWriter)
|
|
|
|
import Database.PG.Query (Query, fromBuilder)
|
|
|
|
import Hasura.Backends.Postgres.SQL.DML qualified as S
|
2022-07-22 18:27:42 +03:00
|
|
|
import Hasura.Backends.Postgres.SQL.RenameIdentifiers (renameIdentifiers)
|
2022-04-22 16:38:35 +03:00
|
|
|
import Hasura.Backends.Postgres.SQL.Types
|
|
|
|
import Hasura.Backends.Postgres.SQL.Value (withConstructorFn)
|
|
|
|
import Hasura.Backends.Postgres.Translate.Select.AnnotatedFieldJSON
|
2022-10-11 13:42:15 +03:00
|
|
|
import Hasura.Backends.Postgres.Translate.Select.Internal.Aliases (contextualizeBaseTableColumn)
|
2022-04-22 20:18:20 +03:00
|
|
|
import Hasura.Backends.Postgres.Translate.Select.Internal.Extractor (asJsonAggExtr)
|
|
|
|
import Hasura.Backends.Postgres.Translate.Select.Internal.GenerateSelect (generateSQLSelectFromArrayNode)
|
|
|
|
import Hasura.Backends.Postgres.Translate.Select.Internal.Process (processAnnSimpleSelect)
|
2022-04-22 16:38:35 +03:00
|
|
|
import Hasura.Backends.Postgres.Translate.Types
|
|
|
|
( MultiRowSelectNode (MultiRowSelectNode),
|
|
|
|
PermissionLimitSubQuery (PLSQNotRequired),
|
|
|
|
SelectNode (SelectNode),
|
|
|
|
SourcePrefixes (SourcePrefixes),
|
|
|
|
orderByForJsonAgg,
|
|
|
|
)
|
|
|
|
import Hasura.Backends.Postgres.Types.Column (unsafePGColumnToBackend)
|
|
|
|
import Hasura.Prelude
|
|
|
|
import Hasura.RQL.IR.BoolExp
|
|
|
|
( AnnBoolExpFld (AVColumn),
|
2022-07-12 12:25:22 +03:00
|
|
|
GBoolExp (BoolField),
|
2022-04-22 16:38:35 +03:00
|
|
|
OpExpG (AGT, ALT),
|
|
|
|
andAnnBoolExps,
|
|
|
|
)
|
|
|
|
import Hasura.RQL.IR.OrderBy (OrderByItemG (OrderByItemG))
|
|
|
|
import Hasura.RQL.IR.Select
|
|
|
|
import Hasura.RQL.Types.Backend (Backend)
|
|
|
|
import Hasura.RQL.Types.Column
|
2022-09-22 12:51:21 +03:00
|
|
|
( ColumnInfo (ciColumn, ciName),
|
2022-04-22 16:38:35 +03:00
|
|
|
ColumnValue (cvType),
|
|
|
|
)
|
|
|
|
import Hasura.RQL.Types.Common
|
|
|
|
( FieldName (FieldName),
|
|
|
|
JsonAggSelect (JASMultipleRows),
|
2022-10-11 13:42:15 +03:00
|
|
|
getFieldNameTxt,
|
2022-04-22 16:38:35 +03:00
|
|
|
)
|
|
|
|
import Hasura.RQL.Types.Subscription
|
|
|
|
( CursorOrdering (CODescending),
|
|
|
|
)
|
|
|
|
import Hasura.SQL.Backend (BackendType (Postgres))
|
|
|
|
import Hasura.SQL.Types
|
|
|
|
( CollectableType (CollectableTypeArray, CollectableTypeScalar),
|
|
|
|
ToSQL (toSQL),
|
|
|
|
)
|
2022-09-22 12:51:21 +03:00
|
|
|
import Language.GraphQL.Draft.Syntax qualified as G
|
2022-04-22 16:38:35 +03:00
|
|
|
|
|
|
|
selectStreamQuerySQL ::
|
|
|
|
forall pgKind.
|
|
|
|
(Backend ('Postgres pgKind), PostgresAnnotatedFieldJSON pgKind) =>
|
|
|
|
AnnSimpleStreamSelect ('Postgres pgKind) ->
|
|
|
|
Query
|
|
|
|
selectStreamQuerySQL sel =
|
|
|
|
fromBuilder $ toSQL $ mkStreamSQLSelect sel
|
|
|
|
|
|
|
|
mkStreamSQLSelect ::
|
|
|
|
forall pgKind.
|
|
|
|
( Backend ('Postgres pgKind),
|
|
|
|
PostgresAnnotatedFieldJSON pgKind
|
|
|
|
) =>
|
|
|
|
AnnSimpleStreamSelect ('Postgres pgKind) ->
|
|
|
|
S.Select
|
|
|
|
mkStreamSQLSelect (AnnSelectStreamG () fields from perm args strfyNum) =
|
|
|
|
let cursorArg = _ssaCursorArg args
|
|
|
|
cursorColInfo = _sciColInfo cursorArg
|
|
|
|
annOrderbyCol = AOCColumn cursorColInfo
|
|
|
|
basicOrderType =
|
|
|
|
bool S.OTAsc S.OTDesc $ _sciOrdering cursorArg == CODescending
|
|
|
|
orderByItems =
|
|
|
|
nonEmpty $ pure $ OrderByItemG (Just basicOrderType) annOrderbyCol Nothing
|
|
|
|
cursorBoolExp =
|
|
|
|
let orderByOpExp = bool ALT AGT $ basicOrderType == S.OTAsc
|
|
|
|
sqlExp =
|
|
|
|
fromResVars
|
|
|
|
(CollectableTypeScalar $ unsafePGColumnToBackend $ cvType (_sciInitialValue cursorArg))
|
2022-09-22 12:51:21 +03:00
|
|
|
["cursor", G.unName $ ciName cursorColInfo]
|
2022-07-12 12:25:22 +03:00
|
|
|
in BoolField $ AVColumn cursorColInfo [(orderByOpExp sqlExp)]
|
2022-04-22 16:38:35 +03:00
|
|
|
|
|
|
|
selectArgs =
|
|
|
|
noSelectArgs
|
|
|
|
{ _saWhere =
|
|
|
|
Just $ maybe cursorBoolExp (andAnnBoolExps cursorBoolExp) $ _ssaWhere args,
|
|
|
|
_saOrderBy = orderByItems,
|
|
|
|
_saLimit = Just $ _ssaBatchSize args
|
|
|
|
}
|
2022-07-19 09:55:42 +03:00
|
|
|
sqlSelect = AnnSelectG fields from perm selectArgs strfyNum Nothing
|
2022-04-22 16:38:35 +03:00
|
|
|
permLimitSubQuery = PLSQNotRequired
|
|
|
|
((selectSource, nodeExtractors), joinTree) =
|
|
|
|
runWriter $
|
|
|
|
flip runReaderT strfyNum $
|
|
|
|
processAnnSimpleSelect sourcePrefixes rootFldName permLimitSubQuery sqlSelect
|
|
|
|
selectNode = SelectNode nodeExtractors joinTree
|
|
|
|
topExtractor =
|
|
|
|
asJsonAggExtr JASMultipleRows rootFldAls permLimitSubQuery $
|
|
|
|
orderByForJsonAgg selectSource
|
|
|
|
cursorLatestValueExp :: S.SQLExp =
|
2022-09-22 12:51:21 +03:00
|
|
|
let columnAlias = ciName cursorColInfo
|
|
|
|
pgColumn = ciColumn cursorColInfo
|
2022-04-22 16:38:35 +03:00
|
|
|
mkMaxOrMinSQLExp maxOrMin col =
|
|
|
|
S.SEFnApp maxOrMin [S.SEIdentifier col] Nothing
|
|
|
|
maxOrMinTxt = bool "MIN" "MAX" $ basicOrderType == S.OTAsc
|
|
|
|
-- text encoding the cursor value while it's fetched from the DB, because
|
|
|
|
-- we can then directly reuse this value, otherwise if this were json encoded
|
|
|
|
-- then we'd have to parse the value and then convert it into a text encoded value
|
|
|
|
colExp =
|
2022-09-22 12:51:21 +03:00
|
|
|
[ S.SELit (G.unName columnAlias),
|
2022-04-22 16:38:35 +03:00
|
|
|
S.SETyAnn
|
2022-09-22 12:51:21 +03:00
|
|
|
( mkMaxOrMinSQLExp maxOrMinTxt $
|
|
|
|
toIdentifier $
|
2022-10-11 13:42:15 +03:00
|
|
|
contextualizeBaseTableColumn rootFldIdentifier pgColumn
|
2022-09-22 12:51:21 +03:00
|
|
|
)
|
2022-04-22 16:38:35 +03:00
|
|
|
S.textTypeAnn
|
|
|
|
]
|
|
|
|
in -- SELECT json_build_object ('col1', MAX(col1) :: text)
|
|
|
|
|
|
|
|
S.SEFnApp "json_build_object" colExp Nothing
|
2022-07-18 12:44:17 +03:00
|
|
|
cursorLatestValueExtractor = S.Extractor cursorLatestValueExp (Just $ S.toColumnAlias $ Identifier "cursor")
|
2022-04-22 16:38:35 +03:00
|
|
|
arrayNode = MultiRowSelectNode [topExtractor, cursorLatestValueExtractor] selectNode
|
2022-07-22 18:27:42 +03:00
|
|
|
in renameIdentifiers $
|
2022-04-22 16:38:35 +03:00
|
|
|
generateSQLSelectFromArrayNode selectSource arrayNode $ S.BELit True
|
|
|
|
where
|
2022-10-11 13:42:15 +03:00
|
|
|
rootFldIdentifier = TableIdentifier $ getFieldNameTxt rootFldName
|
|
|
|
sourcePrefixes = SourcePrefixes (tableIdentifierToIdentifier rootFldIdentifier) (tableIdentifierToIdentifier rootFldIdentifier)
|
2022-04-22 16:38:35 +03:00
|
|
|
rootFldName = FieldName "root"
|
2022-07-18 12:44:17 +03:00
|
|
|
rootFldAls = S.toColumnAlias $ toIdentifier rootFldName
|
2022-04-22 16:38:35 +03:00
|
|
|
|
|
|
|
-- TODO: these functions also exist in `resolveMultiplexedValue`, de-duplicate these!
|
|
|
|
fromResVars pgType jPath =
|
|
|
|
addTypeAnnotation pgType $
|
|
|
|
S.SEOpApp
|
|
|
|
(S.SQLOp "#>>")
|
2022-10-11 13:42:15 +03:00
|
|
|
[ S.SEQIdentifier $ S.QIdentifier (S.QualifiedIdentifier (TableIdentifier "_subs") Nothing) (Identifier "result_vars"),
|
2022-04-22 16:38:35 +03:00
|
|
|
S.SEArray $ map S.SELit jPath
|
|
|
|
]
|
|
|
|
addTypeAnnotation pgType =
|
|
|
|
flip S.SETyAnn (S.mkTypeAnn pgType) . case pgType of
|
|
|
|
CollectableTypeScalar scalarType -> withConstructorFn scalarType
|
|
|
|
CollectableTypeArray _ -> id
|