@Override
    public Stream<? extends Ticket> getSessionsWithAttributes(final Map<String, List<Object>> queryAttributes) {
        val factory = getJpaTicketEntityFactory();
        val criteria = queryAttributes.entrySet()
            .stream()
            .map(entry -> {
                val criteriaValues = entry.getValue()
                    .stream()
                    .map(queryValue -> {
                        if (factory.isOracle()) {
                            return String.format("JSON_EXISTS(t.attributes, '$?(@.\"%s\" == \"%s\")')",
                                digestIdentifier(entry.getKey()), digestIdentifier(queryValue.toString()));
                        }
                        if (factory.isPostgres()) {
                            return String.format("(t.attributes->'%s')\\:\\:jsonb \\?\\? '%s'", digestIdentifier(entry.getKey()), digestIdentifier(queryValue.toString()));
                        }
                        if (factory.isMariaDb()) {
                            val attributeKey = isCipherExecutorEnabled()
                                ? entry.getKey()
                                : entry.getKey().replace("-", "\\\\-");
                            return String.format("JSON_SEARCH(JSON_QUERY(t.attributes, '$.%s'), 'one', '%s') is not null",
                                String.format("\"%s\"", digestIdentifier(attributeKey)), digestIdentifier(queryValue.toString()));
                        }
                        if (factory.isMySql()) {
                            return String.format("JSON_SEARCH(JSON_EXTRACT(t.attributes, '$.%s'), 'one', '%s') is not null",
                                String.format("\"%s\"", digestIdentifier(entry.getKey())), digestIdentifier(queryValue.toString()));
                        }
                        if (factory.isMsSqlServer()) {
                            return String.format("%s.value='%s'", String.format("\"%s\"", digestIdentifier(entry.getKey())), digestIdentifier(queryValue.toString()));
                        }
                        return "1=2";
                    })
                    .collect(Collectors.joining(" OR "));
                return String.format("(%s)", criteriaValues);
            })
            .collect(Collectors.joining(" AND "));

        val selectClause = new StringBuilder(String.format("SELECT t.* FROM %s t ", factory.getTableName()));
        if (factory.isMsSqlServer()) {
            for (val entry : queryAttributes.entrySet()) {
                val name = String.format("\"%s\"", digestIdentifier(entry.getKey()));
                selectClause.append(String.format("CROSS APPLY OPENJSON(t.attributes, '$.%s') %s ", name, name));
            }
        }

        val sql = String.format("%s WHERE t.type='%s' AND %s", selectClause,
            getTicketTypeName(TicketGrantingTicket.class), criteria);
        LOGGER.debug("Executing SQL query [{}]", sql);
        entityManager.flush();
        val query = entityManager.createNativeQuery(sql, factory.getType());
        return jpaBeanFactory.streamQuery(query)
            .map(BaseTicketEntity.class::cast)
            .map(factory::toTicket)
            .map(this::decodeTicket)
            .filter(ticket -> !ticket.isExpired());
    }
This method purposefully doesn't lock any rows, because the stream traversing can take an indeterminate
amount of time, and logging in to an application with an existing TGT will update the TGT row in the database.

@return streamable results
support/cas-server-support-jpa-ticket-registry/src/main/java/org/apereo/cas/ticket/registry/JpaTicketRegistry.java