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;
@@ -1922,6 +1923,21 @@ private TableDescriptor ensureTableCreated(byte[] physicalTableName,
         }
       }
 
+      // PHOENIX-7788: recover an orphaned disabled physical table before modifyTable runs on it.
+      if (
+        tableExist && tableType == PTableType.TABLE
+          && admin.isTableDisabled(TableName.valueOf(physicalTableName))
+      ) {
+        byte[] viewIndexMarker = existingDesc == null ? null
+          : existingDesc.getValue(MetaDataUtil.IS_VIEW_INDEX_TABLE_PROP_BYTES);
+        // Classify by the descriptor marker, falling back to the _IDX_ name prefix for older
+        // view-index tables that predate the marker.
+        boolean isViewIndexTable =
+          (viewIndexMarker != null && Boolean.TRUE.equals(PBoolean.INSTANCE.toObject(viewIndexMarker)))
+            || MetaDataUtil.isViewIndex(Bytes.toString(physicalTableName));
+        reenableOrphanedDisabledHBaseTable(physicalTableName, isViewIndexTable, admin);
+      }
+
       TableDescriptorBuilder newDesc =
         generateTableDescriptor(physicalTableName, parentPhysicalTableName, existingDesc, tableType,
           props, families, splits, isNamespaceMapped);
@@ -2459,6 +2475,50 @@ 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. Caller
+   * must have already confirmed the physical table exists and is disabled.
+   */
+  private void reenableOrphanedDisabledHBaseTable(byte[] physicalTableNameBytes,
+    boolean isViewIndexTable, Admin admin) throws SQLException {
+    TableName physicalTableName = TableName.valueOf(physicalTableNameBytes);
+    // For a view-index physical table, resolve to the base table name; otherwise the physical
+    // name is itself the metadata key.
+    String physicalName = Bytes.toString(physicalTableNameBytes);
+    String metadataName = isViewIndexTable
+      ? MetaDataUtil.getViewIndexUserTableName(physicalName)
+      : physicalName;
+    byte[] schemaBytes = Bytes.toBytes(SchemaUtil.getSchemaNameFromFullName(metadataName));
+    byte[] tableBytes = Bytes.toBytes(SchemaUtil.getTableNameFromFullName(metadataName));
+    // No-cache read straight from the server; a stale/absent client-cache entry must not drive
+    // the orphan decision.
+    PTable existingTable = getTable(null, schemaBytes, tableBytes, HConstants.LATEST_TIMESTAMP,
+      HConstants.LATEST_TIMESTAMP).getTable();
+    if (existingTable != null) {
+      LOGGER.info(
+        "Physical HBase table {} is disabled but {} has metadata for it; "
+          + "leaving it disabled to preserve any intentional admin action.",
+        physicalTableName, PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME);
+      return;
+    }
+    LOGGER.info("Re-enabling orphaned disabled HBase table {} during CREATE TABLE",
+      physicalTableName);
+    try {
+      enableTable(admin, physicalTableName);
+    } catch (IOException e) {
+      throw ClientUtil.parseServerException(e);
+    }
+  }
+
   private boolean ensureViewIndexTableDropped(byte[] physicalTableName, long timestamp)
     throws SQLException {
     byte[] physicalIndexName = MetaDataUtil.getViewIndexPhysicalName(physicalTableName);

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

@@ -82,6 +82,7 @@
 import org.apache.phoenix.util.ByteUtil;
 import org.apache.phoenix.util.EnvironmentEdgeManager;
 import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.MetaDataUtil;
 import org.apache.phoenix.util.PropertiesUtil;
 import org.apache.phoenix.util.QueryUtil;
 import org.apache.phoenix.util.ReadOnlyProps;
@@ -1739,6 +1740,129 @@ 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));
+    }
+  }
+
+  // PHOENIX-7788: when the base table still has metadata, a disabled shared view-index physical
+  // table must be left disabled -- an admin may have disabled it intentionally, and re-creating
+  // the base table must not silently undo that (same conservative rule as plain base tables).
+  @Test
+  public void testCreateTableDoesNotReenableDisabledViewIndexTableWhenBaseTableExists()
+    throws Exception {
+    String baseTable = generateUniqueName();
+    String ddl = "CREATE TABLE IF NOT EXISTS " + baseTable + " (T_ID VARCHAR NOT NULL, "
+      + "K VARCHAR NOT NULL, V VARCHAR CONSTRAINT PK PRIMARY KEY (T_ID, K)) "
+      + "MULTI_TENANT=true, 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 physicalIndexTable =
+      TableName.valueOf(MetaDataUtil.getViewIndexPhysicalName(baseTable));
+
+    try (Admin admin = services.getAdmin()) {
+      admin.disableTable(physicalIndexTable);
+      assertTrue(admin.isTableDisabled(physicalIndexTable));
+    }
+
+    try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+      conn.unwrap(PhoenixConnection.class).getQueryServices().clearCache();
+      conn.createStatement().execute(ddl);
+    }
+
+    try (Admin admin = services.getAdmin()) {
+      assertTrue("Shared view-index table with a live base table must stay disabled",
+        admin.isTableDisabled(physicalIndexTable));
+    }
+  }
+
   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,15 @@
 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.PTable;
 import org.apache.phoenix.util.ReadOnlyProps;
 import org.junit.Before;
 import org.junit.ClassRule;
