diff --git a/core/trino-main/src/main/java/io/trino/metadata/Metadata.java b/core/trino-main/src/main/java/io/trino/metadata/Metadata.java
index bcc3761c3cc0..bc376d67aa8e 100644
--- a/core/trino-main/src/main/java/io/trino/metadata/Metadata.java
+++ b/core/trino-main/src/main/java/io/trino/metadata/Metadata.java
@@ -87,8 +87,6 @@ public interface Metadata
 
     Optional<SystemTable> getSystemTable(Session session, QualifiedObjectName tableName);
 
-    Optional<TableHandle> getTableHandleForStatisticsCollection(Session session, QualifiedObjectName tableName, Map<String, Object> analyzeProperties);
-
     Optional<TableExecuteHandle> getTableHandleForExecute(
             Session session,
             TableHandle tableHandle,
@@ -273,7 +271,7 @@ Optional<TableExecuteHandle> getTableHandleForExecute(
     /**
      * Describe statistics that must be collected during a statistics collection
      */
-    TableStatisticsMetadata getStatisticsCollectionMetadata(Session session, String catalogName, ConnectorTableMetadata tableMetadata);
+    AnalyzeMetadata getStatisticsCollectionMetadata(Session session, TableHandle tableHandle, Map<String, Object> analyzeProperties);
 
     /**
      * Begin statistics collection
diff --git a/core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java b/core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java
index 43cb5cf3bd6d..9964f0fc0155 100644
--- a/core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java
+++ b/core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java
@@ -39,6 +39,7 @@
 import io.trino.spi.connector.CatalogSchemaTableName;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorAnalyzeMetadata;
 import io.trino.spi.connector.ConnectorCapabilities;
 import io.trino.spi.connector.ConnectorInsertTableHandle;
 import io.trino.spi.connector.ConnectorMaterializedViewDefinition;
@@ -286,28 +287,6 @@ public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName
         });
     }
 
-    @Override
-    public Optional<TableHandle> getTableHandleForStatisticsCollection(Session session, QualifiedObjectName table, Map<String, Object> analyzeProperties)
-    {
-        requireNonNull(table, "table is null");
-
-        Optional<CatalogMetadata> catalog = getOptionalCatalogMetadata(session, table.getCatalogName());
-        if (catalog.isPresent()) {
-            CatalogMetadata catalogMetadata = catalog.get();
-            CatalogName catalogName = catalogMetadata.getConnectorId(session, table);
-            ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogName);
-
-            ConnectorTableHandle tableHandle = metadata.getTableHandleForStatisticsCollection(session.toConnectorSession(catalogName), table.asSchemaTableName(), analyzeProperties);
-            if (tableHandle != null) {
-                return Optional.of(new TableHandle(
-                        catalogName,
-                        tableHandle,
-                        catalogMetadata.getTransactionHandleFor(catalogName)));
-            }
-        }
-        return Optional.empty();
-    }
-
     @Override
     public Optional<TableExecuteHandle> getTableHandleForExecute(Session session, TableHandle tableHandle, String procedure, Map<String, Object> executeProperties)
     {
@@ -791,12 +770,14 @@ public TableStatisticsMetadata getStatisticsCollectionMetadataForWrite(Session s
     }
 
     @Override
-    public TableStatisticsMetadata getStatisticsCollectionMetadata(Session session, String catalogName, ConnectorTableMetadata tableMetadata)
+    public AnalyzeMetadata getStatisticsCollectionMetadata(Session session, TableHandle tableHandle, Map<String, Object> analyzeProperties)
     {
+        CatalogName catalogName = tableHandle.getCatalogName();
         CatalogMetadata catalogMetadata = getCatalogMetadataForWrite(session, catalogName);
         ConnectorMetadata metadata = catalogMetadata.getMetadata(session);
-        CatalogName catalog = catalogMetadata.getCatalogName();
-        return metadata.getStatisticsCollectionMetadata(session.toConnectorSession(catalog), tableMetadata);
+
+        ConnectorAnalyzeMetadata analyze = metadata.getStatisticsCollectionMetadata(session.toConnectorSession(catalogName), tableHandle.getConnectorHandle(), analyzeProperties);
+        return new AnalyzeMetadata(analyze.getStatisticsMetadata(), new TableHandle(catalogName, analyze.getTableHandle(), tableHandle.getTransaction()));
     }
 
     @Override
diff --git a/core/trino-main/src/main/java/io/trino/sql/analyzer/Analysis.java b/core/trino-main/src/main/java/io/trino/sql/analyzer/Analysis.java
index 7a7490ab6a2d..18139ab44471 100644
--- a/core/trino-main/src/main/java/io/trino/sql/analyzer/Analysis.java
+++ b/core/trino-main/src/main/java/io/trino/sql/analyzer/Analysis.java
@@ -24,6 +24,7 @@
 import com.google.common.collect.Multiset;
 import com.google.common.collect.Streams;
 import io.trino.connector.CatalogName;
+import io.trino.metadata.AnalyzeMetadata;
 import io.trino.metadata.QualifiedObjectName;
 import io.trino.metadata.ResolvedFunction;
 import io.trino.metadata.TableExecuteHandle;
@@ -217,7 +218,7 @@ public class Analysis
     private Optional<Insert> insert = Optional.empty();
     private Optional<RefreshMaterializedViewAnalysis> refreshMaterializedView = Optional.empty();
     private Optional<QualifiedObjectName> delegatedRefreshMaterializedView = Optional.empty();
-    private Optional<TableHandle> analyzeTarget = Optional.empty();
+    private Optional<AnalyzeMetadata> analyzeMetadata = Optional.empty();
     private Optional<List<ColumnSchema>> updatedColumns = Optional.empty();
 
     private final QueryType queryType;
@@ -728,14 +729,14 @@ public ColumnHandle getColumn(Field field)
         return columns.get(field);
     }
 
-    public Optional<TableHandle> getAnalyzeTarget()
+    public Optional<AnalyzeMetadata> getAnalyzeMetadata()
     {
-        return analyzeTarget;
+        return analyzeMetadata;
     }
 
-    public void setAnalyzeTarget(TableHandle analyzeTarget)
+    public void setAnalyzeMetadata(AnalyzeMetadata analyzeMetadata)
     {
-        this.analyzeTarget = Optional.of(analyzeTarget);
+        this.analyzeMetadata = Optional.of(analyzeMetadata);
     }
 
     public void setCreate(Create create)
diff --git a/core/trino-main/src/main/java/io/trino/sql/analyzer/StatementAnalyzer.java b/core/trino-main/src/main/java/io/trino/sql/analyzer/StatementAnalyzer.java
index 3afa185153ef..e694d9bce386 100644
--- a/core/trino-main/src/main/java/io/trino/sql/analyzer/StatementAnalyzer.java
+++ b/core/trino-main/src/main/java/io/trino/sql/analyzer/StatementAnalyzer.java
@@ -784,11 +784,13 @@ protected Scope visitAnalyze(Analyze node, Optional<Scope> scope)
             analysis.setUpdateType("ANALYZE");
             analysis.setUpdateTarget(tableName, Optional.empty(), Optional.empty());
 
-            // verify the target table exists and it's not a view
             if (metadata.isView(session, tableName)) {
                 throw semanticException(NOT_SUPPORTED, node, "Analyzing views is not supported");
             }
 
+            TableHandle tableHandle = metadata.getTableHandle(session, tableName)
+                    .orElseThrow(() -> semanticException(TABLE_NOT_FOUND, node, "Table '%s' does not exist", tableName));
+
             validateProperties(node.getProperties(), scope);
             CatalogName catalogName = getRequiredCatalogHandle(metadata, session, node, tableName.getCatalogName());
 
@@ -800,8 +802,8 @@ protected Scope visitAnalyze(Analyze node, Optional<Scope> scope)
                     accessControl,
                     analysis.getParameters(),
                     true);
-            TableHandle tableHandle = metadata.getTableHandleForStatisticsCollection(session, tableName, analyzeProperties)
-                    .orElseThrow(() -> semanticException(TABLE_NOT_FOUND, node, "Table '%s' does not exist", tableName));
+
+            analysis.setAnalyzeMetadata(metadata.getStatisticsCollectionMetadata(session, tableHandle, analyzeProperties));
 
             // user must have read and insert permission in order to analyze stats of a table
             analysis.addTableColumnReferences(
@@ -817,7 +819,6 @@ protected Scope visitAnalyze(Analyze node, Optional<Scope> scope)
                 throw new AccessDeniedException(format("Cannot ANALYZE (missing insert privilege) table %s", tableName));
             }
 
-            analysis.setAnalyzeTarget(tableHandle);
             return createAndAssignScope(node, scope, Field.newUnqualified("rows", BIGINT));
         }
 
diff --git a/core/trino-main/src/main/java/io/trino/sql/planner/LogicalPlanner.java b/core/trino-main/src/main/java/io/trino/sql/planner/LogicalPlanner.java
index 8f13a84907c4..cdf7c463d851 100644
--- a/core/trino-main/src/main/java/io/trino/sql/planner/LogicalPlanner.java
+++ b/core/trino-main/src/main/java/io/trino/sql/planner/LogicalPlanner.java
@@ -25,6 +25,7 @@
 import io.trino.cost.StatsCalculator;
 import io.trino.cost.StatsProvider;
 import io.trino.execution.warnings.WarningCollector;
+import io.trino.metadata.AnalyzeMetadata;
 import io.trino.metadata.Metadata;
 import io.trino.metadata.QualifiedObjectName;
 import io.trino.metadata.ResolvedFunction;
@@ -336,7 +337,9 @@ private RelationPlan createExplainAnalyzePlan(Analysis analysis, ExplainAnalyze
 
     private RelationPlan createAnalyzePlan(Analysis analysis, Analyze analyzeStatement)
     {
-        TableHandle targetTable = analysis.getAnalyzeTarget().orElseThrow();
+        AnalyzeMetadata analyzeMetadata = analysis.getAnalyzeMetadata().orElseThrow();
+        TableHandle targetTable = analyzeMetadata.getTableHandle();
+        TableStatisticsMetadata tableStatisticsMetadata = analyzeMetadata.getStatisticsMetadata();
 
         // Plan table scan
         Map<String, ColumnHandle> columnHandles = metadata.getColumnHandles(session, targetTable);
@@ -351,11 +354,6 @@ private RelationPlan createAnalyzePlan(Analysis analysis, Analyze analyzeStateme
             columnNameToSymbol.put(column.getName(), symbol);
         }
 
-        TableStatisticsMetadata tableStatisticsMetadata = metadata.getStatisticsCollectionMetadata(
-                session,
-                targetTable.getCatalogName().getCatalogName(),
-                tableMetadata.getMetadata());
-
         TableStatisticAggregation tableStatisticAggregation = statisticsAggregationPlanner.createStatisticsAggregation(tableStatisticsMetadata, columnNameToSymbol.buildOrThrow());
         StatisticAggregations statisticAggregations = tableStatisticAggregation.getAggregations();
         List<Symbol> groupingSymbols = statisticAggregations.getGroupingSymbols();
diff --git a/core/trino-main/src/main/java/io/trino/testing/TestingMetadata.java b/core/trino-main/src/main/java/io/trino/testing/TestingMetadata.java
index 1290eb3aa7b7..4a1d42115cec 100644
--- a/core/trino-main/src/main/java/io/trino/testing/TestingMetadata.java
+++ b/core/trino-main/src/main/java/io/trino/testing/TestingMetadata.java
@@ -22,6 +22,7 @@
 import io.trino.spi.TrinoException;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorAnalyzeMetadata;
 import io.trino.spi.connector.ConnectorInsertTableHandle;
 import io.trino.spi.connector.ConnectorMaterializedViewDefinition;
 import io.trino.spi.connector.ConnectorMetadata;
@@ -42,6 +43,7 @@
 import io.trino.spi.security.Privilege;
 import io.trino.spi.security.TrinoPrincipal;
 import io.trino.spi.statistics.ComputedStatistics;
+import io.trino.spi.statistics.TableStatisticsMetadata;
 import io.trino.spi.type.Type;
 
 import java.util.ArrayList;
@@ -92,12 +94,6 @@ public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTable
         return new TestingTableHandle(tableName);
     }
 
-    @Override
-    public ConnectorTableHandle getTableHandleForStatisticsCollection(ConnectorSession session, SchemaTableName tableName, Map<String, Object> analyzeProperties)
-    {
-        return getTableHandle(session, tableName);
-    }
-
     @Override
     public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
@@ -136,6 +132,12 @@ public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSess
         return tableColumns.buildOrThrow();
     }
 
+    @Override
+    public ConnectorAnalyzeMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, Map<String, Object> analyzeProperties)
+    {
+        return new ConnectorAnalyzeMetadata(tableHandle, TableStatisticsMetadata.empty());
+    }
+
     @Override
     public ColumnMetadata getColumnMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnHandle columnHandle)
     {
diff --git a/core/trino-main/src/test/java/io/trino/metadata/AbstractMockMetadata.java b/core/trino-main/src/test/java/io/trino/metadata/AbstractMockMetadata.java
index 4e61af25490f..dbe508f57b6f 100644
--- a/core/trino-main/src/test/java/io/trino/metadata/AbstractMockMetadata.java
+++ b/core/trino-main/src/test/java/io/trino/metadata/AbstractMockMetadata.java
@@ -120,12 +120,6 @@ public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName
         throw new UnsupportedOperationException();
     }
 
-    @Override
-    public Optional<TableHandle> getTableHandleForStatisticsCollection(Session session, QualifiedObjectName tableName, Map<String, Object> analyzeProperties)
-    {
-        throw new UnsupportedOperationException();
-    }
-
     @Override
     public Optional<TableExecuteHandle> getTableHandleForExecute(Session session, TableHandle tableHandle, String procedureName, Map<String, Object> executeProperties)
     {
@@ -349,7 +343,7 @@ public TableStatisticsMetadata getStatisticsCollectionMetadataForWrite(Session s
     }
 
     @Override
-    public TableStatisticsMetadata getStatisticsCollectionMetadata(Session session, String catalogName, ConnectorTableMetadata tableMetadata)
+    public AnalyzeMetadata getStatisticsCollectionMetadata(Session session, TableHandle tableHandle, Map<String, Object> analyzeProperties)
     {
         throw new UnsupportedOperationException();
     }
diff --git a/core/trino-main/src/test/java/io/trino/metadata/CountingAccessMetadata.java b/core/trino-main/src/test/java/io/trino/metadata/CountingAccessMetadata.java
index e4121debc18b..823fc9fc9d91 100644
--- a/core/trino-main/src/test/java/io/trino/metadata/CountingAccessMetadata.java
+++ b/core/trino-main/src/test/java/io/trino/metadata/CountingAccessMetadata.java
@@ -132,12 +132,6 @@ public Optional<SystemTable> getSystemTable(Session session, QualifiedObjectName
         return delegate.getSystemTable(session, tableName);
     }
 
-    @Override
-    public Optional<TableHandle> getTableHandleForStatisticsCollection(Session session, QualifiedObjectName tableName, Map<String, Object> analyzeProperties)
-    {
-        return delegate.getTableHandleForStatisticsCollection(session, tableName, analyzeProperties);
-    }
-
     @Override
     public Optional<TableExecuteHandle> getTableHandleForExecute(Session session, TableHandle tableHandle, String procedureName, Map<String, Object> executeProperties)
     {
@@ -356,9 +350,9 @@ public TableStatisticsMetadata getStatisticsCollectionMetadataForWrite(Session s
     }
 
     @Override
-    public TableStatisticsMetadata getStatisticsCollectionMetadata(Session session, String catalogName, ConnectorTableMetadata tableMetadata)
+    public AnalyzeMetadata getStatisticsCollectionMetadata(Session session, TableHandle tableHandle, Map<String, Object> analyzeProperties)
     {
-        return delegate.getStatisticsCollectionMetadata(session, catalogName, tableMetadata);
+        return delegate.getStatisticsCollectionMetadata(session, tableHandle, analyzeProperties);
     }
 
     @Override
diff --git a/core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorAnalyzeMetadata.java b/core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorAnalyzeMetadata.java
new file mode 100644
index 000000000000..4ce49f52fe79
--- /dev/null
+++ b/core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorAnalyzeMetadata.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed 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 io.trino.spi.connector;
+
+import io.trino.spi.statistics.TableStatisticsMetadata;
+
+import static java.util.Objects.requireNonNull;
+
+public class ConnectorAnalyzeMetadata
+{
+    private final ConnectorTableHandle tableHandle;
+    private final TableStatisticsMetadata statisticsMetadata;
+
+    public ConnectorAnalyzeMetadata(ConnectorTableHandle tableHandle, TableStatisticsMetadata statisticsMetadata)
+    {
+        this.statisticsMetadata = requireNonNull(statisticsMetadata, "statisticsMetadata is null");
+        this.tableHandle = requireNonNull(tableHandle, "tableHandle is null");
+    }
+
+    public ConnectorTableHandle getTableHandle()
+    {
+        return tableHandle;
+    }
+
+    public TableStatisticsMetadata getStatisticsMetadata()
+    {
+        return statisticsMetadata;
+    }
+}
diff --git a/core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorMetadata.java b/core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorMetadata.java
index c128f1bd28b9..7870900eb2ad 100644
--- a/core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorMetadata.java
+++ b/core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorMetadata.java
@@ -85,7 +85,10 @@ default ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTabl
     /**
      * Returns a table handle for the specified table name, or null if the connector does not contain the table.
      * The returned table handle can contain information in analyzeProperties.
+     *
+     * @deprecated use {@link #getStatisticsCollectionMetadata(ConnectorSession, ConnectorTableHandle, Map)}
      */
+    @Deprecated
     @Nullable
     default ConnectorTableHandle getTableHandleForStatisticsCollection(ConnectorSession session, SchemaTableName tableName, Map<String, Object> analyzeProperties)
     {
@@ -437,12 +440,26 @@ default TableStatisticsMetadata getStatisticsCollectionMetadataForWrite(Connecto
 
     /**
      * Describe statistics that must be collected during a statistics collection
+     *
+     * @deprecated use {@link #getStatisticsCollectionMetadata(ConnectorSession, ConnectorTableHandle, Map)}
      */
+    @Deprecated
     default TableStatisticsMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableMetadata tableMetadata)
     {
         throw new TrinoException(GENERIC_INTERNAL_ERROR, "ConnectorMetadata getTableHandleForStatisticsCollection() is implemented without getStatisticsCollectionMetadata()");
     }
 
+    /**
+     * Describe statistics that must be collected during a statistics collection
+     */
+    default ConnectorAnalyzeMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, Map<String, Object> analyzeProperties)
+    {
+        SchemaTableName tableName = getTableMetadata(session, tableHandle).getTable();
+        ConnectorTableHandle analyzeHandle = getTableHandleForStatisticsCollection(session, tableName, analyzeProperties);
+        TableStatisticsMetadata statisticsMetadata = getStatisticsCollectionMetadata(session, getTableMetadata(session, analyzeHandle));
+        return new ConnectorAnalyzeMetadata(analyzeHandle, statisticsMetadata);
+    }
+
     /**
      * Begin statistics collection
      */
diff --git a/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorMetadata.java b/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorMetadata.java
index 14d437a9a1a0..04168be556c5 100644
--- a/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorMetadata.java
+++ b/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorMetadata.java
@@ -22,6 +22,7 @@
 import io.trino.spi.connector.CatalogSchemaTableName;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorAnalyzeMetadata;
 import io.trino.spi.connector.ConnectorInsertTableHandle;
 import io.trino.spi.connector.ConnectorMaterializedViewDefinition;
 import io.trino.spi.connector.ConnectorMetadata;
@@ -144,6 +145,14 @@ public TableStatisticsMetadata getStatisticsCollectionMetadata(ConnectorSession
         }
     }
 
+    @Override
+    public ConnectorAnalyzeMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, Map<String, Object> analyzeProperties)
+    {
+        try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
+            return delegate.getStatisticsCollectionMetadata(session, tableHandle, analyzeProperties);
+        }
+    }
+
     @Override
     public ConnectorTableHandle beginStatisticsCollection(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
diff --git a/plugin/trino-blackhole/src/main/java/io/trino/plugin/blackhole/BlackHoleMetadata.java b/plugin/trino-blackhole/src/main/java/io/trino/plugin/blackhole/BlackHoleMetadata.java
index 5fd1626f0ca7..aa16a8716caf 100644
--- a/plugin/trino-blackhole/src/main/java/io/trino/plugin/blackhole/BlackHoleMetadata.java
+++ b/plugin/trino-blackhole/src/main/java/io/trino/plugin/blackhole/BlackHoleMetadata.java
@@ -22,6 +22,7 @@
 import io.trino.spi.TrinoException;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorAnalyzeMetadata;
 import io.trino.spi.connector.ConnectorInsertTableHandle;
 import io.trino.spi.connector.ConnectorMetadata;
 import io.trino.spi.connector.ConnectorOutputMetadata;
@@ -104,15 +105,9 @@ public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTable
     }
 
     @Override
