Fix to avoid deleting non ScalarDB tables when dropping namespaces

core/src/integration-test/java/com/scalar/db/storage/cassandra/CassandraAdminTestUtils.java

@@ -57,6 +57,11 @@ public void corruptMetadata(String namespace, String table) {
     // Do nothing
   }
 
+  @Override
+  public void deleteMetadata(String namespace, String table) throws Exception {
+    // Do nothing
+  }
+
   @Override
   public void dropNamespace(String namespace) {
     String dropKeyspaceQuery =

core/src/integration-test/java/com/scalar/db/storage/cosmos/CosmosAdminTestUtils.java

@@ -86,6 +86,15 @@ public void corruptMetadata(String namespace, String table) {
     container.upsertItem(corruptedMetadata);
   }
 
+  @Override
+  public void deleteMetadata(String namespace, String table) {
+    String fullTableName = getFullTableName(namespace, table);
+    CosmosContainer container =
+        client.getDatabase(metadataDatabase).getContainer(CosmosAdmin.TABLE_METADATA_CONTAINER);
+    container.deleteItem(
+        fullTableName, new PartitionKey(fullTableName), new CosmosItemRequestOptions());
+  }
+
   /**
    * Retrieve the stored procedure for the given table
    *

core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminCaseSensitivityIntegrationTest.java

@@ -29,6 +29,11 @@ protected AdminTestUtils getAdminTestUtils(String testName) {
     return new DynamoAdminTestUtils(getProperties(testName));
   }
 
+  @Override
+  @Disabled("DynamoDB does not have a concept of namespaces")
+  public void
+      dropNamespace_ForNamespaceWithNonScalarDBManagedTables_ShouldThrowIllegalArgumentException() {}
+
   @Override
   @Disabled("DynamoDB does not support dropping columns")
   public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly() {}

core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminIntegrationTest.java

@@ -28,6 +28,11 @@ protected AdminTestUtils getAdminTestUtils(String testName) {
     return new DynamoAdminTestUtils(getProperties(testName));
   }
 
+  @Override
+  @Disabled("DynamoDB does not have a concept of namespaces")
+  public void
+      dropNamespace_ForNamespaceWithNonScalarDBManagedTables_ShouldThrowIllegalArgumentException() {}
+
   @Override
   @Disabled("DynamoDB does not support dropping columns")
   public void dropColumnFromTable_DropColumnForEachExistingDataType_ShouldDropColumnsCorrectly() {}

core/src/integration-test/java/com/scalar/db/storage/dynamo/DynamoAdminTestUtils.java

@@ -171,6 +171,20 @@ public void corruptMetadata(String namespace, String table) {
             .build());
   }
 
+  @Override
+  public void deleteMetadata(String namespace, String table) {
+    String fullTableName =
+        getFullTableName(Namespace.of(namespacePrefix, namespace).prefixed(), table);
+    Map<String, AttributeValue> keyToDelete = new HashMap<>();
+    keyToDelete.put("table", AttributeValue.builder().s(fullTableName).build());
+
+    client.deleteItem(
+        DeleteItemRequest.builder()
+            .tableName(getFullTableName(metadataNamespace, DynamoAdmin.METADATA_TABLE))
+            .key(keyToDelete)
+            .build());
+  }
+
   @Override
   public void dropTable(String nonPrefixedNamespace, String table) {
     String namespace = Namespace.of(namespacePrefix, nonPrefixedNamespace).prefixed();

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

@@ -89,6 +89,12 @@ private boolean isWideningColumnTypeConversionNotFullySupported() {
     return JdbcTestUtils.isOracle(rdbEngine) || JdbcTestUtils.isSqlite(rdbEngine);
   }
 
+  @Test
+  @Override
+  @DisabledIf("isSqlite")
+  public void
+      dropNamespace_ForNamespaceWithNonScalarDBManagedTables_ShouldThrowIllegalArgumentException() {}
+
   @Test
   @Override
   @DisabledIf("isDb2")

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

@@ -90,6 +90,12 @@ private boolean isWideningColumnTypeConversionNotFullySupported() {
     return JdbcTestUtils.isOracle(rdbEngine) || JdbcTestUtils.isSqlite(rdbEngine);
   }
 
+  @Test
+  @Override
+  @DisabledIf("isSqlite")
+  public void
+      dropNamespace_ForNamespaceWithNonScalarDBManagedTables_ShouldThrowIllegalArgumentException() {}
+
   @Test
   @Override
   @DisabledIf("isDb2")

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

@@ -64,6 +64,22 @@ public void corruptMetadata(String namespace, String table) throws Exception {
     execute(insertCorruptedMetadataStatement);
   }
 
+  @Override
+  public void deleteMetadata(String namespace, String table) throws Exception {
+    String deleteMetadataStatement =
+        "DELETE FROM "
+            + rdbEngine.encloseFullTableName(metadataSchema, JdbcAdmin.METADATA_TABLE)
+            + " WHERE "
+            + rdbEngine.enclose(JdbcAdmin.METADATA_COL_FULL_TABLE_NAME)
+            + " = ?";
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement preparedStatement =
+            connection.prepareStatement(deleteMetadataStatement)) {
+      preparedStatement.setString(1, getFullTableName(namespace, table));
+      preparedStatement.executeUpdate();
+    }
+  }
+
   private void execute(String sql) throws SQLException {
     try (Connection connection = dataSource.getConnection()) {
       JdbcAdmin.execute(connection, sql);

core/src/integration-test/java/com/scalar/db/storage/multistorage/MultiStorageAdminTestUtils.java

@@ -116,6 +116,25 @@ public void corruptMetadata(String namespace, String table) throws Exception {
     execute(insertCorruptedMetadataStatement);
   }
 
+  @Override
+  public void deleteMetadata(String namespace, String table) throws Exception {
+    // Do nothing for Cassandra
+
+    // for JDBC
+    String deleteMetadataStatement =
+        "DELETE FROM "
+            + rdbEngine.encloseFullTableName(jdbcMetadataSchema, JdbcAdmin.METADATA_TABLE)
+            + " WHERE "
+            + rdbEngine.enclose(JdbcAdmin.METADATA_COL_FULL_TABLE_NAME)
+            + " = ?";
+    try (Connection connection = dataSource.getConnection();
+        PreparedStatement preparedStatement =
+            connection.prepareStatement(deleteMetadataStatement)) {
+      preparedStatement.setString(1, getFullTableName(namespace, table));
+      preparedStatement.executeUpdate();
+    }
+  }
+
   @Override
   public void dropNamespace(String namespace) throws SQLException {
     boolean existsOnCassandra = namespaceExistsOnCassandra(namespace);

core/src/main/java/com/scalar/db/common/CoreError.java

@@ -843,6 +843,12 @@ public enum CoreError implements ScalarDbError {
       "With Oracle, setting a condition on a BLOB column when using a selection operation is not supported. Condition: %s",
       "",
       ""),
+  NAMESPACE_WITH_NON_SCALARDB_TABLES_CANNOT_BE_DROPPED(
+      Category.USER_ERROR,
+      "0249",
+      "The namespace has non-ScalarDB tables and cannot be dropped. Namespace: %s; Tables in the namespace: %s",
+      "",
+      ""),
 
   //
   // Errors for the concurrency error category

core/src/main/java/com/scalar/db/storage/cosmos/CosmosAdmin.java

@@ -344,10 +344,18 @@ private void deleteTableMetadata(String namespace, String table) throws Executio
   @Override
   public void dropNamespace(String namespace) throws ExecutionException {
     try {
+      Set<String> remainingTables = getRawTableNames(namespace);
+      if (!remainingTables.isEmpty()) {
+        throw new IllegalArgumentException(
+            CoreError.NAMESPACE_WITH_NON_SCALARDB_TABLES_CANNOT_BE_DROPPED.buildMessage(
+                namespace, remainingTables));
+      }
       client.getDatabase(namespace).delete();
       getNamespacesContainer()
           .deleteItem(new CosmosNamespace(namespace), new CosmosItemRequestOptions());
       deleteMetadataDatabaseIfEmpty();
+    } catch (IllegalArgumentException e) {
+      throw e;
     } catch (RuntimeException e) {
       throw new ExecutionException(String.format("Deleting the %s database failed", namespace), e);
     }
@@ -785,4 +793,10 @@ private boolean containerExists(String databaseId, String containerId) throws Co
   public StorageInfo getStorageInfo(String namespace) {
     return STORAGE_INFO;
   }
+
+  private Set<String> getRawTableNames(String namespace) {
+    return client.getDatabase(namespace).readAllContainers().stream()
+        .map(CosmosContainerProperties::getId)
+        .collect(Collectors.toSet());
+  }
 }

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

@@ -45,7 +45,7 @@
 public class JdbcAdmin implements DistributedStorageAdmin {
   public static final String METADATA_TABLE = "metadata";
   public static final String NAMESPACES_TABLE = "namespaces";
-  @VisibleForTesting static final String METADATA_COL_FULL_TABLE_NAME = "full_table_name";
+  @VisibleForTesting public static final String METADATA_COL_FULL_TABLE_NAME = "full_table_name";
   @VisibleForTesting static final String METADATA_COL_COLUMN_NAME = "column_name";
   @VisibleForTesting static final String METADATA_COL_DATA_TYPE = "data_type";
   @VisibleForTesting static final String METADATA_COL_KEY_TYPE = "key_type";
@@ -481,6 +481,12 @@ private void deleteMetadataSchema(Connection connection) throws SQLException {
   @Override
   public void dropNamespace(String namespace) throws ExecutionException {
     try (Connection connection = dataSource.getConnection()) {
+      Set<String> remainingTables = getInternalTableNames(connection, namespace);
+      if (!remainingTables.isEmpty()) {
+        throw new IllegalArgumentException(
+            CoreError.NAMESPACE_WITH_NON_SCALARDB_TABLES_CANNOT_BE_DROPPED.buildMessage(
+                namespace, remainingTables));
+      }
       execute(connection, rdbEngine.dropNamespaceSql(namespace));
       deleteFromNamespacesTable(connection, namespace);
       deleteNamespacesTableAndMetadataSchemaIfEmpty(connection);
@@ -703,6 +709,24 @@ public Set<String> getNamespaceTableNames(String namespace) throws ExecutionExce
     }
   }
 
+  private Set<String> getInternalTableNames(Connection connection, String namespace)
+      throws SQLException {
+    String sql = rdbEngine.getTableNamesInNamespaceSql();
+    if (Strings.isNullOrEmpty(sql)) {
+      return Collections.emptySet();
+    }
+    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

@@ -535,4 +535,9 @@ public void bindBlobColumnToPreparedStatement(
       preparedStatement.setBinaryStream(index, inputStream);
     }
   }
+
+  @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,10 @@ 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() {
+    // Do nothing. Namespace is just a table prefix in the SQLite implementation.
+    return null;
+  }
 }

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

@@ -325,4 +325,6 @@ default void throwIfCrossPartitionScanOrderingOnBlobColumnNotSupported(
    */
   default void throwIfConjunctionsColumnNotSupported(
       Set<Conjunction> conjunctions, TableMetadata metadata) {}
+
+  String getTableNamesInNamespaceSql();
 }