Skip to content

Commit 5d7a804

Browse files
stonesmistonesmi
authored
Fixed duplicate references in Indices after updates (#2049)
1 parent f30e9f7 commit 5d7a804

15 files changed

Lines changed: 635 additions & 146 deletions

File tree

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

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ package org.finos.vuu.plugin.virtualized.table
22

33
import org.finos.toolbox.collection.window.MovingWindow
44
import org.finos.toolbox.time.Clock
5-
import org.finos.vuu.core.table._
5+
import org.finos.vuu.core.table.*
66

77
class VirtualizedSessionTableData(cacheSize: Int)(implicit clock: Clock) extends TableData {
88

@@ -18,14 +18,18 @@ class VirtualizedSessionTableData(cacheSize: Int)(implicit clock: Clock) extends
1818
}
1919
}
2020

21-
override def update(key: String, update: RowData): (TableData, RowData) = {
22-
rowCache.put(key, update)
23-
(this, update)
21+
override def update(key: String, update: RowData): TableDataUpdate = {
22+
rowCache.put(key, update) match {
23+
case Some(value) => TableDataUpdated(this, value, update)
24+
case None => TableDataInserted(this, update)
25+
}
2426
}
2527

26-
override def delete(key: String): TableData = {
27-
rowCache.remove(key)
28-
this
28+
override def delete(key: String): TableDataDelete = {
29+
rowCache.remove(key) match {
30+
case Some(value) => TableDataDeleted(this, value)
31+
case None => TableDataNothingDeleted
32+
}
2933
}
3034

3135
override def deleteAll(): TableData = {

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,9 @@ import org.finos.vuu.core.table.RowData
55
import org.finos.vuu.plugin.virtualized.table.cache.CaffeineWindowedRowDataCache
66

77
trait WindowedCache[KEY, VALUE] {
8-
def put(key: KEY, v: VALUE): Unit
8+
def put(key: KEY, v: VALUE): Option[VALUE]
99
def get(key: KEY): Option[VALUE]
10-
def remove(key: KEY): Unit
10+
def remove(key: KEY): Option[VALUE]
1111
def removeAll(): Unit
1212
}
1313

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

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,14 @@ class CaffeineWindowedRowDataCache(val cacheSize: Int)(implicit clock: Clock) ex
2525
})
2626
.build()
2727

28-
override def put(key: String, v: RowData): Unit = {
29-
cache.put(key, v)
28+
override def put(key: String, v: RowData): Option[RowData] = {
29+
if (key != null) {
30+
val currentValue = Option(cache.getIfPresent(key))
31+
cache.put(key, v)
32+
currentValue
33+
} else {
34+
None
35+
}
3036
}
3137

3238
override def get(key: String): Option[RowData] = {
@@ -38,5 +44,13 @@ class CaffeineWindowedRowDataCache(val cacheSize: Int)(implicit clock: Clock) ex
3844
}
3945
override def removeAll(): Unit = cache.invalidateAll()
4046

41-
override def remove(key: String): Unit = cache.invalidate(key)
47+
override def remove(key: String): Option[RowData] = {
48+
if (key != null) {
49+
val currentValue = Option(cache.getIfPresent(key))
50+
cache.invalidate(key)
51+
currentValue
52+
} else {
53+
None
54+
}
55+
}
4256
}

‎toolbox/src/main/scala/org/finos/toolbox/collection/set/ImmutableArraySet.scala‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,14 @@ object ImmutableArraySet {
3838
}
3939
}
4040

41+
def from[T <: Object](set: Set[T])(implicit c: ClassTag[T]): ImmutableArraySet[T] = {
42+
if (set.isEmpty) {
43+
empty()
44+
} else {
45+
VectorImmutableArraySet.from(set)
46+
}
47+
}
48+
4149
def empty[T <: Object](implicit c: ClassTag[T]): ImmutableArraySet[T] = {
4250
VectorImmutableArraySet.empty()
4351
}

‎toolbox/src/main/scala/org/finos/toolbox/collection/set/VectorImmutableArraySet.scala‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,12 @@ object VectorImmutableArraySet {
3030
VectorImmutableArraySet(builder.result(), seen.toSet)
3131
}
3232

