ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2024-12-20 21:43:14 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Vault Query Pagination simplification (#997)

Browse Source 
* Pagination improvements (fail-fast on too many results without pagination specification)
* Fix incorrectly returned results count.
* Performance optimisation: only return totalStatesAvailable count on Pagination specification.
* Changed DEFAULT_PAGE_NUMBER to 1 (eg. page numbering starts from 1)
* Changed MAX_PAGE_SIZE to Int.MAX_VALUE
* Fixed compiler WARNINGs in Unit tests.
* Fixed minimum page size check (1).
* Updated API-RST docs with behavioural notes.
* Updated documentation (RST and API);
This commit is contained in:
josecoll 2017-07-12 09:53:15 +01:00 committed by GitHub
parent ac07b3fe94
commit 5f7b8f6ec3
9 changed files with 182 additions and 147 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
client/rpc/src/smoke-test/kotlin/net/corda/kotlin/rpc/StandaloneCordaRPClientTest.kt
View File

@ -13,10 +13,7 @@ import net.corda.core.getOrThrow
import net.corda.core.messaging.* import net.corda.core.messaging.*
import net.corda.core.node.NodeInfo import net.corda.core.node.NodeInfo
import net.corda.core.node.services.Vault import net.corda.core.node.services.Vault
import net.corda.core.node.services.vault.PageSpecification import net.corda.core.node.services.vault.*
import net.corda.core.node.services.vault.QueryCriteria
import net.corda.core.node.services.vault.Sort
import net.corda.core.node.services.vault.SortAttribute
import net.corda.core.seconds import net.corda.core.seconds
import net.corda.core.utilities.OpaqueBytes import net.corda.core.utilities.OpaqueBytes
import net.corda.core.sizedInputStreamAndHash import net.corda.core.sizedInputStreamAndHash
@ -190,7 +187,7 @@ class StandaloneCordaRPClientTest {
.returnValue.getOrThrow(timeout) .returnValue.getOrThrow(timeout)
val criteria = QueryCriteria.VaultQueryCriteria(status = Vault.StateStatus.ALL) val criteria = QueryCriteria.VaultQueryCriteria(status = Vault.StateStatus.ALL)
val paging = PageSpecification(0, 10) val paging = PageSpecification(DEFAULT_PAGE_NUM, 10)
val sorting = Sort(setOf(Sort.SortColumn(SortAttribute.Standard(Sort.VaultStateAttribute.RECORDED_TIME), Sort.Direction.DESC))) val sorting = Sort(setOf(Sort.SortColumn(SortAttribute.Standard(Sort.VaultStateAttribute.RECORDED_TIME), Sort.Direction.DESC)))
val queryResults = rpcProxy.vaultQueryBy<Cash.State>(criteria, paging, sorting) val queryResults = rpcProxy.vaultQueryBy<Cash.State>(criteria, paging, sorting)

19
core/src/main/kotlin/net/corda/core/messaging/CordaRPCOps.kt
View File

@ -13,6 +13,8 @@ import net.corda.core.identity.Party
import net.corda.core.node.NodeInfo import net.corda.core.node.NodeInfo
import net.corda.core.node.services.NetworkMapCache import net.corda.core.node.services.NetworkMapCache
import net.corda.core.node.services.Vault import net.corda.core.node.services.Vault
import net.corda.core.node.services.VaultQueryException
import net.corda.core.node.services.vault.DEFAULT_PAGE_SIZE
import net.corda.core.node.services.vault.PageSpecification import net.corda.core.node.services.vault.PageSpecification
import net.corda.core.node.services.vault.QueryCriteria import net.corda.core.node.services.vault.QueryCriteria
import net.corda.core.node.services.vault.Sort import net.corda.core.node.services.vault.Sort
@ -78,13 +80,18 @@ interface CordaRPCOps : RPCOps {
* and returns a [Vault.Page] object containing the following: * and returns a [Vault.Page] object containing the following:
* 1. states as a List of <StateAndRef> (page number and size defined by [PageSpecification]) * 1. states as a List of <StateAndRef> (page number and size defined by [PageSpecification])
* 2. states metadata as a List of [Vault.StateMetadata] held in the Vault States table. * 2. states metadata as a List of [Vault.StateMetadata] held in the Vault States table.
* 3. the [PageSpecification] used in the query * 3. total number of results available if [PageSpecification] supplied (otherwise returns -1)
* 4. a total number of results available (for subsequent paging if necessary) * 4. status types used in this query: UNCONSUMED, CONSUMED, ALL
* 5. status types used in this query: UNCONSUMED, CONSUMED, ALL * 5. other results (aggregate functions with/without using value groups)
* 6. other results (aggregate functions with/without using value groups)
* *
* Note: a default [PageSpecification] is applied to the query returning the 1st page (indexed from 0) with up to 200 entries. * @throws VaultQueryException if the query cannot be executed for any reason
* It is the responsibility of the Client to request further pages and/or specify a more suitable [PageSpecification]. * (missing criteria or parsing error, paging errors, unsupported query, underlying database error)
*
* Notes
* If no [PageSpecification] is provided, a maximum of [DEFAULT_PAGE_SIZE] results will be returned.
* API users must specify a [PageSpecification] if they are expecting more than [DEFAULT_PAGE_SIZE] results,
* otherwise a [VaultQueryException] will be thrown alerting to this condition.
* It is the responsibility of the API user to request further pages and/or specify a more suitable [PageSpecification].
*/ */
// DOCSTART VaultQueryByAPI // DOCSTART VaultQueryByAPI
@RPCReturnsObservables @RPCReturnsObservables

31
core/src/main/kotlin/net/corda/core/node/services/Services.kt
View File

@ -15,6 +15,7 @@ import net.corda.core.messaging.DataFeed
import net.corda.core.node.services.vault.PageSpecification import net.corda.core.node.services.vault.PageSpecification
import net.corda.core.node.services.vault.QueryCriteria import net.corda.core.node.services.vault.QueryCriteria
import net.corda.core.node.services.vault.Sort import net.corda.core.node.services.vault.Sort
import net.corda.core.node.services.vault.DEFAULT_PAGE_SIZE
import net.corda.core.serialization.CordaSerializable import net.corda.core.serialization.CordaSerializable
import net.corda.core.utilities.OpaqueBytes import net.corda.core.utilities.OpaqueBytes
import net.corda.core.toFuture import net.corda.core.toFuture
@ -118,12 +119,10 @@ class Vault<out T : ContractState>(val states: Iterable<StateAndRef<T>>) {
* A Page contains: * A Page contains:
* 1) a [List] of actual [StateAndRef] requested by the specified [QueryCriteria] to a maximum of [MAX_PAGE_SIZE] * 1) a [List] of actual [StateAndRef] requested by the specified [QueryCriteria] to a maximum of [MAX_PAGE_SIZE]
* 2) a [List] of associated [Vault.StateMetadata], one per [StateAndRef] result * 2) a [List] of associated [Vault.StateMetadata], one per [StateAndRef] result
* 3) the [PageSpecification] definition used to bound this result set * 3) a total number of states that met the given [QueryCriteria] if a [PageSpecification] was provided
* 4) a total number of states that met the given [QueryCriteria] * (otherwise defaults to -1)
* Note that this may be more than the specified [PageSpecification.pageSize], and should be used to perform * 4) Status types used in this query: UNCONSUMED, CONSUMED, ALL
* further pagination (by issuing new queries). * 5) Other results as a [List] of any type (eg. aggregate function results with/without group by)
* 5) Status types used in this query: UNCONSUMED, CONSUMED, ALL
* 6) Other results as a [List] of any type (eg. aggregate function results with/without group by)
* *
* Note: currently otherResults are used only for Aggregate Functions (in which case, the states and statesMetadata * Note: currently otherResults are used only for Aggregate Functions (in which case, the states and statesMetadata
* results will be empty) * results will be empty)
@ -131,8 +130,7 @@ class Vault<out T : ContractState>(val states: Iterable<StateAndRef<T>>) {
@CordaSerializable @CordaSerializable
data class Page<out T : ContractState>(val states: List<StateAndRef<T>>, data class Page<out T : ContractState>(val states: List<StateAndRef<T>>,
val statesMetadata: List<StateMetadata>, val statesMetadata: List<StateMetadata>,
val pageable: PageSpecification, val totalStatesAvailable: Long,
val totalStatesAvailable: Int,
val stateTypes: StateStatus, val stateTypes: StateStatus,
val otherResults: List<Any>) val otherResults: List<Any>)
@ -353,17 +351,18 @@ interface VaultQueryService {
* and returns a [Vault.Page] object containing the following: * and returns a [Vault.Page] object containing the following:
* 1. states as a List of <StateAndRef> (page number and size defined by [PageSpecification]) * 1. states as a List of <StateAndRef> (page number and size defined by [PageSpecification])
* 2. states metadata as a List of [Vault.StateMetadata] held in the Vault States table. * 2. states metadata as a List of [Vault.StateMetadata] held in the Vault States table.
* 3. the [PageSpecification] used in the query * 3. total number of results available if [PageSpecification] supplied (otherwise returns -1)
* 4. a total number of results available (for subsequent paging if necessary) * 4. status types used in this query: UNCONSUMED, CONSUMED, ALL
* 5. status types used in this query: UNCONSUMED, CONSUMED, ALL * 5. other results (aggregate functions with/without using value groups)
* 6. other results (aggregate functions with/without using value groups)
* *
* @throws VaultQueryException if the query cannot be executed for any reason * @throws VaultQueryException if the query cannot be executed for any reason
* (missing criteria or parsing error, invalid operator, unsupported query, underlying database error) * (missing criteria or parsing error, paging errors, unsupported query, underlying database error)
* *
* Note: a default [PageSpecification] is applied to the query returning the 1st page (indexed from 0) with up to 200 entries. * Notes
* It is the responsibility of the Client to request further pages and/or specify a more suitable [PageSpecification]. * If no [PageSpecification] is provided, a maximum of [DEFAULT_PAGE_SIZE] results will be returned.
* Note2: you can also annotate entity fields with JPA OrderBy annotation to achieve the same effect as explicit sorting * API users must specify a [PageSpecification] if they are expecting more than [DEFAULT_PAGE_SIZE] results,
* otherwise a [VaultQueryException] will be thrown alerting to this condition.
* It is the responsibility of the API user to request further pages and/or specify a more suitable [PageSpecification].
*/ */
@Throws(VaultQueryException::class) @Throws(VaultQueryException::class)
fun <T : ContractState> _queryBy(criteria: QueryCriteria, fun <T : ContractState> _queryBy(criteria: QueryCriteria,

18
core/src/main/kotlin/net/corda/core/node/services/vault/QueryCriteriaUtils.kt
View File

@ -119,21 +119,25 @@ fun <O, C> getColumnName(column: Column<O, C>): String {
* paging and sorting capability: * paging and sorting capability:
* https://docs.spring.io/spring-data/commons/docs/current/api/org/springframework/data/repository/PagingAndSortingRepository.html * https://docs.spring.io/spring-data/commons/docs/current/api/org/springframework/data/repository/PagingAndSortingRepository.html
*/ */
const val DEFAULT_PAGE_NUM = 0 const val DEFAULT_PAGE_NUM = 1
const val DEFAULT_PAGE_SIZE = 200 const val DEFAULT_PAGE_SIZE = 200
/** /**
* Note: this maximum size will be configurable in future (to allow for large JVM heap sized node configurations) * Note: use [PageSpecification] to correctly handle a number of bounded pages of a pre-configured page size.
* Use [PageSpecification] to correctly handle a number of bounded pages of [MAX_PAGE_SIZE].
*/ */
const val MAX_PAGE_SIZE = 512 const val MAX_PAGE_SIZE = Int.MAX_VALUE
/** /**
* PageSpecification allows specification of a page number (starting from 0 as default) and page size (defaulting to * [PageSpecification] allows specification of a page number (starting from [DEFAULT_PAGE_NUM]) and page size
* [DEFAULT_PAGE_SIZE] with a maximum page size of [MAX_PAGE_SIZE] * (defaulting to [DEFAULT_PAGE_SIZE] with a maximum page size of [MAX_PAGE_SIZE])
* Note: we default the page number to [DEFAULT_PAGE_SIZE] to enable queries without requiring a page specification
* but enabling detection of large results sets that fall out of the [DEFAULT_PAGE_SIZE] requirement.
* [MAX_PAGE_SIZE] should be used with extreme caution as results may exceed your JVM memory footprint.
*/ */
@CordaSerializable @CordaSerializable
data class PageSpecification(val pageNumber: Int = DEFAULT_PAGE_NUM, val pageSize: Int = DEFAULT_PAGE_SIZE) data class PageSpecification(val pageNumber: Int = -1, val pageSize: Int = DEFAULT_PAGE_SIZE) {
val isDefault = (pageSize == DEFAULT_PAGE_SIZE && pageNumber == -1)
}
/** /**
* Sort allows specification of a set of entity attribute names and their associated directionality * Sort allows specification of a set of entity attribute names and their associated directionality

31
docs/source/api-vault-query.rst
View File

@ -48,7 +48,7 @@ The API provides both static (snapshot) and dynamic (snapshot with streaming upd
.. note:: Streaming updates are only filtered based on contract type and state status (UNCONSUMED, CONSUMED, ALL) .. note:: Streaming updates are only filtered based on contract type and state status (UNCONSUMED, CONSUMED, ALL)
Simple pagination (page number and size) and sorting (directional ordering using standard or custom property attributes) is also specifiable. Simple pagination (page number and size) and sorting (directional ordering using standard or custom property attributes) is also specifiable.
Defaults are defined for Paging (pageNumber = 0, pageSize = 200) and Sorting (direction = ASC). Defaults are defined for Paging (pageNumber = 1, pageSize = 200) and Sorting (direction = ASC).
The ``QueryCriteria`` interface provides a flexible mechanism for specifying different filtering criteria, including and/or composition and a rich set of operators to include: binary logical (AND, OR), comparison (LESS_THAN, LESS_THAN_OR_EQUAL, GREATER_THAN, GREATER_THAN_OR_EQUAL), equality (EQUAL, NOT_EQUAL), likeness (LIKE, NOT_LIKE), nullability (IS_NULL, NOT_NULL), and collection based (IN, NOT_IN). Standard SQL-92 aggregate functions (SUM, AVG, MIN, MAX, COUNT) are also supported. The ``QueryCriteria`` interface provides a flexible mechanism for specifying different filtering criteria, including and/or composition and a rich set of operators to include: binary logical (AND, OR), comparison (LESS_THAN, LESS_THAN_OR_EQUAL, GREATER_THAN, GREATER_THAN_OR_EQUAL), equality (EQUAL, NOT_EQUAL), likeness (LIKE, NOT_LIKE), nullability (IS_NULL, NOT_NULL), and collection based (IN, NOT_IN). Standard SQL-92 aggregate functions (SUM, AVG, MIN, MAX, COUNT) are also supported.
@ -104,6 +104,15 @@ An example of a custom query in Java is illustrated here:
.. note:: Current queries by ``Party`` specify the ``AbstractParty`` which may be concrete or anonymous. In the later case, where an anonymous party does not have an associated X500Name, then no query results will ever be produced. For performance reasons, queries do not use PublicKey as search criteria. Ongoing design work on identity manangement is likely to enhance identity based queries (including composite key criteria selection). .. note:: Current queries by ``Party`` specify the ``AbstractParty`` which may be concrete or anonymous. In the later case, where an anonymous party does not have an associated X500Name, then no query results will ever be produced. For performance reasons, queries do not use PublicKey as search criteria. Ongoing design work on identity manangement is likely to enhance identity based queries (including composite key criteria selection).
Pagination
----------
The API provides support for paging where large numbers of results are expected (by default, a page size is set to 200 results).
Defining a sensible default page size enables efficient checkpointing within flows, and frees the developer from worrying about pagination where
result sets are expected to be constrained to 200 or fewer entries. Where large result sets are expected (such as using the RPC API for reporting and/or UI display), it is strongly recommended to define a ``PageSpecification`` to correctly process results with efficient memory utilistion. A fail-fast mode is in place to alert API users to the need for pagination where a single query returns more than 200 results and no ``PageSpecification``
has been supplied.
.. note:: A pages maximum size ``MAX_PAGE_SIZE`` is defined as ``Int.MAX_VALUE`` and should be used with extreme caution as results returned may exceed your JVM's memory footprint.
Example usage Example usage
------------- -------------
@ -284,7 +293,7 @@ Track unconsumed linear states:
:end-before: DOCEND VaultQueryExample16 :end-before: DOCEND VaultQueryExample16
.. note:: This will return both Deal and Linear states. .. note:: This will return both Deal and Linear states.
Track unconsumed deal states: Track unconsumed deal states:
.. literalinclude:: ../../node/src/test/kotlin/net/corda/node/services/vault/VaultQueryTests.kt .. literalinclude:: ../../node/src/test/kotlin/net/corda/node/services/vault/VaultQueryTests.kt
@ -369,6 +378,17 @@ Track unconsumed deal states or linear states (with snapshot including specifica
:start-after: DOCSTART VaultJavaQueryExample4 :start-after: DOCSTART VaultJavaQueryExample4
:end-before: DOCEND VaultJavaQueryExample4 :end-before: DOCEND VaultJavaQueryExample4
Behavioural notes
-----------------
1. **TrackBy** updates do not take into account the full criteria specification due to different and more restrictive syntax
in `observables <https://github.com/ReactiveX/RxJava/wiki>`_ filtering (vs full SQL-92 JDBC filtering as used in snapshot views).
Specifically, dynamic updates are filtered by ``contractType`` and ``stateType`` (UNCONSUMED, CONSUMED, ALL) only.
2. **QueryBy** and **TrackBy snapshot views** using pagination may return different result sets as each paging request is a
separate SQL query on the underlying database, and it is entirely conceivable that state modifications are taking
place in between and/or in parallel to paging requests.
When using pagination, always check the value of the ``totalStatesAvailable`` (from the ``Vault.Page`` result) and
adjust further paging requests appropriately.
Other use case scenarios Other use case scenarios
------------------------ ------------------------
@ -410,10 +430,11 @@ This query returned an ``Iterable<StateAndRef<T>>``
The query returns a ``Vault.Page`` result containing: The query returns a ``Vault.Page`` result containing:
- states as a ``List<StateAndRef<T : ContractState>>`` sized according to the default Page specification of ``DEFAULT_PAGE_NUM`` (0) and ``DEFAULT_PAGE_SIZE`` (200). - states as a ``List<StateAndRef<T : ContractState>>`` up to a maximum of ``DEFAULT_PAGE_SIZE`` (200) where no ``PageSpecification`` provided, otherwise returns results according to the parameters ``pageNumber`` and ``pageSize`` specified in the supplied ``PageSpecification``.
- states metadata as a ``List<Vault.StateMetadata>`` containing Vault State metadata held in the Vault states table. - states metadata as a ``List<Vault.StateMetadata>`` containing Vault State metadata held in the Vault states table.
- the ``PagingSpecification`` used in the query - a ``total`` number of results available if ``PageSpecification`` provided (otherwise returns -1). For pagination, this value can be used to issue subsequent queries with appropriately specified ``PageSpecification`` parameters (according to your paging needs and/or maximum memory capacity for holding large data sets). Note it is your responsibility to manage page numbers and sizes.
- a ``total`` number of results available. This value can be used issue subsequent queries with appropriately specified ``PageSpecification`` (according to your paging needs and/or maximum memory capacity for holding large data sets). Note it is your responsibility to manage page numbers and sizes. - status types used in this query: UNCONSUMED, CONSUMED, ALL
- other results as a [List] of any type (eg. aggregate function results with/without group by)
2. ServiceHub usage obtaining linear heads for a given contract state type 2. ServiceHub usage obtaining linear heads for a given contract state type

5
node/src/main/kotlin/net/corda/node/services/vault/HibernateQueryCriteriaParser.kt
View File

@ -95,6 +95,7 @@ class HibernateQueryCriteriaParser(val contractType: Class<out ContractState>,
} }
is ColumnPredicate.BinaryComparison -> { is ColumnPredicate.BinaryComparison -> {
column as Path<Comparable<Any?>?> column as Path<Comparable<Any?>?>
@Suppress("UNCHECKED_CAST")
val literal = columnPredicate.rightLiteral as Comparable<Any?>? val literal = columnPredicate.rightLiteral as Comparable<Any?>?
when (columnPredicate.operator) { when (columnPredicate.operator) {
BinaryComparisonOperator.GREATER_THAN -> criteriaBuilder.greaterThan(column, literal) BinaryComparisonOperator.GREATER_THAN -> criteriaBuilder.greaterThan(column, literal)
@ -117,8 +118,11 @@ class HibernateQueryCriteriaParser(val contractType: Class<out ContractState>,
} }
} }
is ColumnPredicate.Between -> { is ColumnPredicate.Between -> {
@Suppress("UNCHECKED_CAST")
column as Path<Comparable<Any?>?> column as Path<Comparable<Any?>?>
@Suppress("UNCHECKED_CAST")
val fromLiteral = columnPredicate.rightFromLiteral as Comparable<Any?>? val fromLiteral = columnPredicate.rightFromLiteral as Comparable<Any?>?
@Suppress("UNCHECKED_CAST")
val toLiteral = columnPredicate.rightToLiteral as Comparable<Any?>? val toLiteral = columnPredicate.rightToLiteral as Comparable<Any?>?
criteriaBuilder.between(column, fromLiteral, toLiteral) criteriaBuilder.between(column, fromLiteral, toLiteral)
} }
@ -164,6 +168,7 @@ class HibernateQueryCriteriaParser(val contractType: Class<out ContractState>,
val columnPredicate = expression.predicate val columnPredicate = expression.predicate
when (columnPredicate) { when (columnPredicate) {
is ColumnPredicate.AggregateFunction -> { is ColumnPredicate.AggregateFunction -> {
@Suppress("UNCHECKED_CAST")
column as Path<Long?>? column as Path<Long?>?
val aggregateExpression = val aggregateExpression =
when (columnPredicate.type) { when (columnPredicate.type) {

65
node/src/main/kotlin/net/corda/node/services/vault/HibernateVaultQueryImpl.kt
View File

@ -11,10 +11,8 @@ import net.corda.core.messaging.DataFeed
import net.corda.core.node.services.Vault import net.corda.core.node.services.Vault
import net.corda.core.node.services.VaultQueryException import net.corda.core.node.services.VaultQueryException
import net.corda.core.node.services.VaultQueryService import net.corda.core.node.services.VaultQueryService
import net.corda.core.node.services.vault.MAX_PAGE_SIZE import net.corda.core.node.services.vault.*
import net.corda.core.node.services.vault.PageSpecification import net.corda.core.node.services.vault.QueryCriteria.VaultCustomQueryCriteria
import net.corda.core.node.services.vault.QueryCriteria
import net.corda.core.node.services.vault.Sort
import net.corda.core.serialization.SingletonSerializeAsToken import net.corda.core.serialization.SingletonSerializeAsToken
import net.corda.core.serialization.deserialize import net.corda.core.serialization.deserialize
import net.corda.core.serialization.storageKryo import net.corda.core.serialization.storageKryo
@ -43,6 +41,15 @@ class HibernateVaultQueryImpl(hibernateConfig: HibernateConfiguration,
override fun <T : ContractState> _queryBy(criteria: QueryCriteria, paging: PageSpecification, sorting: Sort, contractType: Class<out T>): Vault.Page<T> { override fun <T : ContractState> _queryBy(criteria: QueryCriteria, paging: PageSpecification, sorting: Sort, contractType: Class<out T>): Vault.Page<T> {
log.info("Vault Query for contract type: $contractType, criteria: $criteria, pagination: $paging, sorting: $sorting") log.info("Vault Query for contract type: $contractType, criteria: $criteria, pagination: $paging, sorting: $sorting")
// calculate total results where a page specification has been defined
var totalStates = -1L
if (!paging.isDefault) {
val count = builder { VaultSchemaV1.VaultStates::recordedTime.count() }
val countCriteria = VaultCustomQueryCriteria(count)
val results = queryBy(contractType, criteria.and(countCriteria))
totalStates = results.otherResults[0] as Long
}
val session = sessionFactory.withOptions(). val session = sessionFactory.withOptions().
connection(TransactionManager.current().connection). connection(TransactionManager.current().connection).
openSession() openSession()
@ -62,43 +69,45 @@ class HibernateVaultQueryImpl(hibernateConfig: HibernateConfiguration,
// prepare query for execution // prepare query for execution
val query = session.createQuery(criteriaQuery) val query = session.createQuery(criteriaQuery)
// pagination // pagination checks
if (paging.pageNumber < 0) throw VaultQueryException("Page specification: invalid page number ${paging.pageNumber} [page numbers start from 0]") if (!paging.isDefault) {
if (paging.pageSize < 0 || paging.pageSize > MAX_PAGE_SIZE) throw VaultQueryException("Page specification: invalid page size ${paging.pageSize} [maximum page size is ${MAX_PAGE_SIZE}]") // pagination
if (paging.pageNumber < DEFAULT_PAGE_NUM) throw VaultQueryException("Page specification: invalid page number ${paging.pageNumber} [page numbers start from $DEFAULT_PAGE_NUM]")
if (paging.pageSize < 1) throw VaultQueryException("Page specification: invalid page size ${paging.pageSize} [must be a value between 1 and $MAX_PAGE_SIZE]")
}
// count total results available query.firstResult = (paging.pageNumber - 1) * paging.pageSize
val countQuery = criteriaBuilder.createQuery(Long::class.java) query.maxResults = paging.pageSize + 1 // detection too many results
countQuery.select(criteriaBuilder.count(countQuery.from(VaultSchemaV1.VaultStates::class.java)))
val totalStates = session.createQuery(countQuery).singleResult.toInt()
if ((paging.pageNumber != 0) && (paging.pageSize * paging.pageNumber >= totalStates))
throw VaultQueryException("Requested more results than available [${paging.pageSize} * ${paging.pageNumber} >= ${totalStates}]")
query.firstResult = paging.pageNumber * paging.pageSize
query.maxResults = paging.pageSize
// execution // execution
val results = query.resultList val results = query.resultList
val statesAndRefs: MutableList<StateAndRef<*>> = mutableListOf()
// final pagination check (fail-fast on too many results when no pagination specified)
if (paging.isDefault && results.size > DEFAULT_PAGE_SIZE)
throw VaultQueryException("Please specify a `PageSpecification` as there are more results [${results.size}] than the default page size [$DEFAULT_PAGE_SIZE]")
val statesAndRefs: MutableList<StateAndRef<T>> = mutableListOf()
val statesMeta: MutableList<Vault.StateMetadata> = mutableListOf() val statesMeta: MutableList<Vault.StateMetadata> = mutableListOf()
val otherResults: MutableList<Any> = mutableListOf() val otherResults: MutableList<Any> = mutableListOf()
results.asSequence() results.asSequence()
.forEach { it -> .forEachIndexed { index, result ->
if (it[0] is VaultSchemaV1.VaultStates) { if (result[0] is VaultSchemaV1.VaultStates) {
val it = it[0] as VaultSchemaV1.VaultStates if (!paging.isDefault && index == paging.pageSize) // skip last result if paged
val stateRef = StateRef(SecureHash.parse(it.stateRef!!.txId!!), it.stateRef!!.index!!) return@forEachIndexed
val state = it.contractState.deserialize<TransactionState<T>>(storageKryo()) val vaultState = result[0] as VaultSchemaV1.VaultStates
statesMeta.add(Vault.StateMetadata(stateRef, it.contractStateClassName, it.recordedTime, it.consumedTime, it.stateStatus, it.notaryName, it.notaryKey, it.lockId, it.lockUpdateTime)) val stateRef = StateRef(SecureHash.parse(vaultState.stateRef!!.txId!!), vaultState.stateRef!!.index!!)
val state = vaultState.contractState.deserialize<TransactionState<T>>(storageKryo())
statesMeta.add(Vault.StateMetadata(stateRef, vaultState.contractStateClassName, vaultState.recordedTime, vaultState.consumedTime, vaultState.stateStatus, vaultState.notaryName, vaultState.notaryKey, vaultState.lockId, vaultState.lockUpdateTime))
statesAndRefs.add(StateAndRef(state, stateRef)) statesAndRefs.add(StateAndRef(state, stateRef))
} }
else { else {
log.debug { "OtherResults: ${Arrays.toString(it.toArray())}" } log.debug { "OtherResults: ${Arrays.toString(result.toArray())}" }
otherResults.addAll(it.toArray().asList()) otherResults.addAll(result.toArray().asList())
} }
} }
return Vault.Page(states = statesAndRefs, statesMetadata = statesMeta, pageable = paging, stateTypes = criteriaParser.stateTypes, totalStatesAvailable = totalStates, otherResults = otherResults) as Vault.Page<T> return Vault.Page(states = statesAndRefs, statesMetadata = statesMeta, stateTypes = criteriaParser.stateTypes, totalStatesAvailable = totalStates, otherResults = otherResults)
} catch (e: Exception) { } catch (e: Exception) {
log.error(e.message) log.error(e.message)
@ -132,6 +141,7 @@ class HibernateVaultQueryImpl(hibernateConfig: HibernateConfiguration,
val contractInterfaceToConcreteTypes = mutableMapOf<String, MutableList<String>>() val contractInterfaceToConcreteTypes = mutableMapOf<String, MutableList<String>>()
distinctTypes.forEach { it -> distinctTypes.forEach { it ->
@Suppress("UNCHECKED_CAST")
val concreteType = Class.forName(it) as Class<ContractState> val concreteType = Class.forName(it) as Class<ContractState>
val contractInterfaces = deriveContractInterfaces(concreteType) val contractInterfaces = deriveContractInterfaces(concreteType)
contractInterfaces.map { contractInterfaces.map {
@ -146,6 +156,7 @@ class HibernateVaultQueryImpl(hibernateConfig: HibernateConfiguration,
val myInterfaces: MutableSet<Class<T>> = mutableSetOf() val myInterfaces: MutableSet<Class<T>> = mutableSetOf()
clazz.interfaces.forEach { clazz.interfaces.forEach {
if (!it.equals(ContractState::class.java)) { if (!it.equals(ContractState::class.java)) {
@Suppress("UNCHECKED_CAST")
myInterfaces.add(it as Class<T>) myInterfaces.add(it as Class<T>)
myInterfaces.addAll(deriveContractInterfaces(it)) myInterfaces.addAll(deriveContractInterfaces(it))
} }

97
node/src/test/java/net/corda/node/services/vault/VaultQueryJavaTests.java
View File

@ -1,62 +1,45 @@
package net.corda.node.services.vault; package net.corda.node.services.vault;
import com.google.common.collect.ImmutableSet; import com.google.common.collect.*;
import kotlin.Pair; import kotlin.*;
import net.corda.contracts.DealState; import net.corda.contracts.*;
import net.corda.contracts.asset.Cash; import net.corda.contracts.asset.*;
import net.corda.core.contracts.*; import net.corda.core.contracts.*;
import net.corda.testing.contracts.DummyLinearContract;
import net.corda.core.crypto.*; import net.corda.core.crypto.*;
import net.corda.core.identity.AbstractParty; import net.corda.core.identity.*;
import net.corda.core.messaging.DataFeed; import net.corda.core.messaging.*;
import net.corda.core.node.services.Vault; import net.corda.core.node.services.*;
import net.corda.core.node.services.VaultQueryException;
import net.corda.core.node.services.VaultQueryService;
import net.corda.core.node.services.VaultService;
import net.corda.core.node.services.vault.*; import net.corda.core.node.services.vault.*;
import net.corda.core.node.services.vault.QueryCriteria.LinearStateQueryCriteria; import net.corda.core.node.services.vault.QueryCriteria.*;
import net.corda.core.node.services.vault.QueryCriteria.VaultCustomQueryCriteria; import net.corda.core.schemas.*;
import net.corda.core.node.services.vault.QueryCriteria.VaultQueryCriteria; import net.corda.core.schemas.testing.*;
import net.corda.core.schemas.MappedSchema; import net.corda.core.transactions.*;
import net.corda.core.schemas.testing.DummyLinearStateSchemaV1; import net.corda.core.utilities.*;
import net.corda.core.utilities.OpaqueBytes; import net.corda.node.services.database.*;
import net.corda.core.transactions.SignedTransaction; import net.corda.node.services.schema.*;
import net.corda.core.transactions.WireTransaction; import net.corda.schemas.*;
import net.corda.node.services.database.HibernateConfiguration; import net.corda.testing.*;
import net.corda.node.services.schema.NodeSchemaService; import net.corda.testing.contracts.*;
import net.corda.schemas.CashSchemaV1; import net.corda.testing.node.*;
import net.corda.testing.TestConstants; import org.jetbrains.annotations.*;
import net.corda.testing.contracts.VaultFiller; import org.jetbrains.exposed.sql.*;
import net.corda.testing.node.MockServices; import org.junit.*;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.exposed.sql.Database;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import rx.Observable; import rx.Observable;
import java.io.Closeable; import java.io.*;
import java.io.IOException; import java.lang.reflect.*;
import java.lang.reflect.Field;
import java.util.*; import java.util.*;
import java.util.stream.Collectors; import java.util.stream.*;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import static net.corda.contracts.asset.CashKt.getDUMMY_CASH_ISSUER; import static net.corda.contracts.asset.CashKt.*;
import static net.corda.contracts.asset.CashKt.getDUMMY_CASH_ISSUER_KEY; import static net.corda.core.contracts.ContractsDSL.*;
import static net.corda.testing.CoreTestUtils.getBOC; import static net.corda.core.node.services.vault.QueryCriteriaUtils.*;
import static net.corda.testing.CoreTestUtils.getBOC_KEY; import static net.corda.node.utilities.DatabaseSupportKt.*;
import static net.corda.testing.CoreTestUtils.getBOC_PUBKEY;
import static net.corda.core.contracts.ContractsDSL.USD;
import static net.corda.core.node.services.vault.QueryCriteriaUtils.MAX_PAGE_SIZE;
import static net.corda.node.utilities.DatabaseSupportKt.configureDatabase;
import static net.corda.node.utilities.DatabaseSupportKt.transaction; import static net.corda.node.utilities.DatabaseSupportKt.transaction;
import static net.corda.testing.CoreTestUtils.getMEGA_CORP; import static net.corda.testing.CoreTestUtils.*;
import static net.corda.testing.CoreTestUtils.getMEGA_CORP_KEY; import static net.corda.testing.node.MockServicesKt.*;
import static net.corda.testing.node.MockServicesKt.makeTestDataSourceProperties;
import static net.corda.core.utilities.ByteArrays.toHexString; import static net.corda.core.utilities.ByteArrays.toHexString;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.*;
public class VaultQueryJavaTests { public class VaultQueryJavaTests {
@ -146,7 +129,7 @@ public class VaultQueryJavaTests {
List<StateRef> stateRefs = stateRefsStream.collect(Collectors.toList()); List<StateRef> stateRefs = stateRefsStream.collect(Collectors.toList());
SortAttribute.Standard sortAttribute = new SortAttribute.Standard(Sort.CommonStateAttribute.STATE_REF_TXN_ID); SortAttribute.Standard sortAttribute = new SortAttribute.Standard(Sort.CommonStateAttribute.STATE_REF_TXN_ID);
Sort sorting = new Sort(Arrays.asList(new Sort.SortColumn(sortAttribute, Sort.Direction.ASC))); Sort sorting = new Sort(Collections.singletonList(new Sort.SortColumn(sortAttribute, Sort.Direction.ASC)));
VaultQueryCriteria criteria = new VaultQueryCriteria(Vault.StateStatus.UNCONSUMED, null, stateRefs); VaultQueryCriteria criteria = new VaultQueryCriteria(Vault.StateStatus.UNCONSUMED, null, stateRefs);
Vault.Page<DummyLinearContract.State> results = vaultQuerySvc.queryBy(DummyLinearContract.State.class, criteria, sorting); Vault.Page<DummyLinearContract.State> results = vaultQuerySvc.queryBy(DummyLinearContract.State.class, criteria, sorting);
@ -219,7 +202,7 @@ public class VaultQueryJavaTests {
QueryCriteria compositeCriteria1 = dealCriteriaAll.or(linearCriteriaAll); QueryCriteria compositeCriteria1 = dealCriteriaAll.or(linearCriteriaAll);
QueryCriteria compositeCriteria2 = vaultCriteria.and(compositeCriteria1); QueryCriteria compositeCriteria2 = vaultCriteria.and(compositeCriteria1);
PageSpecification pageSpec = new PageSpecification(0, MAX_PAGE_SIZE); PageSpecification pageSpec = new PageSpecification(DEFAULT_PAGE_NUM, MAX_PAGE_SIZE);
Sort.SortColumn sortByUid = new Sort.SortColumn(new SortAttribute.Standard(Sort.LinearStateAttribute.UUID), Sort.Direction.DESC); Sort.SortColumn sortByUid = new Sort.SortColumn(new SortAttribute.Standard(Sort.LinearStateAttribute.UUID), Sort.Direction.DESC);
Sort sorting = new Sort(ImmutableSet.of(sortByUid)); Sort sorting = new Sort(ImmutableSet.of(sortByUid));
Vault.Page<LinearState> results = vaultQuerySvc.queryBy(LinearState.class, compositeCriteria2, pageSpec, sorting); Vault.Page<LinearState> results = vaultQuerySvc.queryBy(LinearState.class, compositeCriteria2, pageSpec, sorting);
@ -232,6 +215,7 @@ public class VaultQueryJavaTests {
} }
@Test @Test
@SuppressWarnings("unchecked")
public void customQueryForCashStatesWithAmountOfCurrencyGreaterOrEqualThanQuantity() { public void customQueryForCashStatesWithAmountOfCurrencyGreaterOrEqualThanQuantity() {
transaction(database, tx -> { transaction(database, tx -> {
@ -328,7 +312,7 @@ public class VaultQueryJavaTests {
QueryCriteria dealOrLinearIdCriteria = dealCriteria.or(linearCriteria); QueryCriteria dealOrLinearIdCriteria = dealCriteria.or(linearCriteria);
QueryCriteria compositeCriteria = dealOrLinearIdCriteria.and(vaultCriteria); QueryCriteria compositeCriteria = dealOrLinearIdCriteria.and(vaultCriteria);
PageSpecification pageSpec = new PageSpecification(0, MAX_PAGE_SIZE); PageSpecification pageSpec = new PageSpecification(DEFAULT_PAGE_NUM, MAX_PAGE_SIZE);
Sort.SortColumn sortByUid = new Sort.SortColumn(new SortAttribute.Standard(Sort.LinearStateAttribute.UUID), Sort.Direction.DESC); Sort.SortColumn sortByUid = new Sort.SortColumn(new SortAttribute.Standard(Sort.LinearStateAttribute.UUID), Sort.Direction.DESC);
Sort sorting = new Sort(ImmutableSet.of(sortByUid)); Sort sorting = new Sort(ImmutableSet.of(sortByUid));
DataFeed<Vault.Page<ContractState>, Vault.Update> results = vaultQuerySvc.trackBy(ContractState.class, compositeCriteria, pageSpec, sorting); DataFeed<Vault.Page<ContractState>, Vault.Update> results = vaultQuerySvc.trackBy(ContractState.class, compositeCriteria, pageSpec, sorting);
@ -408,6 +392,7 @@ public class VaultQueryJavaTests {
*/ */
@Test @Test
@SuppressWarnings("unchecked")
public void aggregateFunctionsWithoutGroupClause() { public void aggregateFunctionsWithoutGroupClause() {
transaction(database, tx -> { transaction(database, tx -> {
@ -452,6 +437,7 @@ public class VaultQueryJavaTests {
} }
@Test @Test
@SuppressWarnings("unchecked")
public void aggregateFunctionsWithSingleGroupClause() { public void aggregateFunctionsWithSingleGroupClause() {
transaction(database, tx -> { transaction(database, tx -> {
@ -472,11 +458,11 @@ public class VaultQueryJavaTests {
Field pennies = CashSchemaV1.PersistentCashState.class.getDeclaredField("pennies"); Field pennies = CashSchemaV1.PersistentCashState.class.getDeclaredField("pennies");
Field currency = CashSchemaV1.PersistentCashState.class.getDeclaredField("currency"); Field currency = CashSchemaV1.PersistentCashState.class.getDeclaredField("currency");
QueryCriteria sumCriteria = new VaultCustomQueryCriteria(Builder.sum(pennies, Arrays.asList(currency))); QueryCriteria sumCriteria = new VaultCustomQueryCriteria(Builder.sum(pennies, Collections.singletonList(currency)));
QueryCriteria countCriteria = new VaultCustomQueryCriteria(Builder.count(pennies)); QueryCriteria countCriteria = new VaultCustomQueryCriteria(Builder.count(pennies));
QueryCriteria maxCriteria = new VaultCustomQueryCriteria(Builder.max(pennies, Arrays.asList(currency))); QueryCriteria maxCriteria = new VaultCustomQueryCriteria(Builder.max(pennies, Collections.singletonList(currency)));
QueryCriteria minCriteria = new VaultCustomQueryCriteria(Builder.min(pennies, Arrays.asList(currency))); QueryCriteria minCriteria = new VaultCustomQueryCriteria(Builder.min(pennies, Collections.singletonList(currency)));
QueryCriteria avgCriteria = new VaultCustomQueryCriteria(Builder.avg(pennies, Arrays.asList(currency))); QueryCriteria avgCriteria = new VaultCustomQueryCriteria(Builder.avg(pennies, Collections.singletonList(currency)));
QueryCriteria criteria = sumCriteria.and(countCriteria).and(maxCriteria).and(minCriteria).and(avgCriteria); QueryCriteria criteria = sumCriteria.and(countCriteria).and(maxCriteria).and(minCriteria).and(avgCriteria);
Vault.Page<Cash.State> results = vaultQuerySvc.queryBy(Cash.State.class, criteria); Vault.Page<Cash.State> results = vaultQuerySvc.queryBy(Cash.State.class, criteria);
@ -522,6 +508,7 @@ public class VaultQueryJavaTests {
} }
@Test @Test
@SuppressWarnings("unchecked")
public void aggregateFunctionsSumByIssuerAndCurrencyAndSortByAggregateSum() { public void aggregateFunctionsSumByIssuerAndCurrencyAndSortByAggregateSum() {
transaction(database, tx -> { transaction(database, tx -> {

56
node/src/test/kotlin/net/corda/node/services/vault/VaultQueryTests.kt
View File

@ -37,10 +37,8 @@ import org.assertj.core.api.Assertions.assertThat
import org.assertj.core.api.Assertions.assertThatThrownBy import org.assertj.core.api.Assertions.assertThatThrownBy
import org.bouncycastle.asn1.x500.X500Name import org.bouncycastle.asn1.x500.X500Name
import org.jetbrains.exposed.sql.Database import org.jetbrains.exposed.sql.Database
import org.junit.After import org.junit.*
import org.junit.Before import org.junit.rules.ExpectedException
import org.junit.Ignore
import org.junit.Test
import java.io.Closeable import java.io.Closeable
import java.lang.Thread.sleep import java.lang.Thread.sleep
import java.math.BigInteger import java.math.BigInteger
@ -825,7 +823,7 @@ class VaultQueryTests {
// Last page implies we need to perform a row count for the Query first, // Last page implies we need to perform a row count for the Query first,
// and then re-query for a given offset defined by (count - pageSize) // and then re-query for a given offset defined by (count - pageSize)
val pagingSpec = PageSpecification(9, 10) val pagingSpec = PageSpecification(10, 10)
val criteria = VaultQueryCriteria(status = Vault.StateStatus.ALL) val criteria = VaultQueryCriteria(status = Vault.StateStatus.ALL)
val results = vaultQuerySvc.queryBy<ContractState>(criteria, paging = pagingSpec) val results = vaultQuerySvc.queryBy<ContractState>(criteria, paging = pagingSpec)
@ -834,48 +832,54 @@ class VaultQueryTests {
} }
} }
@get:Rule
val expectedEx = ExpectedException.none()!!
// pagination: invalid page number // pagination: invalid page number
@Test(expected = VaultQueryException::class) @Test
fun `invalid page number`() { fun `invalid page number`() {
expectedEx.expect(VaultQueryException::class.java)
expectedEx.expectMessage("Page specification: invalid page number")
database.transaction { database.transaction {
services.fillWithSomeTestCash(100.DOLLARS, DUMMY_NOTARY, 100, 100, Random(0L)) services.fillWithSomeTestCash(100.DOLLARS, DUMMY_NOTARY, 100, 100, Random(0L))
val pagingSpec = PageSpecification(-1, 10) val pagingSpec = PageSpecification(0, 10)
val criteria = VaultQueryCriteria(status = Vault.StateStatus.ALL) val criteria = VaultQueryCriteria(status = Vault.StateStatus.ALL)
val results = vaultQuerySvc.queryBy<ContractState>(criteria, paging = pagingSpec) vaultQuerySvc.queryBy<ContractState>(criteria, paging = pagingSpec)
assertThat(results.states).hasSize(10) // should retrieve states 90..99
} }
} }
// pagination: invalid page size // pagination: invalid page size
@Test(expected = VaultQueryException::class) @Test
fun `invalid page size`() { fun `invalid page size`() {
expectedEx.expect(VaultQueryException::class.java)
expectedEx.expectMessage("Page specification: invalid page size")
database.transaction { database.transaction {
services.fillWithSomeTestCash(100.DOLLARS, DUMMY_NOTARY, 100, 100, Random(0L)) services.fillWithSomeTestCash(100.DOLLARS, DUMMY_NOTARY, 100, 100, Random(0L))
val pagingSpec = PageSpecification(0, MAX_PAGE_SIZE + 1) val pagingSpec = PageSpecification(DEFAULT_PAGE_NUM, MAX_PAGE_SIZE + 1) // overflow = -2147483648
val criteria = VaultQueryCriteria(status = Vault.StateStatus.ALL) val criteria = VaultQueryCriteria(status = Vault.StateStatus.ALL)
vaultQuerySvc.queryBy<ContractState>(criteria, paging = pagingSpec) vaultQuerySvc.queryBy<ContractState>(criteria, paging = pagingSpec)
assertFails { }
} }
} }
// pagination: out or range request (page number * page size) > total rows available // pagination not specified but more than DEFAULT_PAGE_SIZE results available (fail-fast test)
@Test(expected = VaultQueryException::class) @Test
fun `out of range page request`() { fun `pagination not specified but more than default results available`() {
expectedEx.expect(VaultQueryException::class.java)
expectedEx.expectMessage("Please specify a `PageSpecification`")
database.transaction { database.transaction {
services.fillWithSomeTestCash(100.DOLLARS, DUMMY_NOTARY, 100, 100, Random(0L)) services.fillWithSomeTestCash(201.DOLLARS, DUMMY_NOTARY, 201, 201, Random(0L))
val pagingSpec = PageSpecification(10, 10) // this requests results 101 .. 110
val criteria = VaultQueryCriteria(status = Vault.StateStatus.ALL) val criteria = VaultQueryCriteria(status = Vault.StateStatus.ALL)
val results = vaultQuerySvc.queryBy<ContractState>(criteria, paging = pagingSpec) vaultQuerySvc.queryBy<ContractState>(criteria)
assertFails { println("Query should throw an exception [${results.states.count()}]") }
} }
} }
@ -1776,7 +1780,7 @@ class VaultQueryTests {
updates updates
} }
updates?.expectEvents { updates.expectEvents {
sequence( sequence(
expect { (consumed, produced, flowId) -> expect { (consumed, produced, flowId) ->
require(flowId == null) {} require(flowId == null) {}
@ -1823,7 +1827,7 @@ class VaultQueryTests {
updates updates
} }
updates?.expectEvents { updates.expectEvents {
sequence( sequence(
expect { (consumed, produced, flowId) -> expect { (consumed, produced, flowId) ->
require(flowId == null) {} require(flowId == null) {}
@ -1870,7 +1874,7 @@ class VaultQueryTests {
updates updates
} }
updates?.expectEvents { updates.expectEvents {
sequence( sequence(
expect { (consumed, produced, flowId) -> expect { (consumed, produced, flowId) ->
require(flowId == null) {} require(flowId == null) {}
@ -1926,7 +1930,7 @@ class VaultQueryTests {
updates updates
} }
updates?.expectEvents { updates.expectEvents {
sequence( sequence(
expect { (consumed, produced, flowId) -> expect { (consumed, produced, flowId) ->
require(flowId == null) {} require(flowId == null) {}
@ -1976,7 +1980,7 @@ class VaultQueryTests {
updates updates
} }
updates?.expectEvents { updates.expectEvents {
sequence( sequence(
expect { (consumed, produced, flowId) -> expect { (consumed, produced, flowId) ->
require(flowId == null) {} require(flowId == null) {}