Adding fetchSize as an optional configuration option on driver config and session config.

Author: Zhen Li

src/driver.js

@@ -30,9 +30,16 @@ import {
 } from './internal/pool-config'
 import Session from './session'
 import RxSession from './session-rx'
+import { ALL } from './internal/request-message'
 
 const DEFAULT_MAX_CONNECTION_LIFETIME = 60 * 60 * 1000 // 1 hour
 
+/**
+ * The default record fetch size. This is used in Bolt V4 protocol to pull query execution result in batches.
+ * @type {number}
+ */
+const DEFAULT_FETCH_SIZE = 1000
+
 /**
  * Constant that represents read session access mode.
  * Should be used like this: `driver.session({ defaultAccessMode: neo4j.session.READ })`.
@@ -132,19 +139,23 @@ class Driver {
    * @param {string} param.defaultAccessMode=WRITE - the access mode of this session, allowed values are {@link READ} and {@link WRITE}.
    * @param {string|string[]} param.bookmarks - the initial reference or references to some previous
    * transactions. Value is optional and absence indicates that that the bookmarks do not exist or are unknown.
+   * @param {number} param.fetchSize - the record fetch size of each batch of this session.
+   * Use {@link ALL} to always pull all records in one batch. This will override the config value set on driver config.
    * @param {string} param.database - the database this session will operate on.
    * @return {Session} new session.
    */
   session ({
     defaultAccessMode = WRITE,
     bookmarks: bookmarkOrBookmarks,
-    database = ''
+    database = '',
+    fetchSize
   } = {}) {
     return this._newSession({
       defaultAccessMode,
       bookmarkOrBookmarks,
       database,
-      reactive: false
+      reactive: false,
+      fetchSize: validateFetchSizeValue(fetchSize, this._config.fetchSize)
     })
   }
 
@@ -229,7 +240,13 @@ class Driver {
   /**
    * @private
    */
-  _newSession ({ defaultAccessMode, bookmarkOrBookmarks, database, reactive }) {
+  _newSession ({
+    defaultAccessMode,
+    bookmarkOrBookmarks,
+    database,
+    reactive,
+    fetchSize
+  }) {
     const sessionMode = Driver._validateSessionMode(defaultAccessMode)
     const connectionProvider = this._getOrCreateConnectionProvider()
     const bookmark = bookmarkOrBookmarks
@@ -241,7 +258,8 @@ class Driver {
       connectionProvider,
       bookmark,
       config: this._config,
-      reactive
+      reactive,
+      fetchSize
     })
   }
 
@@ -277,6 +295,10 @@ function sanitizeConfig (config) {
     config.connectionAcquisitionTimeout,
     DEFAULT_ACQUISITION_TIMEOUT
   )
+  config.fetchSize = validateFetchSizeValue(
+    config.fetchSize,
+    DEFAULT_FETCH_SIZE
+  )
 }
 
 /**
@@ -293,6 +315,23 @@ function sanitizeIntValue (rawValue, defaultWhenAbsent) {
   }
 }
 
+/**
+ * @private
+ */
+function validateFetchSizeValue (rawValue, defaultWhenAbsent) {
+  const fetchSize = parseInt(rawValue, 10)
+  if (fetchSize > 0 || fetchSize === ALL) {
+    return fetchSize
+  } else if (fetchSize === 0 || fetchSize < 0) {
+    throw new Error(
+      'The fetch size can only be a positive value or -1 for ALL. However fetchSize = ' +
+        fetchSize
+    )
+  } else {
+    return defaultWhenAbsent
+  }
+}
+
 export { Driver, READ, WRITE }
 
 export default Driver

src/internal/bolt-protocol-v4.js

@@ -17,7 +17,7 @@
  * limitations under the License.
  */
 import BoltProtocolV3 from './bolt-protocol-v3'
-import RequestMessage from './request-message'
+import RequestMessage, { ALL } from './request-message'
 import { ResultStreamObserver } from './stream-observers'
 import { BOLT_PROTOCOL_V4 } from './constants'
 