@@ -410,6 +420,111 @@ 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 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);
+    MetaDataMutationResult existing =
+      new MetaDataMutationResult(MutationCode.TABLE_ALREADY_EXISTS, 0L, Mockito.mock(PTable.class));
+    doReturn(existing).when(mockCqs).getTable(Mockito.<PName> any(), any(byte[].class),
+      any(byte[].class), anyLong(), anyLong());
+    invokeReenableOrphanedDisabledHBaseTable(mockCqs, name, mockAdmin);
+    verify(mockAdmin, never()).enableTable(physical);
+  }
+
+  @Test
+  public void testReenableOrphanedDisabledHBaseTableReenablesOrphanedTable() throws Exception {
+    byte[] name = "TEST_TABLE".getBytes(StandardCharsets.UTF_8);
+    TableName physical = TableName.valueOf(name);
+    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());
+    invokeReenableOrphanedDisabledHBaseTable(mockCqs, name, mockAdmin);
+    verify(mockAdmin, Mockito.times(1)).enableTable(physical);
+  }
+
+  @Test
+  public void testReenableOrphanedDisabledHBaseTableDerivesLogicalNameForNamespaceMapped()
+    throws Exception {
+    // PHOENIX-7788: for a namespace-mapped physical name "MYSCHEMA:MYTABLE" the metadata
+    // lookup must be against logical schema="MYSCHEMA", table="MYTABLE".
+    byte[] name = "MYSCHEMA:MYTABLE".getBytes(StandardCharsets.UTF_8);
+    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());
+    invokeReenableOrphanedDisabledHBaseTable(mockCqs, name, mockAdmin);
+    verify(mockCqs).getTable(Mockito.<PName> any(), eq("MYSCHEMA".getBytes(StandardCharsets.UTF_8)),
+      eq("MYTABLE".getBytes(StandardCharsets.UTF_8)), anyLong(), anyLong());
+  }
+
+  @Test
+  public void testReenableOrphanedDisabledHBaseTableDerivesLogicalNameForNonNamespaceMapped()
+    throws Exception {
+    // PHOENIX-7788: for a non-namespace-mapped physical name "MYSCHEMA.MYTABLE" the metadata
+    // lookup must be against logical schema="MYSCHEMA", table="MYTABLE".
+    byte[] name = "MYSCHEMA.MYTABLE".getBytes(StandardCharsets.UTF_8);
+    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());
+    invokeReenableOrphanedDisabledHBaseTable(mockCqs, name, mockAdmin);
+    verify(mockCqs).getTable(Mockito.<PName> any(), eq("MYSCHEMA".getBytes(StandardCharsets.UTF_8)),
+      eq("MYTABLE".getBytes(StandardCharsets.UTF_8)), anyLong(), anyLong());
+  }
+
+  private static void invokeReenableOrphanedDisabledHBaseTable(ConnectionQueryServicesImpl cqs,
+    byte[] physicalTableNameBytes, Admin admin) throws Exception {
+    Method m = ConnectionQueryServicesImpl.class.getDeclaredMethod(
+      "reenableOrphanedDisabledHBaseTable", byte[].class, boolean.class, Admin.class);
+    m.setAccessible(true);
+    try {
+      // These cases exercise base (non-view-index) physical tables, so isViewIndexTable is false.
+      m.invoke(cqs, physicalTableNameBytes, false, admin);
+    } 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