spline #1155 Introduce secondary transaction ID

arangodb-foxx-services/src/main/persistence/model.ts

@@ -101,7 +101,8 @@ export type TxNum = number
  */
 export type WriteTxInfo = {
  num: TxNum
- uid: TxId
+ uid: TxId // primary key - universally unique // TODO: double check with ArangoDB if it's indeed unique for entire collection life even if records are constantly removed
+ sid: TxId // secondary key - for deduplication purpose
  params: TxParams
 }
 

arangodb-foxx-services/src/main/services/execution-event-store.ts

@@ -70,14 +70,16 @@ export function storeExecutionEvent(progress: Progress): void {
  const progressEdge: Partial<ArangoDB.Edge> =
  edge(CollectionName.Progress, progress._key, CollectionName.ExecutionPlan, progress.planKey, progress._key)
 
- const wtxInfo: WriteTxInfo = TxManager.startWrite({
- execEventInfo: {
- _key: progress._key,
- planKey: progress.planKey,
- timestamp: progress.timestamp,
- error: progress.error,
- }
- })
+ const wtxInfo: WriteTxInfo = TxManager.startWrite(
+ progress._key,
+ {
+ execEventInfo: {
+ _key: progress._key,
+ planKey: progress.planKey,
+ timestamp: progress.timestamp,
+ error: progress.error,
+ }
+ })
 
  try {
  store.insertOne(progressWithPlanDetails, CollectionName.Progress, wtxInfo)

arangodb-foxx-services/src/main/services/execution-plan-store.ts

@@ -23,12 +23,15 @@ import { TxManager } from './txm'
 
 
 export function storeExecutionPlan(eppm: ExecutionPlanPersistentModel): void {
- withTimeTracking(`STORE PLAN ${eppm.executionPlan._key}`, () => {
- const txInfo: WriteTxInfo = TxManager.startWrite({
- execPlanInfo: {
- _key: eppm.executionPlan._key,
- }
- })
+ const execPlanKey = eppm.executionPlan._key
+ withTimeTracking(`STORE PLAN ${execPlanKey}`, () => {
+ const txInfo: WriteTxInfo = TxManager.startWrite(
+ execPlanKey,
+ {
+ execPlanInfo: {
+ _key: execPlanKey,
+ }
+ })
 
  try {
  // execution plan

arangodb-foxx-services/src/main/services/txm/subcribable-tx-manager-decorator.ts

@@ -16,7 +16,7 @@
 
 import { TxManager } from './tx-manager'
 import events from 'events'
-import { ReadTxInfo, TxAwareDocument, TxEvent, TxParams, WriteTxInfo } from '../../persistence/model'
+import { ReadTxInfo, TxAwareDocument, TxEvent, TxId, TxParams, WriteTxInfo } from '../../persistence/model'
 import * as Logger from '../../utils/logger'
 
 
@@ -64,8 +64,8 @@ export class SubscribableTxManagerDecorator implements TxManager {
  return this.internalTxManager.startRead()
  }
 
- startWrite(txParams: TxParams): WriteTxInfo {
- const wtxInfo = this.internalTxManager.startWrite(txParams)
+ startWrite(sid: TxId, txParams: TxParams): WriteTxInfo {
+ const wtxInfo = this.internalTxManager.startWrite(sid, txParams)
  this.eventsEmitter.emit(TxEvent.StartWrite, wtxInfo)
  return wtxInfo
  }

arangodb-foxx-services/src/main/services/txm/tx-manager-impl.ts

@@ -58,7 +58,7 @@ export class TxManagerImpl implements TxManager {
  return newCnt.curVal
  }
 
- startWrite(txParams: TxParams = {}): WriteTxInfo {
+ startWrite(sid: TxId, txParams: TxParams = {}): WriteTxInfo {
  // The following steps must be executed in the given exact order
  // (opposite to one for the READ transaction) as follows:
 
@@ -74,6 +74,7 @@ export class TxManagerImpl implements TxManager {
  const wtxInfo: WriteTxInfo = {
  num: txNum,
  uid: txId,
+ sid: sid,
  params: txParams,
  }
  Logger.debug('[TX] WRITE STARTED', wtxInfo)

arangodb-foxx-services/src/main/services/txm/tx-manager.ts

@@ -14,15 +14,18 @@
  * limitations under the License.
  */
 
-import { ReadTxInfo, TxAwareDocument, TxParams, WriteTxInfo } from '../../persistence/model'
+import { ReadTxInfo, TxAwareDocument, TxId, TxParams, WriteTxInfo } from '../../persistence/model'
 
 
 export interface TxManager {
  /**
  * Start new logical transaction for WRITE
+ * @param sid secondary transaction ID - serves for deduplication purposes.
+ * It MUST be unique if and only if the data written by the transaction is logically unique.
+ * @param txParams additional parameters associated and stored with this transaction info.
  * @return new WRITE transaction metadata
  */
- startWrite(txParams: TxParams): WriteTxInfo
+ startWrite(sid: TxId, txParams: TxParams): WriteTxInfo
 
  /**
  * Start new logical transaction for READ
@@ -45,7 +48,7 @@ export interface TxManager {
  /**
  * Checks if all the `docs` are visible from the `rtx` READ transaction.
  * The document is visible from the READ transaction if:
- * = the document was created out of any logical transaction, OR
+ * - the document was created out of any logical transaction, OR
  * - the doc's associated WRITE transaction has been committed
  * before the current READ one started.
  * @param rtx Read transaction info object

arangodb-foxx-services/tests/main/services/txm/subscribable-tx-manager-decorator.spec.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import { ReadTxInfo, TxAwareDocument, TxEvent, TxParams, WriteTxInfo } from '../../../../src/main/persistence/model'
+import { ReadTxInfo, TxAwareDocument, TxEvent, TxId, TxParams, WriteTxInfo } from '../../../../src/main/persistence/model'
 import { SubscribableTxManagerDecorator } from '../../../../src/main/services/txm/subcribable-tx-manager-decorator'
 
 
@@ -60,11 +60,12 @@ test('startRead() should be delegated', () => {
 })
 
 test('startWrite() should be delegated', () => {
+ const dummySID: TxId = "dummy_tx_sid"
  const dummyTxParams: TxParams = {}
  const dummyResult = {}
  mockTxManagerImpl.startWrite.mockReturnValue(dummyResult)
 
- const res = txManagerDecorator.startWrite(dummyTxParams)
+ const res = txManagerDecorator.startWrite(dummySID, dummyTxParams)
  expect(res).toBe(dummyResult)
 
  expect(mockTxManagerImpl.startRead.mock.calls.length).toBe(0)
@@ -73,7 +74,8 @@ test('startWrite() should be delegated', () => {
  expect(mockTxManagerImpl.rollback.mock.calls.length).toBe(0)
  expect(mockTxManagerImpl.isVisibleFromTx.mock.calls.length).toBe(0)
 
- expect(mockTxManagerImpl.startWrite.mock.lastCall[0]).toBe(dummyTxParams)
+ expect(mockTxManagerImpl.startWrite.mock.lastCall[0]).toBe(dummySID)
+ expect(mockTxManagerImpl.startWrite.mock.lastCall[1]).toBe(dummyTxParams)
 })
 
 test('commit() should be delegated', () => {
@@ -133,7 +135,7 @@ test('on(TX_START_WRITE)', () => {
  const dummyTxInfo: WriteTxInfo = {}
  mockTxManagerImpl.startWrite.mockReturnValue(dummyTxInfo)
 
- txManagerDecorator.startWrite({})
+ txManagerDecorator.startWrite(null, null)
 
  expect(startWriteHandler.mock.calls.length).toBe(1)
  expect(preCommitHandler.mock.calls.length).toBe(0)

persistence/src/main/scala/za/co/absa/spline/persistence/model/persistentDefs.scala

@@ -226,6 +226,10 @@ object CollectionDef {
  override def collectionType = CollectionType.DOCUMENT
 
  override def name: String = "txInfo"
+
+ override def indexDefs: Seq[IndexDef] = Seq(
+ IndexDef(Seq("sid"), (new PersistentIndexOptions).unique(true))
+ )
  }
 
 }