-    public ConnectorTableHandle getTableHandleForStatisticsCollection(ConnectorSession session, SchemaTableName tableName, Map<String, Object> analyzeProperties)
+    public ConnectorAnalyzeMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, Map<String, Object> analyzeProperties)
     {
-        return getTableHandle(session, tableName);
-    }
-
-    @Override
-    public TableStatisticsMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableMetadata tableMetadata)
-    {
-        return TableStatisticsMetadata.empty();
+        return new ConnectorAnalyzeMetadata(tableHandle, TableStatisticsMetadata.empty());
     }
 
     @Override
diff --git a/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/AnalyzeHandle.java b/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/AnalyzeHandle.java
index 6edaf03dbf52..64dd5f248aff 100644
--- a/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/AnalyzeHandle.java
+++ b/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/AnalyzeHandle.java
@@ -25,31 +25,22 @@
 
 public class AnalyzeHandle
 {
-    private final long tableVersion;
     private final boolean initialAnalyze;
     private final Optional<Instant> filesModifiedAfter;
     private final Optional<Set<String>> columns;
 
     @JsonCreator
     public AnalyzeHandle(
-            @JsonProperty("tableVersion") long tableVersion,
             @JsonProperty("initialAnalyze") boolean initialAnalyze,
             @JsonProperty("startTime") Optional<Instant> filesModifiedAfter,
             @JsonProperty("columns") Optional<Set<String>> columns)
     {
-        this.tableVersion = tableVersion;
         this.initialAnalyze = initialAnalyze;
         this.filesModifiedAfter = requireNonNull(filesModifiedAfter, "filesModifiedAfter is null");
         requireNonNull(columns, "columns is null");
         this.columns = columns.map(ImmutableSet::copyOf);
     }
 
-    @JsonProperty
-    public long getTableVersion()
-    {
-        return tableVersion;
-    }
-
     @JsonProperty
     public boolean isInitialAnalyze()
     {
diff --git a/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java b/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java
index 038cc35e30fb..f73561ba42a3 100644
--- a/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java
+++ b/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java
@@ -64,6 +64,7 @@
 import io.trino.spi.connector.CatalogSchemaTableName;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorAnalyzeMetadata;
 import io.trino.spi.connector.ConnectorInsertTableHandle;
 import io.trino.spi.connector.ConnectorMetadata;
 import io.trino.spi.connector.ConnectorOutputMetadata;
@@ -149,7 +150,6 @@
 import static io.trino.plugin.deltalake.DeltaLakeErrorCode.DELTA_LAKE_INVALID_SCHEMA;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isExtendedStatisticsEnabled;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isTableStatisticsEnabled;
-import static io.trino.plugin.deltalake.DeltaLakeTableProperties.ANALYZE_COLUMNS_PROPERTY;
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.CHECKPOINT_INTERVAL_PROPERTY;
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.LOCATION_PROPERTY;
 import static io.trino.plugin.deltalake.DeltaLakeTableProperties.PARTITIONED_BY_PROPERTY;
@@ -400,10 +400,6 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
         Optional<Long> checkpointInterval = tableHandle.getMetadataEntry().getCheckpointInterval();
         checkpointInterval.ifPresent(value -> properties.put(CHECKPOINT_INTERVAL_PROPERTY, value));
 
-        tableHandle.getAnalyzeHandle().flatMap(AnalyzeHandle::getColumns).ifPresent(
-                // we use table properties as a vehicle to pass to the analyzer the subset of columns to be analyzed
-                analyzeColumns -> properties.put(ANALYZE_COLUMNS_PROPERTY, analyzeColumns));
-
         return new ConnectorTableMetadata(
                 tableHandle.getSchemaTableName(),
                 columns,
@@ -1855,27 +1851,21 @@ public Optional<TableScanRedirectApplicationResult> applyTableScanRedirect(Conne
         return deltaLakeRedirectionsProvider.getTableScanRedirection(session, (DeltaLakeTableHandle) tableHandle);
     }
 
-    @Nullable
     @Override
-    public ConnectorTableHandle getTableHandleForStatisticsCollection(ConnectorSession session, SchemaTableName tableName, Map<String, Object> analyzeProperties)
+    public ConnectorAnalyzeMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, Map<String, Object> analyzeProperties)
     {
-        Optional<Table> table = metastore.getTable(tableName.getSchemaName(), tableName.getTableName());
-        if (table.isEmpty()) {
-            return null;
-        }
-
         if (!isExtendedStatisticsEnabled(session)) {
             throw new TrinoException(
                     NOT_SUPPORTED,
                     "ANALYZE not supported if extended statistics are disabled. Enable via delta.extended-statistics.enabled config property or extended_statistics_enabled session property.");
         }
 
+        DeltaLakeTableHandle handle = (DeltaLakeTableHandle) tableHandle;
+        MetadataEntry metadata = handle.getMetadataEntry();
+
         Optional<Instant> filesModifiedAfterFromProperties = DeltaLakeAnalyzeProperties.getFilesModifiedAfterProperty(analyzeProperties);
-        TableSnapshot tableSnapshot = metastore.getSnapshot(tableName, session);
-        long version = tableSnapshot.getVersion();
 
-        String tableLocation = metastore.getTableLocation(tableName, session);
-        Optional<ExtendedStatistics> statistics = statisticsAccess.readExtendedStatistics(session, tableLocation);
+        Optional<ExtendedStatistics> statistics = statisticsAccess.readExtendedStatistics(session, handle.getLocation());
 
         Optional<Instant> alreadyAnalyzedModifiedTimeMax = statistics.map(ExtendedStatistics::getAlreadyAnalyzedModifiedTimeMax);
 
@@ -1887,9 +1877,6 @@ public ConnectorTableHandle getTableHandleForStatisticsCollection(ConnectorSessi
                     alreadyAnalyzedModifiedTimeMax.orElse(Instant.ofEpochMilli(0))));
         }
 
-        MetadataEntry metadata = metastore.getMetadata(tableSnapshot, session)
-                .orElseThrow(() -> new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Metadata not found in transaction log for " + table));
-
         Optional<Set<String>> analyzeColumnNames = DeltaLakeAnalyzeProperties.getColumnNames(analyzeProperties);
         if (analyzeColumnNames.isPresent()) {
             Set<String> columnNames = analyzeColumnNames.get();
@@ -1917,11 +1904,11 @@ public ConnectorTableHandle getTableHandleForStatisticsCollection(ConnectorSessi
             }
         }
 
-        AnalyzeHandle analyzeHandle = new AnalyzeHandle(version, statistics.isEmpty(), filesModifiedAfter, analyzeColumnNames);
-        return new DeltaLakeTableHandle(
-                tableName.getSchemaName(),
-                tableName.getTableName(),
-                tableLocation,
+        AnalyzeHandle analyzeHandle = new AnalyzeHandle(statistics.isEmpty(), filesModifiedAfter, analyzeColumnNames);
+        DeltaLakeTableHandle newHandle = new DeltaLakeTableHandle(
+                handle.getSchemaTableName().getSchemaName(),
+                handle.getSchemaTableName().getTableName(),
+                handle.getLocation(),
                 Optional.of(metadata),
                 TupleDomain.all(),
                 TupleDomain.all(),
@@ -1930,16 +1917,11 @@ public ConnectorTableHandle getTableHandleForStatisticsCollection(ConnectorSessi
                 Optional.empty(),
                 Optional.empty(),
                 Optional.of(analyzeHandle),
-                version,
+                handle.getReadVersion(),
                 false);
-    }
 
-    @Override
-    public TableStatisticsMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableMetadata tableMetadata)
-    {
         ImmutableSet.Builder<ColumnStatisticMetadata> columnStatistics = ImmutableSet.builder();
-        Optional<Set<String>> analyzeColumnNames = DeltaLakeTableProperties.getAnalyzeColumns(tableMetadata.getProperties());
-        tableMetadata.getColumns().stream()
+        extractSchema(metadata, typeManager).stream()
                 .filter(DeltaLakeMetadata::shouldCollectExtendedStatistics)
                 .filter(columnMetadata ->
                         analyzeColumnNames
@@ -1951,10 +1933,12 @@ public TableStatisticsMetadata getStatisticsCollectionMetadata(ConnectorSession
         // collect max(file modification time) for sake of incremental ANALYZE
         columnStatistics.add(new ColumnStatisticMetadata(FILE_MODIFIED_TIME_COLUMN_NAME, MAX_VALUE));
 
-        return new TableStatisticsMetadata(
+        TableStatisticsMetadata statisticsMetadata = new TableStatisticsMetadata(
                 columnStatistics.build(),
                 ImmutableSet.of(),
                 ImmutableList.of());
+
+        return new ConnectorAnalyzeMetadata(newHandle, statisticsMetadata);
     }
 
     private static boolean shouldCollectExtendedStatistics(ColumnMetadata columnMetadata)
diff --git a/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeTableProperties.java b/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeTableProperties.java
index d594a64663c7..c97f1564b9f9 100644
--- a/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeTableProperties.java
+++ b/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeTableProperties.java
@@ -24,7 +24,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 import java.util.stream.Collectors;
 
 import static io.trino.spi.StandardErrorCode.INVALID_TABLE_PROPERTY;
@@ -39,11 +38,6 @@ public class DeltaLakeTableProperties
     public static final String LOCATION_PROPERTY = "location";
     public static final String PARTITIONED_BY_PROPERTY = "partitioned_by";
     public static final String CHECKPOINT_INTERVAL_PROPERTY = "checkpoint_interval";
-    // TODO: This property represents the subset of columns to be analyzed. This exists mainly because there is no way
-    //       to pass the column names to ConnectorMetadata#getStatisticsCollectionMetadata; we should consider passing
-    //       ConnectorTableHandle instead of ConnectorTableMetadata as an argument since it makes more information
-    //       available (including the names of the columns to be analyzed)
-    public static final String ANALYZE_COLUMNS_PROPERTY = "$trino.analyze_columns";
 
     private final List<PropertyMetadata<?>> tableProperties;
 
@@ -103,10 +97,4 @@ public static Optional<Long> getCheckpointInterval(Map<String, Object> tableProp
 
         return checkpointInterval;
     }
-
-    @SuppressWarnings("unchecked")
-    public static Optional<Set<String>> getAnalyzeColumns(Map<String, Object> tableProperties)
-    {
-        return Optional.ofNullable((Set<String>) tableProperties.get(ANALYZE_COLUMNS_PROPERTY));
-    }
 }
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java
index 639de1407d72..c030c525f7e8 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java
@@ -62,6 +62,7 @@
 import io.trino.spi.connector.CatalogSchemaTableName;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorAnalyzeMetadata;
 import io.trino.spi.connector.ConnectorInsertTableHandle;
 import io.trino.spi.connector.ConnectorMaterializedViewDefinition;
 import io.trino.spi.connector.ConnectorOutputMetadata;
@@ -199,7 +200,6 @@
 import static io.trino.plugin.hive.HiveSessionProperties.isRespectTableFormat;
 import static io.trino.plugin.hive.HiveSessionProperties.isSortedWritingEnabled;
 import static io.trino.plugin.hive.HiveSessionProperties.isStatisticsEnabled;
-import static io.trino.plugin.hive.HiveTableProperties.ANALYZE_COLUMNS_PROPERTY;
 import static io.trino.plugin.hive.HiveTableProperties.AUTO_PURGE;
 import static io.trino.plugin.hive.HiveTableProperties.AVRO_SCHEMA_URL;
 import static io.trino.plugin.hive.HiveTableProperties.BUCKETED_BY_PROPERTY;
@@ -218,7 +218,6 @@
 import static io.trino.plugin.hive.HiveTableProperties.STORAGE_FORMAT_PROPERTY;
 import static io.trino.plugin.hive.HiveTableProperties.TEXTFILE_FIELD_SEPARATOR;
 import static io.trino.plugin.hive.HiveTableProperties.TEXTFILE_FIELD_SEPARATOR_ESCAPE;
-import static io.trino.plugin.hive.HiveTableProperties.getAnalyzeColumns;
 import static io.trino.plugin.hive.HiveTableProperties.getAvroSchemaUrl;
 import static io.trino.plugin.hive.HiveTableProperties.getBucketProperty;
 import static io.trino.plugin.hive.HiveTableProperties.getExternalLocation;
@@ -470,17 +469,15 @@ public HiveTableHandle getTableHandle(ConnectorSession session, SchemaTableName
     }
 
     @Override
-    public ConnectorTableHandle getTableHandleForStatisticsCollection(ConnectorSession session, SchemaTableName tableName, Map<String, Object> analyzeProperties)
+    public ConnectorAnalyzeMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, Map<String, Object> analyzeProperties)
     {
-        HiveTableHandle handle = getTableHandle(session, tableName);
-        if (handle == null) {
-            return null;
-        }
+        HiveTableHandle handle = (HiveTableHandle) tableHandle;
         Optional<List<List<String>>> partitionValuesList = getPartitionList(analyzeProperties);
         Optional<Set<String>> analyzeColumnNames = getColumnNames(analyzeProperties);
-        ConnectorTableMetadata tableMetadata = getTableMetadata(session, handle.getSchemaTableName());
 
-        List<String> partitionedBy = getPartitionedBy(tableMetadata.getProperties());
+        List<String> partitionedBy = handle.getPartitionColumns().stream()
+                .map(HiveColumnHandle::getName)
+                .collect(toImmutableList());
 
         if (partitionValuesList.isPresent()) {
             List<List<String>> list = partitionValuesList.get();
@@ -501,19 +498,26 @@ public ConnectorTableHandle getTableHandleForStatisticsCollection(ConnectorSessi
 
         if (analyzeColumnNames.isPresent()) {
             Set<String> columnNames = analyzeColumnNames.get();
-            Set<String> allColumnNames = tableMetadata.getColumns().stream()
-                    .map(ColumnMetadata::getName)
-                    .collect(toImmutableSet());
+            Set<String> allColumnNames = ImmutableSet.<String>builder()
+                    .addAll(handle.getDataColumns().stream()
+                            .map(HiveColumnHandle::getName)
+                            .collect(toImmutableSet()))
+                    .addAll(partitionedBy)
+                    .build();
             if (!allColumnNames.containsAll(columnNames)) {
                 throw new TrinoException(
                         INVALID_ANALYZE_PROPERTY,
                         format("Invalid columns specified for analysis: %s", Sets.difference(columnNames, allColumnNames)));
             }
-
-            handle = handle.withAnalyzeColumnNames(columnNames);
         }
 
-        return handle;
+        List<ColumnMetadata> columns = handle.getDataColumns().stream()
+                .map(HiveColumnHandle::getColumnMetadata)
+                .collect(toImmutableList());
+
+        TableStatisticsMetadata metadata = getStatisticsCollectionMetadata(columns, partitionedBy, analyzeColumnNames, true);
+
+        return new ConnectorAnalyzeMetadata(handle, metadata);
     }
 
     @Override
@@ -532,20 +536,7 @@ public Optional<SystemTable> getSystemTable(ConnectorSession session, SchemaTabl
     @Override
     public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
-        HiveTableHandle hiveTableHandle = (HiveTableHandle) tableHandle;
-        ConnectorTableMetadata tableMetadata = getTableMetadata(session, hiveTableHandle.getSchemaTableName());
-
-        return hiveTableHandle.getAnalyzeColumnNames()
-                .map(columnNames -> new ConnectorTableMetadata(
-                        tableMetadata.getTable(),
-                        tableMetadata.getColumns(),
-                        ImmutableMap.<String, Object>builder()
-                                .putAll(tableMetadata.getProperties())
-                                // we use table properties as a vehicle to pass to the analyzer the subset of columns to be analyzed
-                                .put(ANALYZE_COLUMNS_PROPERTY, columnNames)
-                                .buildOrThrow(),
-                        tableMetadata.getComment()))
-                .orElse(tableMetadata);
+        return getTableMetadata(session, ((HiveTableHandle) tableHandle).getSchemaTableName());
     }
 
     private ConnectorTableMetadata getTableMetadata(ConnectorSession session, SchemaTableName tableName)
@@ -2935,7 +2926,6 @@ public ConnectorTableHandle makeCompatiblePartitioning(ConnectorSession session,
                         bucketHandle.getSortedBy())),
                 hiveTable.getBucketFilter(),
                 hiveTable.getAnalyzePartitionValues(),
-                hiveTable.getAnalyzeColumnNames(),
                 ImmutableSet.of(),
                 ImmutableSet.of(), // Projected columns is used only during optimization phase of planning
                 hiveTable.getTransaction(),
@@ -3112,13 +3102,6 @@ public TableStatisticsMetadata getStatisticsCollectionMetadataForWrite(Connector
         return getStatisticsCollectionMetadata(tableMetadata.getColumns(), partitionedBy, Optional.empty(), false);
     }
 
-    @Override
-    public TableStatisticsMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableMetadata tableMetadata)
-    {
-        List<String> partitionedBy = firstNonNull(getPartitionedBy(tableMetadata.getProperties()), ImmutableList.of());
-        return getStatisticsCollectionMetadata(tableMetadata.getColumns(), partitionedBy, getAnalyzeColumns(tableMetadata.getProperties()), true);
-    }
-
     private TableStatisticsMetadata getStatisticsCollectionMetadata(List<ColumnMetadata> columns, List<String> partitionedBy, Optional<Set<String>> analyzeColumns, boolean includeRowCount)
     {
         Set<ColumnStatisticMetadata> columnStatistics = columns.stream()
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HivePartitionManager.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HivePartitionManager.java
index e2e9061c5700..03f799c42484 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HivePartitionManager.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HivePartitionManager.java
@@ -203,7 +203,6 @@ public HiveTableHandle applyPartitionResult(HiveTableHandle handle, HivePartitio
                 partitions.getBucketHandle(),
                 partitions.getBucketFilter(),
                 handle.getAnalyzePartitionValues(),
-                handle.getAnalyzeColumnNames(),
                 Sets.union(handle.getConstraintColumns(), constraint.getPredicateColumns().orElseGet(ImmutableSet::of)),
                 handle.getProjectedColumns(),
                 handle.getTransaction(),
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveTableHandle.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveTableHandle.java
index e8153b6ca957..a42830ef250d 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveTableHandle.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveTableHandle.java
@@ -53,7 +53,6 @@ public class HiveTableHandle
     private final Optional<HiveBucketHandle> bucketHandle;
     private final Optional<HiveBucketFilter> bucketFilter;
     private final Optional<List<List<String>>> analyzePartitionValues;
-    private final Optional<Set<String>> analyzeColumnNames;
     private final Set<ColumnHandle> constraintColumns;
     private final Set<ColumnHandle> projectedColumns;
     private final AcidTransaction transaction;
@@ -71,7 +70,6 @@ public HiveTableHandle(
             @JsonProperty("bucketHandle") Optional<HiveBucketHandle> bucketHandle,
             @JsonProperty("bucketFilter") Optional<HiveBucketFilter> bucketFilter,
             @JsonProperty("analyzePartitionValues") Optional<List<List<String>>> analyzePartitionValues,
-            @JsonProperty("analyzeColumnNames") Optional<Set<String>> analyzeColumnNames,
             @JsonProperty("transaction") AcidTransaction transaction)
     {
         this(
@@ -87,7 +85,6 @@ public HiveTableHandle(
                 bucketHandle,
                 bucketFilter,
                 analyzePartitionValues,
-                analyzeColumnNames,
                 ImmutableSet.of(),
                 ImmutableSet.of(),
                 transaction,
@@ -116,7 +113,6 @@ public HiveTableHandle(
                 bucketHandle,
                 Optional.empty(),
                 Optional.empty(),
-                Optional.empty(),
                 ImmutableSet.of(),
                 ImmutableSet.of(),
                 NO_ACID_TRANSACTION,
@@ -137,7 +133,6 @@ public HiveTableHandle(
             Optional<HiveBucketHandle> bucketHandle,
             Optional<HiveBucketFilter> bucketFilter,
             Optional<List<List<String>>> analyzePartitionValues,
-            Optional<Set<String>> analyzeColumnNames,
             Set<ColumnHandle> constraintColumns,
             Set<ColumnHandle> projectedColumns,
             AcidTransaction transaction,
@@ -157,7 +152,6 @@ public HiveTableHandle(
         this.bucketHandle = requireNonNull(bucketHandle, "bucketHandle is null");
         this.bucketFilter = requireNonNull(bucketFilter, "bucketFilter is null");
         this.analyzePartitionValues = requireNonNull(analyzePartitionValues, "analyzePartitionValues is null").map(ImmutableList::copyOf);
-        this.analyzeColumnNames = requireNonNull(analyzeColumnNames, "analyzeColumnNames is null").map(ImmutableSet::copyOf);
         this.constraintColumns = ImmutableSet.copyOf(requireNonNull(constraintColumns, "constraintColumns is null"));
         this.projectedColumns = ImmutableSet.copyOf(requireNonNull(projectedColumns, "projectedColumns is null"));
         this.transaction = requireNonNull(transaction, "transaction is null");
@@ -180,7 +174,6 @@ public HiveTableHandle withAnalyzePartitionValues(List<List<String>> analyzePart
                 bucketHandle,
                 bucketFilter,
                 Optional.of(analyzePartitionValues),
-                analyzeColumnNames,
                 constraintColumns,
                 projectedColumns,
                 transaction,
@@ -188,7 +181,7 @@ public HiveTableHandle withAnalyzePartitionValues(List<List<String>> analyzePart
                 maxScannedFileSize);
     }
 
-    public HiveTableHandle withAnalyzeColumnNames(Set<String> analyzeColumnNames)
+    public HiveTableHandle withAnalyzeColumnNames()
     {
         return new HiveTableHandle(
                 schemaName,
@@ -203,7 +196,6 @@ public HiveTableHandle withAnalyzeColumnNames(Set<String> analyzeColumnNames)
                 bucketHandle,
                 bucketFilter,
                 analyzePartitionValues,
-                Optional.of(analyzeColumnNames),
                 constraintColumns,
                 projectedColumns,
                 transaction,
@@ -226,7 +218,6 @@ public HiveTableHandle withTransaction(AcidTransaction transaction)
                 bucketHandle,
                 bucketFilter,
                 analyzePartitionValues,
-                analyzeColumnNames,
                 constraintColumns,
                 projectedColumns,
                 transaction,
@@ -250,7 +241,6 @@ public HiveTableHandle withUpdateProcessor(AcidTransaction transaction, HiveUpda
                 bucketHandle,
                 bucketFilter,
                 analyzePartitionValues,
-                analyzeColumnNames,
                 constraintColumns,
                 projectedColumns,
                 transaction,
@@ -273,7 +263,6 @@ public HiveTableHandle withProjectedColumns(Set<ColumnHandle> projectedColumns)
                 bucketHandle,
                 bucketFilter,
                 analyzePartitionValues,
-                analyzeColumnNames,
                 constraintColumns,
                 projectedColumns,
                 transaction,
@@ -296,7 +285,6 @@ public HiveTableHandle withRecordScannedFiles(boolean recordScannedFiles)
                 bucketHandle,
                 bucketFilter,
                 analyzePartitionValues,
-                analyzeColumnNames,
                 constraintColumns,
                 projectedColumns,
                 transaction,
@@ -319,7 +307,6 @@ public HiveTableHandle withMaxScannedFileSize(Optional<Long> maxScannedFileSize)
                 bucketHandle,
                 bucketFilter,
                 analyzePartitionValues,
-                analyzeColumnNames,
                 constraintColumns,
                 projectedColumns,
                 transaction,
@@ -411,12 +398,6 @@ public Optional<List<List<String>>> getAnalyzePartitionValues()
         return analyzePartitionValues;
     }
 
-    @JsonProperty
-    public Optional<Set<String>> getAnalyzeColumnNames()
-    {
-        return analyzeColumnNames;
-    }
-
     @JsonProperty
     public AcidTransaction getTransaction()
     {
diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveTableProperties.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveTableProperties.java
index 0a9a0346476a..2c31074aa5f5 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveTableProperties.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveTableProperties.java
@@ -27,7 +27,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static io.trino.plugin.hive.util.HiveBucketing.BucketingVersion.BUCKETING_V1;
@@ -51,11 +50,6 @@ public class HiveTableProperties
     public static final String BUCKETING_VERSION = "bucketing_version";
     public static final String BUCKET_COUNT_PROPERTY = "bucket_count";
     public static final String SORTED_BY_PROPERTY = "sorted_by";
-    // TODO: This property represents the subset of columns to be analyzed. This exists mainly because there is no way
-    //       to pass the column names to ConnectorMetadata#getStatisticsCollectionMetadata; we should consider passing
-    //       ConnectorTableHandle instead of ConnectorTableMetadata as an argument since it makes more information
-    //       available (including the names of the columns to be analyzed)
-    public static final String ANALYZE_COLUMNS_PROPERTY = "presto.analyze_columns";
     public static final String ORC_BLOOM_FILTER_COLUMNS = "orc_bloom_filter_columns";
     public static final String ORC_BLOOM_FILTER_FPP = "orc_bloom_filter_fpp";
     public static final String AVRO_SCHEMA_URL = "avro_schema_url";
@@ -200,12 +194,6 @@ public static List<String> getPartitionedBy(Map<String, Object> tableProperties)
         return partitionedBy == null ? ImmutableList.of() : ImmutableList.copyOf(partitionedBy);
     }
 
-    @SuppressWarnings("unchecked")
-    public static Optional<Set<String>> getAnalyzeColumns(Map<String, Object> tableProperties)
-    {
-        return Optional.ofNullable((Set<String>) tableProperties.get(ANALYZE_COLUMNS_PROPERTY));
-    }
-
     public static Optional<HiveBucketProperty> getBucketProperty(Map<String, Object> tableProperties)
     {
         List<String> bucketedBy = getBucketedBy(tableProperties);
diff --git a/plugin/trino-tpch/src/main/java/io/trino/plugin/tpch/TpchMetadata.java b/plugin/trino-tpch/src/main/java/io/trino/plugin/tpch/TpchMetadata.java
index 8609ae0bd1c7..3b3555a012b5 100644
--- a/plugin/trino-tpch/src/main/java/io/trino/plugin/tpch/TpchMetadata.java
+++ b/plugin/trino-tpch/src/main/java/io/trino/plugin/tpch/TpchMetadata.java
@@ -29,6 +29,7 @@
 import io.trino.spi.connector.CatalogSchemaTableName;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorAnalyzeMetadata;
 import io.trino.spi.connector.ConnectorMetadata;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.ConnectorTableHandle;
@@ -201,12 +202,6 @@ public TpchTableHandle getTableHandle(ConnectorSession session, SchemaTableName
         return new TpchTableHandle(tableName.getSchemaName(), tableName.getTableName(), scaleFactor);
     }
 
-    @Override
-    public ConnectorTableHandle getTableHandleForStatisticsCollection(ConnectorSession session, SchemaTableName tableName, Map<String, Object> analyzeProperties)
-    {
-        return getTableHandle(session, tableName);
-    }
-
     private Set<NullableValue> filterValues(Set<NullableValue> nullableValues, TpchColumn<?> column, Constraint constraint)
     {
         return nullableValues.stream()
@@ -375,9 +370,9 @@ private static double toDouble(Object value, Type columnType)
     }
 
     @Override
-    public TableStatisticsMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableMetadata tableMetadata)
+    public ConnectorAnalyzeMetadata getStatisticsCollectionMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, Map<String, Object> analyzeProperties)
     {
-        return new TableStatisticsMetadata(ImmutableSet.of(), ImmutableSet.of(ROW_COUNT), ImmutableList.of());
+        return new ConnectorAnalyzeMetadata(tableHandle, new TableStatisticsMetadata(ImmutableSet.of(), ImmutableSet.of(ROW_COUNT), ImmutableList.of()));
     }
 
     @Override