Fix to avoid deleting non ScalarDB tables when dropping namespaces

core/src/integration-test/java/com/scalar/db/storage/jdbc/ConsensusCommitAdminImportTableIntegrationTestWithJdbcDatabase.java

@@ -3,6 +3,7 @@
 import com.scalar.db.api.DistributedStorageAdminImportTableIntegrationTestBase.TestData;
 import com.scalar.db.exception.storage.ExecutionException;
 import com.scalar.db.transaction.consensuscommit.ConsensusCommitAdminImportTableIntegrationTestBase;
+import com.scalar.db.util.AdminTestUtils;
 import java.sql.SQLException;
 import java.util.List;
 import java.util.Properties;
@@ -53,6 +54,11 @@ protected void dropNonImportableTable(String table) throws SQLException {
     testUtils.dropTable(getNamespace(), table);
   }
 
+  @Override
+  protected AdminTestUtils getAdminTestUtils(String testName) {
+    return new JdbcAdminTestUtils(getProperties(testName));
+  }
+
   @SuppressWarnings("unused")
   private boolean isSqlite() {
     return JdbcEnv.isSqlite();
@@ -72,4 +78,9 @@ public void importTable_ForUnsupportedDatabase_ShouldThrowUnsupportedOperationEx
       throws ExecutionException {
     super.importTable_ForUnsupportedDatabase_ShouldThrowUnsupportedOperationException();
   }
+
+  @Test
+  @Override
+  @DisabledIf("isSqlite")
+  public void dropNamespace_ShouldNotDropNonScalarDBTables() {}
 }

core/src/integration-test/java/com/scalar/db/storage/jdbc/JdbcAdminImportTableIntegrationTest.java

@@ -6,6 +6,7 @@
 import com.scalar.db.api.TableMetadata;
 import com.scalar.db.exception.storage.ExecutionException;
 import com.scalar.db.io.DataType;
+import com.scalar.db.util.AdminTestUtils;
 import java.sql.SQLException;
 import java.util.Collections;
 import java.util.List;
@@ -68,6 +69,11 @@ protected void dropNonImportableTable(String table) throws SQLException {
     testUtils.dropTable(getNamespace(), table);
   }
 
+  @Override
+  protected AdminTestUtils getAdminTestUtils(String testName) {
+    return new JdbcAdminTestUtils(getProperties(testName));
+  }
+
   @SuppressWarnings("unused")
   private boolean isOracle() {
     return JdbcEnv.isOracle();
@@ -263,4 +269,9 @@ public void alterColumnType_Oracle_WideningConversion_ForImportedTable_ShouldAlt
       }
     }
   }
+
+  @Test
+  @Override
+  @DisabledIf("isSqlite")
+  public void dropNamespace_ShouldNotDropNonScalarDBTables() {}
 }

core/src/main/java/com/scalar/db/storage/jdbc/JdbcAdmin.java

@@ -481,6 +481,11 @@ private void deleteMetadataSchema(Connection connection) throws SQLException {
   @Override
   public void dropNamespace(String namespace) throws ExecutionException {
     try (Connection connection = dataSource.getConnection()) {
+      Set<String> remainingTables = getNamespaceTableNamesInternal(connection, namespace);
+      if (!remainingTables.isEmpty()) {
+        throw new IllegalStateException(
+            CoreError.NAMESPACE_NOT_EMPTY.buildMessage(namespace, remainingTables));
+      }
       execute(connection, rdbEngine.dropNamespaceSql(namespace));
       deleteFromNamespacesTable(connection, namespace);
       deleteNamespacesTableAndMetadataSchemaIfEmpty(connection);
@@ -703,6 +708,21 @@ public Set<String> getNamespaceTableNames(String namespace) throws ExecutionExce
     }
   }
 
+  private Set<String> getNamespaceTableNamesInternal(Connection connection, String namespace)
+      throws SQLException {
+    String sql = rdbEngine.getTableNamesInNamespaceSql();
+    Set<String> tableNames = new HashSet<>();
+    try (PreparedStatement statement = connection.prepareStatement(sql)) {
+      statement.setString(1, namespace);
+      try (ResultSet resultSet = statement.executeQuery()) {
+        while (resultSet.next()) {
+          tableNames.add(resultSet.getString(1));
+        }
+      }
+    }
+    return tableNames;
+  }
+
   @Override
   public boolean namespaceExists(String namespace) throws ExecutionException {
     String selectQuery =

core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineDb2.java

@@ -583,4 +583,9 @@ public void throwIfCrossPartitionScanOrderingOnBlobColumnNotSupported(
               .buildMessage(orderingOnBlobColumn.get()));
     }
   }
