Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,12 @@ public static long getMaxLookbackInMillis(Configuration conf) {
public static final String SKIP_REGION_BOUNDARY_CHECK = "_SKIP_REGION_BOUNDARY_CHECK";
public static final String TX_SCN = "_TxScn";
public static final String TTL = "_TTL";
// Literal TTL threaded per-mutation for the server-side internal current-row scan
// (IndexRegionObserver.getCurrentRowStates). Kept distinct from TTL (_TTL), which on a mutation
// means conditional TTL, so the server never has to disambiguate an overloaded attribute and an
// old RegionServer that predates this simply ignores the unknown attribute instead of
// mis-parsing it as a conditional expression.
public static final String LITERAL_TTL = "_LITERAL_TTL";
public static final String IS_STRICT_TTL = "_IS_STRICT_TTL";
public static final String SCAN_ACTUAL_START_ROW = "_ScanActualStartRow";
public static final String REPLAY_WRITES = "_IGNORE_NEWER_MUTATIONS";
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*/
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -1862,6 +1863,100 @@ 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 on the dedicated {@code _LITERAL_TTL} attribute (distinct from
* {@code _TTL}, which on a mutation means conditional 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)) {
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 on its own _LITERAL_TTL attribute (not _TTL, which on a
// mutation means conditional 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 attribute 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 nothing.
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.LITERAL_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()) {
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
/*
* 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.
*/
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));
}
}
Loading