From fefdf82492610b1ae671d371e35576ab4018a040 Mon Sep 17 00:00:00 2001 From: lokiore Date: Mon, 3 Aug 2026 13:47:14 -0700 Subject: [PATCH 1/2] PHOENIX-7907 :- Cutover lifecycle with PENDING_PARTIAL_PASS state, UCF-wait, and link removal at cutover commit Replace the inline partial pass that previously ran during transform cutover with an explicit, restartable cutover state machine in TransformMonitorTask, and tear down the dual-write links at the cutover commit so a single client cache-invalidation cycle propagates both the physical-table pointer swap and the dual-write shutoff. Lifecycle (all states committed to SYSTEM.TRANSFORM, monitored by the self-healing TransformMonitorTask): * PENDING_CUTOVER -> PENDING_PARTIAL_PASS: capture the cutover instant (CUTOVER_TS) BEFORE doCutover, then after doCutover swaps the physical-table pointer, persist a wait deadline instead of running the partial pass immediately. The deadline is the logical table's update-cache-frequency scaled by a safety margin (1.10), floored at 30 minutes and capped at 24 hours, so clients still holding a cached pointer to the old physical table refresh before the partial pass runs and late writes are not stranded as unverified rows. The raw frequency is clamped to the 24-hour ceiling BEFORE scaling because a table configured to never refresh its cache resolves its update-cache-frequency to Long.MAX_VALUE; scaling that and adding it to the current time would saturate into a negative (past) deadline that would defeat the wait entirely. The deadline is stored in a new nullable BIGINT column PENDING_PARTIAL_PASS_UNTIL_TS on SYSTEM.TRANSFORM. * PENDING_PARTIAL_PASS -> PARTIAL_PASS_RUNNING: once the wait window elapses, commit the transition (clearing the inherited full-pass job id so the monitoring branch cannot mistake the already-successful full pass for the partial pass and complete early), then kick the partial-pass TransformTool run. * PARTIAL_PASS_RUNNING -> COMPLETED / FAILED: monitor the partial-pass job. A job that cannot be confirmed successful (no job id registered because the initial kick failed before its STARTED transition, an unsuccessful job, or a job id that no longer resolves) is routed through a retry budget; once retries are exhausted the record reaches terminal FAILED rather than stranding a pointer-swapped table with unverified rows. Partial-pass repair floor (correctness): the partial pass re-verifies rows on the new physical table from a lower-bound timestamp. Deriving that floor from the post-wait lastStateTs stranded every write made to the old physical pointer during the [cutover, cutover + waitWindow] window -- exactly the writes the wait exists to let stale-cached clients drain -- because that floor sits past the window. The floor is instead derived from CUTOVER_TS (captured before doCutover, the most conservative instant) minus one so it is inclusive of writes stamped exactly at cutover; repairScanFloor centralizes this and falls back to lastStateTs only for pre-existing records that predate the CUTOVER_TS column. CUTOVER_TS is persisted durably (in its own commit) BEFORE doCutover swaps the pointer, and preserved across every downstream transition by the record copy-constructor. doCutover commits the pointer swap durably, so persisting CUTOVER_TS only in the later PENDING_PARTIAL_PASS commit would leave a crash window in between: a crash there would lose the instant and, on re-entry, recapture a later one that pushes the repair floor past the real cutover and silently drops the post-cutover-window writes. The PENDING_CUTOVER handling therefore commits CUTOVER_TS first and, on re-entry, reuses the already-persisted instant (resolveCutoverTs) rather than recapturing; a crash before that first commit is harmless because the pointer has not yet swapped. Dual-write link teardown at cutover commit (Transform.doCutover): * The base-table TRANSFORMING_NEW_TABLE link is deleted uncommitted and batched into the same commit as the base-table pointer swap. * Each child view's link is deleted inside the existing MUTATE_BATCH_SIZE view loop, paired with that view's pointer swap. A base table can have millions of views, so folding per-view link teardown into the bounded batch loop keeps every commit bounded while still pairing each link removal with its swap in one cache-invalidation cycle. doGetTable attaches the transforming-new-table per row from link presence, so a surviving view link would keep dual-write alive for view-routed writes after cutover. Schema: two nullable BIGINT columns are added to SYSTEM.TRANSFORM -- PENDING_PARTIAL_PASS_UNTIL_TS (the wait deadline) and CUTOVER_TS (the cutover instant used as the partial-pass repair floor); they are the first columns ever added to SYSTEM.TRANSFORM. A column added to a system table on upgrade only takes effect if SYSTEM.CATALOG's own header timestamp advances to the new min system-table timestamp, because the client upgrade gate reports SYSTEM.CATALOG's header timestamp: a min not backed by a genuine SYSTEM.CATALOG column-add at that timestamp leaves the catalog below the gate after an in-place upgrade and loops clients on UpgradeRequiredException. MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 is therefore bumped by one and a no-op marker column, UPGRADE_TS_ANCHOR_5_4_0, is added to SYSTEM.CATALOG at the new min so the header genuinely advances; the existing 5.4.0 SYSTEM.CATALOG column-add cascade is re-offset by one so every column keeps its original absolute timestamp (INDEX_CONSISTENCY stays at min-1, the anchor lands at min). The two transform columns are then added unconditionally via idempotent addColumnsIfNotExists; the previous strict-less-than gate is dropped because it was unreachable on a cluster already snapshotted at the pre-bump timestamp and would have left the columns unadded (ColumnNotFoundException on every transform read thereafter). A fresh install gets the anchor from the SYSTEM.CATALOG CREATE DDL and both transform columns from the SYSTEM.TRANSFORM CREATE DDL. SystemTransformRecord / TransformClient read and write both new columns with explicit BIGINT null handling. retry-count accounting: TransformTool's STARTED transition unconditionally increments (and auto-commits) the retry count. The first partial-pass kick pre-decrements to net zero (the initial pass is not a retry and must not consume budget); a genuine retry skips the decrement so the count strictly increases and the retries-exhausted -> FAILED transition stays reachable. Testing: CutoverLifecycleIT drives real and seeded cutovers with an injected clock (no real 30-minute sleep) and a job-lookup seam, covering the happy path (mutable / immutable / secondary-index / child-view tables), the wait-deadline honoring, the inherited-job-id clearing, the strand regressions (null job id, not-found job, retries exhausted) each asserting a terminal state, a never-cached table (update-cache-frequency NEVER) yielding a bounded future wait deadline rather than an overflowed past one, and a repair-floor regression (testPartialPassRepairFloorCoversPostCutoverWaitWindow) asserting CUTOVER_TS is captured at cutover, preserved across the transition to PARTIAL_PASS_RUNNING, and strictly precedes the post-wait lastStateTs -- the exact interval that a lastStateTs-derived floor would strand, and a re-entry regression (testCutoverReentryReusesPersistedCutoverTs) that resets a pointer-swapped record back to PENDING_CUTOVER with its CUTOVER_TS preserved, advances the clock far past it, re-runs the monitor, and asserts the persisted instant is reused unchanged rather than recaptured at the later clock. TransformMonitorTaskWaitTest is a fast unit test that pins the clamp-before-scale wait arithmetic (boundedPartialPassWaitMs) across the whole input domain (Long.MAX_VALUE / zero / negative / mid-range / at-and-above ceiling) so the deadline is always bounded and positive, the repair floor (repairScanFloor) resolving to cutoverTs-1, to the lastStateTs-1 fallback, and to 0 when neither is set, and the cutover-instant resolution (resolveCutoverTs) reusing a persisted instant on re-entry and capturing the current time on a first run. MetaDataUtilTest asserts the min system-table timestamp equals the highest timestamp SYSTEM.CATALOG's header reaches and pins its absolute offset, guarding against a future timestamp bump not backed by a SYSTEM.CATALOG column-add. Heavy user-table cutover ITs run on CI; the seeded strand and repair-floor regressions run locally. Generated-by: Claude Code (Opus 4.8) Co-authored-by: Claude Opus 4.8 (1M context) --- .../coprocessorclient/MetaDataProtocol.java | 2 +- .../phoenix/jdbc/PhoenixDatabaseMetaData.java | 20 + .../query/ConnectionQueryServicesImpl.java | 43 +- .../apache/phoenix/query/QueryConstants.java | 13 +- .../org/apache/phoenix/schema/PTable.java | 16 +- .../transform/SystemTransformRecord.java | 40 +- .../schema/transform/TransformClient.java | 19 +- .../tasks/TransformMonitorTask.java | 436 ++++++- .../mapreduce/transform/TransformTool.java | 14 +- .../phoenix/schema/transform/Transform.java | 95 ++ .../end2end/transform/CutoverLifecycleIT.java | 1057 +++++++++++++++++ .../tasks/TransformMonitorTaskWaitTest.java | 157 +++ .../apache/phoenix/util/MetaDataUtilTest.java | 35 + 13 files changed, 1896 insertions(+), 51 deletions(-) create mode 100644 phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/CutoverLifecycleIT.java create mode 100644 phoenix-core/src/test/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTaskWaitTest.java diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java b/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java index 08cac7cbe9a..497b4380423 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/coprocessorclient/MetaDataProtocol.java @@ -92,7 +92,7 @@ public abstract class MetaDataProtocol extends MetaDataService { public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_1_0 = MIN_SYSTEM_TABLE_TIMESTAMP_4_16_0; public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_2_0 = MIN_TABLE_TIMESTAMP + 38; public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_3_0 = MIN_TABLE_TIMESTAMP + 44; - public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 = MIN_TABLE_TIMESTAMP + 45; + public static final long MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 = MIN_TABLE_TIMESTAMP + 46; // MIN_SYSTEM_TABLE_TIMESTAMP needs to be set to the max of all the MIN_SYSTEM_TABLE_TIMESTAMP_* // constants public static final long MIN_SYSTEM_TABLE_TIMESTAMP = MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0; diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java index ed63485440f..563976296fd 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java @@ -215,6 +215,18 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData { public static final byte[] INDEX_TYPE_BYTES = Bytes.toBytes(INDEX_TYPE); public static final String INDEX_CONSISTENCY = "INDEX_CONSISTENCY"; public static final byte[] INDEX_CONSISTENCY_BYTES = Bytes.toBytes(INDEX_CONSISTENCY); + // No-op SYSTEM.CATALOG marker column, never populated or read. It exists solely to advance the + // SYSTEM.CATALOG header timestamp to MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 during an in-place upgrade. + // The isUpgradeRequired() gate keys off SYSTEM.CATALOG's header timestamp; the paired + // SYSTEM.TRANSFORM column-add does not itself advance that timestamp, so on a cluster + // bootstrapped + // in the window between sibling 5.4.0 features (already at the old threshold via + // INDEX_CONSISTENCY) + // the transform column-add would be unreachable and clients would loop on + // UpgradeRequiredException. + // Adding this genuinely-new column at the new threshold makes the upgrade reachable. Single-use: + // a future timestamp bump needs its own new marker (re-adding an existing column is a no-op). + public static final String UPGRADE_TS_ANCHOR_5_4_0 = "UPGRADE_TS_ANCHOR_5_4_0"; public static final String LINK_TYPE = "LINK_TYPE"; public static final byte[] LINK_TYPE_BYTES = Bytes.toBytes(LINK_TYPE); public static final String TASK_TYPE = "TASK_TYPE"; @@ -236,6 +248,14 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData { public static final String OLD_METADATA = "OLD_METADATA"; public static final String NEW_METADATA = "NEW_METADATA"; public static final String TRANSFORM_FUNCTION = "TRANSFORM_FUNCTION"; + // Epoch-millis (BIGINT, nullable) marking the earliest time the transform monitor may leave the + // PENDING_PARTIAL_PASS wait window. Compared against EnvironmentEdgeManager.currentTimeMillis(). + public static final String PENDING_PARTIAL_PASS_UNTIL_TS = "PENDING_PARTIAL_PASS_UNTIL_TS"; + // Epoch-millis (BIGINT, nullable) captured just before the cutover pointer swap and preserved + // across later transitions. The partial pass derives its repair-scan lower bound from this so + // rows written to the old pointer during the post-cutover cache-refresh window are re-verified + // rather than skipped. + public static final String CUTOVER_TS = "CUTOVER_TS"; public static final String TRANSFORM_TABLE_TTL = "7776000"; // 90 days public static final int TTL_FOR_MUTEX = 15 * 60; // 15min diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java index cb2a47230f4..5c38d7d3cc6 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java @@ -4801,22 +4801,22 @@ protected PhoenixConnection upgradeSystemCatalogIfRequired(PhoenixConnection met } if (currentServerSideTableTimeStamp < MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0) { metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG, - MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 9, + MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 10, PhoenixDatabaseMetaData.PHYSICAL_TABLE_NAME + " " + PVarchar.INSTANCE.getSqlTypeName()); metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG, - MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 8, + MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 9, PhoenixDatabaseMetaData.SCHEMA_VERSION + " " + PVarchar.INSTANCE.getSqlTypeName()); metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG, - MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 7, + MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 8, PhoenixDatabaseMetaData.EXTERNAL_SCHEMA_ID + " " + PVarchar.INSTANCE.getSqlTypeName()); metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG, - MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 6, + MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 7, PhoenixDatabaseMetaData.STREAMING_TOPIC_NAME + " " + PVarchar.INSTANCE.getSqlTypeName()); metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG, - MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 5, + MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 6, PhoenixDatabaseMetaData.INDEX_WHERE + " " + PVarchar.INSTANCE.getSqlTypeName()); metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG, - MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 4, + MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 5, PhoenixDatabaseMetaData.CDC_INCLUDE_TABLE + " " + PVarchar.INSTANCE.getSqlTypeName()); /** @@ -4824,16 +4824,23 @@ protected PhoenixConnection upgradeSystemCatalogIfRequired(PhoenixConnection met * PHOENIX_TTL Column. See PHOENIX-7023 */ metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG, - MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 3, + MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 4, PhoenixDatabaseMetaData.TTL + " " + PVarchar.INSTANCE.getSqlTypeName()); metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG, - MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 2, + MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 3, PhoenixDatabaseMetaData.ROW_KEY_MATCHER + " " + PVarbinary.INSTANCE.getSqlTypeName()); metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG, - MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 1, + MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 2, PhoenixDatabaseMetaData.IS_STRICT_TTL + " " + PBoolean.INSTANCE.getSqlTypeName()); metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG, - MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0, PhoenixDatabaseMetaData.INDEX_CONSISTENCY + " CHAR(1)"); + MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 - 1, + PhoenixDatabaseMetaData.INDEX_CONSISTENCY + " CHAR(1)"); + // No-op catalog schema-version anchor: advances the SYSTEM.CATALOG header timestamp to the + // new MIN so in-place SNAPSHOT clusters re-enter the upgrade path and pick up the new + // SYSTEM.TRANSFORM columns. Never populated or read. + metaConnection = addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_CATALOG, + MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0, + PhoenixDatabaseMetaData.UPGRADE_TS_ANCHOR_5_4_0 + " CHAR(1)"); // move TTL values stored in descriptor to SYSCAT TTL column. moveTTLFromHBaseLevelTTLToPhoenixLevelTTL(metaConnection); @@ -5301,8 +5308,20 @@ private PhoenixConnection upgradeSystemTransform(PhoenixConnection metaConnectio Map systemTableToSnapshotMap) throws SQLException { try (Statement statement = metaConnection.createStatement()) { statement.executeUpdate(getTransformDDL()); - } catch (TableAlreadyExistsException ignored) { - + } catch (NewerTableAlreadyExistsException ignored) { + } catch (TableAlreadyExistsException e) { + // This is the first-ever column add to SYSTEM.TRANSFORM, so take a snapshot before altering. + takeSnapshotOfSysTable(systemTableToSnapshotMap, e); + // addColumnsIfNotExists is idempotent, so call it unconditionally rather than gating on the + // table timestamp. A gate keyed on MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 is unreachable on + // SNAPSHOT clusters whose SYSTEM.TRANSFORM header already reached that timestamp without the + // columns, and would strand transform reads on a missing-column error. + metaConnection = + addColumnsIfNotExists(metaConnection, PhoenixDatabaseMetaData.SYSTEM_TRANSFORM_NAME, + MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0, + PhoenixDatabaseMetaData.PENDING_PARTIAL_PASS_UNTIL_TS + " " + + PLong.INSTANCE.getSqlTypeName() + ", " + PhoenixDatabaseMetaData.CUTOVER_TS + " " + + PLong.INSTANCE.getSqlTypeName()); } return metaConnection; } diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java index aa73c833145..f9385250cfe 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryConstants.java @@ -38,6 +38,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_QUALIFIER_COUNTER; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_SIZE; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CURRENT_VALUE; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CUTOVER_TS; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CYCLE_FLAG; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DATA_TABLE_NAME; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DATA_TYPE; @@ -99,6 +100,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PARTITION_ID; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PARTITION_START_KEY; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PARTITION_START_TIME; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PENDING_PARTIAL_PASS_UNTIL_TS; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PHOENIX_TTL; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PHOENIX_TTL_HWM; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PHYSICAL_NAME; @@ -172,6 +174,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_NAME; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_SEQUENCE; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.UPDATE_CACHE_FREQUENCY; +import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.UPGRADE_TS_ANCHOR_5_4_0; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.USER; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.USE_STATS_FOR_PARALLELIZATION; import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_CONSTANT; @@ -419,7 +422,8 @@ enum JoinType { + " BOOLEAN, \n" + SCHEMA_VERSION + " VARCHAR, \n" + EXTERNAL_SCHEMA_ID + " VARCHAR, \n" + STREAMING_TOPIC_NAME + " VARCHAR, \n" + INDEX_WHERE + " VARCHAR, \n" + CDC_INCLUDE_TABLE + " VARCHAR, \n" + TTL + " VARCHAR, \n" + ROW_KEY_MATCHER + " VARBINARY_ENCODED, \n" - + IS_STRICT_TTL + " BOOLEAN, \n" + INDEX_CONSISTENCY + " CHAR(1), \n" + + + IS_STRICT_TTL + " BOOLEAN, \n" + INDEX_CONSISTENCY + " CHAR(1), \n" + + UPGRADE_TS_ANCHOR_5_4_0 + " CHAR(1), \n" + // Column metadata (will be null for table row) DATA_TYPE + " INTEGER," + COLUMN_SIZE + " INTEGER," + DECIMAL_DIGITS + " INTEGER," + NULLABLE + " INTEGER," + ORDINAL_POSITION + " INTEGER," + SORT_ORDER + " INTEGER," + ARRAY_SIZE @@ -566,9 +570,10 @@ enum JoinType { TRANSFORM_STATUS + " VARCHAR NULL," + TRANSFORM_JOB_ID + " VARCHAR NULL," + TRANSFORM_RETRY_COUNT + " INTEGER NULL," + TRANSFORM_START_TS + " TIMESTAMP NULL," + TRANSFORM_LAST_STATE_TS + " TIMESTAMP NULL," + OLD_METADATA + " VARBINARY NULL,\n" - + NEW_METADATA + " VARCHAR NULL,\n" + TRANSFORM_FUNCTION + " VARCHAR NULL\n" + "CONSTRAINT " - + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + TENANT_ID + "," + TABLE_SCHEM + "," - + LOGICAL_TABLE_NAME + "))\n" + HConstants.VERSIONS + "=%s,\n" + + NEW_METADATA + " VARCHAR NULL,\n" + TRANSFORM_FUNCTION + " VARCHAR NULL,\n" + + PENDING_PARTIAL_PASS_UNTIL_TS + " BIGINT NULL,\n" + CUTOVER_TS + " BIGINT NULL\n" + + "CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + TENANT_ID + "," + TABLE_SCHEM + + "," + LOGICAL_TABLE_NAME + "))\n" + HConstants.VERSIONS + "=%s,\n" + ColumnFamilyDescriptorBuilder.KEEP_DELETED_CELLS + "=%s,\n" + ColumnFamilyDescriptorBuilder.TTL + "=" + TRANSFORM_TABLE_TTL + ",\n" + // 90 days TableDescriptorBuilder.SPLIT_POLICY + "='" + SYSTEM_TASK_SPLIT_POLICY_CLASSNAME + "',\n" diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java index effd7773ec2..dc100d19559 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java @@ -297,7 +297,11 @@ public static TransformType fromSerializedValue(int serializedValue) { } public static TransformType getPartialTransform(TransformType transformType) { - if (transformType == METADATA_TRANSFORM) { + // A full transform's partial variant is the partial type. Asking for the partial variant of a + // record that is already partial-type yields the partial type itself: this is what lets a + // failed partial pass be re-kicked (the record is already partial-type on retry) instead of + // being mistaken for "no partial pass needed" and completed early. + if (transformType == METADATA_TRANSFORM || transformType == METADATA_TRANSFORM_PARTIAL) { return METADATA_TRANSFORM_PARTIAL; } return null; @@ -326,6 +330,16 @@ public String toString() { return "PENDING_CUTOVER"; } }, + PENDING_PARTIAL_PASS { + public String toString() { + return "PENDING_PARTIAL_PASS"; + } + }, + PARTIAL_PASS_RUNNING { + public String toString() { + return "PARTIAL_PASS_RUNNING"; + } + }, COMPLETED { public String toString() { return "COMPLETED"; diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/SystemTransformRecord.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/SystemTransformRecord.java index 0b73bf341b2..9fb7e2bf1f6 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/SystemTransformRecord.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/SystemTransformRecord.java @@ -45,11 +45,14 @@ public class SystemTransformRecord { private final byte[] oldMetadata; private final String newMetadata; private final String transformFunction; + private final Long pendingPartialPassUntilTs; + private final Long cutoverTs; public SystemTransformRecord(PTable.TransformType transformType, String schemaName, String logicalTableName, String tenantId, String newPhysicalTableName, String logicalParentName, String transformStatus, String transformJobId, Integer transformRetryCount, Timestamp startTs, - Timestamp lastStateTs, byte[] oldMetadata, String newMetadata, String transformFunction) { + Timestamp lastStateTs, byte[] oldMetadata, String newMetadata, String transformFunction, + Long pendingPartialPassUntilTs, Long cutoverTs) { this.transformType = transformType; this.schemaName = schemaName; this.tenantId = tenantId; @@ -64,6 +67,8 @@ public SystemTransformRecord(PTable.TransformType transformType, String schemaNa this.oldMetadata = oldMetadata; this.newMetadata = newMetadata; this.transformFunction = transformFunction; + this.pendingPartialPassUntilTs = pendingPartialPassUntilTs; + this.cutoverTs = cutoverTs; } public String getString() { @@ -130,10 +135,20 @@ public String getTransformFunction() { return transformFunction; } + public Long getPendingPartialPassUntilTs() { + return pendingPartialPassUntilTs; + } + + public Long getCutoverTs() { + return cutoverTs; + } + public boolean isActive() { return (transformStatus.equals(PTable.TransformStatus.STARTED.name()) || transformStatus.equals(PTable.TransformStatus.CREATED.name()) - || transformStatus.equals(PTable.TransformStatus.PENDING_CUTOVER.name())); + || transformStatus.equals(PTable.TransformStatus.PENDING_CUTOVER.name()) + || transformStatus.equals(PTable.TransformStatus.PENDING_PARTIAL_PASS.name()) + || transformStatus.equals(PTable.TransformStatus.PARTIAL_PASS_RUNNING.name())); } @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = { "EI_EXPOSE_REP", "EI_EXPOSE_REP2" }, @@ -154,6 +169,8 @@ public static class SystemTransformBuilder { private byte[] oldMetadata; private String newMetadata; private String transformFunction; + private Long pendingPartialPassUntilTs; + private Long cutoverTs; public SystemTransformBuilder() { @@ -174,6 +191,8 @@ public SystemTransformBuilder(SystemTransformRecord systemTransformRecord) { this.setOldMetadata(systemTransformRecord.getOldMetadata()); this.setNewMetadata(systemTransformRecord.getNewMetadata()); this.setTransformFunction(systemTransformRecord.getTransformFunction()); + this.setPendingPartialPassUntilTs(systemTransformRecord.getPendingPartialPassUntilTs()); + this.setCutoverTs(systemTransformRecord.getCutoverTs()); } public SystemTransformBuilder setTransformType(PTable.TransformType transformType) { @@ -246,6 +265,16 @@ public SystemTransformBuilder setTransformFunction(String transformFunction) { return this; } + public SystemTransformBuilder setPendingPartialPassUntilTs(Long pendingPartialPassUntilTs) { + this.pendingPartialPassUntilTs = pendingPartialPassUntilTs; + return this; + } + + public SystemTransformBuilder setCutoverTs(Long cutoverTs) { + this.cutoverTs = cutoverTs; + return this; + } + public SystemTransformRecord build() { Timestamp lastTs = lastStateTs; if ( @@ -256,7 +285,8 @@ public SystemTransformRecord build() { } return new SystemTransformRecord(transformType, schemaName, logicalTableName, tenantId, newPhysicalTableName, logicalParentName, transformStatus, transformJobId, - transformRetryCount, startTs, lastTs, oldMetadata, newMetadata, transformFunction); + transformRetryCount, startTs, lastTs, oldMetadata, newMetadata, transformFunction, + pendingPartialPassUntilTs, cutoverTs); } public static SystemTransformRecord build(ResultSet resultSet) throws SQLException { @@ -276,6 +306,10 @@ public static SystemTransformRecord build(ResultSet resultSet) throws SQLExcepti builder.setOldMetadata(resultSet.getBytes(col++)); builder.setNewMetadata(resultSet.getString(col++)); builder.setTransformFunction(resultSet.getString(col++)); + long pendingPartialPassUntilTs = resultSet.getLong(col++); + builder.setPendingPartialPassUntilTs(resultSet.wasNull() ? null : pendingPartialPassUntilTs); + long cutoverTs = resultSet.getLong(col++); + builder.setCutoverTs(resultSet.wasNull() ? null : cutoverTs); return builder.build(); } diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/TransformClient.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/TransformClient.java index f37c414f5b7..fff43b52386 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/TransformClient.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/TransformClient.java @@ -82,8 +82,9 @@ public class TransformClient { + PhoenixDatabaseMetaData.TRANSFORM_START_TS + ", " + PhoenixDatabaseMetaData.TRANSFORM_LAST_STATE_TS + ", " + PhoenixDatabaseMetaData.OLD_METADATA + " , " + PhoenixDatabaseMetaData.NEW_METADATA + " , " - + PhoenixDatabaseMetaData.TRANSFORM_FUNCTION + " FROM " - + PhoenixDatabaseMetaData.SYSTEM_TRANSFORM_NAME; + + PhoenixDatabaseMetaData.TRANSFORM_FUNCTION + " , " + + PhoenixDatabaseMetaData.PENDING_PARTIAL_PASS_UNTIL_TS + " , " + + PhoenixDatabaseMetaData.CUTOVER_TS + " FROM " + PhoenixDatabaseMetaData.SYSTEM_TRANSFORM_NAME; public static SystemTransformRecord getTransformRecord(PName schema, PName logicalTableName, PName logicalParentName, PName tenantId, PhoenixConnection connection) throws SQLException { @@ -330,7 +331,9 @@ public static void upsertTransform(SystemTransformRecord systemTransformParams, + PhoenixDatabaseMetaData.TRANSFORM_START_TS + ", " + PhoenixDatabaseMetaData.TRANSFORM_LAST_STATE_TS + ", " + PhoenixDatabaseMetaData.OLD_METADATA + " , " + PhoenixDatabaseMetaData.NEW_METADATA + " , " - + PhoenixDatabaseMetaData.TRANSFORM_FUNCTION + " ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)")) { + + PhoenixDatabaseMetaData.TRANSFORM_FUNCTION + " , " + + PhoenixDatabaseMetaData.PENDING_PARTIAL_PASS_UNTIL_TS + " , " + + PhoenixDatabaseMetaData.CUTOVER_TS + " ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")) { int colNum = 1; if (systemTransformParams.getSchemaName() != null) { stmt.setString(colNum++, systemTransformParams.getSchemaName()); @@ -383,6 +386,16 @@ public static void upsertTransform(SystemTransformRecord systemTransformParams, } else { stmt.setNull(colNum++, Types.VARCHAR); } + if (systemTransformParams.getPendingPartialPassUntilTs() != null) { + stmt.setLong(colNum++, systemTransformParams.getPendingPartialPassUntilTs()); + } else { + stmt.setNull(colNum++, Types.BIGINT); + } + if (systemTransformParams.getCutoverTs() != null) { + stmt.setLong(colNum++, systemTransformParams.getCutoverTs()); + } else { + stmt.setNull(colNum++, Types.BIGINT); + } LOGGER.info("Adding transform type: " + systemTransformParams.getString()); stmt.execute(); } diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTask.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTask.java index bd808d5650a..e3bae8db915 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTask.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTask.java @@ -22,10 +22,13 @@ import static org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtilHelper.DEFAULT_TRANSFORM_MONITOR_ENABLED; import static org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtilHelper.TRANSFORM_MONITOR_ENABLED; +import java.sql.SQLException; +import java.sql.Timestamp; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.mapreduce.Cluster; import org.apache.hadoop.mapreduce.Job; +import org.apache.hadoop.mapreduce.JobID; import org.apache.phoenix.coprocessor.TaskRegionObserver; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.mapreduce.transform.TransformTool; @@ -35,6 +38,7 @@ import org.apache.phoenix.schema.task.Task; import org.apache.phoenix.schema.transform.SystemTransformRecord; import org.apache.phoenix.schema.transform.Transform; +import org.apache.phoenix.util.EnvironmentEdgeManager; import org.apache.phoenix.util.QueryUtil; import org.apache.phoenix.util.SchemaUtil; import org.slf4j.Logger; @@ -50,14 +54,59 @@ public class TransformMonitorTask extends BaseTask { public static final Logger LOGGER = LoggerFactory.getLogger(TransformMonitorTask.class); + // After the cutover pointer swap, clients may still hold a cached pointer to the old physical + // table until their update-cache-frequency window elapses. We wait for that window (with a small + // safety margin) before running the partial pass, so late writes routed to the old table are not + // stranded as unverified rows. The multiplier adds headroom over the raw cache frequency and the + // floor guarantees a minimum wait even when the cache frequency is very small or zero. + private static final double CACHE_FREQUENCY_SAFETY_MULTIPLIER = 1.10; + // 30 minutes + private static final long MIN_PARTIAL_PASS_WAIT_MS = 30L * 60L * 1000L; + // 24 hours. Upper bound on the wait. A table configured to never refresh its cache resolves its + // update-cache-frequency to Long.MAX_VALUE; scaling that and adding it to the current time would + // saturate and overflow into a negative (past) deadline, which would defeat the wait entirely. + // Clamping the wait to this ceiling keeps the persisted deadline a bounded, valid future + // timestamp while still deferring the partial pass long enough for clients to refresh. + private static final long MAX_PARTIAL_PASS_WAIT_MS = 24L * 60L * 60L * 1000L; + private static boolean isDisabled = false; - // Called from testong + // Called from testing @VisibleForTesting public static void disableTransformMonitorTask(boolean disabled) { isDisabled = disabled; } + /** + * Resolves the running MapReduce job for a given job id. Extracted behind an overridable seam so + * a test can inject a completed/failed job and exercise the PARTIAL_PASS_RUNNING branch's + * retries-exhausted -> FAILED transition deterministically, without submitting a real MR job + * that fails. The default implementation looks the job up on the real cluster. + */ + @VisibleForTesting + public interface JobLookup { + Job getJob(Configuration configuration, String jobId) throws Exception; + } + + private static JobLookup defaultJobLookup() { + return (configuration, jobId) -> { + Cluster cluster = new Cluster(configuration); + return cluster.getJob(JobID.forName(jobId)); + }; + } + + private static JobLookup jobLookup = defaultJobLookup(); + + @VisibleForTesting + public static void setJobLookupForTesting(JobLookup lookup) { + jobLookup = lookup; + } + + @VisibleForTesting + public static void resetJobLookupForTesting() { + jobLookup = defaultJobLookup(); + } + @Override public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) { Configuration conf = HBaseConfiguration.create(env.getConfiguration()); @@ -84,7 +133,7 @@ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) { if ( systemTransformRecord.getTransformStatus().equals(PTable.TransformStatus.CREATED.name()) ) { - LOGGER.info("Transform is created, starting the TransformTool ", tableName); + LOGGER.info("Transform is created, starting the TransformTool {}", tableName); // Kick a TransformTool run, it will already update transform record status and job id TransformTool transformTool = TransformTool.runTransformTool(systemTransformRecord, conf, false, null, null, false, false); @@ -97,44 +146,183 @@ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) { } else if ( systemTransformRecord.getTransformStatus().equals(PTable.TransformStatus.COMPLETED.name()) ) { - LOGGER.info("Transform is completed, TransformMonitor is done ", tableName); + LOGGER.info("Transform is completed, TransformMonitor is done {}", tableName); return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.SUCCESS, ""); } else if ( systemTransformRecord.getTransformStatus() .equals(PTable.TransformStatus.PENDING_CUTOVER.name()) && !PTable.TransformType.isPartialTransform(systemTransformRecord.getTransformType()) ) { - LOGGER.info("Transform is pending cutover ", tableName); + LOGGER.info("Transform is pending cutover {}", tableName); + // Persist the cutover instant (repair-floor anchor) durably before doCutover. doCutover + // commits the pointer swap durably, but this instant is otherwise only in the buffered + // PENDING_PARTIAL_PASS upsert below, so a crash in that gap would lose it and a re-entry + // would re-capture a later one, pushing the repair floor past the real cutover. A re-entry + // instead reuses the persisted value (resolveCutoverTs). + long cutoverTs = resolveCutoverTs(systemTransformRecord); + if (systemTransformRecord.getCutoverTs() == null) { + Transform.updateTransformRecord(conn, systemTransformRecord, + PTable.TransformStatus.PENDING_CUTOVER, + systemTransformRecord.getPendingPartialPassUntilTs(), cutoverTs); + conn.commit(); + } Transform.doCutover(conn, systemTransformRecord); PTable.TransformType partialTransform = PTable.TransformType.getPartialTransform(systemTransformRecord.getTransformType()); if (partialTransform != null) { - // Update transform to be partial - SystemTransformRecord.SystemTransformBuilder builder = - new SystemTransformRecord.SystemTransformBuilder(systemTransformRecord); - builder.setTransformType(partialTransform); - // Decrement retry count since TransformTool will increment it. Should we set it to 0? - builder.setTransformRetryCount(systemTransformRecord.getTransformRetryCount() - 1); - Transform.upsertTransform(builder.build(), conn); - - // Fix unverified rows. Running partial transform will make the transform status go back - // to started - long startFromTs = 0; - if (systemTransformRecord.getTransformLastStateTs() != null) { - startFromTs = systemTransformRecord.getTransformLastStateTs().getTime() - 1; - } - TransformTool.runTransformTool(systemTransformRecord, conf, true, startFromTs, null, true, - false); - - // In the future, if we are changing the PK structure, we need to run indextools as well + // After the pointer swap, wait for clients to refresh their cached physical-table pointer + // before running the partial pass. Persist the earliest time the wait may end and move to + // PENDING_PARTIAL_PASS; the partial pass is NOT kicked here. + long waitUntilTs = EnvironmentEdgeManager.currentTimeMillis() + + computePartialPassWaitMs(conn, systemTransformRecord); + // One INFO per transform recording the deferral deadline; the per-scan "still waiting" + // line is logged at DEBUG (the wait spans 30 min to 24 h at ~60 s scans) so the wait + // window is observable without flooding the log. + LOGGER.info( + "Cutover complete for {}; deferring the partial pass until ts {} so clients " + + "can refresh their cached physical-table pointer before it runs", + tableName, waitUntilTs); + Transform.updateTransformRecord(conn, systemTransformRecord, + PTable.TransformStatus.PENDING_PARTIAL_PASS, waitUntilTs, cutoverTs); } else { // No partial transform needed so, we update state of the transform - LOGGER.warn("No partial type of the transform is found. Completing the transform ", + LOGGER.warn("No partial type of the transform is found. Completing the transform {}", tableName); Transform.updateTransformRecord(conn, systemTransformRecord, PTable.TransformStatus.COMPLETED); } + } else if ( + systemTransformRecord.getTransformStatus() + .equals(PTable.TransformStatus.PENDING_PARTIAL_PASS.name()) + ) { + Long waitUntilTs = systemTransformRecord.getPendingPartialPassUntilTs(); + if (waitUntilTs != null && EnvironmentEdgeManager.currentTimeMillis() < waitUntilTs) { + // Still inside the client cache-refresh window; re-poll on the next scan. Logged at DEBUG + // because the monitor scans every ~60s while this wait can span 30 min to 24 h, so an + // INFO here would emit the same line hundreds of times per transform. + LOGGER.debug( + "Transform is pending partial pass, still waiting for cache refresh window {}", + tableName); + } else { + LOGGER.info("Transform wait window elapsed, starting the partial pass {}", tableName); + // Make the PARTIAL_PASS_RUNNING transition authoritative and committed BEFORE launching + // the partial pass. Committing here (rather than relying on the monitor's ServerTask + // commit at the end of run()) means a later monitor scan can observe + // PARTIAL_PASS_RUNNING, + // and the partial pass's own reducer-committed COMPLETED (written on a separate + // connection when the async MR job ends) is never clobbered by a stale buffered upsert. + // + // Clear the inherited job id on this transition. The record still carries the full-pass + // job id (the PENDING_PARTIAL_PASS record inherited it and neither the cutover transition + // nor the builder copy ctor clears it). That old job id points at the already-completed, + // successful full pass. If a monitor scan observes the record while still + // PARTIAL_PASS_RUNNING before the kicked partial pass has registered its own job id, the + // PARTIAL_PASS_RUNNING branch below would look up that stale job, see it successful, and + // drive the record straight to COMPLETED -- skipping the partial pass that repairs + // unverified rows. Nulling it here makes the `if (jobId != null)` guard correctly no-op + // until the partial pass has registered its own job id. + updateTransformRecordClearingJobId(conn, systemTransformRecord, + PTable.TransformStatus.PARTIAL_PASS_RUNNING, waitUntilTs); + conn.commit(); + // Re-read so the kick builds its upsert off the PARTIAL_PASS_RUNNING record rather than + // the pre-kick PENDING_PARTIAL_PASS state. + SystemTransformRecord runningRecord = Transform.getTransformRecord( + systemTransformRecord.getSchemaName(), systemTransformRecord.getLogicalTableName(), + null, systemTransformRecord.getTenantId(), conn); + // First partial-pass kick (not a retry): the initial partial pass must not consume retry + // budget. + kickPartialPass(conn, conf, runningRecord, tableName, false); + } + } else if ( + systemTransformRecord.getTransformStatus() + .equals(PTable.TransformStatus.PARTIAL_PASS_RUNNING.name()) + ) { + LOGGER.info("Partial pass is running, we will monitor {}", tableName); + // Monitor the partial-pass job to completion, then advance to COMPLETED. + String jobId = systemTransformRecord.getTransformJobId(); + // Defense-in-depth alongside the job-id clearing on the PENDING_PARTIAL_PASS -> + // PARTIAL_PASS_RUNNING transition: only a partial-type record may be driven to a terminal + // state by this branch. A record still carrying the full transform type has not yet had its + // partial pass registered (the kick flips the type to the partial variant before + // launching), + // so any job id it carries is the stale, already-successful full-pass job. Refusing to act + // on it here prevents a premature COMPLETED that would skip the partial pass; such a record + // is transient and is re-evaluated on the next scan once the kick has flipped the type. + if (PTable.TransformType.isPartialTransform(systemTransformRecord.getTransformType())) { + // A null job id here means no partial-pass job is currently registered. This is NOT a + // benign no-op: the pointer swap already happened, so the record must still reach a + // terminal state. It occurs when the initial partial-pass kick failed before + // TransformTool's STARTED transition (e.g. connection acquisition, index-table creation, + // or argument validation threw, so runTransformTool returned null without registering a + // job id and without throwing), leaving a committed (PARTIAL_PASS_RUNNING, partial-type, + // jobId=null) row. A job id that resolves to null (aged out of the job-history server, + // resource-manager restart, etc.) is likewise unconfirmable. In every one of these cases + // the pass cannot be confirmed successful, so it is routed through the same + // retry-budgeted path as an outright failed job below -- never left to no-op forever. + Job job = jobId != null ? jobLookup.getJob(configuration, jobId) : null; + if (job != null && !job.isComplete()) { + // Partial pass is still running; re-evaluate on the next monitor scan. + LOGGER.info("Partial pass job is still running, we will keep monitoring {}", tableName); + } else if (job != null && job.isSuccessful()) { + Transform.updateTransformRecord(conn, systemTransformRecord, + PTable.TransformStatus.COMPLETED); + } else { + // The partial pass could not be confirmed successful: no job id is registered (the + // initial kick failed before its STARTED transition), the job completed unsuccessfully, + // or the job id could not be resolved at all (aged out of the job-history server, + // resource-manager restart, etc.). All are treated as a failed partial pass and routed + // through the retry-budgeted path. Returning SKIPPED here instead would strand the + // transform: a SKIPPED result leaves the self-healing task in the STARTED state, and + // the monitor scan re-picks up only CREATED/RETRY tasks, so PARTIAL_PASS_RUNNING would + // never be re-evaluated and the already-pointer-swapped table would never get its + // repairing partial pass. + if (jobId == null) { + LOGGER.warn("No partial-pass job is registered for {}; the initial kick did not " + + "register one. Treating as a failed partial pass and retrying.", tableName); + } else if (job == null) { + LOGGER.warn(String.format( + "Transform job with Id=%s is not found; treating as a failed partial pass", jobId)); + } + // Account for a pre-STARTED failure. TransformTool increments the retry count only on + // its STARTED transition, so a kick that failed before STARTED (jobId == null) was + // never counted. Left uncounted, a deterministically pre-STARTED-failing partial pass + // (e.g. connection acquisition or index-table creation throws every attempt) would + // resubmit forever and never reach the retries-exhausted -> FAILED transition below. + // A jobId that merely resolved to null (job aged out of the history server) already + // reached STARTED and was counted there, so it is deliberately not re-counted here. + SystemTransformRecord failedRecord = systemTransformRecord; + if (jobId == null) { + SystemTransformRecord.SystemTransformBuilder builder = + new SystemTransformRecord.SystemTransformBuilder(systemTransformRecord); + builder.setTransformRetryCount(systemTransformRecord.getTransformRetryCount() + 1); + failedRecord = builder.build(); + } + int maxRetryCount = + configuration.getInt(TRANSFORM_RETRY_COUNT_VALUE, DEFAULT_TRANSFORM_RETRY_COUNT); + if (failedRecord.getTransformRetryCount() < maxRetryCount) { + // Retry the partial pass. A kick that reaches STARTED has its count strictly + // increased by TransformTool's STARTED transition; a pre-STARTED failure has it + // increased by the block above. Either way the count strictly increases per failed + // attempt, so the retries-exhausted -> FAILED transition below stays reachable and a + // repeatedly-failing partial pass cannot resubmit forever. + kickPartialPass(conn, conf, failedRecord, tableName, true); + } else { + // Retries are exhausted. Move to a terminal FAILED state so the record does not + // re-enter PARTIAL_PASS_RUNNING forever on subsequent scans. + LOGGER + .error("Partial pass failed and retries are exhausted. Marking transform as failed " + + tableName); + Transform.updateTransformRecord(conn, failedRecord, PTable.TransformStatus.FAILED); + } + } + } else { + // Defensive guard: no code path produces a full-type PARTIAL_PASS_RUNNING record, since + // the transition to PARTIAL_PASS_RUNNING commits the partial type. Reachable only by an + // externally seeded record; log without acting on the stale full-pass job id. + LOGGER.info("Partial pass not yet registered as partial-type, will re-evaluate {}", + tableName); + } } else if ( systemTransformRecord.getTransformStatus().equals(PTable.TransformStatus.STARTED.name()) || (systemTransformRecord.getTransformStatus() @@ -149,9 +337,7 @@ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) { // Monitor the job of transform tool and decide to retry String jobId = systemTransformRecord.getTransformJobId(); if (jobId != null) { - Cluster cluster = new Cluster(configuration); - - Job job = cluster.getJob(org.apache.hadoop.mapreduce.JobID.forName(jobId)); + Job job = jobLookup.getJob(configuration, jobId); if (job == null) { LOGGER.warn(String.format("Transform job with Id=%s is not found", jobId)); return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.SKIPPED, @@ -212,6 +398,204 @@ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) { } } + /** + * Updates the transform record to the given status while clearing any inherited job id. Used on + * the PENDING_PARTIAL_PASS -> PARTIAL_PASS_RUNNING transition so the record does not carry the + * completed full-pass job id into the PARTIAL_PASS_RUNNING monitoring branch. Mirrors + * {@link Transform#updateTransformRecord} (bumps last-state-ts, preserves the wait timestamp) but + * additionally sets the job id to null; the kicked partial pass registers its own job id when it + * launches. It also flips the transform type to the partial variant so the committed + * PARTIAL_PASS_RUNNING record is restart-safe. + */ + private void updateTransformRecordClearingJobId(PhoenixConnection conn, + SystemTransformRecord systemTransformRecord, PTable.TransformStatus newStatus, + Long pendingPartialPassUntilTs) throws SQLException { + SystemTransformRecord.SystemTransformBuilder builder = + new SystemTransformRecord.SystemTransformBuilder(systemTransformRecord); + builder.setTransformStatus(newStatus.name()); + builder.setTransformJobId(null); + builder.setLastStateTs(new Timestamp(EnvironmentEdgeManager.currentTimeMillis())); + builder.setPendingPartialPassUntilTs(pendingPartialPassUntilTs); + // Flip to the partial transform type in this same committed transition so the record is + // restart-safe: a crash after this commit but before the partial-pass kick leaves a + // partial-type, null-job-id record, which the PARTIAL_PASS_RUNNING branch routes through the + // retry path instead of the log-only branch that never re-kicks. + PTable.TransformType partialTransform = + PTable.TransformType.getPartialTransform(systemTransformRecord.getTransformType()); + if (partialTransform != null) { + builder.setTransformType(partialTransform); + } + Transform.upsertTransform(builder.build(), conn); + } + + /** + * Computes how long to wait after cutover before running the partial pass. The wait is the + * logical (parent) table's update-cache-frequency scaled by a safety margin, clamped to + * [{@link #MIN_PARTIAL_PASS_WAIT_MS}, {@link #MAX_PARTIAL_PASS_WAIT_MS}] so a small or zero cache + * frequency still yields a meaningful wait and a table that never refreshes its cache (whose + * update-cache-frequency resolves to Long.MAX_VALUE) does not produce an unbounded wait that + * would overflow the deadline arithmetic in the caller. The returned value is always a small, + * positive number of milliseconds, so adding it to the current time cannot overflow. + */ + private long computePartialPassWaitMs(PhoenixConnection conn, + SystemTransformRecord systemTransformRecord) { + long updateCacheFrequency = 0; + try { + String logicalTableName = SchemaUtil.getTableName(systemTransformRecord.getSchemaName(), + systemTransformRecord.getLogicalTableName()); + PTable logicalTable = conn.getTable(systemTransformRecord.getTenantId(), logicalTableName); + updateCacheFrequency = logicalTable.getUpdateCacheFrequency(); + } catch (Exception e) { + LOGGER.warn("Could not resolve update cache frequency for the logical table; " + + "falling back to the minimum partial-pass wait", e); + } + return boundedPartialPassWaitMs(updateCacheFrequency); + } + + /** + * Clamps and scales a raw update-cache-frequency into a bounded partial-pass wait. Extracted as a + * pure function so the overflow-safety of the arithmetic can be unit-tested without a cluster. + * The raw frequency is clamped to the ceiling BEFORE scaling so the multiplication cannot + * saturate (a never-refreshed table reports Long.MAX_VALUE); the scaled result is then clamped to + * the [{@link #MIN_PARTIAL_PASS_WAIT_MS}, {@link #MAX_PARTIAL_PASS_WAIT_MS}] window. The return + * is always in that window -- small, positive, and safe to add to the current time without + * overflow. + */ + @VisibleForTesting + static long boundedPartialPassWaitMs(long updateCacheFrequency) { + long bounded = Math.min(updateCacheFrequency, MAX_PARTIAL_PASS_WAIT_MS); + long scaled = (long) (bounded * CACHE_FREQUENCY_SAFETY_MULTIPLIER); + return Math.min(Math.max(scaled, MIN_PARTIAL_PASS_WAIT_MS), MAX_PARTIAL_PASS_WAIT_MS); + } + + /** + * Resolves the cutover instant that anchors the partial-pass repair floor + * ({@link #repairScanFloor}). A first run captures the current time -- taken before the pointer + * swap, the most conservative floor -- while a run re-entering the PENDING_CUTOVER handling after + * a crash reuses the instant the prior run already persisted, so the floor cannot drift past the + * real cutover. The PENDING_CUTOVER branch of {@link #run} persists this instant durably before + * the swap so it survives such a crash. + */ + @VisibleForTesting + static long resolveCutoverTs(SystemTransformRecord record) { + return record.getCutoverTs() != null + ? record.getCutoverTs() + : EnvironmentEdgeManager.currentTimeMillis(); + } + + /** + * Repair-scan lower bound for the partial pass. Derived from the cutover instant (minus one, so + * the floor is inclusive of writes stamped exactly at cutover) so the pass re-verifies every row + * written to the old pointer during {@code [cutover, cutover + waitWindow]}. A floor derived from + * the post-wait {@code lastStateTs} would sit past that window and strand those rows. Falls back + * to {@code lastStateTs} only for records that predate the CUTOVER_TS column, and to 0 (full + * scan) when neither is set. The cutover instant is persisted durably before the swap and reused + * on crash re-entry (see {@link #resolveCutoverTs}) so it never drifts later than the real + * cutover. + */ + @VisibleForTesting + static long repairScanFloor(SystemTransformRecord record) { + if (record.getCutoverTs() != null) { + return record.getCutoverTs() - 1; + } + if (record.getTransformLastStateTs() != null) { + return record.getTransformLastStateTs().getTime() - 1; + } + return 0; + } + + /** + * Kicks the partial-pass TransformTool run that fixes unverified rows on the new physical table. + * This preserves the partial-pass invocation that previously ran inline during cutover. The + * partial-type marker is committed before launching the tool; the tool is launched asynchronously + * (TransformTool submits the MR job and returns without blocking). + *

