jerryshao commented on code in PR #12157: URL: https://github.com/apache/gravitino/pull/12157#discussion_r3664397476
########## design-docs/treelock-necessity-and-concurrency-design.md: ########## @@ -0,0 +1,528 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. +--> + +# Design: Concurrency Control for Multi-Node Gravitino (TreeLock) + +> Tracking issue: [#10474](https://github.com/apache/gravitino/issues/10474) — *Address TreeLock limitations for Gravitino HA deployment* + +**Short words used in this doc:** + +- **HA** = High Availability = running more than one Gravitino server at the same time behind a load balancer. +- **OCC** = Optimistic Concurrency Control = the caller does not take an application/path lock before work; when it writes, it checks "is the row still the version I read?" If yes, write; if no, someone else changed it first, so handle the conflict. The database still takes its normal short statement/transaction locks. +- **External catalog** = the real system that holds the data, such as Hive, Iceberg, MySQL, or Kafka. +- **Gravitino store** = Gravitino's own database (`RelationalEntityStore`) that keeps a copy of the metadata. +- **Source of truth** = the system whose data we trust as correct when two copies disagree. + +--- + +## Background + +Gravitino uses an in-memory lock called `TreeLock` (`core/src/main/java/org/apache/gravitino/lock/`) to run metadata operations one at a time. For every operation, `TreeLockUtils.doWithTreeLock` locks the whole path from the root down: a read lock on every parent, and a read or write lock on the target (or on its parent for rename/drop). + +``` +loadTable(metalake.cat.db.t1) alterTable(... rename t1) dropTable(metalake.cat.db.t1) + / READ / READ / READ + /metalake READ /metalake READ /metalake READ + /metalake/cat READ /metalake/cat READ /metalake/cat READ + /metalake/cat/db READ /metalake/cat/db WRITE /metalake/cat/db WRITE + /metalake/cat/db/t1 READ (parent write-locked) (parent write-locked) +``` + +This is built on `LockManager`, which keeps an in-memory tree of `TreeLockNode`s (each one wraps a `ReentrantReadWriteLock`), plus reference counting, a background thread that removes unused nodes, and another background thread that checks for deadlocks inside the same JVM. It is about 800 lines of code in total (`LockManager` ~300, `TreeLockNode` ~250, `TreeLock` ~180, `TreeLockUtils` ~70). + +### The problem: TreeLock only works inside one JVM + +Each Gravitino server has its **own** `LockManager` with its own lock tree in its own memory. A write lock taken on server A means nothing to server B. Behind a load balancer, two servers can each pass their *local* TreeLock and change the same resource at the same time. So the moment Gravitino runs in HA, TreeLock stops protecting anything across servers. This is the reason #10474 was opened. + +This leaves us with a decision. There are two directions: + +1. **Remove TreeLock's correctness role** and let explicit rules in the shared database keep data correct. +2. **Keep the lock but make it work across nodes** (a distributed lock). + +The rest of this document analyses what TreeLock really does today, then compares these two directions, then picks one. + +--- + +## Goals + +1. **Understand what TreeLock protects today**: Describe what TreeLock actually guards, and which of those guarantees the shared database could provide instead. +2. **Evaluate the candidate directions on equal footing**: Assess each direction — database-native concurrency, and a cross-node distributed lock — against the same criteria (correctness, performance, maintainability, operational cost), informed by how comparable systems solve the two-store problem. Neither direction is assumed better going in. +3. **Correctness for one and for many servers**: Whatever is chosen must be correct with a single server and under HA. +4. **Decide with evidence**: End with a direction and the reasons behind it, plus a plan a developer can start on. + +--- + +## Non-Goals + +1. **No single transaction across the external catalog and the Gravitino store**: We will not try 2PC/XA across the two stores. External catalogs do not all offer the same transaction guarantees, so the existing pattern (re-sync on read, which is safe to repeat) is kept. +2. **Write correctness only, not read staleness**: This document is about correct concurrent *writes*. Keeping each server's *cached reads* fresh across nodes (the `EntityChangeLogPoller` work) is a separate effort, out of scope here. +3. **No change to external-catalog behavior**: We will not change how Hive/Iceberg/JDBC connectors keep their own data correct; they are the source of truth for their own data. + +--- + +## Analysis and Investigation + +### A metadata write touches two stores, with no shared transaction + +The key fact that is easy to miss: **every catalog metadata operation touches two separate stores, and there is no single transaction that covers both of them**: + +1. The **external catalog** (Hive, Iceberg REST, JDBC/MySQL, Kafka, …). +2. Gravitino's **own database**, the Gravitino store. + +The order is always the same: **the external system first, the Gravitino store second.** From `TableOperationDispatcher`: + +```text +internalCreateTable(): catalog.createTable(...) → store.put(tableEntity) // lines 642-689 +dropTable(): catalog.dropTable(...) → store.delete(ident, TABLE) // lines 366-386 +alterTable(): catalog.alterTable(...) → store column-sync // lines 267-340 +importTable(): catalog.loadTable(...) → store.put(tableEntity) // lines 474-527 +``` + + + +When a table is created, its Gravitino id is written **into the external table's own properties** as a `StringIdentifier` (`internalCreateTable`, line 636). Later, when the table is read, `importTable` uses the data from the external system to **overwrite and correct** the stored copy. So for external-backed catalogs, the external system is the source of truth, and the Gravitino store is a copy that is updated later and fixes itself on the next read. + +An important result: **no lock can make the two stores update as one unit.** If the process crashes after `catalog.createTable` finished but before `store.put` runs, the external system has a table with no matching Gravitino entity. This is a crash problem, not a "two things at once" problem, and neither a local nor a distributed lock can fix it. This already tells us that a lock is not the tool that keeps the two stores matched. + +### How the store fixes itself, and where it stops working + +Here is how the self-fix works, traced through the code. If create succeeds in the external system but the store write fails, the error is **hidden** — `internalCreateTable:688-696` catches it, logs it, and still returns success to the client with no stored entity. Nothing is fixed until someone reads the table again. On that next read, `loadTable:142` sees `imported == false` and runs `importTable`, which writes the entity into the store. Schemas behave the same way (`internalLoadSchema` + `importSchema`). So a failed store write **does** auto-correct on the next load. + + + +The import step does not need the id to run; the id only decides how stable the result is. The details: + +| External system | How "needs import" is detected | Id after the self-fix | +| ----------------------------------------------------------------------- | ------------------------------------------------------------------------------ | --------------------------------------------------------------------------- | +| Can store the id (e.g. Hive, Iceberg) — `stringId != null` | by id — `store.get(stringId.id())` (`internalLoadTable:596`) | **the original id is reused** (`importTable` sets `uid = stringId.id()`) | +| Cannot store the id (e.g. JDBC schema, PostgreSQL) — `stringId == null` | by name — `getEntity(ident)` (`internalLoadTable:570`, code comment at `:590`) | **a new id is generated** (`importTable` sets `uid = idGenerator.nextId()`) | + +The narrow limit: for an id-less external system the id is really owned by the Gravitino store, so if that row is lost the id cannot be recovered, and anything that references the entity by id (owner, tag, policy, role) points at the old, now-missing id. This does not appear in the plain "create then store write fails" flow, because no id-based references exist yet. + +### Not every catalog has an external system that decides the winner + +Whether the external system can act as the judge depends on the catalog. In the code this is the `managedStorage` capability (`Capability` in `core/.../connector/capability/Capability.java`; the default returns "managed" only for functions). The catalogs split into two groups: + +| Group | Catalogs | Source of truth for create/drop | +| ------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------- | +| **External-backed** (`managedStorage` = false for the entity) | Hive, Glue, JDBC (MySQL/PostgreSQL/Doris/OceanBase/StarRocks/BigQuery), Iceberg, Hudi, Paimon | the **external system** | +| **Gravitino-managed** (`managedStorage` = true) | fileset (schema, fileset), model (schema, model), lakehouse-generic (schema, table), kafka (schema; the topic itself is external), function | the **Gravitino store** | + +This matters for the comparison below: + +- For **external-backed** catalogs, the external system is a single shared authority that all Gravitino nodes talk to. It decides the final state according to its own concurrency semantics. For example, a database normally rejects a duplicate `CREATE TABLE`; Hive Metastore accepts full-table replacement alters without exposing a Gravitino-compatible version/CAS contract, so concurrent alters are not guaranteed to merge and may be last-writer-wins. Gravitino does not promise stronger merge semantics than the external catalog. Because TreeLock is per-node — and ordinary table alter currently takes a TreeLock read lock, so same-table alters can overlap even in one JVM — **cross-node behavior here already depends on the external system today, not on TreeLock.** +- For **Gravitino-managed** catalogs (fileset, model, and so on) there is **no external judge**. A fileset's "external" side is just a directory on HDFS or S3, which has no uniqueness check and no parent/child rule. For these, correctness can only come from the Gravitino store. + +#### A concrete lost write on HMS and Glue, and what can be done + +HMS `alter_table(db, table, newTable)` carries only the new object, never the base the caller read, so the server has nothing to compare against. The connector is a plain read-modify-write (`HiveCatalogOperations:692-746`, still carrying `// TODO(@Minghuang): require a table lock to avoid race condition`): + +| Step | Server A | Server B | +|------|--------------------------------------------|-----------------------------------------| +| 1 | loads t1: columns `[a]`, comment `c0` | | +| 2 | | loads t1: the same base | +| 3 | writes back columns `[a, b]`, comment `c0` | | +| 4 | | writes back columns `[a]`, comment `c1` | + +Both calls succeed, the table ends with comment `c1`, and **column `b` is gone** although B never touched columns. Same-field conflicts resolving to last-writer-wins would be expected, as in a SQL `UPDATE`; the defect is the collateral loss of a field the winner never edited. Gravitino does not notice either, because the connector returns the locally built object instead of re-reading (`HiveCatalogOperations:750`), so the store row mirrors a state HMS never held — and that row does not self-heal on read. + +**Glue has the same defect**, for the same reason: `getTable` → build a full `TableInput` in memory → `UpdateTable` → return the locally built object (`GlueCatalogOperations:500-583`; its Iceberg tables take a separate branch and are unaffected). The other external-backed backends do not have this shape: JDBC translates the changes into an incremental `ALTER TABLE` statement and then re-reads, Iceberg commits an incremental change list under its native OCC, Paimon applies incremental `SchemaChange`s and re-reads. So this is a defect of the two HMS-shaped connectors, not of external-backed catalogs in general. + +A lock does not fix this: a fencing token has nothing to validate against on HMS, and it would only serialize callers going through Gravitino while Spark, Trino and the Hive CLI write to the same HMS directly. The affordable fix is detection rather than mutual exclusion: **re-read after `alter_table`, verify the requested changes are present, and on a mismatch re-apply the change list to the fresh base with bounded retries**, then mirror the re-read result. One extra RPC, no new component, converges for disjoint changes, and detects out-of-band writers too; only the two connectors above need it, so a capability flag keeps the rest on the fast path. Hive 4's `AlterTableRequest.expectedParameterKey/Value` is a real single-parameter CAS, but Gravitino builds against Hive `2.3.9`/`3.1.3`, so it stays a future option. Review Comment: The read after alter check mechanism seems reasonable. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
