PHOENIX-7961 Secondary index diverges from data table after TTL expiry on partial-touch upserts (#2574)
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 52517a8..b0bfac0 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 @@
// 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/schema/MetaDataClient.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 22833e2..e70aba7 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 @@
/**
* 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 1dfc5b1..0be1157 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,104 @@
}
}
+ /**
+ * 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.
+ * <p>
+ * The guiding principle is to thread precisely what the read path
+ * ({@link #setScanAttributesForPhoenixTTL}) would set as the {@code _TTL} scan attribute:
+ * <ul>
+ * <li>For a base <b>TABLE</b>, the numeric literal TTL lives on the HBase CF descriptor, so the
+ * server's {@link org.apache.phoenix.coprocessor.TTLRegionScanner} CF-descriptor fallback derives
+ * it. We set no {@code _TTL} attribute (matching the read path for base tables).</li>
+ * <li>For a <b>VIEW</b>, the view-level TTL is stored in SYSTEM.CATALOG and not on the shared CF
+ * descriptor (which carries the base table's TTL), so the view's compiled literal expression must
+ * be threaded per-mutation as {@code _TTL}, but only when {@code serialize()} is non-null — the
+ * exact non-null filter the read path uses. {@code serialize()} returns null only for NONE, which
+ * correctly threads nothing (matching the read path). This reuses the single {@code _TTL}
+ * attribute for both literal and conditional TTL, exactly as the scan path does; the server
+ * disambiguates by the deserialized expression type
+ * ({@link org.apache.phoenix.hbase.index.IndexRegionObserver#extractLiteralTTLForInternalScan}).</li>
+ * <li>For either type, {@code IS_STRICT_TTL=false} is set only when the table/view is non-strict,
+ * so absence defaults to strict, matching the read-path convention and avoiding over-masking of a
+ * non-strict table.</li>
+ * <li>For either type, the empty-column CF/CQ are threaded <b>unconditionally</b> (for any
+ * mutable literal-TTL table/view, regardless of the TTL value or the view-TTL flag), exactly as
+ * the client read path does: {@link #setScanAttributesForClient} sets the empty column on every
+ * non-analyze scan. They merely identify the table's empty column and only <i>enable</i> masking;
+ * {@link org.apache.phoenix.coprocessor.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 <i>identically</i> to a client read rather than
+ * diverging from it. They are also the only source of these values for the no-index current-row
+ * read (an atomic / ON DUPLICATE KEY / {@code returnResult} / row-delete on a TTL table), which
+ * has no {@code IndexMaintainer} on the server. Derived exactly as the read path derives them via
+ * {@link SchemaUtil#getEmptyColumnFamily(PTable)} /
+ * {@link SchemaUtil#getEmptyColumnQualifier}.</li>
+ * </ul>
+ */
+ public static void annotateMutationWithLiteralTTL(PhoenixConnection connection, PTable table,
+ List<? extends Mutation> mutations) throws SQLException {
+
+ if (!(table.getTTLExpression() instanceof LiteralTTLExpression)) {
+ 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.
+ // The server disambiguates a mutation's _TTL by deserialized type, so reusing the same
+ // attribute for both literal and conditional TTL (as the scan path does) never confuses the
+ // conditional path.
+ 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();
+ }
+
+ 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<Filter> 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 0000000..ecd4d9e
--- /dev/null
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/ServerScanUtil.java
@@ -0,0 +1,123 @@
+/*
+ * 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.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.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
+import org.apache.phoenix.schema.types.PBoolean;
+
+/**
+ * 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 the
+ * TTL-masking step for server-side callers (e.g. {@code IndexRegionObserver} current-row reads) so
+ * an internal scan masks identically to a client scan.
+ * <p>
+ * Note: server paging is intentionally <b>not</b> reproduced here. Paging exists to bound the work
+ * a single RPC handler thread does before yielding; these internal scans are region-local (opened
+ * directly on the {@link Region}, not through the RPC scan path), so they consume no handler thread
+ * and there is nothing for paging to protect.
+ */
+public class ServerScanUtil {
+
+ private ServerScanUtil() {
+ }
+
+ /**
+ * Sets the Phoenix TTL scan attributes on an internal data-table scan so it masks exactly like a
+ * client read.
+ * <p>
+ * TTL masking attributes ({@link TTLRegionScanner} reads these):
+ * <ul>
+ * <li>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;</li>
+ * <li>{@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);</li>
+ * <li>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.</li>
+ * </ul>
+ */
+ public static void setInternalScanAttributes(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);
+ }
+ }
+
+ /**
+ * Opens a region scanner wrapped in a {@link TTLRegionScanner} 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.
+ * <p>
+ * The raw HBase {@link RegionScanner} is wrapped in a {@link NonPagingRegionScanner} rather than
+ * passed to {@link TTLRegionScanner} directly: {@link TTLRegionScanner} requires its delegate to
+ * be a {@link DelegateRegionScanner} — its gap-analysis and re-scan paths cast the delegate and
+ * call {@code getNewRegionScanner} — so a raw HBase scanner would fail those casts. On the client
+ * read path {@code PagingRegionScanner} fills this role; here we use a non-paging equivalent
+ * because the scan is region-local (opened directly on the {@link Region}, off the RPC path), so
+ * it holds no handler thread and has nothing for paging to protect.
+ */
+ public static RegionScanner openRegionScanner(RegionCoprocessorEnvironment env, Region region,
+ Scan scan) throws IOException {
+ return new TTLRegionScanner(env, scan,
+ new NonPagingRegionScanner(region, region.getScanner(scan)));
+ }
+
+ /**
+ * A minimal {@link DelegateRegionScanner} over a raw HBase {@link RegionScanner} that can re-open
+ * a fresh scanner from the {@link Region} via {@link #getNewRegionScanner(Scan)} but adds no
+ * paging. {@link TTLRegionScanner} needs a {@link DelegateRegionScanner} delegate (it casts and
+ * calls {@code getNewRegionScanner} during gap analysis and re-scans); this fills that role for
+ * internal region-local scans without reproducing the paging machinery that
+ * {@code PagingRegionScanner} adds for the RPC read path.
+ */
+ private static final class NonPagingRegionScanner extends DelegateRegionScanner {
+ private final Region region;
+
+ private NonPagingRegionScanner(Region region, RegionScanner scanner) {
+ super(scanner);
+ this.region = region;
+ }
+
+ @Override
+ public RegionScanner getNewRegionScanner(Scan scan) throws IOException {
+ return new NonPagingRegionScanner(region, region.getScanner(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 9763388..b1145c7 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
@@ -32,6 +32,7 @@
import java.io.DataInputStream;
import java.io.EOFException;
import java.io.IOException;
+import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -85,6 +86,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 +121,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;
@@ -378,8 +382,40 @@
private boolean hasTransform;
private boolean returnResult;
private boolean returnOldRow;
- private boolean hasConditionalTTL; // table has Conditional TTL
private boolean immutableRows;
+ // The deserialized CompiledTTLExpression for this batch, or null when no _TTL attribute was
+ // set. LiteralTTLExpression for a view with literal TTL; CompiledConditionalTTLExpression for
+ // a conditional TTL table. Populated once by extractTTLKindForInternalScan before
+ // identifyMutationTypes runs. Consumers gate on the concrete instance type via the
+ // hasLiteralTTL() / hasConditionalTTL() methods.
+ private CompiledTTLExpression ttlExpressionForBatch;
+ // 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;
+ // Whether the table/view uses strict TTL, captured (via isStrictTTLEnabled) alongside the other
+ // internal-scan attributes so getCurrentRowStates masks exactly like a client read: a
+ // non-strict table is never masked. Defaults to strict, matching isStrictTTLEnabled's default.
+ private boolean hasStrictTTL = PTable.DEFAULT_IS_STRICT_TTL;
+
+ public boolean hasLiteralTTL() {
+ return ttlExpressionForBatch instanceof LiteralTTLExpression;
+ }
+
+ public boolean hasConditionalTTL() {
+ return ttlExpressionForBatch instanceof CompiledConditionalTTLExpression;
+ }
+
+ public boolean hasStrictTTL() {
+ return hasStrictTTL;
+ }
+
+ public boolean hasStrictConditionalTTL() {
+ return hasConditionalTTL() && hasStrictTTL();
+ }
public BatchMutateContext() {
this.clientVersion = 0;
@@ -679,7 +715,7 @@
Mutation m = miniBatchOp.getOperation(i);
if (
this.builder.isAtomicOp(m) || context.returnResult || this.builder.isEnabled(m)
- || (this.builder.hasConditionalTTL(m) && isStrictTTLEnabled(miniBatchOp))
+ || context.hasStrictConditionalTTL()
) {
ImmutableBytesPtr row = new ImmutableBytesPtr(m.getRow());
context.rowsToLock.add(row);
@@ -775,18 +811,16 @@
*/
private void updateMutationsForConditionalTTL(MiniBatchOperationInProgress<Mutation> miniBatchOp,
BatchMutateContext context) throws IOException {
- // If TTL is not strict, skip conditional TTL processing
- if (!isStrictTTLEnabled(miniBatchOp)) {
+ if (!context.hasConditionalTTL()) {
return;
}
+ CompiledConditionalTTLExpression ttlExpr =
+ (CompiledConditionalTTLExpression) context.ttlExpressionForBatch;
// mapping from row key to indices in mini batch
Map<ImmutableBytesPtr, List<Integer>> expiredVersions = Maps.newHashMap();
Set<ImmutableBytesPtr> notExpiredVersions = Sets.newHashSet();
for (int i = 0; i < miniBatchOp.size(); i++) {
Mutation m = miniBatchOp.getOperation(i);
- if (!builder.hasConditionalTTL(m)) {
- continue;
- }
if (IndexUtil.isDeleteFamily(m)) {
// no need to fix DeleteFamily mutation
continue;
@@ -808,9 +842,6 @@
positions.add(i);
continue;
}
- byte[] ttl = m.getAttribute(BaseScannerRegionObserverConstants.TTL);
- CompiledConditionalTTLExpression ttlExpr =
- (CompiledConditionalTTLExpression) TTLExpressionFactory.create(ttl);
List<Cell> currentRow = flattenCells(currentVersion);
// isRaw is false because we are looking at a Put mutation
if (ttlExpr.isExpired(currentRow, false)) {
@@ -913,7 +944,7 @@
}
public static void setTimestamps(MiniBatchOperationInProgress<Mutation> miniBatchOp,
- IndexBuildManager builder, long ts, boolean isTTLStrict) throws IOException {
+ IndexBuildManager builder, long ts, boolean hasStrictConditionalTTL) throws IOException {
for (Integer i = 0; i < miniBatchOp.size(); i++) {
if (isAtomicOperationComplete(miniBatchOp.getOperationStatus(i))) {
continue;
@@ -922,9 +953,8 @@
// skip this mutation if we aren't enabling indexing or Conditional TTL
// or not an atomic op or if it is an atomic op
// and its timestamp is already set(not LATEST)
- // Also, skip conditional TTL if TTL is not strict
if (
- !builder.isEnabled(m) && (!builder.hasConditionalTTL(m) || !isTTLStrict)
+ !builder.isEnabled(m) && !hasStrictConditionalTTL
&& !((builder.isAtomicOp(m) || builder.returnResult(m))
&& IndexUtil.getMaxTimestamp(m) == HConstants.LATEST_TIMESTAMP)
) {
@@ -1161,9 +1191,24 @@
/**
* Retrieve the data row state either from memory or disk. The rows are locked by the caller.
+ * <p>
+ * 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.
+ * When the batch has a literal TTL ({@link BatchMutateContext#hasLiteralTTL()}), the TTL
+ * expression is re-serialized and set on the scan so {@link TTLRegionScanner} masks with the
+ * view's TTL (null for base tables, which use the CF-descriptor fallback).
+ * {@link BatchMutateContext#hasStrictTTL()} avoids over-masking a non-strict table. Masking is a
+ * strict no-op when Phoenix compaction is disabled. Unlike a client read the scan is not paged:
+ * it is region-local (off the RPC path), so it holds no handler thread and there is nothing for
+ * paging to protect.
*/
private void getCurrentRowStates(ObserverContext<RegionCoprocessorEnvironment> c,
- BatchMutateContext context) throws IOException {
+ BatchMutateContext context, long batchTimestamp) throws IOException {
Set<KeyRange> keys = new HashSet<KeyRange>(context.rowsToLock.size());
for (ImmutableBytesPtr rowKeyPtr : context.rowsToLock) {
PendingRow pendingRow = new PendingRow(rowKeyPtr, context);
@@ -1212,6 +1257,19 @@
return;
}
+ byte[] emptyCF = context.emptyCFForInternalScan;
+ byte[] emptyCQ = context.emptyCQForInternalScan;
+ // Re-serialize the literal TTL expression for the scan attribute. Only views carry a literal
+ // TTL here; a base table relies on the CF-descriptor fallback (null is correct for them).
+ byte[] literalTTLForScan = null;
+ try {
+ literalTTLForScan =
+ context.hasLiteralTTL() ? context.ttlExpressionForBatch.serialize() : null;
+ } catch (SQLException e) {
+ throw new IOException("Failed to serialize literal TTL expression for internal scan", e);
+ }
+ boolean isStrictTTL = context.hasStrictTTL();
+
if (this.useBloomFilter) {
for (KeyRange key : keys) {
// Scan.java usage alters scan instances, safer to create scan instance per usage
@@ -1220,6 +1278,16 @@
// 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. The range upper bound is
+ // batchTimestamp + 1 (not batchTimestamp) because HBase scan time ranges are half-open
+ // [min, max): a bound of batchTimestamp would exclude any committed cell sitting at exactly
+ // batchTimestamp. That collision cannot happen in production (getBatchTimestamp forces a
+ // distinct timestamp for same-row batches), but a frozen test clock can produce it, and the
+ // read must see those cells to build a correct current-row state. The masking clock is then
+ // batchTimestamp + 1, trimming expired cells as of the timestamp the index is built at.
+ scan.setTimeRange(0, batchTimestamp + 1);
+ ServerScanUtil.setInternalScanAttributes(scan, emptyCF, emptyCQ, literalTTLForScan,
+ isStrictTTL);
readDataTableRows(c, context, scan);
}
} else {
@@ -1228,13 +1296,27 @@
scanRanges.initializeScan(scan);
SkipScanFilter skipScanFilter = scanRanges.getSkipScanFilter();
scan.setFilter(skipScanFilter);
+ // Anchor TTLRegionScanner's masking clock at batchTimestamp. The range upper bound is
+ // batchTimestamp + 1 (not batchTimestamp) because HBase scan time ranges are half-open
+ // [min, max): a bound of batchTimestamp would exclude any committed cell sitting at exactly
+ // batchTimestamp. That collision cannot happen in production (getBatchTimestamp forces a
+ // distinct timestamp for same-row batches), but a frozen test clock can produce it, and the
+ // read must see those cells to build a correct current-row state. The masking clock is then
+ // batchTimestamp + 1, trimming expired cells as of the timestamp the index is built at.
+ scan.setTimeRange(0, batchTimestamp + 1);
+ ServerScanUtil.setInternalScanAttributes(scan, emptyCF, emptyCQ, literalTTLForScan,
+ isStrictTTL);
readDataTableRows(c, context, scan);
}
}
private void readDataTableRows(ObserverContext<RegionCoprocessorEnvironment> 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<Cell> cells = new ArrayList<Cell>();
@@ -1664,6 +1746,42 @@
}
}
+ /**
+ * Captures the TTL-related internal-scan attributes the client threads off the representative
+ * mutation and deserializes the {@code _TTL} attribute into
+ * {@link BatchMutateContext#ttlExpressionForBatch}.
+ * <p>
+ * Two things are captured:
+ * <ul>
+ * <li>The <b>empty-column CF/CQ</b> — threaded whenever an effective (non-NONE) literal TTL
+ * applies, for both base tables and views. They are inert on the write path, so they are left on
+ * the mutations.</li>
+ * <li>The <b>{@code _TTL}</b> — deserialized once into {@code context.ttlExpressionForBatch}.
+ * When needed for an internal scan, the expression is re-serialized from this field.</li>
+ * </ul>
+ * Must run before {@link #identifyMutationTypes} so that {@code context.hasStrictTTL()} is
+ * available. A table has exactly one TTL kind and a batch never mixes views, so the deserialized
+ * expression is uniform across the batch.
+ */
+ private void extractTTLKindForInternalScan(MiniBatchOperationInProgress<Mutation> miniBatchOp,
+ BatchMutateContext context) throws IOException {
+ if (miniBatchOp.size() == 0) {
+ return;
+ }
+ Mutation representative = miniBatchOp.getOperation(0);
+ context.emptyCFForInternalScan =
+ representative.getAttribute(BaseScannerRegionObserverConstants.EMPTY_COLUMN_FAMILY_NAME);
+ context.emptyCQForInternalScan =
+ representative.getAttribute(BaseScannerRegionObserverConstants.EMPTY_COLUMN_QUALIFIER_NAME);
+ context.hasStrictTTL = isStrictTTLEnabled(miniBatchOp);
+
+ byte[] ttlBytes = representative.getAttribute(BaseScannerRegionObserverConstants.TTL);
+ if (ttlBytes == null) {
+ return;
+ }
+ context.ttlExpressionForBatch = TTLExpressionFactory.create(ttlBytes);
+ }
+
private void identifyMutationTypes(MiniBatchOperationInProgress<Mutation> miniBatchOp,
BatchMutateContext context) throws IOException {
for (int i = 0; i < miniBatchOp.size(); i++) {
@@ -1678,9 +1796,7 @@
context.returnOldRow = true;
}
}
- if (this.builder.hasConditionalTTL(m) && isStrictTTLEnabled(miniBatchOp)) {
- context.hasConditionalTTL = true;
- }
+
if (this.builder.isAtomicOp(m) || this.builder.returnResult(m)) {
context.hasAtomic = true;
if (context.hasRowDelete) {
@@ -1781,13 +1897,19 @@
// 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 HashSet<>(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 HashSet<>(context.rowsToLock));
return ts;
}
}
@@ -1806,7 +1928,7 @@
// 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 HashSet<>(context.rowsToLock));
return ts;
}
}
@@ -1817,6 +1939,7 @@
BatchMutateContext context = new BatchMutateContext(indexMetaData.getClientVersion());
setBatchMutateContext(c, context);
identifyIndexMaintainerTypes(indexMetaData, context);
+ extractTTLKindForInternalScan(miniBatchOp, context);
identifyMutationTypes(miniBatchOp, context);
context.populateOriginalMutations(miniBatchOp);
@@ -1838,9 +1961,18 @@
context.currentPhase = BatchMutatePhase.PRE;
long onDupCheckTime = 0;
+ // Compute the batch timestamp now, while holding the row locks and before the current-row read,
+ // so the internal current-row scan can anchor its TTL masking clock at exactly the timestamp
+ // the
+ // index is built at (see getCurrentRowStates). getBatchTimestamp only needs the locked rows,
+ // which are already populated, and the "got the next timestamp while holding the row locks"
+ // invariant still holds since the locks are held through the rest of this method.
+ TableName table = c.getEnvironment().getRegion().getRegionInfo().getTable();
+ long batchTimestamp = getBatchTimestamp(context, table);
+
if (
context.hasAtomic || context.returnResult || context.hasGlobalIndex
- || context.hasUncoveredIndex || context.hasTransform || context.hasConditionalTTL
+ || context.hasUncoveredIndex || context.hasTransform || context.hasStrictConditionalTTL()
) {
// Retrieve the current row states from the data table while holding the lock.
// This is needed for both atomic mutations and global indexes
@@ -1850,16 +1982,16 @@
if (
!context.immutableRows && context.hasGlobalIndex || context.hasTransform
|| context.hasAtomic || context.returnResult || context.hasRowDelete
- || context.hasConditionalTTL
+ || context.hasStrictConditionalTTL()
|| !context.immutableRows && context.hasUncoveredIndex
&& isPartialUncoveredIndexMutation(indexMetaData, miniBatchOp)
) {
- getCurrentRowStates(c, context);
+ getCurrentRowStates(c, context, batchTimestamp);
}
onDupCheckTime += (EnvironmentEdgeManager.currentTimeMillis() - start);
}
- if (context.hasConditionalTTL) {
+ if (context.hasStrictConditionalTTL()) {
// If the table has conditional TTL, then before making any update to a row
// we need to evaluate the ttl expression to check if the current row version has
// expired. If the current row version has expired then the incoming mutation has to
@@ -1885,11 +2017,9 @@
}
}
- TableName table = c.getEnvironment().getRegion().getRegionInfo().getTable();
- long batchTimestamp = getBatchTimestamp(context, table);
// Update the timestamps of the data table mutations to prevent overlapping timestamps
// (which prevents index inconsistencies as this case is not handled).
- setTimestamps(miniBatchOp, builder, batchTimestamp, isStrictTTLEnabled(miniBatchOp));
+ setTimestamps(miniBatchOp, builder, batchTimestamp, context.hasStrictConditionalTTL());
if (context.hasGlobalIndex || context.hasUncoveredIndex || context.hasTransform) {
// Prepare next data rows states for pending mutations (for global indexes)
prepareDataRowStates(c, miniBatchOp, context, batchTimestamp);
@@ -2448,7 +2578,6 @@
delete.add(cell);
}
}
-
if (!put.isEmpty() || !delete.isEmpty()) {
PTable table = operations.get(0).getFirst();
addEmptyKVCellToPut(put, tuple, table);
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/BaseIndexBuilder.java b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/BaseIndexBuilder.java
index 768a751..e5d6fef 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/BaseIndexBuilder.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/BaseIndexBuilder.java
@@ -146,11 +146,6 @@
return false;
}
- @Override
- public boolean hasConditionalTTL(Mutation m) {
- return false;
- }
-
public RegionCoprocessorEnvironment getEnv() {
return this.env;
}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuildManager.java b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuildManager.java
index 53a8644..1d45dfd 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuildManager.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuildManager.java
@@ -185,7 +185,4 @@
return delegate.returnResult(m);
}
- public boolean hasConditionalTTL(Mutation m) {
- return delegate.hasConditionalTTL(m);
- }
}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuilder.java b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuilder.java
index 1d13448..6db7ed3 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuilder.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuilder.java
@@ -156,11 +156,4 @@
*/
boolean returnResult(Mutation m);
- /**
- * True if mutation has a Conditional TTL expression
- * @param m Mutation object
- * @return True if mutation has a Conditional TTL expression, False otherwise
- */
- boolean hasConditionalTTL(Mutation m);
-
}
diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/index/PhoenixIndexBuilder.java b/phoenix-core-server/src/main/java/org/apache/phoenix/index/PhoenixIndexBuilder.java
index 490b0ad..1532480 100644
--- a/phoenix-core-server/src/main/java/org/apache/phoenix/index/PhoenixIndexBuilder.java
+++ b/phoenix-core-server/src/main/java/org/apache/phoenix/index/PhoenixIndexBuilder.java
@@ -44,7 +44,6 @@
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.io.WritableUtils;
import org.apache.phoenix.coprocessor.generated.PTableProtos;
-import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants;
import org.apache.phoenix.coprocessorclient.BaseScannerRegionObserverConstants.ReplayWrite;
import org.apache.phoenix.exception.DataExceedsCapacityException;
import org.apache.phoenix.expression.Expression;
@@ -291,9 +290,4 @@
&& (Arrays.equals(returnResult, PhoenixIndexBuilderHelper.RETURN_RESULT_ROW)
|| Arrays.equals(returnResult, PhoenixIndexBuilderHelper.RETURN_RESULT_OLD_ROW));
}
-
- @Override
- public boolean hasConditionalTTL(Mutation m) {
- return m.getAttribute(BaseScannerRegionObserverConstants.TTL) != null;
- }
}
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 0000000..4412193
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexDataTableSyncIT.java
@@ -0,0 +1,772 @@
+/*
+ * 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/TTL masking.
+ * <p>
+ * 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<Object[]> 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<String, String> baseServerProps() {
+ Map<String, String> 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, committed past covcol@t0's TTL
+ // boundary. With autoCommit off the UPSERT VALUES only buffers on the client; the server-side
+ // masked current-row read and the mutationTimestamp are both established at commit, by which
+ // point the clock is past covcol@t0's expiry. So the masked read drops covcol@t0 and the
+ // index is rebuilt WITHOUT covcol.
+ injectEdge.incrementValue(TTL * 1000L + 1);
+ conn.createStatement()
+ .executeUpdate("UPSERT INTO " + tableName + " (id, touchcol) VALUES ('r1', 'x2')");
+ 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");
+ }
+ }
+
+ @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 + 50_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<Exception> 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 (61s) 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 + 61_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+50s) keeps the row alive at the current clock (t0+61s, gap
+ // 11s < 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<RegionCoprocessorEnvironment> c,
+ MiniBatchOperationInProgress<Mutation> 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);
+ }
+ }
+ }
+ }
+}
