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 a83fa5e212cb..f8e4a9f618bd 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
@@ -741,16 +741,6 @@ default boolean isMaterializedView(Session session, QualifiedObjectName viewName
      */
     RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion);
 
-    /**
-     * Returns true if the connector reports number of written bytes for an existing table. Otherwise, it returns false.
-     */
-    boolean supportsReportingWrittenBytes(Session session, TableHandle tableHandle);
-
-    /**
-     * Returns true if the connector reports number of written bytes for a new table. Otherwise, it returns false.
-     */
-    boolean supportsReportingWrittenBytes(Session session, QualifiedObjectName tableName, Map<String, Object> tableProperties);
-
     /**
      * Returns a table handle for the specified table name with a specified version
      */
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 4f88f3d7b4b7..06f8fb13fa33 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
@@ -2603,27 +2603,6 @@ private synchronized void finish()
         }
     }
 
-    @Override
-    public boolean supportsReportingWrittenBytes(Session session, QualifiedObjectName tableName, Map<String, Object> tableProperties)
-    {
-        Optional<CatalogMetadata> catalog = getOptionalCatalogMetadata(session, tableName.getCatalogName());
-        if (catalog.isEmpty()) {
-            return false;
-        }
-
-        CatalogMetadata catalogMetadata = catalog.get();
-        CatalogHandle catalogHandle = catalogMetadata.getCatalogHandle(session, tableName);
-        ConnectorMetadata metadata = catalogMetadata.getMetadataFor(session, catalogHandle);
-        return metadata.supportsReportingWrittenBytes(session.toConnectorSession(catalogHandle), tableName.asSchemaTableName(), tableProperties);
-    }
-
-    @Override
-    public boolean supportsReportingWrittenBytes(Session session, TableHandle tableHandle)
-    {
-        ConnectorMetadata metadata = getMetadata(session, tableHandle.getCatalogHandle());
-        return metadata.supportsReportingWrittenBytes(session.toConnectorSession(tableHandle.getCatalogHandle()), tableHandle.getConnectorHandle());
-    }
-
     @Override
     public OptionalInt getMaxWriterTasks(Session session, String catalogName)
     {
diff --git a/core/trino-main/src/main/java/io/trino/tracing/TracingConnectorMetadata.java b/core/trino-main/src/main/java/io/trino/tracing/TracingConnectorMetadata.java
index 6304fb0094ba..c74f4c52b408 100644
--- a/core/trino-main/src/main/java/io/trino/tracing/TracingConnectorMetadata.java
+++ b/core/trino-main/src/main/java/io/trino/tracing/TracingConnectorMetadata.java
@@ -1226,24 +1226,6 @@ public Optional<CatalogSchemaTableName> redirectTable(ConnectorSession session,
         }
     }
 
-    @Override
-    public boolean supportsReportingWrittenBytes(ConnectorSession session, SchemaTableName schemaTableName, Map<String, Object> tableProperties)
-    {
-        Span span = startSpan("supportsReportingWrittenBytes", schemaTableName);
-        try (var ignored = scopedSpan(span)) {
-            return delegate.supportsReportingWrittenBytes(session, schemaTableName, tableProperties);
-        }
-    }
-
-    @Override
-    public boolean supportsReportingWrittenBytes(ConnectorSession session, ConnectorTableHandle connectorTableHandle)
-    {
-        Span span = startSpan("supportsReportingWrittenBytes", connectorTableHandle);
-        try (var ignored = scopedSpan(span)) {
-            return delegate.supportsReportingWrittenBytes(session, connectorTableHandle);
-        }
-    }
-
     @Override
     public OptionalInt getMaxWriterTasks(ConnectorSession session)
     {
diff --git a/core/trino-main/src/main/java/io/trino/tracing/TracingMetadata.java b/core/trino-main/src/main/java/io/trino/tracing/TracingMetadata.java
index 91f57c093336..e9551912e9d9 100644
--- a/core/trino-main/src/main/java/io/trino/tracing/TracingMetadata.java
+++ b/core/trino-main/src/main/java/io/trino/tracing/TracingMetadata.java
@@ -1340,24 +1340,6 @@ public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session sessio
         }
     }
 
