PHOENIX-7906 :- Add UNKNOWN forward-compat sentinel for TransformType

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

@@ -267,7 +267,16 @@ public String toString() {
 
   public enum TransformType {
     METADATA_TRANSFORM((byte) 1),
-    METADATA_TRANSFORM_PARTIAL((byte) 2);
+    METADATA_TRANSFORM_PARTIAL((byte) 2),
+    /**
+     * Forward-compatibility sentinel returned by {@link #fromSerializedValue(int)} when the
+     * serialized byte read from SYSTEM.TRANSFORM does not correspond to any known
+     * {@code TransformType} constant. This allows an older binary to deserialize a row written by a
+     * newer binary that introduced an additional transform type without crashing the calling code
+     * path; callers that observe {@code UNKNOWN} must handle it defensively (skip, block, or
+     * fail-fast with an operator-readable message) rather than treat it as a known type.
+     */
+    UNKNOWN((byte) -1);
 
     private final byte[] byteValue;
     private final int serializedValue;
@@ -289,11 +298,20 @@ public static TransformType getDefault() {
       return METADATA_TRANSFORM;
     }
 
+    /**
+     * Resolve a serialized transform type to its enum constant. Returns {@link #UNKNOWN} for any
+     * value that does not match a known constant (including the sentinel value itself), so that
+     * rows written by a newer binary do not crash an older binary. The sentinel constant
+     * {@code UNKNOWN} itself is never returned via its own serialized value through normal write
+     * paths — it is only produced when the persisted value is unrecognized.
+     */
     public static TransformType fromSerializedValue(int serializedValue) {
-      if (serializedValue < 1 || serializedValue > TransformType.values().length) {
-        throw new IllegalArgumentException("Invalid TransformType " + serializedValue);
+      for (TransformType type : TransformType.values()) {
+        if (type != UNKNOWN && type.serializedValue == serializedValue) {
+          return type;
+        }
       }
-      return TransformType.values()[serializedValue - 1];
+      return UNKNOWN;
     }
 
     public static TransformType getPartialTransform(TransformType transformType) {

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

@@ -152,6 +152,7 @@ public static boolean checkIsTransformNeeded(MetaDataClient.MetaProperties metaP
     if (isTransformNeeded) {
       SystemTransformRecord existingTransform =
         getTransformRecord(schemaName, logicalTableName, parentTableName, tenantId, connection);
+      enforceKnownTransformType(existingTransform, schemaName, logicalTableName);
       if (existingTransform != null && existingTransform.isActive()) {
         throw new SQLExceptionInfo.Builder(
           SQLExceptionCode.CANNOT_TRANSFORM_ALREADY_TRANSFORMING_TABLE)
@@ -162,6 +163,30 @@ public static boolean checkIsTransformNeeded(MetaDataClient.MetaProperties metaP
     return isTransformNeeded;
   }
 
+  /**
+   * Forward-compatibility guard: throw
+   * {@link SQLExceptionCode#CANNOT_TRANSFORM_ALREADY_TRANSFORMING_TABLE} when an existing
+   * SYSTEM.TRANSFORM row carries a transform type this binary does not recognise. Such a row was
+   * very likely written by a newer binary; this binary cannot reason about whether the on-disk
+   * schema state is consistent with a new ALTER, so it blocks defensively regardless of the row's
+   * status — even a {@code COMPLETED} row of an unknown type may have left the schema in a state we
+   * do not understand. Package-private to allow direct unit testing without a live SYSTEM.TRANSFORM
+   * table.
+   */
+  static void enforceKnownTransformType(SystemTransformRecord existingTransform, String schemaName,
+    String logicalTableName) throws SQLException {
+    if (
+      existingTransform != null
+        && existingTransform.getTransformType() == PTable.TransformType.UNKNOWN
+    ) {
+      throw new SQLExceptionInfo.Builder(
+        SQLExceptionCode.CANNOT_TRANSFORM_ALREADY_TRANSFORMING_TABLE)
+          .setMessage(" Existing transform record has an unrecognized transform type "
+            + "(likely written by a newer binary); blocking new transform on this table ")
+          .setSchemaName(schemaName).setTableName(logicalTableName).build().buildException();
+    }
+  }
+
   protected static SystemTransformRecord getTransformRecord(PhoenixConnection connection,
     PTableType tableType, PName schemaName, PName tableName, PName dataTableName, PName tenantId,
     PName parentLogicalName) throws SQLException {

phoenix-core-server/src/main/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTask.java

@@ -78,6 +78,14 @@ public TaskRegionObserver.TaskResult run(Task.TaskRecord taskRecord) {
         return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.FAIL,
           "No transform record is found");
       }
+      // Forward-compatibility guard: if the persisted transform type was written by a newer binary
+      // that introduced a new TransformType constant, this binary will see UNKNOWN. Skip the row
+      // without retrying — the monitor task cannot make a safe decision about how to advance it.
+      TaskRegionObserver.TaskResult unknownTypeResult =
+        skipIfUnknownTransformType(systemTransformRecord);
+      if (unknownTypeResult != null) {
+        return unknownTypeResult;
+      }
       String tableName = SchemaUtil.getTableName(systemTransformRecord.getSchemaName(),
         systemTransformRecord.getLogicalTableName());
 
@@ -218,4 +226,23 @@ public TaskRegionObserver.TaskResult checkCurrentResult(Task.TaskRecord taskReco
     // We don't need to check MR job result here since the job itself changes task state.
     return null;
   }
+
+  /**
+   * If the supplied transform record carries an {@link PTable.TransformType#UNKNOWN} transform type
+   * (typically because the row was written by a newer binary that introduced a new transform type
+   * this binary does not understand), return a {@link TaskRegionObserver.TaskResultCode#SKIPPED}
+   * result with an operator-readable message. Otherwise return {@code null} to signal that normal
+   * processing should continue. Package-private to keep the gate independently testable without
+   * standing up a full coprocessor environment.
+   */
+  static TaskRegionObserver.TaskResult
+    skipIfUnknownTransformType(SystemTransformRecord systemTransformRecord) {
+    if (systemTransformRecord.getTransformType() == PTable.TransformType.UNKNOWN) {
+      String msg = "Skipping transform monitor for record with UNKNOWN transform type; "
+        + "likely written by a newer binary. " + systemTransformRecord.getString();
+      LOGGER.warn(msg);
+      return new TaskRegionObserver.TaskResult(TaskRegionObserver.TaskResultCode.SKIPPED, msg);
+    }
+    return null;
+  }
 }

phoenix-core-server/src/main/java/org/apache/phoenix/mapreduce/transform/TransformTool.java

@@ -440,6 +440,21 @@ public void validateTransform(PTable argPDataTable, PTable argIndexTable,
         "ALTER statement has not been run and the transform has not been created for this table");
     }
 
+    // Forward-compatibility guard: the persisted transform type does not correspond to any
+    // TransformType known to this binary. Fail fast with an operator-readable message rather than
+    // attempt a transform we cannot reason about. The operator must upgrade this binary to a
+    // version that recognises the type, or remove the SYSTEM.TRANSFORM row, before re-running.
+    if (transformRecord.getTransformType() == PTable.TransformType.UNKNOWN) {
+      throw new IllegalStateException(String.format(
+        "TransformTool cannot run for SYSTEM.TRANSFORM record with an unrecognized transform type. "
+          + "This row was likely written by a newer Phoenix binary that introduced a new "
+          + "TransformType this binary does not understand. Upgrade this binary to a version that "
+          + "supports the new type, or remove the SYSTEM.TRANSFORM row, before retrying. "
+          + "schema=%s, logicalTable=%s, newPhysicalTable=%s, tenantId=%s",
+        transformRecord.getSchemaName(), transformRecord.getLogicalTableName(),
+        transformRecord.getNewPhysicalTableName(), transformRecord.getTenantId()));
+    }
+
     if (pDataTable != null && pIndexTable != null) {
       if (!IndexTool.isValidIndexTable(connection, qDataTable, indexTable, tenantId)) {
         throw new IllegalArgumentException(String

phoenix-core-server/src/main/java/org/apache/phoenix/schema/transform/Transform.java

@@ -138,6 +138,17 @@ public static void removeTransformRecord(SystemTransformRecord transformRecord,
    */
   public static void doCutover(PhoenixConnection connection,
     SystemTransformRecord systemTransformRecord) throws Exception {
+    // Forward-compatibility guard: a row written by a newer binary may carry a transform type this
+    // binary does not recognize. Skipping the cutover (rather than throwing) keeps the monitor task
+    // alive so that other transforms are not starved while the operator decides how to act.
+    if (systemTransformRecord.getTransformType() == PTable.TransformType.UNKNOWN) {
+      LOGGER.warn(
+        "Skipping cutover for transform with UNKNOWN transform type (likely written by a newer "
+          + "binary). Schema={}, logicalTable={}, newPhysicalTable={}, tenantId={}",
+        systemTransformRecord.getSchemaName(), systemTransformRecord.getLogicalTableName(),
+        systemTransformRecord.getNewPhysicalTableName(), systemTransformRecord.getTenantId());
+      return;
+    }
     String tenantId = systemTransformRecord.getTenantId();
     String schema = systemTransformRecord.getSchemaName();
     String tableName = systemTransformRecord.getLogicalTableName();
@@ -507,6 +518,28 @@ public static void doForceCutover(Connection connection, Configuration configura
     SystemTransformRecord transformRecord = getTransformRecord(schemaName, logicaTableName,
       logicalParentName, tenantId, phoenixConnection);
     Transform.doCutover(phoenixConnection, transformRecord);
+    finishForceCutover(phoenixConnection, transformRecord);
+  }
+
+  /**
+   * Marks the transform record COMPLETED and commits after a successful force cutover. This must be
+   * skipped when the record carries {@link PTable.TransformType#UNKNOWN}: in that case
+   * {@link #doCutover(PhoenixConnection, SystemTransformRecord)} short-circuited without performing
+   * any cutover, so marking the record COMPLETED here would falsely report success for a transform
+   * this binary could not run. Mirrors the skip-with-warn behaviour of the other forward-compat
+   * guards. Package-private so the gate is independently testable without standing up a Reducer.
+   */
+  static void finishForceCutover(PhoenixConnection phoenixConnection,
+    SystemTransformRecord transformRecord) throws SQLException {
+    if (transformRecord.getTransformType() == PTable.TransformType.UNKNOWN) {
+      LOGGER.warn(
+        "Skipping COMPLETED update for force cutover of transform with UNKNOWN transform type "
+          + "(likely written by a newer binary); cutover did not run. Schema={}, logicalTable={}, "
+          + "newPhysicalTable={}, tenantId={}",
+        transformRecord.getSchemaName(), transformRecord.getLogicalTableName(),
+        transformRecord.getNewPhysicalTableName(), transformRecord.getTenantId());
+      return;
+    }
     updateTransformRecord(phoenixConnection, transformRecord, PTable.TransformStatus.COMPLETED);
     phoenixConnection.commit();
   }

phoenix-core/src/test/java/org/apache/phoenix/coprocessor/tasks/TransformMonitorTaskUnknownTypeTest.java

@@ -0,0 +1,71 @@
+/*
+ * 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.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.phoenix.coprocessor.TaskRegionObserver;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.transform.SystemTransformRecord;
+import org.junit.Test;
+
+/**
+ * Unit test for the forward-compatibility gate in
+ * {@link TransformMonitorTask#skipIfUnknownTransformType(SystemTransformRecord)}: when the
+ * persisted transform type is unrecognised, the monitor task must short-circuit with a SKIPPED
+ * result rather than treat the row as a known transform and either crash or push it into a wrong
+ * state branch.
+ */
+public class TransformMonitorTaskUnknownTypeTest {
+
+  @Test
+  public void skipIfUnknownTransformTypeReturnsSkippedForUnknownType() {
+    SystemTransformRecord record = mock(SystemTransformRecord.class);
+    when(record.getTransformType()).thenReturn(PTable.TransformType.UNKNOWN);
+    when(record.getString()).thenReturn("schema=S logicalTable=T newPhysical=S.T_2");
+
+    TaskRegionObserver.TaskResult result = TransformMonitorTask.skipIfUnknownTransformType(record);
+
+    assertNotNull("UNKNOWN type must produce a non-null TaskResult", result);
+    assertEquals(TaskRegionObserver.TaskResultCode.SKIPPED, result.getResultCode());
+  }
+
+  @Test
+  public void skipIfUnknownTransformTypeReturnsNullForKnownType() {
+    SystemTransformRecord record = mock(SystemTransformRecord.class);
+    when(record.getTransformType()).thenReturn(PTable.TransformType.METADATA_TRANSFORM);
+
+    TaskRegionObserver.TaskResult result = TransformMonitorTask.skipIfUnknownTransformType(record);
+
+    assertNull("Known transform types must let normal processing proceed", result);
+  }
+
+  @Test
+  public void skipIfUnknownTransformTypeReturnsNullForKnownPartialType() {
+    SystemTransformRecord record = mock(SystemTransformRecord.class);
+    when(record.getTransformType()).thenReturn(PTable.TransformType.METADATA_TRANSFORM_PARTIAL);
+
+    TaskRegionObserver.TaskResult result = TransformMonitorTask.skipIfUnknownTransformType(record);
+
+    assertNull("Partial transform type must let normal processing proceed", result);
+  }
+}

phoenix-core/src/test/java/org/apache/phoenix/mapreduce/transform/TransformToolValidateTransformUnknownTypeTest.java

@@ -0,0 +1,78 @@
+/*
+ * 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.mapreduce.transform;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.schema.transform.SystemTransformRecord;
+import org.junit.Test;
+
+/**
+ * Unit test for the forward-compatibility fail-fast guard in
+ * {@link TransformTool#validateTransform}: when the SYSTEM.TRANSFORM record carries an unrecognised
+ * transform type, the tool must refuse to run with an operator-readable
+ * {@link IllegalStateException} rather than attempt a transform it cannot reason about.
+ */
+public class TransformToolValidateTransformUnknownTypeTest {
+
+  @Test
+  public void validateTransformFailsFastForUnknownTransformType() throws Exception {
+    PTable dataTable = mock(PTable.class);
+    when(dataTable.getType()).thenReturn(PTableType.TABLE);
+    when(dataTable.isTransactional()).thenReturn(false);
+
+    SystemTransformRecord record = mock(SystemTransformRecord.class);
+    when(record.getTransformType()).thenReturn(PTable.TransformType.UNKNOWN);
+    when(record.getSchemaName()).thenReturn("S");
+    when(record.getLogicalTableName()).thenReturn("T");
+    when(record.getNewPhysicalTableName()).thenReturn("S.T_2");
+    when(record.getTenantId()).thenReturn(null);
+
+    TransformTool tool = new TransformTool();
+    try {
+      tool.validateTransform(dataTable, null, record);
+      fail("Expected IllegalStateException for UNKNOWN transform type");
+    } catch (IllegalStateException e) {
+      String msg = e.getMessage();
+      assertTrue("Expected operator-readable message; got: " + msg,
+        msg != null && msg.contains("unrecognized transform type"));
+      assertTrue("Expected message to mention the schema; got: " + msg, msg.contains("S"));
+      assertTrue("Expected message to mention the logical table; got: " + msg, msg.contains("T"));
+    }
+  }
+
+  @Test
+  public void validateTransformAllowsKnownTransformType() throws Exception {
+    PTable dataTable = mock(PTable.class);
+    when(dataTable.getType()).thenReturn(PTableType.TABLE);
+    when(dataTable.isTransactional()).thenReturn(false);
+
+    SystemTransformRecord record = mock(SystemTransformRecord.class);
+    when(record.getTransformType()).thenReturn(PTable.TransformType.METADATA_TRANSFORM);
+
+    TransformTool tool = new TransformTool();
+    // No exception — a known transform type passes the forward-compat guard. The remaining
+    // index-table-shape checks inside validateTransform are skipped because argIndexTable is null.
+    tool.validateTransform(dataTable, null, record);
+  }
+}