33+
def from[T <: Object : ClassTag](set: Set[T]): ImmutableArraySet[T] = {
34+
VectorImmutableArraySet(set.toVector, set)
35+
}
36+
3337
def empty[T <: Object : ClassTag](): ImmutableArraySet[T] = {
34-
new VectorImmutableArraySet[T](Vector.empty, Set.empty)
38+
VectorImmutableArraySet[T](Vector.empty, Set.empty)
3539
}
3640

3741
}

‎toolbox/src/test/scala/org/finos/toolbox/collection/set/ImmutableArraySetTest.scala‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,14 @@ class ImmutableArraySetTest extends AnyFunSuite with Matchers {
77

88
def set(xs: String*): ImmutableArraySet[String] = ImmutableArraySet.from(xs.toArray)
99

10+
test("creating from a set is the same as from an iterable") {
11+
val original = set("a", "b", "c")
12+
13+
val result = ImmutableArraySet.from(Set("a", "b", "c"))
14+
15+
result shouldEqual original
16+
}
17+
1018
test("adding a new element returns a new set and preserves the original") {
1119
val original = set("a", "b", "c")
1220

‎vuu/src/main/scala/org/finos/vuu/core/index/InMemColumnIndices.scala‎

Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,137 @@
1+
package org.finos.vuu.core.index
2+
3+
import org.finos.vuu.api.TableDef
4+
import org.finos.vuu.core.table.datatype.EpochTimestamp
5+
import org.finos.vuu.core.table.{Column, DataType, RowData}
6+
7+
trait InMemColumnIndices {
8+
9+
def indexForColumn(column: Column): Option[IndexedField[_]]
10+
11+
def insert(rowData: RowData): Unit
12+
13+
def update(originalRow: RowData, updatedRow: RowData): Unit
14+
15+
def remove(rowData: RowData): Unit
16+
17+
}
18+
19+
object InMemColumnIndices {
20+
21+
def apply(tableDef: TableDef): InMemColumnIndices = {
22+
val indices = tableDef.indices.indices
23+
.map(index => tableDef.columnForName(index.column))
24+
.map(c => c -> buildIndexForColumn(c))
25+
.toMap[Column, IndexedField[_]]
26+
val updaters = buildUpdaters(indices)
27+
InMemColumnIndicesImpl(indices, updaters)
28+
}
29+
30+
private def buildIndexForColumn(c: Column): IndexedField[_] = {
31+
c.dataType match {
32+
case DataType.StringDataType =>
33+
new HashMapIndexedStringField(c)
34+
case DataType.IntegerDataType =>
35+
new SkipListIndexedIntField(c)
36+
case DataType.LongDataType =>
37+
new SkipListIndexedLongField(c)
38+
case DataType.DoubleDataType =>
39+
new SkipListIndexedDoubleField(c)
40+
case DataType.BooleanDataType =>
41+
new SkipListIndexedBooleanField(c)
42+
case DataType.EpochTimestampType =>
43+
new SkipListIndexedEpochTimestampField(c)
44+
case DataType.CharDataType =>
45+
new SkipListIndexedCharField(c)
46+
case _ =>
47+
throw new UnsupportedOperationException(s"Unsupported type ${c.dataType} in column ${c.name}")
48+
}
49+
}
50+
51+
private def buildUpdaters(indices: Map[Column, IndexedField[?]]): Array[IndexUpdater] = {
52+
indices.map {
53+
case (column, index) =>
54+
55+
//helper method (handy!)
56+
def create[T](index: IndexedField[T]): IndexUpdater = IndexUpdater(
57+
column,
58+
insertFunction = (v, k) => index.insert(v.asInstanceOf[T], k),
59+
replaceFunction = (v1, v2, k) => index.replace(v1.asInstanceOf[T], v2.asInstanceOf[T], k),
60+
removeFunction = (v, k) => index.remove(v.asInstanceOf[T], k)
61+
)
62+
63+
column.dataType match {
64+
case DataType.StringDataType => create(index.asInstanceOf[IndexedField[String]])
65+
case DataType.IntegerDataType => create(index.asInstanceOf[IndexedField[Int]])
66+
case DataType.LongDataType => create(index.asInstanceOf[IndexedField[Long]])
67+
case DataType.DoubleDataType => create(index.asInstanceOf[IndexedField[Double]])
68+
case DataType.BooleanDataType => create(index.asInstanceOf[IndexedField[Boolean]])
69+
case DataType.EpochTimestampType => create(index.asInstanceOf[IndexedField[EpochTimestamp]])
70+
case DataType.CharDataType => create(index.asInstanceOf[IndexedField[Char]])
71+
case _ => throw new UnsupportedOperationException(s"Unsupported type ${column.dataType} in column ${column.name}")
72+
}
73+
}.toArray
74+
}
75+
76+
}
77+
78+
private case class IndexUpdater(column: Column,
79+
insertFunction: (Any, String) => Unit,
80+
replaceFunction: (Any, Any, String) => Unit,
81+
removeFunction: (Any, String) => Unit)
82+
83+
private case class InMemColumnIndicesImpl(indices: Map[Column, IndexedField[?]],
84+
updaters: Array[IndexUpdater]) extends InMemColumnIndices {
85+
86+
override def indexForColumn(column: Column): Option[IndexedField[_]] = indices.get(column)
87+
88+
override def insert(rowData: RowData): Unit = {
89+
val rowKey = rowData.key
90+
91+
var i = 0
92+
while (i < updaters.length) {
93+
val updater = updaters(i)
94+
val value = rowData.get(updater.column)
95+
if (value != null) {
96+
updater.insertFunction.apply(value, rowKey)
97+
}
98+
i += 1
99+
}
100+
}
101+
102+
override def update(originalRow: RowData, updatedRow: RowData): Unit = {
103+
val rowKey = updatedRow.key
104+
105+
var i = 0
106+
while (i < updaters.length) {
107+
val updater = updaters(i)
108+
val originalValue = originalRow.get(updater.column)
109+
val currentValue = updatedRow.get(updater.column)
110+
if (originalValue != currentValue) {
111+
if (originalValue == null && currentValue != null) {
112+
updater.insertFunction.apply(currentValue, rowKey)
113+
} else if (originalValue != null && currentValue == null) {
114+
updater.removeFunction.apply(originalValue, rowKey)
115+
} else {
116+
updater.replaceFunction.apply(originalValue, currentValue, rowKey)
117+
}
118+
}
119+
i += 1
120+
}
121+
}
122+
123+
override def remove(rowData: RowData): Unit = {
124+
val rowKey = rowData.key
125+
126+
var i = 0
127+
while (i < updaters.length) {
128+
val updater = updaters(i)
129+
val value = rowData.get(updater.column)
130+
if (value != null) {
131+
updater.removeFunction.apply(value, rowKey)
132+
}
133+
i += 1
134+
}
135+
}
136+
137+
}

‎vuu/src/main/scala/org/finos/vuu/core/index/IndexedField.scala‎

Lines changed: 43 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
package org.finos.vuu.core.index
22

33
import com.typesafe.scalalogging.StrictLogging
4-
import org.finos.toolbox.collection.array.ImmutableArray
4+
import org.finos.toolbox.collection.array.{ImmutableArray, VectorImmutableArray}
55
import org.finos.toolbox.collection.set.ImmutableArraySet
66
import org.finos.vuu.core.table.Column
77
import org.finos.vuu.core.table.datatype.EpochTimestamp
@@ -13,9 +13,11 @@ trait IndexedField[TYPE] {
1313

1414
protected val empty: ImmutableArray[String] = ImmutableArray.empty
1515

16-
def insert(indexedValue: TYPE, rowKeys: String): Unit
16+
def insert(indexedValue: TYPE, rowKey: String): Unit
1717

18-
def remove(indexedValue: TYPE, rowKeys: String): Unit
18+
def replace(oldIndexedValue: TYPE, newIndexedValue: TYPE, rowKey: String): Unit
19+
20+
def remove(indexedValue: TYPE, rowKey: String): Unit
1921

2022
def column: Column
2123

@@ -56,7 +58,7 @@ class HashMapIndexedStringField(val column: Column) extends StringIndexedField w
5658
private final val indexMap = new ConcurrentHashMap[String, ImmutableArraySet[String]]()
5759

5860
override def remove(indexKey: String, rowKey: String): Unit = {
59-
logger.trace(s"Remove Index: ${column.name}")
61+
logger.trace(s"Removing value $rowKey from ${column.name} index")
6062
indexMap.computeIfPresent(indexKey, (_, value) => {
6163
if (value.length > 1) {
6264
value.-(rowKey)
@@ -67,7 +69,7 @@ class HashMapIndexedStringField(val column: Column) extends StringIndexedField w
6769
}
6870

6971
override def insert(indexKey: String, rowKey: String): Unit = {
70-
logger.trace(s"Update Index: ${column.name}")
72+
logger.trace(s"Inserting value $rowKey into ${column.name} index")
7173
indexMap.compute(indexKey, (_, value) => {
7274
value match {
7375
case null => ImmutableArraySet.of(rowKey)
@@ -76,6 +78,23 @@ class HashMapIndexedStringField(val column: Column) extends StringIndexedField w
7678
})
7779
}
7880

81+
override def replace(oldIndexKey: String, newIndexKey: String, rowKey: String): Unit = {
82+
logger.trace(s"Moving value $rowKey in ${column.name} index")
83+
indexMap.computeIfPresent(oldIndexKey, (_, value) => {
84+
if (value.length > 1) {
85+
value.-(rowKey)
86+
} else {
87+
null
88+
}
89+
})
90+
indexMap.compute(newIndexKey, (_, value) => {
91+
value match {
92+
case null => ImmutableArraySet.of(rowKey)
93+
case array: ImmutableArraySet[String] => array.+(rowKey)
94+
}
95+
})
96+
}
97+
7998
override def find(indexKey: String): ImmutableArray[String] = {
8099
logger.debug(s"Hit Index: ${column.name} for key $indexKey")
81100
val result = indexMap.get(indexKey)
@@ -100,7 +119,7 @@ class SkipListIndexedField[TYPE](val column: Column) extends IndexedField[TYPE]
100119
private final val skipList = new ConcurrentSkipListMap[TYPE, ImmutableArraySet[String]]()
101120

102121
override def remove(indexKey: TYPE, rowKey: String): Unit = {
103-
logger.trace(s"Remove Index: ${column.name}")
122+
logger.trace(s"Removing value $rowKey from ${column.name} index")
104123
skipList.computeIfPresent(indexKey, (_, value) => {
105124
if (value.length > 1) {
106125
value.-(rowKey)
@@ -110,8 +129,25 @@ class SkipListIndexedField[TYPE](val column: Column) extends IndexedField[TYPE]
110129
})
111130
}
112131

132+
override def replace(oldIndexKey: TYPE, newIndexKey: TYPE, rowKey: String): Unit = {
133+
logger.trace(s"Moving value $rowKey in ${column.name} index")
134+
skipList.computeIfPresent(oldIndexKey, (_, value) => {
135+
if (value.length > 1) {
136+
value.-(rowKey)
137+
} else {
138+
null
139+
}
140+
})
141+
skipList.compute(newIndexKey, (_, value) => {
142+
value match {
143+
case null => ImmutableArraySet.of(rowKey)
144+
case array: ImmutableArraySet[String] => array.+(rowKey)
145+
}
146+
})
147+
}
148+
113149
override def insert(indexKey: TYPE, rowKey: String): Unit = {
114-
logger.trace(s"Update Index: ${column.name}")
150+
logger.trace(s"Inserting value $rowKey into ${column.name} index")
115151
skipList.compute(indexKey, (_, value) => {
116152
value match {
117153
case null => ImmutableArraySet.of(rowKey)

‎vuu/src/main/scala/org/finos/vuu/core/index/QueryOptimizer.scala‎

Lines changed: 0 additions & 7 deletions
This file was deleted.

0 commit comments

Comments
 (0)