-    @Override
-    public boolean supportsReportingWrittenBytes(Session session, TableHandle tableHandle)
-    {
-        Span span = startSpan("supportsReportingWrittenBytes", tableHandle);
-        try (var ignored = scopedSpan(span)) {
-            return delegate.supportsReportingWrittenBytes(session, tableHandle);
-        }
-    }
-
-    @Override
-    public boolean supportsReportingWrittenBytes(Session session, QualifiedObjectName tableName, Map<String, Object> tableProperties)
-    {
-        Span span = startSpan("supportsReportingWrittenBytes", tableName);
-        try (var ignored = scopedSpan(span)) {
-            return delegate.supportsReportingWrittenBytes(session, tableName, tableProperties);
-        }
-    }
-
     @Override
     public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion)
     {
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 b8f750a46f51..361b58e600d6 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
@@ -923,18 +923,6 @@ public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session sessio
         throw new UnsupportedOperationException();
     }
 
-    @Override
-    public boolean supportsReportingWrittenBytes(Session session, TableHandle tableHandle)
-    {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public boolean supportsReportingWrittenBytes(Session session, QualifiedObjectName tableName, Map<String, Object> tableProperties)
-    {
-        throw new UnsupportedOperationException();
-    }
-
     @Override
     public Optional<TableHandle> getTableHandle(Session session, QualifiedObjectName table, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion)
     {
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 898abbe2a285..1c24573a267a 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
@@ -1555,20 +1555,6 @@ default Optional<CatalogSchemaTableName> redirectTable(ConnectorSession session,
         return Optional.empty();
     }
 
-    // TODO - Remove this method since now it is only used in test BaseConnectorTest#testWrittenDataSize()
-    @Deprecated
-    default boolean supportsReportingWrittenBytes(ConnectorSession session, SchemaTableName schemaTableName, Map<String, Object> tableProperties)
-    {
-        return false;
-    }
-
-    // TODO - Remove this method since now it is only used in test BaseConnectorTest#testWrittenDataSize()
-    @Deprecated
-    default boolean supportsReportingWrittenBytes(ConnectorSession session, ConnectorTableHandle connectorTableHandle)
-    {
-        return false;
-    }
-
     default OptionalInt getMaxWriterTasks(ConnectorSession session)
     {
         return OptionalInt.empty();
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 a9eb90752531..f4054e661093 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
@@ -1117,22 +1117,6 @@ public RowChangeParadigm getRowChangeParadigm(ConnectorSession session, Connecto
         }
     }
 
-    @Override
-    public boolean supportsReportingWrittenBytes(ConnectorSession session, SchemaTableName schemaTableName, Map<String, Object> tableProperties)
-    {
-        try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
-            return delegate.supportsReportingWrittenBytes(session, schemaTableName, tableProperties);
-        }
-    }
-
-    @Override
-    public boolean supportsReportingWrittenBytes(ConnectorSession session, ConnectorTableHandle connectorTableHandle)
-    {
-        try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
-            return delegate.supportsReportingWrittenBytes(session, connectorTableHandle);
-        }
-    }
-
     @Override
     public OptionalInt getMaxWriterTasks(ConnectorSession session)
     {
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 ab9a12178896..1f1858547e22 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
@@ -3054,18 +3054,6 @@ private static String toPhysicalColumnName(String columnName, Map</* lowercase*/
         return originalColumnName;
     }
 
-    @Override
-    public boolean supportsReportingWrittenBytes(ConnectorSession session, ConnectorTableHandle connectorTableHandle)
-    {
-        return true;
-    }
-
-    @Override
-    public boolean supportsReportingWrittenBytes(ConnectorSession session, SchemaTableName fullTableName, Map<String, Object> tableProperties)
-    {
-        return true;
-    }
-
     private void cleanExtraOutputFiles(ConnectorSession session, Location baseLocation, List<DataFileInfo> validDataFiles)
     {
         Set<Location> writtenFilePaths = validDataFiles.stream()
diff --git a/plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeConnectorTest.java b/plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeConnectorTest.java
index 7b4299e81695..7ead2b5816c2 100644
--- a/plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeConnectorTest.java
+++ b/plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeConnectorTest.java
@@ -162,6 +162,9 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_CREATE_MATERIALIZED_VIEW:
                 return false;
 
+            case SUPPORTS_REPORTING_WRITTEN_BYTES:
+                return true;
+
             default:
                 return super.hasBehavior(connectorBehavior);
         }
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 c16b7d0cdfab..38c23f6cddff 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
@@ -3902,16 +3902,4 @@ private static boolean isQueryPartitionFilterRequiredForTable(ConnectorSession s
         return isQueryPartitionFilterRequired(session) &&
                 requiredSchemas.isEmpty() || requiredSchemas.contains(schemaTableName.getSchemaName());
     }
-
-    @Override
-    public boolean supportsReportingWrittenBytes(ConnectorSession session, ConnectorTableHandle connectorTableHandle)
-    {
-        return true;
-    }
-
-    @Override
-    public boolean supportsReportingWrittenBytes(ConnectorSession session, SchemaTableName schemaTableName, Map<String, Object> tableProperties)
-    {
-        return true;
-    }
 }
diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/BaseHiveConnectorTest.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/BaseHiveConnectorTest.java
index 1855f8af65e3..9e58658d6186 100644
--- a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/BaseHiveConnectorTest.java
+++ b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/BaseHiveConnectorTest.java
@@ -279,6 +279,8 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
 
             case SUPPORTS_MULTI_STATEMENT_WRITES:
                 return true;
+            case SUPPORTS_REPORTING_WRITTEN_BYTES:
+                return true;
 
             default:
                 return super.hasBehavior(connectorBehavior);
diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java
index cd81cc8a2ed5..b44627563803 100644
--- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java
+++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java
@@ -2878,18 +2878,6 @@ private TableChangeInfo getTableChangeInfo(ConnectorSession session, IcebergTabl
                 .orElse(new UnknownTableChange());
     }
 
-    @Override
-    public boolean supportsReportingWrittenBytes(ConnectorSession session, ConnectorTableHandle connectorTableHandle)
-    {
-        return true;
-    }
-
-    @Override
-    public boolean supportsReportingWrittenBytes(ConnectorSession session, SchemaTableName fullTableName, Map<String, Object> tableProperties)
-    {
-        return true;
-    }
-
     @Override
     public void setColumnComment(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnHandle column, Optional<String> comment)
     {
diff --git a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java
index f5b58a0b2048..9fec165a10cb 100644
--- a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java
+++ b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java
@@ -235,6 +235,9 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
             case SUPPORTS_ADD_COLUMN_NOT_NULL_CONSTRAINT:
                 return false;
 
+            case SUPPORTS_REPORTING_WRITTEN_BYTES:
+                return true;
+
             default:
                 return super.hasBehavior(connectorBehavior);
         }
diff --git a/testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java b/testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java
index 51cf244b6e6f..d566126a54aa 100644
--- a/testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java
+++ b/testing/trino-testing/src/main/java/io/trino/testing/BaseConnectorTest.java
@@ -30,9 +30,7 @@
 import io.trino.metadata.FunctionManager;
 import io.trino.metadata.Metadata;
 import io.trino.metadata.QualifiedObjectName;
-import io.trino.security.AllowAllAccessControl;
 import io.trino.server.BasicQueryInfo;
-import io.trino.server.testing.TestingTrinoServer;
 import io.trino.spi.connector.ConnectorSession;
 import io.trino.spi.connector.MaterializedViewFreshness;
 import io.trino.spi.security.Identity;
@@ -145,6 +143,7 @@
 import static io.trino.testing.TestingConnectorBehavior.SUPPORTS_RENAME_SCHEMA;
 import static io.trino.testing.TestingConnectorBehavior.SUPPORTS_RENAME_TABLE;
 import static io.trino.testing.TestingConnectorBehavior.SUPPORTS_RENAME_TABLE_ACROSS_SCHEMAS;
+import static io.trino.testing.TestingConnectorBehavior.SUPPORTS_REPORTING_WRITTEN_BYTES;
 import static io.trino.testing.TestingConnectorBehavior.SUPPORTS_ROW_LEVEL_DELETE;
 import static io.trino.testing.TestingConnectorBehavior.SUPPORTS_ROW_TYPE;
 import static io.trino.testing.TestingConnectorBehavior.SUPPORTS_SET_COLUMN_TYPE;
