This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 6c35c8ed35 [#13860] improvement(core): Chunk the versioned cascade
soft delete of schemas and catalogs (#13861)
6c35c8ed35 is described below
commit 6c35c8ed35ab19a8d691dfdd2089f8c042619222
Author: Qi Yu <[email protected]>
AuthorDate: Mon Oct 26 19:29:47 2026 +0800
[#13860] improvement(core): Chunk the versioned cascade soft delete of
schemas and catalogs (#13861)
### What changes were proposed in this pull request?
- `OccWriteSupport#deleteChildrenWithVersions` runs the versioned batch
soft delete in chunks of 500 children. All chunks stay in the caller's
transaction, so a version miss in any chunk still rolls back the whole
cascade.
- `softDeleteSchemaMetasWithVersion` and
`softDeleteCatalogMetasWithVersion` (base and PostgreSQL) add `schema_id
IN (...)` / `catalog_id IN (...)` before the version OR chain, so each
chunk finds its rows by primary key.
### Why are the changes needed?
A cascading catalog or metalake drop deleted all schemas with one OR
chain. MySQL planned it superlinearly, H2 scanned the whole table for
it, and PostgreSQL rejected it above 32,767 schemas because of its
65,535 bind parameter limit.
Cascade drop of one catalog, time in ms:
| Schemas | H2 before | H2 after | MySQL before | MySQL after | PG
before | PG after | PG after, with `ANALYZE` |
|--------:|----------:|---------:|-------------:|------------:|----------:|---------:|-------------------------:|
| 2,000 | 636 | 389 | 246 | 162 | 77 | 73 | 72 |
| 17,000 | 4,475 | 956 | 22,649 | 766 | 1,117 | 1,239 | 399 |
| 40,000 | 20,909 | 2,371 | 117,160 | 1,806 | failed | 8,648 | 971 |
The schemas are bulk-inserted right before the drop. Until PostgreSQL
analyzes the table, it estimates the table as nearly empty and scans a
secondary index instead of the primary key. That accounts for the "PG
after" column; with fresh statistics it uses the primary key.
Fix: #13860
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- `TestOccWriteSupport`: the children are split into chunks, and a short
chunk stops the delete with `OptimisticLockException`.
-
`TestCatalogMetaService#testCascadeDeleteCatalogWithMoreSchemasThanOneChunk`:
a stale version in the second chunk fails the cascade and rolls back the
first chunk; a retry then drops everything. Runs on H2, MySQL and
PostgreSQL.
- `TestCatalogMetaService`, `TestMetalakeMetaService`,
`TestSchemaMetaService` and `TestOccWriteSupport` pass on H2, MySQL and
PostgreSQL (`-PskipDockerTests=false`).
- The timings above come from a local stress harness that is not part of
this PR.
---
.../provider/base/CatalogMetaBaseSQLProvider.java | 11 ++++-
.../provider/base/SchemaMetaBaseSQLProvider.java | 11 ++++-
.../postgresql/CatalogMetaPostgreSQLProvider.java | 6 ++-
.../postgresql/SchemaMetaPostgreSQLProvider.java | 6 ++-
.../relational/service/OccWriteSupport.java | 23 +++++++--
.../relational/service/TestCatalogMetaService.java | 57 ++++++++++++++++++++++
.../relational/service/TestOccWriteSupport.java | 51 +++++++++++++++++++
7 files changed, 154 insertions(+), 11 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java
index 41a20f22ae..3cbcc707f0 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/CatalogMetaBaseSQLProvider.java
@@ -248,7 +248,10 @@ public class CatalogMetaBaseSQLProvider {
+ " AND current_version = #{currentVersion} AND deleted_at = 0";
}
- /** Returns SQL that soft-deletes catalogs using identifier-and-version
pairs. */
+ /**
+ * Returns SQL that soft-deletes catalogs using identifier-and-version
pairs. The leading IN list
+ * lets the database find the rows by primary key instead of scanning the
table for the OR chain.
+ */
public String softDeleteCatalogMetasWithVersion(
@Param("catalogMetas") List<CatalogPO> catalogPOs) {
return "<script>"
@@ -256,7 +259,11 @@ public class CatalogMetaBaseSQLProvider {
+ TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.MYSQL
- + " WHERE deleted_at = 0 AND "
+ + " WHERE deleted_at = 0 AND catalog_id IN "
+ + "<foreach collection='catalogMetas' item='item' separator=','
open='(' close=')'>"
+ + "#{item.catalogId}"
+ + "</foreach>"
+ + " AND "
+ "<foreach collection='catalogMetas' item='item' separator=' OR '
open='(' close=')'>"
+ "(catalog_id = #{item.catalogId} AND current_version =
#{item.currentVersion})"
+ "</foreach>"
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java
index b7560254df..c6357291d3 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/SchemaMetaBaseSQLProvider.java
@@ -375,14 +375,21 @@ public class SchemaMetaBaseSQLProvider {
+ " AND current_version = #{currentVersion} AND deleted_at = 0";
}
- /** Returns SQL that soft-deletes schemas using identifier-and-version
pairs. */
+ /**
+ * Returns SQL that soft-deletes schemas using identifier-and-version pairs.
The leading IN list
+ * lets the database find the rows by primary key instead of scanning the
table for the OR chain.
+ */
public String softDeleteSchemaMetasWithVersion(@Param("schemaMetas")
List<SchemaPO> schemaPOs) {
return "<script>"
+ "UPDATE "
+ TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.MYSQL
- + " WHERE deleted_at = 0 AND "
+ + " WHERE deleted_at = 0 AND schema_id IN "
+ + "<foreach collection='schemaMetas' item='item' separator=','
open='(' close=')'>"
+ + "#{item.schemaId}"
+ + "</foreach>"
+ + " AND "
+ "<foreach collection='schemaMetas' item='item' separator=' OR '
open='(' close=')'>"
+ "(schema_id = #{item.schemaId} AND current_version =
#{item.currentVersion})"
+ "</foreach>"
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java
index 83997da0ae..cb4a2d8500 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/CatalogMetaPostgreSQLProvider.java
@@ -50,7 +50,11 @@ public class CatalogMetaPostgreSQLProvider extends
CatalogMetaBaseSQLProvider {
+ TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.POSTGRESQL
- + " WHERE deleted_at = 0 AND "
+ + " WHERE deleted_at = 0 AND catalog_id IN "
+ + "<foreach collection='catalogMetas' item='item' separator=','
open='(' close=')'>"
+ + "#{item.catalogId}"
+ + "</foreach>"
+ + " AND "
+ "<foreach collection='catalogMetas' item='item' separator=' OR '
open='(' close=')'>"
+ "(catalog_id = #{item.catalogId} AND current_version =
#{item.currentVersion})"
+ "</foreach>"
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java
index a7e52587fa..ecbb0e25b7 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/SchemaMetaPostgreSQLProvider.java
@@ -159,7 +159,11 @@ public class SchemaMetaPostgreSQLProvider extends
SchemaMetaBaseSQLProvider {
+ TABLE_NAME
+ " SET deleted_at = "
+ DatabaseTimeSQL.POSTGRESQL
- + " WHERE deleted_at = 0 AND "
+ + " WHERE deleted_at = 0 AND schema_id IN "
+ + "<foreach collection='schemaMetas' item='item' separator=','
open='(' close=')'>"
+ + "#{item.schemaId}"
+ + "</foreach>"
+ + " AND "
+ "<foreach collection='schemaMetas' item='item' separator=' OR '
open='(' close=')'>"
+ "(schema_id = #{item.schemaId} AND current_version =
#{item.currentVersion})"
+ "</foreach>"
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java
index c1cc7af577..f8fd83a4d9 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/OccWriteSupport.java
@@ -18,6 +18,8 @@
*/
package org.apache.gravitino.storage.relational.service;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
import java.util.List;
import java.util.Locale;
import java.util.function.Function;
@@ -38,6 +40,13 @@ import
org.apache.gravitino.storage.relational.utils.ExceptionUtils;
*/
public class OccWriteSupport {
+ /**
+ * The most children one versioned batch soft delete may guard. Every child
adds an {@code (id = ?
+ * AND current_version = ?)} term to the statement, and a long OR chain is
slow to plan on MySQL
+ * and exceeds the 65,535 bind parameter limit of PostgreSQL beyond 32,767
children.
+ */
+ @VisibleForTesting static final int CHILD_DELETE_BATCH_SIZE = 500;
+
private OccWriteSupport() {}
/**
@@ -161,8 +170,10 @@ public class OccWriteSupport {
/**
* Executes a batch soft delete of child entities guarded by their
individual versions.
*
- * <p>If the affected row count does not match the size of the child list,
throws a concurrent
- * child modification exception via {@link
ExceptionUtils#concurrentChildModification}.
+ * <p>The children are deleted in chunks of at most {@link
#CHILD_DELETE_BATCH_SIZE}. All chunks
+ * run in the caller's transaction, so a failure rolls back the chunks
already deleted. If the
+ * affected row count does not match the size of the child list, throws a
concurrent child
+ * modification exception via {@link
ExceptionUtils#concurrentChildModification}.
*
* @param <T> the persistent object (PO) type of the child entities
* @param parentIdentifier the name identifier of the parent entity
@@ -180,9 +191,11 @@ public class OccWriteSupport {
if (children == null || children.isEmpty()) {
return;
}
- int deleted = softDeleteOps.applyAsInt(children);
- if (deleted != children.size()) {
- throw ExceptionUtils.concurrentChildModification(childType, parentType,
parentIdentifier);
+ for (List<T> chunk : Lists.partition(children, CHILD_DELETE_BATCH_SIZE)) {
+ int deleted = softDeleteOps.applyAsInt(chunk);
+ if (deleted != chunk.size()) {
+ throw ExceptionUtils.concurrentChildModification(childType,
parentType, parentIdentifier);
+ }
}
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java
index 2a6bc7c441..890a1f12e8 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestCatalogMetaService.java
@@ -29,6 +29,7 @@ import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
@@ -61,8 +62,10 @@ import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
import org.apache.gravitino.storage.relational.po.CatalogPO;
import org.apache.gravitino.storage.relational.po.MetalakePO;
+import org.apache.gravitino.storage.relational.po.SchemaPO;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
import org.apache.gravitino.storage.relational.utils.POConverters;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
@@ -72,6 +75,7 @@ import org.apache.ibatis.session.SqlSession;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
+import org.mockito.Mockito;
public class TestCatalogMetaService extends TestJDBCBackend {
@@ -622,6 +626,53 @@ public class TestCatalogMetaService extends
TestJDBCBackend {
assertEquals(0, countActiveTableVersionRows(table.id()));
}
+ @TestTemplate
+ public void testCascadeDeleteCatalogWithMoreSchemasThanOneChunk() throws
IOException {
+ String catalogName = "catalog_many_schemas";
+ CatalogEntity catalog = createAndInsertCatalog(metalakeName, catalogName);
+ int schemaCount = OccWriteSupport.CHILD_DELETE_BATCH_SIZE + 1;
+ for (int i = 0; i < schemaCount; i++) {
+ createAndInsertSchema(metalakeName, catalogName, "schema_" + i);
+ }
+
+ // Make the snapshot of the last schema stale, so only the second chunk
misses a row. The
+ // first chunk is already deleted by then and must roll back with the
failed cascade.
+ CatalogMetaService service = Mockito.spy(CatalogMetaService.getInstance());
+ Mockito.doAnswer(
+ invocation -> {
+ @SuppressWarnings("unchecked")
+ List<SchemaPO> schemaPOs = (List<SchemaPO>)
invocation.callRealMethod();
+ List<SchemaPO> staleSchemaPOs = new ArrayList<>(schemaPOs);
+ SchemaPO last = staleSchemaPOs.get(staleSchemaPOs.size() - 1);
+ staleSchemaPOs.set(
+ staleSchemaPOs.size() - 1,
+ SchemaPO.builder()
+ .withSchemaId(last.getSchemaId())
+ .withSchemaName(last.getSchemaName())
+ .withMetalakeId(last.getMetalakeId())
+ .withCatalogId(last.getCatalogId())
+ .withSchemaComment(last.getSchemaComment())
+ .withProperties(last.getProperties())
+ .withAuditInfo(last.getAuditInfo())
+ .withCurrentVersion(last.getCurrentVersion() + 1)
+ .withLastVersion(last.getLastVersion() + 1)
+ .withDeletedAt(last.getDeletedAt())
+ .build());
+ return staleSchemaPOs;
+ })
+ .when(service)
+ .listSchemaPOsForCascade(catalog.id());
+
+ assertThrows(
+ OptimisticLockException.class, () ->
service.deleteCatalog(catalog.nameIdentifier(), true));
+ assertTrue(backend.exists(catalog.nameIdentifier(),
Entity.EntityType.CATALOG));
+ assertEquals(schemaCount, countActiveSchemas(catalog.id()));
+
+
assertTrue(CatalogMetaService.getInstance().deleteCatalog(catalog.nameIdentifier(),
true));
+ assertFalse(backend.exists(catalog.nameIdentifier(),
Entity.EntityType.CATALOG));
+ assertEquals(0, countActiveSchemas(catalog.id()));
+ }
+
private List<Throwable> insertCatalogsConcurrently(CatalogEntity first,
CatalogEntity second)
throws Exception {
ExecutorService executor = Executors.newFixedThreadPool(2);
@@ -706,6 +757,12 @@ public class TestCatalogMetaService extends
TestJDBCBackend {
}
}
+ private int countActiveSchemas(Long catalogId) {
+ return SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class, mapper ->
mapper.listSchemaPOsByCatalogId(catalogId))
+ .size();
+ }
+
private int countActiveTableVersionRows(Long tableId) {
try (SqlSession sqlSession =
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOccWriteSupport.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOccWriteSupport.java
index c09c433ad1..ed359d5f04 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOccWriteSupport.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestOccWriteSupport.java
@@ -25,6 +25,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
@@ -234,6 +235,56 @@ public class TestOccWriteSupport {
list -> 1));
}
+ @Test
+ void testDeleteChildrenWithVersionsSplitsIntoChunks() {
+ NameIdentifier parentIdent = NameIdentifier.of("parent");
+ int batchSize = OccWriteSupport.CHILD_DELETE_BATCH_SIZE;
+ List<DummyPO> children = new ArrayList<>();
+ for (int i = 0; i < batchSize * 2 + 1; i++) {
+ children.add(new DummyPO("c" + i, 1L));
+ }
+ List<Integer> chunkSizes = new ArrayList<>();
+
+ OccWriteSupport.deleteChildrenWithVersions(
+ parentIdent,
+ Entity.EntityType.SCHEMA,
+ Entity.EntityType.CATALOG,
+ children,
+ chunk -> {
+ chunkSizes.add(chunk.size());
+ return chunk.size();
+ });
+
+ assertEquals(List.of(batchSize, batchSize, 1), chunkSizes);
+ }
+
+ @Test
+ void testDeleteChildrenWithVersionsStopsAtFirstShortChunk() {
+ NameIdentifier parentIdent = NameIdentifier.of("parent");
+ int batchSize = OccWriteSupport.CHILD_DELETE_BATCH_SIZE;
+ List<DummyPO> children = new ArrayList<>();
+ for (int i = 0; i < batchSize * 2 + 1; i++) {
+ children.add(new DummyPO("c" + i, 1L));
+ }
+ List<Integer> chunkSizes = new ArrayList<>();
+
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ OccWriteSupport.deleteChildrenWithVersions(
+ parentIdent,
+ Entity.EntityType.SCHEMA,
+ Entity.EntityType.CATALOG,
+ children,
+ chunk -> {
+ chunkSizes.add(chunk.size());
+ // The second chunk misses one child, as if it was changed
concurrently.
+ return chunkSizes.size() == 2 ? chunk.size() - 1 :
chunk.size();
+ }));
+
+ assertEquals(List.of(batchSize, batchSize), chunkSizes);
+ }
+
@Test
void testLockParentForChildWriteSuccess() {
DummyPO parent = new DummyPO("parent_name", 10L);