+ * The running partial pass must be observable in status PARTIAL_PASS_RUNNING so that the + * PARTIAL_PASS_RUNNING branch of {@link #run} -- not the STARTED branch -- monitors it and can + * drive it to a terminal FAILED state once retries are exhausted. TransformTool.runTransform + * unconditionally moves the record to STARTED and registers the partial-pass job id under + * STARTED; left as-is, a running partial pass would live in STARTED, whose retries-exhausted case + * does nothing, stranding a permanently-failing partial pass. To prevent that, after the async + * submit returns we re-assert PARTIAL_PASS_RUNNING while preserving the just-registered + * partial-pass job id, so the PARTIAL_PASS_RUNNING branch owns the running partial pass. We only + * re-assert when the record is still STARTED: if the (background) job already finished and its + * reducer committed a terminal status, re-asserting is skipped so the reducer-committed COMPLETED + * is never clobbered. + *

+ * {@code isRetry} governs retry-count accounting. Launching the partial pass runs it through + * TransformTool.runTransform, whose STARTED transition unconditionally increments the retry + * count. For the FIRST partial-pass kick that increment is spurious -- the initial partial pass + * is not a retry and must not consume retry budget -- so we compensate with a matching decrement + * ({@code isRetry == false}). For a genuine RETRY after a failed partial pass ({@code isRetry == + * true}) we let the increment stand, so the retry count strictly increases and the + * retries-exhausted -> terminal FAILED transition in the PARTIAL_PASS_RUNNING branch is + * reachable. Cancelling the increment on the retry path would pin the count and resubmit a + * deterministically-failing partial pass on every monitor tick forever. + */ + private void kickPartialPass(PhoenixConnection conn, Configuration conf, + SystemTransformRecord systemTransformRecord, String tableName, boolean isRetry) + throws Exception { + PTable.TransformType partialTransform = + PTable.TransformType.getPartialTransform(systemTransformRecord.getTransformType()); + if (partialTransform == null) { + LOGGER.warn("No partial type of the transform is found. Completing the transform {}", + tableName); + Transform.updateTransformRecord(conn, systemTransformRecord, + PTable.TransformStatus.COMPLETED); + return; + } + // Update transform to be partial + SystemTransformRecord.SystemTransformBuilder builder = + new SystemTransformRecord.SystemTransformBuilder(systemTransformRecord); + builder.setTransformType(partialTransform); + if (!isRetry) { + // First (non-retry) partial-pass kick: TransformTool's STARTED transition will increment the + // retry count, but the initial partial pass must not consume retry budget, so pre-decrement + // to net zero. On the retry path we deliberately skip this so the count strictly increases + // and retries-exhausted -> FAILED remains reachable. + builder.setTransformRetryCount(systemTransformRecord.getTransformRetryCount() - 1); + } + SystemTransformRecord partialRecord = builder.build(); + Transform.upsertTransform(partialRecord, conn); + // Commit the partial-type marker before launching the async tool run so the tool (which runs on + // a separate connection) sees committed state and the monitor holds no stale buffered upsert + // that could later overwrite the reducer-committed COMPLETED. + conn.commit(); + + // Fix unverified rows. TransformTool moves the record to STARTED, submits the MR job + // asynchronously, registers the partial-pass job id, and returns; the reducer advances the + // record to COMPLETED when the job finishes successfully. + // Derive the repair-scan lower bound from the cutover instant, not the post-wait lastStateTs + // (see repairScanFloor), so writes to the old pointer during [cutover, cutover + waitWindow] + // are re-verified rather than stranded. + long startFromTs = repairScanFloor(partialRecord); + TransformTool.runTransformTool(partialRecord, conf, true, startFromTs, null, true, false); + + // Re-assert PARTIAL_PASS_RUNNING so the running partial pass is monitored by the + // PARTIAL_PASS_RUNNING branch (which handles job failure, retry, and retries-exhausted -> + // FAILED) rather than the STARTED branch (which has no terminal transition for a + // repeatedly-failing partial pass and would strand it forever). The partial-pass job id that + // TransformTool just registered is preserved. We re-assert only if the record is still STARTED + // to narrow the window in which a background job that already finished and had its reducer + // commit a terminal status gets clobbered. This is a read-then-write with no lock, so a narrow + // TOCTOU window remains: the reducer could commit COMPLETED between this read and the upsert + // below. That is self-correcting -- the next monitor scan looks the (successful) job up and + // re-drives the record to COMPLETED -- so the worst case is one extra scan, not a stranded + // transform. + SystemTransformRecord afterLaunch = Transform.getTransformRecord(partialRecord.getSchemaName(), + partialRecord.getLogicalTableName(), partialRecord.getLogicalParentName(), + partialRecord.getTenantId(), conn); + if ( + afterLaunch != null + && PTable.TransformStatus.STARTED.name().equals(afterLaunch.getTransformStatus()) + ) { + Transform.updateTransformRecord(conn, afterLaunch, + PTable.TransformStatus.PARTIAL_PASS_RUNNING); + conn.commit(); + } + // In the future, if we are changing the PK structure, we need to run indextools as well + } + @Override public TaskRegionObserver.TaskResult checkCurrentResult(Task.TaskRecord taskRecord) throws Exception { diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/transform/TransformTool.java b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/transform/TransformTool.java index daca9a04616..860450b067c 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/transform/TransformTool.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/transform/TransformTool.java @@ -833,9 +833,21 @@ public void abortTransform() throws Exception { public void pauseTransform() throws Exception { SystemTransformRecord transformRecord = getTransformRecord(connection.unwrap(PhoenixConnection.class)); - if (transformRecord.getTransformStatus().equals(PTable.TransformStatus.COMPLETED.name())) { + String status = transformRecord.getTransformStatus(); + if (status.equals(PTable.TransformStatus.COMPLETED.name())) { throw new IllegalStateException("A completed transform cannot be paused"); } + // A transform that has already cut over must not be paused: resume re-runs a FULL transform + // (resumeTransform -> runTransform), but after cutover the physical-table pointer has already + // been swapped, so a full re-run would operate against the swapped-in table. These post-cutover + // partial-pass states are driven to completion by the TransformMonitor, not by pause/resume. + if ( + status.equals(PTable.TransformStatus.PENDING_PARTIAL_PASS.name()) + || status.equals(PTable.TransformStatus.PARTIAL_PASS_RUNNING.name()) + ) { + throw new IllegalStateException( + "A transform that has already cut over cannot be paused; it is in state " + status); + } updateTransformRecord(connection.unwrap(PhoenixConnection.class), PTable.TransformStatus.PAUSED); diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/schema/transform/Transform.java b/phoenix-core-server/src/main/java/org/apache/phoenix/schema/transform/Transform.java index 86e3233831c..e5d3da60957 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/schema/transform/Transform.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/schema/transform/Transform.java @@ -130,6 +130,65 @@ public static void removeTransformRecord(SystemTransformRecord transformRecord, .execute(); } + /** + * Tears down the base-table TRANSFORMING_NEW_TABLE link that drives dual-write to the new + * physical table, mirroring the base-table link installed in TransformClient.addTransform. The + * DELETE is left uncommitted so the caller can batch it with the base-table pointer swap into a + * single commit, letting one client cache-invalidation cycle propagate both the pointer swap and + * the dual-write shutoff for the base table. + *

+ * Child-view links are torn down separately by the caller ({@link #doCutover}), one per view, + * folded into the same {@code MUTATE_BATCH_SIZE}-bounded loop that swaps the view rows. A base + * table can have millions of views, so deleting every view link into a single commit would build + * an unbounded mutation set (risking client-side OOM and an oversized commit); batching per view + * keeps each commit bounded while still pairing each view's link teardown with its own swap in + * one cache-invalidation cycle. Currently invoked only by doCutover. + */ + public static void disableDualWrite(PhoenixConnection connection, + SystemTransformRecord systemTransformRecord) throws SQLException { + String tenantId = systemTransformRecord.getTenantId(); + String schema = systemTransformRecord.getSchemaName(); + String tableName = systemTransformRecord.getLogicalTableName(); + // The link row uses the full new physical table name as its COLUMN_FAMILY, matching the value + // bound when the link was installed. + String newPhysicalName = systemTransformRecord.getNewPhysicalTableName(); + deleteTransformingNewTableLink(connection, tenantId, schema, tableName, newPhysicalName); + } + + /** + * Deletes a single TRANSFORMING_NEW_TABLE link row. The DELETE is left uncommitted so the caller + * can batch it with related mutations. The COLUMN_FAMILY match uses the full new physical table + * name, matching the value bound when the link was installed in TransformClient.addTransform. + * NULL tenant/schema are matched with IS NULL to mirror the install's NULL handling. + */ + private static void deleteTransformingNewTableLink(PhoenixConnection connection, String tenantId, + String schema, String tableName, String newPhysicalName) throws SQLException { + // Constrain COLUMN_NAME IS NULL so the predicate resolves to the single table-level link row + // (a link row carries no COLUMN_NAME) instead of range-scanning every column row of the table. + // COLUMN_NAME is a PK column between TABLE_NAME and COLUMN_FAMILY, so leaving it unbound forces + // a range scan that matches COLUMN_FAMILY/LINK_TYPE as non-PK filters. An unset PK VARCHAR + // matches IS NULL, so this is a point delete, not a silent no-op. + String deleteLink = "DELETE FROM " + PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME + " WHERE " + + PhoenixDatabaseMetaData.TENANT_ID + (tenantId == null ? " IS NULL" : " = ?") + " AND " + + PhoenixDatabaseMetaData.TABLE_SCHEM + (schema == null ? " IS NULL" : " = ?") + " AND " + + PhoenixDatabaseMetaData.TABLE_NAME + " = ? AND " + PhoenixDatabaseMetaData.COLUMN_NAME + + " IS NULL AND " + PhoenixDatabaseMetaData.COLUMN_FAMILY + " = ? AND " + + PhoenixDatabaseMetaData.LINK_TYPE + " = ?"; + try (PreparedStatement stmt = connection.prepareStatement(deleteLink)) { + int param = 0; + if (tenantId != null) { + stmt.setString(++param, tenantId); + } + if (schema != null) { + stmt.setString(++param, schema); + } + stmt.setString(++param, tableName); + stmt.setString(++param, newPhysicalName); + stmt.setByte(++param, PTable.LinkType.TRANSFORMING_NEW_TABLE.getSerializedValue()); + stmt.execute(); + } + } + /** * Disable caching re-design if you use Online Data Format Change since the cutover logic is * currently incompatible and clients may not learn about the physical table change. See @@ -211,6 +270,13 @@ public static void doCutover(PhoenixConnection connection, columnMap); } } + // Tear down the base-table dual-write link in the same commit as the base-table pointer swap + // so a single client cache-invalidation cycle propagates both the physical table pointer + // change and the dual-write shutoff. Per-view link teardown is folded into the batched view + // loop below (a base table can have millions of views, so deleting every view link here would + // build an unbounded commit). + String newPhysicalName = systemTransformRecord.getNewPhysicalTableName(); + disableDualWrite(connection.unwrap(PhoenixConnection.class), systemTransformRecord); connection.commit(); // We can have millions of views. We need to send it in batches @@ -240,6 +306,19 @@ public static void doCutover(PhoenixConnection connection, } stmt.execute(); } + // Tear down this view's TRANSFORMING_NEW_TABLE link in the same batch as its pointer swap. + // addTransform installs a link on every child-view row; doGetTable attaches the + // transforming + // new table per-row from link presence (not gated by isActive), so a surviving view link + // would keep dual-write alive for view-routed writes after cutover. + String viewTenantId = view.getTenantId() == null || view.getTenantId().length == 0 + ? null + : Bytes.toString(view.getTenantId()); + String viewSchema = view.getSchemaName() == null || view.getSchemaName().length == 0 + ? null + : Bytes.toString(view.getSchemaName()); + deleteTransformingNewTableLink(connection.unwrap(PhoenixConnection.class), viewTenantId, + viewSchema, Bytes.toString(view.getTableName()), newPhysicalName); viewsToUpdateCache.add(view); batchSize++; if (batchSize >= maxBatchSize) { @@ -358,10 +437,26 @@ public static void completeTransform(Connection connection, Configuration config public static void updateTransformRecord(PhoenixConnection connection, SystemTransformRecord transformRecord, PTable.TransformStatus newStatus) throws SQLException { + updateTransformRecord(connection, transformRecord, newStatus, + transformRecord.getPendingPartialPassUntilTs()); + } + + public static void updateTransformRecord(PhoenixConnection connection, + SystemTransformRecord transformRecord, PTable.TransformStatus newStatus, + Long pendingPartialPassUntilTs) throws SQLException { + updateTransformRecord(connection, transformRecord, newStatus, pendingPartialPassUntilTs, + transformRecord.getCutoverTs()); + } + + public static void updateTransformRecord(PhoenixConnection connection, + SystemTransformRecord transformRecord, PTable.TransformStatus newStatus, + Long pendingPartialPassUntilTs, Long cutoverTs) throws SQLException { SystemTransformRecord.SystemTransformBuilder builder = new SystemTransformRecord.SystemTransformBuilder(transformRecord); builder.setTransformStatus(newStatus.name()); builder.setLastStateTs(new Timestamp(EnvironmentEdgeManager.currentTimeMillis())); + builder.setPendingPartialPassUntilTs(pendingPartialPassUntilTs); + builder.setCutoverTs(cutoverTs); if (newStatus == PTable.TransformStatus.STARTED) { builder.setTransformRetryCount(transformRecord.getTransformRetryCount() + 1); } diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/CutoverLifecycleIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/CutoverLifecycleIT.java new file mode 100644 index 00000000000..7a7249be3ac --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/CutoverLifecycleIT.java @@ -0,0 +1,1057 @@ +/* + * 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.transform; + +import static org.apache.phoenix.query.QueryConstants.UNVERIFIED_BYTES; +import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.util.List; +import java.util.Properties; +import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; +import org.apache.hadoop.mapreduce.Job; +import org.apache.phoenix.coprocessor.TaskRegionObserver; +import org.apache.phoenix.coprocessor.tasks.TransformMonitorTask; +import org.apache.phoenix.end2end.ParallelStatsDisabledIT; +import org.apache.phoenix.end2end.ParallelStatsDisabledTest; +import org.apache.phoenix.jdbc.PhoenixConnection; +import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData; +import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.query.QueryServicesOptions; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.task.ServerTask; +import org.apache.phoenix.schema.task.SystemTaskParams; +import org.apache.phoenix.schema.task.Task; +import org.apache.phoenix.schema.transform.SystemTransformRecord; +import org.apache.phoenix.schema.transform.Transform; +import org.apache.phoenix.util.EnvironmentEdge; +import org.apache.phoenix.util.EnvironmentEdgeManager; +import org.apache.phoenix.util.PropertiesUtil; +import org.apache.phoenix.util.SchemaUtil; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.mockito.Mockito; + +/** + * Integration tests for the cutover lifecycle: after the physical-table pointer swap the transform + * monitor waits for clients to refresh their cached pointer (status PENDING_PARTIAL_PASS) before + * running the partial pass (status PARTIAL_PASS_RUNNING) and finally completing. The wait uses an + * injectable clock so the tests never sleep for the real wait window. + */ +@Category(ParallelStatsDisabledTest.class) +public class CutoverLifecycleIT extends ParallelStatsDisabledIT { + + private static RegionCoprocessorEnvironment taskRegionEnvironment; + + private final Properties testProps = PropertiesUtil.deepCopy(TEST_PROPERTIES); + + public CutoverLifecycleIT() throws IOException, InterruptedException { + testProps.put(QueryServices.DEFAULT_IMMUTABLE_STORAGE_SCHEME_ATTRIB, "ONE_CELL_PER_COLUMN"); + testProps.put(QueryServices.DEFAULT_COLUMN_ENCODED_BYTES_ATRRIB, "0"); + + taskRegionEnvironment = (RegionCoprocessorEnvironment) getUtility() + .getRSForFirstRegionInTable(PhoenixDatabaseMetaData.SYSTEM_TASK_HBASE_TABLE_NAME) + .getRegions(PhoenixDatabaseMetaData.SYSTEM_TASK_HBASE_TABLE_NAME).get(0).getCoprocessorHost() + .findCoprocessorEnvironment(TaskRegionObserver.class.getName()); + } + + @Before + public void setupTest() throws Exception { + try (Connection conn = DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + conn.createStatement() + .execute("DELETE FROM " + PhoenixDatabaseMetaData.SYSTEM_TRANSFORM_NAME); + conn.createStatement().execute("DELETE FROM " + PhoenixDatabaseMetaData.SYSTEM_TASK_NAME); + } + } + + @After + public void tearDownTest() { + EnvironmentEdgeManager.reset(); + TransformMonitorTask.resetJobLookupForTesting(); + } + + /** + * Advances the monitor clock so it is comfortably past the persisted wait deadline, letting the + * PENDING_PARTIAL_PASS branch proceed without waiting for the real 30-minute floor. + */ + private static class AdvancingClock extends EnvironmentEdge { + private long value = System.currentTimeMillis(); + + @Override + public long currentTime() { + return value; + } + + void setValue(long millis) { + value = millis; + } + } + + private void runMonitorOnce() { + TaskRegionObserver.SelfHealingTask task = new TaskRegionObserver.SelfHealingTask( + taskRegionEnvironment, QueryServicesOptions.DEFAULT_TASK_HANDLING_MAX_INTERVAL_MS); + task.run(); + } + + private SystemTransformRecord fetch(PhoenixConnection conn, String schemaName, String tableName, + String parentName) throws SQLException { + return Transform.getTransformRecord(schemaName, tableName, parentName, null, conn); + } + + /** + * Drives the monitor, advancing the injected clock as needed, until the transform reaches the + * requested status or we run out of attempts. + */ + private SystemTransformRecord driveMonitorToStatus(PhoenixConnection conn, String schemaName, + String tableName, String parentName, PTable.TransformStatus target, AdvancingClock clock) + throws Exception { + for (int i = 0; i < 60; i++) { + SystemTransformRecord record = fetch(conn, schemaName, tableName, parentName); + if (record != null && target.name().equals(record.getTransformStatus())) { + return record; + } + // If we are inside the wait window, jump the clock past the deadline so the monitor advances. + if ( + record != null + && PTable.TransformStatus.PENDING_PARTIAL_PASS.name().equals(record.getTransformStatus()) + && record.getPendingPartialPassUntilTs() != null + ) { + clock.setValue(record.getPendingPartialPassUntilTs() + 1); + } + runMonitorOnce(); + Thread.sleep(200); + } + SystemTransformRecord record = fetch(conn, schemaName, tableName, parentName); + fail("Ran out of attempts waiting for transform status " + target + " but it was " + + (record == null ? "" : record.getTransformStatus())); + return null; + } + + private long countLinkRows(Connection conn, String schemaName, String tableName) + throws SQLException { + String sql = "SELECT COUNT(*) FROM " + PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME + " WHERE " + + PhoenixDatabaseMetaData.TABLE_SCHEM + " = ? AND " + PhoenixDatabaseMetaData.TABLE_NAME + + " = ? AND " + PhoenixDatabaseMetaData.LINK_TYPE + " = ?"; + try (PreparedStatement stmt = conn.prepareStatement(sql)) { + stmt.setString(1, schemaName); + stmt.setString(2, tableName); + stmt.setByte(3, PTable.LinkType.TRANSFORMING_NEW_TABLE.getSerializedValue()); + ResultSet rs = stmt.executeQuery(); + rs.next(); + return rs.getLong(1); + } + } + + private long countUnverified(Connection conn, String physicalTableFullName) throws Exception { + return org.apache.phoenix.end2end.index.ImmutableIndexExtendedIT + .getRowCountForEmptyColValue(conn, physicalTableFullName, UNVERIFIED_BYTES); + } + + /** + * Counts every TRANSFORMING_NEW_TABLE link row that points at the given new physical table, + * regardless of which logical entity (base table or child view) owns the row. The link's + * COLUMN_FAMILY holds the full new physical table name, so this catches base-table and view links + * alike. + */ + private long countLinksToNewPhysicalTable(Connection conn, String newPhysicalFullName) + throws SQLException { + String sql = "SELECT COUNT(*) FROM " + PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME + " WHERE " + + PhoenixDatabaseMetaData.COLUMN_FAMILY + " = ? AND " + PhoenixDatabaseMetaData.LINK_TYPE + + " = ?"; + try (PreparedStatement stmt = conn.prepareStatement(sql)) { + stmt.setString(1, newPhysicalFullName); + stmt.setByte(2, PTable.LinkType.TRANSFORMING_NEW_TABLE.getSerializedValue()); + ResultSet rs = stmt.executeQuery(); + rs.next(); + return rs.getLong(1); + } + } + + private void runCutoverLifecycle(boolean createIndex, boolean isImmutable) throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + String schemaName = generateUniqueName(); + String dataTableName = "TBL_" + generateUniqueName(); + String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName); + String newTableName = dataTableName + "_1"; + String indexName = "IDX_" + generateUniqueName(); + String createIndexStmt = "CREATE INDEX %s ON " + dataTableFullName + " (NAME) INCLUDE (ZIP) "; + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + int numOfRows = 10; + TransformToolIT.createTableAndUpsertRows(conn, dataTableFullName, numOfRows, + isImmutable ? " IMMUTABLE_ROWS=true" : ""); + if (createIndex) { + conn.createStatement().execute(String.format(createIndexStmt, indexName)); + } + + // Kick off the transform. The monitor task is registered as part of the ALTER. + conn.createStatement().execute("ALTER TABLE " + dataTableFullName + + " SET IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + + SystemTransformRecord record = fetch(conn, schemaName, dataTableName, null); + assertNotNull(record); + + List taskRecordList = Task.queryTaskTable(conn, null); + assertEquals(1, taskRecordList.size()); + assertEquals(PTable.TaskType.TRANSFORM_MONITOR, taskRecordList.get(0).getTaskType()); + + // Drive to PENDING_PARTIAL_PASS: the pointer swap has happened and the monitor is waiting. + SystemTransformRecord pendingPartial = driveMonitorToStatus(conn, schemaName, dataTableName, + null, PTable.TransformStatus.PENDING_PARTIAL_PASS, clock); + + // The wait deadline is set and lies in the future relative to when it was computed. + Long untilTs = pendingPartial.getPendingPartialPassUntilTs(); + assertNotNull("PENDING_PARTIAL_PASS must persist a wait deadline", untilTs); + + // The pointer swap is already visible and the dual-write link row is gone in the same cache + // generation (the DELETE was committed together with the pointer swap). + PTable swappedTable = conn.getTableNoCache(dataTableFullName); + assertEquals(newTableName, swappedTable.getPhysicalName(true).getString()); + assertEquals("TRANSFORMING_NEW_TABLE link must be deleted at cutover", 0, + countLinkRows(conn, schemaName, dataTableName)); + + // While the clock is still before the deadline, the monitor must NOT advance. + clock.setValue(untilTs - 1); + runMonitorOnce(); + Thread.sleep(200); + SystemTransformRecord stillWaiting = fetch(conn, schemaName, dataTableName, null); + assertEquals("Monitor must no-op while clock < deadline", + PTable.TransformStatus.PENDING_PARTIAL_PASS.name(), stillWaiting.getTransformStatus()); + + // Once the clock reaches the deadline the monitor advances through the partial pass to + // COMPLETED. + clock.setValue(untilTs + 1); + SystemTransformRecord completed = driveMonitorToStatus(conn, schemaName, dataTableName, null, + PTable.TransformStatus.COMPLETED, clock); + assertNotNull(completed); + + // No stranded unverified rows on the new physical table after the partial pass. + assertEquals("No unverified rows should remain after the partial pass", 0, + countUnverified(conn, completed.getNewPhysicalTableName())); + + // Sanity: pointer still points at the new physical table. + PTable finalTable = conn.getTableNoCache(dataTableFullName); + assertEquals(newTableName, finalTable.getPhysicalName(true).getString()); + } + } + + @Test + public void testCutoverLifecycleMutableTableWithoutIndex() throws Exception { + runCutoverLifecycle(false, false); + } + + @Test + public void testCutoverLifecycleImmutableTableWithoutIndex() throws Exception { + runCutoverLifecycle(false, true); + } + + @Test + public void testCutoverLifecycleTableWithSecondaryIndex() throws Exception { + runCutoverLifecycle(true, false); + } + + /** + * Cutover on a base table that HAS a child view must tear down the dual-write links installed on + * both the base table AND the child view. Asserts that after cutover reaches PENDING_PARTIAL_PASS + * (the point at which the pointer swap and link teardown are committed) no TRANSFORMING_NEW_TABLE + * link rows pointing at the new physical table remain, for the base table or the view. + */ + @Test + public void testCutoverTearsDownViewDualWriteLinks() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + String schemaName = generateUniqueName(); + String dataTableName = "TBL_" + generateUniqueName(); + String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName); + String newTableName = dataTableName + "_1"; + String newTableFullName = SchemaUtil.getTableName(schemaName, newTableName); + String viewName = "VW_" + generateUniqueName(); + String viewFullName = SchemaUtil.getTableName(schemaName, viewName); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + TransformToolIT.createTableAndUpsertRows(conn, dataTableFullName, 10, ""); + // Create a child view over the base table before starting the transform, so the transform + // installs a dual-write link on the view as well as the base table. + conn.createStatement() + .execute("CREATE VIEW " + viewFullName + " AS SELECT * FROM " + dataTableFullName); + + conn.createStatement().execute("ALTER TABLE " + dataTableFullName + + " SET IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + + SystemTransformRecord record = fetch(conn, schemaName, dataTableName, null); + assertNotNull(record); + + // Before cutover, links exist on both the base table and the view (2 total). + assertEquals("Base table and child view must each have a TRANSFORMING_NEW_TABLE link", 2, + countLinksToNewPhysicalTable(conn, newTableFullName)); + + // Drive to PENDING_PARTIAL_PASS: pointer swap and link teardown are committed at this point. + SystemTransformRecord pendingPartial = driveMonitorToStatus(conn, schemaName, dataTableName, + null, PTable.TransformStatus.PENDING_PARTIAL_PASS, clock); + assertNotNull(pendingPartial); + + PTable swappedTable = conn.getTableNoCache(dataTableFullName); + assertEquals(newTableName, swappedTable.getPhysicalName(true).getString()); + + // No TRANSFORMING_NEW_TABLE links (base OR view) may survive cutover. + assertEquals("All TRANSFORMING_NEW_TABLE links (base + view) must be deleted at cutover", 0, + countLinksToNewPhysicalTable(conn, newTableFullName)); + assertEquals("Base-table TRANSFORMING_NEW_TABLE link must be deleted at cutover", 0, + countLinkRows(conn, schemaName, dataTableName)); + assertEquals("View TRANSFORMING_NEW_TABLE link must be deleted at cutover", 0, + countLinkRows(conn, schemaName, viewName)); + } + } + + /** + * Seeds a PENDING_PARTIAL_PASS record plus its monitor task, then drives the monitor to prove it + * honors the persisted wait deadline: while the injected clock is before the deadline a monitor + * run is a no-op (still PENDING_PARTIAL_PASS), and once the clock reaches the deadline a monitor + * run advances the record to PARTIAL_PASS_RUNNING. The PARTIAL_PASS_RUNNING transition is + * committed before the partial-pass tool is launched, so it is observable even though the tool + * run for this seeded (backing-table-less) record does not itself finish. No real sleep is used. + */ + @Test + public void testMonitorHonorsWaitDeadline() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + String logicalTableName = generateUniqueName(); + long deadline = clock.currentTime() + (60L * 60L * 1000L); + + SystemTransformRecord.SystemTransformBuilder builder = + new SystemTransformRecord.SystemTransformBuilder(); + builder.setLogicalTableName(logicalTableName); + builder.setNewPhysicalTableName(logicalTableName + "_1"); + // METADATA_TRANSFORM has a defined partial-transform variant, so the monitor is able to move + // the record out of PENDING_PARTIAL_PASS once the wait window elapses. + builder.setTransformType(PTable.TransformType.METADATA_TRANSFORM); + builder.setTransformStatus(PTable.TransformStatus.PENDING_PARTIAL_PASS.name()); + builder.setPendingPartialPassUntilTs(deadline); + Transform.upsertTransform(builder.build(), conn); + + SystemTransformRecord readBack = fetch(conn, null, logicalTableName, null); + assertNotNull(readBack); + // The BIGINT column round-trips exactly. + assertEquals(Long.valueOf(deadline), readBack.getPendingPartialPassUntilTs()); + + // Register the monitor task so runMonitorOnce() dispatches to this transform record. + Timestamp startTs = new Timestamp(EnvironmentEdgeManager.currentTimeMillis()); + ServerTask.addTask(new SystemTaskParams.SystemTaskParamsBuilder().setConn(conn) + .setTaskType(PTable.TaskType.TRANSFORM_MONITOR).setTenantId(null).setSchemaName(null) + .setTableName(logicalTableName).setTaskStatus(PTable.TaskStatus.CREATED.toString()) + .setData(null).setPriority(null).setStartTs(startTs).setEndTs(null).build()); + + // Before the deadline: a monitor run must NOT advance the record. + clock.setValue(deadline - 1); + runMonitorOnce(); + Thread.sleep(200); + SystemTransformRecord stillWaiting = fetch(conn, null, logicalTableName, null); + assertEquals("Monitor must no-op while clock < deadline", + PTable.TransformStatus.PENDING_PARTIAL_PASS.name(), stillWaiting.getTransformStatus()); + + // At/after the deadline: a monitor run advances the record to PARTIAL_PASS_RUNNING, which is + // committed before the (asynchronous) partial pass is launched. + clock.setValue(deadline + 1); + SystemTransformRecord running = driveMonitorToStatus(conn, null, logicalTableName, null, + PTable.TransformStatus.PARTIAL_PASS_RUNNING, clock); + assertEquals(PTable.TransformStatus.PARTIAL_PASS_RUNNING.name(), + running.getTransformStatus()); + } + } + + /** + * Drives a real cutover to PENDING_PARTIAL_PASS and asserts the record carries the completed + * full-pass job id at that point, then advances into PARTIAL_PASS_RUNNING and asserts the stale + * full-pass job id never leaks into that state. The PENDING_PARTIAL_PASS -> + * PARTIAL_PASS_RUNNING transition clears the inherited full-pass job id; the launched partial + * pass then registers its OWN job id under PARTIAL_PASS_RUNNING. Either way the record must never + * carry the already-successful full-pass job id while PARTIAL_PASS_RUNNING, which is what keeps + * the PARTIAL_PASS_RUNNING monitoring branch from mistaking that job for the partial-pass job and + * prematurely completing the transform. + */ + @Test + public void testPartialPassRunningTransitionClearsInheritedJobId() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + String schemaName = generateUniqueName(); + String dataTableName = "TBL_" + generateUniqueName(); + String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + TransformToolIT.createTableAndUpsertRows(conn, dataTableFullName, 10, ""); + conn.createStatement().execute("ALTER TABLE " + dataTableFullName + + " SET IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + + // Drive the full pass through cutover to PENDING_PARTIAL_PASS. The full pass ran a + // TransformTool job, so the record carries that (now completed) job id here. + SystemTransformRecord pendingPartial = driveMonitorToStatus(conn, schemaName, dataTableName, + null, PTable.TransformStatus.PENDING_PARTIAL_PASS, clock); + String fullPassJobId = pendingPartial.getTransformJobId(); + assertNotNull("Full-pass job id must be present on the PENDING_PARTIAL_PASS record", + fullPassJobId); + + // Advance past the wait deadline and into PARTIAL_PASS_RUNNING. The transition nulls out the + // inherited full-pass job id, and the partial pass registers its own job id afterward. The + // record's job id while PARTIAL_PASS_RUNNING is therefore either null (before the partial + // pass + // registers) or the partial-pass job id -- but never the stale full-pass job id. + clock.setValue(pendingPartial.getPendingPartialPassUntilTs() + 1); + SystemTransformRecord running = driveMonitorToStatus(conn, schemaName, dataTableName, null, + PTable.TransformStatus.PARTIAL_PASS_RUNNING, clock); + assertTrue("PARTIAL_PASS_RUNNING must not carry the stale full-pass job id", + running.getTransformJobId() == null || !fullPassJobId.equals(running.getTransformJobId())); + } + } + + /** + * Defense-in-depth guard for the PARTIAL_PASS_RUNNING monitoring branch: a record that is + * PARTIAL_PASS_RUNNING but still carries the FULL transform type together with a stale + * (already-successful) full-pass job id must NOT be driven to COMPLETED. Only a partial-type + * record whose own partial pass has registered may complete via that branch. Seeds exactly that + * hazardous state and asserts a monitor run leaves the record untouched. + */ + @Test + public void testMonitorDoesNotCompletePartialPassRunningWithStaleFullPassJob() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + // Inject a completed-and-successful job matching the seeded full-pass job id. This is what + // makes + // the test discriminating: without the type gate the PARTIAL_PASS_RUNNING branch would look up + // this (successful) job and drive the record straight to COMPLETED, failing the assertion + // below. + // The gate keeps a full-type record from ever consulting the lookup, so the record stays put. + // Reset via @After resetJobLookupForTesting(). + Job successfulJob = Mockito.mock(Job.class); + Mockito.when(successfulJob.isComplete()).thenReturn(true); + Mockito.when(successfulJob.isSuccessful()).thenReturn(true); + TransformMonitorTask.setJobLookupForTesting((configuration, jobId) -> successfulJob); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + String logicalTableName = generateUniqueName(); + + SystemTransformRecord.SystemTransformBuilder builder = + new SystemTransformRecord.SystemTransformBuilder(); + builder.setLogicalTableName(logicalTableName); + builder.setNewPhysicalTableName(logicalTableName + "_1"); + // FULL transform type + a stale full-pass job id is the pre-fix hazard state: without the + // type gate the branch would look up that job, find it successful, and complete the + // transform. + builder.setTransformType(PTable.TransformType.METADATA_TRANSFORM); + builder.setTransformStatus(PTable.TransformStatus.PARTIAL_PASS_RUNNING.name()); + builder.setTransformJobId("job_000000000000_0001"); + Transform.upsertTransform(builder.build(), conn); + + // Register the monitor task so runMonitorOnce() dispatches to this transform record. + Timestamp startTs = new Timestamp(EnvironmentEdgeManager.currentTimeMillis()); + ServerTask.addTask(new SystemTaskParams.SystemTaskParamsBuilder().setConn(conn) + .setTaskType(PTable.TaskType.TRANSFORM_MONITOR).setTenantId(null).setSchemaName(null) + .setTableName(logicalTableName).setTaskStatus(PTable.TaskStatus.CREATED.toString()) + .setData(null).setPriority(null).setStartTs(startTs).setEndTs(null).build()); + + runMonitorOnce(); + Thread.sleep(200); + + SystemTransformRecord after = fetch(conn, null, logicalTableName, null); + assertNotNull(after); + assertEquals( + "Full-type PARTIAL_PASS_RUNNING record with a stale full-pass job must not be completed", + PTable.TransformStatus.PARTIAL_PASS_RUNNING.name(), after.getTransformStatus()); + } + } + + /** + * A running partial pass must be observable in status PARTIAL_PASS_RUNNING (partial transform + * type plus a non-null partial-pass job id), not in STARTED. Only then does the + * PARTIAL_PASS_RUNNING branch of the monitor -- which alone can drive a repeatedly-failing + * partial pass to a terminal FAILED state -- own the running partial pass. If the running partial + * pass were left in STARTED (the state TransformTool.runTransform sets), the STARTED branch, + * which has no terminal transition for an exhausted partial pass, would monitor it and could + * strand it forever. Drives a real cutover through the partial pass and asserts the record is + * observed at least once in PARTIAL_PASS_RUNNING with a partial type and a registered job id + * before completing. + */ + @Test + public void testRunningPartialPassIsObservableAsPartialPassRunning() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + String schemaName = generateUniqueName(); + String dataTableName = "TBL_" + generateUniqueName(); + String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + TransformToolIT.createTableAndUpsertRows(conn, dataTableFullName, 10, ""); + conn.createStatement().execute("ALTER TABLE " + dataTableFullName + + " SET IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + + // Drive to PENDING_PARTIAL_PASS, then past the wait deadline so the next monitor step kicks + // the partial pass and re-asserts PARTIAL_PASS_RUNNING with the partial-pass job id. + SystemTransformRecord pendingPartial = driveMonitorToStatus(conn, schemaName, dataTableName, + null, PTable.TransformStatus.PENDING_PARTIAL_PASS, clock); + clock.setValue(pendingPartial.getPendingPartialPassUntilTs() + 1); + + // Reach the running partial pass deterministically. driveMonitorToStatus fails the test if + // the record never reaches PARTIAL_PASS_RUNNING, so a regression that ran the partial pass in + // STARTED is caught here rather than passing silently. + SystemTransformRecord running = driveMonitorToStatus(conn, schemaName, dataTableName, null, + PTable.TransformStatus.PARTIAL_PASS_RUNNING, clock); + assertTrue("Running partial pass must carry the partial transform type", + PTable.TransformType.isPartialTransform(running.getTransformType())); + assertNotNull("Running partial pass must have a registered partial-pass job id", + running.getTransformJobId()); + + // The transform must ultimately complete. + SystemTransformRecord completed = driveMonitorToStatus(conn, schemaName, dataTableName, null, + PTable.TransformStatus.COMPLETED, clock); + assertNotNull(completed); + } + } + + /** + * A partial pass whose job repeatedly fails must not strand the transform. Once retries are + * exhausted the PARTIAL_PASS_RUNNING branch must drive the record to a terminal FAILED state + * rather than re-entering PARTIAL_PASS_RUNNING (or STARTED) forever. Seeds the exact observable + * state a running partial pass now produces -- PARTIAL_PASS_RUNNING, partial transform type, a + * registered job id, and a retry count already at/above the maximum -- and injects a + * completed-and-failed job through the monitor's job-lookup seam. Asserts a single monitor run + * moves the record to FAILED. + */ + @Test + public void testPartialPassRetriesExhaustedReachesFailed() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + // Inject a job that is complete and unsuccessful so the PARTIAL_PASS_RUNNING branch takes the + // failure path. The retry count is seeded above the maximum so the branch treats retries as + // exhausted and must transition to FAILED. + Job failedJob = Mockito.mock(Job.class); + Mockito.when(failedJob.isComplete()).thenReturn(true); + Mockito.when(failedJob.isSuccessful()).thenReturn(false); + TransformMonitorTask.setJobLookupForTesting((configuration, jobId) -> failedJob); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + String logicalTableName = generateUniqueName(); + + SystemTransformRecord.SystemTransformBuilder builder = + new SystemTransformRecord.SystemTransformBuilder(); + builder.setLogicalTableName(logicalTableName); + builder.setNewPhysicalTableName(logicalTableName + "_1"); + // This is the real observable state of a running partial pass: partial transform type, + // PARTIAL_PASS_RUNNING, and a registered partial-pass job id. + builder.setTransformType(PTable.TransformType.METADATA_TRANSFORM_PARTIAL); + builder.setTransformStatus(PTable.TransformStatus.PARTIAL_PASS_RUNNING.name()); + builder.setTransformJobId("job_000000000000_0002"); + // Well above the default maximum retry count so retries are exhausted. + builder.setTransformRetryCount(1000); + Transform.upsertTransform(builder.build(), conn); + + Timestamp startTs = new Timestamp(EnvironmentEdgeManager.currentTimeMillis()); + ServerTask.addTask(new SystemTaskParams.SystemTaskParamsBuilder().setConn(conn) + .setTaskType(PTable.TaskType.TRANSFORM_MONITOR).setTenantId(null).setSchemaName(null) + .setTableName(logicalTableName).setTaskStatus(PTable.TaskStatus.CREATED.toString()) + .setData(null).setPriority(null).setStartTs(startTs).setEndTs(null).build()); + + SystemTransformRecord failed = driveMonitorToStatus(conn, null, logicalTableName, null, + PTable.TransformStatus.FAILED, clock); + assertEquals( + "A partial pass whose retries are exhausted must reach terminal FAILED, not strand", + PTable.TransformStatus.FAILED.name(), failed.getTransformStatus()); + } + } + + /** + * A partial-pass job that cannot be resolved (aged out of the job-history server, + * resource-manager restart, etc.) must be treated like a failed job and routed through the + * retry-budgeted path, not left to strand. A job that cannot be found cannot be confirmed + * successful, so the PARTIAL_PASS_RUNNING branch must NOT return SKIPPED for it: after the + * pointer swap the task sits in the STARTED task state and only CREATED/RETRY tasks are re-picked + * up, so a SKIPPED result would leave the already-cut-over table's repairing partial pass forever + * unrun. Seeds an exhausted PARTIAL_PASS_RUNNING record and injects a null (not-found) job + * through the lookup seam; asserts a monitor run drives the record to terminal FAILED rather than + * stranding it in PARTIAL_PASS_RUNNING. + */ + @Test + public void testPartialPassRunningJobNotFoundDoesNotStrand() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + // Inject a not-found job (the lookup returns null). The retry count is seeded above the maximum + // so retries are exhausted and the not-found job must be treated as a terminal failure without + // needing a real TransformTool run. + TransformMonitorTask.setJobLookupForTesting((configuration, jobId) -> null); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + String logicalTableName = generateUniqueName(); + + SystemTransformRecord.SystemTransformBuilder builder = + new SystemTransformRecord.SystemTransformBuilder(); + builder.setLogicalTableName(logicalTableName); + builder.setNewPhysicalTableName(logicalTableName + "_1"); + // The real observable state of a running partial pass: partial transform type, + // PARTIAL_PASS_RUNNING, and a registered partial-pass job id (which the lookup cannot + // resolve). + builder.setTransformType(PTable.TransformType.METADATA_TRANSFORM_PARTIAL); + builder.setTransformStatus(PTable.TransformStatus.PARTIAL_PASS_RUNNING.name()); + builder.setTransformJobId("job_000000000000_0004"); + // Well above the default maximum retry count so retries are exhausted. + builder.setTransformRetryCount(1000); + Transform.upsertTransform(builder.build(), conn); + + Timestamp startTs = new Timestamp(EnvironmentEdgeManager.currentTimeMillis()); + ServerTask.addTask(new SystemTaskParams.SystemTaskParamsBuilder().setConn(conn) + .setTaskType(PTable.TaskType.TRANSFORM_MONITOR).setTenantId(null).setSchemaName(null) + .setTableName(logicalTableName).setTaskStatus(PTable.TaskStatus.CREATED.toString()) + .setData(null).setPriority(null).setStartTs(startTs).setEndTs(null).build()); + + SystemTransformRecord failed = driveMonitorToStatus(conn, null, logicalTableName, null, + PTable.TransformStatus.FAILED, clock); + assertEquals( + "A PARTIAL_PASS_RUNNING record whose job cannot be found must reach terminal " + + "FAILED, not strand in PARTIAL_PASS_RUNNING", + PTable.TransformStatus.FAILED.name(), failed.getTransformStatus()); + } + } + + /** + * Regression test: a partial-pass record left in PARTIAL_PASS_RUNNING with NO registered job id + * must still reach a terminal state, not no-op forever. This is the state a failed initial + * partial-pass kick produces: the cutover transition commits (PARTIAL_PASS_RUNNING, partial type, + * job id cleared) and then kicks the partial pass; if that first TransformTool run fails before + * its STARTED transition (connection acquisition, index-table creation, or argument validation + * throws, so the run returns without registering a job id and without propagating an exception), + * the committed row stays partial-type/PARTIAL_PASS_RUNNING with a null job id. The pointer swap + * has already happened, so the record must reach a terminal state so the unverified rows are + * either repaired (via retry) or the transform is surfaced as FAILED. A monitoring branch that + * only acted when a job id was present would silently no-op on every subsequent scan, stranding + * the already-cut-over table with no repairing partial pass and no terminal status. + *

+ * Seeds exactly that observable state -- PARTIAL_PASS_RUNNING, partial transform type, a null job + * id, retry count above the maximum -- and installs a job-lookup seam that FAILS if invoked, to + * prove the null-job-id path reaches FAILED without ever consulting the (irrelevant) job lookup. + * A single monitor run must move the record to terminal FAILED. + */ + @Test + public void testPartialPassRunningNullJobIdDoesNotStrand() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + // The job lookup must never be consulted when no job id is registered: a null job id is handled + // directly as an unconfirmable (failed) partial pass. Fail loudly if the lookup is invoked. + TransformMonitorTask.setJobLookupForTesting((configuration, jobId) -> { + throw new AssertionError( + "Job lookup must not be called when no partial-pass job id is registered"); + }); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + String logicalTableName = generateUniqueName(); + + SystemTransformRecord.SystemTransformBuilder builder = + new SystemTransformRecord.SystemTransformBuilder(); + builder.setLogicalTableName(logicalTableName); + builder.setNewPhysicalTableName(logicalTableName + "_1"); + // The observable state a failed initial partial-pass kick leaves behind: partial transform + // type and PARTIAL_PASS_RUNNING, but no job id was ever registered. + builder.setTransformType(PTable.TransformType.METADATA_TRANSFORM_PARTIAL); + builder.setTransformStatus(PTable.TransformStatus.PARTIAL_PASS_RUNNING.name()); + builder.setTransformJobId(null); + // Well above the default maximum retry count so retries are exhausted and the branch must + // move + // straight to terminal FAILED without launching a real TransformTool run. + builder.setTransformRetryCount(1000); + Transform.upsertTransform(builder.build(), conn); + + Timestamp startTs = new Timestamp(EnvironmentEdgeManager.currentTimeMillis()); + ServerTask.addTask(new SystemTaskParams.SystemTaskParamsBuilder().setConn(conn) + .setTaskType(PTable.TaskType.TRANSFORM_MONITOR).setTenantId(null).setSchemaName(null) + .setTableName(logicalTableName).setTaskStatus(PTable.TaskStatus.CREATED.toString()) + .setData(null).setPriority(null).setStartTs(startTs).setEndTs(null).build()); + + SystemTransformRecord failed = driveMonitorToStatus(conn, null, logicalTableName, null, + PTable.TransformStatus.FAILED, clock); + assertEquals( + "A PARTIAL_PASS_RUNNING record with no registered job id must reach terminal FAILED, not " + + "strand in PARTIAL_PASS_RUNNING", + PTable.TransformStatus.FAILED.name(), failed.getTransformStatus()); + } + } + + /** + * Regression test for the partial-pass retry-count accounting. A genuine retry must strictly + * ADVANCE the retry count, tick by tick, so it eventually reaches the maximum and the + * retries-exhausted -> terminal FAILED transition becomes reachable. + *

+ * Mechanism under test: on the retry path {@code kickPartialPass} skips the compensating + * pre-decrement it applies to the very first (non-retry) partial-pass kick. TransformTool's + * STARTED transition unconditionally increments the retry count and auto-commits it on + * TransformTool's own connection, so skipping the decrement lets the persisted count strictly + * increase. A prior implementation cancelled that increment with a matching decrement on the + * retry path too, pinning the count so the terminal FAILED transition was unreachable and a + * deterministically-failing partial pass resubmitted forever; this test fails against that + * implementation (the count nets back to the seed) and passes once the decrement is skipped on + * the retry path. + *

+ * The retry's kick runs a real TransformTool invocation whose pre-run validation resolves both + * the logical table and the new physical table before the STARTED increment is reached. The test + * therefore drives a real cutover to PENDING_PARTIAL_PASS, which creates both backing tables + * through the production machinery, before forcing the retry -- so validation passes and the + * increment actually fires. A backing-table-less seed would throw inside validation before the + * increment and would false-fail against the correct fix. + */ + @Test + public void testPartialPassRetryAdvancesRetryCount() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + String schemaName = generateUniqueName(); + String dataTableName = "TBL_" + generateUniqueName(); + String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + // Real backing tables via the production ALTER/cutover machinery, so the retry's + // TransformTool + // run passes pre-run validation and reaches the STARTED increment. + TransformToolIT.createTableAndUpsertRows(conn, dataTableFullName, 10, ""); + conn.createStatement().execute("ALTER TABLE " + dataTableFullName + + " SET IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + + // Drive the real transform through cutover to PENDING_PARTIAL_PASS using the REAL job-lookup + // seam: this drive depends on the actual full-pass MR job completing successfully. Only after + // reaching PENDING_PARTIAL_PASS do we swap in the failing-job seam below -- installing it + // earlier would make the monitor see the real full-pass job as failed and never cut over. + // At this point both the logical table and the new physical table exist, so the partial-pass + // TransformTool run will pass validation. + SystemTransformRecord pendingPartial = driveMonitorToStatus(conn, schemaName, dataTableName, + null, PTable.TransformStatus.PENDING_PARTIAL_PASS, clock); + + // Now force the PARTIAL_PASS_RUNNING branch down the retry path: the injected job is complete + // and unsuccessful, so the monitor retries the partial pass rather than completing it. + Job failedJob = Mockito.mock(Job.class); + Mockito.when(failedJob.isComplete()).thenReturn(true); + Mockito.when(failedJob.isSuccessful()).thenReturn(false); + TransformMonitorTask.setJobLookupForTesting((configuration, jobId) -> failedJob); + + // Seed the observable state of a running-but-failing partial pass, one retry below the + // maximum so the failure path takes a genuine retry (not the already-exhausted + // straight-to-FAILED path). Build from the driven record so schema/logical/new-physical names + // and the last-state timestamp carry over and the seeded record stays consistent with the + // real backing tables (the partial pass validates the last-transform time against that ts). + int seededRetryCount = PhoenixConfigurationUtil.DEFAULT_TRANSFORM_RETRY_COUNT - 1; + SystemTransformRecord.SystemTransformBuilder builder = + new SystemTransformRecord.SystemTransformBuilder(pendingPartial); + builder.setTransformType(PTable.TransformType.METADATA_TRANSFORM_PARTIAL); + builder.setTransformStatus(PTable.TransformStatus.PARTIAL_PASS_RUNNING.name()); + builder.setTransformJobId("job_000000000000_0003"); + builder.setTransformRetryCount(seededRetryCount); + Transform.upsertTransform(builder.build(), conn); + + // Replace the monitor task chain left by the drive with a single fresh CREATED task, so the + // next runMonitorOnce() dispatches exactly once to the seeded record. + conn.createStatement().execute("DELETE FROM " + PhoenixDatabaseMetaData.SYSTEM_TASK_NAME); + Timestamp startTs = new Timestamp(EnvironmentEdgeManager.currentTimeMillis()); + ServerTask.addTask(new SystemTaskParams.SystemTaskParamsBuilder().setConn(conn) + .setTaskType(PTable.TaskType.TRANSFORM_MONITOR).setTenantId(null).setSchemaName(schemaName) + .setTableName(dataTableName).setTaskStatus(PTable.TaskStatus.CREATED.toString()) + .setData(null).setPriority(null).setStartTs(startTs).setEndTs(null).build()); + + // A monitor run takes the PARTIAL_PASS_RUNNING failure path and retries the partial pass; + // TransformTool's STARTED transition then increments the (auto-committed) retry count. Poll + // for the persisted count to exceed the seed. + int observedRetryCount = seededRetryCount; + for (int i = 0; i < 60; i++) { + SystemTransformRecord record = fetch(conn, schemaName, dataTableName, null); + if (record != null && record.getTransformRetryCount() > seededRetryCount) { + observedRetryCount = record.getTransformRetryCount(); + break; + } + runMonitorOnce(); + Thread.sleep(200); + } + assertTrue("A genuine partial-pass retry must strictly advance the retry count (seed " + + seededRetryCount + ", observed " + observedRetryCount + + ") so retries-exhausted -> FAILED is reachable", observedRetryCount > seededRetryCount); + } + } + + /** + * Verifies the new PENDING_PARTIAL_PASS_UNTIL_TS column exists on a freshly created + * SYSTEM.TRANSFORM and that a BIGINT value, including NULL, round-trips through upsert and + * select. + */ + @Test + public void testSystemTransformNewColumnReadWrite() throws Exception { + try (Connection conn = DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + String logicalTableName = generateUniqueName(); + String upsert = "UPSERT INTO " + PhoenixDatabaseMetaData.SYSTEM_TRANSFORM_NAME + " (" + + PhoenixDatabaseMetaData.LOGICAL_TABLE_NAME + ", " + + PhoenixDatabaseMetaData.NEW_PHYS_TABLE_NAME + ", " + + PhoenixDatabaseMetaData.PENDING_PARTIAL_PASS_UNTIL_TS + ") VALUES (?, ?, ?)"; + try (PreparedStatement stmt = conn.prepareStatement(upsert)) { + stmt.setString(1, logicalTableName); + stmt.setString(2, logicalTableName + "_1"); + stmt.setLong(3, 1234567890123L); + stmt.execute(); + } + String select = "SELECT " + PhoenixDatabaseMetaData.PENDING_PARTIAL_PASS_UNTIL_TS + " FROM " + + PhoenixDatabaseMetaData.SYSTEM_TRANSFORM_NAME + " WHERE " + + PhoenixDatabaseMetaData.LOGICAL_TABLE_NAME + " = ?"; + try (PreparedStatement stmt = conn.prepareStatement(select)) { + stmt.setString(1, logicalTableName); + ResultSet rs = stmt.executeQuery(); + assertTrue(rs.next()); + assertEquals(1234567890123L, rs.getLong(1)); + } + // A NULL round-trips as NULL (wasNull path). + String logicalTableName2 = generateUniqueName(); + try (PreparedStatement stmt = + conn.prepareStatement("UPSERT INTO " + PhoenixDatabaseMetaData.SYSTEM_TRANSFORM_NAME + " (" + + PhoenixDatabaseMetaData.LOGICAL_TABLE_NAME + ", " + + PhoenixDatabaseMetaData.NEW_PHYS_TABLE_NAME + ") VALUES (?, ?)")) { + stmt.setString(1, logicalTableName2); + stmt.setString(2, logicalTableName2 + "_1"); + stmt.execute(); + } + SystemTransformRecord record = Transform.getTransformRecord(null, logicalTableName2, null, + null, ((PhoenixConnection) conn)); + assertNotNull(record); + assertNull(record.getPendingPartialPassUntilTs()); + } + } + + /** + * A logical table configured to never refresh its cache resolves its update-cache-frequency to + * Long.MAX_VALUE. The persisted wait deadline must still be a bounded timestamp strictly in the + * future: scaling Long.MAX_VALUE and adding it to the current time would overflow into a negative + * (past) deadline, which would make the monitor skip the wait entirely and run the partial pass + * with no delay -- the opposite of the intended behavior for a table whose clients cache + * indefinitely. This drives a real cutover on a NEVER-cache table and asserts the deadline lands + * within a sane bounded window rather than overflowing. + */ + @Test + public void testNeverCachedTableYieldsBoundedWaitDeadline() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + String schemaName = generateUniqueName(); + String dataTableName = "TBL_" + generateUniqueName(); + String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + long beforeCutover = clock.currentTime(); + // UPDATE_CACHE_FREQUENCY=NEVER makes the logical table's cache frequency Long.MAX_VALUE. + TransformToolIT.createTableAndUpsertRows(conn, dataTableFullName, 10, + " UPDATE_CACHE_FREQUENCY=NEVER"); + conn.createStatement().execute("ALTER TABLE " + dataTableFullName + + " SET IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + + SystemTransformRecord record = fetch(conn, schemaName, dataTableName, null); + assertNotNull(record); + + SystemTransformRecord pendingPartial = driveMonitorToStatus(conn, schemaName, dataTableName, + null, PTable.TransformStatus.PENDING_PARTIAL_PASS, clock); + Long untilTs = pendingPartial.getPendingPartialPassUntilTs(); + assertNotNull( + "PENDING_PARTIAL_PASS must persist a wait deadline even for a NEVER-cache table", untilTs); + // The deadline must be strictly in the future (the overflow bug produced a negative value). + assertTrue("Wait deadline for a NEVER-cache table must be in the future, was " + untilTs + + " vs cutover time " + beforeCutover, untilTs > beforeCutover); + // And it must be bounded, not effectively infinite: the wait is capped, so the deadline is at + // most the cutover time plus the ceiling (with slack for clock advances during the drive). + long ceiling = beforeCutover + (24L * 60L * 60L * 1000L) + (60L * 60L * 1000L); + assertTrue("Wait deadline for a NEVER-cache table must be bounded by the ceiling, was " + + untilTs + " vs ceiling " + ceiling, untilTs <= ceiling); + } + } + + /** + * Regression test for the partial-pass repair-scan floor. After cutover the monitor waits for + * clients to refresh their cached physical-table pointer; a stale client can still write to the + * old pointer during that window. The partial pass must re-verify those writes, so its scan lower + * bound has to reach back to the cutover instant -- not to lastStateTs, which is re-stamped only + * after the wait window elapses. A floor derived from lastStateTs would exclude every write in + * (cutover, cutover + waitWindow], the exact rows the wait exists to protect, and silently drop + * them. + *

+ * Drives a real cutover to PENDING_PARTIAL_PASS (which captures the cutover instant), advances + * the clock past the wait deadline, and drives into PARTIAL_PASS_RUNNING (whose transition + * re-stamps lastStateTs to the post-wait clock). Asserts the captured cutover instant survives + * every transition unchanged and lies strictly before the post-wait lastStateTs, so the + * cutover-derived repair floor genuinely covers the wait window that a lastStateTs-derived floor + * would strand. + */ + @Test + public void testPartialPassRepairFloorCoversPostCutoverWaitWindow() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + String schemaName = generateUniqueName(); + String dataTableName = "TBL_" + generateUniqueName(); + String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + TransformToolIT.createTableAndUpsertRows(conn, dataTableFullName, 10, ""); + conn.createStatement().execute("ALTER TABLE " + dataTableFullName + + " SET IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + + // Drive to PENDING_PARTIAL_PASS: the pointer swap has happened and the cutover instant is + // captured on the record. + SystemTransformRecord pendingPartial = driveMonitorToStatus(conn, schemaName, dataTableName, + null, PTable.TransformStatus.PENDING_PARTIAL_PASS, clock); + Long cutoverTs = pendingPartial.getCutoverTs(); + assertNotNull("Cutover instant must be captured at PENDING_PARTIAL_PASS", cutoverTs); + Long deadline = pendingPartial.getPendingPartialPassUntilTs(); + assertNotNull("PENDING_PARTIAL_PASS must persist a wait deadline", deadline); + // The cutover instant strictly precedes the wait deadline it seeds (the bounded wait is + // always positive), so there is a genuine [cutover, deadline] window to protect. + assertTrue("Cutover instant (" + cutoverTs + ") must strictly precede the wait deadline (" + + deadline + ")", cutoverTs < deadline); + + // Advance well past the wait deadline (simulating the full client cache-refresh wait) and + // drive into PARTIAL_PASS_RUNNING, whose transition re-stamps lastStateTs to the post-wait + // clock. + clock.setValue(deadline + 1); + SystemTransformRecord running = driveMonitorToStatus(conn, schemaName, dataTableName, null, + PTable.TransformStatus.PARTIAL_PASS_RUNNING, clock); + + // The cutover instant survives the transition unchanged... + assertEquals("Cutover instant must be preserved across transitions", cutoverTs, + running.getCutoverTs()); + // ...and lies strictly before the post-wait lastStateTs. The interval (cutoverTs, + // lastStateTs] is precisely the window a lastStateTs-derived repair floor would strand; the + // cutover-derived floor used by kickPartialPass covers it. + assertNotNull(running.getTransformLastStateTs()); + assertTrue( + "Post-wait lastStateTs (" + running.getTransformLastStateTs().getTime() + + ") must be strictly after the cutover instant (" + cutoverTs + "); the intervening" + + " window is exactly what a lastStateTs-derived repair floor would drop", + running.getTransformLastStateTs().getTime() > cutoverTs); + } + } + + /** + * Regression test for the crash-gated form of the repair-floor bug. doCutover commits the pointer + * swap durably, but the cutover instant that anchors the repair floor is persisted at the + * PENDING_CUTOVER handling. A crash after the swap but before that persist would, on re-entry, + * re-capture a later instant and push the repair floor past the real cutover -- silently dropping + * the post-cutover-window writes the partial pass exists to re-verify. The monitor must instead + * reuse the instant already persisted on the record rather than re-capturing the current time. + *

+ * Simulates the re-entry by driving a real cutover to PENDING_PARTIAL_PASS (pointer swapped, + * cutover instant persisted), then resetting the record back to PENDING_CUTOVER preserving that + * instant and the still-full transform type, advancing the clock far past it, and running the + * monitor again. Asserts the re-entry advances the record without moving the persisted cutover + * instant to the later clock -- doCutover is an idempotent no-op on the already-swapped pointer. + */ + @Test + public void testCutoverReentryReusesPersistedCutoverTs() throws Exception { + AdvancingClock clock = new AdvancingClock(); + EnvironmentEdgeManager.injectEdge(clock); + + String schemaName = generateUniqueName(); + String dataTableName = "TBL_" + generateUniqueName(); + String dataTableFullName = SchemaUtil.getTableName(schemaName, dataTableName); + + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + TransformToolIT.createTableAndUpsertRows(conn, dataTableFullName, 10, ""); + conn.createStatement().execute("ALTER TABLE " + dataTableFullName + + " SET IMMUTABLE_STORAGE_SCHEME=SINGLE_CELL_ARRAY_WITH_OFFSETS, COLUMN_ENCODED_BYTES=2"); + + // Drive to PENDING_PARTIAL_PASS: the pointer swap has happened and the cutover instant is + // persisted on the record. + SystemTransformRecord pendingPartial = driveMonitorToStatus(conn, schemaName, dataTableName, + null, PTable.TransformStatus.PENDING_PARTIAL_PASS, clock); + Long cutoverTs = pendingPartial.getCutoverTs(); + assertNotNull("Cutover instant must be persisted at PENDING_PARTIAL_PASS", cutoverTs); + // Precondition for the PENDING_CUTOVER re-entry guard: the transform type is still full at + // PENDING_PARTIAL_PASS (it flips to the partial variant only at PARTIAL_PASS_RUNNING), so a + // record reset to PENDING_CUTOVER re-enters the same branch. + assertFalse("Transform type must still be full at PENDING_PARTIAL_PASS", + PTable.TransformType.isPartialTransform(pendingPartial.getTransformType())); + + // Simulate a crash re-entry: reset the record to PENDING_CUTOVER preserving the persisted + // cutover instant and the full transform type, exactly the on-disk state a crash between the + // durable pre-commit and the PENDING_PARTIAL_PASS transition would leave behind. + Transform.updateTransformRecord(conn, pendingPartial, PTable.TransformStatus.PENDING_CUTOVER, + null, cutoverTs); + + // Move the clock far past the cutover instant so a re-capture (the bug) would produce a + // visibly later instant, then re-run the monitor. + clock.setValue(cutoverTs + (30L * 24L * 60L * 60L * 1000L)); + SystemTransformRecord reentered = driveMonitorToStatus(conn, schemaName, dataTableName, null, + PTable.TransformStatus.PENDING_PARTIAL_PASS, clock); + + // The re-entry advanced the record but reused the persisted cutover instant verbatim rather + // than the far-later clock -- so the repair floor still tracks the real cutover. + assertEquals("Re-entry must reuse the persisted cutover instant, not re-capture a later one", + cutoverTs, reentered.getCutoverTs()); + // Sanity: the pointer still points at the new physical table (doCutover was an idempotent + // no-op on re-entry). + PTable finalTable = conn.getTableNoCache(dataTableFullName); + assertEquals(dataTableName + "_1", finalTable.getPhysicalName(true).getString()); + } + } +} diff --git a/phoenix-core/src/test/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTaskWaitTest.java b/phoenix-core/src/test/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTaskWaitTest.java new file mode 100644 index 00000000000..44d33eecc02 --- /dev/null +++ b/phoenix-core/src/test/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTaskWaitTest.java @@ -0,0 +1,157 @@ +/* + * 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.tasks; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.sql.Timestamp; +import org.apache.phoenix.schema.transform.SystemTransformRecord; +import org.apache.phoenix.util.EnvironmentEdgeManager; +import org.apache.phoenix.util.ManualEnvironmentEdge; +import org.junit.Test; + +/** + * Unit coverage for the partial-pass wait arithmetic in {@link TransformMonitorTask}. The wait is + * added to the current time to form a persisted deadline, so the arithmetic must never overflow + * into a negative (past) deadline. In particular a table configured to never refresh its cache + * reports an update-cache-frequency of {@link Long#MAX_VALUE}; scaling that unbounded would + * saturate and, once added to the current time, wrap negative and defeat the wait entirely. These + * assertions pin the clamp-before-scale behavior deterministically, without standing up a cluster. + */ +public class TransformMonitorTaskWaitTest { + + private static final long MIN_WAIT_MS = 30L * 60L * 1000L; + private static final long MAX_WAIT_MS = 24L * 60L * 60L * 1000L; + + @Test + public void testNeverCachedFrequencyClampsToCeilingNotOverflow() { + // A never-refreshed table resolves update-cache-frequency to Long.MAX_VALUE. + long wait = TransformMonitorTask.boundedPartialPassWaitMs(Long.MAX_VALUE); + assertEquals("A never-refreshed table must clamp to the 24h ceiling, not overflow", MAX_WAIT_MS, + wait); + } + + @Test + public void testZeroAndSmallFrequencyFloorToMinimum() { + assertEquals("Zero frequency floors to the minimum wait", MIN_WAIT_MS, + TransformMonitorTask.boundedPartialPassWaitMs(0)); + assertEquals("A frequency below the floor (after scaling) floors to the minimum wait", + MIN_WAIT_MS, TransformMonitorTask.boundedPartialPassWaitMs(1000)); + } + + @Test + public void testNegativeFrequencyFloorsToMinimum() { + // Defensive: a negative frequency should never yield a negative or past deadline. + assertEquals("A negative frequency floors to the minimum wait", MIN_WAIT_MS, + TransformMonitorTask.boundedPartialPassWaitMs(-1L)); + assertEquals("Long.MIN_VALUE floors to the minimum wait", MIN_WAIT_MS, + TransformMonitorTask.boundedPartialPassWaitMs(Long.MIN_VALUE)); + } + + @Test + public void testMidRangeFrequencyScalesWithSafetyMargin() { + // A one-hour cache frequency, well inside the window, scales by the 1.10 safety margin. + long oneHour = 60L * 60L * 1000L; + long wait = TransformMonitorTask.boundedPartialPassWaitMs(oneHour); + assertEquals("A mid-range frequency scales by the 1.10 safety margin", (long) (oneHour * 1.10), + wait); + } + + @Test + public void testFrequencyAtOrAboveCeilingClampsToCeiling() { + assertEquals("A frequency exactly at the ceiling clamps to the ceiling", MAX_WAIT_MS, + TransformMonitorTask.boundedPartialPassWaitMs(MAX_WAIT_MS)); + assertEquals("A frequency above the ceiling clamps to the ceiling", MAX_WAIT_MS, + TransformMonitorTask.boundedPartialPassWaitMs(MAX_WAIT_MS + 1)); + } + + private static SystemTransformRecord recordWith(Long cutoverTs, Long lastStateTs) { + SystemTransformRecord.SystemTransformBuilder builder = + new SystemTransformRecord.SystemTransformBuilder(); + builder.setCutoverTs(cutoverTs); + builder.setLastStateTs(lastStateTs == null ? null : new Timestamp(lastStateTs)); + return builder.build(); + } + + @Test + public void testRepairScanFloorUsesCutoverInstantNotPostWaitLastStateTs() { + // Regression guard for the strand/data-loss bug: lastStateTs is stamped AFTER the post-cutover + // wait window, so a floor derived from it (5000 - 1) would skip every row written to the old + // pointer during [cutover, cutover + waitWindow]. The floor must instead track the cutover + // instant (1000 - 1). This is the exact scenario the fix exists to prevent. + long cutoverTs = 1000L; + long postWaitLastStateTs = 5000L; + long floor = TransformMonitorTask.repairScanFloor(recordWith(cutoverTs, postWaitLastStateTs)); + assertEquals("Repair-scan floor must track the cutover instant, not the post-wait lastStateTs", + cutoverTs - 1, floor); + } + + @Test + public void testRepairScanFloorFallsBackToLastStateTsForLegacyRecords() { + // Records predating the CUTOVER_TS column carry a null cutoverTs; preserve the prior behavior + // (floor derived from lastStateTs) rather than rescanning the whole table. + long lastStateTs = 5000L; + long floor = TransformMonitorTask.repairScanFloor(recordWith(null, lastStateTs)); + assertEquals("With no cutover instant, the floor falls back to lastStateTs", lastStateTs - 1, + floor); + } + + @Test + public void testRepairScanFloorFullScanWhenNeitherSet() { + assertEquals("With neither timestamp set, the floor is 0 (full scan)", 0L, + TransformMonitorTask.repairScanFloor(recordWith(null, null))); + } + + @Test + public void testResolveCutoverTsReusesPersistedInstantOnReentry() { + // A run re-entering the PENDING_CUTOVER handling after a crash must reuse the instant the prior + // run persisted (1000), never re-capture a later one -- otherwise the repair floor would drift + // past the real cutover and strand the post-cutover-window writes the partial pass re-verifies. + assertEquals("A persisted cutover instant must be reused verbatim on re-entry", 1000L, + TransformMonitorTask.resolveCutoverTs(recordWith(1000L, 5000L))); + } + + @Test + public void testResolveCutoverTsCapturesNowOnFirstRun() { + // A first run (no persisted instant) captures the current time -- taken before the pointer + // swap, this is the most conservative floor. + ManualEnvironmentEdge edge = new ManualEnvironmentEdge(); + edge.setValue(4242L); + EnvironmentEdgeManager.injectEdge(edge); + try { + assertEquals("A first run captures the current time as the cutover instant", 4242L, + TransformMonitorTask.resolveCutoverTs(recordWith(null, null))); + } finally { + EnvironmentEdgeManager.reset(); + } + } + + @Test + public void testResultAlwaysBoundedAndPositiveAcrossDomain() { + long[] samples = { Long.MIN_VALUE, -1L, 0L, 1L, 1000L, MIN_WAIT_MS, MAX_WAIT_MS / 2, + MAX_WAIT_MS, MAX_WAIT_MS + 1, Long.MAX_VALUE / 2, Long.MAX_VALUE - 1, Long.MAX_VALUE }; + for (long f : samples) { + long wait = TransformMonitorTask.boundedPartialPassWaitMs(f); + assertTrue("wait must be >= floor for input " + f, wait >= MIN_WAIT_MS); + assertTrue("wait must be <= ceiling for input " + f, wait <= MAX_WAIT_MS); + // The deadline is currentTime + wait; a bounded positive wait cannot overflow it. + assertTrue("wait must stay positive for input " + f, wait > 0); + } + } +} diff --git a/phoenix-core/src/test/java/org/apache/phoenix/util/MetaDataUtilTest.java b/phoenix-core/src/test/java/org/apache/phoenix/util/MetaDataUtilTest.java index 9589d06ecf0..87f20e2d07a 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/util/MetaDataUtilTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/util/MetaDataUtilTest.java @@ -293,6 +293,41 @@ public void testEncodeDecode() { assertEquals(expectedPhoenixVersion, phoenixVersion); } + /** + * The client's upgrade gate throws UpgradeRequiredException whenever the server's SYSTEM.CATALOG + * timestamp is below {@link MetaDataProtocol#MIN_SYSTEM_TABLE_TIMESTAMP}, and the server reports + * that timestamp as SYSTEM.CATALOG's own header timestamp (see MetaDataEndpointImpl#getVersion). + * SYSTEM.CATALOG's header only advances to a given timestamp when a genuinely new column is added + * to SYSTEM.CATALOG at that timestamp during upgrade. The last such column-add in + * ConnectionQueryServicesImpl#upgradeSystemCatalogIfRequired lands at + * {@link MetaDataProtocol#MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0}. If the gate threshold is bumped past + * that (for instance to accommodate a column added only to a non-CATALOG system table), a real + * in-place upgrade would leave SYSTEM.CATALOG below the threshold and every client would throw + * UpgradeRequiredException forever. This invariant guards that: a new MIN timestamp must be + * accompanied by a SYSTEM.CATALOG column-add at that timestamp, and this constant updated to + * match. + */ + @Test + public void testMinSystemTableTimestampIsSystemCatalogReachable() { + assertEquals( + "MIN_SYSTEM_TABLE_TIMESTAMP must equal the highest timestamp at which a SYSTEM.CATALOG column" + + " is added during upgrade; otherwise clients throw UpgradeRequiredException perpetually" + + " after an in-place upgrade. If you bump the min system-table timestamp, add a" + + " SYSTEM.CATALOG column-add at the new timestamp in upgradeSystemCatalogIfRequired and" + + " update this assertion.", + MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0, + MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP); + // Concrete tripwire on the absolute offset. The assertion above is satisfied by the constant + // definition itself, so it cannot catch a bump that forgets the paired SYSTEM.CATALOG + // column-add. Pin the exact offset so any future change to MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 + // fails here and forces the author to add a SYSTEM.CATALOG column-add at the new timestamp + // (see UPGRADE_TS_ANCHOR_5_4_0 in upgradeSystemCatalogIfRequired) and update this offset. + assertEquals( + "MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0 changed. Add a SYSTEM.CATALOG column-add at the new" + + " timestamp during upgrade before updating this offset.", + MetaDataProtocol.MIN_TABLE_TIMESTAMP + 46, MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0); + } + private Put generateOriginalPut() { String version = VersionInfo.getVersion(); KeyValueBuilder builder = KeyValueBuilder.get(version); From 73344db723bbb8723fa8e8d3651c097ae89d5751 Mon Sep 17 00:00:00 2001 From: lokesh-khurana Date: Mon, 14 Sep 2026 23:20:46 -0700 Subject: [PATCH 2/2] PHOENIX-7907 :- Harden cutover lifecycle monitor and add upgrade coverage Follow-up on the cutover lifecycle review: - Resolve the effective UPDATE_CACHE_FREQUENCY (explicit table value vs the connection default sentinel) before scaling the bounded partial-pass wait, mirroring MetaDataClient#avoidRpcToGetTable so a table left at the ALWAYS default no longer collapses the wait to the floor. - Persist the PENDING_CUTOVER status with an explicit commit so a crash after the pointer swap but before the status write cannot re-run the swap. - Capture the MapReduce job's completion/success as an immutable snapshot while the Cluster is open, then close it, instead of returning a live Job whose later isComplete()/isSuccessful() calls would re-query a closed client. - Widen upgradeSystemTransform to @VisibleForTesting (mirroring upgradeSystemLog) and add SystemTransformUpgradeIT covering the idempotent, ungated column-add upgrade path for SNAPSHOT clusters. - Add unit coverage for the effective-frequency resolution and the wait bounds; document the new lifecycle enum values and include them in the transform record debug string. Lower the per-scan monitor logging to DEBUG while keeping the state-transition lines at INFO. Generated-by: Claude Code (Opus 4.8) Co-Authored-By: Claude Opus 4.8 (1M context) --- .../query/ConnectionQueryServicesImpl.java | 6 +- .../org/apache/phoenix/schema/PTable.java | 4 + .../transform/SystemTransformRecord.java | 5 +- .../tasks/TransformMonitorTask.java | 102 ++++++++++++-- .../end2end/transform/CutoverLifecycleIT.java | 20 +-- .../transform/SystemTransformUpgradeIT.java | 130 ++++++++++++++++++ .../tasks/TransformMonitorTaskWaitTest.java | 35 +++++ 7 files changed, 273 insertions(+), 29 deletions(-) create mode 100644 phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/SystemTransformUpgradeIT.java diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java index 5c38d7d3cc6..d1fd173b83e 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java @@ -5304,11 +5304,15 @@ private PhoenixConnection upgradeSystemTask(PhoenixConnection metaConnection, return metaConnection; } - private PhoenixConnection upgradeSystemTransform(PhoenixConnection metaConnection, + @VisibleForTesting + public PhoenixConnection upgradeSystemTransform(PhoenixConnection metaConnection, Map systemTableToSnapshotMap) throws SQLException { try (Statement statement = metaConnection.createStatement()) { statement.executeUpdate(getTransformDDL()); } catch (NewerTableAlreadyExistsException ignored) { + // A newer SYSTEM.TRANSFORM header means a same-or-newer client already ran this DDL, whose + // CREATE statement carries the two new columns; the column-add below is therefore already + // done and skipping it is safe. } catch (TableAlreadyExistsException e) { // This is the first-ever column add to SYSTEM.TRANSFORM, so take a snapshot before altering. takeSnapshotOfSysTable(systemTableToSnapshotMap, e); diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java index dc100d19559..f4b0f695454 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/PTable.java @@ -330,11 +330,15 @@ public String toString() { return "PENDING_CUTOVER"; } }, + /** + * Cutover is done; the post-cutover partial-pass repair scan is deferred until the UCF wait. + */ PENDING_PARTIAL_PASS { public String toString() { return "PENDING_PARTIAL_PASS"; } }, + /** The post-cutover partial-pass repair scan has been launched and is being monitored. */ PARTIAL_PASS_RUNNING { public String toString() { return "PARTIAL_PASS_RUNNING"; diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/SystemTransformRecord.java b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/SystemTransformRecord.java index 9fb7e2bf1f6..d9699256ea2 100644 --- a/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/SystemTransformRecord.java +++ b/phoenix-core-client/src/main/java/org/apache/phoenix/schema/transform/SystemTransformRecord.java @@ -73,10 +73,11 @@ public SystemTransformRecord(PTable.TransformType transformType, String schemaNa public String getString() { return String.format( - "transformType: %s, schameName: %s, logicalTableName: %s, newPhysicalTableName: %s, logicalParentName: %s, status: %s", + "transformType: %s, schameName: %s, logicalTableName: %s, newPhysicalTableName: %s, logicalParentName: %s, status: %s, pendingPartialPassUntilTs: %s, cutoverTs: %s", String.valueOf(transformType), String.valueOf(schemaName), String.valueOf(logicalTableName), String.valueOf(newPhysicalTableName), String.valueOf(logicalParentName), - String.valueOf(transformStatus)); + String.valueOf(transformStatus), String.valueOf(pendingPartialPassUntilTs), + String.valueOf(cutoverTs)); } public PTable.TransformType getTransformType() { diff --git a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTask.java b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTask.java index e3bae8db915..3c2840d9b78 100644 --- a/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTask.java +++ b/phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTask.java @@ -32,6 +32,9 @@ import org.apache.phoenix.coprocessor.TaskRegionObserver; import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.mapreduce.transform.TransformTool; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.query.QueryServicesOptions; +import org.apache.phoenix.schema.ConnectionProperty; import org.apache.phoenix.schema.PTable; import org.apache.phoenix.schema.task.ServerTask; import org.apache.phoenix.schema.task.SystemTaskParams; @@ -78,20 +81,60 @@ public static void disableTransformMonitorTask(boolean disabled) { } /** - * Resolves the running MapReduce job for a given job id. Extracted behind an overridable seam so - * a test can inject a completed/failed job and exercise the PARTIAL_PASS_RUNNING branch's - * retries-exhausted -> FAILED transition deterministically, without submitting a real MR job - * that fails. The default implementation looks the job up on the real cluster. + * Resolves the completion/success of the MapReduce job for a given job id, as an immutable + * snapshot. Extracted behind an overridable seam so a test can inject a completed/failed status + * and exercise the PARTIAL_PASS_RUNNING branch's retries-exhausted -> FAILED transition + * deterministically, without submitting a real MR job that fails. Returns {@code null} when the + * job id cannot be resolved. The default implementation looks the job up on the real cluster. */ @VisibleForTesting public interface JobLookup { - Job getJob(Configuration configuration, String jobId) throws Exception; + JobStatus getJobStatus(Configuration configuration, String jobId) throws Exception; + } + + /** + * Immutable snapshot of a looked-up MR job's completion and success. The status must be captured + * while the owning {@link Cluster} is still open: a Job obtained from a Cluster cannot outlive it + * -- once the Cluster is closed its client is torn down and the Job's isComplete()/isSuccessful() + * can no longer issue their status RPCs. The seam therefore materializes both booleans up front + * and hands back this snapshot rather than a live Job whose Cluster the caller has no handle to + * close. + */ + @VisibleForTesting + public static final class JobStatus { + private final boolean complete; + private final boolean successful; + + public JobStatus(boolean complete, boolean successful) { + this.complete = complete; + this.successful = successful; + } + + public boolean isComplete() { + return complete; + } + + public boolean isSuccessful() { + return successful; + } } private static JobLookup defaultJobLookup() { + // Materialize the job's completion/success while the Cluster is open, then close it. Cluster + // opens a YARN/job-history client (RPC proxies, threads, file descriptors); the + // PARTIAL_PASS_RUNNING branch looks a job up on every ~60s monitor scan for the duration of the + // partial pass, so a leaked Cluster per scan would accumulate abandoned clients. Cluster is not + // AutoCloseable, and the returned Job cannot outlive its Cluster (a closed cluster's client can + // no longer serve status RPCs), so we snapshot the status here and close in a finally rather + // than return a live Job. return (configuration, jobId) -> { Cluster cluster = new Cluster(configuration); - return cluster.getJob(JobID.forName(jobId)); + try { + Job job = cluster.getJob(JobID.forName(jobId)); + return job == null ? null : new JobStatus(job.isComplete(), job.isSuccessful()); + } finally { + cluster.close(); + } }; } @@ -192,6 +235,12 @@ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) { Transform.updateTransformRecord(conn, systemTransformRecord, PTable.TransformStatus.COMPLETED); } + // Commit this post-cutover transition durably here rather than relying on the ServerTask + // commit at the tail of run(): the pointer swap in doCutover is already committed, so a + // throw between here and that tail commit would discard the buffered status upsert and + // leave a swapped-but-still-PENDING_CUTOVER record. Every other transition in this method + // commits explicitly for the same reason. + conn.commit(); } else if ( systemTransformRecord.getTransformStatus() .equals(PTable.TransformStatus.PENDING_PARTIAL_PASS.name()) @@ -238,7 +287,9 @@ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) { systemTransformRecord.getTransformStatus() .equals(PTable.TransformStatus.PARTIAL_PASS_RUNNING.name()) ) { - LOGGER.info("Partial pass is running, we will monitor {}", tableName); + // DEBUG, not INFO: this branch is re-entered on every ~60s monitor scan for the whole + // duration of the partial pass, so an INFO here would repeat the same line many times. + LOGGER.debug("Partial pass is running, we will monitor {}", tableName); // Monitor the partial-pass job to completion, then advance to COMPLETED. String jobId = systemTransformRecord.getTransformJobId(); // Defense-in-depth alongside the job-id clearing on the PENDING_PARTIAL_PASS -> @@ -260,10 +311,12 @@ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) { // resource-manager restart, etc.) is likewise unconfirmable. In every one of these cases // the pass cannot be confirmed successful, so it is routed through the same // retry-budgeted path as an outright failed job below -- never left to no-op forever. - Job job = jobId != null ? jobLookup.getJob(configuration, jobId) : null; + JobStatus job = jobId != null ? jobLookup.getJobStatus(configuration, jobId) : null; if (job != null && !job.isComplete()) { - // Partial pass is still running; re-evaluate on the next monitor scan. - LOGGER.info("Partial pass job is still running, we will keep monitoring {}", tableName); + // Partial pass is still running; re-evaluate on the next monitor scan. DEBUG, not INFO: + // this repeats every ~60s scan until the job completes. + LOGGER.debug("Partial pass job is still running, we will keep monitoring {}", + tableName); } else if (job != null && job.isSuccessful()) { Transform.updateTransformRecord(conn, systemTransformRecord, PTable.TransformStatus.COMPLETED); @@ -337,7 +390,7 @@ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) { // Monitor the job of transform tool and decide to retry String jobId = systemTransformRecord.getTransformJobId(); if (jobId != null) { - Job job = jobLookup.getJob(configuration, jobId); + JobStatus job = jobLookup.getJobStatus(configuration, jobId); if (job == null) { LOGGER.warn(String.format("Transform job with Id=%s is not found", jobId)); return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.SKIPPED, @@ -444,7 +497,11 @@ private long computePartialPassWaitMs(PhoenixConnection conn, String logicalTableName = SchemaUtil.getTableName(systemTransformRecord.getSchemaName(), systemTransformRecord.getLogicalTableName()); PTable logicalTable = conn.getTable(systemTransformRecord.getTenantId(), logicalTableName); - updateCacheFrequency = logicalTable.getUpdateCacheFrequency(); + long connectionDefaultUpdateCacheFrequency = + (Long) ConnectionProperty.UPDATE_CACHE_FREQUENCY.getValue(conn.getQueryServices().getProps() + .get(QueryServices.DEFAULT_UPDATE_CACHE_FREQUENCY_ATRRIB)); + updateCacheFrequency = effectiveUpdateCacheFrequency(logicalTable.getUpdateCacheFrequency(), + connectionDefaultUpdateCacheFrequency); } catch (Exception e) { LOGGER.warn("Could not resolve update cache frequency for the logical table; " + "falling back to the minimum partial-pass wait", e); @@ -452,6 +509,27 @@ private long computePartialPassWaitMs(PhoenixConnection conn, return boundedPartialPassWaitMs(updateCacheFrequency); } + /** + * Resolves the update-cache-frequency that governs how long clients may keep writing to the old + * physical pointer after cutover. A table with an explicit (non-default) UPDATE_CACHE_FREQUENCY + * pins every client's cache lifetime to that value. When the table carries no explicit value (its + * stored frequency equals the ALWAYS/default sentinel), a client instead caches for its own + * connection-level {@code phoenix.default.update.cache.frequency} -- the same precedence + * {@code MetaDataClient#avoidRpcToGetTable} applies. In that case the stored sentinel understates + * the real cache lifetime, so we substitute the server's configured default as the best available + * proxy for the clients' default; otherwise a fleet running a large or NEVER default would keep + * writing to the old pointer well past the 30-minute floor and strand those late rows as + * unverified. The result is fed through {@link #boundedPartialPassWaitMs}, which scales and + * clamps it. + */ + @VisibleForTesting + static long effectiveUpdateCacheFrequency(long tableUpdateCacheFrequency, + long connectionDefaultUpdateCacheFrequency) { + return tableUpdateCacheFrequency != QueryServicesOptions.DEFAULT_UPDATE_CACHE_FREQUENCY + ? tableUpdateCacheFrequency + : connectionDefaultUpdateCacheFrequency; + } + /** * Clamps and scales a raw update-cache-frequency into a bounded partial-pass wait. Extracted as a * pure function so the overflow-safety of the arithmetic can be unit-tested without a cluster. diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/CutoverLifecycleIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/CutoverLifecycleIT.java index 7a7249be3ac..23814f8c207 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/CutoverLifecycleIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/CutoverLifecycleIT.java @@ -36,7 +36,6 @@ import java.util.List; import java.util.Properties; import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; -import org.apache.hadoop.mapreduce.Job; import org.apache.phoenix.coprocessor.TaskRegionObserver; import org.apache.phoenix.coprocessor.tasks.TransformMonitorTask; import org.apache.phoenix.end2end.ParallelStatsDisabledIT; @@ -60,7 +59,6 @@ import org.junit.Before; import org.junit.Test; import org.junit.experimental.categories.Category; -import org.mockito.Mockito; /** * Integration tests for the cutover lifecycle: after the physical-table pointer swap the transform @@ -467,10 +465,8 @@ public void testMonitorDoesNotCompletePartialPassRunningWithStaleFullPassJob() t // below. // The gate keeps a full-type record from ever consulting the lookup, so the record stays put. // Reset via @After resetJobLookupForTesting(). - Job successfulJob = Mockito.mock(Job.class); - Mockito.when(successfulJob.isComplete()).thenReturn(true); - Mockito.when(successfulJob.isSuccessful()).thenReturn(true); - TransformMonitorTask.setJobLookupForTesting((configuration, jobId) -> successfulJob); + TransformMonitorTask.setJobLookupForTesting( + (configuration, jobId) -> new TransformMonitorTask.JobStatus(true, true)); try (PhoenixConnection conn = (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { @@ -574,10 +570,8 @@ public void testPartialPassRetriesExhaustedReachesFailed() throws Exception { // Inject a job that is complete and unsuccessful so the PARTIAL_PASS_RUNNING branch takes the // failure path. The retry count is seeded above the maximum so the branch treats retries as // exhausted and must transition to FAILED. - Job failedJob = Mockito.mock(Job.class); - Mockito.when(failedJob.isComplete()).thenReturn(true); - Mockito.when(failedJob.isSuccessful()).thenReturn(false); - TransformMonitorTask.setJobLookupForTesting((configuration, jobId) -> failedJob); + TransformMonitorTask.setJobLookupForTesting( + (configuration, jobId) -> new TransformMonitorTask.JobStatus(true, false)); try (PhoenixConnection conn = (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { @@ -783,10 +777,8 @@ public void testPartialPassRetryAdvancesRetryCount() throws Exception { // Now force the PARTIAL_PASS_RUNNING branch down the retry path: the injected job is complete // and unsuccessful, so the monitor retries the partial pass rather than completing it. - Job failedJob = Mockito.mock(Job.class); - Mockito.when(failedJob.isComplete()).thenReturn(true); - Mockito.when(failedJob.isSuccessful()).thenReturn(false); - TransformMonitorTask.setJobLookupForTesting((configuration, jobId) -> failedJob); + TransformMonitorTask.setJobLookupForTesting( + (configuration, jobId) -> new TransformMonitorTask.JobStatus(true, false)); // Seed the observable state of a running-but-failing partial pass, one retry below the // maximum so the failure path takes a genuine retry (not the already-exhausted diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/SystemTransformUpgradeIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/SystemTransformUpgradeIT.java new file mode 100644 index 00000000000..360990853bd --- /dev/null +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/transform/SystemTransformUpgradeIT.java @@ -0,0 +1,130 @@ +/* + * 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.transform; + +import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.sql.DriverManager; +import java.sql.SQLException; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest; +import org.apache.phoenix.end2end.ParallelStatsDisabledIT; +import org.apache.phoenix.jdbc.PhoenixConnection; +import org.apache.phoenix.query.ConnectionQueryServicesImpl; +import org.apache.phoenix.schema.PTable; +import org.apache.phoenix.schema.transform.SystemTransformRecord; +import org.apache.phoenix.schema.transform.Transform; +import org.apache.phoenix.util.PropertiesUtil; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +/** + * Verifies the SYSTEM.TRANSFORM column-add upgrade path in + * {@link ConnectionQueryServicesImpl#upgradeSystemTransform}. The two lifecycle columns + * (PENDING_PARTIAL_PASS_UNTIL_TS, CUTOVER_TS) are added by an unconditional, idempotent + * {@code addColumnsIfNotExists} rather than a timestamp gate, so that a SNAPSHOT cluster whose + * SYSTEM.TRANSFORM header already advanced past the version timestamp still gets the columns + * instead of stranding transform reads on a missing-column error. This test pins that contract: + * running the upgrade on a cluster that already carries the columns takes a snapshot, runs the + * ungated add as a safe no-op, and leaves the columns intact and readable. + *

+ * Note on coverage: a companion "re-add when the columns are missing" test is deliberately not + * included. Production never drops these columns, so the only way to simulate the missing state is + * a test-only DROP COLUMN; and because the upgrade re-adds at exactly + * {@code MIN_SYSTEM_TABLE_TIMESTAMP_5_4_0} -- which equals the SYSTEM-table upgrade guard ceiling + * {@code MIN_SYSTEM_TABLE_TIMESTAMP} -- any DROP that actually removes the original column cells + * must run at that same timestamp, so its delete marker masks the re-add PUT at equal timestamp + * (HBase delete-wins-at-equal-ts). That collision is a simulation artifact with no production + * analogue, so re-add coverage is left to the ungated {@code addColumnsIfNotExists} exercised by + * the TableAlreadyExists branch below. + *

+ * This exercises the same production upgrade helper the real EXECUTE UPGRADE flow calls, so it must + * boot its own mini-cluster: it mutates the shared SYSTEM.TRANSFORM schema and takes a snapshot of + * it. As with the other cutover-lifecycle integration tests, the heavy mini-cluster paths are run + * in CI (they can wedge on some local, e.g. Apple-Silicon, environments during region assignment). + */ +@Category(NeedsOwnMiniClusterTest.class) +public class SystemTransformUpgradeIT extends ParallelStatsDisabledIT { + + private final Properties testProps = PropertiesUtil.deepCopy(TEST_PROPERTIES); + + /** + * Re-running the upgrade on a cluster that already has the two columns must be a safe no-op: the + * CREATE throws {@link org.apache.phoenix.schema.TableAlreadyExistsException}, a snapshot is + * taken, and the idempotent add leaves the columns intact and readable. This is the fresh / + * SNAPSHOT-cluster path that the gate removal exists to keep safe. + */ + @Test + public void testUpgradeIsIdempotentWhenColumnsAlreadyPresent() throws Exception { + try (PhoenixConnection conn = + (PhoenixConnection) DriverManager.getConnection(getUrl(), testProps)) { + conn.setAutoCommit(true); + ConnectionQueryServicesImpl cqs = (ConnectionQueryServicesImpl) conn.getQueryServices(); + + Map snapshotMap = new HashMap<>(); + // The columns are already present on a fresh cluster, so this must not throw. + cqs.upgradeSystemTransform(conn, snapshotMap); + assertSnapshotTakenForTransform(snapshotMap); + + conn.getQueryServices().clearCache(); + assertColumnsRoundTrip(conn); + } + } + + /** Asserts the upgrade took a snapshot of SYSTEM.TRANSFORM before altering it. */ + private static void assertSnapshotTakenForTransform(Map snapshotMap) { + assertFalse("The upgrade must snapshot SYSTEM.TRANSFORM before adding columns", + snapshotMap.isEmpty()); + assertTrue( + "The snapshot map must key on the SYSTEM.TRANSFORM physical name, was " + snapshotMap, + snapshotMap.keySet().stream().anyMatch(k -> k.contains("TRANSFORM"))); + } + + /** + * Round-trips a transform record through both new BIGINT columns to prove they are present, + * writable, and readable after the upgrade. + */ + private void assertColumnsRoundTrip(PhoenixConnection conn) throws SQLException { + String logicalTableName = generateUniqueName(); + long pendingUntil = 4242L; + long cutover = 1000L; + + SystemTransformRecord.SystemTransformBuilder builder = + new SystemTransformRecord.SystemTransformBuilder(); + builder.setLogicalTableName(logicalTableName); + builder.setNewPhysicalTableName(logicalTableName + "_1"); + builder.setTransformType(PTable.TransformType.METADATA_TRANSFORM); + builder.setTransformStatus(PTable.TransformStatus.PENDING_PARTIAL_PASS.name()); + builder.setPendingPartialPassUntilTs(pendingUntil); + builder.setCutoverTs(cutover); + Transform.upsertTransform(builder.build(), conn); + + SystemTransformRecord readBack = + Transform.getTransformRecord(null, logicalTableName, null, null, conn); + assertNotNull("The transform record must read back after the upgrade", readBack); + assertEquals("PENDING_PARTIAL_PASS_UNTIL_TS must round-trip", Long.valueOf(pendingUntil), + readBack.getPendingPartialPassUntilTs()); + assertEquals("CUTOVER_TS must round-trip", Long.valueOf(cutover), readBack.getCutoverTs()); + } +} diff --git a/phoenix-core/src/test/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTaskWaitTest.java b/phoenix-core/src/test/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTaskWaitTest.java index 44d33eecc02..84812cd69b5 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTaskWaitTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTaskWaitTest.java @@ -21,6 +21,7 @@ import static org.junit.Assert.assertTrue; import java.sql.Timestamp; +import org.apache.phoenix.query.QueryServicesOptions; import org.apache.phoenix.schema.transform.SystemTransformRecord; import org.apache.phoenix.util.EnvironmentEdgeManager; import org.apache.phoenix.util.ManualEnvironmentEdge; @@ -81,6 +82,40 @@ public void testFrequencyAtOrAboveCeilingClampsToCeiling() { TransformMonitorTask.boundedPartialPassWaitMs(MAX_WAIT_MS + 1)); } + @Test + public void testEffectiveUpdateCacheFrequencyPrefersExplicitTableValue() { + // A table with an explicit (non-default) UPDATE_CACHE_FREQUENCY pins every client's cache + // lifetime, so it wins outright and the connection default is ignored. + long tableUcf = 90L * 60L * 1000L; + long connectionDefault = 5L * 60L * 1000L; + assertEquals("An explicit table frequency must be used verbatim", tableUcf, + TransformMonitorTask.effectiveUpdateCacheFrequency(tableUcf, connectionDefault)); + } + + @Test + public void testEffectiveUpdateCacheFrequencyFallsBackToConnectionDefaultForSentinel() { + // A table carrying the ALWAYS/default sentinel understates the real cache lifetime: clients + // fall back to their connection-level phoenix.default.update.cache.frequency, so we must too. + long connectionDefault = 90L * 60L * 1000L; + assertEquals("The sentinel table frequency must defer to the connection default", + connectionDefault, TransformMonitorTask.effectiveUpdateCacheFrequency( + QueryServicesOptions.DEFAULT_UPDATE_CACHE_FREQUENCY, connectionDefault)); + } + + @Test + public void testSentinelTableWithLargeConnectionDefaultDrivesWaitPastFloor() { + // End-to-end of the fix: a sentinel-UCF table under a large connection default must produce a + // wait derived from that default (scaled by 1.10), not collapse to the 30-minute floor as it + // did when the monitor read the stored sentinel (0) directly. + long connectionDefault = 90L * 60L * 1000L; + long effective = TransformMonitorTask.effectiveUpdateCacheFrequency( + QueryServicesOptions.DEFAULT_UPDATE_CACHE_FREQUENCY, connectionDefault); + long wait = TransformMonitorTask.boundedPartialPassWaitMs(effective); + assertEquals("A large connection default must drive the wait above the floor", + (long) (connectionDefault * 1.10), wait); + assertTrue("The resulting wait must exceed the minimum floor", wait > MIN_WAIT_MS); + } + private static SystemTransformRecord recordWith(Long cutoverTs, Long lastStateTs) { SystemTransformRecord.SystemTransformBuilder builder = new SystemTransformRecord.SystemTransformBuilder();