Skip to content
Merged
Show file tree
Hide file tree
Changes from 8 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 25 additions & 13 deletions src/datachain/catalog/catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
DatasetInvalidVersionError,
DatasetNotFoundError,
DatasetVersionNotFoundError,
NamespaceNotFoundError,
ProjectNotFoundError,
QueryScriptCancelError,
QueryScriptRunError,
Expand Down Expand Up @@ -1107,21 +1108,26 @@ def get_dataset_with_remote_fallback(
namespace_name: str,
project_name: str,
version: Optional[str] = None,
pull_dataset: bool = False,
update: bool = False,
) -> DatasetRecord:
try:
project = self.metastore.get_project(project_name, namespace_name)
ds = self.get_dataset(name, project)
if version and not ds.has_version(version):
raise DatasetVersionNotFoundError(
f"Dataset {name} does not have version {version}"
)
return ds
if self.metastore.is_local_dataset(namespace_name) or not update:
try:
project = self.metastore.get_project(project_name, namespace_name)
ds = self.get_dataset(name, project)
if not version or ds.has_version(version):
return ds
except (NamespaceNotFoundError, ProjectNotFoundError, DatasetNotFoundError):
pass

if self.metastore.is_local_dataset(namespace_name):
raise DatasetNotFoundError(
f"Dataset {name}"
+ (f" version {version} " if version else " ")
+ "not found"
)