@@ -69,7 +69,8 @@ export default class BoltProtocol extends BoltProtocolV3 {
       beforeComplete,
       afterComplete,
       flush = true,
-      reactive = false
+      reactive = false,
+      fetchSize = ALL
     } = {}
   ) {
     const observer = new ResultStreamObserver({

src/internal/request-message.js

@@ -43,7 +43,7 @@ const READ_MODE = 'r'
 /* eslint-enable no-unused-vars */
 
 const NO_STATEMENT_ID = -1
-const ALL = -1
+export const ALL = -1
 
 export default class RequestMessage {
   constructor (signature, fields, toString) {
@@ -220,19 +220,19 @@ export default class RequestMessage {
 function buildTxMetadata (bookmark, txConfig, database, mode) {
   const metadata = {}
   if (!bookmark.isEmpty()) {
-    metadata['bookmarks'] = bookmark.values()
+    metadata.bookmarks = bookmark.values()
   }
   if (txConfig.timeout) {
-    metadata['tx_timeout'] = txConfig.timeout
+    metadata.tx_timeout = txConfig.timeout
   }
   if (txConfig.metadata) {
-    metadata['tx_metadata'] = txConfig.metadata
+    metadata.tx_metadata = txConfig.metadata
   }
   if (database) {
-    metadata['db'] = assertString(database, 'database')
+    metadata.db = assertString(database, 'database')
   }
   if (mode === ACCESS_MODE_READ) {
-    metadata['mode'] = READ_MODE
+    metadata.mode = READ_MODE
   }
   return metadata
 }
@@ -246,7 +246,7 @@ function buildTxMetadata (bookmark, txConfig, database, mode) {
 function buildStreamMetadata (stmtId, n) {
   const metadata = { n: int(n) }
   if (stmtId !== NO_STATEMENT_ID) {
-    metadata['qid'] = int(stmtId)
+    metadata.qid = int(stmtId)
   }
   return metadata
 }

src/internal/stream-observers.js

@@ -21,8 +21,7 @@ import Connection from './connection'
 import { newError, PROTOCOL_ERROR } from '../error'
 import { isString } from './util'
 import Integer from '../integer'
-
-const DefaultBatchSize = 50
+import { ALL } from './request-message'
 
 class StreamObserver {
   onNext (rawRecord) {}
@@ -63,7 +62,7 @@ class ResultStreamObserver extends StreamObserver {
     reactive = false,
     moreFunction,
     discardFunction,
-    batchSize = DefaultBatchSize,
+    batchSize = ALL,
     beforeError,
     afterError,
     beforeKeys,
@@ -108,7 +107,7 @@ class ResultStreamObserver extends StreamObserver {
    * @param {Array} rawRecord - An array with the raw record
    */
   onNext (rawRecord) {
-    let record = new Record(this._fieldKeys, rawRecord, this._fieldLookup)
+    const record = new Record(this._fieldKeys, rawRecord, this._fieldLookup)
     if (this._observers.some(o => o.onNext)) {
       this._observers.forEach(o => {
         if (o.onNext) {

src/session.js

@@ -49,18 +49,21 @@ class Session {
    * @param {string} args.database the database name
    * @param {Object} args.config={} - this driver configuration.
    * @param {boolean} args.reactive - whether this session should create reactive streams
+   * @param {number} args.fetchSize - defines how many records is pulled in each pulling batch
    */
   constructor ({
     mode,
     connectionProvider,
     bookmark,
     database,
     config,
-    reactive
+    reactive,
+    fetchSize
   }) {
     this._mode = mode
     this._database = database
     this._reactive = reactive
+    this._fetchSize = fetchSize
     this._readConnectionHolder = new ConnectionHolder({
       mode: ACCESS_MODE_READ,
       database,
@@ -107,7 +110,8 @@ class Session {
         mode: this._mode,
         database: this._database,
         afterComplete: this._onComplete,
-        reactive: this._reactive
+        reactive: this._reactive,
+        fetchSize: this._fetchSize
       })
     )
   }

test/driver.test.js

@@ -26,6 +26,7 @@ import {
 } from '../src/internal/pool-config'
 import { ServerVersion, VERSION_4_0_0 } from '../src/internal/server-version'
 import testUtils from './internal/test-utils'
+import { AuthToken } from '../types'
 
 describe('#integration driver', () => {
   let driver
@@ -253,6 +254,28 @@ describe('#integration driver', () => {
     )
   })
 
+  it('should validate fetch size in the config', async () => {
+    await validateConfigSanitizing({}, 1000)
+    await validateConfigSanitizing({ fetchSize: 42 }, 42)
+    await validateConfigSanitizing({ fetchSize: -1 }, -1)
+    await validateConfigSanitizing({ fetchSize: '42' }, 42)
+    await validateConfigSanitizing({ fetchSize: '-1' }, -1)
+  })
+
+  it('should fail when fetch size is negative', () => {
+    expect(() =>
+      neo4j.driver('bolt://localhost', sharedNeo4j.authToken, {
+        fetchSize: -77
+      })
+    ).toThrow()
+  })
+
+  it('should fail when fetch size is 0', () => {
+    expect(() =>
+      neo4j.driver('bolt://localhost', sharedNeo4j.authToken, { fetchSize: 0 })
+    ).toThrow()
+  })
+
   it('should discard closed connections', async () => {
     driver = neo4j.driver('bolt://localhost', sharedNeo4j.authToken)
 

types/driver.d.ts

@@ -49,9 +49,11 @@ declare interface Config {
   trust?: TrustStrategy
   trustedCertificates?: string[]
   knownHosts?: string
+  fetchSize?: number
   maxConnectionPoolSize?: number
   maxTransactionRetryTime?: number
   maxConnectionLifetime?: number
+  connectionAcquisitionTimeout?: number
   connectionTimeout?: number
   disableLosslessIntegers?: boolean
   logging?: LoggingConfig
@@ -71,6 +73,7 @@ declare interface Driver {
   }?: {
     defaultAccessMode?: SessionMode
     bookmarks?: string | string[]
+    fetchSize?: number
     database?: string
   }): Session