+
+  @Override
+  public String getTableNamesInNamespaceSql() {
+    return "SELECT TABNAME FROM SYSCAT.TABLES WHERE TABSCHEMA = ? AND TYPE = 'T'";
+  }
 }

core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineMysql.java

@@ -504,4 +504,9 @@ public void setConnectionToReadOnly(Connection connection, boolean readOnly) thr
     // Observed performance degradation when using read-only connections in MySQL. So we do not
     // set the read-only mode for MySQL connections.
   }
+
+  @Override
+  public String getTableNamesInNamespaceSql() {
+    return "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = ?";
+  }
 }

core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineOracle.java

@@ -451,4 +451,9 @@ public void throwIfAlterColumnTypeNotSupported(DataType from, DataType to) {
               from.toString(), to.toString()));
     }
   }
+
+  @Override
+  public String getTableNamesInNamespaceSql() {
+    return "SELECT TABLE_NAME FROM ALL_TABLES WHERE OWNER = ?";
+  }
 }

core/src/main/java/com/scalar/db/storage/jdbc/RdbEnginePostgresql.java

@@ -395,4 +395,9 @@ public String tryAddIfNotExistsToCreateIndexSql(String createIndexSql) {
       getTimeTypeStrategy() {
     return timeTypeEngine;
   }
+
+  @Override
+  public String getTableNamesInNamespaceSql() {
+    return "SELECT table_name FROM information_schema.tables WHERE table_schema = ?";
+  }
 }

core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineSqlServer.java

@@ -431,4 +431,9 @@ public Map<String, String> getConnectionProperties(JdbcConfig config) {
       getTimeTypeStrategy() {
     return timeTypeEngine;
   }
+
+  @Override
+  public String getTableNamesInNamespaceSql() {
+    return "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = ?";
+  }
 }

core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineSqlite.java

@@ -367,4 +367,9 @@ public void throwIfAlterColumnTypeNotSupported(DataType from, DataType to) {
   public void setConnectionToReadOnly(Connection connection, boolean readOnly) {
     // Do nothing. SQLite does not support read-only mode.
   }
+
+  @Override
+  public String getTableNamesInNamespaceSql() {
+    return "SELECT name FROM sqlite_master WHERE type='table' AND name LIKE ?";
+  }
 }

core/src/main/java/com/scalar/db/storage/jdbc/RdbEngineStrategy.java

@@ -305,4 +305,6 @@ default void setConnectionToReadOnly(Connection connection, boolean readOnly)
    */
   default void throwIfCrossPartitionScanOrderingOnBlobColumnNotSupported(
       ScanAll scanAll, TableMetadata metadata) {}
+
+  String getTableNamesInNamespaceSql();
 }

core/src/test/java/com/scalar/db/storage/jdbc/JdbcAdminTest.java

@@ -1957,6 +1957,7 @@ public void dropNamespace_forSqlite_shouldDropNamespace() throws Exception {
     JdbcAdmin admin = createJdbcAdminFor(RdbEngine.SQLITE);
 
     Connection connection = mock(Connection.class);
+    PreparedStatement getTableNamesPrepStmt = mock(PreparedStatement.class);
     PreparedStatement deleteFromNamespaceTablePrepStmt = mock(PreparedStatement.class);
     Statement selectAllFromNamespaceTablePrepStmt = mock(Statement.class);
     Statement selectAllFromMetadataTablePrepStmt = mock(Statement.class);
@@ -1967,7 +1968,13 @@ public void dropNamespace_forSqlite_shouldDropNamespace() throws Exception {
             selectAllFromNamespaceTablePrepStmt,
             selectAllFromMetadataTablePrepStmt,
             dropNamespaceTableStmt);
-    when(connection.prepareStatement(anyString())).thenReturn(deleteFromNamespaceTablePrepStmt);
+    // Mock for getNamespaceTableNamesInternal() check - returns empty ResultSet (namespace is
+    // empty)
+    ResultSet emptyResultSet = mock(ResultSet.class);
+    when(emptyResultSet.next()).thenReturn(false);
+    when(getTableNamesPrepStmt.executeQuery()).thenReturn(emptyResultSet);
+    when(connection.prepareStatement(anyString()))
+        .thenReturn(getTableNamesPrepStmt, deleteFromNamespaceTablePrepStmt);
     when(dataSource.getConnection()).thenReturn(connection);
     // Only the metadata schema is left
     ResultSet resultSet1 =
@@ -2005,6 +2012,7 @@ private void dropNamespace_WithOnlyNamespaceSchemaLeftForX_shouldDropSchemaAndNa
 
     Connection connection = mock(Connection.class);
     Statement dropNamespaceStmt = mock(Statement.class);
+    PreparedStatement isNamespaceEmptyStatementMock = mock(PreparedStatement.class);
     PreparedStatement deleteFromNamespaceTablePrepStmt = mock(PreparedStatement.class);
     Statement selectAllFromNamespaceTablePrepStmt = mock(Statement.class);
     Statement selectAllFromMetadataTablePrepStmt = mock(Statement.class);
@@ -2018,7 +2026,12 @@ private void dropNamespace_WithOnlyNamespaceSchemaLeftForX_shouldDropSchemaAndNa
             selectAllFromMetadataTablePrepStmt,
             dropNamespaceTableStmt,
             dropMetadataSchemaStmt);
