This is an automated email from the ASF dual-hosted git repository.
virajjasani pushed a commit to branch 5.3
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/5.3 by this push:
new 35ce1b0655 PHOENIX-8015 Enforce uncovered global index limits after
verification (#2637)
35ce1b0655 is described below
commit 35ce1b0655ee9773c8a2ab7877d894de00017fbe
Author: Jing Yu <[email protected]>
AuthorDate: Tue Oct 20 12:27:19 2026 -0700
PHOENIX-8015 Enforce uncovered global index limits after verification
(#2637)
---
.../UncoveredGlobalIndexRegionScanner.java | 119 +++++-
.../end2end/index/UncoveredGlobalIndexLimitIT.java | 421 +++++++++++++++++++++
.../index/UncoveredGlobalIndexRegionScannerIT.java | 117 ++++++
.../UncoveredGlobalIndexRegionScannerTest.java | 165 ++++++++
4 files changed, 804 insertions(+), 18 deletions(-)
diff --git
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java
index f457a9f633..8af5f19008 100644
---
a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java
+++
b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScanner.java
@@ -28,15 +28,19 @@ import java.util.TreeSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
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.client.metrics.ScanMetrics;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.regionserver.Region;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
+import org.apache.hadoop.hbase.regionserver.ScannerContext;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.compat.hbase.CompatScanMetrics;
@@ -76,6 +80,8 @@ public class UncoveredGlobalIndexRegionScanner extends
UncoveredIndexRegionScann
protected final HTableFactory hTableFactory;
private final boolean isScanMetricsEnabled;
private List<DataTableScanMetricsWithScanTime> dataTableScanMetrics;
+ private final long maxRowCount;
+ private long returnedRowCount;
// This relies on Hadoop Configuration to handle warning about deprecated
configs and
// to set the correct non-deprecated configs when an old one shows up.
@@ -89,28 +95,105 @@ public class UncoveredGlobalIndexRegionScanner extends
UncoveredIndexRegionScann
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);
- final Configuration config = env.getConfiguration();
- hTableFactory = IndexWriterUtils.getDefaultDelegateHTableFactory(env);
- rowCountPerTask =
- config.getInt(INDEX_ROW_COUNTS_PER_TASK_CONF_KEY,
DEFAULT_INDEX_ROW_COUNTS_PER_TASK);
- byte[] dataTableName = scan.getAttribute(PHYSICAL_DATA_TABLE_NAME);
- dataHTable = hTableFactory.getTable(new ImmutableBytesPtr(dataTableName));
- regionEndKeys =
-
hTableFactory.getConnection().getRegionLocator(dataHTable.getName()).getEndKeys();
- if (indexMaintainer.isUncovered()) {
- // Empty column should also be added to the data columns to join for
uncovered
- // global indexes. This is required to verify the index row against the
data table row and
- // repair it
+ // 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. This must happen before
TTLRegionScanner.init().
+ this(innerScanner, region, scan, env, dataTableScan, tupleProjector,
indexMaintainer,
+ viewConstants, ptr, pageSizeMs, queryLimit, isDistinct,
+ !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 {
+ // Keep a stable delegate for the superclass, and reopen only inside the
guarded body.
+ // If superclass initialization fails, the caller still owns the original
scanner.
+ super(pageFilter == null ? innerScanner : new
DelegateRegionScanner(innerScanner), region, scan,
+ env, dataTableScan, tupleProjector, indexMaintainer, viewConstants, ptr,
pageSizeMs,
+ pageFilter == null ? queryLimit : pageFilter.getPageSize(), isDistinct);
+ maxRowCount = pageFilter == null ? Long.MAX_VALUE :
pageFilter.getPageSize();
+ RegionScanner reopened = null;
+ HTableFactory tableFactory = null;
+ Table table = null;
+ try {
+ if (pageFilter != null) {
+ reopened = reopenScannerWithoutPageFilter(innerScanner, scan,
pageFilter);
+ ((DelegateRegionScanner) this.innerScanner).delegate = reopened;
+ }
+ final Configuration config = env.getConfiguration();
+ tableFactory = IndexWriterUtils.getDefaultDelegateHTableFactory(env);
+ rowCountPerTask =
+ config.getInt(INDEX_ROW_COUNTS_PER_TASK_CONF_KEY,
DEFAULT_INDEX_ROW_COUNTS_PER_TASK);
+ byte[] dataTableName = scan.getAttribute(PHYSICAL_DATA_TABLE_NAME);
+ table = tableFactory.getTable(new ImmutableBytesPtr(dataTableName));
+ regionEndKeys =
tableFactory.getConnection().getRegionLocator(table.getName()).getEndKeys();
+ // Include the empty column to verify the index row against its data row
and repair it.
ScanUtil.addEmptyColumnToScan(dataTableScan,
indexMaintainer.getDataEmptyKeyValueCF(),
indexMaintainer.getEmptyKeyValueQualifierForDataTable());
+ isScanMetricsEnabled =
+ scan.isScanMetricsEnabled() &&
CompatScanMetrics.supportsFineGrainedReadMetrics();
+ if (isScanMetricsEnabled) {
+ dataTableScanMetrics = new ArrayList<>();
+ }
+ hTableFactory = tableFactory;
+ dataHTable = table;
+ } catch (IOException | RuntimeException | Error failure) {
+ // Until construction succeeds, the caller cannot see or close the
replacement scanner.
+ // The caller still owns the original scanner, so close only what was
opened here.
+ if (reopened != null) {
+ closeOnFailure(reopened, failure);
+ }
+ if (table != null) {
+ closeOnFailure(table, failure);
+ }
+ if (tableFactory != null) {
+ closeOnFailure(tableFactory::shutdown, failure);
+ }
+ throw failure;
}
- isScanMetricsEnabled =
- scan.isScanMetricsEnabled() &&
CompatScanMetrics.supportsFineGrainedReadMetrics();
- if (isScanMetricsEnabled) {
- dataTableScanMetrics = new ArrayList<>();
+ }
+
+ private static void closeOnFailure(AutoCloseable resource, Throwable
failure) {
+ try {
+ resource.close();
+ } catch (Throwable cleanupFailure) {
+ if (cleanupFailure != failure) {
+ failure.addSuppressed(cleanupFailure);
+ }
+ }
+ }
+
+ private static RegionScanner reopenScannerWithoutPageFilter(RegionScanner
scanner, Scan scan,
+ PageFilter pageFilter) throws IOException {
+ if (pageFilter == null) {
+ return scanner;
+ }
+ try {
+ DelegateRegionScanner delegate = (DelegateRegionScanner) scanner;
+ scanner.close();
+ return delegate.getNewRegionScanner(scan);
+ } catch (ClassCastException e) {
+ throw new DoNotRetryIOException(e);
+ }
+ }
+
+ @Override
+ public boolean next(List<Cell> result, ScannerContext scannerContext) throws
IOException {
+ if (returnedRowCount >= maxRowCount) {
+ result.clear();
+ return false;
+ }
+ boolean hasMore = super.next(result, scannerContext);
+ if (!result.isEmpty() && !ScanUtil.isDummy(result)) {
+ returnedRowCount++;
+ if (returnedRowCount >= maxRowCount) {
+ return false;
+ }
}
+ return hasMore;
}
@Override
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexLimitIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexLimitIT.java
new file mode 100644
index 0000000000..afd7f28c5b
--- /dev/null
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexLimitIT.java
@@ -0,0 +1,421 @@
+/*
+ * 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.query.QueryServicesOptions;
+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;
+ // Force multiple index batches before and after the first valid row.
+ private static final long INDEX_PAGE_SIZE_IN_ROWS = 3;
+ 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));
+ props.put(QueryServices.INDEX_PAGE_SIZE_IN_ROWS,
Long.toString(INDEX_PAGE_SIZE_IN_ROWS));
+ // 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;
+ LimitTestObserver.lookupTable = 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.dummyResults.set(0);
+ LimitTestObserver.indexTable = fixture.indexTable;
+ LimitTestObserver.dataTable = fixture.dataTable;
+ assertRows(conn, fixture.indexQuery("") + " LIMIT " + LIMIT, LIMIT);
+ assertTrue("The query must return actual PagingFilter dummy results",
+ LimitTestObserver.dummyResults.get() > 0);
+ assertStaleEntriesPresent(fixture);
+ }
+ }
+
+ @Test
+ public void testSmallLimitCapsDataLookupsAtDefaultBatchSize() throws
Exception {
+ // Use the production default rather than the small batches used by the
other regressions.
+ getUtility().getHBaseCluster().getRegionServerThreads()
+ .forEach(thread -> thread.getRegionServer().getConfiguration().setLong(
+ QueryServices.INDEX_PAGE_SIZE_IN_ROWS,
+ QueryServicesOptions.DEFAULT_INDEX_PAGE_SIZE_IN_ROWS));
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ for (int staleRowCount : new int[] { 0, 3 }) {
+ try {
+ Fixture fixture = createFixture(conn, false, 1000, staleRowCount);
+ LimitTestObserver.lookupRows.set(0);
+ LimitTestObserver.lookupTable = fixture.dataTable;
+ String query = fixture.indexQuery("") + " LIMIT 1";
+ assertPlan(conn, query).table(fixture.indexTable).serverRowLimit(1L);
+ assertRows(conn, query, 1);
+ assertEquals("LIMIT 1 must fetch only the stale prefix and the first
valid candidate",
+ staleRowCount + 1, LimitTestObserver.lookupRows.get());
+ assertStaleEntriesPresent(fixture);
+ } finally {
+ EnvironmentEdgeManager.reset();
+ }
+ }
+ } finally {
+ getUtility().getHBaseCluster().getRegionServerThreads()
+ .forEach(thread -> thread.getRegionServer().getConfiguration()
+ .setLong(QueryServices.INDEX_PAGE_SIZE_IN_ROWS,
INDEX_PAGE_SIZE_IN_ROWS));
+ }
+ }
+
+ @Test
+ public void testServerLimitCountsOnlyVerifiedRows() throws Exception {
+ for (boolean finiteTTL : new boolean[] { false, true }) {
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ Fixture fixture = createFixture(conn, finiteTTL, 1000);
+ LimitTestObserver.returnedRows.set(0);
+ LimitTestObserver.limitedTable = fixture.indexTable;
+ assertRows(conn, fixture.indexQuery("") + " LIMIT " + LIMIT, LIMIT);
+ // Client-side LIMIT can hide an unbounded server scan. Count rows
actually sent by HBase.
+ assertEquals("Server LIMIT must count verified rows", LIMIT,
+ LimitTestObserver.returnedRows.get());
+ assertStaleEntriesPresent(fixture);
+ } finally {
+ LimitTestObserver.limitedTable = null;
+ EnvironmentEdgeManager.reset();
+ }
+ }
+ }
+
+ private Fixture createFixture(Connection conn, boolean finiteTTL, long
updateDelayMs)
+ throws Exception {
+ return createFixture(conn, finiteTTL, updateDelayMs, ROW_COUNT);
+ }
+
+ private Fixture createFixture(Connection conn, boolean finiteTTL, long
updateDelayMs,
+ int staleRowCount) throws Exception {
+ String dataTable = generateUniqueName();
+ String indexTable = generateUniqueName();
+ try (Statement stmt = conn.createStatement()) {
+ stmt.execute("CREATE TABLE " + dataTable
+ + " (ID VARCHAR NOT NULL PRIMARY KEY, UPDATED_TIME TIMESTAMP, PAYLOAD
VARCHAR)"
+ + (finiteTTL ? " TTL=" + TTL_SECONDS + ", IS_STRICT_TTL=true" : ""));
+ stmt.execute("CREATE UNCOVERED INDEX " + indexTable + " ON " + dataTable
+ " (UPDATED_TIME)");
+ }
+ clock.setValue(EnvironmentEdgeManager.currentTimeMillis() + 1);
+ EnvironmentEdgeManager.injectEdge(clock);
+ long insertTime = clock.currentTime();
+ upsertRows(conn, dataTable, false, staleRowCount);
+ List<Result> initialRows = readPhysicalRows(indexTable);
+ assertEquals(ROW_COUNT, initialRows.size());
+ // This unsalted index orders NULL-keyed entries before the non-NULL
entries.
+ List<Result> staleRows = new ArrayList<>(initialRows.subList(0,
staleRowCount));
+ assertEquals("Initial NULL-keyed index entries", staleRowCount,
staleRows.size());
+ for (Result stale : staleRows) {
+ assertEquals(insertTime, stale.rawCells()[0].getTimestamp());
+ }
+
+ clock.incrementValue(updateDelayMs);
+ // Supply the indexed column explicitly to take the write-optimized
uncovered-index path.
+ upsertRows(conn, dataTable, true, staleRowCount);
+ Fixture fixture = new Fixture(dataTable, indexTable, staleRows);
+ assertEquals("Both stale and current index entries must physically exist",
+ ROW_COUNT + staleRowCount, readPhysicalRows(indexTable).size());
+ assertStaleEntriesPresent(fixture);
+ for (Result data : readPhysicalRows(dataTable)) {
+ assertEquals("Base rows must have the recent update timestamp",
clock.currentTime(),
+ data.rawCells()[0].getTimestamp());
+ }
+ return fixture;
+ }
+
+ private void upsertRows(Connection conn, String dataTable, boolean
nonNullTimestamp,
+ int staleRowCount) throws Exception {
+ try (PreparedStatement stmt = conn.prepareStatement(
+ "UPSERT INTO " + dataTable + " (ID, UPDATED_TIME, PAYLOAD) VALUES (?, ?,
?)")) {
+ for (int i = 0; i < ROW_COUNT; i++) {
+ stmt.setString(1, "tenant-" + i);
+ if (nonNullTimestamp || i >= staleRowCount) {
+ stmt.setTimestamp(2, new Timestamp(UPDATED_TIME_BASE + i * 1000L));
+ } else {
+ stmt.setNull(2, Types.TIMESTAMP);
+ }
+ stmt.setString(3, "payload-" + i);
+ stmt.executeUpdate();
+ }
+ conn.commit();
+ }
+ }
+
+ private List<Result> readPhysicalRows(String tableName) throws Exception {
+ // No Phoenix scan attributes: inspect live HBase entries without
uncovered verification.
+ List<Result> rows = new ArrayList<>();
+ try (Table table =
getUtility().getConnection().getTable(TableName.valueOf(tableName));
+ ResultScanner scanner = table.getScanner(new Scan())) {
+ for (Result result : scanner) {
+ rows.add(result);
+ }
+ }
+ return rows;
+ }
+
+ private void assertStaleEntriesPresent(Fixture fixture) throws Exception {
+ try (
+ Table index =
getUtility().getConnection().getTable(TableName.valueOf(fixture.indexTable))) {
+ for (Result stale : fixture.staleRows) {
+ Result current = index.get(new Get(stale.getRow()));
+ assertFalse("Fixture lost a stale index entry", current.isEmpty());
+ assertEquals("Stale entry timestamp must not advance with the data
row",
+ stale.rawCells()[0].getTimestamp(),
current.rawCells()[0].getTimestamp());
+ }
+ }
+ }
+
+ private void assertRows(Connection conn, String query, int expectedRows)
throws Exception {
+ assertRows(conn, query, expectedRows, 0);
+ }
+
+ private void assertRows(Connection conn, String query, int expectedRows, int
firstRow)
+ throws Exception {
+ try (Statement stmt = conn.createStatement(); ResultSet rs =
stmt.executeQuery(query)) {
+ int count = 0;
+ while (rs.next()) {
+ assertEquals("tenant-" + (firstRow + count), rs.getString(1));
+ assertEquals("payload-" + (firstRow + count), rs.getString(2));
+ assertEquals(new Timestamp(UPDATED_TIME_BASE + (firstRow + count) *
1000L),
+ rs.getTimestamp(3));
+ count++;
+ }
+ assertEquals("Valid rows returned by: " + query, expectedRows, count);
+ }
+ }
+
+ public static class LimitTestObserver extends SimpleRegionObserver {
+ private static volatile String indexTable;
+ private static volatile String dataTable;
+ private static final AtomicInteger dummyResults = new AtomicInteger();
+ private static volatile String limitedTable;
+ private static final AtomicInteger returnedRows = new AtomicInteger();
+ private static volatile String lookupTable;
+ private static final AtomicInteger lookupRows = new AtomicInteger();
+
+ @Override
+ public void preScannerOpen(ObserverContext<RegionCoprocessorEnvironment>
context, Scan scan) {
+ String table =
+
context.getEnvironment().getRegion().getTableDescriptor().getTableName().getNameAsString();
+ if (table.equals(indexTable) || table.equals(dataTable)) {
+ scan.setAttribute(SERVER_PAGE_SIZE_MS, Bytes.toBytes(0L));
+ if (scan.getFilter() instanceof PagingFilter) {
+ scan
+ .setFilter(new PagingFilter(((PagingFilter)
scan.getFilter()).getDelegateFilter(), 0));
+ }
+ }
+ }
+
+ @Override
+ public boolean
postScannerNext(ObserverContext<RegionCoprocessorEnvironment> context,
+ InternalScanner scanner, List<Result> results, int limit, boolean
hasMore) {
+ String table =
+
context.getEnvironment().getRegion().getTableDescriptor().getTableName().getNameAsString();
+ for (Result result : results) {
+ if (table.equals(indexTable) || table.equals(dataTable)) {
+ if (ScanUtil.isDummy(result)) {
+ dummyResults.incrementAndGet();
+ }
+ }
+ if (table.equals(lookupTable) && !result.isEmpty() &&
!ScanUtil.isDummy(result)) {
+ lookupRows.incrementAndGet();
+ }
+ }
+ if (table.equals(limitedTable)) {
+ for (Result result : results) {
+ if (!result.isEmpty() && !ScanUtil.isDummy(result)) {
+ returnedRows.incrementAndGet();
+ }
+ }
+ }
+ return hasMore;
+ }
+ }
+
+ private static class Fixture {
+ final String dataTable;
+ final String indexTable;
+ final List<Result> staleRows;
+
+ Fixture(String dataTable, String indexTable, List<Result> staleRows) {
+ this.dataTable = dataTable;
+ this.indexTable = indexTable;
+ this.staleRows = staleRows;
+ }
+
+ String indexQuery(String predicate) {
+ return "SELECT /*+ INDEX(" + dataTable + " " + indexTable
+ + ") */ ID, PAYLOAD, UPDATED_TIME FROM " + dataTable + predicate
+ + " ORDER BY UPDATED_TIME ASC";
+ }
+ }
+}
diff --git
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexRegionScannerIT.java
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexRegionScannerIT.java
index f3c8503457..a29dc7b1e5 100644
---
a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexRegionScannerIT.java
+++
b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/UncoveredGlobalIndexRegionScannerIT.java
@@ -912,6 +912,123 @@ public class UncoveredGlobalIndexRegionScannerIT extends
BaseTest {
}
}
+ @Test
+ public void testUncoveredIndexLimitSkipsStaleRows() throws Exception {
+ assumeTrue(uncovered);
+ assertUncoveredIndexLimit(false, 0);
+ }
+
+ @Test
+ public void testUncoveredIndexLimitWithTTLMasking() throws Exception {
+ assumeTrue(uncovered);
+ assertUncoveredIndexLimit(true, 0);
+ }
+
+ @Test
+ public void testUncoveredIndexLimitWithOffset() throws Exception {
+ assumeTrue(uncovered);
+ assertUncoveredIndexLimit(false, 2);
+ }
+
+ private void assertUncoveredIndexLimit(boolean finiteTTL, int offset) throws
Exception {
+ Map<Configuration, String> originalThresholds = new LinkedHashMap<>();
+ Configuration clusterConfig = getUtility().getConfiguration();
+ originalThresholds.put(clusterConfig,
+
clusterConfig.getRaw(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB));
+ getUtility().getHBaseCluster().getRegionServerThreads().forEach(thread -> {
+ Configuration config = thread.getRegionServer().getConfiguration();
+ originalThresholds.putIfAbsent(config,
+
config.getRaw(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB));
+ });
+ EnvironmentEdge originalClock = EnvironmentEdgeManager.getDelegate();
+ ManualEnvironmentEdge clock = new ManualEnvironmentEdge();
+ try {
+ // Keep stale entries young so LIMIT correctness cannot depend on
read-repair deletion.
+ originalThresholds.keySet().forEach(config -> config
+
.setLong(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB,
60_000));
+ try (Connection conn = DriverManager.getConnection(getUrl());
+ Statement stmt = conn.createStatement()) {
+ String dataTable = generateUniqueName();
+ String indexTable = generateUniqueName();
+ stmt.execute("CREATE TABLE " + dataTable
+ + " (ID VARCHAR NOT NULL PRIMARY KEY, UPDATED_TIME TIMESTAMP, VAL
VARCHAR)"
+ + (finiteTTL ? " TTL=3600, IS_STRICT_TTL=true" : "")
+ + (salted ? (finiteTTL ? ", " : " ") + "SALT_BUCKETS=4" : ""));
+ stmt.execute("CREATE UNCOVERED INDEX " + indexTable + " ON " +
dataTable + " (UPDATED_TIME)"
+ + (salted ? " SALT_BUCKETS=4" : ""));
+ long insertTime = EnvironmentEdgeManager.currentTimeMillis() + 1;
+ long timestampBase = 1_700_000_000_000L;
+ int rows = 20;
+ int limit = 3;
+ clock.setValue(insertTime);
+ EnvironmentEdgeManager.injectEdge(clock);
+ upsertLimitRows(conn, dataTable, rows, timestampBase, false);
+ List<byte[]> staleKeys = new ArrayList<>();
+ try (Table index =
getUtility().getConnection().getTable(TableName.valueOf(indexTable))) {
+ // No Phoenix attributes: inspect the physical fixture without
verification or repair.
+ try (ResultScanner scanner = index.getScanner(new Scan())) {
+ for (Result row : scanner) {
+ staleKeys.add(row.getRow());
+ assertEquals(insertTime, row.rawCells()[0].getTimestamp());
+ }
+ }
+ assertEquals(rows, staleKeys.size());
+ clock.incrementValue(1000);
+ upsertLimitRows(conn, dataTable, rows, timestampBase, true);
+ for (byte[] key : staleKeys) {
+ assertFalse("The fixture must retain its stale index entries",
+ index.get(new Get(key)).isEmpty());
+ }
+ String sql = "SELECT /*+ INDEX(" + dataTable + " " + indexTable
+ + ") */ ID, VAL, UPDATED_TIME FROM " + dataTable + " ORDER BY
UPDATED_TIME ASC LIMIT "
+ + limit + (offset == 0 ? "" : " OFFSET " + offset);
+ assertPlan(conn, sql).table(indexTable).scanType("FULL SCAN");
+ try (ResultSet rs = stmt.executeQuery(sql)) {
+ for (int i = offset; i < offset + limit; i++) {
+ assertTrue("LIMIT must return valid rows after the stale
prefix", rs.next());
+ assertEquals("id-" + i, rs.getString(1));
+ assertEquals("value-" + i, rs.getString(2));
+ assertEquals(new Timestamp(timestampBase + i * 1000L),
rs.getTimestamp(3));
+ }
+ assertFalse(rs.next());
+ }
+ for (byte[] key : staleKeys) {
+ Result stale = index.get(new Get(key));
+ assertFalse("LIMIT must work while young stale entries remain",
stale.isEmpty());
+ assertEquals(insertTime, stale.rawCells()[0].getTimestamp());
+ }
+ }
+ }
+ } finally {
+ EnvironmentEdgeManager.injectEdge(originalClock);
+ originalThresholds.forEach((config, value) -> {
+ if (value == null) {
+
config.unset(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB);
+ } else {
+
config.set(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB,
value);
+ }
+ });
+ }
+ }
+
+ private void upsertLimitRows(Connection conn, String dataTable, int rows,
long timestampBase,
+ boolean nonNullTimestamp) throws SQLException {
+ try (PreparedStatement stmt = conn
+ .prepareStatement("UPSERT INTO " + dataTable + " (ID, UPDATED_TIME, VAL)
VALUES (?, ?, ?)")) {
+ for (int i = 0; i < rows; i++) {
+ stmt.setString(1, "id-" + i);
+ if (nonNullTimestamp) {
+ stmt.setTimestamp(2, new Timestamp(timestampBase + i * 1000L));
+ } else {
+ stmt.setNull(2, Types.TIMESTAMP);
+ }
+ stmt.setString(3, "value-" + i);
+ stmt.executeUpdate();
+ }
+ conn.commit();
+ }
+ }
+
/** PHOENIX-8016: an old index entry must be aged independently of its
recent base row. */
@Test
public void testOldStaleEntriesAreDeletedWhileDataRowsAreRecent() throws
Exception {
diff --git
a/phoenix-core/src/test/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScannerTest.java
b/phoenix-core/src/test/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScannerTest.java
new file mode 100644
index 0000000000..1aa2f556df
--- /dev/null
+++
b/phoenix-core/src/test/java/org/apache/phoenix/coprocessor/UncoveredGlobalIndexRegionScannerTest.java
@@ -0,0 +1,165 @@
+/*
+ * 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.coprocessor;
+
+import static
org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants.PHYSICAL_DATA_TABLE_NAME;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.RegionLocator;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.filter.PageFilter;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.regionserver.Region;
+import org.apache.hadoop.hbase.regionserver.RegionScanner;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.index.IndexMaintainer;
+import org.apache.phoenix.util.ServerUtil.ConnectionFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class UncoveredGlobalIndexRegionScannerTest {
+ private final RegionCoprocessorEnvironment env =
mock(RegionCoprocessorEnvironment.class);
+ private final Connection connection = mock(Connection.class);
+ private final Table table = mock(Table.class);
+ private final RegionLocator locator = mock(RegionLocator.class);
+ private final DelegateRegionScanner original =
mock(DelegateRegionScanner.class);
+ private final RegionScanner replacement = mock(RegionScanner.class);
+ private final IndexMaintainer maintainer = mock(IndexMaintainer.class);
+ private final Scan scan = new Scan().withStartRow(Bytes.toBytes("start"));
+
+ @Before
+ public void setUp() throws Exception {
+ ConnectionFactory.shutdown();
+ when(env.getConfiguration()).thenReturn(new Configuration(false));
+
when(env.createConnection(any(Configuration.class))).thenReturn(connection);
+ when(connection.getTable(any(TableName.class))).thenReturn(table);
+
when(connection.getRegionLocator(any(TableName.class))).thenReturn(locator);
+ when(table.getName()).thenReturn(TableName.valueOf("DATA"));
+
when(original.getNewRegionScanner(any(Scan.class))).thenReturn(replacement);
+ scan.setAttribute(PHYSICAL_DATA_TABLE_NAME, Bytes.toBytes("DATA"));
+ scan.setFilter(new PageFilter(1));
+ }
+
+ @After
+ public void tearDown() {
+ ConnectionFactory.shutdown();
+ }
+
+ @Test
+ public void testTableOpenFailureClosesReplacement() throws Exception {
+ IOException failure = new IOException("table open failed");
+ when(connection.getTable(any(TableName.class))).thenThrow(failure);
+ assertConstructionFails(original, failure);
+ verify(original).close();
+ verify(replacement).close();
+ verify(table, never()).close();
+ }
+
+ @Test
+ public void testRegionLookupFailureClosesReplacementAndTable() throws
Exception {
+ IOException failure = new IOException("region lookup failed");
+ when(locator.getEndKeys()).thenThrow(failure);
+ assertConstructionFails(original, failure);
+ verify(replacement).close();
+ verify(table).close();
+ }
+
+ @Test
+ public void testCleanupFailuresPreserveInitializationFailure() throws
Exception {
+ IOException failure = new IOException("region lookup failed");
+ IOException scannerCloseFailure = new IOException("scanner close failed");
+ IOException tableCloseFailure = new IOException("table close failed");
+ when(locator.getEndKeys()).thenThrow(failure);
+ doThrow(scannerCloseFailure).when(replacement).close();
+ doThrow(tableCloseFailure).when(table).close();
+ assertConstructionFails(original, failure);
+ assertEquals(2, failure.getSuppressed().length);
+ assertSame(scannerCloseFailure, failure.getSuppressed()[0]);
+ assertSame(tableCloseFailure, failure.getSuppressed()[1]);
+ verify(table).close();
+ }
+
+ @Test
+ public void testSuperclassFailureDoesNotReopenScanner() throws Exception {
+ RuntimeException failure = new IllegalArgumentException("configuration
failed");
+ when(env.getConfiguration()).thenThrow(failure);
+ try {
+ construct(original);
+ fail("Expected superclass initialization to fail");
+ } catch (RuntimeException actual) {
+ assertSame(failure, actual);
+ }
+ verify(original, never()).getNewRegionScanner(any(Scan.class));
+ verify(original, never()).close();
+ }
+
+ @Test
+ public void testFailureWithoutPageFilterLeavesOriginalOwnedByCaller() throws
Exception {
+ scan.setFilter(null);
+ IOException failure = new IOException("region lookup failed");
+ when(locator.getEndKeys()).thenThrow(failure);
+ assertConstructionFails(original, failure);
+ verify(original, never()).close();
+ verify(original, never()).getNewRegionScanner(any(Scan.class));
+ verify(table).close();
+ }
+
+ @Test
+ public void testUnsupportedScannerIsNotRetriedAndLeftOwnedByCaller() throws
Exception {
+ RegionScanner unsupported = mock(RegionScanner.class);
+ try {
+ construct(unsupported);
+ fail("Expected incompatible scanner to fail");
+ } catch (DoNotRetryIOException failure) {
+ assertTrue(failure.getCause() instanceof ClassCastException);
+ }
+ verify(unsupported, never()).close();
+ }
+
+ private void assertConstructionFails(RegionScanner scanner, IOException
expected)
+ throws Exception {
+ try {
+ construct(scanner);
+ fail("Expected scanner initialization to fail");
+ } catch (IOException actual) {
+ assertSame(expected, actual);
+ }
+ }
+
+ private void construct(RegionScanner scanner) throws IOException {
+ new UncoveredGlobalIndexRegionScanner(scanner, mock(Region.class), scan,
env, new Scan(), null,
+ maintainer, null, new ImmutableBytesWritable(), 60_000, -1, false);
+ }
+}
