Skip to content

Commit

Permalink
Merge pull request #2198 from ergoplatform/indexer-api-ordering
Browse files Browse the repository at this point in the history 
Rework segment retrieval, added ordering test
  • Loading branch information
@kushti
kushti authored Feb 6, 2025
2 parents ad564c9 + d697672 commit 16bbdb8
Show file tree
Hide file tree
Showing 2 changed files with 117 additions and 122 deletions.
86 changes: 27 additions & 59 deletions src/main/scala/org/ergoplatform/nodeView/history/extra/Segment.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -132,19 +132,6 @@ abstract class Segment[T <: Segment[_] : ClassTag](val parentId: ModifierId,
data.toArray
}

/**
* Calculate the segment offsets for the given range.
*
* @param offset - items to skip from the start
* @param limit - items to retrieve
* @return array of offsets
*/
private[extra] def getSegmentsForRange(offset: Int, limit: Int)(implicit segmentTreshold: Int): Array[Int] = {
val floor = math.max(math.floor(offset * 1F / segmentTreshold).toInt, 0)
val ceil = math.ceil((offset + limit) * 1F / segmentTreshold).toInt
(floor to ceil).toArray
}

/**
* Get an array of transactions with full bodies from an array of numeric transaction indexes
*
Expand Down Expand Up @@ -187,55 +174,36 @@ abstract class Segment[T <: Segment[_] : ClassTag](val parentId: ModifierId,
arr: ArrayBuffer[Long],
idOf: (ModifierId, Int) => ModifierId,
arraySelector: T => ArrayBuffer[Long],
retrieve: (ArrayBuffer[Long], ErgoHistoryReader) => Array[B],
txsFlag: Boolean)
retrieve: (ArrayBuffer[Long], ErgoHistoryReader) => Array[B])
(implicit segmentTreshold: Int): Array[B] = {
val array = if(txsFlag) {
arr.reverse
} else {
arr
}

val total: Int = segmentTreshold * segmentCount + array.length
val total: Int = segmentTreshold * segmentCount + arr.length
if (offset >= total)
return Array.empty[B] // return empty array if all elements are skipped
if (offset + limit > array.length && segmentCount > 0) {

val target = offset + limit

val collected: ArrayBuffer[Long] = ArrayBuffer.empty[Long]
collected ++= (if (offset < array.length) array.slice(offset, Math.min(offset + limit, array.length)) else Nil)
val segments = getSegmentsForRange(offset - array.length, limit).map(n => math.min(segmentCount - 1, n)).distinct
segments.foreach { num =>
val lowerBound = array.length + num * segmentTreshold
val upperBound = lowerBound + segmentTreshold

if (collected.length < limit && target > lowerBound) {
val arr = if(txsFlag) {
arraySelector(
history.typedExtraIndexById[T](idMod(idOf(parentId, (segmentCount - 1) - num))).get
).reverse
} else {
arraySelector(
history.typedExtraIndexById[T](idMod(idOf(parentId, num))).get
).reverse
}
if (target > upperBound) {
collected ++= arr.slice(offset - lowerBound, arr.size)
} else {
if (offset > lowerBound) {
collected ++= arr.slice(offset - lowerBound, offset - lowerBound + limit)
} else {
collected ++= arr.slice(0, target - lowerBound)
}
}
}
}

retrieve(collected, history)
} else {
retrieve(array.slice(offset, offset + limit), history)
val collected: ArrayBuffer[Long] = ArrayBuffer.empty[Long]
var off = offset
var lim = limit
var segment = segmentCount - 1
if(off < arr.size) { // all elements are in memory
val x = arr.dropRight(off).takeRight(lim)
collected ++= x.reverse
off = 0
lim -= x.size
}else { // skip all elements in memory
off -= arr.size
}
while(off > segmentTreshold && segment >= 0) { // skip segments until offset gets smaller than one segment
off -= segmentTreshold
segment -= 1
}
while(lim > 0 && segment >= 0) { // take limit elements from remaining segments (also skip remaining offset)
val x = arraySelector(history.typedExtraIndexById[T](idMod(idOf(parentId, segment))).get).dropRight(off).takeRight(lim)
collected ++= x.reverse
lim -= x.size
off = 0
segment -= 1
}
retrieve(collected, history)
}

/**
Expand All @@ -247,7 +215,7 @@ abstract class Segment[T <: Segment[_] : ClassTag](val parentId: ModifierId,
* @return array of transactions with full bodies
*/
def retrieveTxs(history: ErgoHistoryReader, offset: Int, limit: Int)(implicit segmentTreshold: Int): Array[IndexedErgoTransaction] =
getFromSegments(history, offset, limit, txSegmentCount, txs, txSegmentId, _.txs, getTxs, txsFlag = true)
getFromSegments(history, offset, limit, txSegmentCount, txs, txSegmentId, _.txs, getTxs)

