rpuch commented on code in PR #814:
URL: https://github.com/apache/ignite-3/pull/814#discussion_r881681131
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/datapage/DataPageReader.java:
##########
@@ -54,129 +49,49 @@ public DataPageReader(PageMemory pageMemory, int groupId,
IoStatisticsHolder sta
}
/**
- * Returns a row by link. To get the row bytes, more than one pages may be
traversed (if the corresponding row
- * was fragmented when stored).
+ * Traverses page memory starting at the given link. At each step, reads
the current data row and feeds it to the given
+ * {@link PageMemoryTraversal} object which updates itself (usually) and
returns next link to continue traversal
+ * (or {@link PageMemoryTraversal#STOP_TRAVERSAL} to stop.
*
* @param link Row link
- * @return row object assembled from the row bytes
+ * @param traversal object consuming payloads and controlling the traversal
+ * @param argument argument that is passed to the traversal
* @throws IgniteInternalCheckedException If failed
* @see org.apache.ignite.internal.pagememory.util.PageIdUtils#link(long,
int)
+ * @see PageMemoryTraversal
*/
- @Nullable
- public T getRowByLink(final long link) throws
IgniteInternalCheckedException {
+ public <T> void traverse(final long link, PageMemoryTraversal<T>
traversal, @Nullable T argument)
+ throws IgniteInternalCheckedException {
assert link != 0;
int pageSize = pageMemory.realPageSize(groupId);
- ByteArrayOutputStream baos = null;
-
- long nextLink = link;
+ long currentLink = link;
do {
- final long pageId = pageId(nextLink);
-
+ final long pageId = pageId(currentLink);
final long page = pageMemory.acquirePage(groupId, pageId,
statisticsHolder);
try {
long pageAddr = pageMemory.readLock(groupId, pageId, page);
-
- assert pageAddr != 0L : nextLink;
+ assert pageAddr != 0L : currentLink;
try {
AbstractDataPageIo<?> dataIo =
pageMemory.ioRegistry().resolve(pageAddr);
- int itemId = itemId(nextLink);
-
- if (handleNonExistentItemsGracefully() &&
!dataIo.itemExists(pageAddr, itemId, pageSize)) {
- assert nextLink == link : "It is not first page of a
fragmented row, but the item does not exist, pageId="
- + pageId + ", itemId=" + itemId;
-
- return rowForNonExistingItem(pageId, itemId);
- }
+ int itemId = itemId(currentLink);
DataPagePayload data = dataIo.readPayload(pageAddr,
itemId, pageSize);
- if (data.nextLink() == 0 && nextLink == link) {
- // Good luck: we can read the row without fragments.
- return readRowFromAddress(link, pageAddr +
data.offset());
- }
-
- ByteBuffer dataBuf = wrapPointer(pageAddr, pageSize);
-
- dataBuf.position(data.offset());
- dataBuf.limit(data.offset() + data.payloadSize());
-
- if (baos == null) {
- baos = new ByteArrayOutputStream();
- }
-
- byte[] bytes = new byte[data.payloadSize()];
- dataBuf.get(bytes);
- try {
- baos.write(bytes);
- } catch (IOException e) {
- throw new IllegalStateException("Should not happen",
e);
- }
-
- nextLink = data.nextLink();
+ currentLink = traversal.consumePagePayload(currentLink,
pageAddr, data, argument);
Review Comment:
Any actionable suggestions? :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