except (
ProjectNotFoundError,
DatasetNotFoundError,
DatasetVersionNotFoundError,
):
if pull_dataset:
print("Dataset not found in local catalog, trying to get from studio")
remote_ds_uri = create_dataset_uri(
name, namespace_name, project_name, version
Expand All @@ -1136,6 +1142,8 @@ def get_dataset_with_remote_fallback(
name, self.metastore.get_project(project_name, namespace_name)
)

return self.get_remote_dataset(namespace_name, project_name, name)

def get_dataset_with_version_uuid(self, uuid: str) -> DatasetRecord:
"""Returns dataset that contains version with specific uuid"""
for dataset in self.ls_datasets():
Expand All @@ -1152,6 +1160,10 @@ def get_remote_dataset(

info_response = studio_client.dataset_info(namespace, project, name)
if not info_response.ok:
if info_response.status == 404:
raise DatasetNotFoundError(
f"Dataset {namespace}.{project}.{name} not found"
)
raise DataChainError(info_response.message)

dataset_info = info_response.data
Expand Down
39 changes: 34 additions & 5 deletions src/datachain/dataset.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,9 @@
)
from urllib.parse import urlparse

from packaging.specifiers import SpecifierSet
from packaging.version import Version

from datachain import semver
from datachain.error import DatasetVersionNotFoundError, InvalidDatasetNameError
from datachain.namespace import Namespace
Expand Down Expand Up @@ -661,13 +664,39 @@ def latest_major_version(self, major: int) -> Optional[str]:
return None
return max(versions).version

@property
def prev_version(self) -> Optional[str]:
"""Returns previous version of a dataset"""
if len(self.versions) == 1:
def latest_compatible_version(self, version_spec: str) -> Optional[str]:
"""
Returns the latest version that matches the given version specifier.

Supports Python version specifiers like:
- ">=1.0.0,<2.0.0" (compatible release range)
- "~=1.4.2" (compatible release clause)
- "==1.2.*" (prefix matching)
- ">1.0.0" (exclusive ordered comparison)
- ">=1.0.0" (inclusive ordered comparison)
- "!=1.3.0" (version exclusion)

Args:
version_spec: Version specifier string following PEP 440

Returns:
Latest compatible version string, or None if no compatible version found
"""
spec_set = SpecifierSet(version_spec)

# Convert dataset versions to packaging.Version objects
# and filter compatible ones
compatible_versions = []
for v in self.versions:
pkg_version = Version(v.version)
if spec_set.contains(pkg_version):
compatible_versions.append(v)

if not compatible_versions:
return None

return sorted(self.versions)[-2].version
# Return the latest compatible version
return max(compatible_versions).version
Comment thread
shcheklein marked this conversation as resolved.

@classmethod
def from_dict(cls, d: dict[str, Any]) -> "DatasetRecord":
Expand Down
80 changes: 45 additions & 35 deletions src/datachain/lib/dc/datasets.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,6 @@
ProjectNotFoundError,
)
from datachain.lib.dataset_info import DatasetInfo
from datachain.lib.file import (
File,
)
from datachain.lib.projects import get as get_project
from datachain.lib.settings import Settings
from datachain.lib.signal_schema import SignalSchema
Expand All @@ -34,7 +31,6 @@ def read_dataset(
version: Optional[Union[str, int]] = None,
session: Optional[Session] = None,
settings: Optional[dict] = None,
fallback_to_studio: bool = True,
delta: Optional[bool] = False,
delta_on: Optional[Union[str, Sequence[str]]] = (
"file.path",
Expand All @@ -44,6 +40,7 @@ def read_dataset(
delta_result_on: Optional[Union[str, Sequence[str]]] = None,
delta_compare: Optional[Union[str, Sequence[str]]] = None,
delta_retry: Optional[Union[bool, str]] = None,
update: bool = False,
) -> "DataChain":
"""Get data from a saved Dataset. It returns the chain itself.
If dataset or version is not found locally, it will try to pull it from Studio.
Expand All @@ -55,11 +52,12 @@ def read_dataset(
set; otherwise, default values will be applied.
namespace : optional name of namespace in which dataset to read is created
project : optional name of project in which dataset to read is created
version : dataset version
version : dataset version. Supports:
- Exact version strings: "1.2.3"
- Legacy integer versions: 1, 2, 3 (finds latest major version)
- Version specifiers (PEP 440): ">=1.0.0,<2.0.0", "~=1.4.2", "==1.2.*", etc.
session : Session to use for the chain.
settings : Settings to use for the chain.
fallback_to_studio : Try to pull dataset from Studio if not found locally.
Default is True.
delta: If True, only process new or changed files instead of reprocessing
everything. This saves time by skipping files that were already processed in
previous versions. The optimization is working when a new version of the
Expand All @@ -79,6 +77,8 @@ def read_dataset(
(error mode)
- True: Reprocess records missing from the result dataset (missing mode)
- None: No retry processing (default)
update: If True, it checks updates for the updates of the dataset on the Studio
Comment thread
shcheklein marked this conversation as resolved.
Outdated
side.

Example:
```py
Expand All @@ -92,11 +92,17 @@ def read_dataset(
```

```py
chain = dc.read_dataset("my_cats", fallback_to_studio=False)
chain = dc.read_dataset("my_cats", version="1.0.0")

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why don't we make this a part of dataset name? like my_cats@1.0.0 or package_name>=1.0,<2.0

So, we can give up a whole parameter from almost every API call 🙂

```

```py
chain = dc.read_dataset("my_cats", version="1.0.0")
# Using version specifiers (PEP 440)
chain = dc.read_dataset("my_cats", version=">=1.0.0,<2.0.0")
```

```py
# Legacy integer version support (finds latest in major version)
chain = dc.read_dataset("my_cats", version=1) # Latest 1.x.x version
```

```py
Expand All @@ -113,14 +119,15 @@ def read_dataset(
version="1.0.0",
session=session,
settings=settings,
fallback_to_studio=True,
Comment thread
shcheklein marked this conversation as resolved.
)
```
"""
from datachain.telemetry import telemetry

from .datachain import DataChain

telemetry.send_event_once("class", "datachain_init", name=name, version=version)

session = Session.get(session)
catalog = session.catalog

Expand All @@ -131,31 +138,37 @@ def read_dataset(
)

if version is not None:
dataset = session.catalog.get_dataset_with_remote_fallback(
name, namespace_name, project_name, update=update
)

# Convert legacy integer versions to version specifiers
# For backward compatibility we still allow users to put version as integer
# in which case we convert it to a version specifier that finds the latest
# version where major part is equal to that input version.
# For example if user sets version=2, we convert it to ">=2.0.0,<3.0.0"
# which will find something like 2.4.3 (assuming 2.4.3 is the biggest among
# all 2.* dataset versions)
if isinstance(version, int):
version_spec = f">={version}.0.0,<{version + 1}.0.0"
else:
version_spec = str(version)

from packaging.specifiers import InvalidSpecifier, SpecifierSet
Comment thread
shcheklein marked this conversation as resolved.

try:
# for backward compatibility we still allow users to put version as integer
# in which case we are trying to find latest version where major part is
# equal to that input version. For example if user sets version=2, we could
# continue with something like 2.4.3 (assuming 2.4.3 is the biggest among
# all 2.* dataset versions). If dataset doesn't have any versions where
# major part is equal to that input, exception is thrown.
major = int(version)
try:
ds_project = get_project(project_name, namespace_name, session=session)
except ProjectNotFoundError:
raise DatasetNotFoundError(
f"Dataset {name} not found in namespace {namespace_name} and",
f" project {project_name}",
) from None

dataset = session.catalog.get_dataset(name, ds_project)
latest_major = dataset.latest_major_version(major)
if not latest_major:
# Try to parse as version specifier
SpecifierSet(version_spec)
# If it's a valid specifier set, find the latest compatible version
latest_compatible = dataset.latest_compatible_version(version_spec)
if not latest_compatible:
raise DatasetVersionNotFoundError(
f"Dataset {name} does not have version {version}"
f"No dataset {name} version matching specifier {version_spec}"
)
version = latest_major
except ValueError:
# version is in new semver string format, continuing as normal
version = latest_compatible
except InvalidSpecifier:
# If not a valid specifier, treat as exact version string
# This handles cases like "1.2.3" which are exact versions, not specifiers
pass

if settings:
Expand All @@ -169,11 +182,8 @@ def read_dataset(
namespace_name=namespace_name,
version=version, # type: ignore[arg-type]
session=session,
indexing_column_types=File._datachain_column_types,
fallback_to_studio=fallback_to_studio,
)

telemetry.send_event_once("class", "datachain_init", name=name, version=version)
signals_schema = SignalSchema({"sys": Sys})
if query.feature_schema:
signals_schema |= SignalSchema.deserialize(query.feature_schema)
Expand Down
8 changes: 2 additions & 6 deletions src/datachain/lib/dc/listings.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,12 +127,8 @@ def read_listing_dataset(
if version is None:
version = dataset.latest_version

query = DatasetQuery(
name=name,
session=session,
indexing_column_types=File._datachain_column_types,
fallback_to_studio=False,
)
query = DatasetQuery(name=name, session=session)

if settings:
cfg = {**settings}
if "prefetch" not in cfg:
Expand Down
2 changes: 1 addition & 1 deletion src/datachain/lib/projects.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ def get(name: str, namespace: str, session: Optional[Session]) -> Project:
```py
import datachain as dc
from datachain.lib.projects import get as get_project
project = get_project("my-project", "local")
project = get_project("my-project", "local")
```
"""
return Session.get(session).catalog.metastore.get_project(name, namespace)
Expand Down
10 changes: 2 additions & 8 deletions src/datachain/query/dataset.py
Original file line number Diff line number Diff line change
Expand Up @@ -1099,13 +1099,9 @@ def __init__(
namespace_name: Optional[str] = None,
catalog: Optional["Catalog"] = None,
session: Optional[Session] = None,
indexing_column_types: Optional[dict[str, Any]] = None,
in_memory: bool = False,
fallback_to_studio: bool = True,
update: bool = False,
) -> None:
from datachain.remote.studio import is_token_set

self.session = Session.get(session, catalog=catalog, in_memory=in_memory)
self.catalog = catalog or self.session.catalog
self.steps: list[Step] = []
Expand Down Expand Up @@ -1137,18 +1133,16 @@ def __init__(
# not setting query step yet as listing dataset might not exist at
# this point
self.list_ds_name = name
elif fallback_to_studio and is_token_set():
else:
self._set_starting_step(
self.catalog.get_dataset_with_remote_fallback(
name,
namespace_name=namespace_name,
project_name=project_name,
version=version,
pull_dataset=True,
)
)
else:
project = self.catalog.metastore.get_project(project_name, namespace_name)
self._set_starting_step(self.catalog.get_dataset(name, project=project))

def _set_starting_step(self, ds: "DatasetRecord") -> None:
if not self.version:
Expand Down
7 changes: 4 additions & 3 deletions src/datachain/remote/studio.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,10 +78,11 @@ def _parse_dates(obj: dict, date_fields: list[str]):


class Response(Generic[T]):
def __init__(self, data: T, ok: bool, message: str) -> None:
def __init__(self, data: T, ok: bool, message: str, status: int) -> None:
self.data = data
self.ok = ok
self.message = message
self.status = status

def __repr__(self):
return (
Expand Down Expand Up @@ -186,7 +187,7 @@ def _send_request_msgpack(
message = "Indexing in progress"
else:
message = content.get("message", "")
return Response(response_data, ok, message)
return Response(response_data, ok, message, response.status_code)

@retry_with_backoff(retries=3, errors=(HTTPError, Timeout))
def _send_request(
Expand Down Expand Up @@ -236,7 +237,7 @@ def _send_request(
else:
message = ""

return Response(data, ok, message)
return Response(data, ok, message, response.status_code)

@staticmethod
def _unpacker_hook(code, data):
Expand Down
6 changes: 2 additions & 4 deletions tests/func/test_dataset_query.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,7 @@
import sqlalchemy

from datachain.dataset import DatasetDependencyType, DatasetStatus
from datachain.error import (
DatasetVersionNotFoundError,
)
from datachain.error import DatasetNotFoundError
from datachain.lib.listing import parse_listing_uri
from datachain.query import C, DatasetQuery, Object, Stream
from datachain.sql.functions import path as pathfunc
Expand Down Expand Up @@ -70,7 +68,7 @@ def test_save_multiple_versions(cloud_test_catalog, animal_dataset):
assert DatasetQuery(name=ds_name, version="1.0.1", catalog=catalog).count() == 3
assert DatasetQuery(name=ds_name, version="1.0.2", catalog=catalog).count() == 3

with pytest.raises(DatasetVersionNotFoundError):
with pytest.raises(DatasetNotFoundError):
DatasetQuery(name=ds_name, version="4.0.0", catalog=catalog).count()


Expand Down
Loading