PHOENIX-7961 Secondary index diverges from data table after TTL expiry on partial-touch upserts

phoenix-core-client/src/main/java/org/apache/phoenix/execute/MutationState.java

@@ -1528,6 +1528,10 @@ private void sendMutations(Iterator<Entry<TableInfo, List<Mutation>>> 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.

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;

phoenix-core-client/src/main/java/org/apache/phoenix/schema/MetaDataClient.java

@@ -4923,9 +4923,9 @@ public MutationState addColumn(PTable table, List<ColumnDef> 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.
            */
         }
 

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.
+   * <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).</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)) {
+      // 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<Filter> filterIterator = filterList.getFilters().iterator();
     while (filterIterator.hasNext()) {

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.
+ * <p>
+ * 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.
+   * <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(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.
+   * <p>
+   * 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));
+  }
+}