[Feature] S3-based lock to mitigate snapshot conflicts
- Dominant language
- Java
- Stars
- 3.4k
- Forks
- 1.4k
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 396
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.
### Motivation
Current concurrency controls in Paimon rely on an external hive or jdbc metastore to provide locks to mitigate the risk of [snapshot conflicts](https://paimon.apache.org/docs/master/concepts/concurrency-control/).
We would like to run multiple concurrent writers, but without needing to also maintain a metastore service.
Taking inspiration form Hudi, we would like to propose a similar implementation that uses conditional-put semantics, and so the lock can be based on the storage system.
**References**:
- Hudi
- [AWS S3-based lock using conditional requests](https://hudi.apache.org/docs/next/concurrency_control/#storage-with-conditional-writes-based)
- [hudi.client.transaction.lock.StorageBasedLockProvider](https://github.com/apache/hudi/blob/f5f0ef6549fedf93863526a2110fe262a3460432/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java)
- [hudi.aws.transaction.lock.S3StorageLockClient](https://github.com/apache/hudi/blob/master/hudi-aws/src/main/java/org/apache/hudi/aws/transaction/lock/S3StorageLockClient.java)
- [hudi.gcp.transaction.lock.GCSStorageLockClient](https://github.com/apache/hudi/blob/master/hudi-gcp/src/main/java/org/apache/hudi/gcp/transaction/lock/GCSStorageLockClient.java)
- AWS S3
- [S3 If-None-Match header](https://docs.aws.amazon.com/AmazonS3/latest/userguide/conditional-writes.html)
- GCP Storage
- [Conditional write example](https://docs.cloud.google.com/storage/docs/uploading-objects#storage-upload-object-client-libraries)
- [Storage.BlobWriteOption.doesNotExist()](https://docs.cloud.google.com/java/docs/reference/google-cloud-storage/latest/com.google.cloud.storage.Storage.BlobWriteOption#com_google_cloud_storage_Storage_BlobWriteOption_doesNotExist__)
- [Request preconditions](https://docs.cloud.google.com/storage/docs/request-preconditions)
- [Storage JSON api - objects.insert](https://docs.cloud.google.com/storage/docs/json_api/v1/objects/insert)
### Solution
Intended implementation approach:
- New classes in [org.apache.paimon.s3](https://github.com/apache/paimon/tree/master/paimon-filesystems/paimon-s3-impl/src/main/java/org/apache/paimon/s3) and [org.apache.paimon.gs](https://github.com/apache/paimon/tree/master/paimon-filesystems/paimon-gs-impl/src/main/java/org/apache/paimon/gs)
- Use existing methods to extract hadoop configurations (particularly for auth)
- Construct new AWS / GCP storage clients, not relying on Hadoop filesystem (similar to implementation in Hudi)
- Note: Hadoop now has some support for conditional-write semantics, e.g. [HADOOP-19256](https://issues.apache.org/jira/browse/HADOOP-19256), however this is available in Hadoop 3.4.2 onwards, which is not well supported in Flink/Paimon yet
- New classes implement [CatalogLockFactory](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLockFactory.java#L26)
- Extend [CatalogEnvironment](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java) to support adding a lock factory on top of `CatalogEnvironment::empty`, e.g. new method `.withLockFactory(CatalogLockFactory lockFactory)`
To use the lock:
```kotlin
import org.apache.paimon.flink.FlinkFileIO
import org.apache.paimon.fs.Path
import org.apache.paimon.options.Options
import org.apache.paimon.schema.SchemaManager
import org.apache.paimon.table.CatalogEnvironment
import org.apache.paimon.table.FileStoreTableFactory
// assuming an existing table for simplicity
val tablePath = Path("s3a://bucket/prefix")
val fileIO = FlinkFileIO(tablePath)
val writerOptions = Options()
val schema = SchemaManager(fileIO, tablePath).latest().get()
val emptyCatalogWithLock =
CatalogEnvironment
.empty()
.withLockFactory(S3LockFactory()) // NEW
val table = FileStoreTableFactory.create(fileIO, tablePath, schema, writerOptions, emptyCatalogWithLock)
```
**Guidance needed**
- For the purpose of commit locks, the option `lock.type` does not appear to be used when we construct the table via `FileStoreTableFactory.create(...)`.
- Is it necessary to extend catalogs and the catalog option `lock.type` to also support s3/gs locks?
### Anything else?
_No response_
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with CatalogLockFactory, CatalogEnvironment, and FileStoreTableFactory.create(...), then inspect the S3 and GCS implementation directories under paimon-filesystems. Compare the referenced Hudi lock clients and existing Hadoop configuration extraction. Done means the proposed storage-based lock factories and CatalogEnvironment integration support concurrent writers without requiring a metastore.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- aws, gcp, java
- Domain
- cloud, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 30/100