-    when(connection.prepareStatement(anyString())).thenReturn(deleteFromNamespaceTablePrepStmt);
+    // Mock for isNamespaceEmpty() check - returns empty ResultSet (namespace is empty)
+    ResultSet emptyResultSet = mock(ResultSet.class);
+    when(emptyResultSet.next()).thenReturn(false);
+    when(isNamespaceEmptyStatementMock.executeQuery()).thenReturn(emptyResultSet);
+    when(connection.prepareStatement(anyString()))
+        .thenReturn(isNamespaceEmptyStatementMock, deleteFromNamespaceTablePrepStmt);
     when(dataSource.getConnection()).thenReturn(connection);
     // Only the metadata schema is left
     ResultSet resultSet =
@@ -2116,6 +2129,7 @@ private void dropNamespace_WithOtherNamespaceLeftForX_shouldOnlyDropNamespace(
 
     Connection connection = mock(Connection.class);
     Statement dropNamespaceStatementMock = mock(Statement.class);
+    PreparedStatement isNamespaceEmptyStatementMock = mock(PreparedStatement.class);
     PreparedStatement deleteFromNamespaceTableMock = mock(PreparedStatement.class);
     Statement selectNamespaceStatementMock = mock(Statement.class);
     if (rdbEngine != RdbEngine.SQLITE) {
@@ -2124,7 +2138,12 @@ private void dropNamespace_WithOtherNamespaceLeftForX_shouldOnlyDropNamespace(
     } else {
       when(connection.createStatement()).thenReturn(selectNamespaceStatementMock);
     }
-    when(connection.prepareStatement(anyString())).thenReturn(deleteFromNamespaceTableMock);
+    // Mock for isNamespaceEmpty() check - returns empty ResultSet (namespace is empty)
+    ResultSet emptyResultSet = mock(ResultSet.class);
+    when(emptyResultSet.next()).thenReturn(false);
+    when(isNamespaceEmptyStatementMock.executeQuery()).thenReturn(emptyResultSet);
+    when(connection.prepareStatement(anyString()))
+        .thenReturn(isNamespaceEmptyStatementMock, deleteFromNamespaceTableMock);
     when(dataSource.getConnection()).thenReturn(connection);
     // Namespaces table contains other namespaces
     ResultSet resultSet =

integration-test/src/main/java/com/scalar/db/api/DistributedStorageAdminImportTableIntegrationTestBase.java

@@ -1,12 +1,14 @@
 package com.scalar.db.api;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.scalar.db.exception.storage.ExecutionException;
 import com.scalar.db.io.Column;
 import com.scalar.db.io.DataType;
 import com.scalar.db.service.StorageFactory;
+import com.scalar.db.util.AdminTestUtils;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -59,7 +61,7 @@ private void dropTable() throws Exception {
       if (!testData.isImportableTable()) {
         dropNonImportableTable(testData.getTableName());
       } else {
-        admin.dropTable(getNamespace(), testData.getTableName());
+        admin.dropTable(getNamespace(), testData.getTableName(), true);
       }
     }
     if (!admin.namespaceExists(getNamespace())) {
@@ -108,6 +110,8 @@ protected abstract List<String> getFloatCompatibleColumnNamesOnExistingDatabase(
 
   protected abstract void dropNonImportableTable(String table) throws Exception;
 
+  protected abstract AdminTestUtils getAdminTestUtils(String testName);
+
   @Test
   public void importTable_ShouldWorkProperly() throws Exception {
     // Arrange
@@ -222,6 +226,43 @@ public void alterColumnType_WideningConversion_ForImportedTable_ShouldAlterPrope
     }
   }
 
+  @Test
+  public void dropNamespace_ShouldNotDropNonScalarDBTables() throws Exception {
+    AdminTestUtils adminTestUtils = getAdminTestUtils(TEST_NAME);
+    try {
+      // Arrange
+      testDataList.addAll(createExistingDatabaseWithAllDataTypes());
+      for (TestData testData : testDataList) {
+        if (testData.isImportableTable()) {
+          admin.importTable(
+              getNamespace(),
+              testData.getTableName(),
+              Collections.emptyMap(),
+              testData.getOverrideColumnsType());
+        }
+      }
+      for (TestData testData : testDataList) {
+        if (testData.isImportableTable()) {
+          admin.dropTable(getNamespace(), testData.getTableName());
+        }
+      }
+
+      // Act
+      assertThatCode(() -> admin.dropNamespace(getNamespace()))
+          .isInstanceOf(IllegalStateException.class);
+
+      // Assert
+      assertThat(admin.namespaceExists(getNamespace())).isTrue();
+      for (TestData testData : testDataList) {
+        if (!testData.isImportableTable()) {
+          assertThat(adminTestUtils.tableExists(getNamespace(), testData.getTableName())).isTrue();
+        }
+      }
+    } finally {
+      adminTestUtils.close();
+    }
+  }
+
   private void importTable_ForImportableTable_ShouldImportProperly(
       String table, Map<String, DataType> overrideColumnsType, TableMetadata metadata)
       throws ExecutionException {

integration-test/src/main/java/com/scalar/db/api/DistributedTransactionAdminImportTableIntegrationTestBase.java

@@ -1,6 +1,7 @@
 package com.scalar.db.api;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import com.scalar.db.api.DistributedStorageAdminImportTableIntegrationTestBase.TestData;
@@ -9,6 +10,7 @@
 import com.scalar.db.io.Column;
 import com.scalar.db.io.DataType;
 import com.scalar.db.service.TransactionFactory;
+import com.scalar.db.util.AdminTestUtils;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
@@ -104,6 +106,8 @@ protected void afterAll() throws Exception {}
 
   protected abstract void dropNonImportableTable(String table) throws Exception;
 
+  protected abstract AdminTestUtils getAdminTestUtils(String testName);
+
   @Test
   public void importTable_ShouldWorkProperly() throws Exception {
     // Arrange
@@ -134,6 +138,43 @@ public void importTable_ForUnsupportedDatabase_ShouldThrowUnsupportedOperationEx
         .isInstanceOf(UnsupportedOperationException.class);
   }
 
+  @Test
+  public void dropNamespace_ShouldNotDropNonScalarDBTables() throws Exception {
+    AdminTestUtils adminTestUtils = getAdminTestUtils(TEST_NAME);
+    try {
+      // Arrange
+      testDataList.addAll(createExistingDatabaseWithAllDataTypes());
+      for (TestData testData : testDataList) {
+        if (testData.isImportableTable()) {
+          admin.importTable(
+              getNamespace(),
+              testData.getTableName(),
+              Collections.emptyMap(),
+              testData.getOverrideColumnsType());
+        }
+      }
+      for (TestData testData : testDataList) {
+        if (testData.isImportableTable()) {
+          admin.dropTable(getNamespace(), testData.getTableName());
+        }
+      }
+
+      // Act
+      assertThatCode(() -> admin.dropNamespace(getNamespace()))
+          .isInstanceOf(IllegalStateException.class);
+
+      // Assert
+      assertThat(admin.namespaceExists(getNamespace())).isTrue();
+      for (TestData testData : testDataList) {
+        if (!testData.isImportableTable()) {
+          assertThat(adminTestUtils.tableExists(getNamespace(), testData.getTableName())).isTrue();
+        }
+      }
+    } finally {
+      adminTestUtils.close();
+    }
+  }
+
   private void importTable_ForImportableTable_ShouldImportProperly(
       String table, Map<String, DataType> overrideColumnsType, TableMetadata metadata)
       throws ExecutionException {