diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCache.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCache.java
index 8cb3a47..ed26f7c 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCache.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCache.java
@@ -24,14 +24,7 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.asterix.common.api.IApplicationContext;
-import org.apache.asterix.common.exceptions.AsterixException;
-import org.apache.asterix.common.exceptions.CompilationException;
-import org.apache.asterix.common.exceptions.ErrorCode;
 import org.apache.asterix.common.external.IExternalCredentialsCache;
-import org.apache.asterix.common.metadata.DatasetFullyQualifiedName;
-import org.apache.asterix.common.metadata.DataverseName;
-import org.apache.asterix.common.metadata.IFullyQualifiedName;
-import org.apache.asterix.common.metadata.MetadataConstants;
 import org.apache.asterix.external.util.ExternalDataConstants;
 import org.apache.asterix.external.util.aws.s3.S3Constants;
 import org.apache.commons.lang3.tuple.Pair;
@@ -46,41 +39,35 @@
     private static final Logger LOGGER = LogManager.getLogger();
     private final ConcurrentMap<String, Pair<Span, Object>> cache = new ConcurrentHashMap<>();
     private final int awsAssumeRoleDuration;
-    private final double refreshAwsAssumeRoleThreshold;
+    private final int refreshAwsAssumeRoleThresholdPercentage;
 
     public ExternalCredentialsCache(IApplicationContext appCtx) {
         this.awsAssumeRoleDuration = appCtx.getExternalProperties().getAwsAssumeRoleDuration();
-        this.refreshAwsAssumeRoleThreshold = appCtx.getExternalProperties().getAwsRefreshAssumeRoleThreshold();
+        this.refreshAwsAssumeRoleThresholdPercentage =
+                appCtx.getExternalProperties().getAwsRefreshAssumeRoleThresholdPercentage();
     }
 
     @Override
