From 5f7aa006ca8d11c00997fbdd09cf90e52df78e5e Mon Sep 17 00:00:00 2001
From: jtian <86805957+rsyncv@users.noreply.github.com>
Date: Wed, 2 Feb 2022 16:56:43 -0800
Subject: [PATCH] Integrate caching for Iceberg table fetch

---
 .../polaris/PolarisConnectorTableService.java | 26 ++++++++++++++++---
 .../PolarisConnectorTableServiceTest.java     |  1 -
 2 files changed, 23 insertions(+), 4 deletions(-)

diff --git a/metacat-connector-polaris/src/main/java/com/netflix/metacat/connector/polaris/PolarisConnectorTableService.java b/metacat-connector-polaris/src/main/java/com/netflix/metacat/connector/polaris/PolarisConnectorTableService.java
index 0a5eb890c..80ed2cb1e 100644
--- a/metacat-connector-polaris/src/main/java/com/netflix/metacat/connector/polaris/PolarisConnectorTableService.java
+++ b/metacat-connector-polaris/src/main/java/com/netflix/metacat/connector/polaris/PolarisConnectorTableService.java
@@ -26,6 +26,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections.MapUtils;
 import org.apache.commons.lang.StringUtils;
+import org.springframework.cache.annotation.Cacheable;
 import org.springframework.dao.DataIntegrityViolationException;
 
 import javax.annotation.Nullable;
@@ -140,9 +141,8 @@ public TableInfo get(final ConnectorRequestContext requestContext, final Qualifi
                 .orElseThrow(() -> new TableNotFoundException(name));
             final TableInfo info = polarisTableMapper.toInfo(polarisTableEntity);
             final String tableLoc = HiveTableUtil.getIcebergTableMetadataLocation(info);
-            final IcebergTableWrapper icebergTable = icebergTableHandler.getIcebergTable(
-                name, tableLoc, requestContext.isIncludeMetadata());
-            return connectorConverter.fromIcebergTableToTableInfo(name, icebergTable, tableLoc, info);
+            return getIcebergTable(name, tableLoc, info,
+                requestContext.isIncludeMetadata(), connectorContext.getConfig().isIcebergCacheEnabled());
         } catch (TableNotFoundException | IllegalArgumentException exception) {
             log.error(String.format("Not found exception for polaris table %s", name), exception);
             throw exception;
@@ -320,4 +320,24 @@ public List<TableInfo> list(
             throw new ConnectorException(msg, exception);
         }
     }
+
+    /**
+     * Return the table metadata from cache if exists else make the iceberg call and refresh it.
+     * @param tableName             table name
+     * @param tableMetadataLocation table metadata location
+     * @param info                  table info stored in hive metastore
+     * @param includeInfoDetails    if true, will include more details like the manifest file content
+     * @param useCache              true, if table can be retrieved from cache
+     * @return TableInfo
+     */
+    @Cacheable(key = "'iceberg.table.' + #includeInfoDetails + '.' + #tableMetadataLocation", condition = "#useCache")
+    public TableInfo getIcebergTable(final QualifiedName tableName,
+                                     final String tableMetadataLocation,
+                                     final TableInfo info,
+                                     final boolean includeInfoDetails,
+                                     final boolean useCache) {
+        final IcebergTableWrapper icebergTable =
+            this.icebergTableHandler.getIcebergTable(tableName, tableMetadataLocation, includeInfoDetails);
+        return connectorConverter.fromIcebergTableToTableInfo(tableName, icebergTable, tableMetadataLocation, info);
+    }
 }
diff --git a/metacat-connector-polaris/src/test/java/com/netflix/metacat/connector/polaris/PolarisConnectorTableServiceTest.java b/metacat-connector-polaris/src/test/java/com/netflix/metacat/connector/polaris/PolarisConnectorTableServiceTest.java
index 7421620ba..cb2b4f4c4 100644
--- a/metacat-connector-polaris/src/test/java/com/netflix/metacat/connector/polaris/PolarisConnectorTableServiceTest.java
+++ b/metacat-connector-polaris/src/test/java/com/netflix/metacat/connector/polaris/PolarisConnectorTableServiceTest.java
@@ -339,4 +339,3 @@ public void testUpdateTableAccept() {
         Assert.assertEquals(tableResult1.getMetadata().get("metadata_location"), location1);
     }
 }
-