jerryshao commented on code in PR #8696:
URL: https://github.com/apache/gravitino/pull/8696#discussion_r2444316969
##########
docs/lakehouse-iceberg-catalog.md:
##########
@@ -167,6 +168,18 @@ Users can use the following properties to configure the
security of the catalog
| `authentication.kerberos.check-interval-sec` | The check interval of
Kerberos credential for Iceberg catalog.
| 60 | No
| 0.6.0-incubating |
| `authentication.kerberos.keytab-fetch-timeout-sec` | The fetch timeout of
retrieving Kerberos keytab from `authentication.kerberos.keytab-uri`.
| 60 |
No
| 0.6.0-incubating |
+#### Table metadata cache
+
+Gravitino features a pluggable cache system for updating or retrieving table
metadata caches. It validates the location of table metadata against the
catalog backend to ensure the correctness of cached data.
Review Comment:
"..updating or retrieving table metadata in cache."
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java:
##########
@@ -306,4 +336,30 @@ public IcebergTableChange(TableIdentifier tableIdentifier,
Transaction transacti
this.transaction = transaction;
}
}
+
+ private TableMetadataCache loadTableMetadataCache(IcebergConfig config,
Catalog catalog) {
+ String impl = config.get(IcebergConfig.TABLE_METADATA_CACHE_IMPL);
+ if (StringUtils.isBlank(impl)) {
+ return TableMetadataCache.DUMMY;
+ }
+
+ Preconditions.checkArgument(
+ catalog instanceof SupportsMetadataLocation,
+ "You shouldn't enable Iceberg metadata cache for the catalog %s,
because the catalog impl does not support get metadata location.",
Review Comment:
This line is too long.
##########
iceberg/iceberg-common/src/main/java/org/apache/iceberg/jdbc/JdbcCatalogWithMetadataLocation.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * 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.
+ */
+
+package org.apache.iceberg.jdbc;
+
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.iceberg.common.cache.SupportsMetadataLocation;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.jdbc.JdbcUtil.SchemaVersion;
+
+// Use Iceberg package to reuse JdbcUtil related classes.
+public class JdbcCatalogWithMetadataLocation extends JdbcCatalog
Review Comment:
Also here.
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/cache/LocalMetadataCache.java:
##########
@@ -0,0 +1,86 @@
+/*
+ * 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.
+ */
+
+package org.apache.gravitino.iceberg.common.cache;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LocalMetadataCache extends BaseMetadataCache {
+ public static final Logger LOG =
LoggerFactory.getLogger(LocalMetadataCache.class);
+ private Cache<TableIdentifier, TableMetadata> tableMetadataCache;
+
+ @Override
+ public void initialize(
+ int capacity,
+ int expireMinutes,
+ Map<String, String> catalogProperties,
+ SupportsMetadataLocation supportsMetadataLocation) {
+ super.initialize(supportsMetadataLocation);
+ this.tableMetadataCache =
+ Caffeine.newBuilder()
+ .maximumSize(capacity)
+ .expireAfterAccess(expireMinutes, TimeUnit.MINUTES)
+ .build();
+ }
+
+ @Override
+ public void invalidate(TableIdentifier tableIdentifier) {
+ LOG.debug("Invalidate table cache, table identifier: {}", tableIdentifier);
+ tableMetadataCache.invalidate(tableIdentifier);
+ }
+
+ @Override
+ public void updateTableMetadata(TableIdentifier tableIdentifier,
TableMetadata tableMetadata) {
+ LOG.debug(
+ "Update table cache, table identifier: {}, table metadata location:
{}",
+ tableIdentifier,
+ tableMetadata.metadataFileLocation());
+ tableMetadataCache.put(tableIdentifier, tableMetadata);
+ // Clean up expired entries
+ tableMetadataCache.cleanUp();
Review Comment:
I don't think so. Caffeine has two cleanup mechanisms: one is time-based,
another is size-based. It will be invoked internally. I think Caffeine is
mature enough to handle this problem. Do you have any documentation that
mentions this problem?
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/cache/LocalMetadataCache.java:
##########
@@ -0,0 +1,87 @@
+/*
+ * 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.
+ */
+
+package org.apache.gravitino.iceberg.common.cache;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LocalMetadataCache extends BaseMetadataCache {
Review Comment:
`LocalTableMetadataCache`.
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/cache/BaseMetadataCache.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.
+ */
+
+package org.apache.gravitino.iceberg.common.cache;
+
+import java.util.Optional;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Abstract base class implementing {@link TableMetadataCache} that provides
core metadata caching
+ * functionality with validation of metadata location against the latest
version.
+ */
+public abstract class BaseMetadataCache implements TableMetadataCache {
Review Comment:
`BaseTableMetadataCache`.
##########
iceberg/iceberg-common/src/main/java/org/apache/iceberg/hive/HiveCatalogWithMetadataLocation.java:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.
+ */
+
+package org.apache.iceberg.hive;
+
+import com.google.common.base.Preconditions;
+import java.util.Map;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.iceberg.common.ClosableHiveCatalog;
+import org.apache.gravitino.iceberg.common.cache.SupportsMetadataLocation;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.iceberg.BaseMetastoreTableOperations;
+import org.apache.iceberg.ClientPool;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.thrift.TException;
+
+public class HiveCatalogWithMetadataLocation extends ClosableHiveCatalog
Review Comment:
I would suggest renaming it to `MetadataLocationSupportedHiveCatalog` or
`HiveCatalogWithMetadataLocationSupport`
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/cache/BaseMetadataCache.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.
+ */
+
+package org.apache.gravitino.iceberg.common.cache;
+
+import java.util.Optional;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Abstract base class implementing {@link TableMetadataCache} that provides
core metadata caching
+ * functionality with validation of metadata location against the latest
version.
+ */
+public abstract class BaseMetadataCache implements TableMetadataCache {
+
+ public static final Logger LOG =
LoggerFactory.getLogger(BaseMetadataCache.class);
+ /** Component to retrieve the latest metadata location for a table, used for
validation. */
+ private SupportsMetadataLocation supportsMetadataLocation;
+
+ /**
+ * Abstract method to retrieve cached {@link TableMetadata} for a table.
Subclasses must implement
+ * this to provide actual cache retrieval logic.
+ *
+ * @param tableIdentifier Identifier of the table to retrieve cached
metadata for
+ * @return Cached {@link TableMetadata} wrapped in an {@link Optional}, or
an empty {@link
+ * Optional} if not found in cache
+ */
+ protected abstract Optional<TableMetadata>
doGetTableMetadata(TableIdentifier tableIdentifier);
+
+ protected void initialize(SupportsMetadataLocation supportsMetadataLocation)
{
+ this.supportsMetadataLocation = supportsMetadataLocation;
+ }
+
+ /**
+ * Retrieves and validates cached table metadata by comparing the cached
metadata's location with
+ * the latest known location. Invalidates the cache if locations mismatch.
+ *
+ * @param tableIdentifier Identifier of the table to retrieve metadata for
+ * @return Valid {@link TableMetadata} if cache is valid and locations
match; {@code null} if
+ * cache miss, location is invalid, or locations mismatch (after
invalidation)
+ */
+ @Override
+ public Optional<TableMetadata> getTableMetadata(TableIdentifier
tableIdentifier) {
+ Optional<TableMetadata> tableMetadataOptional =
doGetTableMetadata(tableIdentifier);
+ if (!tableMetadataOptional.isPresent()) {
+ return Optional.empty();
+ }
+ TableMetadata tableMetadata = tableMetadataOptional.get();
+ String latestLocation =
supportsMetadataLocation.metadataLocation(tableIdentifier);
+ if (latestLocation == null) {
+ return Optional.empty();
+ }
+ if (latestLocation.equals(tableMetadata.metadataFileLocation())) {
+ return Optional.of(tableMetadata);
+ }
+
+ LOG.debug(
+ "The cached table metadata is not latest, table identifier: {}, "
+ + "table metadata location in cache: {}, latest metadata location:
{}",
+ tableIdentifier,
+ tableMetadata.metadataFileLocation(),
+ latestLocation);
+ invalidate(tableIdentifier);
+ return Optional.empty();
+ }
Review Comment:
Do we need to consider the concurrency problem here?
##########
iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java:
##########
@@ -306,4 +336,30 @@ public IcebergTableChange(TableIdentifier tableIdentifier,
Transaction transacti
this.transaction = transaction;
}
}
+
+ private TableMetadataCache loadTableMetadataCache(IcebergConfig config,
Catalog catalog) {
+ String impl = config.get(IcebergConfig.TABLE_METADATA_CACHE_IMPL);
+ if (StringUtils.isBlank(impl)) {
+ return TableMetadataCache.DUMMY;
+ }
+
+ Preconditions.checkArgument(
+ catalog instanceof SupportsMetadataLocation,
+ "You shouldn't enable Iceberg metadata cache for the catalog %s,
because the catalog impl does not support get metadata location.",
+ catalog.name());
+
+ TableMetadataCache cache =
+ ClassUtils.loadAndGetInstance(impl,
Thread.currentThread().getContextClassLoader());
+ int capacity = config.get(IcebergConfig.TABLE_METADATA_CACHE_CAPACITY);
+ int expireMinutes =
config.get(IcebergConfig.TABLE_METADATA_CACHE_EXPIRE_MINUTES);
+ cache.initialize(
+ capacity, expireMinutes, config.getAllConfig(),
(SupportsMetadataLocation) catalog);
+ LOG.info(
+ "Load Iceberg table metadata cache for catalog: {}, impl:{}, capacity:
{}, expire minutes: {}",
Review Comment:
You mean this cache is per catalog?
--
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]
