diff --git a/plugin/off-heap-table-plugin/pom.xml b/plugin/off-heap-table-plugin/pom.xml new file mode 100644 index 0000000000..e54e6d374b --- /dev/null +++ b/plugin/off-heap-table-plugin/pom.xml @@ -0,0 +1,157 @@ + + + 4.0.0 + + + org.finos.vuu + plugin + 0.17.0-SNAPSHOT + + + org.finos.vuu.plugin + off-heap-table-plugin + ${project.artifactId} + + + + org.finos.vuu + vuu + 0.17.0-SNAPSHOT + + + + org.scala-lang + scala3-library_3 + ${scala.version} + + + + org.finos.vuu + vuu + 0.17.0-SNAPSHOT + tests + test-jar + test + + + + org.scalatest + scalatest_3 + ${scalatest.version} + test + + + org.scala-lang + scala3-library_3 + + + + + + + + src/main/java + src/test/java + + + + + org.apache.maven.plugins + maven-source-plugin + + + compile + + jar + + + + + + + org.apache.maven.plugins + maven-release-plugin + + + + org.apache.maven.plugins + maven-javadoc-plugin + + + attach-javadocs + + jar + + + + + + + net.alchim31.maven + scala-maven-plugin + + + + compile + testCompile + + + + + src/main/scala + src/test/scala + + -deprecation + + + -Xms64m + -Xmx1024m + + + + + + org.apache.maven.plugins + maven-jar-plugin + + + + test-jar + + + + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + + + + + org.scalatest + scalatest-maven-plugin + + ${project.build.directory}/surefire-reports + . + test-reports.txt + + + + test + + test + + + + + + + + + \ No newline at end of file diff --git a/plugin/off-heap-table-plugin/src/main/java/org/finos/vuu/plugin/offheap/MemoryMappedBuffer.java b/plugin/off-heap-table-plugin/src/main/java/org/finos/vuu/plugin/offheap/MemoryMappedBuffer.java new file mode 100644 index 0000000000..bb1ef0ba3a --- /dev/null +++ b/plugin/off-heap-table-plugin/src/main/java/org/finos/vuu/plugin/offheap/MemoryMappedBuffer.java @@ -0,0 +1,51 @@ +package org.finos.vuu.plugin.offheap; + +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.MappedByteBuffer; +import java.nio.channels.FileChannel; +import java.util.Arrays; + +public class MemoryMappedBuffer implements AutoCloseable { + + private static final long SEGMENT_SIZE = 1L << 30; //1GB + private final MappedByteBuffer[] segments; + private final long totalCapacity; + + public MemoryMappedBuffer(String filePath, long capacity) throws IOException { + this.totalCapacity = capacity; + int numSegments = (int) ((capacity + SEGMENT_SIZE - 1) / SEGMENT_SIZE); + this.segments = new MappedByteBuffer[numSegments]; + + try (var raf = new RandomAccessFile(filePath, "rw"); + var channel = raf.getChannel()) { + + for (int i = 0; i < numSegments; i++) { + long position = i * SEGMENT_SIZE; + long size = Math.min(SEGMENT_SIZE, capacity - position); + segments[i] = channel.map(FileChannel.MapMode.READ_WRITE, position, size); + } + } + } + + public void put(long index, byte b) { + int segmentIndex = (int) (index / SEGMENT_SIZE); + int segmentOffset = (int) (index % SEGMENT_SIZE); + segments[segmentIndex].put(segmentOffset, b); + } + + public byte get(long index) { + int segmentIndex = (int) (index / SEGMENT_SIZE); + int segmentOffset = (int) (index % SEGMENT_SIZE); + return segments[segmentIndex].get(segmentOffset); + } + + public long capacity() { + return totalCapacity; + } + + @Override + public void close() { + Arrays.fill(segments, null); + } +} diff --git a/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/OffHeapTablePlugin.scala b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/OffHeapTablePlugin.scala new file mode 100644 index 0000000000..087e6fd9fa --- /dev/null +++ b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/OffHeapTablePlugin.scala @@ -0,0 +1,42 @@ +package org.finos.vuu.plugin.offheap + +import org.finos.toolbox.jmx.MetricsProvider +import org.finos.toolbox.time.Clock +import org.finos.vuu.api.TableDef +import org.finos.vuu.core.table.TableContainer +import org.finos.vuu.feature.{FilterFactory, JoinTableFactory, SessionTableFactory, SortFactory, TableFactory, TableFeature, ViewPortCallableFactory, ViewPortFactory, ViewPortKeysCreator, ViewPortTableCreator, ViewPortTreeCallableFactory} +import org.finos.vuu.plugin.offheap.table.OffHeapDataTable +import org.finos.vuu.plugin.{DefaultPlugin, PluginType} +import org.finos.vuu.provider.JoinTableProvider + +object OffHeapTablePlugin extends DefaultPlugin { + + registerFeature(TableFeature) + + override def tableFactory(implicit metrics: MetricsProvider): TableFactory = (tableDef: TableDef, tableContainer: TableContainer, joinTableProvider: JoinTableProvider) => { + given clock: Clock = tableContainer.timeProvider + val table = new OffHeapDataTable(tableDef, joinTableProvider) + tableContainer.addTable(table) + table + } + + override def pluginType: PluginType = OffHeapTablePluginType + + override def joinTableFactory(implicit metrics: MetricsProvider, timeProvider: Clock): JoinTableFactory = ??? + + override def sessionTableFactory: SessionTableFactory = ??? + + override def viewPortKeysCreator: ViewPortKeysCreator = ??? + + override def viewPortFactory: ViewPortFactory = ??? + + override def filterFactory: FilterFactory = ??? + + override def sortFactory: SortFactory = ??? + + override def viewPortCallableFactory: ViewPortCallableFactory = ??? + + override def viewPortTreeCallableFactory: ViewPortTreeCallableFactory = ??? + + override def viewPortTableCreator: ViewPortTableCreator = ??? +} \ No newline at end of file diff --git a/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/OffHeapTablePluginType.scala b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/OffHeapTablePluginType.scala new file mode 100644 index 0000000000..1809e49bb4 --- /dev/null +++ b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/OffHeapTablePluginType.scala @@ -0,0 +1,7 @@ +package org.finos.vuu.plugin.offheap + +import org.finos.vuu.plugin.PluginType + +object OffHeapTablePluginType extends PluginType { + +} diff --git a/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/store/BufferToMapAdaptor.scala b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/store/BufferToMapAdaptor.scala new file mode 100644 index 0000000000..d62069f74c --- /dev/null +++ b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/store/BufferToMapAdaptor.scala @@ -0,0 +1,8 @@ +package org.finos.vuu.plugin.offheap.store + +class BufferToMapAdaptor { + + + + +} diff --git a/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/store/FastStore.scala b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/store/FastStore.scala new file mode 100644 index 0000000000..b024e027a8 --- /dev/null +++ b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/store/FastStore.scala @@ -0,0 +1,123 @@ +package org.finos.vuu.plugin.offheap.store + +import java.nio.{ByteBuffer, ByteOrder} +import java.nio.charset.StandardCharsets +import scala.collection.immutable.SortedMap + +object FastStore { + // Type Tags (1 byte) + private val T_SHORT: Byte = 0 + private val T_INT: Byte = 1 + private val T_LONG: Byte = 2 + private val T_CHAR: Byte = 3 + private val T_STRING: Byte = 4 + + // Header entry is 6 bytes: [Key: 1] + [Offset: 4] + [Tag: 1] + private val ENTRY_SIZE = 6 + + /** + * Packs data into a read-only ByteBuffer. + * Entries are sorted by key to enable O(log N) lookup. + */ + def pack(data: SortedMap[Byte, Any], target: ByteBuffer): Unit = { + val entryCount = data.size + val headerSize = 2 + (entryCount * ENTRY_SIZE) + + // Skip header for now, we will write it last once we know offsets + target.position(headerSize) + + val index = data.map { case (key, value) => + val offset = target.position() + val tag = value match { + case v: Short => target.putShort(v); T_SHORT + case v: Int => target.putInt(v); T_INT + case v: Long => target.putLong(v); T_LONG + case v: Char => target.putChar(v); T_CHAR + case v: String => + val bytes = v.getBytes(StandardCharsets.UTF_8) + putVarInt(bytes.length, target) + target.put(bytes) + T_STRING + case _ => throw new IllegalArgumentException(s"Unsupported type: ${value.getClass}") + } + (key, offset, tag) + } + + // Write the actual Header at the beginning + val finalSize = target.position() + target.position(0) + target.putShort(entryCount.toShort) + + index.foreach { case (key, offset, tag) => + target.putShort(key) + target.putInt(offset) + target.put(tag) + } + + target.position(finalSize) + target.flip() // Prepare for reading + } + + /** + * Performs a Binary Search on the buffer header to find a key. + * Zero-allocation for primitive returns (except Strings). + */ + def lookup(buf: ByteBuffer, targetKey: Byte): Any = { + val count = buf.getShort(0) + var low = 0 + var high = count - 1 + + while (low <= high) { + val mid = (low + high) / 2 + val pos = 2 + (mid * ENTRY_SIZE) + val key = buf.get(pos) + + if (key == targetKey) { + val offset = buf.getInt(pos + 1) + val tag = buf.get(pos + 5) + return readValue(buf, offset, tag) + } else if (key < targetKey) { + low = mid + 1 + } else { + high = mid - 1 + } + } + null + } + + private def readValue(buf: ByteBuffer, offset: Int, tag: Byte): Any = { + buf.position(offset) + tag match { + case T_SHORT => buf.getShort() + case T_INT => buf.getInt() + case T_LONG => buf.getLong() + case T_CHAR => buf.getChar() + case T_STRING => + val len = getVarInt(buf) + val bytes = new Array[Byte](len) + buf.get(bytes) + new String(bytes, StandardCharsets.UTF_8) + } + } + + private def putVarInt(value: Int, buf: ByteBuffer): Unit = { + var v = value + while ((v & ~0x7F) != 0) { + buf.put(((v & 0x7F) | 0x80).toByte) + v >>>= 7 + } + buf.put(v.toByte) + } + + private def getVarInt(buf: ByteBuffer): Int = { + var b = buf.get() + var value = b & 0x7F + var shift = 7 + while ((b & 0x80) != 0) { + b = buf.get() + value |= (b & 0x7F) << shift + shift += 7 + } + value + } +} diff --git a/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/table/OffHeapDataTable.scala b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/table/OffHeapDataTable.scala new file mode 100644 index 0000000000..ce5919e1ea --- /dev/null +++ b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/table/OffHeapDataTable.scala @@ -0,0 +1,87 @@ +package org.finos.vuu.plugin.offheap.table + +import com.typesafe.scalalogging.StrictLogging +import org.finos.toolbox.jmx.MetricsProvider +import org.finos.toolbox.time.Clock +import org.finos.vuu.api.TableDef +import org.finos.vuu.core.index.IndexedField +import org.finos.vuu.core.row.{InMemMapRowBuilder, RowBuilder} +import org.finos.vuu.core.table.{Column, ColumnValueProvider, DataTable, EmptyRowData, KeyedObservableHelper, RowData, RowKeyUpdate, RowWithData, TableData, TablePrimaryKeys} +import org.finos.vuu.provider.JoinTableProvider +import org.finos.vuu.viewport.{RowProcessor, ViewPortColumns} + +import java.util.concurrent.atomic.AtomicLong + +class OffHeapDataTable(private val tableDef: TableDef, private val joinProvider: JoinTableProvider)(using metrics: MetricsProvider, timeProvider: Clock) extends DataTable with KeyedObservableHelper[RowKeyUpdate] with StrictLogging { + + private final val indices: Map[Column, IndexedField[?]] = Map.empty + private final val internalUpdateCounter: AtomicLong = AtomicLong(0) + private final val data = createDataTableData(); + + override protected def createDataTableData(): TableData = { + OffHeapTableData(tableDef) + } + + override def updateCounter: Long = internalUpdateCounter.get() + + override def newRow(key: String): RowBuilder = new InMemMapRowBuilder().setKey(key) + + override def rowBuilder: RowBuilder = new InMemMapRowBuilder() + + override def incrementUpdateCounter(): Unit = internalUpdateCounter.incrementAndGet() + + override def indexForColumn(column: Column): Option[IndexedField[_]] = indices.get(column) + + override def getColumnValueProvider: ColumnValueProvider = ??? + + override def getTableDef: TableDef = tableDef + + override def processUpdate(rowKey: String, rowUpdate: RowData): Unit = ??? + + override def processDelete(rowKey: String): Unit = ??? + + override def name: String = tableDef.name + + override def linkableName: String = name + + override def notifyListeners(rowKey: String, isDelete: Boolean): Unit = { + getObserversByKey(rowKey).foreach(obs => { + obs.onUpdate(RowKeyUpdate(rowKey, this, isDelete)) + }) + } + + override def readRow(key: String, columns: List[String], processor: RowProcessor): Unit = ??? + + override def primaryKeys: TablePrimaryKeys = data.primaryKeyValues + + override def pullRow(key: String): RowData = { + data.dataByKey(key) match { + case rowWithData: RowWithData => rowWithData + case _ => EmptyRowData + } + } + + override def pullRow(key: String, columns: ViewPortColumns): RowData = { + data.dataByKey(key) match { + case rowWithData: RowWithData => columns.pullRow(key, rowWithData) + case _ => EmptyRowData + } + } + + override def pullRowFiltered(key: String, columns: ViewPortColumns): RowData = { + data.dataByKey(key) match { + case rowWithData: RowWithData => columns.pullRowAlwaysFilter(key, rowWithData) + case _ => EmptyRowData + } + } + + override def pullRowAsArray(key: String, columns: ViewPortColumns): Array[Any] = { + data.dataByKey(key) match { + case rowWithData: RowWithData => rowWithData.toArray(columns.getColumns) + case _ => Array.empty + } + } + + override def toString: String = s"OffHeapDataTable($name, rows=${this.primaryKeys.length})" + +} diff --git a/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/table/OffHeapTableData.scala b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/table/OffHeapTableData.scala new file mode 100644 index 0000000000..ee59dfe2e1 --- /dev/null +++ b/plugin/off-heap-table-plugin/src/main/scala/org/finos/vuu/plugin/offheap/table/OffHeapTableData.scala @@ -0,0 +1,28 @@ +package org.finos.vuu.plugin.offheap.table + +import org.finos.vuu.api.TableDef +import org.finos.vuu.core.table.{RowData, TableData, TablePrimaryKeys} + +import java.util.concurrent.ConcurrentHashMap + +class OffHeapTableData(private val tableDef: TableDef) extends TableData { + + private final val keyToIndexMap = ConcurrentHashMap[String, Long]() + + override def dataByKey(key: String): RowData = { + keyToIndexMap.get(key) match { + case index: Int => ??? + case _ => null + } + } + + override def update(key: String, update: RowData): (TableData, RowData) = ??? + + override def delete(key: String): TableData = ??? + + override def deleteAll(): TableData = ??? + + override def primaryKeyValues: TablePrimaryKeys = + + override def setKeyAt(index: Int, key: String): Unit = ??? +} diff --git a/plugin/pom.xml b/plugin/pom.xml index 8c6959c8b1..dc6552f847 100644 --- a/plugin/pom.xml +++ b/plugin/pom.xml @@ -14,6 +14,7 @@ ignite-plugin virtualized-table-plugin + off-heap-table-plugin \ No newline at end of file diff --git a/vuu/src/main/scala/org/finos/vuu/core/table/DataTable.scala b/vuu/src/main/scala/org/finos/vuu/core/table/DataTable.scala new file mode 100644 index 0000000000..4cdf66848c --- /dev/null +++ b/vuu/src/main/scala/org/finos/vuu/core/table/DataTable.scala @@ -0,0 +1,96 @@ +package org.finos.vuu.core.table + +import org.finos.toolbox.text.AsciiUtil +import org.finos.vuu.api.TableDef +import org.finos.vuu.core.index.IndexedField +import org.finos.vuu.core.row.RowBuilder +import org.finos.vuu.provider.Provider +import org.finos.vuu.viewport.RowSource + +trait DataTable extends KeyedObservable[RowKeyUpdate] with RowSource { + + @volatile private var provider: Provider = null + + protected def createDataTableData(): TableData + + def updateCounter: Long + + def newRow(key: String): RowBuilder + + def rowBuilder: RowBuilder + + def incrementUpdateCounter(): Unit + + def indexForColumn(column: Column): Option[IndexedField[_]] + + def setProvider(aProvider: Provider): Unit = provider = aProvider + + def getProvider: Provider = provider + + def getColumnValueProvider: ColumnValueProvider + + def asTable: DataTable = this + + def columnForName(name: String): Column = getTableDef.columnForName(name) + + def columnsForNames(names: String*): List[Column] = names.map(getTableDef.columnForName(_)).toList + + def columnsForNames(names: List[String]): List[Column] = names.map(getTableDef.columnForName(_)) + + def getTableDef: TableDef + + def processUpdate(rowUpdate: RowData): Unit = { + processUpdate(rowUpdate.key, rowUpdate) + } + + def processUpdate(rowKey: String, rowUpdate: RowData): Unit + + def hasRowChanged(row: RowWithData): Boolean = { + val existingRow = this.pullRow(row.key) + !existingRow.equals(row) + } + + def processDelete(rowKey: String): Unit + + def isSelectedVal(key: String, selected: Map[String, Any]): Int = { + if (selected.contains(key)) 1 else 0 + } + + def size(): Long = { + primaryKeys.length + } + + def toAscii(count: Int): String = { + val columns = getTableDef.getColumns + val keys = primaryKeys + + val selectedKeys = keys.toArray.take(count) + + val rows = selectedKeys.map(key => pullRowAsArray(key, ViewPortColumnCreator.create(this, columns.map(_.name).toList))) + + val columnNames = columns.map(_.name) + + AsciiUtil.asAsciiTable(columnNames, rows) + } + + def toAscii(start: Int, end: Int): String = { + val columns = getTableDef.getColumns + val keys = primaryKeys + + val selectedKeys = keys.toArray.slice(start, end) //.sliceToArray(start, end)//drop(start).take(end - start) + + val rows = selectedKeys.map(key => pullRowAsArray(key, ViewPortColumnCreator.create(this, columns.map(_.name).toList))) + + val columnNames = columns.map(_.name) + + AsciiUtil.asAsciiTable(columnNames, rows) + } +} + +case class RowKeyUpdate(key: String, source: RowSource, isDelete: Boolean = false) { + override def toString: String = s"RowKeyUpdate($key, ${source.name})" +} + +case class JoinTableUpdate(joinTable: DataTable, rowUpdate: RowWithData) { + override def toString: String = "JoinTableUpdate(" + joinTable.toString + ",updates=" + rowUpdate.data.size + ")" +} diff --git a/vuu/src/main/scala/org/finos/vuu/core/table/InMemDataTable.scala b/vuu/src/main/scala/org/finos/vuu/core/table/InMemDataTable.scala index c1d731d641..34bec2e488 100644 --- a/vuu/src/main/scala/org/finos/vuu/core/table/InMemDataTable.scala +++ b/vuu/src/main/scala/org/finos/vuu/core/table/InMemDataTable.scala @@ -9,6 +9,7 @@ import org.finos.vuu.api.TableDef import org.finos.vuu.core.index.{HashMapIndexedStringField, IndexedField, SkipListIndexedBooleanField, SkipListIndexedCharField, SkipListIndexedDoubleField, SkipListIndexedEpochTimestampField, SkipListIndexedIntField, SkipListIndexedLongField} import org.finos.vuu.core.row.{InMemMapRowBuilder, RowBuilder} import org.finos.vuu.core.table.datatype.EpochTimestamp +import org.finos.vuu.core.table.{RowData, RowWithData, EmptyRowData} import org.finos.vuu.feature.inmem.InMemTablePrimaryKeys import org.finos.vuu.provider.{JoinTableProvider, Provider} import org.finos.vuu.viewport.{RowProcessor, RowSource, ViewPortColumns} @@ -17,161 +18,6 @@ import java.util import java.util.concurrent.ConcurrentHashMap import scala.jdk.CollectionConverters.MapHasAsScala -trait DataTable extends KeyedObservable[RowKeyUpdate] with RowSource { - - @volatile private var provider: Provider = null - - protected def createDataTableData(): TableData - - def updateCounter: Long - - def newRow(key: String): RowBuilder - - def rowBuilder: RowBuilder - - def incrementUpdateCounter(): Unit - - def indexForColumn(column: Column): Option[IndexedField[_]] - - def setProvider(aProvider: Provider): Unit = provider = aProvider - - def getProvider: Provider = provider - - def getColumnValueProvider: ColumnValueProvider - - def asTable: DataTable = this - - def columnForName(name: String): Column = getTableDef.columnForName(name) - - def columnsForNames(names: String*): List[Column] = names.map(getTableDef.columnForName(_)).toList - - def columnsForNames(names: List[String]): List[Column] = names.map(getTableDef.columnForName(_)) - - def getTableDef: TableDef - - def processUpdate(rowUpdate: RowData): Unit = { - processUpdate(rowUpdate.key, rowUpdate) - } - - def processUpdate(rowKey: String, rowUpdate: RowData): Unit - - def hasRowChanged(row: RowWithData): Boolean = { - val existingRow = this.pullRow(row.key) - !existingRow.equals(row) - } - - def processDelete(rowKey: String): Unit - - def isSelectedVal(key: String, selected: Map[String, Any]): Int = { - if (selected.contains(key)) 1 else 0 - } - - def size(): Long = { - primaryKeys.length - } - - def toAscii(count: Int): String = { - val columns = getTableDef.getColumns - val keys = primaryKeys - - val selectedKeys = keys.toArray.take(count) - - val rows = selectedKeys.map(key => pullRowAsArray(key, ViewPortColumnCreator.create(this, columns.map(_.name).toList))) - - val columnNames = columns.map(_.name) - - AsciiUtil.asAsciiTable(columnNames, rows) - } - - def toAscii(start: Int, end: Int): String = { - val columns = getTableDef.getColumns - val keys = primaryKeys - - val selectedKeys = keys.toArray.slice(start, end) //.sliceToArray(start, end)//drop(start).take(end - start) - - val rows = selectedKeys.map(key => pullRowAsArray(key, ViewPortColumnCreator.create(this, columns.map(_.name).toList))) - - val columnNames = columns.map(_.name) - - AsciiUtil.asAsciiTable(columnNames, rows) - } -} - -case class RowKeyUpdate(key: String, source: RowSource, isDelete: Boolean = false) { - override def toString: String = s"RowKeyUpdate($key, ${source.name})" -} - -trait RowData { - def key: String - - def get(field: String): Any - - def get(column: Column): Any - - def getFullyQualified(column: Column): Any - - def set(field: String, value: Any): RowData - - def toArray(columns: List[Column]): Array[Any] - - def size: Int -} - -case class JoinTableUpdate(joinTable: DataTable, rowUpdate: RowWithData) { - override def toString: String = "JoinTableUpdate(" + joinTable.toString + ",updates=" + rowUpdate.data.size + ")" -} - -case class RowWithData(key: String, data: Map[String, Any]) extends RowData { - - def this(key: String, data: java.util.Map[String, Any]) = this(key, data.asScala.toMap) - - override def size: Int = data.size - - override def getFullyQualified(column: Column): Any = column.getDataFullyQualified(this) - - override def toArray(columns: List[Column]): Array[Any] = { - columns.map(c => this.get(c)).toArray - } - - override def get(column: Column): Any = { - if (column != null) { - column.getData(this) - } else { - null - } - } - - def get(column: String): Any = { - if (data == null) { - null - } else { - data.get(column).orNull - } - } - - def set(field: String, value: Any): RowWithData = { - RowWithData(key, data ++ Map[String, Any](field -> value)) - } -} - -object EmptyRowData extends RowData { - - override def key: String = null - - override def size: Int = 0 - - override def toArray(columns: List[Column]): Array[Any] = Array() - - override def get(field: String): Any = null - - override def get(column: Column): Any = null - - override def getFullyQualified(column: Column): Any = null - - override def set(field: String, value: Any): RowData = EmptyRowData -} - - case class InMemDataTableData(data: ConcurrentHashMap[String, RowData], private val primaryKeyValuesInternal: TablePrimaryKeys)(implicit timeProvider: Clock) extends TableData { def primaryKeyValues: TablePrimaryKeys = this.primaryKeyValuesInternal @@ -228,7 +74,6 @@ case class InMemDataTableData(data: ConcurrentHashMap[String, RowData], private } - class InMemDataTable(val tableDef: TableDef, val joinProvider: JoinTableProvider)(implicit val metrics: MetricsProvider, timeProvider: Clock) extends DataTable with KeyedObservableHelper[RowKeyUpdate] with StrictLogging { private final val indices = tableDef.indices.indices diff --git a/vuu/src/main/scala/org/finos/vuu/core/table/RowData.scala b/vuu/src/main/scala/org/finos/vuu/core/table/RowData.scala new file mode 100644 index 0000000000..a2c9a3bf73 --- /dev/null +++ b/vuu/src/main/scala/org/finos/vuu/core/table/RowData.scala @@ -0,0 +1,70 @@ +package org.finos.vuu.core.table + +import scala.jdk.CollectionConverters.MapHasAsScala + +trait RowData { + + def key: String + + def get(field: String): Any + + def get(column: Column): Any + + def getFullyQualified(column: Column): Any + + def set(field: String, value: Any): RowData + + def toArray(columns: List[Column]): Array[Any] + + def size: Int +} + +case class RowWithData(key: String, data: Map[String, Any]) extends RowData { + + def this(key: String, data: java.util.Map[String, Any]) = this(key, data.asScala.toMap) + + override def size: Int = data.size + + override def getFullyQualified(column: Column): Any = column.getDataFullyQualified(this) + + override def toArray(columns: List[Column]): Array[Any] = { + columns.map(c => this.get(c)).toArray + } + + override def get(column: Column): Any = { + if (column != null) { + column.getData(this) + } else { + null + } + } + + def get(column: String): Any = { + if (data == null) { + null + } else { + data.get(column).orNull + } + } + + def set(field: String, value: Any): RowWithData = { + RowWithData(key, data ++ Map[String, Any](field -> value)) + } +} + +object EmptyRowData extends RowData { + + override def key: String = null + + override def size: Int = 0 + + override def toArray(columns: List[Column]): Array[Any] = Array() + + override def get(field: String): Any = null + + override def get(column: Column): Any = null + + override def getFullyQualified(column: Column): Any = null + + override def set(field: String, value: Any): RowData = EmptyRowData +}