Skip to content

PHOENIX-6339 Older client using aggregate queries shows incorrect res…#1111

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kadirozde merged 1 commit into from
Jan 27, 2021
Merged

PHOENIX-6339 Older client using aggregate queries shows incorrect res… #1111

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -253,7 +253,7 @@ public RegionScanner preScannerOpen(final ObserverContext<RegionCoprocessorEnvir
// last possible moment. You need to swap the start/stop and make the
// start exclusive and the stop inclusive.
ScanUtil.setupReverseScan(scan);
if (!(scan.getFilter() instanceof PagedFilter)) {
if (scan.getFilter() != null && !(scan.getFilter() instanceof PagedFilter)) {
byte[] pageSizeMsBytes = scan.getAttribute(BaseScannerRegionObserver.SERVER_PAGE_SIZE_MS);
if (pageSizeMsBytes != null) {
scan.setFilter(new PagedFilter(scan.getFilter(), getPageSizeMsForFilter(scan)));
Expand Down
11 changes: 9 additions & 2 deletions ...ix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionScanner.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -29,6 +29,7 @@
import static org.apache.phoenix.query.QueryConstants.AGG_TIMESTAMP;
import static org.apache.phoenix.query.QueryConstants.SINGLE_COLUMN;
import static org.apache.phoenix.query.QueryConstants.SINGLE_COLUMN_FAMILY;
import static org.apache.phoenix.query.QueryConstants.UNGROUPED_AGG_ROW_KEY;
import static org.apache.phoenix.query.QueryServices.MUTATE_BATCH_SIZE_ATTRIB;
import static org.apache.phoenix.query.QueryServices.MUTATE_BATCH_SIZE_BYTES_ATTRIB;
import static org.apache.phoenix.query.QueryServices.SOURCE_OPERATION_ATTRIB;
Expand DownExpand Up@@ -625,8 +626,14 @@ public boolean next(List<Cell> resultsToReturn) throws IOException {
Cell cell;
if (hasAny) {
byte[] value = aggregators.toBytes(rowAggregators);
cell = CellUtil.createCell(CellUtil.cloneRow(lastCell), SINGLE_COLUMN_FAMILY, SINGLE_COLUMN,
AGG_TIMESTAMP, KeyValue.Type.Put.getCode(), value);
if (pageSizeMs == Long.MAX_VALUE) {
// Paging is not set. To be compatible with older clients, do not set the row key
cell = CellUtil.createCell(UNGROUPED_AGG_ROW_KEY, SINGLE_COLUMN_FAMILY, SINGLE_COLUMN,
AGG_TIMESTAMP, KeyValue.Type.Put.getCode(), value);
} else {
cell = CellUtil.createCell(CellUtil.cloneRow(lastCell), SINGLE_COLUMN_FAMILY, SINGLE_COLUMN,
AGG_TIMESTAMP, KeyValue.Type.Put.getCode(), value);
}
resultsToReturn.add(cell);
}
return hasMore;
Expand Down
21 changes: 13 additions & 8 deletions phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -55,6 +55,7 @@
import org.apache.phoenix.monitoring.ScanMetricsHolder;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.util.ByteUtil;
Expand DownExpand Up@@ -139,15 +140,19 @@ public TableResultIterator(MutationState mutationState, Scan scan, ScanMetricsHo
.getInt(QueryConstants.HASH_JOIN_CACHE_RETRIES, QueryConstants.DEFAULT_HASH_JOIN_CACHE_RETRIES);
ScanUtil.setScanAttributesForIndexReadRepair(scan, table, plan.getContext().getConnection());
ScanUtil.setScanAttributesForPhoenixTTL(scan, table, plan.getContext().getConnection());
long pageSizeMs = plan.getContext().getConnection().getQueryServices().getProps()
.getInt(QueryServices.PHOENIX_SERVER_PAGE_SIZE_MS, -1);
if (pageSizeMs == -1) {
// Use the half of the HBase RPC timeout value as the the server page size to make sure that the HBase
// region server will be able to send a heartbeat message to the client before the client times out
pageSizeMs = (long) (plan.getContext().getConnection().getQueryServices().getProps()
.getLong(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT) * 0.5);
if (plan.getContext().getConnection().getQueryServices().getProps().getBoolean(
QueryServices.PHOENIX_SERVER_PAGING_ENABLED_ATTRIB,
QueryServicesOptions.DEFAULT_PHOENIX_SERVER_PAGING_ENABLED)) {
long pageSizeMs = plan.getContext().getConnection().getQueryServices().getProps()
.getInt(QueryServices.PHOENIX_SERVER_PAGE_SIZE_MS, -1);
if (pageSizeMs == -1) {
// Use the half of the HBase RPC timeout value as the the server page size to make sure that the HBase
// region server will be able to send a heartbeat message to the client before the client times out
pageSizeMs = (long) (plan.getContext().getConnection().getQueryServices().getProps()
.getLong(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT) * 0.5);
}
scan.setAttribute(BaseScannerRegionObserver.SERVER_PAGE_SIZE_MS, Bytes.toBytes(Long.valueOf(pageSizeMs)));
}
scan.setAttribute(BaseScannerRegionObserver.SERVER_PAGE_SIZE_MS, Bytes.toBytes(Long.valueOf(pageSizeMs)));
}

@Override
Expand Down
2 changes: 2 additions & 0 deletions phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -326,6 +326,8 @@ public interface QueryServices extends SQLCloseable {
public static final String GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB = "phoenix.global.index.row.age.threshold.to.delete.ms";
// Enable the IndexRegionObserver Coprocessor
public static final String INDEX_REGION_OBSERVER_ENABLED_ATTRIB = "phoenix.index.region.observer.enabled";
// Enable Phoenix server paging
public static final String PHOENIX_SERVER_PAGING_ENABLED_ATTRIB = "phoenix.server.paging.enabled";
// Enable support for long view index(default is false)
public static final String LONG_VIEW_INDEX_ENABLED_ATTRIB = "phoenix.index.longViewIndex.enabled";
// The number of index rows to be rebuild in one RPC call
Expand Down
1 change: 1 addition & 0 deletions phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -341,6 +341,7 @@ public class QueryServicesOptions {

public static final long DEFAULT_GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS = 7*24*60*60*1000; /* 7 days */
public static final boolean DEFAULT_INDEX_REGION_OBSERVER_ENABLED = true;
public static final boolean DEFAULT_PHOENIX_SERVER_PAGING_ENABLED = true;
public static final long DEFAULT_INDEX_REBUILD_PAGE_SIZE_IN_ROWS = 32*1024;
public static final boolean DEFAULT_ALLOW_SPLITTABLE_SYSTEM_CATALOG_ROLLBACK = false;

Expand Down
9 changes: 4 additions & 5 deletions phoenix-core/src/main/java/org/apache/phoenix/util/ScanUtil.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -1304,21 +1304,20 @@ public static PagedFilter getPhoenixPagedFilter(Scan scan) {
* each HBase RegionScanner#next() time which is controlled by PagedFilter is set to 0.3 * SERVER_PAGE_SIZE_MS.
*
*/
private static long getPageSizeMs(Scan scan) {
private static long getPageSizeMs(Scan scan, double factor) {
long pageSizeMs = Long.MAX_VALUE;
byte[] pageSizeMsBytes = scan.getAttribute(BaseScannerRegionObserver.SERVER_PAGE_SIZE_MS);
if (pageSizeMsBytes != null) {
pageSizeMs = Bytes.toLong(pageSizeMsBytes);
pageSizeMs = (long) (pageSizeMs * factor);
}
return pageSizeMs;
}

public static long getPageSizeMsForRegionScanner(Scan scan) {
return (long) (getPageSizeMs(scan) * 0.6);
}
public static long getPageSizeMsForRegionScanner(Scan scan) { return getPageSizeMs(scan, 0.6); }

@jpisaacjpisaacJan 26, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is/should be the thought process in choosing the factor? May be good to add some comments around it?

@kadirozdekadirozdeJan 26, 2021

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The comments are just right there before getPageSizeMs()


public static long getPageSizeMsForFilter(Scan scan) {
return (long) (getPageSizeMs(scan) * 0.3);
return getPageSizeMs(scan, 0.3);
}

/**
Expand Down