SQL Azure database compatibility (#113)

* Coin selection for SQL Server/Azure
* Transaction isolation level change for updateInfoDB in PersistentNetworkMapCache from "repeatableRead" to "readCommitted"
* Configuration option to prefix all Hibernate generated SQL with a schema name via a configuration property `database.schema`
* Fix negative value in SELECT TOP query (fix for Oracle db)

core/src/main/kotlin/net/corda/core/internal/schemas/NodeInfoSchema.kt

@@ -103,7 +103,6 @@ object NodeInfoSchemaV1 : MappedSchema(
             @Column(name = "party_cert_binary")
             val partyCertBinary: ByteArray,
 
-
             val isMain: Boolean,
 
             @ManyToMany(mappedBy = "legalIdentitiesAndCerts", cascade = arrayOf(CascadeType.ALL)) // ManyToMany because of distributed services.

docs/source/changelog.rst

@@ -93,6 +93,10 @@ UNRELEASED
 
 * ``waitForAllNodesToFinish()`` method in ``DriverDSLExposedInterface`` has instead become a parameter on driver creation.
 
+* Enterprise Corda only: Compatibility with SQL Server 2017 and SQL Azure databases.
+
+* Enterprise Corda only: node configuration property ``database.schema`` and documented existing database properties.
+
 .. _changelog_v1:
 
 Release 1.0

docs/source/corda-configuration-file.rst

@@ -72,6 +72,19 @@ path to the node's base directory.
     Currently the defaults in ``/node/src/main/resources/reference.conf`` are as shown in the first example. This is currently
     the only configuration that has been tested, although in the future full support for other storage layers will be validated.
 
+:database:  This section is used to configure JDBC and Hibernate related properties:
+
+        :initDatabase: ``true`` - triggers creation of the database schema at a node startup, ``false`` - existing database schema is not altered,
+                    the value is mapped to Hibernate property 'hibernate.hbm2ddl.auto' (``true`` -> update, ``false`` -> validate)
+
+        :schema: (optional) some database providers require a schema name when generating DDL and SQL statements.
+             (the value is passed to Hibernate property 'hibernate.hbm2ddl.auto').
+
+        :transactionIsolationLevel: Database transaction isolation level ( allowed values ``none``, ``readUncommitted, ``readCommitted``,
+             ``repeatableRead``, ``serializable``), the default isolation level is ``repeatableRead``.
+
+        :serverNameTablePrefix: (optional) A prefix added to each database table.
+
 :messagingServerAddress: The address of the ArtemisMQ broker instance. If not provided the node will run one locally.
 
 :p2pAddress: The host and port on which the node is available for protocol operations over ArtemisMQ.

docs/source/node-database.rst

@@ -1,7 +1,10 @@
 Node database
 =============
 
-Currently, nodes store their data in an H2 database. In the future, we plan to support a wide range of databases.
+Default in-memory database
+--------------------------
+
+By default nodes store their data in an H2 database.
 
 You can connect directly to a running node's database to see its stored states, transactions and attachments as
 follows:
@@ -25,4 +28,33 @@ follows:
 * Paste this string into the JDBC URL field and click ``Connect``, using the default username and password.
 
 You will be presented with a web interface that shows the contents of your node's storage and vault, and provides an
-interface for you to query them using SQL.
+interface for you to query them using SQL.
+
+Standalone database
+-------------------
+
+To run a node against a remote database modify node JDBC connection properties in `dataSourceProperties` entry
+and Hibernate properties in `database` entry - see `:ref:`dataSourceProperties`.
+
+SQL Azure and SQL Server
+````````````````````````
+Corda supports SQL Server 2017 adn SQL Azure.
+The minimum transaction isolation level ``database.transactionIsolationLevel`` is 'readCommitted'.
+The property ``database.schema`` is optional.
+Corda ships with Microsoft JDBC Driver 6.2 for SQLServer out-of-the-box.
+
+Example node configuration for SQL Azure:
+
+.. sourcecode:: none
+
+dataSourceProperties {
+    dataSourceClassName = "com.microsoft.sqlserver.jdbc.SQLServerDataSource"
+    dataSourceProperties.dataSource.url = "jdbc:sqlserver://[DATABASE_SERVER].database.windows.net:1433;databaseName=[DATABASE];
+        encrypt=true;trustServerCertificate=false;hostNameInCertificate=*.database.windows.net;loginTimeout=30"
+    dataSourceProperties.dataSource.user = [USER]
+    dataSourceProperties.dataSource.password = [PASSWORD]
+}
+database {
+    transactionIsolationLevel = "readCommitted"
+    schema = [SCHEMA]
+}

finance/src/main/kotlin/net/corda/finance/contracts/asset/cash/selection/CashSelectionSQLServerImpl.kt

@@ -0,0 +1,69 @@
+package net.corda.finance.contracts.asset.cash.selection
+
+import net.corda.core.contracts.Amount
+import net.corda.core.identity.AbstractParty
+import net.corda.core.identity.Party
+import net.corda.core.utilities.OpaqueBytes
+import net.corda.core.utilities.contextLogger
+import net.corda.core.utilities.toBase58String
+import java.sql.Connection
+import java.sql.DatabaseMetaData
+import java.sql.ResultSet
+import java.util.*
+
+/**
+ * SQL Server / SQL Azure
+ */
+class CashSelectionSQLServerImpl : AbstractCashSelection() {
+
+    companion object {
+        val JDBC_DRIVER_NAME = "Microsoft JDBC Driver 6.2 for SQL Server"
+        private val log = contextLogger()
+    }
+
+    override fun isCompatible(metaData: DatabaseMetaData): Boolean {
+        return metaData.driverName == JDBC_DRIVER_NAME
+    }
+
+    override fun toString() = "${this::class.java} for $JDBC_DRIVER_NAME"
+
+    override fun executeQuery(connection: Connection, amount: Amount<Currency>, lockId: UUID, notary: Party?,
+                              onlyFromIssuerParties: Set<AbstractParty>, withIssuerRefs: Set<OpaqueBytes>) : ResultSet {
+
+        val selectJoin = """
+            WITH row(transaction_id, output_index, pennies, total, lock_id) AS
+            (
+            SELECT vs.transaction_id, vs.output_index, ccs.pennies,
+            SUM(ccs.pennies) OVER (ORDER BY ccs.transaction_id RANGE UNBOUNDED PRECEDING), vs.lock_id
+            FROM contract_cash_states AS ccs, vault_states AS vs
+            WHERE vs.transaction_id = ccs.transaction_id AND vs.output_index = ccs.output_index
+                AND vs.state_status = 0
+                AND ccs.ccy_code = ?
+                AND (vs.lock_id = ? OR vs.lock_id is null)"""+
+                (if (notary != null)
+                    " AND vs.notary_name = ?" else "") +
+                (if (onlyFromIssuerParties.isNotEmpty())
+                    " AND ccs.issuer_key IN (?)" else "") +
+                (if (withIssuerRefs.isNotEmpty())
+                    " AND ccs.issuer_ref IN (?)" else "") +
+                """)
+            SELECT row.transaction_id, row.output_index, row.pennies, row.total, row.lock_id
+            FROM row where row.total <= ? + row.pennies"""
+
+        // Use prepared statement for protection against SQL Injection
+        val psSelectJoin = connection.prepareStatement(selectJoin)
+        var pIndex = 0
+        psSelectJoin.setString(++pIndex, amount.token.currencyCode)
+        psSelectJoin.setString(++pIndex, lockId.toString())
+        if (notary != null)
+            psSelectJoin.setString(++pIndex, notary.name.toString())
+        if (onlyFromIssuerParties.isNotEmpty())
+            psSelectJoin.setObject(++pIndex, onlyFromIssuerParties.map { it.owningKey.toBase58String() as Any}.toTypedArray() )
+        if (withIssuerRefs.isNotEmpty())
+            psSelectJoin.setObject(++pIndex, withIssuerRefs.map { it.bytes as Any }.toTypedArray())
+        psSelectJoin.setLong(++pIndex, amount.quantity)
+        log.debug(selectJoin)
+
+        return psSelectJoin.executeQuery()
+    }
+}

finance/src/main/resources/META-INF/services/net.corda.finance.contracts.asset.cash.selection.AbstractCashSelection

@@ -1,3 +1,4 @@
 net.corda.finance.contracts.asset.cash.selection.CashSelectionH2Impl
 net.corda.finance.contracts.asset.cash.selection.CashSelectionMySQLImpl
-net.corda.finance.contracts.asset.cash.selection.CashSelectionPostgreSQLImpl
+net.corda.finance.contracts.asset.cash.selection.CashSelectionPostgreSQLImpl
+net.corda.finance.contracts.asset.cash.selection.CashSelectionSQLServerImpl

node/build.gradle

@@ -162,6 +162,10 @@ dependencies {
     // Jsh: A SSH implementation for tunneling inbound traffic via a relay
     compile group: 'com.jcraft', name: 'jsch', version: '0.1.54'
 
+    //TODO remove once we can put driver jar into a predefined directory
+    //JDBC driver can be passed to the Node at startup using setting the jarDirs property in the Node configuration file.
+    runtime 'com.microsoft.sqlserver:mssql-jdbc:6.2.1.jre8'
+
     // Integration test helpers
     integrationTestCompile "junit:junit:$junit_version"
     integrationTestCompile "org.assertj:assertj-core:${assertj_version}"

node/src/integration-test/kotlin/net/corda/test/node/NodeStatePersistenceTests.kt

@@ -101,10 +101,10 @@ object MessageSchemaV1 : MappedSchema(
     @Entity
     @Table(name = "messages")
     class PersistentMessage(
-            @Column(name = "by")
+            @Column(name = "message_by")
             var by: String,
 
-            @Column(name = "value")
+            @Column(name = "message_value")
             var value: String
     ) : PersistentState()
 }

node/src/main/kotlin/net/corda/node/services/network/PersistentNetworkMapCache.kt

@@ -29,6 +29,7 @@ import org.hibernate.Session
 import rx.Observable
 import rx.subjects.PublishSubject
 import java.security.PublicKey
+import java.sql.Connection
 import java.util.*
 import javax.annotation.concurrent.ThreadSafe
 import kotlin.collections.HashMap
@@ -238,7 +239,7 @@ open class PersistentNetworkMapCache(private val database: CordaPersistence) : S
         //  network map registration on network map node)
         database.dataSource.connection.use {
             val session = database.entityManagerFactory.withOptions().connection(it.apply {
-                transactionIsolation = 1
+                transactionIsolation = Connection.TRANSACTION_READ_COMMITTED
             }).openSession()
             session.use {
                 val tx = session.beginTransaction()

node/src/main/kotlin/net/corda/node/services/persistence/HibernateConfiguration.kt

@@ -60,6 +60,11 @@ class HibernateConfiguration(val schemaService: SchemaService, private val datab
                 .setProperty("hibernate.format_sql", "true")
                 .setProperty("hibernate.connection.isolation", transactionIsolationLevel.toString())
 
+        if (databaseProperties.getProperty("schema") != null) {
+            // This property helps 'hibernate.hbm2ddl.auto' to work properly when many schemas have similar table names.
+            config.setProperty("hibernate.default_schema", databaseProperties.getProperty("schema"))
+        }
+
         schemas.forEach { schema ->
             // TODO: require mechanism to set schemaOptions (databaseSchema, tablePrefix) which are not global to session
             schema.mappedTypes.forEach { config.addAnnotatedClass(it) }

node/src/main/kotlin/net/corda/node/services/vault/NodeVaultService.kt

@@ -423,7 +423,7 @@ class NodeVaultService(
                 if (paging.pageSize < 1) throw VaultQueryException("Page specification: invalid page size ${paging.pageSize} [must be a value between 1 and $MAX_PAGE_SIZE]")
             }
 
-            query.firstResult = (paging.pageNumber - 1) * paging.pageSize
+            query.firstResult = if (paging.pageNumber > 0) (paging.pageNumber - 1) * paging.pageSize else 0 //some DB don't allow a negative value in SELECT TOP query
             query.maxResults = paging.pageSize + 1  // detection too many results
 
             // execution