This is an automated email from the ASF dual-hosted git repository.
jshao pushed a commit to branch branch-0.9
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-0.9 by this push:
new ee7ee5115c [#2904] feat(fileset): add cache for fileset (#6995)
ee7ee5115c is described below
commit ee7ee5115c03a581e40236d421809ae7aabaf1e5
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Apr 17 21:10:28 2025 +0800
[#2904] feat(fileset): add cache for fileset (#6995)
### What changes were proposed in this pull request?
add cache for fileset
### Why are the changes needed?
Fix: #2904
### Does this PR introduce _any_ user-facing change?
no
### How was this patch tested?
tests added
Co-authored-by: mchades <[email protected]>
---
catalogs/catalog-hadoop/build.gradle.kts | 1 +
.../catalog/hadoop/HadoopCatalogOperations.java | 140 +++++++++++++++------
.../hadoop/HadoopCatalogPropertiesMetadata.java | 26 ++++
.../hadoop/integration/test/HadoopCatalogIT.java | 71 +++++++++++
docs/hadoop-catalog.md | 18 +--
5 files changed, 213 insertions(+), 43 deletions(-)
diff --git a/catalogs/catalog-hadoop/build.gradle.kts
b/catalogs/catalog-hadoop/build.gradle.kts
index b8877646f5..cfcf8b0c5e 100644
--- a/catalogs/catalog-hadoop/build.gradle.kts
+++ b/catalogs/catalog-hadoop/build.gradle.kts
@@ -40,6 +40,7 @@ dependencies {
implementation(project(":core")) {
exclude(group = "*")
}
+ implementation(libs.caffeine)
implementation(libs.commons.lang3)
implementation(libs.commons.io)
implementation(libs.hadoop3.client.api)
diff --git
a/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogOperations.java
b/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogOperations.java
index b61000306f..0b94286690 100644
---
a/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogOperations.java
+++
b/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogOperations.java
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.catalog.hadoop;
+import static
org.apache.gravitino.catalog.hadoop.HadoopCatalogPropertiesMetadata.CACHE_VALUE_NOT_SET;
import static org.apache.gravitino.connector.BaseCatalog.CATALOG_BYPASS_PREFIX;
import static org.apache.gravitino.file.Fileset.LOCATION_NAME_UNKNOWN;
import static org.apache.gravitino.file.Fileset.PROPERTY_CATALOG_PLACEHOLDER;
@@ -27,10 +28,14 @@ import static
org.apache.gravitino.file.Fileset.PROPERTY_LOCATION_PLACEHOLDER_PR
import static
org.apache.gravitino.file.Fileset.PROPERTY_MULTIPLE_LOCATIONS_PREFIX;
import static org.apache.gravitino.file.Fileset.PROPERTY_SCHEMA_PLACEHOLDER;
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Scheduler;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.time.Instant;
@@ -41,10 +46,12 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.Entity;
@@ -119,6 +126,8 @@ public class HadoopCatalogOperations extends
ManagedSchemaOperations
private boolean disableFSOps;
+ private Cache<NameIdentifier, HadoopFileset> filesetCache;
+
HadoopCatalogOperations(EntityStore store) {
this.store = store;
}
@@ -181,6 +190,7 @@ public class HadoopCatalogOperations extends
ManagedSchemaOperations
}
this.catalogStorageLocations = getAndCheckCatalogStorageLocations(config);
+ this.filesetCache = initializeFilesetCache(config);
}
@Override
@@ -203,24 +213,28 @@ public class HadoopCatalogOperations extends
ManagedSchemaOperations
@Override
public Fileset loadFileset(NameIdentifier ident) throws
NoSuchFilesetException {
- try {
- FilesetEntity filesetEntity =
- store.get(ident, Entity.EntityType.FILESET, FilesetEntity.class);
-
- return HadoopFileset.builder()
- .withName(ident.name())
- .withType(filesetEntity.filesetType())
- .withComment(filesetEntity.comment())
- .withStorageLocations(filesetEntity.storageLocations())
- .withProperties(filesetEntity.properties())
- .withAuditInfo(filesetEntity.auditInfo())
- .build();
-
- } catch (NoSuchEntityException exception) {
- throw new NoSuchFilesetException(exception, FILESET_DOES_NOT_EXIST_MSG,
ident);
- } catch (IOException ioe) {
- throw new RuntimeException("Failed to load fileset %s" + ident, ioe);
- }
+ return filesetCache.get(
+ ident,
+ k -> {
+ try {
+ FilesetEntity filesetEntity =
+ store.get(ident, Entity.EntityType.FILESET,
FilesetEntity.class);
+
+ return HadoopFileset.builder()
+ .withName(ident.name())
+ .withType(filesetEntity.filesetType())
+ .withComment(filesetEntity.comment())
+ .withStorageLocations(filesetEntity.storageLocations())
+ .withProperties(filesetEntity.properties())
+ .withAuditInfo(filesetEntity.auditInfo())
+ .build();
+
+ } catch (NoSuchEntityException exception) {
+ throw new NoSuchFilesetException(exception,
FILESET_DOES_NOT_EXIST_MSG, ident);
+ } catch (IOException ioe) {
+ throw new RuntimeException("Failed to load fileset %s" + ident,
ioe);
+ }
+ });
}
@Override
@@ -238,6 +252,12 @@ public class HadoopCatalogOperations extends
ManagedSchemaOperations
}
});
+ // Check if the fileset already existed in cache first. If it does, it
means the fileset is
+ // already created, so we should throw an exception.
+ if (filesetCache.getIfPresent(ident) != null) {
+ throw new FilesetAlreadyExistsException("Fileset %s already exists",
ident);
+ }
+
try {
if (store.exists(ident, Entity.EntityType.FILESET)) {
throw new FilesetAlreadyExistsException("Fileset %s already exists",
ident);
@@ -380,14 +400,17 @@ public class HadoopCatalogOperations extends
ManagedSchemaOperations
throw new RuntimeException("Failed to create fileset " + ident, ioe);
}
- return HadoopFileset.builder()
- .withName(ident.name())
- .withComment(comment)
- .withType(type)
- .withStorageLocations(formattedStorageLocations)
- .withProperties(filesetEntity.properties())
- .withAuditInfo(filesetEntity.auditInfo())
- .build();
+ HadoopFileset fileset =
+ HadoopFileset.builder()
+ .withName(ident.name())
+ .withComment(comment)
+ .withType(type)
+ .withStorageLocations(formattedStorageLocations)
+ .withProperties(filesetEntity.properties())
+ .withAuditInfo(filesetEntity.auditInfo())
+ .build();
+ filesetCache.put(ident, fileset);
+ return fileset;
}
private Map<String, String> setDefaultLocationIfAbsent(
@@ -441,6 +464,7 @@ public class HadoopCatalogOperations extends
ManagedSchemaOperations
throw new RuntimeException("Failed to load fileset " + ident, ioe);
}
+ filesetCache.invalidate(ident);
try {
FilesetEntity updatedFilesetEntity =
store.update(
@@ -449,14 +473,17 @@ public class HadoopCatalogOperations extends
ManagedSchemaOperations
Entity.EntityType.FILESET,
e -> updateFilesetEntity(ident, e, changes));
- return HadoopFileset.builder()
- .withName(updatedFilesetEntity.name())
- .withComment(updatedFilesetEntity.comment())
- .withType(updatedFilesetEntity.filesetType())
- .withStorageLocations(updatedFilesetEntity.storageLocations())
- .withProperties(updatedFilesetEntity.properties())
- .withAuditInfo(updatedFilesetEntity.auditInfo())
- .build();
+ HadoopFileset fileset =
+ HadoopFileset.builder()
+ .withName(updatedFilesetEntity.name())
+ .withComment(updatedFilesetEntity.comment())
+ .withType(updatedFilesetEntity.filesetType())
+ .withStorageLocations(updatedFilesetEntity.storageLocations())
+ .withProperties(updatedFilesetEntity.properties())
+ .withAuditInfo(updatedFilesetEntity.auditInfo())
+ .build();
+ filesetCache.put(updatedFilesetEntity.nameIdentifier(), fileset);
+ return fileset;
} catch (IOException ioe) {
throw new RuntimeException("Failed to update fileset " + ident, ioe);
@@ -514,6 +541,7 @@ public class HadoopCatalogOperations extends
ManagedSchemaOperations
}
}
+ filesetCache.invalidate(ident);
return store.delete(ident, Entity.EntityType.FILESET);
} catch (NoSuchEntityException ne) {
LOG.warn("Fileset {} does not exist", ident);
@@ -677,6 +705,8 @@ public class HadoopCatalogOperations extends
ManagedSchemaOperations
throw new RuntimeException("Failed to check if schema " + ident + "
exists", ioe);
}
+ // note: we need to invalidate the related fileset cache when the schema
rename change is
+ // supported.
return super.alterSchema(ident, changes);
}
@@ -702,6 +732,8 @@ public class HadoopCatalogOperations extends
ManagedSchemaOperations
Map<String, Path> schemaPaths = getAndCheckSchemaPaths(ident.name(),
properties);
boolean dropped = super.dropSchema(ident, cascade);
+ filesetCache.invalidateAll(
+
filesets.stream().map(FilesetEntity::nameIdentifier).collect(Collectors.toList()));
if (disableFSOps) {
return dropped;
}
@@ -831,7 +863,45 @@ public class HadoopCatalogOperations extends
ManagedSchemaOperations
}
@Override
- public void close() throws IOException {}
+ public void close() throws IOException {
+ filesetCache.invalidateAll();
+ }
+
+ private Cache<NameIdentifier, HadoopFileset>
initializeFilesetCache(Map<String, String> config) {
+ Caffeine<Object, Object> cacheBuilder =
+ Caffeine.newBuilder()
+ .removalListener(
+ (k, v, c) -> LOG.info("Evicting fileset {} from cache due to
{}", k, c))
+ .scheduler(
+ Scheduler.forScheduledExecutorService(
+ new ScheduledThreadPoolExecutor(
+ 1,
+ new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat("fileset-cleaner-%d")
+ .build())));
+
+ Long cacheEvictionIntervalInMs =
+ (Long)
+ propertiesMetadata
+ .catalogPropertiesMetadata()
+ .getOrDefault(
+ config,
HadoopCatalogPropertiesMetadata.FILESET_CACHE_EVICTION_INTERVAL_MS);
+ if (cacheEvictionIntervalInMs != CACHE_VALUE_NOT_SET) {
+ cacheBuilder.expireAfterAccess(cacheEvictionIntervalInMs,
TimeUnit.MILLISECONDS);
+ }
+
+ Long cacheMaxSize =
+ (Long)
+ propertiesMetadata
+ .catalogPropertiesMetadata()
+ .getOrDefault(config,
HadoopCatalogPropertiesMetadata.FILESET_CACHE_MAX_SIZE);
+ if (cacheMaxSize != CACHE_VALUE_NOT_SET) {
+ cacheBuilder.maximumSize(cacheMaxSize);
+ }
+
+ return cacheBuilder.build();
+ }
private SchemaEntity updateSchemaEntity(
NameIdentifier ident, SchemaEntity schemaEntity, SchemaChange...
changes) {
diff --git
a/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogPropertiesMetadata.java
b/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogPropertiesMetadata.java
index 2aadea59ae..1c9d566b17 100644
---
a/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogPropertiesMetadata.java
+++
b/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogPropertiesMetadata.java
@@ -56,6 +56,16 @@ public class HadoopCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetada
*/
public static final String DEFAULT_FS_PROVIDER =
"default-filesystem-provider";
+ /** The interval in milliseconds to evict the fileset cache. */
+ public static final String FILESET_CACHE_EVICTION_INTERVAL_MS =
+ "fileset-cache-eviction-interval-ms";
+
+ /** The maximum number of the filesets the cache may contain. */
+ public static final String FILESET_CACHE_MAX_SIZE = "fileset-cache-max-size";
+
+ /** The value to indicate the cache value is not set. */
+ public static final long CACHE_VALUE_NOT_SET = -1;
+
static final String FILESYSTEM_CONNECTION_TIMEOUT_SECONDS =
"filesystem-conn-timeout-secs";
static final int DEFAULT_GET_FILESYSTEM_TIMEOUT_SECONDS = 6;
@@ -117,6 +127,22 @@ public class HadoopCatalogPropertiesMetadata extends
BaseCatalogPropertiesMetada
false /* immutable */,
DEFAULT_GET_FILESYSTEM_TIMEOUT_SECONDS,
false /* hidden */))
+ .put(
+ FILESET_CACHE_EVICTION_INTERVAL_MS,
+ PropertyEntry.longOptionalPropertyEntry(
+ FILESET_CACHE_EVICTION_INTERVAL_MS,
+ "The interval in milliseconds to evict the fileset cache, -1
means never evict.",
+ false /* immutable */,
+ 60 * 60 * 1000L /* 1 hour */,
+ false /* hidden */))
+ .put(
+ FILESET_CACHE_MAX_SIZE,
+ PropertyEntry.longOptionalPropertyEntry(
+ FILESET_CACHE_MAX_SIZE,
+ "The maximum number of the filesets the cache may contain,
-1 means no limit.",
+ false /* immutable */,
+ 200_000L,
+ false /* hidden */))
.put(
DISABLE_FILESYSTEM_OPS,
PropertyEntry.booleanPropertyEntry(
diff --git
a/catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java
b/catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java
index ee77d11e51..2cb6bd327b 100644
---
a/catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java
+++
b/catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java
@@ -43,6 +43,7 @@ import org.apache.gravitino.audit.InternalClientType;
import org.apache.gravitino.client.GravitinoMetalake;
import org.apache.gravitino.exceptions.FilesetAlreadyExistsException;
import org.apache.gravitino.exceptions.IllegalNameIdentifierException;
+import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchFilesetException;
import org.apache.gravitino.file.Fileset;
import org.apache.gravitino.file.FilesetChange;
@@ -150,6 +151,76 @@ public class HadoopCatalogIT extends BaseIT {
Assertions.assertFalse(catalog.asSchemas().schemaExists(schemaName));
}
+ @Test
+ public void testFilesetCache() {
+ Assumptions.assumeTrue(getClass() == HadoopCatalogIT.class);
+ String catalogName = GravitinoITUtils.genRandomName("test_fileset_cache");
+ String newCatalogName = catalogName + "_new";
+ String filesetName =
GravitinoITUtils.genRandomName("test_fileset_cache_fileset");
+ String location = defaultBaseLocation() + "/" + filesetName;
+ Map<String, String> catalogProperties = ImmutableMap.of("location",
location);
+
+ Catalog filesetCatalog =
+ metalake.createCatalog(
+ catalogName, Catalog.Type.FILESET, provider, null,
catalogProperties);
+ filesetCatalog.asSchemas().createSchema(schemaName, null, null);
+ filesetCatalog
+ .asFilesetCatalog()
+ .createFileset(NameIdentifier.of(schemaName, filesetName), null,
MANAGED, null, null);
+
+ // Load fileset
+ Fileset fileset =
+
filesetCatalog.asFilesetCatalog().loadFileset(NameIdentifier.of(schemaName,
filesetName));
+ Assertions.assertEquals(filesetName, fileset.name());
+
+ // rename catalog and load fileset
+ Catalog alteredCatalog =
+ metalake.alterCatalog(catalogName,
CatalogChange.rename(newCatalogName));
+ Assertions.assertTrue(
+ alteredCatalog
+ .asFilesetCatalog()
+ .filesetExists(NameIdentifier.of(schemaName, filesetName)));
+ Assertions.assertThrows(
+ NoSuchCatalogException.class,
+ () ->
+ filesetCatalog
+ .asFilesetCatalog()
+ .filesetExists(NameIdentifier.of(schemaName, filesetName)));
+
+ // rename fileset and load fileset
+ String newFilesetName = filesetName + "_new";
+ Fileset alteredFileset =
+ alteredCatalog
+ .asFilesetCatalog()
+ .alterFileset(
+ NameIdentifier.of(schemaName, filesetName),
FilesetChange.rename(newFilesetName));
+ Assertions.assertEquals(newFilesetName, alteredFileset.name());
+ Assertions.assertTrue(
+ alteredCatalog
+ .asFilesetCatalog()
+ .filesetExists(NameIdentifier.of(schemaName, newFilesetName)));
+
+ // drop schema and load fileset
+ alteredCatalog.asSchemas().dropSchema(schemaName, true);
+ Assertions.assertFalse(
+ alteredCatalog
+ .asFilesetCatalog()
+ .filesetExists(NameIdentifier.of(schemaName, newFilesetName)));
+ Assertions.assertFalse(
+ alteredCatalog
+ .asFilesetCatalog()
+ .filesetExists(NameIdentifier.of(schemaName, filesetName)));
+
+ // drop catalog and load fileset
+ metalake.dropCatalog(newCatalogName, true);
+ Assertions.assertThrows(
+ NoSuchCatalogException.class,
+ () ->
+ alteredCatalog
+ .asFilesetCatalog()
+ .filesetExists(NameIdentifier.of(schemaName, newFilesetName)));
+ }
+
@Test
void testAlterCatalogLocation() throws IOException {
Assumptions.assumeTrue(getClass() == HadoopCatalogIT.class);
diff --git a/docs/hadoop-catalog.md b/docs/hadoop-catalog.md
index c187a80f3b..cea25e6490 100644
--- a/docs/hadoop-catalog.md
+++ b/docs/hadoop-catalog.md
@@ -23,14 +23,16 @@ Hadoop 3. If there's any compatibility issue, please create
an [issue](https://g
Besides the [common catalog
properties](./gravitino-server-config.md#apache-gravitino-catalog-properties-configuration),
the Hadoop catalog has the following properties:
-| Property Name | Description
| Default Value | Required | Since Version |
-|--------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------|----------|------------------|
-| `location` | The storage location managed by Hadoop
catalog.
| (none) | No | 0.5.0 |
-| `default-filesystem-provider` | The default filesystem provider of this
Hadoop catalog if users do not specify the scheme in the URI. Candidate values
are 'builtin-local', 'builtin-hdfs', 's3', 'gcs', 'abs' and 'oss'. Default
value is `builtin-local`. For S3, if we set this value to 's3', we can omit the
prefix 's3a://' in the location. | `builtin-local` | No |
0.7.0-incubating |
-| `filesystem-providers` | The file system providers to add. Users
need to set this configuration to support cloud storage or custom HCFS. For
instance, set it to `s3` or a comma separated string that contains `s3` like
`gs,s3` to support multiple kinds of fileset including `s3`.
| (none) | Yes |
0.7.0-incubating |
-| `credential-providers` | The credential provider types, separated by
comma.
| (none) | No | 0.8.0-incubating |
-| `filesystem-conn-timeout-secs` | The timeout of getting the file system
using Hadoop FileSystem client instance. Time unit: seconds.
| 6 | No | 0.8.0-incubating |
-| `disable-filesystem-ops` | The configuration to disable file system
operations in the server side. If set to true, the Hadoop catalog in the server
side will not create, drop files or folder when the schema, fileset is created,
dropped.
| false | No | 0.9.0-incubating |
+| Property Name | Description
| Default Value | Required | Since Version |
+|--------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------|----------|------------------|
+| `location` | The storage location managed by
Hadoop catalog.
| (none) | No | 0.5.0
|
+| `default-filesystem-provider` | The default filesystem provider of
this Hadoop catalog if users do not specify the scheme in the URI. Candidate
values are 'builtin-local', 'builtin-hdfs', 's3', 'gcs', 'abs' and 'oss'.
Default value is `builtin-local`. For S3, if we set this value to 's3', we can
omit the prefix 's3a://' in the location. | `builtin-local` | No |
0.7.0-incubating |
+| `filesystem-providers` | The file system providers to add.
Users need to set this configuration to support cloud storage or custom HCFS.
For instance, set it to `s3` or a comma separated string that contains `s3`
like `gs,s3` to support multiple kinds of fileset including `s3`.
| (none) | Yes |
0.7.0-incubating |
+| `credential-providers` | The credential provider types,
separated by comma.
| (none) | No |
0.8.0-incubating |
+| `filesystem-conn-timeout-secs` | The timeout of getting the file
system using Hadoop FileSystem client instance. Time unit: seconds.
| 6 | No |
0.8.0-incubating |
+| `disable-filesystem-ops` | The configuration to disable file
system operations in the server side. If set to true, the Hadoop catalog in the
server side will not create, drop files or folder when the schema, fileset is
created, dropped.
| false | No |
0.9.0-incubating |
+| `fileset-cache-eviction-interval-ms` | The interval in milliseconds to evict
the fileset cache, -1 means never evict.
| 3600000 | No | 0.9.0-incubating |
+| `fileset-cache-max-size` | The maximum number of the filesets
the cache may contain, -1 means no limit.
| 200000 | No | 0.9.0-incubating
|
Please refer to [Credential vending](./security/credential-vending.md) for
more details about credential vending.
