tkhurana commented on code in PR #2637:
URL: https://github.com/apache/phoenix/pull/2637#discussion_r4275165111
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java:
##########
@@ -89,8 +94,24 @@ public UncoveredGlobalIndexRegionScanner(final RegionScanner
innerScanner, final
final TupleProjector tupleProjector, final IndexMaintainer indexMaintainer,
final byte[][] viewConstants, final ImmutableBytesWritable ptr, final long
pageSizeMs,
final long queryLimit, boolean isDistinct) throws IOException {
- super(innerScanner, region, scan, env, dataTableScan, tupleProjector,
indexMaintainer,
- viewConstants, ptr, pageSizeMs, queryLimit, isDistinct);
+ // Candidates can be rejected when joined with the data table. Enforce the
pushed limit
+ // after verification, before a lower scanner can count those candidates
toward the limit.
+ this(innerScanner, region, scan, env, dataTableScan, tupleProjector,
indexMaintainer,
+ viewConstants, ptr, pageSizeMs, queryLimit, isDistinct,
+ indexMaintainer.isUncovered() && !indexMaintainer.isCDCIndex()
+ ? ScanUtil.removePageFilter(scan)
+ : null);
+ }
+
+ private UncoveredGlobalIndexRegionScanner(final RegionScanner innerScanner,
final Region region,
+ final Scan scan, final RegionCoprocessorEnvironment env, final Scan
dataTableScan,
+ final TupleProjector tupleProjector, final IndexMaintainer indexMaintainer,
+ final byte[][] viewConstants, final ImmutableBytesWritable ptr, final long
pageSizeMs,
+ final long queryLimit, boolean isDistinct, PageFilter pageFilter) throws
IOException {
+ super(reopenScannerWithoutPageFilter(innerScanner, scan, pageFilter),
region, scan, env,
+ dataTableScan, tupleProjector, indexMaintainer, viewConstants, ptr,
pageSizeMs, queryLimit,
Review Comment:
On the PageFilter path the client never sets `INDEX_LIMIT`
(`BaseResultIterators` L316-325), so `queryLimit` is -1 here and the cap in
`UncoveredIndexRegionScanner` L137-139 is skipped. `pageSizeInRows` stays at
`DEFAULT_INDEX_PAGE_SIZE_IN_ROWS` (32K).
Before this change, the PageFilter stopped the inner index scan after LIMIT
rows. Now nothing does, so `LIMIT 1` can scan up to 32K index rows and do as
many data-table lookups per region before returning, even with no stale rows at
all.
`queryLimit` is only used for that cap and the two are mutually exclusive,
so passing the page size should be enough:
```java
super(reopenScannerWithoutPageFilter(innerScanner, scan, pageFilter),
region, scan, env,
dataTableScan, tupleProjector, indexMaintainer, viewConstants, ptr,
pageSizeMs,
pageFilter == null ? queryLimit : pageFilter.getPageSize(), isDistinct);
```
Pages are then capped at LIMIT, and more pages follow when stale rows are
dropped (same as the INDEX_LIMIT path).
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java:
##########
@@ -89,8 +94,24 @@ public UncoveredGlobalIndexRegionScanner(final RegionScanner
innerScanner, final
final TupleProjector tupleProjector, final IndexMaintainer indexMaintainer,
final byte[][] viewConstants, final ImmutableBytesWritable ptr, final long
pageSizeMs,
final long queryLimit, boolean isDistinct) throws IOException {
- super(innerScanner, region, scan, env, dataTableScan, tupleProjector,
indexMaintainer,
- viewConstants, ptr, pageSizeMs, queryLimit, isDistinct);
+ // Candidates can be rejected when joined with the data table. Enforce the
pushed limit
+ // after verification, before a lower scanner can count those candidates
toward the limit.
+ this(innerScanner, region, scan, env, dataTableScan, tupleProjector,
indexMaintainer,
+ viewConstants, ptr, pageSizeMs, queryLimit, isDistinct,
+ indexMaintainer.isUncovered() && !indexMaintainer.isCDCIndex()
+ ? ScanUtil.removePageFilter(scan)
+ : null);
+ }
+
+ private UncoveredGlobalIndexRegionScanner(final RegionScanner innerScanner,
final Region region,
+ final Scan scan, final RegionCoprocessorEnvironment env, final Scan
dataTableScan,
+ final TupleProjector tupleProjector, final IndexMaintainer indexMaintainer,
+ final byte[][] viewConstants, final ImmutableBytesWritable ptr, final long
pageSizeMs,
+ final long queryLimit, boolean isDistinct, PageFilter pageFilter) throws
IOException {
+ super(reopenScannerWithoutPageFilter(innerScanner, scan, pageFilter),
region, scan, env,
Review Comment:
If the rest of the constructor throws after this reopen, the new scanner
leaks. `getTable`, `getRegionLocator(...).getEndKeys()` (a meta RPC) and
`addEmptyColumnToScan` can all throw.
`RegionScannerHolder.overrideDelegate` only assigns `delegate` on success,
so the holder still points at the original scanner, which
`reopenScannerWithoutPageFilter` already closed. RSRpcServices closes that one
again (harmless), and the reopened `RegionScannerImpl` is never closed. It
keeps its read point and store-file references until the region closes. Before
this change the RPC-layer close cleaned up the only scanner.
Either close `innerScanner` in a try/finally when the constructor fails, or
do the strip+reopen lazily on the first `next()` like `TTLRegionScanner.init()`.
Nit: the cast at L143 could wrap `ClassCastException` in
`DoNotRetryIOException` like `DelegateRegionScanner.getNewRegionScanner` does.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java:
##########
@@ -89,8 +94,24 @@ public UncoveredGlobalIndexRegionScanner(final RegionScanner
innerScanner, final
final TupleProjector tupleProjector, final IndexMaintainer indexMaintainer,
final byte[][] viewConstants, final ImmutableBytesWritable ptr, final long
pageSizeMs,
final long queryLimit, boolean isDistinct) throws IOException {
- super(innerScanner, region, scan, env, dataTableScan, tupleProjector,
indexMaintainer,
- viewConstants, ptr, pageSizeMs, queryLimit, isDistinct);
+ // Candidates can be rejected when joined with the data table. Enforce the
pushed limit
+ // after verification, before a lower scanner can count those candidates
toward the limit.
Review Comment:
Could this name the PageFilter and the lower scanners explicitly? The fix
also depends on `TTLRegionScanner.init()` being lazy: if that ever became
eager, TTL would take the PageFilter first, `removePageFilter` here would
return null, and the bug would silently come back. Worth a sentence here so
nobody changes it unknowingly. Something like:
> Index rows can be rejected when joined with the data table. Take the
PageFilter limit here so only verified rows count toward it; otherwise HBase's
PageFilter (or TTLRegionScanner, which takes over the PageFilter lazily on its
first next()) would count rows that are later rejected.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java:
##########
@@ -89,8 +94,24 @@ public UncoveredGlobalIndexRegionScanner(final RegionScanner
innerScanner, final
final TupleProjector tupleProjector, final IndexMaintainer indexMaintainer,
final byte[][] viewConstants, final ImmutableBytesWritable ptr, final long
pageSizeMs,
final long queryLimit, boolean isDistinct) throws IOException {
- super(innerScanner, region, scan, env, dataTableScan, tupleProjector,
indexMaintainer,
- viewConstants, ptr, pageSizeMs, queryLimit, isDistinct);
+ // Candidates can be rejected when joined with the data table. Enforce the
pushed limit
+ // after verification, before a lower scanner can count those candidates
toward the limit.
+ this(innerScanner, region, scan, env, dataTableScan, tupleProjector,
indexMaintainer,
+ viewConstants, ptr, pageSizeMs, queryLimit, isDistinct,
+ indexMaintainer.isUncovered() && !indexMaintainer.isCDCIndex()
Review Comment:
Nit: `isUncovered()` is always true here, since this scanner is only built
for uncovered global indexes (`RegionScannerFactory`).
`!indexMaintainer.isCDCIndex()` alone would do.
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexLimitIT.java:
##########
@@ -0,0 +1,377 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.end2end.index;
+
+import static
org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants.SERVER_PAGE_SIZE_MS;
+import static org.apache.phoenix.query.explain.ExplainPlanTestUtil.assertPlan;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.SimpleRegionObserver;
+import org.apache.hadoop.hbase.regionserver.InternalScanner;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
+import org.apache.phoenix.filter.PagingFilter;
+import org.apache.phoenix.query.BaseTest;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.ManualEnvironmentEdge;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.ScanUtil;
+import org.junit.After;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+/** Regression tests for PHOENIX-8015: LIMIT counts verified uncovered-index
rows. */
+@Category(NeedsOwnMiniClusterTest.class)
+public class UncoveredGlobalIndexLimitIT extends BaseTest {
+ private static final long AGE_THRESHOLD_MS = 60_000;
+ // Keep stale entries live in HBase and in Phoenix while testing the
independent repair threshold.
+ private static final int TTL_SECONDS = 3600;
+ private static final int ROW_COUNT = 20;
+ private static final int LIMIT = 5;
+ private static final long UPDATED_TIME_BASE = 1_700_000_000_000L;
+ private final ManualEnvironmentEdge clock = new ManualEnvironmentEdge();
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> props = new HashMap<>();
+ props.put(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB,
+ Long.toString(AGE_THRESHOLD_MS));
+ props.put(QueryServices.PHOENIX_COMPACTION_ENABLED,
Boolean.toString(true));
+ props.put(QueryServices.USE_STATS_FOR_PARALLELIZATION,
Boolean.toString(false));
+ // Force multiple index batches before and after the first valid row.
+ props.put(QueryServices.INDEX_PAGE_SIZE_IN_ROWS, Long.toString(3));
Review Comment:
With the page size set to 3, the uncapped-page-size problem above doesn't
show up. Could you add a case at the default page size that asserts how many
index rows the server scanned (e.g. via scan metrics or the observer) for a
LIMIT query?
##########
phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexLimitIT.java:
##########
@@ -0,0 +1,377 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.end2end.index;
+
+import static
org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants.SERVER_PAGE_SIZE_MS;
+import static org.apache.phoenix.query.explain.ExplainPlanTestUtil.assertPlan;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.sql.Types;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.SimpleRegionObserver;
+import org.apache.hadoop.hbase.regionserver.InternalScanner;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
+import org.apache.phoenix.filter.PagingFilter;
+import org.apache.phoenix.query.BaseTest;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.ManualEnvironmentEdge;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.ScanUtil;
+import org.junit.After;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+/** Regression tests for PHOENIX-8015: LIMIT counts verified uncovered-index
rows. */
+@Category(NeedsOwnMiniClusterTest.class)
+public class UncoveredGlobalIndexLimitIT extends BaseTest {
+ private static final long AGE_THRESHOLD_MS = 60_000;
+ // Keep stale entries live in HBase and in Phoenix while testing the
independent repair threshold.
+ private static final int TTL_SECONDS = 3600;
+ private static final int ROW_COUNT = 20;
+ private static final int LIMIT = 5;
+ private static final long UPDATED_TIME_BASE = 1_700_000_000_000L;
+ private final ManualEnvironmentEdge clock = new ManualEnvironmentEdge();
+
+ @BeforeClass
+ public static synchronized void doSetup() throws Exception {
+ Map<String, String> props = new HashMap<>();
+ props.put(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB,
+ Long.toString(AGE_THRESHOLD_MS));
+ props.put(QueryServices.PHOENIX_COMPACTION_ENABLED,
Boolean.toString(true));
+ props.put(QueryServices.USE_STATS_FOR_PARALLELIZATION,
Boolean.toString(false));
+ // Force multiple index batches before and after the first valid row.
+ props.put(QueryServices.INDEX_PAGE_SIZE_IN_ROWS, Long.toString(3));
+ // BaseTest defaults to a dummy result after every row; exercise a
complete server batch here.
+ props.put(QueryServices.PHOENIX_SERVER_PAGE_SIZE_MS,
Long.toString(60_000));
+ props.put("hbase.coprocessor.region.classes",
LimitTestObserver.class.getName());
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+ // BaseTest overwrites the age threshold with zero during mini-cluster
configuration.
+ // Restore it before creating user tables so their coprocessors inherit
the intended value.
+ getUtility().getConfiguration()
+
.setLong(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB,
AGE_THRESHOLD_MS);
+ getUtility().getHBaseCluster().getRegionServerThreads().forEach(thread -> {
+ thread.getRegionServer().getConfiguration().setLong(
+ QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB,
AGE_THRESHOLD_MS);
+ assertEquals(AGE_THRESHOLD_MS,
thread.getRegionServer().getConfiguration()
+
.getLong(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB, -1));
+ });
+ }
+
+ @After
+ public void resetClock() throws Exception {
+ LimitTestObserver.indexTable = null;
+ LimitTestObserver.dataTable = null;
+ LimitTestObserver.limitedTable = null;
+ EnvironmentEdgeManager.reset();
+ assertFalse("refCount leaked", isAnyStoreRefCountLeaked());
+ }
+
+ @Test
+ public void testLimitWithLeadingStaleRowsAndTTLMasking() throws Exception {
+ assertLimitWithLeadingStaleRows(true);
+ }
+
+ @Test
+ public void testLimitWithLeadingStaleRowsWithoutTTLMasking() throws
Exception {
+ assertLimitWithLeadingStaleRows(false);
+ }
+
+ private void assertLimitWithLeadingStaleRows(boolean finiteTTL) throws
Exception {
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ Fixture fixture = createFixture(conn, finiteTTL, 1000);
+ String query = fixture.indexQuery("") + " LIMIT " + LIMIT;
+ assertPlan(conn, query).table(fixture.indexTable).scanType("FULL SCAN")
+ .serverRowLimit((long) LIMIT);
+ assertRows(conn, query, LIMIT);
+ assertStaleEntriesPresent(fixture);
+ }
+ }
+
+ @Test
+ public void testLimitControlsWithLeadingStaleRows() throws Exception {
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ // Use the same FOREVER-TTL configuration as the reproduced PageFilter
failure.
+ Fixture fixture = createFixture(conn, false, 1000);
+ String baseQuery = "SELECT /*+ NO_INDEX */ ID, PAYLOAD, UPDATED_TIME
FROM "
+ + fixture.dataTable + " ORDER BY UPDATED_TIME ASC LIMIT " + LIMIT;
+ assertPlan(conn, baseQuery).table(fixture.dataTable);
+ assertRows(conn, baseQuery, LIMIT);
+
+ String boundedQuery =
+ fixture.indexQuery(" WHERE UPDATED_TIME > TO_TIMESTAMP('2000-01-01')")
+ " LIMIT " + LIMIT;
+ assertPlan(conn, boundedQuery).table(fixture.indexTable).scanType("RANGE
SCAN");
+ assertRows(conn, boundedQuery, LIMIT);
+
+ // PAYLOAD comes from the base row, so this exercises INDEX_FILTER /
INDEX_LIMIT.
+ String filteredQuery = fixture.indexQuery(" WHERE PAYLOAD IS NOT NULL")
+ " LIMIT " + LIMIT;
+ assertPlan(conn, filteredQuery).table(fixture.indexTable);
+ assertRows(conn, filteredQuery, LIMIT);
+
+ assertRows(conn, fixture.indexQuery(""), ROW_COUNT);
+ // All controls leave the young stale entries in place for the same
reason as the LIMIT IT.
+ assertStaleEntriesPresent(fixture);
+ }
+ }
+
+ @Test
+ public void testLimitWithPartiallyStaleLeadingRows() throws Exception {
+ for (boolean finiteTTL : new boolean[] { false, true }) {
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ Fixture fixture = createFixture(conn, finiteTTL, 1000, 3);
+ assertRows(conn, fixture.indexQuery("") + " LIMIT " + LIMIT, LIMIT);
+ assertStaleEntriesPresent(fixture);
+ }
+ EnvironmentEdgeManager.reset();
+ }
+ }
+
+ @Test
+ public void testLimitLargerThanValidRowCount() throws Exception {
+ for (boolean finiteTTL : new boolean[] { false, true }) {
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ Fixture fixture = createFixture(conn, finiteTTL, 1000);
+ assertRows(conn, fixture.indexQuery("") + " LIMIT " + (ROW_COUNT * 3),
ROW_COUNT);
+ assertStaleEntriesPresent(fixture);
+ }
+ EnvironmentEdgeManager.reset();
+ }
+ }
+
+ @Test
+ public void testLimitWithOffset() throws Exception {
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ Fixture fixture = createFixture(conn, false, 1000);
+ assertRows(conn, fixture.indexQuery("") + " LIMIT " + LIMIT + " OFFSET
3", LIMIT, 3);
+ assertStaleEntriesPresent(fixture);
+ }
+ }
+
+ @Test
+ public void testLimitAcrossDummyPages() throws Exception {
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ Fixture fixture = createFixture(conn, false, 1000);
+ LimitTestObserver.scanCount.set(0);
+ LimitTestObserver.indexTable = fixture.indexTable;
+ LimitTestObserver.dataTable = fixture.dataTable;
+ assertRows(conn, fixture.indexQuery("") + " LIMIT " + LIMIT, LIMIT);
+ assertTrue("The query must exercise zero-budget paging scans",
+ LimitTestObserver.scanCount.get() > 0);
Review Comment:
`scanCount` goes up in `preScannerOpen` for every scan on either table, so
this is always true and doesn't show that a dummy page was produced. Counting
dummy results in `postScannerNext` and asserting > 0 would make the test check
what its name says.
Also, the clock is frozen with `ManualEnvironmentEdge` while the query runs,
and `PhoenixScannerContext.isTimedOut` uses a strict `>`. So the index-scan
timeout dummies can't fire; only the PagingFilter path can.
--
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]
