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/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,104 @@ 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). 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()) {

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));
+    }
+  }
+}