PHOENIX-7788 CREATE TABLE fails with TableNotEnabledException when un…

phoenix-core-client/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java

@@ -165,6 +165,7 @@
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableExistsException;
 import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.TableNotDisabledException;
 import org.apache.hadoop.hbase.TableNotEnabledException;
 import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.Append;
@@ -2459,6 +2460,53 @@ private void disableTable(Admin admin, TableName tableName) throws IOException {
     }
   }
 
+  private void enableTable(Admin admin, TableName tableName) throws IOException {
+    try {
+      admin.enableTable(tableName);
+    } catch (TableNotDisabledException e) {
+      LOGGER.info("Table already enabled, continuing with next steps", e);
+    }
+  }
+
+  /**
+   * PHOENIX-7788: re-enable a disabled physical HBase table if SYSTEM.CATALOG has no row for it. If
+   * metadata exists, leave it disabled — an admin may have disabled the registered table.
+   */
+  private void reenableOrphanedDisabledHBaseTable(byte[] schemaBytes, byte[] tableBytes,
+    boolean isNamespaceMapped, PTableType tableType) throws SQLException {
+    if (tableType != PTableType.TABLE) {
+      return;
+    }
+    TableName physicalTableName = TableName.valueOf(
+      SchemaUtil.getPhysicalHBaseTableName(schemaBytes, tableBytes, isNamespaceMapped).getBytes());
+    try (Admin admin = getAdmin()) {
+      if (!AdminUtilWithFallback.tableExists(admin, physicalTableName)) {
+        return;
+      }
+      if (!admin.isTableDisabled(physicalTableName)) {
+        return;
+      }
+      MetaDataMutationResult result = getTable(null, schemaBytes, tableBytes,
+        HConstants.LATEST_TIMESTAMP, HConstants.LATEST_TIMESTAMP);
+      if (result.getMutationCode() != MutationCode.TABLE_NOT_FOUND) {
+        LOGGER.info(
+          "Physical HBase table {} is disabled but SYSTEM.CATALOG has metadata for it "
+            + "(mutation code {}); leaving it disabled to preserve any intentional admin action.",
+          physicalTableName, result.getMutationCode());
+        return;
+      }
+      LOGGER.info("Re-enabling orphaned disabled HBase table {} during CREATE TABLE",
+        physicalTableName);
+      enableTable(admin, physicalTableName);
+    } catch (IOException e) {
+      throw ClientUtil.parseServerException(e);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new SQLExceptionInfo.Builder(SQLExceptionCode.INTERRUPTED_EXCEPTION).setRootCause(e)
+        .build().buildException();
+    }
+  }
+
   private boolean ensureViewIndexTableDropped(byte[] physicalTableName, long timestamp)
     throws SQLException {
     byte[] physicalIndexName = MetaDataUtil.getViewIndexPhysicalName(physicalTableName);
@@ -2557,6 +2605,9 @@ public MetaDataMutationResult createTable(final List<Mutation> tableMetaData,
       (tableType != PTableType.CDC) && ((tableType == PTableType.VIEW && physicalTableName != null)
         || (tableType != PTableType.VIEW && (physicalTableName == null || localIndexTable)))
     ) {
+      // PHOENIX-7788: recover from an orphaned disabled physical table before ensureTableCreated
+      // runs modifyTable on it. See the helper for the metadata-preserving contract.
+      reenableOrphanedDisabledHBaseTable(schemaBytes, tableBytes, isNamespaceMapped, tableType);
       // For views this will ensure that metadata already exists
       // For tables and indexes, this will create the metadata if it doesn't already exist
       ensureTableCreated(physicalTableNameBytes, null, tableType, tableProps, families, splits,

phoenix-core/src/it/java/org/apache/phoenix/end2end/CreateTableIT.java

@@ -1739,6 +1739,93 @@ public void testCreateTableWithNoVerify() throws SQLException, IOException, Inte
     }
   }
 
+  @Test
+  public void testCreateTableReenablesExistingDisabledHBaseTable() throws Exception {
+    String tableName = generateUniqueName();
+    String ddl = "CREATE TABLE " + tableName
+      + " (K VARCHAR NOT NULL PRIMARY KEY, V VARCHAR) COLUMN_ENCODED_BYTES=NONE";
+    Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+
+    try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+      conn.createStatement().execute(ddl);
+    }
+
+    ConnectionQueryServices services = driver.getConnectionQueryServices(getUrl(), props);
+    TableName hbaseTableName = TableName.valueOf(tableName);
+
+    // Simulate the "failed drop" state: Phoenix metadata is gone but the physical HBase
+    // table still exists and has been left disabled.
+    try (Admin admin = services.getAdmin();
+      Connection conn = DriverManager.getConnection(getUrl(), props)) {
+      admin.disableTable(hbaseTableName);
+      assertTrue(admin.isTableDisabled(hbaseTableName));
+
+      conn.createStatement()
+        .executeUpdate("DELETE FROM SYSTEM.CATALOG WHERE TABLE_NAME = '" + tableName + "'");
+      conn.commit();
+      conn.unwrap(PhoenixConnection.class).getQueryServices().clearCache();
+    }
+
+    try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+      conn.createStatement().execute(ddl);
+    }
+
+    try (Admin admin = services.getAdmin()) {
+      assertFalse("HBase table should have been re-enabled by CREATE TABLE",
+        admin.isTableDisabled(hbaseTableName));
+      assertTrue(admin.isTableEnabled(hbaseTableName));
+    }
+
+    try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+      conn.setAutoCommit(true);
+      conn.createStatement().execute("UPSERT INTO " + tableName + " VALUES ('a', 'b')");
+      try (ResultSet rs =
+        conn.createStatement().executeQuery("SELECT V FROM " + tableName + " WHERE K = 'a'")) {
+        assertTrue(rs.next());
+        assertEquals("b", rs.getString(1));
+        assertFalse(rs.next());
+      }
+    }
+  }
+
+  // Test for PHOENIX-7788: guard must be gated on metadata absence, not on physical state
+  // alone, so an intentional admin disable of a Phoenix-registered table is not silently
+  // undone by CREATE TABLE IF NOT EXISTS.
+  @Test
+  public void testCreateTableIfNotExistsDoesNotReenableDisabledTableWithMetadata()
+    throws Exception {
+    String tableName = generateUniqueName();
+    String ddl = "CREATE TABLE " + tableName
+      + " (K VARCHAR NOT NULL PRIMARY KEY, V VARCHAR) COLUMN_ENCODED_BYTES=NONE";
+    Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+
+    try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+      conn.createStatement().execute(ddl);
+    }
+
+    ConnectionQueryServices services = driver.getConnectionQueryServices(getUrl(), props);
+    TableName hbaseTableName = TableName.valueOf(tableName);
+
+    // Simulate an admin disabling a registered Phoenix table for maintenance. Metadata
+    // rows in SYSTEM.CATALOG are left intact.
+    try (Admin admin = services.getAdmin()) {
+      admin.disableTable(hbaseTableName);
+      assertTrue(admin.isTableDisabled(hbaseTableName));
+    }
+
+    try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+      conn.unwrap(PhoenixConnection.class).getQueryServices().clearCache();
+      conn.createStatement().execute("CREATE TABLE IF NOT EXISTS " + tableName
+        + " (K VARCHAR NOT NULL PRIMARY KEY, V VARCHAR) COLUMN_ENCODED_BYTES=NONE");
+    }
+
+    try (Admin admin = services.getAdmin()) {
+      assertTrue(
+        "CREATE TABLE IF NOT EXISTS must not re-enable a disabled table with existing metadata",
+        admin.isTableDisabled(hbaseTableName));
+    }
+  }
+
   public static long verifyLastDDLTimestamp(String tableFullName, long startTS, Connection conn)
     throws SQLException {
     long endTS = EnvironmentEdgeManager.currentTimeMillis();

phoenix-core/src/test/java/org/apache/phoenix/query/ConnectionQueryServicesImplTest.java

@@ -35,16 +35,21 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 import java.nio.charset.StandardCharsets;
 import java.sql.SQLException;
 import java.util.Collections;
@@ -58,6 +63,7 @@
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HRegionLocation;
 import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.TableNotDisabledException;
 import org.apache.hadoop.hbase.TableNotEnabledException;
 import org.apache.hadoop.hbase.TableNotFoundException;
 import org.apache.hadoop.hbase.client.Admin;
@@ -69,11 +75,16 @@
 import org.apache.hadoop.hbase.client.TableDescriptor;
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
 import org.apache.phoenix.SystemExitRule;
+import org.apache.phoenix.coprocessorclient.MetaDataProtocol.MetaDataMutationResult;
+import org.apache.phoenix.coprocessorclient.MetaDataProtocol.MutationCode;
 import org.apache.phoenix.exception.PhoenixIOException;
 import org.apache.phoenix.jdbc.ConnectionInfo;
 import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
 import org.apache.phoenix.monitoring.GlobalClientMetrics;
 import org.apache.phoenix.schema.PMetaData;
+import org.apache.phoenix.schema.PName;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.util.ByteUtil;
 import org.apache.phoenix.util.ReadOnlyProps;
 import org.junit.Before;
 import org.junit.ClassRule;
@@ -410,6 +421,126 @@ public void testDropTablesTableEnabled() throws Exception {
     verify(mockConn).getAdmin();
   }
 
+  @Test
+  public void testEnableTableAlreadyEnabledSwallowsException() throws Exception {
+    // PHOENIX-7788: enableTable helper must swallow TableNotDisabledException,
+    // so a concurrent client that already re-enabled the table does not fail the CREATE.
+    TableName tableName = TableName.valueOf("TEST_TABLE");
+    doThrow(new TableNotDisabledException(tableName)).when(mockAdmin).enableTable(tableName);
+    invokeEnableTable(mockCqs, mockAdmin, tableName);
+    verify(mockAdmin, Mockito.times(1)).enableTable(tableName);
+  }
+
+  @Test
+  public void testEnableTablePropagatesOtherIOException() throws Exception {
+    TableName tableName = TableName.valueOf("TEST_TABLE");
+    IOException expected = new IOException("boom");
+    doThrow(expected).when(mockAdmin).enableTable(tableName);
+    try {
+      invokeEnableTable(mockCqs, mockAdmin, tableName);
+      fail("Expected IOException to propagate");
+    } catch (InvocationTargetException e) {
+      assertSame(expected, e.getCause());
+    }
+  }
+
+  private static void invokeEnableTable(ConnectionQueryServicesImpl cqs, Admin admin,
+    TableName tableName) throws Exception {
+    Method m = ConnectionQueryServicesImpl.class.getDeclaredMethod("enableTable", Admin.class,
+      TableName.class);
+    m.setAccessible(true);
+    m.invoke(cqs, admin, tableName);
+  }
+
+  @Test
+  public void testReenableOrphanedDisabledHBaseTableSkipsNonTableTypes() throws Exception {
+    // PHOENIX-7788: helper only runs for PTableType.TABLE; other types must return early.
+    byte[] name = "TEST_TABLE".getBytes(StandardCharsets.UTF_8);
+    invokeReenableOrphanedDisabledHBaseTable(mockCqs, ByteUtil.EMPTY_BYTE_ARRAY, name, false,
+      PTableType.VIEW);
+    invokeReenableOrphanedDisabledHBaseTable(mockCqs, ByteUtil.EMPTY_BYTE_ARRAY, name, false,
+      PTableType.INDEX);
+    verify(mockConn, never()).getAdmin();
+  }
+
+  @Test
+  public void testReenableOrphanedDisabledHBaseTableSkipsWhenTableDoesNotExist() throws Exception {
+    byte[] name = "TEST_TABLE".getBytes(StandardCharsets.UTF_8);
+    TableName physical = TableName.valueOf(name);
+    when(mockConn.getAdmin()).thenReturn(mockAdmin);
+    when(mockAdmin.tableExists(physical)).thenReturn(false);
+    invokeReenableOrphanedDisabledHBaseTable(mockCqs, ByteUtil.EMPTY_BYTE_ARRAY, name, false,
+      PTableType.TABLE);
+    verify(mockAdmin, never()).isTableDisabled(any());
+    verify(mockAdmin, never()).enableTable(any());
+  }
+
+  @Test
+  public void testReenableOrphanedDisabledHBaseTableSkipsWhenTableEnabled() throws Exception {
+    byte[] name = "TEST_TABLE".getBytes(StandardCharsets.UTF_8);
+    TableName physical = TableName.valueOf(name);
+    when(mockConn.getAdmin()).thenReturn(mockAdmin);
+    when(mockAdmin.tableExists(physical)).thenReturn(true);
+    when(mockAdmin.isTableDisabled(physical)).thenReturn(false);
+    invokeReenableOrphanedDisabledHBaseTable(mockCqs, ByteUtil.EMPTY_BYTE_ARRAY, name, false,
+      PTableType.TABLE);
+    verify(mockAdmin, never()).enableTable(any());
+  }
+
+  @Test
+  public void testReenableOrphanedDisabledHBaseTableLeavesTableDisabledWhenMetadataPresent()
+    throws Exception {
+    // PHOENIX-7788: disabled physical table with existing SYSTEM.CATALOG metadata must NOT be
+    // re-enabled; an admin may have disabled a registered table intentionally, and
+    // CREATE TABLE IF NOT EXISTS must preserve that.
+    byte[] name = "TEST_TABLE".getBytes(StandardCharsets.UTF_8);
+    TableName physical = TableName.valueOf(name);
+    when(mockConn.getAdmin()).thenReturn(mockAdmin);
+    when(mockAdmin.tableExists(physical)).thenReturn(true);
+    when(mockAdmin.isTableDisabled(physical)).thenReturn(true);
+    MetaDataMutationResult existing =
+      new MetaDataMutationResult(MutationCode.TABLE_ALREADY_EXISTS, 0L, null);
+    doReturn(existing).when(mockCqs).getTable(Mockito.<PName> any(), any(byte[].class),
+      any(byte[].class), anyLong(), anyLong());
+    invokeReenableOrphanedDisabledHBaseTable(mockCqs, ByteUtil.EMPTY_BYTE_ARRAY, name, false,
+      PTableType.TABLE);
+    verify(mockAdmin, never()).enableTable(any());
+  }
+
+  @Test
+  public void testReenableOrphanedDisabledHBaseTableReenablesOrphanedTable() throws Exception {
+    byte[] name = "TEST_TABLE".getBytes(StandardCharsets.UTF_8);
+    TableName physical = TableName.valueOf(name);
+    when(mockConn.getAdmin()).thenReturn(mockAdmin);
+    when(mockAdmin.tableExists(physical)).thenReturn(true);
+    when(mockAdmin.isTableDisabled(physical)).thenReturn(true);
+    MetaDataMutationResult notFound =
+      new MetaDataMutationResult(MutationCode.TABLE_NOT_FOUND, 0L, null);
+    doReturn(notFound).when(mockCqs).getTable(Mockito.<PName> any(), any(byte[].class),
+      any(byte[].class), anyLong(), anyLong());
+    doNothing().when(mockAdmin).enableTable(physical);
+    invokeReenableOrphanedDisabledHBaseTable(mockCqs, ByteUtil.EMPTY_BYTE_ARRAY, name, false,
+      PTableType.TABLE);
+    verify(mockAdmin, Mockito.times(1)).enableTable(physical);
+  }
+
+  private static void invokeReenableOrphanedDisabledHBaseTable(ConnectionQueryServicesImpl cqs,
+    byte[] schemaBytes, byte[] tableBytes, boolean isNamespaceMapped, PTableType tableType)
+    throws Exception {
+    Method m =
+      ConnectionQueryServicesImpl.class.getDeclaredMethod("reenableOrphanedDisabledHBaseTable",
+        byte[].class, byte[].class, boolean.class, PTableType.class);
+    m.setAccessible(true);
+    try {
+      m.invoke(cqs, schemaBytes, tableBytes, isNamespaceMapped, tableType);
+    } catch (InvocationTargetException e) {
+      if (e.getCause() instanceof Exception) {
+        throw (Exception) e.getCause();
+      }
+      throw e;
+    }
+  }
+
   /**
    * When a connection is closed concurrently with query compilation (e.g. connection pool teardown
    * or cluster failover), the metadata cache is nulled out. getMetaDataCache() must surface the