/**
* Get a range of the boxes associated with the parent object
Expand All @@ -258,7 +226,7 @@ abstract class Segment[T <: Segment[_] : ClassTag](val parentId: ModifierId,
* @return array of boxes
*/
def retrieveBoxes(history: ErgoHistoryReader, offset: Int, limit: Int)(implicit segmentTreshold: Int): Array[IndexedErgoBox] =
getFromSegments(history, offset, limit, boxSegmentCount, boxes, boxSegmentId, _.boxes, getBoxes, txsFlag = false)
getFromSegments(history, offset, limit, boxSegmentCount, boxes, boxSegmentId, _.boxes, getBoxes)


/**
Expand Down
153 changes: 90 additions & 63 deletions src/test/scala/org/ergoplatform/nodeView/history/extra/SegmentSpec.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,81 +11,65 @@ import org.ergoplatform.modifiers.{BlockSection, NonHeaderBlockSection}
import org.ergoplatform.nodeView.history.ErgoHistoryReader
import org.ergoplatform.nodeView.history.storage.HistoryStorage
import org.ergoplatform.settings.ErgoSettings
import scorex.util

import scala.util.Try

class SegmentSpec extends ErgoCorePropertyTest {

property("correct slicing in getFromSegments") {
implicit val segmentTreshold: Int = 512
implicit val segmentTreshold: Int = 512

val hash = ModifierId @@ Base16.encode(Array.fill(32)(0.toByte))
val boxes = new ArrayBuffer[Long]()
(1 to 1706).foreach{i =>
boxes.append(i)
}
val hash: util.ModifierId.Type = ModifierId @@ Base16.encode(Array.fill(32)(0.toByte))
val boxes = new ArrayBuffer[Long]()
(1 to 1706).foreach{i =>
boxes.append(i)
}

val segment0 = IndexedErgoAddress(hash, new ArrayBuffer[Long](), boxes.take(segmentTreshold))
val segment1 = IndexedErgoAddress(hash, new ArrayBuffer[Long](), boxes.slice(segmentTreshold, segmentTreshold * 2))
val segment2 = IndexedErgoAddress(hash, new ArrayBuffer[Long](), boxes.slice(segmentTreshold * 2, segmentTreshold * 3))
val ia = IndexedErgoAddress(hash, new ArrayBuffer[Long](), boxes.slice(segmentTreshold * 3, boxes.size).reverse)

val hr = new ErgoHistoryReader {
override protected[history] val historyStorage: HistoryStorage = new HistoryStorage(null, null, null ,null) {
override def getExtraIndex(id: ModifierId): Option[ExtraIndex] = {
val b = Base16.decode(id).get.head
if(b == 0) {
Some(segment0)
} else if(b == 1) {
Some(segment1)
} else if(b == 2) {
Some(segment2)
} else {
None
}
val segment0: IndexedErgoAddress = IndexedErgoAddress(hash, new ArrayBuffer[Long](), boxes.take(segmentTreshold))
val segment1: IndexedErgoAddress = IndexedErgoAddress(hash, new ArrayBuffer[Long](), boxes.slice(segmentTreshold, segmentTreshold * 2))
val segment2: IndexedErgoAddress = IndexedErgoAddress(hash, new ArrayBuffer[Long](), boxes.slice(segmentTreshold * 2, segmentTreshold * 3))
val ia: IndexedErgoAddress = IndexedErgoAddress(hash, new ArrayBuffer[Long](), boxes.slice(segmentTreshold * 3, boxes.size))

val hr: ErgoHistoryReader = new ErgoHistoryReader {
override protected[history] val historyStorage: HistoryStorage = new HistoryStorage(null, null, null ,null) {
override def getExtraIndex(id: ModifierId): Option[ExtraIndex] = {
val b = Base16.decode(id).get.head
if(b == 0) {
Some(segment0)
} else if(b == 1) {
Some(segment1)
} else if(b == 2) {
Some(segment2)
} else {
None
}
}

override protected val settings: ErgoSettings = null

/**
* Whether state requires to download adProofs before full block application
*/
override protected def requireProofs: Boolean = ???

