chore!: Remove DataSourceTask.get_micro_partitions

Cargo.toml

@@ -263,7 +263,6 @@ arrow-schema = "59.0.0"
 arrow-select = "59.0.0"
 approx = "0.5.1"
 arc-swap = "1.7.1"
-async-compat = "0.2.5"
 async-compression = {version = "0.4.43", features = [
   "tokio",
   "brotli",
@@ -351,21 +350,20 @@ opentelemetry-otlp = {version = "0.33", features = ["grpc-tonic", "logs"]}
 opentelemetry_sdk = {version = "0.33", features = ["logs"]}
 parking_lot = "0.12.5"
 parquet = "59.0.0"
-path_macro = "1.0.0"
 pretty_assertions = "1.4.1"
 proptest = "1.9.0"
 prost = "0.13.5"
 prost-types = "0.13.5"
 rand = "^0.9"
 rayon = "1.12.0"
-regex = "1.12.2"
+regex = "1.13"
 rstest = "0.26.1"
 reqwest = {version = "0.13", default-features = false, features = ["json"]}
 serde_json = "1.0.145"
 sha1 = "^0.11"
 simdutf8 = "0.1.5"
 sketches-ddsketch = {version = "0.2.2", features = ["use_serde"]}
-smallvec = {version = "1.15.1", features = ["serde", "impl_bincode"]}
+smallvec = {version = "1.15.1", features = ["serde"]}
 snafu = {version = "0.9.0", features = ["futures"]}
 sqlparser = "0.59.0"
 sysinfo = "0.39"
@@ -383,7 +381,7 @@ tokio = {version = "1.48.0", features = [
   "rt",
   "rt-multi-thread"
 ]}
-tokio-stream = {version = "0.1.14", features = ["fs", "io-util", "time"]}
+tokio-stream = {version = "0.1.14", default-features = false, features = ["fs", "io-util"]}
 tokio-util = "0.7.17"
 tonic = "0.14.2"
 tonic-build = "0.14.2"
@@ -393,7 +391,7 @@ typetag = "0.2.21"
 rustfft = "6"
 url = "2.4.0"
 uuid = {version = "1.19.0", features = ["v4", "v7", "serde", "fast-rng"]}
-xxhash-rust = {version = "0.8.15", features = ["const_xxh3", "const_xxh64", "const_xxh32", "xxh64", "xxh3", "xxh32"]}
+xxhash-rust = {version = "0.8.15", features = ["const_xxh3", "xxh64", "xxh3", "xxh32"]}
 libc = {version = "^0.2.150", default-features = false}
 hex = "0.4.3"
 
@@ -402,7 +400,7 @@ version = "2.0.1"
 features = ["derive", "serde"]
 
 [workspace.dependencies.derive_more]
-features = ["display", "from"]
+features = ["display"]
 version = "2.1.0"
 
 [workspace.dependencies.log]

daft/file/file.py

@@ -143,7 +143,7 @@ def position(self) -> int | None:
 
     @property
     def offset(self) -> int | None:
-        """Deprecated alias for `position`. The byte offset for range reads, or None for full-file reads."""
+        """Deprecated alias for `position`. The byte offset for range reads, or None for full-file reads. Will be removed in v0.9.0."""
         warnings.warn(
             "`File.offset` is deprecated; use `File.position` instead.",
             DeprecationWarning,
@@ -153,7 +153,7 @@ def offset(self) -> int | None:
 
     @property
     def length(self) -> int | None:
-        """Deprecated alias for the byte-range read window size, or None for full-file reads.
+        """Deprecated alias for the byte-range read window size, or None for full-file reads Will be removed in v0.9.0.
 
         Note: this returns the requested range size (caller intent), not the derived file
         size. Use `File.size()` for the actual file size.

daft/io/source.py

@@ -1,20 +1,18 @@
 from __future__ import annotations
 
-import warnings
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING
 
 from daft.daft import McapSourceConfig, ParquetSourceConfig, PyDataSourceTask, StorageConfig
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator, Iterator
+    from collections.abc import AsyncIterator
 
     from daft.dataframe import DataFrame
     from daft.io.clustering import ClusteringKeys
     from daft.io.partitioning import PartitionField
     from daft.io.pushdowns import Pushdowns
     from daft.recordbatch import RecordBatch
-    from daft.recordbatch.micropartition import MicroPartition
     from daft.schema import Schema
 
 
@@ -124,31 +122,16 @@ def schema(self) -> Schema:
         """Returns the schema of the record batches produced by this task."""
         ...
 
+    @abstractmethod
     async def read(self) -> AsyncIterator[RecordBatch]:
         """Yields record batches. Called from an async execution context.
 
-        The default implementation delegates to the deprecated
-        get_micro_partitions for backwards compatibility.
-        New subclasses should override this method directly.
+        Subclasses must override this method. Each yielded batch must match :attr:`schema`.
         """
-        try:
-            parts = self.get_micro_partitions()
-        except NotImplementedError:
-            raise NotImplementedError(
-                f"{type(self).__name__} must implement async def read(self) -> AsyncIterator[RecordBatch]"
-            )
-        warnings.warn(
-            f"{type(self).__name__}.get_micro_partitions() is deprecated — override 'async def read()' instead.",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-        for mp in parts:
-            for rb in mp.get_record_batches():
-                yield rb
-
-    def get_micro_partitions(self) -> Iterator[MicroPartition]:
-        """Deprecated: override read instead."""
-        raise NotImplementedError
+        ...
+        # https://mypy.readthedocs.io/en/latest/more_types.html#typing-async-generators
+        # This tricks the type checker into treating this as an async generator.
+        yield  # type: ignore[misc]
 
     @staticmethod
     def parquet(

src/daft-csv/Cargo.toml

@@ -1,7 +1,6 @@
 [dependencies]
 log = {workspace = true}
 arrow-schema = {workspace = true}
-async-compat = {workspace = true}
 async-stream = {workspace = true}
 common-error = {path = "../common/error", default-features = false}
 common-py-serde = {path = "../common/py-serde", default-features = false}
@@ -24,7 +23,7 @@ serde = {workspace = true}
 smallvec = {workspace = true}
 snafu = {workspace = true}
 tokio = {workspace = true}
-tokio-util = {workspace = true}
+tokio-util = {workspace = true, features = ["compat"]}
 
 [dev-dependencies]
 rstest = {workspace = true}

src/daft-csv/src/metadata.rs

@@ -1,7 +1,6 @@
 use std::{collections::HashSet, sync::Arc};
 
 use arrow_schema::{DataType, Field, Schema as ArrowSchema};
-use async_compat::CompatExt;
 use common_error::DaftResult;
 use csv_async::{AsyncReaderBuilder, ByteRecord};
 use daft_compression::CompressionCodec;
@@ -14,7 +13,7 @@ use tokio::{
     fs::File,
     io::{AsyncBufRead, AsyncRead, BufReader},
 };
-use tokio_util::io::StreamReader;
+use tokio_util::{compat::TokioAsyncReadCompatExt, io::StreamReader};
 
 use crate::{CsvParseOptions, schema::merge_schema};
 

src/daft-csv/src/read.rs

@@ -2,7 +2,6 @@ use core::str;
 use std::{collections::HashMap, num::NonZeroUsize, sync::Arc};
 
 use arrow_schema::Field as ArrowField;
-use async_compat::{Compat, CompatExt};
 use common_error::{DaftError, DaftResult};
 use common_runtime::get_io_runtime;
 use csv_async::{AsyncReader, AsyncReaderBuilder};
@@ -26,7 +25,10 @@ use tokio::{
     io::{AsyncBufRead, AsyncRead, BufReader},
     task::JoinHandle,
 };
-use tokio_util::io::StreamReader;
+use tokio_util::{
+    compat::{Compat, TokioAsyncReadCompatExt},
+    io::StreamReader,
+};
 
 use crate::{CsvConvertOptions, CsvParseOptions, CsvReadOptions, metadata::read_csv_schema_single};
 

src/daft-io/Cargo.toml

@@ -9,7 +9,7 @@ aws-smithy-http-client = "1.1.2"
 aws-smithy-runtime-api = "1.8.0"
 azure_core = {version = "1.1.0", features = ["reqwest", "reqwest_rustls", "tokio", "hmac_rust", "xml"], default-features = false}
 azure_identity = {version = "1.0.0", features = ["tokio"], default-features = false}
-azure_storage_blob = {version = "1.1.0-beta.1", default-features = false}
+azure_storage_blob = {version = "1.1.0", default-features = false}
 bytes = {workspace = true}
 common-error = {path = "../common/error", default-features = false}
 common-file-formats = {path = "../common/file-formats", default-features = false}
@@ -38,7 +38,7 @@ opendal = {version = "0.58", default-features = false, features = [
 ]}
 pyo3 = {workspace = true, optional = true}
 rand = {workspace = true}
-regex = "1.12.2"
+regex = {workspace = true}
 reqwest = {workspace = true, default-features = false, features = ["stream", "json"]}
 reqwest-middleware = "0.5"
 reqwest-retry = "0.9"