Skip to content

Commit 2ce4fd7

Browse files
tanjjjtanjjj
authored
#1909 fix missing timestamp (#1913)
#1909 fix missing timestamps
1 parent eb09a37 commit 2ce4fd7

File tree

10 files changed

+101
-81
lines changed

10 files changed

+101
-81
lines changed

‎plugin/virtualized-table-plugin/src/main/scala/org/finos/vuu/plugin/virtualized/table/VirtualizedSessionTableData.scala‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,9 @@ class VirtualizedSessionTableData(cacheSize: Int)(implicit clock: Clock) extends
1818
}
1919
}
2020

21-
override def update(key: String, update: RowData): TableData = {
21+
override def update(key: String, update: RowData): (TableData, RowData) = {
2222
rowCache.put(key, update)
23-
this
23+
(this, update)
2424
}
2525

2626
override def delete(key: String): TableData = {

‎vuu/src/main/scala/org/finos/vuu/core/table/InMemDataTable.scala‎

Lines changed: 22 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,9 @@ trait DataTable extends KeyedObservable[RowKeyUpdate] with RowSource {
2626
def updateCounter: Long
2727

2828
def newRow(key: String): RowBuilder
29+
2930
def rowBuilder: RowBuilder
31+
3032
def incrementUpdateCounter(): Unit
3133

3234
def indexForColumn(column: Column): Option[IndexedField[_]]
@@ -101,11 +103,17 @@ case class RowKeyUpdate(key: String, source: RowSource, isDelete: Boolean = fals
101103

102104
trait RowData {
103105
def key: String
106+
104107
def get(field: String): Any
108+
105109
def get(column: Column): Any
110+
106111
def getFullyQualified(column: Column): Any
112+
107113
def set(field: String, value: Any): RowData
114+
108115
def toArray(columns: List[Column]): Array[Any]
116+
109117
def size: Int
110118
}
111119

@@ -179,21 +187,21 @@ case class InMemDataTableData(data: ConcurrentHashMap[String, RowData], private
179187
protected def merge(update: RowData, data: RowData): RowData =
180188
MergeFunctions.mergeLeftToRight(update, data)
181189

182-
def update(key: String, update: RowData): TableData = {
190+
def update(key: String, update: RowData): (TableData, RowData) = {
183191

184192
val table = data.synchronized {
185193
val now = timeProvider.now()
186194
data.getOrDefault(key, EmptyRowData) match {
187195
case row: RowWithData =>
188196
val mergedData = merge(update, row)
189-
mergedData.set(DefaultColumnNames.LastUpdatedTimeColumnName, now)
190-
data.put(key, mergedData)
191-
InMemDataTableData(data, primaryKeyValues)
197+
val newRowData = mergedData.set(DefaultColumnNames.LastUpdatedTimeColumnName, now)
198+
data.put(key, newRowData)
199+
(InMemDataTableData(data, primaryKeyValues), newRowData)
192200
case EmptyRowData =>
193-
update.set(DefaultColumnNames.CreatedTimeColumnName, now)
194-
update.set(DefaultColumnNames.LastUpdatedTimeColumnName, now)
195-
data.put(key, update)
196-
InMemDataTableData(data, primaryKeyValues + key)
201+
var newRowData = update.set(DefaultColumnNames.CreatedTimeColumnName, now)
202+
newRowData = newRowData.set(DefaultColumnNames.LastUpdatedTimeColumnName, now)
203+
data.put(key, newRowData)
204+
(InMemDataTableData(data, primaryKeyValues + key), newRowData)
197205
}
198206

199207
}
@@ -232,6 +240,7 @@ class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider
232240
override def newRow(key: String): RowBuilder = {
233241
new InMemMapRowBuilder().setKey(key)
234242
}
243+
235244
override def rowBuilder: RowBuilder = new InMemMapRowBuilder
236245

237246
private def buildIndexForColumn(c: Column): IndexedField[_] = {
@@ -286,6 +295,7 @@ class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider
286295
@volatile protected var data: TableData = createDataTableData()
287296

288297
@volatile private var updateCounterInternal: Long = 0
298+
289299
override def updateCounter: Long = updateCounterInternal
290300

291301
override def incrementUpdateCounter(): Unit = updateCounterInternal += 1
@@ -355,6 +365,7 @@ class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider
355365
}
356366

357367
def columns(): Array[Column] = tableDef.columns
368+
358369
lazy val viewPortColumns: ViewPortColumns = ViewPortColumnCreator.create(this, tableDef.columns.map(_.name).toList)
359370

360371
private def updateIndices(rowkey: String, rowUpdate: RowData): Unit = {
@@ -414,8 +425,9 @@ class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider
414425
}
415426

416427
def update(rowkey: String, rowUpdate: RowData): Unit = {
417-
data = data.update(rowkey, rowUpdate)
418-
updateIndices(rowkey, rowUpdate)
428+
val updatedData = data.update(rowkey, rowUpdate)
429+
data = updatedData._1
430+
updateIndices(rowkey, updatedData._2)
419431
}
420432

421433
def delete(rowKey: String): RowData = {

‎vuu/src/main/scala/org/finos/vuu/core/table/TableData.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ package org.finos.vuu.core.table
22

33
trait TableData {
44
def dataByKey(key: String): RowData
5-
def update(key: String, update: RowData): TableData
5+
def update(key: String, update: RowData): (TableData, RowData)
66
def delete(key: String): TableData
77
def deleteAll(): TableData
88
def primaryKeyValues: TablePrimaryKeys

‎vuu/src/test/scala/org/finos/vuu/core/filter/type/BaseFilterTest.scala‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package org.finos.vuu.core.filter.`type`
22

3+
import org.finos.toolbox.time.TestFriendlyClock
34
import org.finos.vuu.core.sort.FilterAndSortFixture.{setupTable, setupTableWithCreationTime}
45
import org.finos.vuu.viewport.ViewPortColumns
56
import org.scalatest.featurespec.AnyFeatureSpec
@@ -22,10 +23,11 @@ class BaseFilterTest extends AnyFeatureSpec with Matchers {
2223
}
2324

2425
Scenario("Freeze filter applied") {
25-
26-
val table = setupTableWithCreationTime()
26+
val clock = new TestFriendlyClock(1000L)
27+
val table = setupTableWithCreationTime(List())(using clock)
28+
val now = clock.now()
2729
val permissionFilter = PermissionFilter("ric", Set("VOD.L","AAPL.L"))
28-
val baseFilter = BaseFilter(permissionFilter, Option(10001L))
30+
val baseFilter = BaseFilter(permissionFilter, Option(now))
2931

3032
val results = baseFilter.doFilter(table, table.primaryKeys, ViewPortColumns(table.columns().toList), true)
3133

‎vuu/src/test/scala/org/finos/vuu/core/filter/type/FrozenTimeFilterTest.scala‎

Lines changed: 17 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package org.finos.vuu.core.filter.`type`
22

33
import org.finos.toolbox.collection.array.ImmutableArray
4+
import org.finos.toolbox.time.TestFriendlyClock
45
import org.finos.vuu.core.sort.FilterAndSortFixture.{setupTable, setupTableWithCreationTime}
56
import org.finos.vuu.core.table.{DefaultColumnNames, EmptyTablePrimaryKeys}
67
import org.finos.vuu.feature.inmem.InMemTablePrimaryKeys
@@ -21,8 +22,10 @@ class FrozenTimeFilterTest extends AnyFeatureSpec with Matchers {
2122
}
2223

2324
Scenario("Freeze filter with created time column present") {
24-
val table = setupTableWithCreationTime()
25-
val frozenTimeFilter = FrozenTimeFilter(10001L)
25+
val clock = TestFriendlyClock(1000L)
26+
val table = setupTableWithCreationTime(List())(using clock)
27+
val now = clock.now()
28+
val frozenTimeFilter = FrozenTimeFilter(now)
2629

2730
val results = frozenTimeFilter.doFilter(table, table.primaryKeys, true)
2831

@@ -31,8 +34,10 @@ class FrozenTimeFilterTest extends AnyFeatureSpec with Matchers {
3134
}
3235

3336
Scenario("Freeze filter with indexed created time") {
34-
val table = setupTableWithCreationTime(List(DefaultColumnNames.CreatedTimeColumnName))
35-
val frozenTimeFilter = FrozenTimeFilter(10001L)
37+
val clock = new TestFriendlyClock(1000L)
38+
val table = setupTableWithCreationTime(List(DefaultColumnNames.CreatedTimeColumnName))(using clock)
39+
val now = clock.now()
40+
val frozenTimeFilter = FrozenTimeFilter(now)
3641

3742
val results = frozenTimeFilter.doFilter(table, table.primaryKeys, true)
3843

@@ -41,8 +46,10 @@ class FrozenTimeFilterTest extends AnyFeatureSpec with Matchers {
4146
}
4247

4348
Scenario("Freeze filter with indexed created time and not first in chain") {
44-
val table = setupTableWithCreationTime(List(DefaultColumnNames.CreatedTimeColumnName))
45-
val frozenTimeFilter = FrozenTimeFilter(10001L)
49+
val clock = new TestFriendlyClock(1000L)
50+
val table = setupTableWithCreationTime(List(DefaultColumnNames.CreatedTimeColumnName))(using clock)
51+
val now = clock.now()
52+
val frozenTimeFilter = FrozenTimeFilter(now)
4653

4754
val results = frozenTimeFilter.doFilter(table, InMemTablePrimaryKeys(ImmutableArray.from(Array("LDN-0003"))), false)
4855

@@ -51,8 +58,10 @@ class FrozenTimeFilterTest extends AnyFeatureSpec with Matchers {
5158
}
5259

5360
Scenario("Freeze filter with no input primary keys") {
54-
val table = setupTableWithCreationTime(List(DefaultColumnNames.CreatedTimeColumnName))
55-
val frozenTimeFilter = FrozenTimeFilter(10001L)
61+
val clock = new TestFriendlyClock(1000L)
62+
val table = setupTableWithCreationTime(List(DefaultColumnNames.CreatedTimeColumnName))(using clock)
63+
val now = clock.now()
64+
val frozenTimeFilter = FrozenTimeFilter(now)
5665

5766
val results = frozenTimeFilter.doFilter(table, EmptyTablePrimaryKeys, true)
5867

‎vuu/src/test/scala/org/finos/vuu/core/filter/type/PermissionFilterTest.scala‎

Lines changed: 17 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package org.finos.vuu.core.filter.`type`
22

33
import org.finos.toolbox.collection.array.ImmutableArray
4+
import org.finos.toolbox.time.TestFriendlyClock
45
import org.finos.vuu.core.sort.FilterAndSortFixture.{setupTable, setupTable2, setupTableWithCreationTime}
56
import org.finos.vuu.core.table.{DefaultColumnNames, EmptyTablePrimaryKeys, RowData}
67
import org.finos.vuu.feature.inmem.InMemTablePrimaryKeys
@@ -446,10 +447,10 @@ class PermissionFilterTest extends AnyFeatureSpec with Matchers {
446447
Feature("Contains filter - Epoch Timestamp") {
447448

448449
Scenario("Contains filter with no index") {
449-
450-
val table = setupTableWithCreationTime()
451-
452-
val filter = PermissionFilter(DefaultColumnNames.CreatedTimeColumnName, Set("10001"))
450+
val clock = TestFriendlyClock(1000L)
451+
val table = setupTableWithCreationTime(List())(using clock)
452+
val now = clock.now()
453+
val filter = PermissionFilter(DefaultColumnNames.CreatedTimeColumnName, Set(now.toString))
453454

454455
val result = filter.doFilter(table, table.primaryKeys, true)
455456

@@ -458,10 +459,10 @@ class PermissionFilterTest extends AnyFeatureSpec with Matchers {
458459
}
459460

460461
Scenario("Contains filter with no index, not first in chain") {
461-
462-
val table = setupTableWithCreationTime()
463-
464-
val filter = PermissionFilter(DefaultColumnNames.CreatedTimeColumnName, Set("10001"))
462+
val clock = TestFriendlyClock(1000L)
463+
val table = setupTableWithCreationTime(List())(using clock)
464+
val now = clock.now()
465+
val filter = PermissionFilter(DefaultColumnNames.CreatedTimeColumnName, Set(now.toString))
465466

466467
val result = filter.doFilter(table, InMemTablePrimaryKeys(ImmutableArray.from(Array("LDN-0001", "LDN-0003", "NYC-0002", "NYC-0004"))), false)
467468

@@ -470,10 +471,10 @@ class PermissionFilterTest extends AnyFeatureSpec with Matchers {
470471
}
471472

472473
Scenario("Contains filter with index") {
473-
474-
val table = setupTableWithCreationTime(List(DefaultColumnNames.CreatedTimeColumnName))
475-
476-
val filter = PermissionFilter(DefaultColumnNames.CreatedTimeColumnName, Set("10001"))
474+
val clock = new TestFriendlyClock(1000L)
475+
val table = setupTableWithCreationTime(List(DefaultColumnNames.CreatedTimeColumnName))(using clock)
476+
val now = clock.now()
477+
val filter = PermissionFilter(DefaultColumnNames.CreatedTimeColumnName, Set(now.toString))
477478

478479
val result = filter.doFilter(table, table.primaryKeys, true)
479480

@@ -482,10 +483,10 @@ class PermissionFilterTest extends AnyFeatureSpec with Matchers {
482483
}
483484

484485
Scenario("Contains filter with index, not first in chain") {
485-
486-
val table = setupTableWithCreationTime(List(DefaultColumnNames.CreatedTimeColumnName))
487-
488-
val filter = PermissionFilter(DefaultColumnNames.CreatedTimeColumnName, Set("10001"))
486+
val clock = new TestFriendlyClock(1000L)
487+
val table = setupTableWithCreationTime(List(DefaultColumnNames.CreatedTimeColumnName))(using clock)
488+
val now = clock.now()
489+
val filter = PermissionFilter(DefaultColumnNames.CreatedTimeColumnName, Set(now.toString))
489490

490491
val result = filter.doFilter(table, InMemTablePrimaryKeys(ImmutableArray.from(Array("LDN-0001", "LDN-0003", "NYC-0002", "NYC-0004"))), false)
491492

‎vuu/src/test/scala/org/finos/vuu/core/sort/FilterAndSortFixture.scala‎

Lines changed: 18 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,10 @@ package org.finos.vuu.core.sort
33
import org.finos.toolbox.collection.MapDiffResult
44
import org.finos.toolbox.jmx.MetricsProviderImpl
55
import org.finos.toolbox.text.{AsciiUtil, CodeGenUtil}
6-
import org.finos.toolbox.time.{DefaultClock, TestFriendlyClock}
6+
import org.finos.toolbox.time.{Clock, DefaultClock, TestFriendlyClock}
77
import org.finos.vuu.api.{Index, Indices, TableDef}
88
import org.finos.vuu.core.filter.FilterClause
9+
import org.finos.vuu.core.sort.FilterAndSortFixture.timeProvider
910
import org.finos.vuu.core.filter.`type`.AntlrBasedFilter
1011
import org.finos.vuu.core.table.DefaultColumnNames.CreatedTimeColumnName
1112
import org.finos.vuu.core.table.{Columns, InMemDataTable, RowWithData, ViewPortColumnCreator}
@@ -14,10 +15,6 @@ import org.scalatest.Assertions.fail
1415

1516
object FilterAndSortFixture {
1617
private val timeProvider = new TestFriendlyClock(10001L)
17-
val now: Long = timeProvider.now();
18-
val previousHour: Long = now - 3600000;
19-
val nextHour: Long = now + 3600000;
20-
2118

2219
def getFilteredRows(table: InMemDataTable, clause: FilterClause): Iterable[RowWithData] = {
2320
val vpColumns = ViewPortColumnCreator.create(table, table.columns().map(_.name).toList)
@@ -76,7 +73,7 @@ object FilterAndSortFixture {
7673
row("tradeTime" -> 5L, "quantity" -> 100, "price" -> 180.50, "side" -> 'B', "ric" -> "BT.L", "orderId" -> "LDN-0008", "onMkt" -> true, "trader" -> "chris", "ccyCross" -> "GBPUSD"),
7774
row("tradeTime" -> 6L, "quantity" -> 100, "price" -> 94.12, "side" -> 'S', "ric" -> "VOD.L", "orderId" -> "NYC-0002", "onMkt" -> false, "trader" -> "steve", "ccyCross" -> "GBPUSD"),
7875
row("tradeTime" -> 6L, "quantity" -> 100, "price" -> 94.12, "side" -> 'B', "ric" -> "VOD.L", "orderId" -> "NYC-0010", "onMkt" -> true, "trader" -> "steve", "ccyCross" -> "GBPUSD")
79-
)
76+
)(using timeProvider)
8077
}
8178

8279
def setupTable2(): InMemDataTable = {
@@ -92,25 +89,27 @@ object FilterAndSortFixture {
9289
row("tradeTime" -> 6L, "quantity" -> null, "price" -> 94.12, "side" -> 'B', "ric" -> "VOD\\L", "orderId" -> "NYC-0012", "onMkt" -> true, "trader" -> "steve", "ccyCross" -> "GBPUSD"),
9390
//unicode in trade name and special char in ccycross
9491
row("tradeTime" -> 6L, "quantity" -> null, "price" -> 94.12, "side" -> 'S', "ric" -> "VOD\\L", "orderId" -> "NYC-0013", "onMkt" -> true, "trader" -> "rahúl", "ccyCross" -> "$GBPUSD")
95-
)
92+
)(using timeProvider)
9693
}
9794

9895
def setupTableWithCreationTime(): InMemDataTable = {
99-
setupTableWithCreationTime(List())
96+
val clock = new TestFriendlyClock(1000L)
97+
setupTableWithCreationTime(List())(using clock)
10098
}
10199

102-
def setupTableWithCreationTime(indices: List[String]): InMemDataTable = {
103-
setupTable(indices,
104-
row("tradeTime" -> 5L, "quantity" -> 500, "price" -> 283.10, "side" -> 'B', "ric" -> "AAPL.L", "orderId" -> "NYC-0004", "onMkt" -> false, "trader" -> "chris", "ccyCross" -> "GBPUSD", CreatedTimeColumnName -> previousHour),
105-
row("tradeTime" -> 2L, "quantity" -> 100, "price" -> 94.12, "side" -> 'S', "ric" -> "VOD.L", "orderId" -> "LDN-0001", "onMkt" -> true, "trader" -> "chris", "ccyCross" -> "GBPUSD", CreatedTimeColumnName -> now),
106-
row("tradeTime" -> 1L, "quantity" -> 100, "price" -> 180.50, "side" -> 'B', "ric" -> "BT.L", "orderId" -> "LDN-0002", "onMkt" -> true, "trader" -> "chris", "ccyCross" -> "GBPUSD", CreatedTimeColumnName -> nextHour),
107-
row("tradeTime" -> 3L, "quantity" -> 100, "price" -> 94.12, "side" -> 'S', "ric" -> "VOD.L", "orderId" -> "LDN-0003", "onMkt" -> true, "trader" -> "steve", "ccyCross" -> "GBPUSD", CreatedTimeColumnName -> previousHour),
108-
row("tradeTime" -> 5L, "quantity" -> 100, "price" -> 180.50, "side" -> 'B', "ric" -> "BT.L", "orderId" -> "LDN-0008", "onMkt" -> true, "trader" -> "steve", "ccyCross" -> "GBPUSD", CreatedTimeColumnName -> now),
109-
row("tradeTime" -> 6L, "quantity" -> 100, "price" -> 94.12, "side" -> 'S', "ric" -> "VOD.L", "orderId" -> "NYC-0002", "onMkt" -> false, "trader" -> "steve", "ccyCross" -> "GBPUSD", CreatedTimeColumnName -> nextHour)
110-
)
100+
def setupTableWithCreationTime(indices: List[String])(using clock: TestFriendlyClock): InMemDataTable = {
101+
val now: Long = clock.now();
102+
val table: InMemDataTable = setupTable(indices,
103+
row("tradeTime" -> 5L, "quantity" -> 500, "price" -> 283.10, "side" -> 'B', "ric" -> "AAPL.L", "orderId" -> "NYC-0004", "onMkt" -> false, "trader" -> "chris", "ccyCross" -> "GBPUSD"),
104+
row("tradeTime" -> 3L, "quantity" -> 100, "price" -> 94.12, "side" -> 'S', "ric" -> "VOD.L", "orderId" -> "LDN-0003", "onMkt" -> true, "trader" -> "steve", "ccyCross" -> "GBPUSD"),
105+
)(using clock)
106+
clock.advanceBy(1000L)
107+
table.processUpdate("LDN-0001", row("tradeTime" -> 2L, "quantity" -> 100, "price" -> 94.12, "side" -> 'S', "ric" -> "VOD.L", "orderId" -> "LDN-0001", "onMkt" -> true, "trader" -> "chris", "ccyCross" -> "GBPUSD"))
108+
table.processUpdate("LDN-0008", row("tradeTime" -> 5L, "quantity" -> 100, "price" -> 180.50, "side" -> 'B', "ric" -> "BT.L", "orderId" -> "LDN-0008", "onMkt" -> true, "trader" -> "steve", "ccyCross" -> "GBPUSD"))
109+
table
111110
}
112111

113-
def setupTable(indices: List[String], rows: RowWithData*): InMemDataTable = {
112+
def setupTable(indices: List[String], rows: RowWithData*)(using clock: Clock): InMemDataTable = {
114113
val columns = Columns.fromNames(
115114
"orderId:String",
116115
"trader:String",
@@ -129,7 +128,7 @@ object FilterAndSortFixture {
129128
indices = Indices(indices.map(f => Index(f)) *),
130129
joinFields = "ric", "orderId", "ccyCross"
131130
)
132-
val table: InMemDataTable = new InMemDataTable(tableDef, new TestFriendlyJoinTableProvider)(new MetricsProviderImpl, new DefaultClock)
131+
val table: InMemDataTable = new InMemDataTable(tableDef, new TestFriendlyJoinTableProvider)(new MetricsProviderImpl, clock)
133132
rows.foreach(row => table.processUpdate(row.key, row))
134133
table
135134
}

‎vuu/src/test/scala/org/finos/vuu/core/sort/GenericSortTest.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ import org.scalatest.featurespec.AnyFeatureSpec
88
import org.scalatest.matchers.should.Matchers
99

1010
class GenericSortTest extends AnyFeatureSpec with Matchers {
11-
implicit val clock: Clock = new TestFriendlyClock(1000)
11+
implicit val clock: Clock = new TestFriendlyClock(1000L)
1212
private val table = setupTable()
1313
private val table2 = setupTable2()
1414

0 commit comments

Comments
 (0)