@@ -155,7 +154,6 @@
 import static io.trino.testing.TestingNames.randomNameSuffix;
 import static io.trino.testing.assertions.Assert.assertEventually;
 import static io.trino.testing.assertions.TestUtil.verifyResultOrFailure;
-import static io.trino.transaction.TransactionBuilder.transaction;
 import static java.lang.String.format;
 import static java.lang.String.join;
 import static java.lang.Thread.currentThread;
@@ -5086,41 +5084,14 @@ public void testWrittenStats()
     @Test
     public void testWrittenDataSize()
     {
-        skipTestUnless(hasBehavior(SUPPORTS_CREATE_TABLE_WITH_DATA));
-
-        AtomicBoolean isReportingWrittenBytesSupported = new AtomicBoolean();
-        transaction(getQueryRunner().getTransactionManager(), getQueryRunner().getAccessControl())
-                .singleStatement()
-                .execute(getSession(), session -> {
-                    String catalogName = session.getCatalog().orElseThrow();
-                    TestingTrinoServer coordinator = getDistributedQueryRunner().getCoordinator();
-                    Map<String, Object> properties = coordinator.getTablePropertyManager().getProperties(
-                            catalogName,
-                            coordinator.getMetadata().getCatalogHandle(session, catalogName).orElseThrow(),
-                            List.of(),
-                            session,
-                            null,
-                            new AllowAllAccessControl(),
-                            Map.of(),
-                            true);
-                    QualifiedObjectName fullTableName = new QualifiedObjectName(catalogName, "any", "any");
-                    isReportingWrittenBytesSupported.set(coordinator.getMetadata().supportsReportingWrittenBytes(session, fullTableName, properties));
-                });
-
+        skipTestUnless(hasBehavior(SUPPORTS_REPORTING_WRITTEN_BYTES));
         String tableName = "write_stats_" + randomNameSuffix();
         try {
             String query = "CREATE TABLE " + tableName + " AS SELECT * FROM tpch.tiny.nation";
             assertQueryStats(
                     getSession(),
                     query,
-                    queryStats -> {
-                        if (isReportingWrittenBytesSupported.get()) {
-                            assertThat(queryStats.getPhysicalWrittenDataSize().toBytes()).isPositive();
-                        }
-                        else {
-                            assertThat(queryStats.getPhysicalWrittenDataSize().toBytes()).isZero();
-                        }
-                    },
+                    queryStats -> assertThat(queryStats.getPhysicalWrittenDataSize().toBytes()).isPositive(),
                     results -> {});
         }
         finally {
diff --git a/testing/trino-testing/src/main/java/io/trino/testing/TestingConnectorBehavior.java b/testing/trino-testing/src/main/java/io/trino/testing/TestingConnectorBehavior.java
index 9628dec7d67d..e0a0bbdead69 100644
--- a/testing/trino-testing/src/main/java/io/trino/testing/TestingConnectorBehavior.java
+++ b/testing/trino-testing/src/main/java/io/trino/testing/TestingConnectorBehavior.java
@@ -116,6 +116,8 @@ public enum TestingConnectorBehavior
 
     SUPPORTS_NATIVE_QUERY(true), // system.query or equivalent PTF for query passthrough
 
+    SUPPORTS_REPORTING_WRITTEN_BYTES(false),
+
     /**/;
 
     private final Predicate<Predicate<TestingConnectorBehavior>> hasBehaviorByDefault;
@@ -134,6 +136,7 @@ public enum TestingConnectorBehavior
                         (name().equals("SUPPORTS_CANCELLATION") ||
                                 name().equals("SUPPORTS_DYNAMIC_FILTER_PUSHDOWN") ||
                                 name().equals("SUPPORTS_JOIN_PUSHDOWN") ||
+                                name().equals("SUPPORTS_REPORTING_WRITTEN_BYTES") ||
                                 name().equals("SUPPORTS_MULTI_STATEMENT_WRITES")),
                 "Every behavior should be expected to be true by default. Having mixed defaults makes reasoning about tests harder. False default provided for %s",
                 name());