diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/execute/MutationState.java b/phoenix-core-client/src/main/java/org/apache/phoenix/execute/MutationState.java index 52517a884df..b0bfac0b047 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/execute/MutationState.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/execute/MutationState.java @@ -1528,6 +1528,10 @@ private void sendMutations(Iterator>> mutationsI // no-op if table doesn't have Conditional TTL ScanUtil.annotateMutationWithConditionalTTL(connection, tableInfo.getPTable(), mutationList); + // no-op unless table/view has a literal TTL; threads the empty-column CF/CQ (plus a view's + // literal TTL and any non-strict flag) so the internal current-row scan masks like a client + // read + ScanUtil.annotateMutationWithLiteralTTL(connection, tableInfo.getPTable(), mutationList); // If we haven't retried yet, retry for this case only, as it's possible that // a split will occur after we send the index metadata cache to all known // region servers. diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java index 5a8334ed63e..2c27a57fd37 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java @@ -5471,14 +5471,14 @@ private void createSnapshot(String snapshotName, String tableName) throws SQLExc // is safe. // A second flavor of transient failure is "already running another snapshot on the same table", // produced by RPC-level retries. The original snapshot() RPC has already been accepted by the - // master and the snapshot procedure is in flight, but the client retries the call and the master + // master and the snapshot procedure is in flight, but the client retries the call and the + // master // rejects the duplicate. In that case the existing snapshot procedure will succeed, so we treat // it as success once the snapshot of that name shows up in listSnapshots(). If it never appears // within the polling window we fall through to a normal retry. final int maxAttempts = 5; final long backoffMs = 1000L; - final long alreadyRunningWaitMs = - TimeUnit.MINUTES.toMillis(2L); + final long alreadyRunningWaitMs = TimeUnit.MINUTES.toMillis(2L); SQLException sqlE = null; for (int attempt = 1; attempt <= maxAttempts; attempt++) { sqlE = null; diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java index 22833e24945..e70aba774ff 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java @@ -4923,9 +4923,9 @@ public MutationState addColumn(PTable table, List origColumnDefs, /** * To check if TTL is defined at any of the child below we are checking it at * {@link org.apache.phoenix.coprocessor.MetaDataEndpointImpl#mutateColumn(List, ColumnMutator, int, PTable, PTable, boolean)} - * level where in function - * {@link org.apache.phoenix.coprocessor.MetaDataEndpointImpl# validateIfMutationAllowedOnParent(PTable, List, PTableType, long, byte[], byte[], byte[], List, int)} - * we are already traversing through allDescendantViews. + * level where in function {@link org.apache.phoenix.coprocessor.MetaDataEndpointImpl# + * validateIfMutationAllowedOnParent(PTable, List, PTableType, long, byte[], byte[], + * byte[], List, int)} we are already traversing through allDescendantViews. */ } diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/util/ScanUtil.java b/phoenix-core-client/src/main/java/org/apache/phoenix/util/ScanUtil.java index 1dfc5b1e615..0f40f68ad4e 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/util/ScanUtil.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/util/ScanUtil.java @@ -92,6 +92,7 @@ import org.apache.phoenix.query.QueryServicesOptions; import org.apache.phoenix.schema.CompiledTTLExpression; import org.apache.phoenix.schema.IllegalDataException; +import org.apache.phoenix.schema.LiteralTTLExpression; import org.apache.phoenix.schema.PColumn; import org.apache.phoenix.schema.PName; import org.apache.phoenix.schema.PTable; @@ -1862,6 +1863,107 @@ public static void annotateMutationWithConditionalTTL(PhoenixConnection connecti } } + /** + * Annotates mutations for a table/view with a literal TTL so the server-side internal current-row + * scan (IndexRegionObserver.getCurrentRowStates) can mask expired rows exactly like a client + * read. This is the literal-TTL sibling of {@link #annotateMutationWithConditionalTTL}; the two + * are disjoint (one is guarded on a literal expression, the other on a conditional expression) so + * at most one fires for a given table. + *

+ * The guiding principle is to thread precisely what the read path + * ({@link #setScanAttributesForPhoenixTTL}) would set as the {@code _TTL} scan attribute: + *

+ */ + public static void annotateMutationWithLiteralTTL(PhoenixConnection connection, PTable table, + List mutations) throws SQLException { + + if (!(table.getTTLExpression() instanceof LiteralTTLExpression)) { + // Conditional TTL is handled by annotateMutationWithConditionalTTL; NONE has no literal to + // thread. Only a literal TTL (finite or FOREVER) reaches the internal-scan masking path. + return; + } + // NOTE: unlike annotateMutationWithConditionalTTL, we do NOT skip immutable tables here. For + // conditional TTL the server reads the current row only when context.hasConditionalTTL, which + // is + // itself derived from the _TTL mutation attribute this annotation would set - so skipping + // immutable tables is self-consistent (no attribute => no conditional read). For a LITERAL TTL + // the server-side current-row read in IndexRegionObserver.getCurrentRowStates is triggered by + // table/index structure and mutation type, NOT by any attribute set here: a global index whose + // data/index storage schemes differ leaves context.immutableRows false (see + // identifyIndexMaintainerTypes) and reads the row to rebuild the full index entry, and the + // atomic / returnResult / row-delete paths read it regardless of immutability. If we skipped + // immutable tables, that read would be unmasked and could rebuild the index from TTL-expired + // cells - the very divergence this masking exists to prevent. The threaded attributes are inert + // (they only enable masking on a scan TTLRegionScanner independently gates) when no current-row + // read happens, so annotating immutable tables is safe. + + // For a view, honor the view-TTL feature flag exactly as setScanAttributesForPhoenixTTL does. + boolean isView = table.getType() == PTableType.VIEW; + boolean viewTTLEnabled = !isView || connection.getQueryServices().getConfiguration().getBoolean( + QueryServices.PHOENIX_VIEW_TTL_ENABLED, + QueryServicesOptions.DEFAULT_PHOENIX_VIEW_TTL_ENABLED); + + // The view's literal TTL is threaded as _TTL only for a view with view-TTL enabled, since it is + // not on the shared CF descriptor. A view with view-TTL disabled sets no _TTL on the read path + // (it returns after only IS_STRICT_TTL), so neither do we; a base table's literal TTL is on the + // CF descriptor, so the server's TTLRegionScanner fallback derives it and we thread no _TTL. + byte[] ttlForScan = null; + if (isView && viewTTLEnabled) { + // serialize() is non-null for FOREVER and finite literals and null only for NONE, the exact + // non-null filter the read path uses at setScanAttributesForPhoenixTTL. FOREVER must be + // threaded, not skipped. + ttlForScan = table.getCompiledTTLExpression(connection).serialize(); + } + + // The empty-column CF/CQ are threaded unconditionally, exactly as the client read path + // (setScanAttributesForClient) sets them on every non-analyze scan. They only identify the + // table's empty column and enable masking; TTLRegionScanner still independently requires an + // effective, non-FOREVER, strict TTL to actually mask, so setting them whenever a current-row + // read may happen makes the internal scan mask identically to a client read. They are also the + // only source of these values for the no-index current-row read (atomic / ON DUPLICATE KEY / + // returnResult / row-delete), which has no IndexMaintainer on the server. + byte[] emptyCF = SchemaUtil.getEmptyColumnFamily(table); + byte[] emptyCQ = SchemaUtil.getEmptyColumnQualifier(table); + + byte[] isStrictTTL = + table.isStrictTTL() ? null : PBoolean.INSTANCE.toBytes(table.isStrictTTL()); + for (Mutation mutation : mutations) { + if (ttlForScan != null) { + mutation.setAttribute(BaseScannerRegionObserverConstants.TTL, ttlForScan); + } + if (isStrictTTL != null) { + mutation.setAttribute(BaseScannerRegionObserverConstants.IS_STRICT_TTL, isStrictTTL); + } + mutation.setAttribute(BaseScannerRegionObserverConstants.EMPTY_COLUMN_FAMILY_NAME, emptyCF); + mutation.setAttribute(BaseScannerRegionObserverConstants.EMPTY_COLUMN_QUALIFIER_NAME, + emptyCQ); + } + } + public static PageFilter removePageFilterFromFilterList(FilterList filterList) { Iterator filterIterator = filterList.getFilters().iterator(); while (filterIterator.hasNext()) { diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/ServerScanUtil.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/ServerScanUtil.java new file mode 100644 index 00000000000..fa61cb0b9a8 --- /dev/null +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/ServerScanUtil.java @@ -0,0 +1,138 @@ +/* + * 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 java.io.IOException; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.client.Scan; +import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; +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.coprocessorclient.BaseScannerRegionObserverConstants; +import org.apache.phoenix.filter.PagingFilter; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.query.QueryServicesOptions; +import org.apache.phoenix.schema.types.PBoolean; +import org.apache.phoenix.util.ScanUtil; + +/** + * Utilities for internal server-side region scans that must honor Phoenix TTL exactly like a client + * read. The client normally sets the empty-column and TTL scan attributes + * ({@link org.apache.phoenix.util.ScanUtil#setScanAttributesForPhoenixTTL}) and the coprocessor + * hook {@code BaseScannerRegionObserver.postScannerOpen} wraps the scan in a + * {@link TTLRegionScanner}. Internal scans opened directly via {@code region.getScanner(scan)} + * bypass that hook, so they set no attributes and are never TTL-masked. These helpers reproduce + * both steps for server-side callers (e.g. {@code IndexRegionObserver} current-row reads) so an + * internal scan masks identically to a client scan. + *

+ * This class lives in the {@code org.apache.phoenix.coprocessor} package so it can reference the + * server-only {@link TTLRegionScanner} and {@link PagingRegionScanner}; the client-side + * {@code ScanUtil} cannot. + */ +public class ServerScanUtil { + + private ServerScanUtil() { + } + + /** + * Sets the Phoenix TTL and paging scan attributes on an internal data-table scan so it behaves + * exactly like a client read. + *

+ * TTL masking attributes ({@link TTLRegionScanner} reads these): + *

    + *
  • the empty-column CF/CQ, supplied by the caller from the bytes the client threaded on the + * mutation ({@link org.apache.phoenix.util.ScanUtil#annotateMutationWithLiteralTTL}) — the single + * source for every path, secondary-index and no-index (atomic / ON DUPLICATE KEY / + * {@code returnResult} / row-delete) alike;
    • + *
  • {@code IS_STRICT_TTL=false} when {@code isStrictTTL == false}, so a non-strict table is not + * masked (absence of the attribute defaults to strict, matching the read path);
    • + *
  • the view's literal TTL as the standard {@code _TTL} scan attribute when + * {@code literalTTLForScan != null}. A base table's literal TTL is left unset so + * {@link TTLRegionScanner}'s CF-descriptor fallback derives it.
    • + *
+ */ + public static void setInternalScanAttributes(Configuration conf, Scan scan, byte[] emptyCF, + byte[] emptyCQ, byte[] literalTTLForScan, boolean isStrictTTL) { + scan.setAttribute(BaseScannerRegionObserverConstants.EMPTY_COLUMN_FAMILY_NAME, emptyCF); + scan.setAttribute(BaseScannerRegionObserverConstants.EMPTY_COLUMN_QUALIFIER_NAME, emptyCQ); + if (!isStrictTTL) { + // Absence of the attribute defaults to strict-true (ScanUtil.isStrictTTL), so only set it + // when the table/view is non-strict, mirroring setScanAttributesForPhoenixTTL. + scan.setAttribute(BaseScannerRegionObserverConstants.IS_STRICT_TTL, + PBoolean.INSTANCE.toBytes(false)); + } + if (literalTTLForScan != null) { + // Only views carry a literal TTL here; a base table relies on the CF-descriptor fallback. + scan.setAttribute(BaseScannerRegionObserverConstants.TTL, literalTTLForScan); + } + setInternalScanAttributesForPaging(conf, scan); + } + + /** + * Reproduces the client read path's server-paging setup for an internal scan. On the client the + * {@code SERVER_PAGE_SIZE_MS} attribute is set by + * {@code ScanUtil.setScanAttributeForPaging(Scan, PhoenixConnection)} and the scan filter is + * later wrapped in a {@link PagingFilter} by {@code BaseScannerRegionObserver.preScannerOpen}. + * Internal scans opened directly via {@code region.getScanner(scan)} bypass both, so this method + * performs both steps up-front. The region-server {@link Configuration} is the source of the + * paging props here, standing in for the client's {@code PhoenixConnection} props. + *

+ * Ordering matters: {@code PagingRegionScanner}'s constructor reads the {@link PagingFilter} and + * the page size off the scan, so this must run before + * {@link #openRegionScanner(RegionCoprocessorEnvironment, Region, Scan)} builds the scanner. + */ + public static void setInternalScanAttributesForPaging(Configuration conf, Scan scan) { + if ( + !conf.getBoolean(QueryServices.PHOENIX_SERVER_PAGING_ENABLED_ATTRIB, + QueryServicesOptions.DEFAULT_PHOENIX_SERVER_PAGING_ENABLED) + ) { + return; + } + long pageSizeMs = conf.getInt(QueryServices.PHOENIX_SERVER_PAGE_SIZE_MS, -1); + if (pageSizeMs == -1) { + // Use half of the HBase RPC timeout value as the server page size, mirroring the client + // ScanUtil.setScanAttributeForPaging fallback. + pageSizeMs = + (long) (conf.getLong(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT) + * 0.5); + } + scan.setAttribute(BaseScannerRegionObserverConstants.SERVER_PAGE_SIZE_MS, + Bytes.toBytes(Long.valueOf(pageSizeMs))); + // Wrap the scan filter in a PagingFilter as the top-level filter, matching + // BaseScannerRegionObserver.preScannerOpen. PagingRegionScanner then detects when PagingFilter + // has paged the scan out and returns a dummy result; readDataTableRows skips those dummies. + if (!(scan.getFilter() instanceof PagingFilter)) { + scan.setFilter(new PagingFilter(scan.getFilter(), ScanUtil.getPageSizeMsForFilter(scan))); + } + } + + /** + * Opens a region scanner wrapped exactly as {@code BaseScannerRegionObserver.postScannerOpen} + * wraps a client scan, so TTL masking is applied. This is always safe: + * {@link TTLRegionScanner#isMaskingEnabled} no-ops the masking when Phoenix compaction is + * disabled, the empty-column attributes are absent, the TTL is FOREVER, or the scan is non-strict + * — so wrapping a non-TTL scan changes no behavior. + */ + public static RegionScanner openRegionScanner(RegionCoprocessorEnvironment env, Region region, + Scan scan) throws IOException { + return new TTLRegionScanner(env, scan, + new PagingRegionScanner(region, region.getScanner(scan), scan)); + } +} diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java index 9763388effb..f11b79f455c 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java @@ -85,6 +85,7 @@ import org.apache.htrace.TraceScope; import org.apache.phoenix.compile.ScanRanges; import org.apache.phoenix.coprocessor.DelegateRegionCoprocessorEnvironment; +import org.apache.phoenix.coprocessor.ServerScanUtil; import org.apache.phoenix.coprocessor.generated.IndexMutationsProtos; import org.apache.phoenix.coprocessor.generated.PTableProtos; import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants; @@ -119,6 +120,8 @@ import org.apache.phoenix.query.QueryConstants; import org.apache.phoenix.query.QueryServicesOptions; import org.apache.phoenix.schema.CompiledConditionalTTLExpression; +import org.apache.phoenix.schema.CompiledTTLExpression; +import org.apache.phoenix.schema.LiteralTTLExpression; import org.apache.phoenix.schema.PColumn; import org.apache.phoenix.schema.PRow; import org.apache.phoenix.schema.PTable; @@ -140,6 +143,7 @@ import org.apache.phoenix.util.IndexUtil; import org.apache.phoenix.util.MutationUtil; import org.apache.phoenix.util.PhoenixKeyValueUtil; +import org.apache.phoenix.util.ScanUtil; import org.apache.phoenix.util.SchemaUtil; import org.apache.phoenix.util.ServerIndexUtil; import org.apache.phoenix.util.ServerUtil.ConnectionType; @@ -380,6 +384,18 @@ public static class BatchMutateContext { private boolean returnOldRow; private boolean hasConditionalTTL; // table has Conditional TTL private boolean immutableRows; + // For a VIEW carrying a view-level literal TTL, the serialized compiled literal TTL threaded + // by the client via the _TTL mutation attribute. Captured (and the attribute removed) early so + // the internal current-row scan masks with the view's TTL. Null for base tables (their literal + // TTL is on the CF descriptor) and for conditional TTL (left on the mutation for its own path). + private byte[] literalTTLForInternalScan; + // The empty-column CF/CQ threaded by the client (ScanUtil.annotateMutationWithLiteralTTL) for + // any table/view with an effective literal TTL — the single source that lets the internal + // current-row scan mask, for the secondary-index and no-index (atomic / ON DUPLICATE KEY / + // returnResult / row-delete) paths alike. Null when no literal TTL applies. Unlike the _TTL + // attribute these are left on the mutations (they are inert on the write path). + private byte[] emptyCFForInternalScan; + private byte[] emptyCQForInternalScan; public BatchMutateContext() { this.clientVersion = 0; @@ -1161,9 +1177,36 @@ private boolean isPartialUncoveredIndexMutation(PhoenixIndexMetaData indexMetaDa /** * Retrieve the data row state either from memory or disk. The rows are locked by the caller. + *

+ * The disk read is opened through {@link ServerScanUtil} so it is TTL-masked exactly like a + * client read: the internal scan otherwise bypasses the {@code postScannerOpen} coprocessor hook + * (the only place a scan is wrapped in {@code TTLRegionScanner}) and would rebuild the index from + * logically-expired-but-physically-present cells. The empty-column CF/CQ that enable masking are + * the bytes the client threaded on the mutation + * ({@link org.apache.phoenix.util.ScanUtil#annotateMutationWithLiteralTTL}) and captured into the + * batch context — the single source for every path that reaches here, index or no-index alike; + * {@code literalTTLForScan} carries a view's literal TTL (null for base tables, which use the + * CF-descriptor fallback), and {@code isStrictTTL} avoids over-masking a non-strict table. + * Masking is a strict no-op when Phoenix compaction is disabled. The internal scan is also given + * the client read path's server-paging setup (a {@code SERVER_PAGE_SIZE_MS} attribute and a + * {@code PagingFilter} wrap), so it is bounded by the server page budget just like a client scan; + * {@code readDataTableRows} skips the dummy results that paging can emit. + *

+ * The scan's time range is set to {@code [0, batchTimestamp)} so {@link TTLRegionScanner} anchors + * its masking clock ({@code currentTime = scan.getTimeRange().getMax()}) at + * {@code batchTimestamp} — the exact timestamp at which the index side is rebuilt — rather than + * at the scan-open wall clock. On a partial "touch" upsert, {@code getBatchTimestamp} bumps + * {@code batchTimestamp} slightly past that wall clock to avoid timestamp collisions, so a cell + * whose TTL boundary falls in that sub-millisecond window would otherwise be masked here (as of + * the scan-open instant) yet be visible to the index build (as of {@code batchTimestamp}), + * leaving the data row and index row inconsistent. Anchoring both at {@code batchTimestamp} makes + * the read trim expired cells as of exactly the timestamp the index is built at, so the two never + * disagree. The half-open range drops nothing current: every pre-existing cell of a locked row + * was written by an earlier batch at {@code ts < batchTimestamp} under the Phoenix write path. */ private void getCurrentRowStates(ObserverContext c, - BatchMutateContext context) throws IOException { + BatchMutateContext context, byte[] literalTTLForScan, boolean isStrictTTL, long batchTimestamp) + throws IOException { Set keys = new HashSet(context.rowsToLock.size()); for (ImmutableBytesPtr rowKeyPtr : context.rowsToLock) { PendingRow pendingRow = new PendingRow(rowKeyPtr, context); @@ -1212,6 +1255,18 @@ private void getCurrentRowStates(ObserverContext c return; } + // The empty-column CF/CQ that let TTLRegionScanner mask this internal scan are the bytes the + // client threaded on the mutation (ScanUtil.annotateMutationWithLiteralTTL) and captured into + // the batch context. This is the single source for every path that reaches here — the + // secondary-index case and the no-index atomic / ON DUPLICATE KEY / returnResult / row-delete + // case alike — and it ties internal-scan masking to the same client signal that governs client + // read masking (setScanAttributesForClient), so the two stay consistent. When they are absent + // (a mutation not annotated by the client) the scan is left unmasked, which is safe: such a + // client's own reads are likewise unmasked, and TTLRegionScanner masking is a no-op anyway on a + // non-TTL table (CF-descriptor FOREVER) or when Phoenix compaction is disabled. + byte[] emptyCF = context.emptyCFForInternalScan; + byte[] emptyCQ = context.emptyCQForInternalScan; + if (this.useBloomFilter) { for (KeyRange key : keys) { // Scan.java usage alters scan instances, safer to create scan instance per usage @@ -1220,6 +1275,12 @@ private void getCurrentRowStates(ObserverContext c // for bloom filters scan should be a get scan.withStartRow(key.getLowerRange(), true); scan.withStopRow(key.getLowerRange(), true); + // Anchor TTLRegionScanner's masking clock at batchTimestamp (see method comment): the + // half-open [0, batchTimestamp) range makes getTimeRange().getMax() == batchTimestamp, so + // the read trims expired cells as of exactly the timestamp the index is built at. + scan.setTimeRange(0, batchTimestamp); + ServerScanUtil.setInternalScanAttributes(c.getEnvironment().getConfiguration(), scan, + emptyCF, emptyCQ, literalTTLForScan, isStrictTTL); readDataTableRows(c, context, scan); } } else { @@ -1228,13 +1289,23 @@ private void getCurrentRowStates(ObserverContext c scanRanges.initializeScan(scan); SkipScanFilter skipScanFilter = scanRanges.getSkipScanFilter(); scan.setFilter(skipScanFilter); + // Anchor TTLRegionScanner's masking clock at batchTimestamp (see method comment): the + // half-open [0, batchTimestamp) range makes getTimeRange().getMax() == batchTimestamp, so + // the read trims expired cells as of exactly the timestamp the index is built at. + scan.setTimeRange(0, batchTimestamp); + ServerScanUtil.setInternalScanAttributes(c.getEnvironment().getConfiguration(), scan, emptyCF, + emptyCQ, literalTTLForScan, isStrictTTL); readDataTableRows(c, context, scan); } } private void readDataTableRows(ObserverContext c, BatchMutateContext context, Scan scan) throws IOException { - try (RegionScanner scanner = c.getEnvironment().getRegion().getScanner(scan)) { + // Open through ServerScanUtil so the scan is wrapped in TTLRegionScanner (mirroring + // postScannerOpen) and masks TTL-expired rows exactly like a client read. Masking no-ops when + // Phoenix compaction is off or the empty-column attributes are absent. + try (RegionScanner scanner = + ServerScanUtil.openRegionScanner(c.getEnvironment(), c.getEnvironment().getRegion(), scan)) { boolean more = true; while (more) { List cells = new ArrayList(); @@ -1242,6 +1313,13 @@ private void readDataTableRows(ObserverContext c, if (cells.isEmpty()) { continue; } + // With server paging wired in (ServerScanUtil.setInternalScanAttributes), + // PagingRegionScanner + // returns a dummy result when a page is paged out; skip it and let the loop resume rather + // than build a Put from the dummy cell. + if (ScanUtil.isDummy(cells)) { + continue; + } byte[] rowKey = CellUtil.cloneRow(cells.get(0)); Put put = new Put(rowKey); for (Cell cell : cells) { @@ -1664,6 +1742,61 @@ private static void identifyIndexMaintainerTypes(PhoenixIndexMetaData indexMetaD } } + /** + * Captures the literal-TTL internal-scan attributes the client threads + * ({@link org.apache.phoenix.util.ScanUtil#annotateMutationWithLiteralTTL}) off the + * representative mutation so the internal current-row scan can mask exactly like a client read. + *

+ * Two things are captured: + *

    + *
  • The empty-column CF/CQ — threaded whenever an effective (non-NONE) literal TTL + * applies, for both base tables and views. They are the single source that lets + * {@code setInternalScanAttributes} mask, for the secondary-index and no-index (atomic / ON + * DUPLICATE KEY / {@code returnResult} / row-delete) paths alike. They are inert on the write + * path (nothing reads a mutation's empty-column attribute), so they are left on the mutations, + * not removed.
    • + *
  • The view's literal {@code _TTL} — the client threads it via the same {@code _TTL} + * mutation attribute the server otherwise treats as conditional TTL + * ({@code PhoenixIndexBuilder.hasConditionalTTL}). We make that attribute polymorphic: if the + * representative mutation's {@code _TTL} deserializes to a {@link LiteralTTLExpression} (a view's + * literal TTL), capture its bytes and remove {@code _TTL} from every mutation; if it is + * conditional (or absent), leave it untouched so the conditional-TTL path runs normally.
    • + *
+ * The {@code _TTL} removal must happen before {@link #identifyMutationTypes} so that + * {@code hasConditionalTTL(m)} returns false at every consumer (it sets + * {@code context.hasConditionalTTL}, which would otherwise reach the blind cast to + * {@code CompiledConditionalTTLExpression} in {@link #updateMutationsForConditionalTTL}). A table + * has exactly one TTL kind and a batch never mixes views, so the decision is uniform across the + * batch. HBase has no removeAttribute, so removal is {@code setAttribute(TTL, null)}. + */ + private void extractLiteralTTLForInternalScan(MiniBatchOperationInProgress miniBatchOp, + BatchMutateContext context) throws IOException { + if (miniBatchOp.size() == 0) { + return; + } + Mutation representative = miniBatchOp.getOperation(0); + // Empty-column CF/CQ are threaded independently of _TTL (a base finite-TTL table threads them + // but no _TTL), so capture them regardless of the _TTL attribute below. Left on the mutations. + context.emptyCFForInternalScan = + representative.getAttribute(BaseScannerRegionObserverConstants.EMPTY_COLUMN_FAMILY_NAME); + context.emptyCQForInternalScan = + representative.getAttribute(BaseScannerRegionObserverConstants.EMPTY_COLUMN_QUALIFIER_NAME); + + byte[] ttlBytes = representative.getAttribute(BaseScannerRegionObserverConstants.TTL); + if (ttlBytes == null) { + return; + } + CompiledTTLExpression ttlExpr = TTLExpressionFactory.create(ttlBytes); + if (!(ttlExpr instanceof LiteralTTLExpression)) { + // Conditional TTL: leave the attribute in place for updateMutationsForConditionalTTL. + return; + } + context.literalTTLForInternalScan = ttlBytes; + for (int i = 0; i < miniBatchOp.size(); i++) { + miniBatchOp.getOperation(i).setAttribute(BaseScannerRegionObserverConstants.TTL, null); + } + } + private void identifyMutationTypes(MiniBatchOperationInProgress miniBatchOp, BatchMutateContext context) throws IOException { for (int i = 0; i < miniBatchOp.size(); i++) { @@ -1781,13 +1914,19 @@ private long getBatchTimestamp(BatchMutateContext context, TableName table) // The timestamp for this batch will be different from the last batch processed. lastTimestamp = ts; batchesWithLastTimestamp.clear(); - batchesWithLastTimestamp.add(context.rowsToLock); + // Register a snapshot, not the live rowsToLock: this now runs before + // releaseLocksForOnDupIgnoreMutations may structurally mutate rowsToLock (remove) outside + // synchronized(this), which would corrupt a concurrent shouldSleep iteration under + // synchronized(this). The snapshot is a superset of every later state (rowsToLock only + // shrinks after lockRows), so shouldSleep stays conservative: at most an extra sleep, never + // a missed one. + batchesWithLastTimestamp.add(new TreeSet<>(context.rowsToLock)); return ts; } else { if (!shouldSleep(context)) { // There is no need to sleep as the last batches with the same timestamp // do not have a common row this batch - batchesWithLastTimestamp.add(context.rowsToLock); + batchesWithLastTimestamp.add(new TreeSet<>(context.rowsToLock)); return ts; } } @@ -1806,7 +1945,7 @@ private long getBatchTimestamp(BatchMutateContext context, TableName table) // We do not have to check again if we need to sleep again since we got the next // timestamp while holding the row locks. This mean there cannot be a new // mutation with the same row attempting get the same timestamp - batchesWithLastTimestamp.add(context.rowsToLock); + batchesWithLastTimestamp.add(new TreeSet<>(context.rowsToLock)); return ts; } } @@ -1817,6 +1956,11 @@ public void preBatchMutateWithExceptions(ObserverContext generateOnDupMutations(BatchMutateContext context, Put at delete.add(cell); } } - if (!put.isEmpty() || !delete.isEmpty()) { PTable table = operations.get(0).getFirst(); addEmptyKVCellToPut(put, tuple, table); diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetaDataEndPointIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetaDataEndPointIT.java index 66fdc1540bb..dfd21bf91c7 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetaDataEndPointIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetaDataEndPointIT.java @@ -44,9 +44,9 @@ * {@link MetricsMetadataSourceFactory#getMetadataMetricsSource()} singleton, whose counters * (CREATE_TABLE_COUNT, METADATA_CACHE_ESTIMATED_USED_SIZE, METADATA_CACHE_ADD_COUNT, ...) are * incremented by every server-side metadata operation in the same JVM. Running this test class in - * the {@link ParallelStatsDisabledTest} group lets parallel tests in other classes in the same - * fork mutate those counters between this test's "capture baseline" and "verify after CREATE - * TABLE" calls, producing intermittent strict-equality assertion failures such as: + * the {@link ParallelStatsDisabledTest} group lets parallel tests in other classes in the same fork + * mutate those counters between this test's "capture baseline" and "verify after CREATE TABLE" + * calls, producing intermittent strict-equality assertion failures such as: * * 
  *   testMetadataMetricsOfCreateTable
@@ -54,8 +54,8 @@
  *
* * Categorize as {@link NeedsOwnMiniClusterTest} so failsafe runs this class in its own forked JVM - * (reuseForks=false), guaranteeing exclusive ownership of the metric singleton for the duration - * of the suite. Tests within this class still run sequentially in that fork, which is sufficient + * (reuseForks=false), guaranteeing exclusive ownership of the metric singleton for the duration of + * the suite. Tests within this class still run sequentially in that fork, which is sufficient * because each individual test captures and verifies its own counter deltas in a single thread. */ @Category(NeedsOwnMiniClusterTest.class) diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataGetTableReadLockIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataGetTableReadLockIT.java index f2344fc8c93..58c4b8fbc86 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataGetTableReadLockIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/MetadataGetTableReadLockIT.java @@ -97,7 +97,8 @@ public void testBlockedReadDoesNotBlockAnotherRead() throws Exception { } } - private static void waitForCoprocessorOnSystemCatalog(Class coprocessorClass) throws Exception { + private static void waitForCoprocessorOnSystemCatalog(Class coprocessorClass) + throws Exception { final TableName sysCatalog = TableName.valueOf("SYSTEM.CATALOG"); utility.waitFor(10000, 100, () -> { List regions = utility.getHBaseCluster().getRegions(sysCatalog); @@ -105,8 +106,7 @@ private static void waitForCoprocessorOnSystemCatalog(Class coprocessorClass) return false; } for (HRegion region : regions) { - if (region.getCoprocessorHost() - .findCoprocessor(coprocessorClass.getName()) == null) { + if (region.getCoprocessorHost().findCoprocessor(coprocessorClass.getName()) == null) { return false; } } diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexDataTableSyncIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexDataTableSyncIT.java new file mode 100644 index 00000000000..30298b24834 --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexDataTableSyncIT.java @@ -0,0 +1,796 @@ +/* + * 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.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.Arrays; +import java.util.Collection; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.Admin; +import org.apache.hadoop.hbase.client.Mutation; +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.MiniBatchOperationInProgress; +import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants; +import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest; +import org.apache.phoenix.jdbc.PhoenixConnection; +import org.apache.phoenix.query.BaseTest; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.util.EnvironmentEdgeManager; +import org.apache.phoenix.util.ManualEnvironmentEdge; +import org.apache.phoenix.util.ReadOnlyProps; +import org.apache.phoenix.util.TestUtil; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import org.apache.phoenix.thirdparty.com.google.common.collect.Maps; + +/** + * Verifies that internal server-side index-maintenance current-row reads honor Phoenix TTL exactly + * like a client read, and that the read is anchored at the mutation timestamp so the data table and + * a global index stay aligned under compaction. + *

+ * The scenario is the divergence being fixed: a covered column (covcol) is written once, then never + * re-written by later partial "touch" upserts. On the index side covcol is rebuilt at every touch's + * mutationTimestamp from the masked current-row read; on the data side covcol keeps its original + * timestamp. The fix anchors that internal read's masking clock at mutationTimestamp (the same + * instant the index is built at) via scan.setTimeRange(0, mutationTimestamp), so both sides always + * agree on whether covcol is still alive: whatever the read masks the index rebuild omits, and + * whatever the read keeps a later major compaction keeps on the data side too. The row and its + * index then expire as a unit rather than covcol being dropped on only one side. + * {@link #testNoIndexAtomicUpsertMasksExpiredRow} covers the no-index masking path: an atomic ON + * DUPLICATE KEY UPDATE on a TTL table with no secondary index still triggers an internal + * current-row read. That scan is masked using the empty-column CF/CQ the client threads on the + * mutation — the same single source used for the secondary-index case — so an expired row is + * treated as absent rather than resurrected. + */ +@Category(NeedsOwnMiniClusterTest.class) +@RunWith(Parameterized.class) +public class IndexDataTableSyncIT extends BaseTest { + static final int MAX_LOOKBACK_AGE = 10; + static final int TTL = 60; + static final String COVCOL_VALUE = "cov-original"; + static final String IDXCOL_VALUE = "idx-original"; + + private final boolean columnEncoded; + private ManualEnvironmentEdge injectEdge; + + public IndexDataTableSyncIT(boolean columnEncoded) { + this.columnEncoded = columnEncoded; + } + + @Parameterized.Parameters(name = "columnEncoded={0}") + public static synchronized Collection data() { + return Arrays.asList(new Object[][] { { false }, { true } }); + } + + @BeforeClass + public static synchronized void doSetup() throws Exception { + setUpTestDriver(new ReadOnlyProps(baseServerProps().entrySet().iterator())); + } + + /** + * Common server props for the TTL-sync IT: max-lookback and immediate global-index row aging so a + * major compaction deterministically exercises the covered-column timestamp-skew divergence. + */ + static Map baseServerProps() { + Map props = Maps.newHashMapWithExpectedSize(4); + props.put(QueryServices.GLOBAL_INDEX_ROW_AGE_THRESHOLD_TO_DELETE_MS_ATTRIB, Long.toString(0)); + props.put(BaseScannerRegionObserverConstants.PHOENIX_MAX_LOOKBACK_AGE_CONF_KEY, + Integer.toString(MAX_LOOKBACK_AGE)); + props.put("hbase.procedure.remote.dispatcher.delay.msec", "0"); + // The view case threads the view's literal TTL as the per-mutation _TTL attribute. + props.put(QueryServices.PHOENIX_VIEW_TTL_ENABLED, Boolean.toString(true)); + return props; + } + + @Before + public void beforeTest() { + EnvironmentEdgeManager.reset(); + injectEdge = new ManualEnvironmentEdge(); + injectEdge.setValue(EnvironmentEdgeManager.currentTimeMillis()); + } + + @After + public synchronized void afterTest() throws Exception { + boolean refCountLeaked = isAnyStoreRefCountLeaked(); + EnvironmentEdgeManager.reset(); + Assert.assertFalse("refCount leaked", refCountLeaked); + } + + /** + * Builds the comma-separated table-property clause. When ttlSeconds >= 0 a TTL is emitted as the + * first property; COLUMN_ENCODED_BYTES always follows, and IS_STRICT_TTL=false is appended for a + * non-strict table. + */ + private String withClause(int ttlSeconds, boolean strict) { + StringBuilder sb = new StringBuilder(" "); + if (ttlSeconds >= 0) { + sb.append("TTL=").append(ttlSeconds).append(", "); + } + sb.append("COLUMN_ENCODED_BYTES=").append(columnEncoded ? 2 : 0); + if (!strict) { + sb.append(", IS_STRICT_TTL=false"); + } + return sb.toString(); + } + + /** + * Base table with a literal TTL and a covered global index. After a partial touch that never + * writes covcol, the data-side covcol keeps its original timestamp; the masked internal read + * anchored at mutationTimestamp keeps the data and index agreeing on covcol while the row is + * alive, and a full expiry after that shows the data row and the index expiring as a unit. + */ + @Test + public void testBaseTableCoveredColumnResync() throws Exception { + String tableName = generateUniqueName(); + String indexName = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + conn.createStatement() + .execute("CREATE TABLE " + tableName + " (id VARCHAR NOT NULL PRIMARY KEY, " + + "idxcol VARCHAR, covcol VARCHAR, touchcol VARCHAR)" + withClause(TTL, true)); + conn.createStatement() + .execute("CREATE INDEX " + indexName + " ON " + tableName + " (idxcol) INCLUDE (covcol)"); + conn.commit(); + + EnvironmentEdgeManager.injectEdge(injectEdge); + + // Full initial upsert: covcol is written exactly once here and never again. + conn.createStatement().execute("UPSERT INTO " + tableName + + " (id, idxcol, covcol, touchcol) VALUES ('r1', 'k1', '" + COVCOL_VALUE + "', 'x0')"); + conn.commit(); + + // Partial touch that does not write covcol, well within the TTL so the row is alive. + injectEdge.incrementValue(1000); + conn.createStatement() + .execute("UPSERT INTO " + tableName + " (id, touchcol) VALUES ('r1', 'x1')"); + conn.commit(); + assertTouchColConsistent(conn, tableName, indexName, "r1", "x1"); + // Data and index agree on covcol + assertCovColConsistent(conn, tableName, indexName, "r1", "k1", COVCOL_VALUE); + + // Advancing past TTL expires the row in data table for internal current row read. + injectEdge.incrementValue((TTL + 1) * 1000L); + conn.createStatement() + .execute("UPSERT INTO " + tableName + " (id, touchcol) VALUES ('r1', 'x2')"); + conn.commit(); + assertTouchColConsistent(conn, tableName, indexName, "r1", "x2"); + // Data and index agree on covcol + assertCovColAbsent(conn, tableName, indexName, "r1"); + } + } + + /** + * Uncovered global index on a base table with a literal TTL. An uncovered index stores its + * indexed column (idxcol) in the index row key, rebuilt at every touch's mutationTimestamp, while + * a partial touch that omits idxcol would otherwise leave the data-side idxcol cell at its + * original timestamp. + */ + @Test + public void testUncoveredIndexIndexedColumnResync() throws Exception { + String tableName = generateUniqueName(); + String indexName = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + conn.createStatement() + .execute("CREATE TABLE " + tableName + " (id VARCHAR NOT NULL PRIMARY KEY, " + + "idxcol VARCHAR, touchcol VARCHAR)" + withClause(TTL, true)); + conn.createStatement() + .execute("CREATE UNCOVERED INDEX " + indexName + " ON " + tableName + " (idxcol)"); + conn.commit(); + + EnvironmentEdgeManager.injectEdge(injectEdge); + + // Full initial upsert: idxcol is written exactly once here and never again. + conn.createStatement().execute("UPSERT INTO " + tableName + + " (id, idxcol, touchcol) VALUES ('r1', '" + IDXCOL_VALUE + "', 'x0')"); + conn.commit(); + + // Partial touch that does not write idxcol, well within the TTL so the row is alive and the + // internal scan anchored at batchTimestamp still returns idxcol for the index rebuild. + injectEdge.incrementValue(1000); + conn.createStatement() + .execute("UPSERT INTO " + tableName + " (id, touchcol) VALUES ('r1', 'x1')"); + conn.commit(); + + // An uncovered index carries no data columns, so touchcol is only readable data-side; the + // index side agrees on idxcol below. The row is alive well within the TTL. + assertDataTouchColAlive(conn, tableName, "r1", "x1"); + assertUncoveredIdxColConsistent(conn, tableName, indexName, "r1", IDXCOL_VALUE); + + // Advancing past TTL expires the row in data table for internal current row read. + injectEdge.incrementValue((TTL + 1) * 1000L); + conn.createStatement() + .execute("UPSERT INTO " + tableName + " (id, touchcol) VALUES ('r1', 'x2')"); + conn.commit(); + // Data side: row still alive by PK (touchcol keeps it); the masked idxcol@T0 is trimmed, so + // the uncovered index no longer carries a usable entry keyed by idxcol (asserted below). + assertDataTouchColAlive(conn, tableName, "r1", "x2"); + assertUncoveredIdxColAbsent(conn, tableName, indexName, IDXCOL_VALUE); + } + } + + /** + * View with a view-level literal TTL (not on the shared CF descriptor) and a covered index on the + * view. Exercises the literal-_TTL per-mutation threading: the internal scan masks with the + * view's TTL, anchored at mutationTimestamp, and covcol keeps its original data-side timestamp + * just like the base-table case while the data and index reads stay consistent. A full expiry + * after that shows the view's data row and its index expiring as a unit. + */ + @Test + public void testViewCoveredColumnResync() throws Exception { + String baseTableName = generateUniqueName(); + String viewName = generateUniqueName(); + String indexName = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + // Base table carries NO TTL; the TTL lives only on the view. + conn.createStatement() + .execute("CREATE TABLE " + baseTableName + " (id VARCHAR NOT NULL PRIMARY KEY, " + + "idxcol VARCHAR, covcol VARCHAR, touchcol VARCHAR)" + withClause(-1, true)); + conn.createStatement() + .execute("CREATE VIEW " + viewName + " AS SELECT * FROM " + baseTableName + " TTL=" + TTL); + conn.createStatement() + .execute("CREATE INDEX " + indexName + " ON " + viewName + " (idxcol) INCLUDE (covcol)"); + conn.commit(); + + EnvironmentEdgeManager.injectEdge(injectEdge); + + conn.createStatement().execute("UPSERT INTO " + viewName + + " (id, idxcol, covcol, touchcol) VALUES ('r1', 'k1', '" + COVCOL_VALUE + "', 'x0')"); + conn.commit(); + + injectEdge.incrementValue(1000); + conn.createStatement() + .execute("UPSERT INTO " + viewName + " (id, touchcol) VALUES ('r1', 'x1')"); + conn.commit(); + assertTouchColConsistent(conn, viewName, indexName, "r1", "x1"); + // View and index agree on covcol + assertCovColConsistent(conn, viewName, indexName, "r1", "k1", COVCOL_VALUE); + + // Advancing past the view TTL expires the row in data table for internal current row read. + injectEdge.incrementValue((TTL + 1) * 1000L); + conn.createStatement() + .execute("UPSERT INTO " + viewName + " (id, touchcol) VALUES ('r1', 'x2')"); + conn.commit(); + assertTouchColConsistent(conn, viewName, indexName, "r1", "x2"); + assertCovColAbsent(conn, viewName, indexName, "r1"); + } + } + + /** + * A table with IS_STRICT_TTL=false must not be masked away: the internal scan anchored at + * mutationTimestamp must not mask covcol away. + */ + @Test + public void testNonStrictTableNotOverMasked() throws Exception { + String tableName = generateUniqueName(); + String indexName = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + conn.createStatement() + .execute("CREATE TABLE " + tableName + " (id VARCHAR NOT NULL PRIMARY KEY, " + + "idxcol VARCHAR, covcol VARCHAR, touchcol VARCHAR)" + withClause(TTL, false)); + conn.createStatement() + .execute("CREATE INDEX " + indexName + " ON " + tableName + " (idxcol) INCLUDE (covcol)"); + conn.commit(); + + EnvironmentEdgeManager.injectEdge(injectEdge); + + conn.createStatement().execute("UPSERT INTO " + tableName + + " (id, idxcol, covcol, touchcol) VALUES ('r1', 'k1', '" + COVCOL_VALUE + "', 'x0')"); + conn.commit(); + + // Advance beyond TTL, then touch. A strict table would mask covcol at the internal scan; + // a non-strict table must not. + injectEdge.incrementValue((TTL + 5) * 1000L); + conn.createStatement() + .execute("UPSERT INTO " + tableName + " (id, touchcol) VALUES ('r1', 'x1')"); + conn.commit(); + assertTouchColConsistent(conn, tableName, indexName, "r1", "x1"); + // The decisive check: a non-strict internal read is not masked away, so the index rebuilds + // with covcol and both the data read and the index read still return it. + assertCovColConsistent(conn, tableName, indexName, "r1", "k1", COVCOL_VALUE); + + injectEdge.incrementValue(MAX_LOOKBACK_AGE * 1000L); + flushAndMajorCompact(conn, tableName); + flushAndMajorCompact(conn, indexName); + + // Data side: the row is still alive (touchcol keeps it) but the stale covcol was physically + // collected by compaction, so a forced data-table read returns covcol = NULL. + try (ResultSet rs = conn.createStatement() + .executeQuery("SELECT /*+ NO_INDEX */ covcol FROM " + tableName + " WHERE id = 'r1'")) { + assertTrue("data-table row should still be alive (touchcol keeps it)", rs.next()); + assertNull("stale data-side covcol should be collected by compaction", rs.getString(1)); + assertFalse("exactly one data-table row", rs.next()); + } + + // Index side: the covered index copy survives compaction, so the index read still returns it. + try (ResultSet rs = conn.createStatement() + .executeQuery("SELECT covcol FROM " + tableName + " WHERE idxcol = 'k1'")) { + assertTrue("index-visible row should still exist", rs.next()); + assertEquals("index covcol should survive compaction", COVCOL_VALUE, rs.getString(1)); + assertFalse("exactly one index-visible row", rs.next()); + } + } + } + + /** + * The no-index masking path: an atomic ON DUPLICATE KEY UPDATE on a base table with a literal TTL + * and NO secondary index. + */ + @Test + public void testNoIndexAtomicUpsertMasksExpiredRow() throws Exception { + String tableName = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + conn.createStatement().execute("CREATE TABLE " + tableName + + " (id VARCHAR NOT NULL PRIMARY KEY, counter INTEGER)" + withClause(TTL, true)); + conn.commit(); + + EnvironmentEdgeManager.injectEdge(injectEdge); + + // First atomic upsert: the row does not exist, so the ON DUPLICATE clause is skipped and the + // UPSERT VALUES insert counter = 0. + conn.createStatement().execute("UPSERT INTO " + tableName + + " (id, counter) VALUES ('r1', 0) ON DUPLICATE KEY UPDATE counter = counter + 1"); + conn.commit(); + assertEquals("fresh insert of a new row", 0, atomicCounter(conn, tableName)); + + // Advance past the TTL. No flush/compact: the row's cells are still physically present, so + // only read masking - not compaction - can hide the now-expired row from the internal scan. + injectEdge.incrementValue((TTL + 1) * 1000L); + + // Second atomic upsert on the now-expired row. Post-fix the masked internal scan returns no + // current row, so this inserts counter = 0 again; pre-fix the unmasked scan resurrects the + // row and the ON DUPLICATE clause increments counter to 1. + conn.createStatement().execute("UPSERT INTO " + tableName + + " (id, counter) VALUES ('r1', 0) ON DUPLICATE KEY UPDATE counter = counter + 1"); + conn.commit(); + + assertEquals("expired row must be treated as absent, not resurrected and incremented", 0, + atomicCounter(conn, tableName)); + } + } + + @Test + public void testUpsertSelectTouchNearTtlBoundaryKeepsDataAndIndexConsistent() throws Exception { + runTtlBoundaryScenario(true); + } + + @Test + public void testUpsertValuesTouchNearTtlBoundaryKeepsDataAndIndexConsistent() throws Exception { + runTtlBoundaryScenario(false); + } + + @Test + public void testImmutableDifferentStorageSchemeIndexKeepsDataAndIndexConsistent() + throws Exception { + String tableName = generateUniqueName(); + String indexName = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + conn.setAutoCommit(false); + // The scheme mismatch (independent of the class columnEncoded parameter, which this test does + // not use) is the point: it forces server-side index maintenance and the masked current-row + // read. Direction ONE_CELL data -> SINGLE_CELL index is the only allowed one. + conn.createStatement().execute("CREATE IMMUTABLE TABLE " + tableName + + " (id VARCHAR NOT NULL PRIMARY KEY, idxcol VARCHAR, covcol VARCHAR, touchcol VARCHAR) " + + "TTL=" + TTL + ", IMMUTABLE_STORAGE_SCHEME=ONE_CELL_PER_COLUMN, COLUMN_ENCODED_BYTES=0"); + conn.createStatement() + .execute("CREATE INDEX " + indexName + " ON " + tableName + " (idxcol) INCLUDE (covcol) " + + "IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + conn.commit(); + + // Confirm the two sides really do use different storage schemes. + assertMetadata(conn, PTable.ImmutableStorageScheme.ONE_CELL_PER_COLUMN, + PTable.QualifierEncodingScheme.NON_ENCODED_QUALIFIERS, tableName); + assertMetadata(conn, PTable.ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS, + PTable.QualifierEncodingScheme.TWO_BYTE_QUALIFIERS, indexName); + + // Freeze the clock at t0; every cell of the full initial row lands at t0. + long t0 = EnvironmentEdgeManager.currentTimeMillis(); + injectEdge.setValue(t0); + EnvironmentEdgeManager.injectEdge(injectEdge); + + conn.createStatement().execute("UPSERT INTO " + tableName + + " (id, idxcol, covcol, touchcol) VALUES ('r1', 'k1', '" + COVCOL_VALUE + "', 'x0')"); + conn.commit(); + + // Push the t0 cells onto an HFile so the later major compaction merges two files. + flushTable(conn, tableName); + flushTable(conn, indexName); + + // Phase 1: a partial touch that omits covcol, committed WITHIN the TTL so the row is alive. + // The masked current-row read still returns covcol@t0, so the index is rebuilt WITH covcol + // and + // both sides agree covcol is present (non-null). + injectEdge.incrementValue(1000); + conn.createStatement() + .executeUpdate("UPSERT INTO " + tableName + " (id, touchcol) VALUES ('r1', 'x1')"); + conn.commit(); + assertCovColConsistent(conn, tableName, indexName, "r1", "k1", COVCOL_VALUE); + + // Phase 2: a second partial touch that also omits covcol. It is READ while the row is still + // alive (one tick before the boundary) but COMMITTED past a > TTL gap, so its + // mutationTimestamp is well past covcol@t0's expiry. The masked read now drops covcol@t0, so + // the index is rebuilt WITHOUT covcol. + injectEdge.incrementValue(TTL * 1000L - 1); + conn.createStatement() + .executeUpdate("UPSERT INTO " + tableName + " (id, touchcol) VALUES ('r1', 'x2')"); + injectEdge.incrementValue(2); + conn.commit(); + + assertDataTouchColAlive(conn, tableName, "r1", "x2"); + // Past the TTL boundary covcol reads NULL on both the data table and the index - agreement + // AND + // the specific expired value: the index was rebuilt without the stale covcol. + assertCovColAbsent(conn, tableName, indexName, "r1"); + } + } + + /** + * A major compaction that races an in-flight partial touch. This reproduces the narrow window the + * design notes call out: {@code preBatchMutateWithExceptions} releases the data-table row locks + * before the (potentially slow) first-phase index write and re-acquires them afterwards + * (IndexRegionObserver.java unlockRows before doPre, lockRows after), so a concurrent major + * compaction on the data table can run between the masked current-row read and the moment the + * touch's data-side cells are persisted. + *

+ * The interleaving is made deterministic by a {@link BlockingIndexWriteObserver} installed on the + * index table: the touch's first-phase index write (doPre -> table.batch to the index) blocks in + * the index region's preBatchMutate, which is strictly after the data-table current-row read and + * strictly before doPre completes. While it is parked we fire the major compaction on the data + * table, then release the index write. + *

+ * When the current row is read the clock is well within the TTL, so the masked read returns + * covcol@t0 and the index is rebuilt with covcol. When the compaction runs the clock has advanced + * past the TTL but not past TTL + max-lookback, so covcol@t0 is older than the row's freshest + * cell by more than the TTL yet is retained physically: CompactionScanner keeps the last row + * version unless {@code compactionTime - maxTimestamp > maxLookbackInMillis + ttl} + * (CompactionScanner.retainCellsOfLastRowVersion), which is false here (65s <= 60s + 10s). The + * non-zero max-lookback is exactly what makes the race harmless: the pre-existing covcol@t0 + * cannot be collected out from under the index rebuild, so the data table and the index still + * agree covcol is present after the touch commits. + */ + @Test + public void testConcurrentMajorCompactionDuringIndexWriteKeepsDataAndIndexConsistent() + throws Exception { + String tableName = generateUniqueName(); + String indexName = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + conn.setAutoCommit(false); + conn.createStatement() + .execute("CREATE TABLE " + tableName + " (id VARCHAR NOT NULL PRIMARY KEY, " + + "idxcol VARCHAR, covcol VARCHAR, touchcol VARCHAR)" + withClause(TTL, true)); + conn.createStatement() + .execute("CREATE INDEX " + indexName + " ON " + tableName + " (idxcol) INCLUDE (covcol)"); + conn.commit(); + + // Freeze the clock at t0; every cell of the full initial row lands at t0. + long t0 = EnvironmentEdgeManager.currentTimeMillis(); + injectEdge.setValue(t0); + EnvironmentEdgeManager.injectEdge(injectEdge); + + // Full initial row: covcol is written once here and never again. Flush so covcol@t0 is on an + // HFile and the later major compaction has a file to rewrite. + conn.createStatement().execute("UPSERT INTO " + tableName + + " (id, idxcol, covcol, touchcol) VALUES ('r1', 'k1', '" + COVCOL_VALUE + "', 'x0')"); + conn.commit(); + flushTable(conn, tableName); + flushTable(conn, indexName); + + // Arm the index-table observer so the NEXT index write (the touch's doPre) parks mid-flight, + // then advance to a time still comfortably within the TTL: the touch's current-row read sees + // covcol@t0 as alive and rebuilds the index with covcol. + CountDownLatch indexWriteReached = new CountDownLatch(1); + CountDownLatch indexWriteProceed = new CountDownLatch(1); + BlockingIndexWriteObserver.arm(indexName, indexWriteReached, indexWriteProceed); + TestUtil.addCoprocessor(conn, indexName, BlockingIndexWriteObserver.class); + injectEdge.setValue(t0 + 30_000L); + + // Run the touch on its own thread/connection: it parks inside the index region's + // preBatchMutate (i.e. inside doPre) after the data-table read and after the data locks were + // released, letting us drive the compaction while it is parked. + AtomicReference touchError = new AtomicReference<>(); + Thread toucher = new Thread(() -> { + try (Connection tc = DriverManager.getConnection(getUrl())) { + tc.setAutoCommit(false); + tc.createStatement() + .executeUpdate("UPSERT INTO " + tableName + " (id, touchcol) VALUES ('r1', 'x1')"); + tc.commit(); + } catch (Exception e) { + touchError.set(e); + } + }, "ttl-sync-toucher"); + toucher.start(); + + // Wait until the touch's index write has parked (read done, doPre not yet complete). + assertTrue("index write did not reach the observer in time", + indexWriteReached.await(60, TimeUnit.SECONDS)); + + // The row is now past the TTL (65s) but not past TTL + max-lookback (70s), so the concurrent + // major compaction on the data table must NOT physically collect covcol@t0. + injectEdge.setValue(t0 + 65_000L); + TestUtil.majorCompact(getUtility(), TableName.valueOf(tableName)); + + // Release the parked index write and let the touch finish its data-side persist. + indexWriteProceed.countDown(); + toucher.join(TimeUnit.SECONDS.toMillis(60)); + assertFalse("toucher thread did not finish", toucher.isAlive()); + if (touchError.get() != null) { + throw touchError.get(); + } + + // The touch's fresh heartbeat (@t0+30s) keeps the row alive at the current clock (t0+65s, gap + // 35s < TTL). covcol@t0 survived the concurrent compaction, so the data table and the index + // still agree covcol is present. + assertCovColConsistent(conn, tableName, indexName, "r1", "k1", COVCOL_VALUE); + } + } + + private void runTtlBoundaryScenario(boolean useUpsertSelect) throws Exception { + String tableName = generateUniqueName(); + String indexName = generateUniqueName(); + try (Connection conn = DriverManager.getConnection(getUrl())) { + conn.setAutoCommit(false); + conn.createStatement() + .execute("CREATE TABLE " + tableName + " (id VARCHAR NOT NULL PRIMARY KEY, " + + "idxcol VARCHAR, covcol VARCHAR, touchcol VARCHAR)" + withClause(TTL, true)); + conn.createStatement() + .execute("CREATE INDEX " + indexName + " ON " + tableName + " (idxcol) INCLUDE (covcol)"); + conn.commit(); + + // Freeze the clock at t0; every cell of the full initial row lands at t0. + long t0 = EnvironmentEdgeManager.currentTimeMillis(); + injectEdge.setValue(t0); + EnvironmentEdgeManager.injectEdge(injectEdge); + + conn.createStatement().execute("UPSERT INTO " + tableName + + " (id, idxcol, covcol, touchcol) VALUES ('r1', 'k1', '" + COVCOL_VALUE + "', 'x0')"); + conn.commit(); + + // Push the t0 cells onto an HFile so the later major compaction merges two files. + flushTable(conn, tableName); + flushTable(conn, indexName); + + // Read the partial touch while the row is still alive (one tick before the boundary) but do + // NOT commit yet. For the UPSERT SELECT variant the inner SELECT reads the live row here. + injectEdge.incrementValue(TTL * 1000L - 1); + String expectedTouch; + if (useUpsertSelect) { + int affected = conn.createStatement().executeUpdate("UPSERT INTO " + tableName + + " (id, touchcol) SELECT id, touchcol FROM " + tableName + " WHERE id = 'r1'"); + assertEquals("UPSERT SELECT must read the still-alive row and buffer one touch", 1, + affected); + expectedTouch = "x0"; + } else { + conn.createStatement() + .executeUpdate("UPSERT INTO " + tableName + " (id, touchcol) VALUES ('r1', 'x1')"); + expectedTouch = "x1"; + } + + // Advance two ticks past the boundary and commit: the touch's mutationTimestamp is now + // t0 + TTL + 1 (gap > TTL). idxcol@t0 / covcol@t0 are masked at the internal read anchored at + // mutationTimestamp, so the index is rebuilt without them - matching the data side. + injectEdge.incrementValue(2); + conn.commit(); + + // Settle each side's physical state; the major compaction merges the t0 HFile with this one. + flushAndMajorCompact(conn, tableName); + flushAndMajorCompact(conn, indexName); + + // Data path (forced data-table read): the row is alive by PK, covcol read past the boundary. + String dataCov; + String dataTouch; + try (ResultSet rs = conn.createStatement().executeQuery( + "SELECT /*+ NO_INDEX */ covcol, touchcol FROM " + tableName + " WHERE id = 'r1'")) { + assertTrue("data-table row must still exist (touchcol keeps it alive)", rs.next()); + dataCov = rs.getString(1); + dataTouch = rs.getString(2); + assertFalse("exactly one data-table row", rs.next()); + } + + // Index path (forced index read by PK): idxcol@t0 is masked past the boundary, so the row is + // reachable through the covered index only by its PK suffix, not by idxcol='k1'. + String indexCov; + try (ResultSet rs = conn.createStatement().executeQuery("SELECT /*+ INDEX(" + tableName + " " + + indexName + ") */ covcol FROM " + tableName + " WHERE id = 'r1'")) { + assertTrue("index-visible row must exist", rs.next()); + indexCov = rs.getString(1); + assertFalse("exactly one index-visible row", rs.next()); + } + + assertEquals("touchcol must survive (row alive)", expectedTouch, dataTouch); + assertEquals("covcol must agree between data and index", dataCov, indexCov); + } + } + + // ---- helpers ---- + + /** + * Reads the single {@code counter} value for row {@code r1}, asserting exactly one live row. Used + * by {@link #testNoIndexAtomicUpsertMasksExpiredRow} to observe the atomic upsert's result. + */ + private static int atomicCounter(Connection conn, String tableName) throws SQLException { + try (ResultSet rs = conn.createStatement() + .executeQuery("SELECT counter FROM " + tableName + " WHERE id = 'r1'")) { + assertTrue("row should be present", rs.next()); + int value = rs.getInt(1); + assertFalse("expected exactly one row", rs.next()); + return value; + } + } + + private void assertCovColConsistent(Connection conn, String queryTableName, String indexName, + String id, String idxColValue, String expectedCovVal) throws SQLException { + // Force the data-table read. + try (ResultSet rs = conn.createStatement().executeQuery( + "SELECT /*+ NO_INDEX */ covcol FROM " + queryTableName + " WHERE id = '" + id + "'")) { + assertTrue("data-table row should exist", rs.next()); + assertEquals("data-table covcol", expectedCovVal, rs.getString(1)); + assertFalse(rs.next()); + } + // Read via the index (idxcol is the leading index column). + try (ResultSet rs = + conn.createStatement().executeQuery("SELECT /*+ INDEX(" + queryTableName + " " + indexName + + ") */ id, covcol FROM " + queryTableName + " WHERE idxcol = '" + idxColValue + "'")) { + assertTrue("index-visible row should exist", rs.next()); + assertEquals("index id", id, rs.getString(1)); + assertEquals("index covcol", expectedCovVal, rs.getString(2)); + assertFalse("exactly one index-visible row", rs.next()); + } + } + + private void assertCovColAbsent(Connection conn, String queryTableName, String indexName, + String id) throws SQLException { + try (ResultSet rs = conn.createStatement().executeQuery( + "SELECT /*+ NO_INDEX */ covcol FROM " + queryTableName + " WHERE id = '" + id + "'")) { + assertTrue("data-table row should exist", rs.next()); + assertNull("data-table row should not have covcol", rs.getString(1)); + assertFalse("exactly one data-table row", rs.next()); + } + try (ResultSet rs = conn.createStatement().executeQuery("SELECT /*+ INDEX(" + queryTableName + + " " + indexName + ") */ covcol FROM " + queryTableName + " WHERE id = '" + id + "'")) { + assertTrue("index-visible row should exist", rs.next()); + assertNull("index-visible row should not have covcol", rs.getString(1)); + assertFalse("exactly one index-visible row", rs.next()); + } + } + + private void assertTouchColConsistent(Connection conn, String queryTableName, String indexName, + String id, String expectedTouchVal) throws SQLException { + try (ResultSet rs = conn.createStatement().executeQuery( + "SELECT /*+ NO_INDEX */ touchcol FROM " + queryTableName + " WHERE id = '" + id + "'")) { + assertTrue("data-table row should exist", rs.next()); + assertEquals("data-table touchcol", expectedTouchVal, rs.getString(1)); + assertFalse(rs.next()); + } + try (ResultSet rs = conn.createStatement().executeQuery("SELECT /*+ INDEX(" + queryTableName + + " " + indexName + ") */ touchcol FROM " + queryTableName + " WHERE id = '" + id + "'")) { + assertTrue("index-visible row should exist", rs.next()); + assertEquals("index touchcol", expectedTouchVal, rs.getString(1)); + assertFalse("exactly one index-visible row", rs.next()); + } + } + + /** + * Data-side-only liveness check for the uncovered-index case, where touchcol cannot be read + * through the index (an uncovered index stores no data columns and is reachable only by its + * indexed column). Asserts the data row is alive and carries the expected touchcol. + */ + private void assertDataTouchColAlive(Connection conn, String queryTableName, String id, + String expectedTouchVal) throws SQLException { + try (ResultSet rs = conn.createStatement().executeQuery( + "SELECT /*+ NO_INDEX */ touchcol FROM " + queryTableName + " WHERE id = '" + id + "'")) { + assertTrue("data-table row should exist", rs.next()); + assertEquals("data-table touchcol", expectedTouchVal, rs.getString(1)); + assertFalse(rs.next()); + } + } + + private void assertUncoveredIdxColConsistent(Connection conn, String queryTableName, + String indexName, String id, String idxColValue) throws SQLException { + try (ResultSet rs = + conn.createStatement().executeQuery("SELECT /*+ INDEX(" + queryTableName + " " + indexName + + ") */ id, idxcol FROM " + queryTableName + " WHERE idxcol = '" + idxColValue + "'")) { + assertTrue("index-visible row should exist", rs.next()); + assertEquals("index id", id, rs.getString(1)); + assertEquals("index idxcol", idxColValue, rs.getString(2)); + assertFalse("exactly one index-visible row", rs.next()); + } + } + + private void assertUncoveredIdxColAbsent(Connection conn, String queryTableName, String indexName, + String idxColValue) throws SQLException { + try (ResultSet rs = + conn.createStatement().executeQuery("SELECT /*+ INDEX(" + queryTableName + " " + indexName + + ") */ idxcol FROM " + queryTableName + " WHERE idxcol = '" + idxColValue + "'")) { + assertFalse("no index-visible row should exist", rs.next()); + } + } + + static void flushTable(Connection conn, String tableName) throws Exception { + try (Admin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin()) { + admin.flush(TableName.valueOf(tableName)); + } + } + + static void flushAndMajorCompact(Connection conn, String tableName) throws Exception { + TableName tn = TableName.valueOf(tableName); + try (Admin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin()) { + admin.flush(tn); + } + TestUtil.majorCompact(getUtility(), tn); + } + + /** + * A region observer installed on the index table that parks the first write to a named index in + * {@code preBatchMutate}, letting a test run a concurrent major compaction on the data table + * while the secondary-index write is in flight. This is the "slow index write" seam: the index + * write is driven by {@code IndexRegionObserver.doPre -> table.batch(...)}, which reaches the + * index region's {@code preBatchMutate} strictly after the data-table current-row read and + * strictly before doPre returns. It is a one-shot per arming so only the touch's write blocks, + * not later maintenance. + */ + public static class BlockingIndexWriteObserver extends SimpleRegionObserver { + private static volatile String targetIndexName; + private static volatile CountDownLatch reached; + private static volatile CountDownLatch proceed; + private static final AtomicBoolean fired = new AtomicBoolean(false); + + static void arm(String indexName, CountDownLatch reachedLatch, CountDownLatch proceedLatch) { + targetIndexName = indexName; + reached = reachedLatch; + proceed = proceedLatch; + fired.set(false); + } + + @Override + public void preBatchMutate(ObserverContext c, + MiniBatchOperationInProgress miniBatchOp) throws IOException { + String tableName = c.getEnvironment().getRegionInfo().getTable().getNameAsString(); + if (tableName.equals(targetIndexName) && fired.compareAndSet(false, true)) { + reached.countDown(); + try { + proceed.await(60, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("interrupted while parking index write", e); + } + } + } + } +} diff --git a/phoenix-core/src/it/java/org/apache/phoenix/iterate/PhoenixQueryTimeoutIT.java b/phoenix-core/src/it/java/org/apache/phoenix/iterate/PhoenixQueryTimeoutIT.java index 98724c8dc64..8dd5ea5d7f4 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/iterate/PhoenixQueryTimeoutIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/iterate/PhoenixQueryTimeoutIT.java @@ -201,8 +201,10 @@ public void testScanningResultIteratorQueryTimeoutForPagingWithVeryLowTimeout() return false; } for (HRegion region : regions) { - if (region.getCoprocessorHost().findCoprocessor( - DelayedRegionObserver.class.getName()) == null) { + if ( + region.getCoprocessorHost().findCoprocessor(DelayedRegionObserver.class.getName()) + == null + ) { return false; } }