Remove size_bytes_on_disk from Python

daft/daft.pyi

@@ -707,12 +707,6 @@ class ScanTask:
         """
         ...
 
-    def size_bytes_on_disk(self) -> int:
-        """
-        Get number of bytes that will be scanned by this ScanTask.
-        """
-        ...
-
     def estimate_in_memory_size_bytes(self, cfg: PyDaftExecutionConfig) -> int:
         """
         Estimate the In Memory Size of this ScanTask.

daft/execution/execution_step.py

@@ -5,6 +5,7 @@
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Generic, Protocol
 
+from daft.context import get_context
 from daft.daft import FileFormat, IOConfig, JoinType, ResourceRequest, ScanTask
 from daft.expressions import Expression, ExpressionsProjection, col
 from daft.logical.map_partition_ops import MapPartitionOp
@@ -304,10 +305,12 @@ def _scan(self, inputs: list[MicroPartition]) -> list[MicroPartition]:
     def run_partial_metadata(self, input_metadatas: list[PartialPartitionMetadata]) -> list[PartialPartitionMetadata]:
         assert len(input_metadatas) == 0
 
+        cfg = get_context().daft_execution_config
+
         return [
             PartialPartitionMetadata(
                 num_rows=self.scan_task.num_rows(),
-                size_bytes=self.scan_task.size_bytes_on_disk(),
+                size_bytes=self.scan_task.estimate_in_memory_size_bytes(cfg),
             )
         ]
 

src/daft-scan/src/python.rs

@@ -285,10 +285,6 @@ pub mod pylib {
             Ok(self.0.num_rows().map(i64::try_from).transpose()?)
         }
 
-        pub fn size_bytes_on_disk(&self) -> PyResult<Option<i64>> {
-            Ok(self.0.size_bytes_on_disk().map(i64::try_from).transpose()?)
-        }
-
         pub fn estimate_in_memory_size_bytes(
             &self,
             cfg: PyDaftExecutionConfig,