-    public synchronized Object getCredentials(Map<String, String> configuration) throws CompilationException {
-        IFullyQualifiedName fqn = getFullyQualifiedNameFromConfiguration(configuration);
-        return getCredentials(fqn);
-    }
-
-    @Override
-    public synchronized Object getCredentials(IFullyQualifiedName fqn) {
-        String name = getName(fqn);
-        if (cache.containsKey(name) && !needsRefresh(cache.get(name).getLeft())) {
-            return cache.get(name).getRight();
+    public synchronized Object get(String key) {
+        invalidateCache();
+        if (cache.containsKey(key)) {
+            return cache.get(key).getRight();
         }
         return null;
     }
 
     @Override
-    public synchronized void updateCache(Map<String, String> configuration, Map<String, String> credentials)
-            throws CompilationException {
-        IFullyQualifiedName fqn = getFullyQualifiedNameFromConfiguration(configuration);
-        String type = configuration.get(ExternalDataConstants.KEY_EXTERNAL_SOURCE_TYPE);
-        updateCache(fqn, type, credentials);
+    public void delete(String key) {
+        Object removed = cache.remove(key);
+        if (removed != null) {
+            LOGGER.info("Removed cached credentials for {} because it got deleted", key);
+        }
     }
 
     @Override
-    public synchronized void updateCache(IFullyQualifiedName fqn, String type, Map<String, String> credentials) {
-        String name = getName(fqn);
+    public synchronized void put(String key, String type, Map<String, String> credentials) {
         if (ExternalDataConstants.KEY_ADAPTER_NAME_AWS_S3.equalsIgnoreCase(type)) {
-            updateAwsCache(name, credentials);
+            updateAwsCache(key, credentials);
         }
     }
 
@@ -88,34 +75,23 @@
         String accessKeyId = credentials.get(S3Constants.ACCESS_KEY_ID_FIELD_NAME);
         String secretAccessKey = credentials.get(S3Constants.SECRET_ACCESS_KEY_FIELD_NAME);
         String sessionToken = credentials.get(S3Constants.SESSION_TOKEN_FIELD_NAME);
-        doUpdateAwsCache(name, AwsSessionCredentials.create(accessKeyId, secretAccessKey, sessionToken));
-    }
 
-    private void doUpdateAwsCache(String name, AwsSessionCredentials credentials) {
-        cache.put(name, Pair.of(Span.start(awsAssumeRoleDuration, TimeUnit.SECONDS), credentials));
+        AwsSessionCredentials sessionCreds = AwsSessionCredentials.create(accessKeyId, secretAccessKey, sessionToken);
+        cache.put(name, Pair.of(Span.start(awsAssumeRoleDuration, TimeUnit.SECONDS), sessionCreds));
         LOGGER.info("Received and cached new credentials for {}", name);
     }
 
-    @Override
-    public void deleteCredentials(IFullyQualifiedName fqn) {
-        String name = getName(fqn);
-        Object removed = cache.remove(name);
-        if (removed != null) {
-            LOGGER.info("Removed cached credentials for {}", name);
-        } else {
-            LOGGER.info("No cached credentials found for {}, nothing to remove", name);
-        }
-    }
-
-    @Override
-    public String getName(Map<String, String> configuration) throws CompilationException {
-        IFullyQualifiedName fqn = getFullyQualifiedNameFromConfiguration(configuration);
-        return getName(fqn);
-    }
-
-    @Override
-    public String getName(IFullyQualifiedName fqn) {
-        return fqn.toString();
+    /**
+     * Iterates the cache and removes the credentials that are considered expired
+     */
+    private void invalidateCache() {
+        cache.entrySet().removeIf(entry -> {
+            boolean shouldRemove = needsRefresh(entry.getValue().getLeft());
+            if (shouldRemove) {
+                LOGGER.info("Removing cached credentials for {} because it expired", entry.getKey());
+            }
+            return shouldRemove;
+        });
     }
 
     /**
@@ -125,27 +101,9 @@
      * @return true if the remaining time is less than the configured refresh percentage, false otherwise
      */
     private boolean needsRefresh(Span span) {
-        return (double) span.remaining(TimeUnit.SECONDS)
-                / span.getSpan(TimeUnit.SECONDS) < refreshAwsAssumeRoleThreshold;
-    }
-
-    protected IFullyQualifiedName getFullyQualifiedNameFromConfiguration(Map<String, String> configuration)
-            throws CompilationException {
-        String database = configuration.get(ExternalDataConstants.KEY_DATASET_DATABASE);
-        if (database == null) {
-            database = MetadataConstants.DEFAULT_DATABASE;
-        }
-        String stringDataverse = configuration.get(ExternalDataConstants.KEY_DATASET_DATAVERSE);
-        DataverseName dataverse = getDataverseName(stringDataverse);
-        String dataset = configuration.get(ExternalDataConstants.KEY_DATASET);
-        return new DatasetFullyQualifiedName(database, dataverse, dataset);
-    }
-
-    protected DataverseName getDataverseName(String dataverse) throws CompilationException {
-        try {
-            return DataverseName.createSinglePartName(dataverse);
-        } catch (AsterixException ex) {
-            throw new CompilationException(ErrorCode.INVALID_DATABASE_OBJECT_NAME, dataverse);
-        }
+        double remaining = (double) span.remaining(TimeUnit.SECONDS) / span.getSpan(TimeUnit.SECONDS);
+        double passed = 1 - remaining;
+        int passedPercentage = (int) (passed * 100);
+        return passedPercentage > refreshAwsAssumeRoleThresholdPercentage;
     }
 }
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCacheUpdater.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCacheUpdater.java
index f07caaa..29e3239 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCacheUpdater.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/external/ExternalCredentialsCacheUpdater.java
@@ -40,6 +40,7 @@
 import org.apache.asterix.common.external.IExternalCredentialsCache;
 import org.apache.asterix.common.external.IExternalCredentialsCacheUpdater;
 import org.apache.asterix.common.messaging.api.MessageFuture;
+import org.apache.asterix.external.util.ExternalDataConstants;
 import org.apache.asterix.external.util.aws.s3.S3AuthUtils;
 import org.apache.asterix.external.util.aws.s3.S3Constants;
 import org.apache.asterix.messaging.CCMessageBroker;
@@ -65,7 +66,8 @@
     public synchronized Object generateAndCacheCredentials(Map<String, String> configuration)
             throws HyracksDataException, CompilationException {
         IExternalCredentialsCache cache = appCtx.getExternalCredentialsCache();
-        Object credentials = cache.getCredentials(configuration);
+        String name = configuration.get(ExternalDataConstants.KEY_ENTITY_ID);
+        Object credentials = cache.get(name);
         if (credentials != null) {
             return credentials;
         }
@@ -74,9 +76,7 @@
          * if we are the CC, generate new creds and ask all NCs to update their cache
          * if we are the NC, send a message to the CC to generate new creds and ask all NCs to update their cache
          */
-        String name = cache.getName(configuration);
-        if (appCtx instanceof ICcApplicationContext) {
-            ICcApplicationContext ccAppCtx = (ICcApplicationContext) appCtx;
+        if (appCtx instanceof ICcApplicationContext ccAppCtx) {
             IClusterManagementWork.ClusterState state = ccAppCtx.getClusterStateManager().getState();
             if (!(state == ACTIVE || state == REBALANCE_REQUIRED)) {
                 throw new RuntimeDataException(REJECT_BAD_CLUSTER_STATE, state);
@@ -106,7 +106,8 @@
 
             // request all NCs to update their credentials cache with the latest creds
             updateNcsCredentialsCache(ccAppCtx, name, credentialsMap, configuration);
-            cache.updateCache(configuration, credentialsMap);
+            String type = configuration.get(ExternalDataConstants.KEY_EXTERNAL_SOURCE_TYPE);
+            cache.put(name, type, credentialsMap);
             credentials = AwsSessionCredentials.create(accessKeyId, secretAccessKey, sessionToken);
         } else {
             NCMessageBroker broker = (NCMessageBroker) appCtx.getServiceContext().getMessageBroker();
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/UpdateAwsCredentialsCacheRequest.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/UpdateAwsCredentialsCacheRequest.java
index d1a9ebf..438e425 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/UpdateAwsCredentialsCacheRequest.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/UpdateAwsCredentialsCacheRequest.java
@@ -21,9 +21,8 @@
 import java.util.Map;
 
 import org.apache.asterix.common.api.INcApplicationContext;
-import org.apache.asterix.common.exceptions.CompilationException;
 import org.apache.asterix.common.messaging.api.INcAddressedMessage;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.asterix.external.util.ExternalDataConstants;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -40,13 +39,10 @@
     }
 
     @Override
-    public void handle(INcApplicationContext appCtx) throws HyracksDataException {
-        try {
-            appCtx.getExternalCredentialsCache().updateCache(configuration, credentials);
-        } catch (CompilationException ex) {
-            LOGGER.info("Failed to process request", ex);
-            throw HyracksDataException.create(ex);
-        }
+    public void handle(INcApplicationContext appCtx) {
+        String name = configuration.get(ExternalDataConstants.KEY_ENTITY_ID);
+        String type = configuration.get(ExternalDataConstants.KEY_EXTERNAL_SOURCE_TYPE);
+        appCtx.getExternalCredentialsCache().put(name, type, credentials);
     }
 
     @Override
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index 51371da..97bc3cd 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -274,6 +274,7 @@
 import org.apache.hyracks.api.job.profiling.IOperatorStats;
 import org.apache.hyracks.api.result.IResultSet;
 import org.apache.hyracks.api.result.ResultSetId;
+import org.apache.hyracks.api.util.ExceptionUtils;
 import org.apache.hyracks.control.cc.ClusterControllerService;
 import org.apache.hyracks.control.common.controllers.CCConfig;
 import org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorDescriptor.DropOption;
@@ -578,11 +579,17 @@
             // async queries are completed after their job completes
             if (ResultDelivery.ASYNC != resultDelivery) {
                 appCtx.getRequestTracker().complete(requestParameters.getRequestReference().getUuid());
+                postRequestCompleteCleanup(requestParameters);
             }
             Thread.currentThread().setName(threadName);
         }
     }
 
+    protected void postRequestCompleteCleanup(IRequestParameters requestParameters) {
+        String uuid = requestParameters.getRequestReference().getUuid();
+        appCtx.getExternalCredentialsCache().delete(uuid);
+    }
+
     protected void configureMetadataProvider(MetadataProvider metadataProvider, Map<String, String> config,
             Counter resultSetIdCounter, FileSplit outputFile, IRequestParameters requestParameters,
             Statement statement) {
@@ -1031,8 +1038,12 @@
                     ExternalDataUtils.normalize(properties);
                     ExternalDataUtils.validate(properties);
                     ExternalDataUtils.validateType(properties, (ARecordType) itemType);
-                    validateExternalDatasetProperties(externalDetails, properties, dd.getSourceLocation(), mdTxnCtx,
-                            appCtx, metadataProvider);
+                    Map<String, String> propertiesCopy = preparePropertiesCopyForValidation(externalDetails, properties,
+                            dd.getSourceLocation(), mdTxnCtx, appCtx, metadataProvider);
+                    // do any necessary validation on the copy to avoid changing the original and storing it in the metadata
+                    metadataProvider.setExternalEntityIdFromParts(propertiesCopy, databaseName, dataverseName,
+                            datasetName, false);
+                    validateAdapterSpecificProperties(propertiesCopy, dd.getSourceLocation(), appCtx);
                     datasetDetails = new ExternalDatasetDetails(externalDetails.getAdapter(), properties, new Date(),
                             TransactionState.COMMIT);
                     break;
@@ -2448,9 +2459,7 @@
                     sourceLoc, EnumSet.of(DropOption.IF_EXISTS), requestParameters.isForceDropDataset());
 
             MetadataManager.INSTANCE.commitTransaction(mdTxnCtx.getValue());
-            if (ds.getDatasetType().equals(DatasetType.EXTERNAL)) {
-                appCtx.getExternalCredentialsCache().deleteCredentials(ds.getDatasetFullyQualifiedName());
-            }
+            deleteDatasetCachedCredentials(ds);
             return true;
         } catch (Exception e) {
             LOGGER.error("failed to drop dataset; executing compensating operations", e);
@@ -2498,6 +2507,10 @@
         }
     }
 
+    protected void deleteDatasetCachedCredentials(Dataset dataset) throws CompilationException {
+        appCtx.getExternalCredentialsCache().delete(dataset.getDatasetFullyQualifiedName().toString());
+    }
+
     protected void handleIndexDropStatement(MetadataProvider metadataProvider, Statement stmt,
             IHyracksClientConnection hcc, IRequestParameters requestParameters) throws Exception {
         IndexDropStatement stmtIndexDrop = (IndexDropStatement) stmt;
@@ -3637,7 +3650,7 @@
     protected CreateResult doCreateLibrary(MetadataProvider metadataProvider, String databaseName,
             DataverseName dataverseName, String libraryName, String libraryHash, CreateLibraryStatement cls,
             IHyracksClientConnection hcc, IRequestParameters requestParameters) throws Exception {
-        JobUtils.ProgressState progress = ProgressState.NO_PROGRESS;
+        ProgressState progress = ProgressState.NO_PROGRESS;
         boolean prepareJobSuccessful = false;
         JobSpecification abortJobSpec = null;
         Library existingLibrary = null;
@@ -3773,7 +3786,7 @@
     protected boolean doDropLibrary(MetadataProvider metadataProvider, LibraryDropStatement stmtDropLibrary,
             String databaseName, DataverseName dataverseName, String libraryName, IHyracksClientConnection hcc,
             IRequestParameters requestParameters) throws Exception {
-        JobUtils.ProgressState progress = ProgressState.NO_PROGRESS;
+        ProgressState progress = ProgressState.NO_PROGRESS;
         MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
         boolean bActiveTxn = true;
         metadataProvider.setMetadataTxnContext(mdTxnCtx);
@@ -4044,8 +4057,11 @@
                     copyStmt, itemType, mdTxnCtx, metadataProvider);
             ExternalDataUtils.normalize(properties);
             ExternalDataUtils.validate(properties);
-            validateExternalDatasetProperties(externalDetails, properties, copyStmt.getSourceLocation(), mdTxnCtx,
-                    appCtx, metadataProvider);
+            Map<String, String> propertiesCopy = preparePropertiesCopyForValidation(externalDetails, properties,
+                    copyStmt.getSourceLocation(), mdTxnCtx, appCtx, metadataProvider);
+            // do any necessary validation on the copy to avoid changing the original and storing it in the metadata
+            metadataProvider.setExternalEntityId(propertiesCopy, dataset);
+            validateAdapterSpecificProperties(propertiesCopy, copyStmt.getSourceLocation(), appCtx);
             CompiledCopyFromFileStatement cls = new CompiledCopyFromFileStatement(databaseName, dataverseName,
                     copyStmt.getDatasetName(), itemType, externalDetails.getAdapter(), properties);
             cls.setSourceLocation(stmt.getSourceLocation());
@@ -4102,6 +4118,9 @@
             ResultMetadata outMetadata, IRequestParameters requestParameters, Map<String, IAObject> stmtParams,
             Stats stats) throws Exception {
         CopyToStatement copyTo = (CopyToStatement) stmt;
+        Namespace namespace = copyTo.getNamespace();
+        DataverseName dataverseName = namespace.getDataverseName();
+        String databaseName = namespace.getDatabaseName();
         final IRequestTracker requestTracker = appCtx.getRequestTracker();
         final ClientRequest clientRequest =
                 (ClientRequest) requestTracker.get(requestParameters.getRequestReference().getUuid());
@@ -4130,16 +4149,21 @@
             metadataProvider.setMetadataTxnContext(mdTxnCtx);
             try {
                 ExternalDetailsDecl edd = copyTo.getExternalDetailsDecl();
-                edd.setProperties(createAndValidateAdapterConfigurationForCopyToStmt(edd,
+                Map<String, String> properties = createAndValidateAdapterConfigurationForCopyToStmt(edd,
                         ExternalDataConstants.WRITER_SUPPORTED_ADAPTERS, copyTo.getSourceLocation(), mdTxnCtx,
-                        metadataProvider));
+                        metadataProvider);
+
+                // request id is used to cache credentials if needed, and clear them after request is done
+                String uuid = requestParameters.getRequestReference().getUuid();
+                metadataProvider.setExternalEntityIdFromParts(properties, databaseName, dataverseName, uuid, true);
+                edd.setProperties(properties);
 
                 if (ExternalDataConstants.FORMAT_PARQUET
                         .equalsIgnoreCase(edd.getProperties().get(ExternalDataConstants.KEY_FORMAT))) {
                     if (copyTo.getType() != null) {
-                        DataverseName dataverseName =
+                        DataverseName dummyDataverse =
                                 DataverseName.createFromCanonicalForm(ExternalDataConstants.DUMMY_DATAVERSE_NAME);
-                        IAType iaType = translateType(ExternalDataConstants.DUMMY_DATABASE_NAME, dataverseName,
+                        IAType iaType = translateType(ExternalDataConstants.DUMMY_DATABASE_NAME, dummyDataverse,
                                 ExternalDataConstants.DUMMY_TYPE_NAME, copyTo.getType(), mdTxnCtx);
                         edd.getProperties().put(ExternalDataConstants.PARQUET_SCHEMA_KEY,
                                 SchemaConverterVisitor.convertToParquetSchemaString((ARecordType) iaType));
@@ -4148,14 +4172,14 @@
 
                 if (edd.getProperties().get(ExternalDataConstants.KEY_FORMAT)
                         .equalsIgnoreCase(ExternalDataConstants.FORMAT_CSV_LOWER_CASE)) {
-                    DataverseName dataverseName =
+                    DataverseName dummyDataverse =
                             DataverseName.createFromCanonicalForm(ExternalDataConstants.DUMMY_DATAVERSE_NAME);
                     IAType iaType;
                     if (copyTo.getType() != null) {
-                        iaType = translateType(ExternalDataConstants.DUMMY_DATABASE_NAME, dataverseName,
+                        iaType = translateType(ExternalDataConstants.DUMMY_DATABASE_NAME, dummyDataverse,
                                 ExternalDataConstants.DUMMY_TYPE_NAME, copyTo.getType(), mdTxnCtx);
                     } else if (copyTo.getTypeExpressionItemType() != null) {
-                        iaType = translateType(ExternalDataConstants.DUMMY_DATABASE_NAME, dataverseName,
+                        iaType = translateType(ExternalDataConstants.DUMMY_DATABASE_NAME, dummyDataverse,
                                 ExternalDataConstants.DUMMY_TYPE_NAME, copyTo.getTypeExpressionItemType(), mdTxnCtx);
                     } else {
                         throw new CompilationException(ErrorCode.COMPILATION_ERROR,
@@ -5552,7 +5576,7 @@
             if (atomic && jobId != null) {
                 globalTxManager.abortTransaction(jobId);
             }
-            if (org.apache.hyracks.api.util.ExceptionUtils.getRootCause(e) instanceof InterruptedException) {
+            if (ExceptionUtils.getRootCause(e) instanceof InterruptedException) {
                 Thread.currentThread().interrupt();
                 throw new RuntimeDataException(ErrorCode.REQUEST_CANCELLED, clientRequest.getId());
             }
@@ -5561,6 +5585,7 @@
             // complete async jobs after their job completes
             if (ResultDelivery.ASYNC == resultDelivery) {
                 requestTracker.complete(clientRequest.getId());
+                postRequestCompleteCleanup(requestParameters);
             }
             locker.unlock();
         }
@@ -5777,33 +5802,39 @@
      * @param details external details
      * @param sourceLoc source location
      */
-    private void normalizeAdapters(ExternalDetailsDecl details, SourceLocation sourceLoc) throws CompilationException {
+    protected void normalizeAdapters(ExternalDetailsDecl details, SourceLocation sourceLoc)
+            throws CompilationException {
         String adapter = details.getAdapter();
-        Optional<String> normalizedAdapter = ExternalDataConstants.EXTERNAL_READ_ADAPTERS.stream()
-                .filter(k -> k.equalsIgnoreCase(adapter)).findFirst();
+        Optional<String> normalizedAdapter =
+                getSupportedAdapters().stream().filter(k -> k.equalsIgnoreCase(adapter)).findFirst();
         if (normalizedAdapter.isEmpty()) {
             throw CompilationException.create(ErrorCode.UNKNOWN_ADAPTER, sourceLoc, adapter);
         }
         details.setAdapter(normalizedAdapter.get());
     }
 
-    protected void validateExternalDatasetProperties(ExternalDetailsDecl externalDetails,
+    protected Set<String> getSupportedAdapters() {
+        return ExternalDataConstants.EXTERNAL_READ_ADAPTERS;
+    }
+
+    protected Map<String, String> preparePropertiesCopyForValidation(ExternalDetailsDecl externalDetails,
             Map<String, String> properties, SourceLocation srcLoc, MetadataTransactionContext mdTxnCtx,
             IApplicationContext appCtx, MetadataProvider metadataProvider)
             throws AlgebricksException, HyracksDataException {
-        // Validate adapter specific properties
         normalizeAdapters(externalDetails, srcLoc);
         String adapter = externalDetails.getAdapter();
         Map<String, String> details = new HashMap<>(properties);
         details.put(ExternalDataConstants.KEY_EXTERNAL_SOURCE_TYPE, adapter);
-        validateAdapterSpecificProperties(details, srcLoc, appCtx);
+        return details;
     }
 
     protected Map<String, String> createAndValidateAdapterConfigurationForCopyToStmt(
             ExternalDetailsDecl externalDetailsDecl, Set<String> supportedAdapters, SourceLocation sourceLocation,
             MetadataTransactionContext mdTxnCtx, MetadataProvider md) throws AlgebricksException {
+        normalizeAdapters(externalDetailsDecl, sourceLocation);
         String adapterName = externalDetailsDecl.getAdapter();
         Map<String, String> properties = externalDetailsDecl.getProperties();
+        properties.put(ExternalDataConstants.KEY_EXTERNAL_SOURCE_TYPE, adapterName);
         WriterValidationUtil.validateWriterConfiguration(adapterName, supportedAdapters, properties, sourceLocation);
         return properties;
     }
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1/cluster_state_1.1.regexadm b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1/cluster_state_1.1.regexadm
index 29a4a6e..3ffb080 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1/cluster_state_1.1.regexadm
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1/cluster_state_1.1.regexadm
@@ -9,7 +9,7 @@
     "active\.stop\.timeout" : 3600,
     "active\.suspend\.timeout" : 3600,
     "aws.assume.role.duration" : 900,
-    "aws.refresh.assume.role.threshold" : 0.5,
+    "aws.refresh.assume.role.threshold.percentage" : 75,
     "azure.request.timeout" : 120,
     "cloud.acquire.token.timeout" : 100,
     "cloud.deployment" : false,
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_full/cluster_state_1_full.1.regexadm b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_full/cluster_state_1_full.1.regexadm
index dbbf83f..aeb3cea 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_full/cluster_state_1_full.1.regexadm
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_full/cluster_state_1_full.1.regexadm
@@ -9,7 +9,7 @@
     "active\.stop\.timeout" : 3600,
     "active\.suspend\.timeout" : 3600,
     "aws.assume.role.duration" : 900,
-    "aws.refresh.assume.role.threshold" : 0.5,
+    "aws.refresh.assume.role.threshold.percentage" : 75,
     "azure.request.timeout" : 120,
     "cloud.acquire.token.timeout" : 100,
     "cloud.deployment" : false,
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_less/cluster_state_1_less.1.regexadm b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_less/cluster_state_1_less.1.regexadm
index 4778cb8..b475612 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_less/cluster_state_1_less.1.regexadm
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/cluster_state_1_less/cluster_state_1_less.1.regexadm
@@ -9,7 +9,7 @@
     "active\.stop\.timeout" : 3600,
     "active\.suspend\.timeout" : 3600,
     "aws.assume.role.duration" : 900,
-    "aws.refresh.assume.role.threshold" : 0.5,
+    "aws.refresh.assume.role.threshold.percentage" : 75,
     "azure.request.timeout" : 120,
     "cloud.acquire.token.timeout" : 100,
     "cloud.deployment" : false,
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ExternalProperties.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ExternalProperties.java
index f6559f0..62437b2 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ExternalProperties.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ExternalProperties.java
@@ -18,12 +18,12 @@
  */
 package org.apache.asterix.common.config;
 
-import static org.apache.hyracks.control.common.config.OptionTypes.DOUBLE;
 import static org.apache.hyracks.control.common.config.OptionTypes.LEVEL;
 import static org.apache.hyracks.control.common.config.OptionTypes.NONNEGATIVE_INTEGER;
 import static org.apache.hyracks.control.common.config.OptionTypes.POSITIVE_INTEGER;
 import static org.apache.hyracks.control.common.config.OptionTypes.POSITIVE_INTEGER_BYTE_UNIT;
 import static org.apache.hyracks.control.common.config.OptionTypes.STRING;
+import static org.apache.hyracks.control.common.config.OptionTypes.getRangedIntegerType;
 
 import org.apache.hyracks.api.config.IOption;
 import org.apache.hyracks.api.config.IOptionType;
@@ -55,14 +55,16 @@
         LIBRARY_DEPLOY_TIMEOUT(POSITIVE_INTEGER, 1800, "Timeout to upload a UDF in seconds"),
         AZURE_REQUEST_TIMEOUT(POSITIVE_INTEGER, 120, "Timeout for Azure client requests in seconds"),
         AWS_ASSUME_ROLE_DURATION(
-                POSITIVE_INTEGER,
+                getRangedIntegerType(900, 43200),
                 900,
                 "AWS assuming role duration in seconds. "
                         + "Range from 900 seconds (15 mins) to 43200 seconds (12 hours)"),
-        AWS_REFRESH_ASSUME_ROLE_THRESHOLD(
-                DOUBLE,
-                .5,
-                "Percentage of left duration before assume role credentials " + "needs to be refreshed");
+        AWS_REFRESH_ASSUME_ROLE_THRESHOLD_PERCENTAGE(
+                getRangedIntegerType(25, 90),
+                75,
+                "Percentage of duration passed before assume role credentials need to be refreshed, the value ranges "
+                        + "from 25 to 90, default is 75. For example, if the value is set to 65, this means the "
+                        + "credentials need to be refreshed if 65% of the total expiration duration is already passed");
 
         private final IOptionType type;
         private final Object defaultValue;
@@ -91,7 +93,7 @@
                 case LIBRARY_DEPLOY_TIMEOUT:
                 case AZURE_REQUEST_TIMEOUT:
                 case AWS_ASSUME_ROLE_DURATION:
-                case AWS_REFRESH_ASSUME_ROLE_THRESHOLD:
+                case AWS_REFRESH_ASSUME_ROLE_THRESHOLD_PERCENTAGE:
                     return Section.COMMON;
                 case CC_JAVA_OPTS:
                 case NC_JAVA_OPTS:
@@ -177,7 +179,7 @@
         return accessor.getInt(Option.AWS_ASSUME_ROLE_DURATION);
     }
 
-    public double getAwsRefreshAssumeRoleThreshold() {
-        return accessor.getDouble(Option.AWS_REFRESH_ASSUME_ROLE_THRESHOLD);
+    public int getAwsRefreshAssumeRoleThresholdPercentage() {
+        return accessor.getInt(Option.AWS_REFRESH_ASSUME_ROLE_THRESHOLD_PERCENTAGE);
     }
 }
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalCredentialsCache.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalCredentialsCache.java
index 3ff444d..3a9ae1c 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalCredentialsCache.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/external/IExternalCredentialsCache.java
@@ -20,64 +20,29 @@
 
 import java.util.Map;
 
-import org.apache.asterix.common.exceptions.CompilationException;
-import org.apache.asterix.common.metadata.IFullyQualifiedName;
-
 public interface IExternalCredentialsCache {
 
     /**
-     * Returns the cached credentials. Can be of any supported external credentials type
+     * Returns the cached credentials.
      *
-     * @param configuration configuration containing external collection details
-     * @return credentials if present, null otherwise
+     * @param key credentials key
+     * @return credentials if present and not expired/need refreshing, null otherwise
      */
-    Object getCredentials(Map<String, String> configuration) throws CompilationException;
+    Object get(String key);
 
     /**
-     * Returns the cached credentials. Can be of any supported external credentials type
+     * Deletes the cache for the provided entity
      *
-     * @param fqn fully qualified name of the credentials entity
-     * @return credentials if present, null otherwise
+     * @param key credentials key
      */
-    Object getCredentials(IFullyQualifiedName fqn) throws CompilationException;
+    void delete(String key);
 
     /**
      * Updates the credentials cache with the provided credentials for the specified name
      *
-     * @param configuration configuration containing external collection details
+     * @param key credentials key
+     * @param type credentials type
      * @param credentials credentials to cache
      */
-    void updateCache(Map<String, String> configuration, Map<String, String> credentials) throws CompilationException;
-
-    /**
-     * Updates the credentials cache with the provided credentials for the specified name
-     *
-     * @param fqn fully qualified name for the credentials entity
-     * @param type type of the entity
-     * @param credentials credentials to cache
-     */
-    void updateCache(IFullyQualifiedName fqn, String type, Map<String, String> credentials);
-
-    /**
-     * Deletes the cache for the provided enitty
-     *
-     * @param fqn fully qualified name of entity for which the credentials are to be deleted
-     */
-    void deleteCredentials(IFullyQualifiedName fqn);
-
-    /**
-     * Returns the name of the entity which the cached credentials belong to
-     *
-     * @param configuration configuration containing external collection details
-     * @return name of entity which credentials belong to
-     */
-    String getName(Map<String, String> configuration) throws CompilationException;
-
-    /**
-     * Returns the name of the entity which the cached credentials belong to
-     *
-     * @param fqn fully qualified name for the credentials entity
-     * @return name of entity which credentials belong to
-     */
-    String getName(IFullyQualifiedName fqn) throws CompilationException;
+    void put(String key, String type, Map<String, String> credentials);
 }
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index 4cc1656..9e9df8b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -105,6 +105,7 @@
     public static final String KEY_WAIT_FOR_DATA = "wait-for-data";
     public static final String KEY_FEED_NAME = "feed";
     // a string representing external bucket name
+    public static final String KEY_ENTITY_ID = "entity-id";
     public static final String KEY_EXTERNAL_SOURCE_TYPE = "type";
     // a comma delimited list of nodes
     public static final String KEY_NODES = "nodes";
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3AuthUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3AuthUtils.java
index 57889ea..4e0a435 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3AuthUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/aws/s3/S3AuthUtils.java
@@ -241,7 +241,7 @@
     public static AwsCredentialsProvider getTrustAccountCredentials(IApplicationContext appCtx,
             Map<String, String> configuration) throws CompilationException {
         IExternalCredentialsCache cache = appCtx.getExternalCredentialsCache();
-        Object credentialsObject = cache.getCredentials(configuration);
+        Object credentialsObject = cache.get(configuration.get(ExternalDataConstants.KEY_ENTITY_ID));
         if (credentialsObject != null) {
             return () -> (AwsSessionCredentials) credentialsObject;
         }
@@ -389,7 +389,7 @@
      */
     public static void configureAwsS3HdfsJobConf(IApplicationContext appCtx, JobConf jobConf,
             Map<String, String> configuration, int numberOfPartitions) throws CompilationException {
-        setHadoopCredentials(jobConf, configuration);
+        setHadoopCredentials(appCtx, jobConf, configuration);
         String serviceEndpoint = configuration.get(SERVICE_END_POINT_FIELD_NAME);
         Region region = validateAndGetRegion(configuration.get(REGION_FIELD_NAME));
         jobConf.set(HADOOP_REGION, region.toString());
@@ -421,7 +421,8 @@
      * @param jobConf hadoop job config
      * @param configuration external details configuration
      */
-    private static void setHadoopCredentials(JobConf jobConf, Map<String, String> configuration) {
+    private static void setHadoopCredentials(IApplicationContext appCtx, JobConf jobConf,
+            Map<String, String> configuration) {
         AuthenticationType authenticationType = getAuthenticationType(configuration);
         switch (authenticationType) {
             case ANONYMOUS:
@@ -432,7 +433,11 @@
                 jobConf.set(HADOOP_ASSUME_ROLE_ARN, configuration.get(ROLE_ARN_FIELD_NAME));
                 jobConf.set(HADOOP_ASSUME_ROLE_EXTERNAL_ID, configuration.get(EXTERNAL_ID_FIELD_NAME));
                 jobConf.set(HADOOP_ASSUME_ROLE_SESSION_NAME, "parquet-" + UUID.randomUUID());
-                jobConf.set(HADOOP_ASSUME_ROLE_SESSION_DURATION, "15m");
+
+                // hadoop accepts time 15m to 1h, we will base it on the provided configuration
+                int durationInSeconds = appCtx.getExternalProperties().getAwsAssumeRoleDuration();
+                String hadoopDuration = getHadoopDuration(durationInSeconds);
+                jobConf.set(HADOOP_ASSUME_ROLE_SESSION_DURATION, hadoopDuration);
 
                 // TODO: this assumes basic keys always, also support if we use InstanceProfile to assume a role
                 jobConf.set(HADOOP_CREDENTIALS_TO_ASSUME_ROLE_KEY, HADOOP_SIMPLE);
@@ -456,6 +461,36 @@
     }
 
     /**
+     * Hadoop accepts duration values from 15m to 1h (in this format). We will base this on the configured
+     * duration in seconds. If the time exceeds 1 hour, we will return 1h
+     *
+     * @param seconds configured duration in seconds
+     * @return hadoop updated duration
+     */
+    private static String getHadoopDuration(int seconds) {
+        // constants for time thresholds
+        final int FIFTEEN_MINUTES_IN_SECONDS = 15 * 60;
+        final int ONE_HOUR_IN_SECONDS = 60 * 60;
+
+        // Adjust seconds to fit within bounds
+        if (seconds < FIFTEEN_MINUTES_IN_SECONDS) {
+            seconds = FIFTEEN_MINUTES_IN_SECONDS;
+        } else if (seconds > ONE_HOUR_IN_SECONDS) {
+            seconds = ONE_HOUR_IN_SECONDS;
+        }
+
+        // Convert seconds to minutes
+        int minutes = seconds / 60;
+
+        // Format the result
+        if (minutes == 60) {
+            return "1h";
+        } else {
+            return minutes + "m";
+        }
+    }
+
+    /**
      * Validate external dataset properties
      *
      * @param configuration properties
@@ -471,27 +506,6 @@
             throw new CompilationException(ErrorCode.PARAMETERS_REQUIRED, srcLoc, ExternalDataConstants.KEY_FORMAT);
         }
 
-        String arnRole = configuration.get(ROLE_ARN_FIELD_NAME);
-        String externalId = configuration.get(EXTERNAL_ID_FIELD_NAME);
-        String accessKeyId = configuration.get(ACCESS_KEY_ID_FIELD_NAME);
-        String secretAccessKey = configuration.get(SECRET_ACCESS_KEY_FIELD_NAME);
-
-        if (arnRole != null) {
-            return;
-        } else if (externalId != null) {
-            throw new CompilationException(REQUIRED_PARAM_IF_PARAM_IS_PRESENT, ROLE_ARN_FIELD_NAME,
-                    EXTERNAL_ID_FIELD_NAME);
-        } else if (accessKeyId == null || secretAccessKey == null) {
-            // If one is passed, the other is required
-            if (accessKeyId != null) {
-                throw new CompilationException(REQUIRED_PARAM_IF_PARAM_IS_PRESENT, SECRET_ACCESS_KEY_FIELD_NAME,
-                        ACCESS_KEY_ID_FIELD_NAME);
-            } else if (secretAccessKey != null) {
-                throw new CompilationException(REQUIRED_PARAM_IF_PARAM_IS_PRESENT, ACCESS_KEY_ID_FIELD_NAME,
-                        SECRET_ACCESS_KEY_FIELD_NAME);
-            }
-        }
-
         validateIncludeExclude(configuration);
         try {
             // TODO(htowaileb): maybe something better, this will check to ensure type is supported before creation
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
index ef056c9..6f3a9b6 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
@@ -48,6 +48,7 @@
 import org.apache.asterix.common.external.IDataSourceAdapter;
 import org.apache.asterix.common.external.IExternalFilterEvaluatorFactory;
 import org.apache.asterix.common.functions.FunctionSignature;
+import org.apache.asterix.common.metadata.DatasetFullyQualifiedName;
 import org.apache.asterix.common.metadata.DataverseName;
 import org.apache.asterix.common.metadata.LockList;
 import org.apache.asterix.common.metadata.MetadataConstants;
@@ -998,6 +999,7 @@
             configuration.put(ExternalDataConstants.KEY_DATASET_DATABASE, dataset.getDatabaseName());
             configuration.put(ExternalDataConstants.KEY_DATASET_DATAVERSE,
                     dataset.getDataverseName().getCanonicalForm());
+            setExternalEntityId(configuration, dataset);
             return AdapterFactoryProvider.getAdapterFactory(getApplicationContext().getServiceContext(), adapterName,
                     configuration, itemType, null, warningCollector, filterEvaluatorFactory);
         } catch (Exception e) {
@@ -1005,6 +1007,20 @@
         }
     }
 
+    public void setExternalEntityId(Map<String, String> configuration, Dataset dataset) throws AlgebricksException {
+        configuration.put(ExternalDataConstants.KEY_ENTITY_ID, dataset.getDatasetFullyQualifiedName().toString());
+    }
+
+    public void setExternalEntityIdFromParts(Map<String, String> configuration, String database,
+            DataverseName dataverse, String dataset, boolean isUuid) throws AlgebricksException {
+        if (isUuid) {
+            configuration.put(ExternalDataConstants.KEY_ENTITY_ID, dataset);
+        } else {
+            DatasetFullyQualifiedName fqn = new DatasetFullyQualifiedName(database, dataverse, dataset);
+            configuration.put(ExternalDataConstants.KEY_ENTITY_ID, fqn.toString());
+        }
+    }
+
     public TxnId getTxnId() {
         return txnId;
     }