/**
* @param m - modifier to process
* @return ProgressInfo - info required for State to be consistent with History
*/
override protected def process(m: NonHeaderBlockSection): Try[ProgressInfo[BlockSection]] = ???

/**
* @param m - modifier to validate
* @return Success() if modifier is valid from History point of view, Failure(error) otherwise
*/
override protected def validate(m: NonHeaderBlockSection): Try[Unit] = ???

override val powScheme: AutolykosPowScheme = null
}
override protected val settings: ErgoSettings = null
override protected def requireProofs: Boolean = ???
override protected def process(m: NonHeaderBlockSection): Try[ProgressInfo[BlockSection]] = ???
override protected def validate(m: NonHeaderBlockSection): Try[Unit] = ???
override val powScheme: AutolykosPowScheme = null
}

def getFromSegments(offset: Int, limit: Int) = {
ia.getFromSegments(
history = hr,
offset,
limit,
segmentCount = 3,
ia.boxes,
idOf = {
case (_, i) => ModifierId @@ Base16.encode(Array.fill(32)(i.toByte))
},
arraySelector = { ai => ai.boxes },
retrieve = {
case (arr, _) => arr.toArray
},
txsFlag = false
)
}
def getFromSegments(offset: Int, limit: Int): Array[Long] = {
ia.getFromSegments(
history = hr,
offset,
limit,
segmentCount = 3,
ia.boxes,
idOf = {
case (_, i) => ModifierId @@ Base16.encode(Array.fill(32)(i.toByte))
},
arraySelector = { ai => ai.boxes },
retrieve = {
case (arr, _) => arr.toArray
}
)
}

property("correct slicing in getFromSegments") {
val lim = 200

val a1 = getFromSegments(0, lim)
Expand Down Expand Up @@ -184,4 +168,47 @@ class SegmentSpec extends ErgoCorePropertyTest {
(c1 ++ c2 ++ c3).distinct.length shouldBe 1706
}

property("correct ordering in getFromSegments") {
val lim = 200

val a1 = getFromSegments(0, lim)
val a2 = getFromSegments(lim, lim)
val a3 = getFromSegments(2 * lim, lim)
val a4 = getFromSegments(3 * lim, lim)
val a5 = getFromSegments(4 * lim, lim)
val a6 = getFromSegments(5 * lim, lim)
val a7 = getFromSegments(6 * lim, lim)
val a8 = getFromSegments(7 * lim, lim)
val a9 = getFromSegments(8 * lim, lim)
val a10 = getFromSegments(9 * lim, lim)

val full = a1 ++ a2 ++ a3 ++ a4 ++ a5 ++ a6 ++ a7 ++ a8 ++ a9 ++ a10
full.reverse.sameElements(full.sorted) shouldBe true

val lim2 = 100

val b1 = getFromSegments(0, lim2)
val b2 = getFromSegments(lim2, lim2)
val b3 = getFromSegments(2 * lim2, lim2)
val b4 = getFromSegments(3 * lim2, lim2)
val b5 = getFromSegments(4 * lim2, lim2)
val b6 = getFromSegments(5 * lim2, lim2)
val b7 = getFromSegments(6 * lim2, lim2)
val b8 = getFromSegments(7 * lim2, lim2)
val b9 = getFromSegments(8 * lim2, lim2)
val b10 = getFromSegments(9 * lim2, lim2)
val b11 = getFromSegments(10 * lim2, lim2)
val b12 = getFromSegments(11 * lim2, lim2)
val b13 = getFromSegments(12 * lim2, lim2)
val b14 = getFromSegments(13 * lim2, lim2)
val b15 = getFromSegments(14 * lim2, lim2)
val b16 = getFromSegments(15 * lim2, lim2)
val b17 = getFromSegments(16 * lim2, lim2)
val b18 = getFromSegments(17 * lim2, lim2)
val b19 = getFromSegments(18 * lim2, lim2)

val full2 = b1 ++ b2 ++ b3 ++ b4 ++ b5 ++ b6 ++ b7 ++ b8 ++ b9 ++ b10 ++ b11 ++ b12 ++ b13 ++ b14 ++ b15 ++ b16 ++ b17 ++ b18 ++ b19
full2.reverse.sameElements(full2.sorted) shouldBe true
}

}

0 comments on commit 16bbdb8

Please sign in to comment.