jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

add API to get statistics from pageSource

This commit is contained in:
guojunfei 2022-04-20 20:41:40 +08:00
parent 9b42fbaf38
commit f437b03bbb
3 changed files with 42 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
presto-main/src/main/java/io/prestosql/operator/ScanFilterAndProjectOperator.java
View File

@ -493,7 +493,12 @@ public class ScanFilterAndProjectOperator
page = recordMaterializedBytes(page, sizeInBytes -> processedBytes += sizeInBytes);
// update operator stats
processedPositions += page.getPositionCount();
if (pageSource.getCompletedPositionCount().isPresent()) {
processedPositions = pageSource.getCompletedPositionCount().getAsLong();
}
else {
processedPositions += page.getPositionCount();
}
physicalBytes = pageSource.getCompletedBytes();
readTimeNanos = pageSource.getReadTimeNanos();

31
presto-main/src/main/java/io/prestosql/operator/TableScanOperator.java
View File

@ -267,6 +267,8 @@ public class TableScanOperator
private boolean finished;
// completedPositionCount is used only if connectorPageSource.getCompletedPositionCount is present.
private long completedPositionCount;
private long completedBytes;
private long readTimeNanos;
Optional<TableScanNode> tableScanNodeOptional;
@ -707,17 +709,34 @@ public class TableScanOperator
}
Page page = source.getNextPage();
// if pageSource.getCompletedPositionCount is present, get operator statistics from pageSource
if (source.getCompletedPositionCount().isPresent()) {
long endCompletedPositionCount = source.getCompletedPositionCount().getAsLong();
long endCompletedBytes = source.getCompletedBytes();
long endReadTimeNanos = source.getReadTimeNanos();
long currentPositionCount = endCompletedPositionCount - completedPositionCount;
long currentCompletedBytes = endCompletedBytes - completedBytes;
operatorContext.recordPhysicalInputWithTiming(currentCompletedBytes, currentPositionCount, endReadTimeNanos - readTimeNanos);
operatorContext.recordProcessedInput(currentCompletedBytes, currentPositionCount);
completedPositionCount = endCompletedPositionCount;
completedBytes = endCompletedBytes;
readTimeNanos = endReadTimeNanos;
}
if (page != null) {
// assure the page is in memory before handing to another operator
page = page.getLoadedPage();
// update operator stats
long endCompletedBytes = source.getCompletedBytes();
long endReadTimeNanos = source.getReadTimeNanos();
operatorContext.recordPhysicalInputWithTiming(endCompletedBytes - completedBytes, page.getPositionCount(), endReadTimeNanos - readTimeNanos);
operatorContext.recordProcessedInput(page.getSizeInBytes(), page.getPositionCount());
completedBytes = endCompletedBytes;
readTimeNanos = endReadTimeNanos;
if (!source.getCompletedPositionCount().isPresent()) {
long endCompletedBytes = source.getCompletedBytes();
long endReadTimeNanos = source.getReadTimeNanos();
operatorContext.recordPhysicalInputWithTiming(endCompletedBytes - completedBytes, page.getPositionCount(), endReadTimeNanos - readTimeNanos);
operatorContext.recordProcessedInput(page.getSizeInBytes(), page.getPositionCount());
completedBytes = endCompletedBytes;
readTimeNanos = endReadTimeNanos;
}
// pull bloomFilter from stateStore and filter page
if (existsCrossFilter) {

11
presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorPageSource.java
View File

@ -17,6 +17,7 @@ import io.prestosql.spi.Page;
import java.io.Closeable;
import java.io.IOException;
import java.util.OptionalLong;
import java.util.concurrent.CompletableFuture;
public interface ConnectorPageSource
@ -75,4 +76,14 @@ public interface ConnectorPageSource
{
return false;
}
/**
* Some components can push down some operators to other component for calculation.
* This API is used to return the number of position count before push down for statistics.
* If this API is not used, just return empty.
*/
default OptionalLong getCompletedPositionCount()
{
return OptionalLong.empty();
}
}