diff --git a/CHANGELOG.md b/CHANGELOG.md
index 91815ea..e4c52e3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,8 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.68.0] - 2026-10-29
+
 ### Added
 
+- `tilebox-datasets`: Support scalar and repeated custom enums in schema definitions with `"type": "enum"`,
+  `enum_values`, and optional `"repeated": True`. Ingest and query case-sensitive public strings through
+  sync and async clients, including pandas and xarray inputs.
+- `tilebox-datasets`: Add `field(...).contains()`, `.contains_all()`, and `.contains_any()` filters for
+  queryable repeated enums. Empty `contains_all([])` matches every row; `contains_any([])` matches none.
+  Negated membership includes untagged rows. Requires a backend supporting enum membership predicates.
 - `tilebox-workflows`: Add `workflow` selection by reference or object to job submission and automation creation,
   and `workflows` filters to job queries. Omitting scope preserves unscoped direct-runner behavior.
 - `tilebox-workflows`: Preserve task workflow IDs and inherit the executing parent's scope for children.
@@ -18,6 +26,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- `tilebox-datasets`: Return repeated enums, including built-ins such as SAR polarizations, as one
+  `list[str]` per row in xarray object columns, preserving the returned order. Remove their extra
+  `n_<field>` dimension and retain empty values as `[]`, even when the entire returned column is empty.
+  Update code that indexes the former array dimension to access the per-row list instead.
+  Other repeated fields keep their padded arrays.
 - `tilebox-workflows`: Always create scoped-only workflows with legacy task aliases disabled.
 - `tilebox-workflows`: Use `UnboundTaskIdentifier` for task metadata, runner registries, and release contents;
   keep execution scope in `TaskIdentifier` and resolve each future task's scope before merging submissions.
@@ -716,7 +729,8 @@ the first client that does not cache data (since it's already on the local file
 - Released under the [MIT](https://opensource.org/license/mit) license.
 - Released packages: `tilebox-datasets`, `tilebox-workflows`, `tilebox-storage`, `tilebox-grpc`
 
-[Unreleased]: https://github.com/tilebox/tilebox-python/compare/v0.67.0...HEAD
+[Unreleased]: https://github.com/tilebox/tilebox-python/compare/v0.68.0...HEAD
+[0.68.0]: https://github.com/tilebox/tilebox-python/compare/v0.67.0...v0.68.0
 [0.67.0]: https://github.com/tilebox/tilebox-python/compare/v0.66.0...v0.67.0
 [0.65.0]: https://github.com/tilebox/tilebox-python/compare/v0.64.0...v0.65.0
 [0.64.0]: https://github.com/tilebox/tilebox-python/compare/v0.63.2...v0.64.0
diff --git a/tilebox-datasets/README.md b/tilebox-datasets/README.md
index 78cb19d..18fc154 100644
--- a/tilebox-datasets/README.md
+++ b/tilebox-datasets/README.md
@@ -87,6 +87,97 @@ results = s2a_l1c.query(
 print(f"Found {results.sizes['time']} datapoints")  # Found 979 datapoints
 ```
 
+## Enum fields
+
+Define a scalar enum with an ordered list of public strings:
+
+```python
+from datetime import UTC, datetime
+
+from tilebox.datasets import field
+from tilebox.datasets.data.datasets import DatasetKind
+
+dataset = client.create_or_update_dataset(
+    DatasetKind.TEMPORAL,
+    "scenes",
+    fields=[
+        {"name": "processing_level", "type": "enum", "enum_values": ["L1", "L2-SP", "雪"], "queryable": True},
+    ],
+)
+collection = dataset.get_or_create_collection("example")
+collection.ingest(
+    [
+        {"time": datetime(2026, 10, 28, 12, tzinfo=UTC), "processing_level": "L2-SP"},
+        {"time": datetime(2026, 10, 28, 13, tzinfo=UTC)},
+    ]
+)
+
+results = collection.query(
+    temporal_extent=("2026-10-28", "2026-10-29"),
+    filter=field("processing_level") == "L2-SP",
+)
+assert results.processing_level.to_numpy().tolist() == ["L2-SP"]
+```
+
+Ingestion accepts public strings in records, columns, pandas DataFrames, and xarray Datasets.
+Query results contain public strings in object arrays. Missing values remain missing; use xarray's
+`isnull()` to identify them. You do not need a Python enum class to create or read a dataset.
+Numbers and generated protobuf symbols are not accepted as substitutes for public values.
+
+Scalar enum filters support `==` and `!=`. Combine equality expressions with `|` to match several values.
+The server validates values case-sensitively. Comparisons exclude missing values, including `!=`;
+include them explicitly with `(field("processing_level") != "L1") | field("processing_level").is_null()`.
+
+Vocabularies contain 1–100 unique, nonempty strings. Empty datasets allow replacing, removing,
+renaming, or reordering values. Once a dataset contains data, updates may only append values.
+Supply the full field schema when updating a dataset, and re-fetch the dataset after another client
+updates its vocabulary. The asynchronous client provides the same enum behavior.
+
+### Repeated enums
+
+Add `"repeated": True` to define lists of enum values, including queryable tags:
+
+```python
+fields = [
+    {
+        "name": "use_cases",
+        "type": "enum",
+        "enum_values": ["agriculture", "defense", "wind-power"],
+        "repeated": True,
+        "queryable": True,
+    },
+]
+dataset = client.create_or_update_dataset(DatasetKind.TEMPORAL, "tagged_scenes", fields)
+collection = dataset.get_or_create_collection("example")
+collection.ingest(
+    [
+        {"time": datetime(2026, 10, 28, 12, tzinfo=UTC), "use_cases": ["defense", "agriculture", "defense"]},
+    ]
+)
+results = collection.query(
+    temporal_extent=("2026-10-28", "2026-10-29"),
+    filter=field("use_cases").contains("defense"),
+)
+
+# Alternative membership filters:
+all_tags = field("use_cases").contains_all(["defense", "agriculture"])
+any_tag = field("use_cases").contains_any(["defense", "wind-power"])
+without_defense = ~field("use_cases").contains("defense")
+```
+
+The backend removes duplicates during ingestion, preserving first-occurrence order. The example
+stores `["defense", "agriculture"]`. Ingest lists of public strings through records, columns, or pandas;
+xarray results contain one `list[str]` per row in an object column, `use_cases(time)`, without an
+extra array dimension. Empty values are `[]`, including when every returned value is empty.
+Lists preserve the returned order and can be ingested again without changing enum ordering.
+Other repeated fields retain their existing padded array layout.
+
+Repeated enums use membership filters, not `==`, `!=`, or null checks. Combine predicates with `&`, `|`,
+and `~`. Membership treats omitted and empty lists as the empty set and never returns null.
+Consequently, `without_defense` includes untagged rows. `contains_all([])` matches every row;
+`contains_any([])` matches none. Each predicate accepts up to 100 nonempty public strings, counting
+duplicates toward that limit. The server checks vocabulary membership and whether the field is queryable.
+
 ## Documentation
 
 Check out the [Tilebox Datasets documentation](https://docs.tilebox.com/datasets/introduction) for more information.
diff --git a/tilebox-datasets/tests/data/datasets.py b/tilebox-datasets/tests/data/datasets.py
index c9b22ec..3d88906 100644
--- a/tilebox-datasets/tests/data/datasets.py
+++ b/tilebox-datasets/tests/data/datasets.py
@@ -46,7 +46,10 @@ def field_annotations(draw: DrawFn) -> FieldAnnotation:
     queryable = draw(booleans())
     json_schema_ref = draw(text(alphabet=string.ascii_letters + string.digits + "#/_-", max_size=25) | none())
     roles = draw(lists(sampled_from(FieldRole), unique=True))
-    return FieldAnnotation(description, example_value, source_json_pointer, queryable, json_schema_ref, roles)
+    enum_values = draw(lists(text(min_size=1, max_size=20), unique=True, max_size=5))
+    return FieldAnnotation(
+        description, example_value, source_json_pointer, queryable, json_schema_ref, roles, enum_values
+    )
 
 
 @composite
diff --git a/tilebox-datasets/tests/data/test_datasets.py b/tilebox-datasets/tests/data/test_datasets.py
index 9e97b78..6622c35 100644
--- a/tilebox-datasets/tests/data/test_datasets.py
+++ b/tilebox-datasets/tests/data/test_datasets.py
@@ -1,6 +1,6 @@
 from collections.abc import Callable
 from dataclasses import replace
-from typing import Any, Literal
+from typing import Any, Literal, cast
 from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
@@ -128,6 +128,35 @@ def test_message_types_can_define_dataset_fields(message_type: type, message_nam
     assert repeated.descriptor.label == FieldDescriptorProto.Label.REPEATED
 
 
+@pytest.mark.parametrize(
+    "definition",
+    [
+        {"type": "enum"},
+        {"type": "enum", "enum_values": []},
+        {"type": "enum", "enum_values": [""]},
+        {"type": "enum", "enum_values": ["L1", "L1"]},
+        {"type": "enum", "enum_values": [1]},
+        {"type": "enum", "enum_values": "L1"},
+        {"type": "enum", "enum_values": [str(i) for i in range(101)]},
+        {"type": str, "repeated": True},
+        {"type": str, "enum_values": []},
+    ],
+)
+def test_invalid_enum_definitions(definition: dict[str, Any]) -> None:
+    with pytest.raises(ValueError, match="Field 'level'"):
+        Field.from_dict(cast(FieldDict, {"name": "level", **definition}))
+
+
+@pytest.mark.parametrize("repeated", [False, True])
+def test_enum_vocabulary_limit(repeated: bool) -> None:
+    values = [str(i) for i in range(100)]
+    definition: FieldDict = {"name": "level", "type": "enum", "enum_values": values, "repeated": repeated}
+    assert Field.from_dict(definition).annotation.enum_values == values
+    values.append("100")
+    with pytest.raises(ValueError, match="at most 100"):
+        Field.from_dict(definition)
+
+
 @given(fields())
 def test_fields_to_message_and_back(field: Field) -> None:
     assert Field.from_message(field.to_message()) == field
@@ -159,7 +188,8 @@ def test_list_datasets_responses_to_message_and_back(response: ListDatasetsRespo
 
 
 @pytest.mark.parametrize("operation", ["create", "update"])
-def test_create_and_update_dataset_include_field_annotations(operation: str) -> None:
+@pytest.mark.parametrize("repeated", [False, True])
+def test_create_and_update_dataset_include_field_annotations(operation: str, repeated: bool) -> None:
     """Public dataset writes send required and custom field annotations to native clients."""
     dataset_service = MagicMock()
     if operation == "create":
@@ -172,7 +202,14 @@ def test_create_and_update_dataset_include_field_annotations(operation: str) ->
             "queryable": True,
             "json_schema_ref": "https://example.com/schema.json#/title",
             "roles": [FieldRole.PRIMARY_TITLE],
-        }
+        },
+        {
+            "name": "level",
+            "type": "enum",
+            "enum_values": ["L1", "L2-SP", "雪"],
+            "queryable": True,
+            "repeated": repeated,
+        },
     ]
 
     with (
@@ -195,9 +232,20 @@ def test_create_and_update_dataset_include_field_annotations(operation: str) ->
     assert annotations["geometry"].source_json_pointer == "/geometry"
     assert annotations["geometry"].json_schema_ref == "https://geojson.org/schema/Geometry.json"
     assert all(
-        not annotation.queryable and not annotation.roles for name, annotation in annotations.items() if name != "title"
+        not annotation.queryable and not annotation.roles
+        for name, annotation in annotations.items()
+        if name not in {"title", "level"}
     )
     assert annotations["title"].source_json_pointer == "/properties/title"
     assert annotations["title"].queryable
     assert annotations["title"].json_schema_ref == "https://example.com/schema.json#/title"
     assert list(annotations["title"].roles) == [FieldRole.PRIMARY_TITLE.value]
+    enum_field = request.type.fields[-1]
+    assert enum_field.descriptor.type == FieldDescriptorProto.Type.ENUM
+    assert enum_field.descriptor.label == (
+        FieldDescriptorProto.Label.REPEATED if repeated else FieldDescriptorProto.Label.OPTIONAL
+    )
+    assert not enum_field.descriptor.type_name
+    assert enum_field.annotation.queryable
+    assert [entry.value for entry in enum_field.annotation.enum_values] == ["L1", "L2-SP", "雪"]
+    assert all(field.descriptor.options is None for field in request.type.fields)
diff --git a/tilebox-datasets/tests/protobuf_conversion/test_protobuf_xarray.py b/tilebox-datasets/tests/protobuf_conversion/test_protobuf_xarray.py
index 807fea0..ffc727f 100644
--- a/tilebox-datasets/tests/protobuf_conversion/test_protobuf_xarray.py
+++ b/tilebox-datasets/tests/protobuf_conversion/test_protobuf_xarray.py
@@ -1,4 +1,5 @@
 from enum import Enum
+from typing import Any
 from uuid import UUID
 
 import numpy as np
@@ -343,7 +344,7 @@ class Polarization(Enum):
     message_type = get_message_type(
         AnnotatedType(FileDescriptorSet(file=[file_descriptor]), "tests.protobuf_conversion.enums.EnumDatapoint", [])
     )
-    messages = [
+    messages: list[Any] = [
         message_type(primary_polarization=1, polarizations=[0, 1]),
         message_type(primary_polarization=2, polarizations=[2]),
         message_type(),
@@ -357,11 +358,8 @@ class Polarization(Enum):
     assert dataset.primary_polarization[:2].to_numpy().tolist() == ["HH", "VV"]
     assert pd.isna(dataset.primary_polarization[2].item())
     assert dataset.polarizations.dtype == object
-    assert dataset.polarizations[0].to_numpy().tolist() == ["UNSPECIFIED", "HH"]
-    assert dataset.polarizations[1, 0].item() == "VV"
-    assert pd.isna(dataset.polarizations[1, 1].item())
-    assert pd.isna(dataset.polarizations[2].to_numpy()).all()
-    assert dataset.polarizations.dims == ("time", "n_polarizations")
+    assert dataset.polarizations.to_numpy().tolist() == [["UNSPECIFIED", "HH"], ["VV"], []]
+    assert dataset.polarizations.dims == ("time",)
     assert dataset.primary_polarization.attrs == {}
     assert dataset.polarizations.attrs == {}
     assert to_messages(dataset, message_type) == messages
@@ -387,10 +385,8 @@ def test_convert_sar_polarizations_with_short_names_in_both_directions() -> None
 
     dataset = converter.finalize("item")
 
-    assert dataset.polarizations[0].to_numpy().tolist() == ["HH", "VV"]
-    assert dataset.polarizations[1, 0].item() == "VV"
-    assert pd.isna(dataset.polarizations[1, 1].item())
-    assert pd.isna(dataset.polarizations[2].to_numpy()).all()
+    assert dataset.polarizations.dims == ("item",)
+    assert dataset.polarizations.to_numpy().tolist() == [["HH", "VV"], ["VV"], []]
 
     other_fields = [field.name for field in SARProperties.desc().fields if field.name != "polarizations"]
     assert to_messages(dataset, SARProperties, ignore_fields=other_fields) == messages
diff --git a/tilebox-datasets/tests/protobuf_conversion/test_to_protobuf.py b/tilebox-datasets/tests/protobuf_conversion/test_to_protobuf.py
index f1302e9..7d80fcc 100644
--- a/tilebox-datasets/tests/protobuf_conversion/test_to_protobuf.py
+++ b/tilebox-datasets/tests/protobuf_conversion/test_to_protobuf.py
@@ -105,6 +105,19 @@ def test_record_oriented_data_preserves_absent_fields_and_filters_missing_values
     assert converted[1].some_int == 0
 
 
+@pytest.mark.parametrize(
+    ("values", "expected"), [([], []), ([np.nan], []), ([None, np.nan], []), ([3, np.nan, 7], [3, 7])]
+)
+@pytest.mark.parametrize("as_array", [False, True])
+def test_required_repeated_field_is_not_a_missing_scalar(
+    values: list[float | None], expected: list[int], as_array: bool
+) -> None:
+    value = np.array(values) if as_array else values
+    converted = to_messages([{"some_repeated_int": value}], ExampleDatapoint, required_fields=["some_repeated_int"])
+
+    assert converted[0].some_repeated_int == expected
+
+
 def test_record_oriented_data_requires_every_record_to_have_required_fields() -> None:
     with pytest.raises(ValueError, match=r"Record 1: Missing required field.*time"):
         to_messages(
diff --git a/tilebox-datasets/tests/query/test_expression.py b/tilebox-datasets/tests/query/test_expression.py
index fb3b215..3d439ae 100644
--- a/tilebox-datasets/tests/query/test_expression.py
+++ b/tilebox-datasets/tests/query/test_expression.py
@@ -1,6 +1,6 @@
 from datetime import UTC, datetime, timedelta
-from enum import Enum, IntEnum
-from typing import Literal
+from enum import Enum, IntEnum, StrEnum
+from typing import Any, Literal
 
 import numpy as np
 import pytest
@@ -17,6 +17,15 @@ class NumericQuality(IntEnum):
     GOOD = 1
 
 
+class ProcessingLevel(StrEnum):
+    UPPER = "L1"
+    LOWER = "l1"
+
+
+class StringQuality(str, Enum):  # noqa: UP042 -- Exercise Enum.__str__, unlike StrEnum.
+    GOOD = "good"
+
+
 @pytest.mark.parametrize(
     ("value", "kind", "expected"),
     [
@@ -26,18 +35,23 @@ class NumericQuality(IntEnum):
         (0.0, "double_value", 0.0),
         ("", "string_value", ""),
         (b"", "bytes_value", b""),
-        (Quality.GOOD, "enum_name", "GOOD"),
-        (NumericQuality.GOOD, "enum_name", "GOOD"),
+        (Quality.GOOD.value, "string_value", "good"),
+        (NumericQuality.GOOD, "int64_value", 1),
+        (ProcessingLevel.UPPER, "string_value", "L1"),
+        (ProcessingLevel.LOWER, "string_value", "l1"),
+        (StringQuality.GOOD, "string_value", "good"),
+        (np.str_("L2-SP"), "string_value", "L2-SP"),
+        ("L2-SP", "string_value", "L2-SP"),
+        ("雪", "string_value", "雪"),
     ],
 )
 def test_query_value_types(
     value: object,
-    kind: Literal[
-        "bool_value", "int64_value", "uint64_value", "double_value", "string_value", "bytes_value", "enum_name"
-    ],
+    kind: Literal["bool_value", "int64_value", "uint64_value", "double_value", "string_value", "bytes_value"],
     expected: object,
 ) -> None:
-    comparison = (field("quality") == value).to_message().comparison
+    wire = (field("quality") == value).to_message().to_binary()
+    comparison = data_access_pb.FilterExpression.from_binary(wire).comparison
     assert comparison is not None
     message = comparison.value
     assert message is not None
@@ -45,6 +59,69 @@ def test_query_value_types(
     assert getattr(message, kind) == expected
 
 
+def test_unsupported_query_value() -> None:
+    with pytest.raises(TypeError, match="Unsupported query value type"):
+        _ = field("quality") == Quality.GOOD
+
+
+@pytest.mark.parametrize("method", ["contains", "contains_all", "contains_any"])
+@pytest.mark.parametrize("value", [ProcessingLevel.UPPER, ProcessingLevel.LOWER])
+def test_string_subclass_membership(method: str, value: str) -> None:
+    expression = getattr(field("tags"), method)(value if method == "contains" else [value])
+    message = data_access_pb.FilterExpression.from_binary(expression.to_message().to_binary())
+    assert message.array_predicate is not None
+    assert list(message.array_predicate.values) == [value]
+    assert Expression.from_message(message) == expression
+
+
+@pytest.mark.parametrize("method", ["contains_all", "contains_any"])
+@pytest.mark.parametrize("values", [[], ["defense"], ["wind-power", "雪", "defense", "defense"], ["defense"] * 100])
+def test_array_predicates_round_trip(method: str, values: list[str]) -> None:
+    expression = getattr(field("use_cases"), method)(values)
+    message = expression.to_message()
+    predicate = message.array_predicate
+    assert predicate is not None
+    assert predicate.field_name == "use_cases"
+    assert predicate.operator == (
+        data_access_pb.FieldArrayOperator.CONTAINS_ALL
+        if method == "contains_all"
+        else data_access_pb.FieldArrayOperator.OVERLAPS
+    )
+    assert list(predicate.values) == values
+    assert Expression.from_message(message) == expression
+    combined = ~(expression | field("use_cases").contains("agriculture")) & (field("quality") == "good")
+    assert Expression.from_message(combined.to_message()) == combined
+    assert field("use_cases").contains("defense") == field("use_cases").contains_all(["defense"])
+
+
+@pytest.mark.parametrize("values", ["defense", None, [1], [""], [Quality.GOOD], {"defense": True}, ["defense"] * 101])
+def test_invalid_array_predicate_inputs(values: Any) -> None:
+    for method in (field("tags").contains_all, field("tags").contains_any):
+        with pytest.raises((TypeError, ValueError)):
+            method(values).to_message().to_binary()
+
+
+def test_array_predicate_copies_inputs_and_validates_messages() -> None:
+    values = ["defense"]
+    expression = field("tags").contains_any(values)
+    values.append("agriculture")
+    assert expression == field("tags").contains_any(["defense"])
+    for operator, values in ((0, []), (99, []), (1, [""]), (2, ["defense"] * 101)):
+        message = data_access_pb.FilterExpression(
+            array_predicate=data_access_pb.FieldArrayPredicate(
+                field_name="tags",
+                operator=data_access_pb.FieldArrayOperator(operator),
+                values=values,
+            )
+        )
+        with pytest.raises(ValueError, match=r"[Aa]rray predicate"):
+            Expression.from_message(message)
+    message = expression.to_message()
+    message.is_null = data_access_pb.FieldNullCheck(field_name="tags")
+    with pytest.raises(ValueError, match="exactly one"):
+        Expression.from_message(message)
+
+
 @pytest.mark.parametrize(
     "value",
     [
diff --git a/tilebox-datasets/tests/test_client.py b/tilebox-datasets/tests/test_client.py
index f9e9003..1f998d8 100644
--- a/tilebox-datasets/tests/test_client.py
+++ b/tilebox-datasets/tests/test_client.py
@@ -1,23 +1,92 @@
 import os
 import subprocess
 import sys
+from dataclasses import replace
 from datetime import UTC, datetime
 from pathlib import Path
 from unittest.mock import MagicMock, patch
+from uuid import UUID
 
 import pytest
 import xarray as xr
+from protobuf.wkt.google.protobuf.descriptor_pb import (
+    EnumDescriptorProto,
+    EnumValueDescriptorProto,
+    FileDescriptorProto,
+    FileDescriptorSet,
+)
 from shapely import Polygon
 
 from _tilebox.errors import ArgumentError, AuthenticationError, InternalServerError, NotFoundError
 from _tilebox.replay import Record, Replay
+from tests.example_dataset.schema import DESCRIPTOR
 from tilebox.datasets import Client, DatasetClient
+from tilebox.datasets.aio import DatasetClient as AsyncDatasetClient
 from tilebox.datasets.aio.client import Client as AsyncClient
 from tilebox.datasets.client import _TILEBOX_API_URL, _TILEBOX_DEV_API_URL
 from tilebox.datasets.data.datapoint import QueryResultPage
+from tilebox.datasets.data.datasets import AnnotatedType, Dataset, DatasetKind, FieldAnnotation
 from tilebox.datasets.query.time_interval import us_to_datetime
 
 
+@pytest.mark.parametrize("client_type", [DatasetClient, AsyncDatasetClient])
+@pytest.mark.parametrize("summary", ["", "Synthetic scenes."])
+@pytest.mark.parametrize("kind", [DatasetKind.TEMPORAL, DatasetKind.SPATIOTEMPORAL, None])
+def test_dataset_representation(
+    client_type: type[DatasetClient] | type[AsyncDatasetClient], summary: str, kind: DatasetKind | None
+) -> None:
+    descriptor = FileDescriptorProto.from_binary(DESCRIPTOR.to_binary())
+    descriptor.enum_type.append(
+        EnumDescriptorProto(
+            name="Level",
+            value=[EnumValueDescriptorProto(name=name, number=n) for name, n in [("ZERO", 0), ("A", 17), ("B", 4)]],
+        )
+    )
+    annotations = []
+    for field in descriptor.message_type[0].field:
+        # Deliberately differ from declaration order to check annotation alignment.
+        field.number = 100 - field.number
+        if field.name == "some_enum":
+            field.type_name = ".tests.scientific.Level"
+            annotations.append(
+                FieldAnnotation("Hidden description", "Hidden example", queryable=True, enum_values=["L2-SP", '雪"'])
+            )
+        else:
+            annotations.append(FieldAnnotation("", ""))
+    schema = AnnotatedType(FileDescriptorSet(file=[descriptor]), "tests.scientific.Datapoint", annotations, kind)
+    dataset = Dataset(UUID(int=1), UUID(int=2), schema, "demo", "Scene demo", summary, "", "", slug="tilebox.demo")
+    dataset = Dataset.from_message(dataset.to_message())
+    client = client_type(MagicMock(), dataset)
+    output = repr(client)
+    label = {
+        DatasetKind.TEMPORAL: "Temporal Dataset",
+        DatasetKind.SPATIOTEMPORAL: "Spatiotemporal Dataset",
+        None: "Dataset",
+    }[kind]
+    heading = f"Scene demo [{label}]" + (f": {summary}" if summary else "")
+    assert output.splitlines()[:4] == ["tilebox.demo", heading, "", "Fields:"]
+    rows = [line.split() for line in output.splitlines()[4:]]
+    assert rows[:5] == [
+        ["🕒", "time", "Timestamp"],
+        ["🔑", "id", "UUID"],
+        ["ingestion_time", "Timestamp"],
+        ["geometry", "Geometry"],
+        ["some_string", "string"],
+    ]
+    assert ["some_int", "int64"] in rows
+    assert ["some_repeated_time", "[]", "Timestamp"] in rows
+    assert ["some_repeated_string", "[]", "string"] in rows
+    assert ["🔍", "some_enum", "enum", '("L2-SP",', r'"雪\"")'] in rows
+    assert "Hidden" not in output
+
+    field = descriptor.message_type[0].field[15]
+    field.label = field.Label.REPEATED
+    assert '[] enum ("L2-SP", "雪\\"")' in repr(client_type(MagicMock(), dataset))
+    legacy = replace(dataset, slug="", type=replace(schema, field_annotations=[]))
+    assert repr(client_type(MagicMock(), legacy)).startswith("demo\n")
+    assert '[] enum ("ZERO", "A", "B")' in repr(client_type(MagicMock(), legacy))
+
+
 @pytest.mark.parametrize("client_type", [Client, AsyncClient])
 @pytest.mark.parametrize(
     ("environment_url", "explicit_url", "expected_url"),
diff --git a/tilebox-datasets/tests/test_enums.py b/tilebox-datasets/tests/test_enums.py
new file mode 100644
index 0000000..9fb56b1
--- /dev/null
+++ b/tilebox-datasets/tests/test_enums.py
@@ -0,0 +1,314 @@
+"""Public enum ingestion and xarray reads through both client implementations."""
+
+from dataclasses import replace
+from datetime import UTC, datetime
+from inspect import isawaitable
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import UUID
+
+import pandas as pd
+import pytest
+from protobuf.wkt.google.protobuf.descriptor_pb import (
+    DescriptorProto,
+    Edition,
+    EnumDescriptorProto,
+    EnumValueDescriptorProto,
+    FieldDescriptorProto,
+    FileDescriptorProto,
+    FileDescriptorSet,
+)
+from protobuf.wkt.google.protobuf.timestamp_pb import Timestamp
+
+from _tilebox.proto.tilebox.v1.id_pb import ID
+from tilebox.datasets import DatasetClient, field
+from tilebox.datasets.aio import DatasetClient as AsyncDatasetClient
+from tilebox.datasets.data.datasets import AnnotatedType, Dataset, FieldAnnotation
+from tilebox.datasets.message_pool import get_message_type
+from tilebox.datasets.v1 import core_pb, data_access_pb, data_ingestion_pb
+from tilebox.query.pagination import Pagination
+
+
+@pytest.fixture
+def enum_schema() -> AnnotatedType:
+    schema = FileDescriptorProto(
+        name="enum.proto",
+        package="tests.enums",
+        syntax="editions",
+        edition=Edition.EDITION_2023,
+        dependency=["google/protobuf/timestamp.proto"],
+        enum_type=[
+            EnumDescriptorProto(
+                name="Level",
+                value=[
+                    EnumValueDescriptorProto(name=name, number=number)
+                    for name, number in [("ZERO", 0), ("FIRST", 17), ("SECOND", 4), ("THIRD", 9)]
+                ],
+            )
+        ],
+        message_type=[
+            DescriptorProto(name="Unrelated"),
+            DescriptorProto(
+                name="Row",
+                field=[
+                    FieldDescriptorProto(
+                        name="time",
+                        number=20,
+                        type=FieldDescriptorProto.Type.MESSAGE,
+                        type_name=".google.protobuf.Timestamp",
+                        label=FieldDescriptorProto.Label.OPTIONAL,
+                    ),
+                    FieldDescriptorProto(
+                        name="level",
+                        number=7,
+                        type=FieldDescriptorProto.Type.ENUM,
+                        type_name=".tests.enums.Level",
+                        label=FieldDescriptorProto.Label.OPTIONAL,
+                    ),
+                    FieldDescriptorProto(
+                        name="scene",
+                        number=11,
+                        type=FieldDescriptorProto.Type.STRING,
+                        label=FieldDescriptorProto.Label.OPTIONAL,
+                    ),
+                ],
+            ),
+        ],
+    )
+    return AnnotatedType(
+        FileDescriptorSet(file=[FileDescriptorProto(name="other.proto", package="other"), schema]),
+        "type.googleapis.com/tests.enums.Row",
+        [FieldAnnotation("", ""), FieldAnnotation("", "", enum_values=["L1", "L2-SP", "雪"]), FieldAnnotation("", "")],
+    )
+
+
+def _client(schema: AnnotatedType, asynchronous: bool = False) -> tuple[Any, Any]:
+    service = AsyncMock() if asynchronous else MagicMock()
+    service.get_collection_by_name.return_value = core_pb.CollectionInfo(
+        collection=core_pb.Collection(id=ID(uuid=UUID(int=2).bytes), name="test"),
+    )
+    connection = MagicMock()
+    connection.client.return_value = service
+    dataset = Dataset(UUID(int=1), UUID(int=0), schema, "test", "Test", "", "", "")
+    client_type = AsyncDatasetClient if asynchronous else DatasetClient
+    return client_type(connection, dataset), service
+
+
+async def _resolve(value: Any) -> Any:
+    return await value if isawaitable(value) else value
+
+
+@pytest.fixture
+def repeated_enum_schema(enum_schema: AnnotatedType) -> AnnotatedType:
+    enum_schema.descriptor_set.file[1].message_type[1].field[1].label = FieldDescriptorProto.Label.REPEATED
+    return enum_schema
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("asynchronous", [False, True])
+async def test_repeated_enum_ingestion_and_reads(repeated_enum_schema: AnnotatedType, asynchronous: bool) -> None:
+    client, service = _client(repeated_enum_schema, asynchronous)
+    collection = await _resolve(client.collection("test"))
+    row = get_message_type(repeated_enum_schema)
+    time = Timestamp.from_seconds(1)
+    raw = [row(time=time), row(time=time), row(time=time, level=[4, 17, 4]), row(time=time, level=[9])]
+    stored = [row(time=time), row(time=time), row(time=time, level=[4, 17]), row(time=time, level=[9])]
+    captured = []
+
+    def ingest(request: data_ingestion_pb.IngestRequest) -> data_ingestion_pb.IngestResponse:
+        captured[:] = request.values  # noqa: PD011 -- protobuf field
+        return data_ingestion_pb.IngestResponse()
+
+    service.ingest.side_effect = ingest
+    records = [
+        {"time": datetime.fromtimestamp(1, UTC), "level": values}
+        for values in [None, [], ["L2-SP", "L1", "L2-SP"], ["雪"]]
+    ]
+    columns = {"time": [record["time"] for record in records], "level": [record["level"] for record in records]}
+    for data in (records, columns, pd.DataFrame(records)):
+        await _resolve(collection.ingest(data))
+        assert [row.from_binary(value) for value in captured] == raw
+
+    # Deduplication belongs to the backend. Read its normalized payload, including a first page with only empty lists.
+    def query_page(request: data_access_pb.QueryRequest) -> data_access_pb.QueryResultPage:
+        if request.page is not None and request.page.starting_after is not None:
+            return data_access_pb.QueryResultPage(data=core_pb.RepeatedAny(value=[m.to_binary() for m in stored[2:]]))
+        return data_access_pb.QueryResultPage(
+            data=core_pb.RepeatedAny(value=[m.to_binary() for m in stored[:2]]),
+            next_page=Pagination(starting_after=UUID(int=2)).to_message(),
+        )
+
+    service.query.side_effect = query_page
+    service.query_by_id.return_value = core_pb.Any(value=stored[2].to_binary())
+    for target in (client, collection):
+        result = await _resolve(target.query(temporal_extent=("1970", "1971")))
+        assert result.level.dims == ("time",)
+        assert result.level.shape == (4,)
+        assert result.level.dtype == object
+        assert result.level.to_numpy().tolist() == [[], [], ["L2-SP", "L1"], ["雪"]]
+        assert all(isinstance(value, list) for value in result.level.to_numpy())
+        assert "n_level" not in result.dims
+        assert (await _resolve(target.find(UUID(int=3)))).level.item() == ["L2-SP", "L1"]
+        assert "level" not in await _resolve(target.query(temporal_extent=("1970", "1971"), skip_data=True))
+    await _resolve(collection.ingest(result))
+    assert [row.from_binary(value) for value in captured] == stored
+    result.level[0].item().append("L1")
+    assert result.level[1].item() == []
+    service.query.side_effect = None
+    service.query.return_value = data_access_pb.QueryResultPage(data=core_pb.RepeatedAny(value=[stored[0].to_binary()]))
+    empty = await _resolve(collection.query(temporal_extent=("1970", "1971")))
+    assert empty.level.dims == ("time",)
+    assert empty.level.item() == []
+    for value in ("L1", [0], [17], ["FIRST"], ["L1"] * 100 + ["unknown"]):
+        with pytest.raises((TypeError, ValueError), match="Field 'level'"):
+            await _resolve(collection.ingest([{"time": "2026-01-01", "level": value}]))
+    await _resolve(collection.ingest([{"time": "2026-01-01", "level": ["L1"] * 101}]))
+    message: Any = row.from_binary(captured[0])
+    assert list(message.level) == [17] * 101
+    for number in (0, 99):
+        service.query_by_id.return_value = core_pb.Any(value=bytes([0x3A, 2, 4, number]))
+        with pytest.raises(ValueError, match="Invalid enum value"):
+            await _resolve(client.find(UUID(int=3)))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("asynchronous", [False, True])
+async def test_public_enum_ingestion_and_reads(enum_schema: AnnotatedType, asynchronous: bool) -> None:
+    client, service = _client(enum_schema, asynchronous)
+    collection = await _resolve(client.collection("test"))
+    row = get_message_type(enum_schema)
+    time = Timestamp.from_seconds(1)
+    expected = [row(time=time, level=17), row(time=time, level=4), row(time=time, level=9), row(time=time)]
+    captured = []
+
+    def ingest(request: data_ingestion_pb.IngestRequest) -> data_ingestion_pb.IngestResponse:
+        captured[:] = request.values  # noqa: PD011 -- protobuf field
+        return data_ingestion_pb.IngestResponse(datapoint_ids=[ID(uuid=UUID(int=i).bytes) for i in range(1, 5)])
+
+    service.ingest.side_effect = ingest
+    records = [{"time": datetime.fromtimestamp(1, UTC), "level": value} for value in ["L1", "L2-SP", "雪", None]]
+    columns = {"time": [record["time"] for record in records], "level": [record["level"] for record in records]}
+    for data in (records, columns, pd.DataFrame(records)):
+        ids = await _resolve(collection.ingest(data))
+        assert ids == [UUID(int=i) for i in range(1, 5)]
+        assert [row.from_binary(value) for value in captured] == expected
+
+    # Separate server fixtures from the encoder under test; include an absent field.
+    wire = [message.to_binary() for message in expected]
+
+    def query_page(request: data_access_pb.QueryRequest) -> data_access_pb.QueryResultPage:
+        if request.page is not None and request.page.starting_after is not None:
+            return data_access_pb.QueryResultPage(data=core_pb.RepeatedAny(value=wire[2:]))
+        return data_access_pb.QueryResultPage(
+            data=core_pb.RepeatedAny(value=wire[:2]),
+            next_page=Pagination(starting_after=UUID(int=2)).to_message(),
+        )
+
+    service.query.side_effect = query_page
+    service.query_by_id.return_value = core_pb.Any(value=wire[1])
+    for target in (client, collection):
+        result = await _resolve(target.query(temporal_extent=("1970", "1971")))
+        assert result.level.dtype == object
+        assert result.level[:3].to_numpy().tolist() == ["L1", "L2-SP", "雪"]
+        assert result.level.isnull().to_numpy().tolist() == [False, False, False, True]  # noqa: PD003 -- xarray API
+        assert result.where(result.level.isin(["L2-SP", "雪"]), drop=True).level.to_numpy().tolist() == ["L2-SP", "雪"]
+        assert (await _resolve(target.find(UUID(int=3)))).level.item() == "L2-SP"
+        assert "level" not in await _resolve(target.query(temporal_extent=("1970", "1971"), skip_data=True))
+        assert "level" not in await _resolve(target.find(UUID(int=3), skip_data=True))
+    await _resolve(collection.ingest(result))
+    assert [row.from_binary(value) for value in captured] == expected
+
+    # The SDK sends public strings unchanged; membership and operator checks belong to the server.
+    def query(request: data_access_pb.QueryRequest) -> data_access_pb.QueryResultPage:
+        assert request.filters is not None
+        comparison = request.filters.expressions[0].comparison
+        assert comparison is not None
+        assert comparison.value is not None
+        assert comparison.value.string_value == "L2-SP"
+        return data_access_pb.QueryResultPage(data=core_pb.RepeatedAny(value=[wire[1]]))
+
+    service.query.side_effect = query
+    result = await _resolve(client.query(temporal_extent=("1970", "1971"), filter=field("level") == "L2-SP"))
+    assert result.level.to_numpy().tolist() == ["L2-SP"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("asynchronous", [False, True])
+async def test_invalid_public_enum_values(enum_schema: AnnotatedType, asynchronous: bool) -> None:
+    client, service = _client(enum_schema, asynchronous)
+    collection = await _resolve(client.collection("test"))
+    for value in (0, 17, 4.0, True, "FIRST", "l1", "unknown", ""):
+        with pytest.raises((ValueError, TypeError), match="Record 0: Field 'level'"):
+            await _resolve(collection.ingest([{"time": "2026-01-01", "level": value}]))
+
+    for number in (0, 99):
+        # Field 7, wire type 0. Check explicit zero independently of message constructors.
+        wire = bytes([0x38, number])
+        service.query_by_id.return_value = core_pb.Any(value=wire)
+        service.query.return_value = data_access_pb.QueryResultPage(data=core_pb.RepeatedAny(value=[wire]))
+        with pytest.raises(ValueError, match="Invalid enum value"):
+            await _resolve(client.find(UUID(int=3)))
+        if asynchronous:
+            with pytest.raises(ExceptionGroup) as error:
+                await _resolve(client.query(temporal_extent=("2026", "2027")))
+            assert len(error.value.exceptions) == 1
+            assert isinstance(error.value.exceptions[0], ValueError)
+            assert "Invalid enum value" in str(error.value.exceptions[0])
+        else:
+            with pytest.raises(ValueError, match="Invalid enum value"):
+                await _resolve(client.query(temporal_extent=("2026", "2027")))
+
+
+def test_enum_schema_isolation_and_append(enum_schema: AnnotatedType) -> None:
+    client, service = _client(enum_schema)
+    service.query_by_id.return_value = core_pb.Any(value=b"\x38\x04")
+    renamed = replace(
+        enum_schema,
+        field_annotations=[
+            FieldAnnotation("", ""),
+            FieldAnnotation("", "", enum_values=["a", "b", "c"]),
+            FieldAnnotation("", ""),
+        ],
+    )
+    other, other_service = _client(renamed)
+    other_service.query_by_id.return_value = core_pb.Any(value=b"\x38\x04")
+    assert other.find(UUID(int=3)).level.item() == "b"
+    assert client.find(UUID(int=3)).level.item() == "L2-SP"
+
+    updated = AnnotatedType.from_message(enum_schema.to_message())
+    updated.descriptor_set.file[1].enum_type[0].value.append(EnumValueDescriptorProto(name="FOURTH", number=31))
+    updated.field_annotations[1].enum_values.append("L3")
+    refreshed, refreshed_service = _client(updated)
+    refreshed_service.query_by_id.return_value = core_pb.Any(value=b"\x38\x1f")
+    assert refreshed.find(UUID(int=3)).level.item() == "L3"
+    refreshed_service.query_by_id.return_value = core_pb.Any(value=b"\x38\x04")
+    assert refreshed.find(UUID(int=3)).level.item() == "L2-SP"
+    assert client.find(UUID(int=3)).level.item() == "L2-SP"
+
+
+@pytest.mark.parametrize(
+    "problem", ["alignment", "short", "long", "empty", "duplicate", "wrong_type", "wrong_repeated_type"]
+)
+def test_invalid_enum_annotations(enum_schema: AnnotatedType, problem: str) -> None:
+    if problem == "alignment":
+        enum_schema.field_annotations.pop(0)
+    elif problem == "short":
+        enum_schema.field_annotations[1].enum_values.pop()
+    elif problem == "long":
+        enum_schema.field_annotations[1].enum_values.append("L3")
+    elif problem == "empty":
+        enum_schema.field_annotations[1].enum_values[0] = ""
+    elif problem == "duplicate":
+        enum_schema.field_annotations[1].enum_values[0] = "L2-SP"
+    elif problem == "wrong_type":
+        enum_schema.field_annotations[0], enum_schema.field_annotations[1] = (
+            enum_schema.field_annotations[1],
+            enum_schema.field_annotations[0],
+        )
+    else:
+        descriptor = enum_schema.descriptor_set.file[1].message_type[1].field[1]
+        descriptor.label = FieldDescriptorProto.Label.REPEATED
+        descriptor.type = FieldDescriptorProto.Type.STRING
+        descriptor.type_name = ""
+    with pytest.raises((ValueError, TypeError), match=r"annotation|enum_values"):
+        _client(enum_schema)
diff --git a/tilebox-datasets/tilebox/datasets/aio/dataset.py b/tilebox-datasets/tilebox/datasets/aio/dataset.py
index 631746e..6ee76d1 100644
--- a/tilebox-datasets/tilebox/datasets/aio/dataset.py
+++ b/tilebox-datasets/tilebox/datasets/aio/dataset.py
@@ -1,4 +1,4 @@
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Mapping
 from dataclasses import replace
 from functools import partial
 from uuid import UUID
@@ -21,6 +21,7 @@
 from tilebox.datasets.data.datasets import Dataset
 from tilebox.datasets.message_pool import get_message_type
 from tilebox.datasets.progress import ProgressCallback
+from tilebox.datasets.protobuf_conversion.field_types import ProtobufFieldType, enum_field_types
 from tilebox.datasets.protobuf_conversion.protobuf_xarray import MessageToXarrayConverter
 from tilebox.datasets.protobuf_conversion.to_protobuf import (
     DatapointIDs,
@@ -33,6 +34,7 @@
 from tilebox.datasets.query.id_interval import IDInterval, IDIntervalLike
 from tilebox.datasets.query.pagination import Pagination
 from tilebox.datasets.query.time_interval import TimeInterval, TimeIntervalLike
+from tilebox.datasets.representation import dataset_repr
 from tilebox.datasets.uuid import as_uuid, must_uuid_to_uuid_message, uuid_to_uuid_message
 from tilebox.datasets.v1.collections_connect import CollectionServiceClient
 from tilebox.datasets.v1.collections_pb import (
@@ -65,6 +67,7 @@ def __init__(
         self.name = dataset.name
         self._dataset = dataset
         self._message_type = get_message_type(dataset.type)
+        self._field_types = enum_field_types(dataset.type, self._message_type)
 
     async def collections(
         self, availability: bool | None = None, count: bool | None = None
@@ -217,7 +220,9 @@ async def find(
         data = self._message_type.from_binary(datapoint.value)
 
         converter = MessageToXarrayConverter(
-            initial_capacity=1, fields=DATAPOINT_METADATA_FIELDS if skip_data else None
+            initial_capacity=1,
+            fields=DATAPOINT_METADATA_FIELDS if skip_data else None,
+            field_types=self._field_types,
         )
         converter.convert(data)
         return converter.finalize("time", skip_empty_fields=skip_data).isel(time=0)
@@ -267,7 +272,7 @@ async def query(  # noqa: PLR0913
             message_type=self._message_type,
             show_progress=show_progress,
         )
-        return await _convert_to_dataset(pages, skip_empty_fields=skip_data)
+        return await _convert_to_dataset(pages, skip_empty_fields=skip_data, field_types=self._field_types)
 
     async def _collection_id(self, collection: "UUID | Collection | CollectionInfo | CollectionClient") -> UUID:
         if isinstance(collection, CollectionClient):
@@ -306,9 +311,7 @@ async def _collection_ids(
         return collection_ids
 
     def __repr__(self) -> str:
-        kind = self._dataset.type.kind
-        label = f"{kind.name.title()} Dataset" if kind else "Dataset"
-        return f"{self.name} [{label}]: {self._dataset.summary}"
+        return dataset_repr(self._dataset, self._message_type)
 
 
 # always ingest / delete in batches, to avoid timeout issues for very large datasets
@@ -410,7 +413,9 @@ async def find(self, datapoint_id: str | UUID, skip_data: bool = False) -> xr.Da
         data = self._dataset._message_type.from_binary(datapoint.value)
 
         converter = MessageToXarrayConverter(
-            initial_capacity=1, fields=DATAPOINT_METADATA_FIELDS if skip_data else None
+            initial_capacity=1,
+            fields=DATAPOINT_METADATA_FIELDS if skip_data else None,
+            field_types=self._dataset._field_types,
         )
         converter.convert(data)
         return converter.finalize("time", skip_empty_fields=skip_data).isel(time=0)
@@ -454,7 +459,7 @@ async def request(page: PaginationProtocol) -> QueryResultPage:
         if show_progress:
             pages = with_progressbar(pages, f"Fetching {self._dataset.name}")
 
-        return await _convert_to_dataset(pages, skip_empty_fields=skip_data)
+        return await _convert_to_dataset(pages, skip_empty_fields=skip_data, field_types=self._dataset._field_types)
 
     async def load(
         self,
@@ -553,7 +558,7 @@ async def query(
             message_type=self._dataset._message_type,
             show_progress=show_progress,
         )
-        return await _convert_to_dataset(pages, skip_empty_fields=skip_data)
+        return await _convert_to_dataset(pages, skip_empty_fields=skip_data, field_types=self._dataset._field_types)
 
     async def _iter_pages(
         self,
@@ -604,7 +609,13 @@ async def ingest(
         """
         message_type = self._dataset._message_type
         messages = marshal_messages(
-            to_messages(data, message_type, required_fields=["time"], ignore_fields=["id", "ingestion_time"])
+            to_messages(
+                data,
+                message_type,
+                required_fields=["time"],
+                ignore_fields=["id", "ingestion_time"],
+                field_types=self._dataset._field_types,
+            )
         )
 
         disable_progress_bar = callable(show_progress) or (not show_progress)
@@ -747,7 +758,12 @@ async def _iter_query_pages(  # noqa: PLR0913, PLR0917
         yield page
 
 
-async def _convert_to_dataset(pages: AsyncIterator[QueryResultPage], skip_empty_fields: bool = False) -> xr.Dataset:
+async def _convert_to_dataset(
+    pages: AsyncIterator[QueryResultPage],
+    skip_empty_fields: bool = False,
+    *,
+    field_types: Mapping[str, ProtobufFieldType] | None = None,
+) -> xr.Dataset:
     """
     Convert an async iterator of QueryResultPages into a single xarray Dataset
 
@@ -760,7 +776,10 @@ async def _convert_to_dataset(pages: AsyncIterator[QueryResultPage], skip_empty_
     Returns:
         The datapoints from the individual pages converted and combined into a single xarray dataset
     """
-    converter = MessageToXarrayConverter(fields=DATAPOINT_METADATA_FIELDS if skip_empty_fields else None)
+    converter = MessageToXarrayConverter(
+        fields=DATAPOINT_METADATA_FIELDS if skip_empty_fields else None,
+        field_types=field_types,
+    )
 
     def convert_page(page: QueryResultPage) -> None:
         if not page.data.value:
diff --git a/tilebox-datasets/tilebox/datasets/data/datasets.py b/tilebox-datasets/tilebox/datasets/data/datasets.py
index aca8c3b..8b6ab00 100644
--- a/tilebox-datasets/tilebox/datasets/data/datasets.py
+++ b/tilebox-datasets/tilebox/datasets/data/datasets.py
@@ -47,6 +47,8 @@ class FieldAnnotation:
     """A JSON Schema reference for the field when it corresponds to a well-known schema."""
     roles: list[FieldRole] = dataclass_field(default_factory=list)
     """Semantic display roles fulfilled by the field."""
+    enum_values: list[str] = dataclass_field(default_factory=list)
+    """Ordered public enum values. Non-empty datasets only allow appending values."""
 
     @classmethod
     def from_message(cls, annotation: dataset_type_pb.FieldAnnotation | None) -> "FieldAnnotation":
@@ -59,6 +61,7 @@ def from_message(cls, annotation: dataset_type_pb.FieldAnnotation | None) -> "Fi
             queryable=annotation.queryable,
             json_schema_ref=annotation.json_schema_ref if annotation.has_field("json_schema_ref") else None,
             roles=[_field_role_int_to_enum[role] for role in annotation.roles],
+            enum_values=[entry.value for entry in annotation.enum_values],
         )
 
     def to_message(self) -> dataset_type_pb.FieldAnnotation:
@@ -67,6 +70,7 @@ def to_message(self) -> dataset_type_pb.FieldAnnotation:
             example_value=self.example_value,
             queryable=self.queryable,
             roles=[role.value for role in self.roles],
+            enum_values=[dataset_type_pb.EnumValue(value=value) for value in self.enum_values],
         )
         if self.source_json_pointer is not None:
             annotation.source_json_pointer = self.source_json_pointer
@@ -99,7 +103,11 @@ class FieldDict(TypedDict):
         | type[Geometry]
         | type[list[Geometry]]
         | MessageFieldType
+        | Literal["enum"]
     ]
+    enum_values: NotRequired[list[str]]
+    repeated: NotRequired[bool]
+    """Whether an enum field contains lists. For other types, use list[T]."""
     description: NotRequired[str]
     example_value: NotRequired[str]
     source_json_pointer: NotRequired[str | None]
@@ -138,6 +146,23 @@ def from_message(cls, field: dataset_type_pb.Field) -> "Field":
 
     @classmethod
     def from_dict(cls, field: FieldDict) -> "Field":
+        enum_values = field.get("enum_values", [])
+        if field["type"] == "enum":
+            if (
+                not isinstance(enum_values, list)
+                or not enum_values
+                or any(not isinstance(value, str) or not value for value in enum_values)
+            ):
+                raise ValueError(f"Field {field['name']!r}: enum_values must be a nonempty list of nonempty strings")
+            if len(enum_values) > 100:
+                raise ValueError(f"Field {field['name']!r}: enum_values must contain at most 100 entries")
+            if len(set(enum_values)) != len(enum_values):
+                raise ValueError(f"Field {field['name']!r}: enum_values must be unique")
+        elif "enum_values" in field:
+            raise ValueError(f"Field {field['name']!r}: enum_values requires type 'enum'")
+        elif field.get("repeated", False):
+            raise ValueError(f"Field {field['name']!r}: use list[T] for non-enum repeated fields")
+
         origin = get_origin(field["type"])
         if origin is list:
             label = FieldDescriptorProto.Label.REPEATED
@@ -148,7 +173,11 @@ def from_dict(cls, field: FieldDict) -> "Field":
             inner_type = field["type"]
 
         message_type = _message_field_type(inner_type)
-        if message_type is not None:
+        if inner_type == "enum":
+            field_type = FieldDescriptorProto.Type.ENUM
+            field_type_name = None
+            label = FieldDescriptorProto.Label.REPEATED if field.get("repeated", False) else label
+        elif message_type is not None:
             field_type = FieldDescriptorProto.Type.MESSAGE
             field_type_name = f".{message_type.desc().type_name}"
         else:
@@ -170,6 +199,7 @@ def from_dict(cls, field: FieldDict) -> "Field":
                 roles=[
                     _field_roles_from_string[role] if isinstance(role, str) else role for role in field.get("roles", [])
                 ],
+                enum_values=list(enum_values),
             ),
         )
 
@@ -303,6 +333,7 @@ class Dataset:
     summary: str
     icon: str
     description: str
+    slug: str = ""
 
     @classmethod
     def from_message(cls, dataset: core_pb.Dataset) -> "Dataset":
@@ -317,6 +348,7 @@ def from_message(cls, dataset: core_pb.Dataset) -> "Dataset":
             summary=dataset.summary,
             icon=dataset.icon,
             description=dataset.description,
+            slug=dataset.slug,
         )
 
     def to_message(self) -> core_pb.Dataset:
@@ -329,6 +361,7 @@ def to_message(self) -> core_pb.Dataset:
             summary=self.summary,
             icon=self.icon,
             description=self.description,
+            slug=self.slug,
         )
 
 
diff --git a/tilebox-datasets/tilebox/datasets/protobuf_conversion/field_types.py b/tilebox-datasets/tilebox/datasets/protobuf_conversion/field_types.py
index 7a7d7ef..f45824d 100644
--- a/tilebox-datasets/tilebox/datasets/protobuf_conversion/field_types.py
+++ b/tilebox-datasets/tilebox/datasets/protobuf_conversion/field_types.py
@@ -26,6 +26,7 @@
 from protobuf.wkt.google.protobuf.timestamp_pb import Timestamp
 from shapely import from_wkb
 
+from tilebox.datasets.data.datasets import AnnotatedType
 from tilebox.datasets.schema import _MESSAGE_FIELD_TYPES, Assets
 from tilebox.datasets.v1.well_known_types_pb import UUID as UUIDMessage  # noqa: N811
 from tilebox.datasets.v1.well_known_types_pb import Geometry, LatLon, LatLonAlt, Quaternion, Vec3
@@ -123,6 +124,8 @@ def to_proto(self, value: Any) -> bool:
 
 
 class EnumField(ProtobufFieldType):
+    """Convert custom enum values without exposing protobuf symbols or numbers."""
+
     def __init__(self, name_lookup: dict[int, str]) -> None:
         super().__init__(object)
         self._values_to_name = name_lookup
@@ -136,6 +139,18 @@ def from_proto(self, value: ProtoFieldValue) -> str:
         except KeyError as error:
             raise ValueError(f"Invalid enum value {value}") from error
 
+    def to_proto(self, value: Any) -> int:
+        if not isinstance(value, str):
+            raise TypeError("Custom enum values must be public strings")
+        try:
+            return self._names_to_value[value]
+        except KeyError as error:
+            raise ValueError(f"Unknown public enum value {value!r}") from error
+
+
+class LegacyEnumField(EnumField):
+    """Convert descriptor-defined enums using short names, numbers, or enum members."""
+
     def to_proto(self, value: str | int | Enum) -> int:
         if isinstance(value, Enum):
             value = value.name
@@ -150,6 +165,44 @@ def to_proto(self, value: str | int | Enum) -> int:
         return integer_value
 
 
+def enum_field_types(schema: AnnotatedType, message_type: type[Message]) -> dict[str, ProtobufFieldType]:
+    """Build dataset-local converters; annotations are not part of the message-class cache."""
+    if not any(annotation.enum_values for annotation in schema.field_annotations):
+        return {}
+    descriptor = message_type.desc()
+    # Native descriptor collections may use numeric order. The wire contract uses declaration order.
+    fields = descriptor.proto.field
+    if len(fields) != len(schema.field_annotations):
+        raise ValueError("Custom enum schema requires one annotation per field, including empty annotations")
+    fields_by_name = {field.name: field for field in descriptor.fields}
+    result: dict[str, ProtobufFieldType] = {}
+    for field, annotation in zip(fields, schema.field_annotations, strict=True):
+        if not annotation.enum_values:
+            continue
+        value = fields_by_name[field.name].value
+        if isinstance(value, DescFieldValueList):
+            enum = value.element
+        elif isinstance(value, DescFieldValueEnum):
+            enum = value.enum
+        else:
+            enum = None
+        if not isinstance(enum, DescEnum):
+            raise TypeError(f"Field {field.name!r}: enum_values requires an enum field")
+        members = [member for member in enum.proto.value if member.number != 0]
+        public_values = annotation.enum_values
+        if (
+            len(members) != len(public_values)
+            or any(not name for name in public_values)
+            or len(set(public_values)) != len(public_values)
+            or len({member.number for member in members}) != len(members)
+        ):
+            raise ValueError(f"Field {field.name!r}: invalid enum_values mapping")
+        result[field.name] = EnumField(
+            {member.number: name for member, name in zip(members, public_values, strict=True)}
+        )
+    return result
+
+
 class TimestampField(ProtobufFieldType):
     def __init__(self) -> None:
         super().__init__("datetime64[ns]")
@@ -347,7 +400,7 @@ def infer_field_type(field: DescField) -> ProtobufFieldType:
     if isinstance(element, DescMessage):
         return _MESSAGE_NAMES_TO_FIELDS.get(element.type_name, ProtobufFieldType(object))
     if isinstance(element, DescEnum):
-        return EnumField(_enum_mapping(element))
+        return LegacyEnumField(_enum_mapping(element))
     if element == ScalarType.BOOL:
         return BoolField()
     return ProtobufFieldType(_PROTOBUF_TYPE_TO_NUMPY_TYPE[element])
@@ -409,10 +462,6 @@ def is_missing(value: Any) -> bool:
     This is needed for pandas 3.0+ compatibility where object-dtype columns use
     np.nan instead of None for missing values.
     """
-    try:
-        return bool(pd.isna(value))
-    except ValueError:
-        # pd.isna returns either a bool, or an array of bools. In case of an array, converting the result to bool()
-        # will raise a ValueError. For an array, we know it's not a missing value, even an array of all NaNs is not
-        # a missing value.
-        return False
+    missing = pd.isna(value)
+    # Arrays are not missing scalars, even when empty or containing only missing values.
+    return isinstance(missing, bool | np.bool_) and bool(missing)
diff --git a/tilebox-datasets/tilebox/datasets/protobuf_conversion/protobuf_xarray.py b/tilebox-datasets/tilebox/datasets/protobuf_conversion/protobuf_xarray.py
index bfafa3e..b5b4129 100644
--- a/tilebox-datasets/tilebox/datasets/protobuf_conversion/protobuf_xarray.py
+++ b/tilebox-datasets/tilebox/datasets/protobuf_conversion/protobuf_xarray.py
@@ -3,9 +3,9 @@
 """
 
 import contextlib
-from collections.abc import Callable, Collection, Sequence
+from collections.abc import Callable, Collection, Mapping, Sequence
 from operator import attrgetter
-from typing import Any, TypeVar
+from typing import Any, TypeVar, cast
 
 import numpy as np
 import xarray as xr
@@ -13,6 +13,7 @@
 from protobuf import DescField, DescFieldValueList, DescFieldValueMessage, DescFieldValueSingular, Message
 
 from tilebox.datasets.protobuf_conversion.field_types import (
+    EnumField,
     ProtobufFieldType,
     ProtoFieldValue,
     infer_field_type,
@@ -22,7 +23,13 @@
 
 
 class MessageToXarrayConverter:
-    def __init__(self, initial_capacity: int = 0, *, fields: Collection[str] | None = None) -> None:
+    def __init__(
+        self,
+        initial_capacity: int = 0,
+        *,
+        fields: Collection[str] | None = None,
+        field_types: Mapping[str, ProtobufFieldType] | None = None,
+    ) -> None:
         """
         A converter for converting arbitrary protobuf messages to a xarray dataset
 
@@ -30,10 +37,12 @@ def __init__(self, initial_capacity: int = 0, *, fields: Collection[str] | None
             initial_capacity: The initial capacity (in number of messages) to reserve for the data arrays of each field.
                 If the total number of messages is already known in advance, set it to this number
             fields: Only convert these fields when the query projects a subset of the schema.
+            field_types: Dataset-specific converters for annotated fields, overriding descriptor-only inference.
         """
         self._capacity = initial_capacity
         self.count = 0  # number of messages already converted
         self._fields = fields
+        self._field_types = field_types or {}
         self._converters: list[tuple[Callable[[Message], ProtoFieldValue], _FieldConverter]] | None = None
 
     def convert(self, message: Message) -> None:
@@ -45,7 +54,7 @@ def convert(self, message: Message) -> None:
         """
         self._ensure_capacity(self.count + 1)
         if self._converters is None:
-            self._converters = _create_field_converters(message, self._capacity, self._fields)
+            self._converters = _create_field_converters(message, self._capacity, self._fields, self._field_types)
         self._convert(message)
 
     def convert_all(self, messages: list[AnyMessage]) -> None:
@@ -60,7 +69,7 @@ def convert_all(self, messages: list[AnyMessage]) -> None:
 
         self._ensure_capacity(self.count + len(messages))
         if self._converters is None:
-            self._converters = _create_field_converters(messages[0], self._capacity, self._fields)
+            self._converters = _create_field_converters(messages[0], self._capacity, self._fields, self._field_types)
         for read, converter in self._converters:
             converter.convert_all(self.count, messages, read)
         self.count += len(messages)
@@ -261,6 +270,33 @@ def resize(self, buffer_size: int) -> None:
             self._data = data
 
 
+class _EnumListFieldConverter(_FieldConverter):
+    """Keep each repeated enum as one ordered list, including empty fields."""
+
+    def __init__(self, field_name: str, field_type: EnumField) -> None:
+        self._field_name = field_name
+        self._type = field_type
+        self._data: NDArray[Any] = np.empty(0, dtype=object)
+
+    def __call__(self, index: int, value: ProtoFieldValue) -> None:
+        self._data[index] = [self._type.from_proto(item) for item in cast(Sequence, value)]
+
+    def resize(self, buffer_size: int) -> None:
+        if buffer_size > len(self._data):
+            data = np.empty(buffer_size, dtype=object)
+            data[: len(self._data)] = self._data
+            for index in range(len(self._data), buffer_size):
+                data[index] = []
+            self._data = data
+
+    def finalize(
+        self, dataset: xr.Dataset, count: int, dimension_names: tuple[str, ...], skip_if_empty: bool = False
+    ) -> str:
+        _ = skip_if_empty  # Empty lists are values, not missing data.
+        dataset[self._field_name] = dimension_names, self._data[:count]
+        return self._field_name
+
+
 class _ArrayFieldConverter(_FieldConverter):
     def __init__(
         self,
@@ -339,7 +375,10 @@ def _resize(self) -> None:
 
 
 def _create_field_converters(
-    message: Message, buffer_size: int, fields: Collection[str] | None
+    message: Message,
+    buffer_size: int,
+    fields: Collection[str] | None,
+    field_types: Mapping[str, ProtobufFieldType],
 ) -> list[tuple[Callable[[Message], ProtoFieldValue], _FieldConverter]]:
     """
     Compile field readers and scientific converters once for the selected schema.
@@ -358,7 +397,7 @@ def _create_field_converters(
             continue
         # if we have an unsupported field type we will get a ValueError, so we just skip those fields
         with contextlib.suppress(ValueError):
-            converter = _create_field_converter(field)
+            converter = _create_field_converter(field, field_types.get(field.name))
             converters.append((_field_reader(field), converter))
 
     if buffer_size > 0:
@@ -380,7 +419,7 @@ def _field_reader(field: DescField) -> Callable[[Message], ProtoFieldValue]:
     return lambda message: read(message) if field in message else None
 
 
-def _create_field_converter(field: DescField) -> _FieldConverter:
+def _create_field_converter(field: DescField, field_type: ProtobufFieldType | None = None) -> _FieldConverter:
     """
     Create a field converter for the given protobuf field descriptor.
 
@@ -391,8 +430,10 @@ def _create_field_converter(field: DescField) -> _FieldConverter:
     Returns:
         A field converter for the given protobuf field descriptor
     """
-    field_type = infer_field_type(field)
+    field_type = field_type or infer_field_type(field)
     if isinstance(field.value, DescFieldValueList):
+        if isinstance(field_type, EnumField):
+            return _EnumListFieldConverter(field.name, field_type)
         return _ArrayFieldConverter(field.name, field_type)
 
     return _SimpleFieldConverter(field.name, field_type)
diff --git a/tilebox-datasets/tilebox/datasets/protobuf_conversion/to_protobuf.py b/tilebox-datasets/tilebox/datasets/protobuf_conversion/to_protobuf.py
index bcb33ca..e921e30 100644
--- a/tilebox-datasets/tilebox/datasets/protobuf_conversion/to_protobuf.py
+++ b/tilebox-datasets/tilebox/datasets/protobuf_conversion/to_protobuf.py
@@ -10,6 +10,7 @@
 from protobuf import DescFieldValueList, DescFieldValueMessage, DescMessage, Message
 
 from tilebox.datasets.protobuf_conversion.field_types import (
+    EnumField,
     ProtobufFieldType,
     ProtoFieldValue,
     ScalarProtoFieldValue,
@@ -30,6 +31,8 @@ def to_messages(
     message_type: type[_MessageT],
     required_fields: list[str] | None = None,
     ignore_fields: list[str] | None = None,
+    *,
+    field_types: Mapping[str, ProtobufFieldType] | None = None,
 ) -> list[_MessageT]:
     """Convert supported ingestion inputs into protobuf messages.
 
@@ -40,12 +43,14 @@ def to_messages(
 
     Missing optional values and absent record keys leave the protobuf field
     unset. Required fields must be present and non-missing in every record.
+    field_types supplies dataset-specific converters, such as public enum
+    vocabularies that cannot be inferred from protobuf descriptors alone.
     """
     required = set(required_fields or [])
     ignore = set(ignore_fields or [])
     field_descriptors_by_name = _fields_by_name(message_type)
     records = _iter_records(data, field_descriptors_by_name, ignore)
-    return _records_to_messages(records, message_type, field_descriptors_by_name, required, ignore)
+    return _records_to_messages(records, message_type, required, ignore, field_types or {})
 
 
 def marshal_messages(messages: list[Message]) -> list[bytes]:
@@ -117,7 +122,8 @@ def _xarray_records(
         descriptor = descriptors.get(field_name)
         if descriptor is not None and isinstance(descriptor.value, DescFieldValueList):
             field_type = infer_field_type(descriptor)
-            values = trim_trailing_fill_values(values, field_type.fill_value)
+            if not (isinstance(field_type, EnumField) and values.ndim == 1):
+                values = trim_trailing_fill_values(values, field_type.fill_value)
         columns[field_name] = values
     yield from _transpose_columns(columns, validated_names)
 
@@ -171,13 +177,13 @@ def _validate_field_names(
 def _records_to_messages(
     records: Iterable[Record],
     message_type: type[_MessageT],
-    descriptors: Mapping[str, FieldDescriptor],
     required: set[str],
     ignore: set[str],
+    field_types: Mapping[str, ProtobufFieldType],
 ) -> list[_MessageT]:
     """Convert records and add the record index to conversion errors."""
     return [
-        _record_to_message_with_context(indexed_record, message_type, descriptors, required, ignore)
+        _record_to_message_with_context(indexed_record, message_type, required, ignore, field_types)
         for indexed_record in enumerate(records)
     ]
 
@@ -185,14 +191,14 @@ def _records_to_messages(
 def _record_to_message_with_context(
     indexed_record: tuple[int, Record],
     message_type: type[_MessageT],
-    descriptors: Mapping[str, FieldDescriptor],
     required: set[str],
     ignore: set[str],
+    field_types: Mapping[str, ProtobufFieldType],
 ) -> _MessageT:
     """Convert one record and add its index to conversion errors."""
     index, record = indexed_record
     try:
-        return _record_to_message(record, message_type, descriptors, required, ignore)
+        return _record_to_message(record, message_type, required, ignore, field_types)
     except (TypeError, ValueError) as error:
         raise type(error)(f"Record {index}: {error}") from error
 
@@ -200,21 +206,22 @@ def _record_to_message_with_context(
 def _record_to_message(
     record: Record,
     message_type: type[_MessageT],
-    descriptors: Mapping[str, FieldDescriptor],
     required: set[str],
     ignore: set[str],
+    field_types: Mapping[str, ProtobufFieldType],
 ) -> _MessageT:
     """Convert one field-name-validated record into a protobuf message."""
     missing_required = [name for name in required if name not in record or is_missing(record[name])]
     if missing_required:
         raise ValueError(f"Missing required field(s): {', '.join(sorted(missing_required))}")
 
+    descriptors = _fields_by_name(message_type)
     values: dict[str, ProtoFieldValue] = {}
     for field_name, value in record.items():
         if field_name in ignore or is_missing(value):
             continue
         descriptor = descriptors[field_name]
-        field_type = infer_field_type(descriptor)
+        field_type = field_types.get(field_name) or infer_field_type(descriptor)
         try:
             if isinstance(descriptor.value, DescFieldValueList):
                 converted = _convert_repeated_value(value, field_type)
diff --git a/tilebox-datasets/tilebox/datasets/query/expression.py b/tilebox-datasets/tilebox/datasets/query/expression.py
index 89d12f2..4c069b7 100644
--- a/tilebox-datasets/tilebox/datasets/query/expression.py
+++ b/tilebox-datasets/tilebox/datasets/query/expression.py
@@ -1,6 +1,6 @@
+from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
 from datetime import datetime, timedelta
-from enum import Enum
 from math import isfinite
 from typing import TYPE_CHECKING, ClassVar, NoReturn, TypeAlias
 
@@ -13,7 +13,7 @@
 if TYPE_CHECKING:
     import numpy as np
 
-    _QueryScalar: TypeAlias = bool | int | float | str | bytes | datetime | timedelta | Enum | np.generic
+    _QueryScalar: TypeAlias = bool | int | float | str | bytes | datetime | timedelta | np.generic
 else:
     _QueryScalar: TypeAlias = object
 
@@ -51,9 +51,9 @@ def to_message(self) -> data_access_pb.FilterExpression:
         raise NotImplementedError
 
     @classmethod
-    def from_message(cls, message: data_access_pb.FilterExpression) -> "Expression":
+    def from_message(cls, message: data_access_pb.FilterExpression) -> "Expression":  # noqa: C901
         """Construct an expression from its protobuf representation."""
-        variants = [name for name in ("logical", "comparison", "is_null") if message.has_field(name)]
+        variants = [name for name in ("logical", "comparison", "is_null", "array_predicate") if message.has_field(name)]
         if len(variants) != 1:
             raise ValueError("A filter expression must contain exactly one expression node")
 
@@ -75,6 +75,10 @@ def from_message(cls, message: data_access_pb.FilterExpression) -> "Expression":
         if variant == "is_null":
             assert message.is_null is not None
             return _NullCheck(message.is_null.field_name)
+        if variant == "array_predicate":
+            predicate = message.array_predicate
+            assert predicate is not None
+            return _ArrayPredicate(predicate.field_name, predicate.operator, tuple(predicate.values))
 
         logical = message.logical
         assert logical is not None
@@ -125,6 +129,23 @@ def is_not_null(self) -> Expression:
         """Match datapoints where this field is present and non-null."""
         return ~self.is_null()
 
+    def contains(self, value: str) -> Expression:
+        """Match a queryable repeated enum containing this public value."""
+        return self.contains_all([value])
+
+    def contains_all(self, values: Iterable[str]) -> Expression:
+        """Match a queryable repeated enum containing every value; an empty request matches all rows."""
+        return self._array_predicate(data_access_pb.FieldArrayOperator.CONTAINS_ALL, values)
+
+    def contains_any(self, values: Iterable[str]) -> Expression:
+        """Match a queryable repeated enum containing any value; an empty request matches no rows."""
+        return self._array_predicate(data_access_pb.FieldArrayOperator.OVERLAPS, values)
+
+    def _array_predicate(self, operator: data_access_pb.FieldArrayOperator, values: Iterable[str]) -> Expression:
+        if isinstance(values, str | bytes | Mapping) or not isinstance(values, Iterable):
+            raise TypeError("Array predicates require an iterable of public strings, not a single string or mapping")
+        return _ArrayPredicate(self.name, operator, tuple(values))
+
     def _comparison(self, operator: data_access_pb.FieldComparisonOperator, value: object) -> Expression:
         query_value = _QueryValue.parse(value)
         _validate_comparison(operator, query_value)
@@ -145,8 +166,6 @@ class _QueryValue:
     def parse(cls, value: object) -> "_QueryValue":  # noqa: C901, PLR0911, PLR0912
         import numpy as np  # noqa: PLC0415
 
-        if isinstance(value, Enum):
-            return cls("enum_name", value.name)
         if isinstance(value, bool | np.bool_):
             return cls("bool_value", bool(value))
         if isinstance(value, np.datetime64):
@@ -182,7 +201,8 @@ def parse(cls, value: object) -> "_QueryValue":  # noqa: C901, PLR0911, PLR0912
                 raise ValueError("Query float values must be finite")
             return cls("double_value", number)
         if isinstance(value, str | np.str_):
-            return cls("string_value", str(value))
+            # Preserve string contents even when a subclass overrides __str__, such as str + Enum.
+            return cls("string_value", str.__str__(value))
         if isinstance(value, bytes | np.bytes_):
             return cls("bytes_value", bytes(value))
         if value is None:
@@ -239,6 +259,34 @@ def to_message(self) -> data_access_pb.FilterExpression:
         )
 
 
+@dataclass(frozen=True, slots=True)
+class _ArrayPredicate(Expression):
+    field_name: str
+    operator: data_access_pb.FieldArrayOperator
+    values: tuple[str, ...]
+
+    def __post_init__(self) -> None:
+        if self.operator not in {
+            data_access_pb.FieldArrayOperator.CONTAINS_ALL,
+            data_access_pb.FieldArrayOperator.OVERLAPS,
+        }:
+            raise ValueError(f"Unknown array predicate operator: {self.operator}")
+        if len(self.values) > 100:
+            raise ValueError("Array predicates accept at most 100 values")
+        for value in self.values:
+            if not value:
+                raise ValueError("Array predicate values must be nonempty strings")
+
+    def to_message(self) -> data_access_pb.FilterExpression:
+        return data_access_pb.FilterExpression(
+            array_predicate=data_access_pb.FieldArrayPredicate(
+                field_name=self.field_name,
+                operator=self.operator,
+                values=list(self.values),
+            )
+        )
+
+
 @dataclass(frozen=True, slots=True)
 class _NullCheck(Expression):
     field_name: str
@@ -325,6 +373,5 @@ def _validate_duration(seconds: int, nanos: int) -> None:
     "string_value",
     "timestamp_value",
     "duration_value",
-    "enum_name",
     "bytes_value",
 )
diff --git a/tilebox-datasets/tilebox/datasets/representation.py b/tilebox-datasets/tilebox/datasets/representation.py
new file mode 100644
index 0000000..3471ffb
--- /dev/null
+++ b/tilebox-datasets/tilebox/datasets/representation.py
@@ -0,0 +1,40 @@
+import json
+
+from protobuf import Message
+from protobuf.wkt.google.protobuf.descriptor_pb import FieldDescriptorProto
+
+from tilebox.datasets.data.datasets import Dataset
+from tilebox.datasets.protobuf_conversion.field_types import enum_mapping_from_field_descriptor
+
+
+def dataset_repr(dataset: Dataset, message_type: type[Message]) -> str:
+    kind = dataset.type.kind
+    label = f"{kind.name.title()} Dataset" if kind else "Dataset"
+    heading = f"{dataset.name} [{label}]"
+    if dataset.summary:
+        heading += f": {dataset.summary}"
+    lines = [dataset.slug or dataset.code_name, heading, "", "Fields:"]
+
+    descriptor = message_type.desc()
+    fields = descriptor.proto.field
+    fields_by_name = {field.name: field for field in descriptor.fields}
+    name_width = max((len(field.name) for field in fields), default=0)
+    for index, field in enumerate(fields):
+        annotation = dataset.type.field_annotations[index] if index < len(dataset.type.field_annotations) else None
+        marker = {"time": "🕒", "id": "🔑"}.get(field.name, "🔍" if annotation and annotation.queryable else "  ")
+        if field.type == FieldDescriptorProto.Type.ENUM:
+            values = (
+                annotation.enum_values
+                if annotation and annotation.enum_values
+                else enum_mapping_from_field_descriptor(fields_by_name[field.name]).values()
+            )
+            field_type = "enum (" + ", ".join(json.dumps(value, ensure_ascii=False) for value in values) + ")"
+        elif field.type == FieldDescriptorProto.Type.MESSAGE:
+            field_type = field.type_name.rsplit(".", 1)[-1]
+        else:
+            field_type = field.type.name.lower()
+        if field.label == FieldDescriptorProto.Label.REPEATED:
+            field_type = f"[] {field_type}"
+        lines.append(f"{marker}  {field.name:<{name_width}}  {field_type}")
+
+    return "\n".join(lines)
diff --git a/tilebox-datasets/tilebox/datasets/sync/dataset.py b/tilebox-datasets/tilebox/datasets/sync/dataset.py
index fd07570..554faf1 100644
--- a/tilebox-datasets/tilebox/datasets/sync/dataset.py
+++ b/tilebox-datasets/tilebox/datasets/sync/dataset.py
@@ -1,4 +1,4 @@
-from collections.abc import Iterator
+from collections.abc import Iterator, Mapping
 from dataclasses import replace
 from functools import partial
 from uuid import UUID
@@ -20,6 +20,7 @@
 from tilebox.datasets.data.datasets import Dataset
 from tilebox.datasets.message_pool import get_message_type
 from tilebox.datasets.progress import ProgressCallback
+from tilebox.datasets.protobuf_conversion.field_types import ProtobufFieldType, enum_field_types
 from tilebox.datasets.protobuf_conversion.protobuf_xarray import MessageToXarrayConverter
 from tilebox.datasets.protobuf_conversion.to_protobuf import (
     DatapointIDs,
@@ -32,6 +33,7 @@
 from tilebox.datasets.query.id_interval import IDInterval, IDIntervalLike
 from tilebox.datasets.query.pagination import Pagination
 from tilebox.datasets.query.time_interval import TimeInterval, TimeIntervalLike
+from tilebox.datasets.representation import dataset_repr
 from tilebox.datasets.sync.pagination import (
     with_progressbar,
     with_time_progress_callback,
@@ -69,6 +71,7 @@ def __init__(
         self.name = dataset.name
         self._dataset = dataset
         self._message_type = get_message_type(dataset.type)
+        self._field_types = enum_field_types(dataset.type, self._message_type)
 
     def collections(self, availability: bool | None = None, count: bool | None = None) -> dict[str, "CollectionClient"]:
         """
@@ -219,7 +222,9 @@ def find(
         data = self._message_type.from_binary(datapoint.value)
 
         converter = MessageToXarrayConverter(
-            initial_capacity=1, fields=DATAPOINT_METADATA_FIELDS if skip_data else None
+            initial_capacity=1,
+            fields=DATAPOINT_METADATA_FIELDS if skip_data else None,
+            field_types=self._field_types,
         )
         converter.convert(data)
         return converter.finalize("time", skip_empty_fields=skip_data).isel(time=0)
@@ -269,7 +274,7 @@ def query(  # noqa: PLR0913
             message_type=self._message_type,
             show_progress=show_progress,
         )
-        return _convert_to_dataset(pages, skip_empty_fields=skip_data)
+        return _convert_to_dataset(pages, skip_empty_fields=skip_data, field_types=self._field_types)
 
     def _collection_id(self, collection: "UUID | Collection | CollectionInfo | CollectionClient") -> UUID:
         if isinstance(collection, CollectionClient):
@@ -308,9 +313,7 @@ def _collection_ids(
         return collection_ids
 
     def __repr__(self) -> str:
-        kind = self._dataset.type.kind
-        label = f"{kind.name.title()} Dataset" if kind else "Dataset"
-        return f"{self.name} [{label}]: {self._dataset.summary}"
+        return dataset_repr(self._dataset, self._message_type)
 
 
 # always ingest / delete in batches, to avoid timeout issues for very large datasets
@@ -408,7 +411,9 @@ def find(self, datapoint_id: str | UUID, skip_data: bool = False) -> xr.Dataset:
         data = self._dataset._message_type.from_binary(datapoint.value)
 
         converter = MessageToXarrayConverter(
-            initial_capacity=1, fields=DATAPOINT_METADATA_FIELDS if skip_data else None
+            initial_capacity=1,
+            fields=DATAPOINT_METADATA_FIELDS if skip_data else None,
+            field_types=self._dataset._field_types,
         )
         converter.convert(data)
         return converter.finalize("time", skip_empty_fields=skip_data).isel(time=0)
@@ -452,7 +457,7 @@ def request(page: PaginationProtocol) -> QueryResultPage:
         if show_progress:
             pages = with_progressbar(pages, f"Fetching {self._dataset.name}")
 
-        return _convert_to_dataset(pages, skip_empty_fields=skip_data)
+        return _convert_to_dataset(pages, skip_empty_fields=skip_data, field_types=self._dataset._field_types)
 
     def load(
         self,
@@ -550,7 +555,7 @@ def query(
             message_type=self._dataset._message_type,
             show_progress=show_progress,
         )
-        return _convert_to_dataset(pages, skip_empty_fields=skip_data)
+        return _convert_to_dataset(pages, skip_empty_fields=skip_data, field_types=self._dataset._field_types)
 
     def _iter_pages(
         self,
@@ -600,7 +605,13 @@ def ingest(
         """
         message_type = self._dataset._message_type
         messages = marshal_messages(
-            to_messages(data, message_type, required_fields=["time"], ignore_fields=["id", "ingestion_time"])
+            to_messages(
+                data,
+                message_type,
+                required_fields=["time"],
+                ignore_fields=["id", "ingestion_time"],
+                field_types=self._dataset._field_types,
+            )
         )
 
         disable_progress_bar = callable(show_progress) or (not show_progress)
@@ -740,7 +751,12 @@ def _iter_query_pages(  # noqa: PLR0913, PLR0917
     yield from pages
 
 
-def _convert_to_dataset(pages: Iterator[QueryResultPage], skip_empty_fields: bool = False) -> xr.Dataset:
+def _convert_to_dataset(
+    pages: Iterator[QueryResultPage],
+    skip_empty_fields: bool = False,
+    *,
+    field_types: Mapping[str, ProtobufFieldType] | None = None,
+) -> xr.Dataset:
     """
     Convert an iterator of QueryResultPages into a single xarray Dataset
 
@@ -753,7 +769,10 @@ def _convert_to_dataset(pages: Iterator[QueryResultPage], skip_empty_fields: boo
     Returns:
         The datapoints from the individual pages converted and combined into a single xarray dataset
     """
-    converter = MessageToXarrayConverter(fields=DATAPOINT_METADATA_FIELDS if skip_empty_fields else None)
+    converter = MessageToXarrayConverter(
+        fields=DATAPOINT_METADATA_FIELDS if skip_empty_fields else None,
+        field_types=field_types,
+    )
 
     def convert_page(page: QueryResultPage) -> None:
         if not page.data.value:
diff --git a/tilebox-datasets/tilebox/datasets/v1/data_access_pb.py b/tilebox-datasets/tilebox/datasets/v1/data_access_pb.py
index 57c5789..655d532 100644
--- a/tilebox-datasets/tilebox/datasets/v1/data_access_pb.py
+++ b/tilebox-datasets/tilebox/datasets/v1/data_access_pb.py
@@ -140,11 +140,11 @@ def __init__(
         spatial_extent: SpatialFilter | None
         expressions: list[FilterExpression]
 
-_FilterExpressionFields: TypeAlias = Literal["logical", "comparison", "is_null"]
+_FilterExpressionFields: TypeAlias = Literal["logical", "comparison", "is_null", "array_predicate"]
 
 class FilterExpression(Message[_FilterExpressionFields]):
     """
-    FilterExpression is a typed filter expression compatible with the Basic CQL2 property-to-literal subset.
+    FilterExpression supports Basic CQL2 property-to-literal comparisons and custom enum array predicates.
     Expressions use three-valued Boolean logic compatible with SQL and CQL2: true, false, and null (unknown). A datapoint
     matches only when the complete filter expression evaluates to true; false and null results are excluded.
 
@@ -167,9 +167,13 @@ class FilterExpression(Message[_FilterExpressionFields]):
             ```proto
             optional datasets.v1.FieldNullCheck is_null = 3;
             ```
+        array_predicate:
+            ```proto
+            optional datasets.v1.FieldArrayPredicate array_predicate = 4;
+            ```
     """
 
-    __slots__ = ("logical", "comparison", "is_null")
+    __slots__ = ("logical", "comparison", "is_null", "array_predicate")
 
     if TYPE_CHECKING:
 
@@ -179,12 +183,64 @@ def __init__(
             logical: LogicalExpression | None = None,
             comparison: FieldComparison | None = None,
             is_null: FieldNullCheck | None = None,
+            array_predicate: FieldArrayPredicate | None = None,
         ) -> None:
             pass
 
         logical: LogicalExpression | None
         comparison: FieldComparison | None
         is_null: FieldNullCheck | None
+        array_predicate: FieldArrayPredicate | None
+
+_FieldArrayPredicateFields: TypeAlias = Literal["field_name", "operator", "values"]
+
+class FieldArrayPredicate(Message[_FieldArrayPredicateFields]):
+    """
+    FieldArrayPredicate tests membership of public values in a queryable repeated custom enum field.
+    Values are case-sensitive and must belong to the field's vocabulary. Order and duplicates do not affect matching.
+    Omitted and empty fields both represent the empty set.
+    CONTAINS_ALL with no values is true; OVERLAPS with no values is false. These predicates never return null.
+    Scalar comparisons and null checks are not supported on repeated fields.
+
+    ```proto
+    message datasets.v1.FieldArrayPredicate
+    ```
+
+    Attributes:
+        field_name:
+            Name of a top-level queryable repeated custom enum field.
+
+            ```proto
+            string field_name = 1;
+            ```
+        operator:
+            ```proto
+            datasets.v1.FieldArrayOperator operator = 2;
+            ```
+        values:
+            Public enum values, not protobuf symbols or numbers. An empty list is allowed.
+
+            ```proto
+            repeated string values = 3;
+            ```
+    """
+
+    __slots__ = ("field_name", "operator", "values")
+
+    if TYPE_CHECKING:
+
+        def __init__(
+            self,
+            *,
+            field_name: str = "",
+            operator: FieldArrayOperator | None = None,
+            values: list[str] | None = None,
+        ) -> None:
+            pass
+
+        field_name: str
+        operator: FieldArrayOperator
+        values: list[str]
 
 _LogicalExpressionFields: TypeAlias = Literal["operator", "operands"]
 
@@ -235,6 +291,9 @@ class FieldComparison(Message[_FieldComparisonFields]):
     Every comparison evaluates to null (unknown) when the field is absent or explicitly null, including NOT_EQUAL.
     Consequently, both `field != value` and `NOT(field == value)` require the field to be present. To include absent or
     null fields, explicitly combine the comparison with FieldNullCheck using OR.
+    Scalar enum fields support only EQUAL and NOT_EQUAL. Use a LogicalExpression with OR over equality comparisons to match
+    any of several members. Unknown enum values, numeric enum literals, and ordering comparisons are invalid arguments.
+    For repeated custom enums, use FieldArrayPredicate instead.
 
     ```proto
     message datasets.v1.FieldComparison
@@ -254,7 +313,8 @@ class FieldComparison(Message[_FieldComparisonFields]):
             datasets.v1.FieldComparisonOperator operator = 2;
             ```
         value:
-            Literal value. Its kind must match the field's protobuf descriptor.
+            Literal value, interpreted using the field's schema. For custom enums, use string_value with a value from
+            FieldAnnotation.enum_values.
 
             ```proto
             optional datasets.v1.FieldQueryValue value = 3;
@@ -312,7 +372,7 @@ def __init__(
 
         field_name: str
 
-_FieldQueryValueFields: TypeAlias = Literal["bool_value", "int64_value", "uint64_value", "double_value", "string_value", "timestamp_value", "duration_value", "enum_name", "bytes_value"]
+_FieldQueryValueFields: TypeAlias = Literal["bool_value", "int64_value", "uint64_value", "double_value", "string_value", "timestamp_value", "duration_value", "bytes_value"]
 
 class FieldQueryValue(Message[_FieldQueryValueFields]):
     """
@@ -342,6 +402,10 @@ class FieldQueryValue(Message[_FieldQueryValueFields]):
             optional double double_value = 4;
             ```
         string_value:
+            Text for a string field, or a value from FieldAnnotation.enum_values for a custom enum, e.g. "L1".
+            The server resolves enum values to their numbers. Matching is case-sensitive; unknown values are rejected.
+            Generated protobuf names are not accepted here unless they are also declared public values.
+
             ```proto
             optional string string_value = 5;
             ```
@@ -353,19 +417,13 @@ class FieldQueryValue(Message[_FieldQueryValueFields]):
             ```proto
             optional google.protobuf.Duration duration_value = 7;
             ```
-        enum_name:
-            Symbolic value from the field's enum descriptor.
-
-            ```proto
-            optional string enum_name = 8;
-            ```
         bytes_value:
             ```proto
             optional bytes bytes_value = 9;
             ```
     """
 
-    __slots__ = ("bool_value", "int64_value", "uint64_value", "double_value", "string_value", "timestamp_value", "duration_value", "enum_name", "bytes_value")
+    __slots__ = ("bool_value", "int64_value", "uint64_value", "double_value", "string_value", "timestamp_value", "duration_value", "bytes_value")
 
     if TYPE_CHECKING:
 
@@ -379,7 +437,6 @@ def __init__(
             string_value: str | None = None,
             timestamp_value: Timestamp | None = None,
             duration_value: Duration | None = None,
-            enum_name: str | None = None,
             bytes_value: bytes | None = None,
         ) -> None:
             pass
@@ -391,7 +448,6 @@ def __init__(
         string_value: str
         timestamp_value: Timestamp | None
         duration_value: Duration | None
-        enum_name: str
         bytes_value: bytes
 
 _SpatialFilterFields: TypeAlias = Literal["geometry", "mode", "coordinate_system"]
@@ -546,6 +602,37 @@ def __init__(
         data: RepeatedAny | None
         next_page: Pagination | None
 
+class FieldArrayOperator(Enum):
+    """
+    FieldArrayOperator selects a CQL2 set-membership operation.
+
+    ```proto
+    enum datasets.v1.FieldArrayOperator
+    ```
+
+    Attributes:
+        UNSPECIFIED:
+            ```proto
+            FIELD_ARRAY_OPERATOR_UNSPECIFIED = 0
+            ```
+        CONTAINS_ALL:
+            CQL2 A_CONTAINS: the field contains every requested value.
+
+            ```proto
+            FIELD_ARRAY_OPERATOR_CONTAINS_ALL = 1
+            ```
+        OVERLAPS:
+            CQL2 A_OVERLAPS: the field contains at least one requested value.
+
+            ```proto
+            FIELD_ARRAY_OPERATOR_OVERLAPS = 2
+            ```
+    """
+
+    UNSPECIFIED = 0
+    CONTAINS_ALL = 1
+    OVERLAPS = 2
+
 class LogicalOperator(Enum):
     """
     LogicalOperator specifies how nested filter expressions are combined.
@@ -701,7 +788,7 @@ class SpatialCoordinateSystem(Enum):
 
 
 _DESC = file_desc(
-    b'\n\x1ddatasets/v1/data_access.proto\x12\x0bdatasets.v1\x1a\x16datasets/v1/core.proto\x1a"datasets/v1/well_known_types.proto\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x13tilebox/v1/id.proto\x1a\x16tilebox/v1/query.proto"\xb5\x01\n\x10QueryByIDRequest\x12-\n\ndataset_id\x18\x04 \x01(\x0b2\x0e.tilebox.v1.IDR\tdatasetId\x125\n\x0ecollection_ids\x18\x01 \x03(\x0b2\x0e.tilebox.v1.IDR\rcollectionIds\x12\x1e\n\x02id\x18\x02 \x01(\x0b2\x0e.tilebox.v1.IDR\x02id\x12\x1b\n\tskip_data\x18\x03 \x01(\x08R\x08skipData"\x98\x02\n\x0cQueryFilters\x12=\n\rtime_interval\x18\x01 \x01(\x0b2\x18.tilebox.v1.TimeIntervalR\x0ctimeInterval\x12E\n\x12datapoint_interval\x18\x02 \x01(\x0b2\x16.tilebox.v1.IDIntervalR\x11datapointInterval\x12A\n\x0espatial_extent\x18\x03 \x01(\x0b2\x1a.datasets.v1.SpatialFilterR\rspatialExtent\x12?\n\x0bexpressions\x18\x04 \x03(\x0b2\x1d.datasets.v1.FilterExpressionR\x0bexpressions"\xc0\x01\n\x10FilterExpression\x128\n\x07logical\x18\x01 \x01(\x0b2\x1e.datasets.v1.LogicalExpressionR\x07logical\x12<\n\ncomparison\x18\x02 \x01(\x0b2\x1c.datasets.v1.FieldComparisonR\ncomparison\x124\n\x07is_null\x18\x03 \x01(\x0b2\x1b.datasets.v1.FieldNullCheckR\x06isNull"\x88\x01\n\x11LogicalExpression\x128\n\x08operator\x18\x01 \x01(\x0e2\x1c.datasets.v1.LogicalOperatorR\x08operator\x129\n\x08operands\x18\x02 \x03(\x0b2\x1d.datasets.v1.FilterExpressionR\x08operands"\xa6\x01\n\x0fFieldComparison\x12\x1d\n\nfield_name\x18\x01 \x01(\tR\tfieldName\x12@\n\x08operator\x18\x02 \x01(\x0e2$.datasets.v1.FieldComparisonOperatorR\x08operator\x122\n\x05value\x18\x03 \x01(\x0b2\x1c.datasets.v1.FieldQueryValueR\x05value"/\n\x0eFieldNullCheck\x12\x1d\n\nfield_name\x18\x01 \x01(\tR\tfieldName"\xb0\x03\n\x0fFieldQueryValue\x12$\n\nbool_value\x18\x01 \x01(\x08R\tboolValueB\x05\xaa\x01\x02\x08\x01\x12&\n\x0bint64_value\x18\x02 \x01(\x03R\nint64ValueB\x05\xaa\x01\x02\x08\x01\x12(\n\x0cuint64_value\x18\x03 \x01(\x04R\x0buint64ValueB\x05\xaa\x01\x02\x08\x01\x12(\n\x0cdouble_value\x18\x04 \x01(\x01R\x0bdoubleValueB\x05\xaa\x01\x02\x08\x01\x12(\n\x0cstring_value\x18\x05 \x01(\tR\x0bstringValueB\x05\xaa\x01\x02\x08\x01\x12C\n\x0ftimestamp_value\x18\x06 \x01(\x0b2\x1a.google.protobuf.TimestampR\x0etimestampValue\x12@\n\x0eduration_value\x18\x07 \x01(\x0b2\x19.google.protobuf.DurationR\rdurationValue\x12"\n\tenum_name\x18\x08 \x01(\tR\x08enumNameB\x05\xaa\x01\x02\x08\x01\x12&\n\x0bbytes_value\x18\t \x01(\x0cR\nbytesValueB\x05\xaa\x01\x02\x08\x01"\xc9\x01\n\rSpatialFilter\x121\n\x08geometry\x18\x01 \x01(\x0b2\x15.datasets.v1.GeometryR\x08geometry\x122\n\x04mode\x18\x02 \x01(\x0e2\x1e.datasets.v1.SpatialFilterModeR\x04mode\x12Q\n\x11coordinate_system\x18\x03 \x01(\x0e2$.datasets.v1.SpatialCoordinateSystemR\x10coordinateSystem"\xf9\x01\n\x0cQueryRequest\x12-\n\ndataset_id\x18\x05 \x01(\x0b2\x0e.tilebox.v1.IDR\tdatasetId\x125\n\x0ecollection_ids\x18\x01 \x03(\x0b2\x0e.tilebox.v1.IDR\rcollectionIds\x123\n\x07filters\x18\x02 \x01(\x0b2\x19.datasets.v1.QueryFiltersR\x07filters\x121\n\x04page\x18\x03 \x01(\x0b2\x16.tilebox.v1.PaginationR\x04pageB\x05\xaa\x01\x02\x08\x01\x12\x1b\n\tskip_data\x18\x04 \x01(\x08R\x08skipData"{\n\x0fQueryResultPage\x12,\n\x04data\x18\x01 \x01(\x0b2\x18.datasets.v1.RepeatedAnyR\x04data\x12:\n\tnext_page\x18\x02 \x01(\x0b2\x16.tilebox.v1.PaginationR\x08nextPageB\x05\xaa\x01\x02\x08\x01*\x80\x01\n\x0fLogicalOperator\x12 \n\x1cLOGICAL_OPERATOR_UNSPECIFIED\x10\x00\x12\x18\n\x14LOGICAL_OPERATOR_AND\x10\x01\x12\x17\n\x13LOGICAL_OPERATOR_OR\x10\x02\x12\x18\n\x14LOGICAL_OPERATOR_NOT\x10\x03*\xce\x02\n\x17FieldComparisonOperator\x12)\n%FIELD_COMPARISON_OPERATOR_UNSPECIFIED\x10\x00\x12#\n\x1fFIELD_COMPARISON_OPERATOR_EQUAL\x10\x01\x12\'\n#FIELD_COMPARISON_OPERATOR_NOT_EQUAL\x10\x02\x12\'\n#FIELD_COMPARISON_OPERATOR_LESS_THAN\x10\x03\x120\n,FIELD_COMPARISON_OPERATOR_LESS_THAN_OR_EQUAL\x10\x04\x12*\n&FIELD_COMPARISON_OPERATOR_GREATER_THAN\x10\x05\x123\n/FIELD_COMPARISON_OPERATOR_GREATER_THAN_OR_EQUAL\x10\x06*\xc0\x01\n\x11SpatialFilterMode\x12#\n\x1fSPATIAL_FILTER_MODE_UNSPECIFIED\x10\x00\x12"\n\x1eSPATIAL_FILTER_MODE_INTERSECTS\x10\x01\x120\n,SPATIAL_FILTER_MODE_FILTER_CONTAINS_GEOMETRY\x10\x02\x120\n,SPATIAL_FILTER_MODE_GEOMETRY_CONTAINS_FILTER\x10\x03*\x96\x01\n\x17SpatialCoordinateSystem\x12)\n%SPATIAL_COORDINATE_SYSTEM_UNSPECIFIED\x10\x00\x12\'\n#SPATIAL_COORDINATE_SYSTEM_CARTESIAN\x10\x01\x12\'\n#SPATIAL_COORDINATE_SYSTEM_SPHERICAL\x10\x022\x93\x01\n\x11DataAccessService\x12<\n\tQueryByID\x12\x1d.datasets.v1.QueryByIDRequest\x1a\x10.datasets.v1.Any\x12@\n\x05Query\x12\x19.datasets.v1.QueryRequest\x1a\x1c.datasets.v1.QueryResultPageB\x05\x92\x03\x02\x08\x02b\x08editionsp\xe8\x07',
+    b'\n\x1ddatasets/v1/data_access.proto\x12\x0bdatasets.v1\x1a\x16datasets/v1/core.proto\x1a"datasets/v1/well_known_types.proto\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x13tilebox/v1/id.proto\x1a\x16tilebox/v1/query.proto"\xb5\x01\n\x10QueryByIDRequest\x12-\n\ndataset_id\x18\x04 \x01(\x0b2\x0e.tilebox.v1.IDR\tdatasetId\x125\n\x0ecollection_ids\x18\x01 \x03(\x0b2\x0e.tilebox.v1.IDR\rcollectionIds\x12\x1e\n\x02id\x18\x02 \x01(\x0b2\x0e.tilebox.v1.IDR\x02id\x12\x1b\n\tskip_data\x18\x03 \x01(\x08R\x08skipData"\x98\x02\n\x0cQueryFilters\x12=\n\rtime_interval\x18\x01 \x01(\x0b2\x18.tilebox.v1.TimeIntervalR\x0ctimeInterval\x12E\n\x12datapoint_interval\x18\x02 \x01(\x0b2\x16.tilebox.v1.IDIntervalR\x11datapointInterval\x12A\n\x0espatial_extent\x18\x03 \x01(\x0b2\x1a.datasets.v1.SpatialFilterR\rspatialExtent\x12?\n\x0bexpressions\x18\x04 \x03(\x0b2\x1d.datasets.v1.FilterExpressionR\x0bexpressions"\x8b\x02\n\x10FilterExpression\x128\n\x07logical\x18\x01 \x01(\x0b2\x1e.datasets.v1.LogicalExpressionR\x07logical\x12<\n\ncomparison\x18\x02 \x01(\x0b2\x1c.datasets.v1.FieldComparisonR\ncomparison\x124\n\x07is_null\x18\x03 \x01(\x0b2\x1b.datasets.v1.FieldNullCheckR\x06isNull\x12I\n\x0farray_predicate\x18\x04 \x01(\x0b2 .datasets.v1.FieldArrayPredicateR\x0earrayPredicate"\x89\x01\n\x13FieldArrayPredicate\x12\x1d\n\nfield_name\x18\x01 \x01(\tR\tfieldName\x12;\n\x08operator\x18\x02 \x01(\x0e2\x1f.datasets.v1.FieldArrayOperatorR\x08operator\x12\x16\n\x06values\x18\x03 \x03(\tR\x06values"\x88\x01\n\x11LogicalExpression\x128\n\x08operator\x18\x01 \x01(\x0e2\x1c.datasets.v1.LogicalOperatorR\x08operator\x129\n\x08operands\x18\x02 \x03(\x0b2\x1d.datasets.v1.FilterExpressionR\x08operands"\xa6\x01\n\x0fFieldComparison\x12\x1d\n\nfield_name\x18\x01 \x01(\tR\tfieldName\x12@\n\x08operator\x18\x02 \x01(\x0e2$.datasets.v1.FieldComparisonOperatorR\x08operator\x122\n\x05value\x18\x03 \x01(\x0b2\x1c.datasets.v1.FieldQueryValueR\x05value"/\n\x0eFieldNullCheck\x12\x1d\n\nfield_name\x18\x01 \x01(\tR\tfieldName"\x9d\x03\n\x0fFieldQueryValue\x12$\n\nbool_value\x18\x01 \x01(\x08R\tboolValueB\x05\xaa\x01\x02\x08\x01\x12&\n\x0bint64_value\x18\x02 \x01(\x03R\nint64ValueB\x05\xaa\x01\x02\x08\x01\x12(\n\x0cuint64_value\x18\x03 \x01(\x04R\x0buint64ValueB\x05\xaa\x01\x02\x08\x01\x12(\n\x0cdouble_value\x18\x04 \x01(\x01R\x0bdoubleValueB\x05\xaa\x01\x02\x08\x01\x12(\n\x0cstring_value\x18\x05 \x01(\tR\x0bstringValueB\x05\xaa\x01\x02\x08\x01\x12C\n\x0ftimestamp_value\x18\x06 \x01(\x0b2\x1a.google.protobuf.TimestampR\x0etimestampValue\x12@\n\x0eduration_value\x18\x07 \x01(\x0b2\x19.google.protobuf.DurationR\rdurationValue\x12&\n\x0bbytes_value\x18\t \x01(\x0cR\nbytesValueB\x05\xaa\x01\x02\x08\x01J\x04\x08\x08\x10\tR\tenum_name"\xc9\x01\n\rSpatialFilter\x121\n\x08geometry\x18\x01 \x01(\x0b2\x15.datasets.v1.GeometryR\x08geometry\x122\n\x04mode\x18\x02 \x01(\x0e2\x1e.datasets.v1.SpatialFilterModeR\x04mode\x12Q\n\x11coordinate_system\x18\x03 \x01(\x0e2$.datasets.v1.SpatialCoordinateSystemR\x10coordinateSystem"\xf9\x01\n\x0cQueryRequest\x12-\n\ndataset_id\x18\x05 \x01(\x0b2\x0e.tilebox.v1.IDR\tdatasetId\x125\n\x0ecollection_ids\x18\x01 \x03(\x0b2\x0e.tilebox.v1.IDR\rcollectionIds\x123\n\x07filters\x18\x02 \x01(\x0b2\x19.datasets.v1.QueryFiltersR\x07filters\x121\n\x04page\x18\x03 \x01(\x0b2\x16.tilebox.v1.PaginationR\x04pageB\x05\xaa\x01\x02\x08\x01\x12\x1b\n\tskip_data\x18\x04 \x01(\x08R\x08skipData"{\n\x0fQueryResultPage\x12,\n\x04data\x18\x01 \x01(\x0b2\x18.datasets.v1.RepeatedAnyR\x04data\x12:\n\tnext_page\x18\x02 \x01(\x0b2\x16.tilebox.v1.PaginationR\x08nextPageB\x05\xaa\x01\x02\x08\x01*\x84\x01\n\x12FieldArrayOperator\x12$\n FIELD_ARRAY_OPERATOR_UNSPECIFIED\x10\x00\x12%\n!FIELD_ARRAY_OPERATOR_CONTAINS_ALL\x10\x01\x12!\n\x1dFIELD_ARRAY_OPERATOR_OVERLAPS\x10\x02*\x80\x01\n\x0fLogicalOperator\x12 \n\x1cLOGICAL_OPERATOR_UNSPECIFIED\x10\x00\x12\x18\n\x14LOGICAL_OPERATOR_AND\x10\x01\x12\x17\n\x13LOGICAL_OPERATOR_OR\x10\x02\x12\x18\n\x14LOGICAL_OPERATOR_NOT\x10\x03*\xce\x02\n\x17FieldComparisonOperator\x12)\n%FIELD_COMPARISON_OPERATOR_UNSPECIFIED\x10\x00\x12#\n\x1fFIELD_COMPARISON_OPERATOR_EQUAL\x10\x01\x12\'\n#FIELD_COMPARISON_OPERATOR_NOT_EQUAL\x10\x02\x12\'\n#FIELD_COMPARISON_OPERATOR_LESS_THAN\x10\x03\x120\n,FIELD_COMPARISON_OPERATOR_LESS_THAN_OR_EQUAL\x10\x04\x12*\n&FIELD_COMPARISON_OPERATOR_GREATER_THAN\x10\x05\x123\n/FIELD_COMPARISON_OPERATOR_GREATER_THAN_OR_EQUAL\x10\x06*\xc0\x01\n\x11SpatialFilterMode\x12#\n\x1fSPATIAL_FILTER_MODE_UNSPECIFIED\x10\x00\x12"\n\x1eSPATIAL_FILTER_MODE_INTERSECTS\x10\x01\x120\n,SPATIAL_FILTER_MODE_FILTER_CONTAINS_GEOMETRY\x10\x02\x120\n,SPATIAL_FILTER_MODE_GEOMETRY_CONTAINS_FILTER\x10\x03*\x96\x01\n\x17SpatialCoordinateSystem\x12)\n%SPATIAL_COORDINATE_SYSTEM_UNSPECIFIED\x10\x00\x12\'\n#SPATIAL_COORDINATE_SYSTEM_CARTESIAN\x10\x01\x12\'\n#SPATIAL_COORDINATE_SYSTEM_SPHERICAL\x10\x022\x93\x01\n\x11DataAccessService\x12<\n\tQueryByID\x12\x1d.datasets.v1.QueryByIDRequest\x1a\x10.datasets.v1.Any\x12@\n\x05Query\x12\x19.datasets.v1.QueryRequest\x1a\x1c.datasets.v1.QueryResultPageB\x05\x92\x03\x02\x08\x02b\x08editionsp\xe8\x07',
     [
         core_pb.desc(),
         well_known_types_pb.desc(),
@@ -714,6 +801,7 @@ class SpatialCoordinateSystem(Enum):
         "QueryByIDRequest": QueryByIDRequest,
         "QueryFilters": QueryFilters,
         "FilterExpression": FilterExpression,
+        "FieldArrayPredicate": FieldArrayPredicate,
         "LogicalExpression": LogicalExpression,
         "FieldComparison": FieldComparison,
         "FieldNullCheck": FieldNullCheck,
@@ -721,6 +809,7 @@ class SpatialCoordinateSystem(Enum):
         "SpatialFilter": SpatialFilter,
         "QueryRequest": QueryRequest,
         "QueryResultPage": QueryResultPage,
+        "FieldArrayOperator": FieldArrayOperator,
         "LogicalOperator": LogicalOperator,
         "FieldComparisonOperator": FieldComparisonOperator,
         "SpatialFilterMode": SpatialFilterMode,
diff --git a/tilebox-datasets/tilebox/datasets/v1/dataset_type_pb.py b/tilebox-datasets/tilebox/datasets/v1/dataset_type_pb.py
index 8a4d2e7..a64ddd0 100644
--- a/tilebox-datasets/tilebox/datasets/v1/dataset_type_pb.py
+++ b/tilebox-datasets/tilebox/datasets/v1/dataset_type_pb.py
@@ -39,14 +39,14 @@ class Field(Message[_FieldFields]):
     Attributes:
         descriptor:
             The descriptor contains the name of the field, the type, optional labels (e.g. repeated) and other information.
-            If the type is TYPE_MESSAGE, then the type_name must be a fully qualified name to a well known type, e.g.
-            `datasets.v1.Vec3` or `google.protobuf.Timestamp`.
+            For built-in message and enum types, set type_name to the full name with a leading dot, e.g. `.datasets.v1.Vec3`.
+            For a custom enum, set TYPE_ENUM, leave type_name empty, and provide annotation.enum_values.
 
             ```proto
             optional google.protobuf.FieldDescriptorProto descriptor = 1;
             ```
         annotation:
-            Additional documentation, source mapping, and query metadata for the field.
+            Documentation, source mapping, query metadata, and custom enum values for the field.
 
             ```proto
             optional datasets.v1.FieldAnnotation annotation = 2;
@@ -68,7 +68,39 @@ def __init__(
         descriptor: FieldDescriptorProto | None
         annotation: FieldAnnotation | None
 
-_FieldAnnotationFields: TypeAlias = Literal["description", "example_value", "source_json_pointer", "queryable", "json_schema_ref", "roles"]
+_EnumValueFields: TypeAlias = Literal["value"]
+
+class EnumValue(Message[_EnumValueFields]):
+    """
+    EnumValue is one allowed value of a custom enum.
+
+    ```proto
+    message datasets.v1.EnumValue
+    ```
+
+    Attributes:
+        value:
+            The value users ingest, query, and receive, e.g. "L1".
+
+            ```proto
+            string value = 1;
+            ```
+    """
+
+    __slots__ = ("value",)
+
+    if TYPE_CHECKING:
+
+        def __init__(
+            self,
+            *,
+            value: str = "",
+        ) -> None:
+            pass
+
+        value: str
+
+_FieldAnnotationFields: TypeAlias = Literal["description", "example_value", "source_json_pointer", "queryable", "json_schema_ref", "roles", "enum_values"]
 
 class FieldAnnotation(Message[_FieldAnnotationFields]):
     """
@@ -97,6 +129,7 @@ class FieldAnnotation(Message[_FieldAnnotationFields]):
         queryable:
             Whether the field is projected into query storage and may be used in server-side filter expressions.
             Queryable fields are not necessarily backed by a secondary database index.
+            Enum fields do not count toward the queryable string field limit.
 
             ```proto
             bool queryable = 4;
@@ -115,9 +148,25 @@ class FieldAnnotation(Message[_FieldAnnotationFields]):
             ```proto
             repeated datasets.v1.FieldRole roles = 6 [packed = true];
             ```
+        enum_values:
+            Allowed values for a custom enum field. Values must be unique, with at most 100 entries; leave empty for other types.
+            The server generates the protobuf enum names and numbers. Clients do not need to know its naming rules.
+            Empty datasets allow replacing, removing, renaming, or reordering values, just like other breaking schema changes.
+            Non-empty datasets only allow appending values without changing existing numbers.
+
+            In responses, entries match the enum descriptor's members in order, excluding the generated zero member.
+            Pair each entry with the corresponding member's number to encode and decode datapoints. List positions
+            are not enum numbers. The zero member is not a public value; missing fields remain distinct from zero.
+            Repeated enum fields are deduplicated at ingestion, preserving first-occurrence order in the stored payload.
+            Every element must be a declared nonzero member, so the stored list has at most 100 elements.
+            Omitted and empty repeated fields are equivalent. Queryable repeated enums support FieldArrayPredicate.
+
+            ```proto
+            repeated datasets.v1.EnumValue enum_values = 7;
+            ```
     """
 
-    __slots__ = ("description", "example_value", "source_json_pointer", "queryable", "json_schema_ref", "roles")
+    __slots__ = ("description", "example_value", "source_json_pointer", "queryable", "json_schema_ref", "roles", "enum_values")
 
     if TYPE_CHECKING:
 
@@ -130,6 +179,7 @@ def __init__(
             queryable: bool = False,
             json_schema_ref: str | None = None,
             roles: list[FieldRole] | None = None,
+            enum_values: list[EnumValue] | None = None,
         ) -> None:
             pass
 
@@ -139,6 +189,7 @@ def __init__(
         queryable: bool
         json_schema_ref: str
         roles: list[FieldRole]
+        enum_values: list[EnumValue]
 
 _DatasetTypeFields: TypeAlias = Literal["kind", "fields"]
 
@@ -194,11 +245,13 @@ class AnnotatedType(Message[_AnnotatedTypeFields]):
 
     Attributes:
         descriptor_set:
+            Descriptors for the dataset's datapoint message, including its generated enums.
+
             ```proto
             optional google.protobuf.FileDescriptorSet descriptor_set = 1;
             ```
         type_url:
-            the url of the type, one of the types defined in the descriptor
+            Identifies the datapoint message in descriptor_set by name, independent of file or declaration order.
 
             ```proto
             string type_url = 2;
@@ -402,7 +455,7 @@ class FieldRole(Enum):
 
 
 _DESC = file_desc(
-    b'\n\x1edatasets/v1/dataset_type.proto\x12\x0bdatasets.v1\x1a"datasets/v1/well_known_types.proto\x1a google/protobuf/descriptor.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x13tilebox/v1/id.proto"\x8c\x01\n\x05Field\x12E\n\ndescriptor\x18\x01 \x01(\x0b2%.google.protobuf.FieldDescriptorProtoR\ndescriptor\x12<\n\nannotation\x18\x02 \x01(\x0b2\x1c.datasets.v1.FieldAnnotationR\nannotation"\x8a\x02\n\x0fFieldAnnotation\x12 \n\x0bdescription\x18\x01 \x01(\tR\x0bdescription\x12#\n\rexample_value\x18\x02 \x01(\tR\x0cexampleValue\x125\n\x13source_json_pointer\x18\x03 \x01(\tR\x11sourceJsonPointerB\x05\xaa\x01\x02\x08\x01\x12\x1c\n\tqueryable\x18\x04 \x01(\x08R\tqueryable\x12-\n\x0fjson_schema_ref\x18\x05 \x01(\tR\rjsonSchemaRefB\x05\xaa\x01\x02\x08\x01\x12,\n\x05roles\x18\x06 \x03(\x0e2\x16.datasets.v1.FieldRoleR\x05roles"g\n\x0bDatasetType\x12,\n\x04kind\x18\x01 \x01(\x0e2\x18.datasets.v1.DatasetKindR\x04kind\x12*\n\x06fields\x18\x02 \x03(\x0b2\x12.datasets.v1.FieldR\x06fields"\xf4\x01\n\rAnnotatedType\x12I\n\x0edescriptor_set\x18\x01 \x01(\x0b2".google.protobuf.FileDescriptorSetR\rdescriptorSet\x12\x19\n\x08type_url\x18\x02 \x01(\tR\x07typeUrl\x12I\n\x11field_annotations\x18\x04 \x03(\x0b2\x1c.datasets.v1.FieldAnnotationR\x10fieldAnnotations\x12,\n\x04kind\x18\x05 \x01(\x0e2\x18.datasets.v1.DatasetKindR\x04kindJ\x04\x08\x03\x10\x04"\xa6\x01\n\x11TemporalDatapoint\x12.\n\x04time\x18\x01 \x01(\x0b2\x1a.google.protobuf.TimestampR\x04time\x12\x1e\n\x02id\x18\x02 \x01(\x0b2\x0e.tilebox.v1.IDR\x02id\x12A\n\x0eingestion_time\x18\x03 \x01(\x0b2\x1a.google.protobuf.TimestampR\ringestionTime"\xdf\x01\n\x17SpatioTemporalDatapoint\x12.\n\x04time\x18\x01 \x01(\x0b2\x1a.google.protobuf.TimestampR\x04time\x12\x1e\n\x02id\x18\x02 \x01(\x0b2\x0e.tilebox.v1.IDR\x02id\x12A\n\x0eingestion_time\x18\x03 \x01(\x0b2\x1a.google.protobuf.TimestampR\ringestionTime\x121\n\x08geometry\x18\x04 \x01(\x0b2\x15.datasets.v1.GeometryR\x08geometry*g\n\x0bDatasetKind\x12\x1c\n\x18DATASET_KIND_UNSPECIFIED\x10\x00\x12\x19\n\x15DATASET_KIND_TEMPORAL\x10\x01\x12\x1f\n\x1bDATASET_KIND_SPATIOTEMPORAL\x10\x02*E\n\tFieldRole\x12\x1a\n\x16FIELD_ROLE_UNSPECIFIED\x10\x00\x12\x1c\n\x18FIELD_ROLE_PRIMARY_TITLE\x10\x01B\x05\x92\x03\x02\x08\x02b\x08editionsp\xe8\x07',
+    b'\n\x1edatasets/v1/dataset_type.proto\x12\x0bdatasets.v1\x1a"datasets/v1/well_known_types.proto\x1a google/protobuf/descriptor.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a\x13tilebox/v1/id.proto"\x8c\x01\n\x05Field\x12E\n\ndescriptor\x18\x01 \x01(\x0b2%.google.protobuf.FieldDescriptorProtoR\ndescriptor\x12<\n\nannotation\x18\x02 \x01(\x0b2\x1c.datasets.v1.FieldAnnotationR\nannotation"!\n\tEnumValue\x12\x14\n\x05value\x18\x01 \x01(\tR\x05value"\xc3\x02\n\x0fFieldAnnotation\x12 \n\x0bdescription\x18\x01 \x01(\tR\x0bdescription\x12#\n\rexample_value\x18\x02 \x01(\tR\x0cexampleValue\x125\n\x13source_json_pointer\x18\x03 \x01(\tR\x11sourceJsonPointerB\x05\xaa\x01\x02\x08\x01\x12\x1c\n\tqueryable\x18\x04 \x01(\x08R\tqueryable\x12-\n\x0fjson_schema_ref\x18\x05 \x01(\tR\rjsonSchemaRefB\x05\xaa\x01\x02\x08\x01\x12,\n\x05roles\x18\x06 \x03(\x0e2\x16.datasets.v1.FieldRoleR\x05roles\x127\n\x0benum_values\x18\x07 \x03(\x0b2\x16.datasets.v1.EnumValueR\nenumValues"g\n\x0bDatasetType\x12,\n\x04kind\x18\x01 \x01(\x0e2\x18.datasets.v1.DatasetKindR\x04kind\x12*\n\x06fields\x18\x02 \x03(\x0b2\x12.datasets.v1.FieldR\x06fields"\xf4\x01\n\rAnnotatedType\x12I\n\x0edescriptor_set\x18\x01 \x01(\x0b2".google.protobuf.FileDescriptorSetR\rdescriptorSet\x12\x19\n\x08type_url\x18\x02 \x01(\tR\x07typeUrl\x12I\n\x11field_annotations\x18\x04 \x03(\x0b2\x1c.datasets.v1.FieldAnnotationR\x10fieldAnnotations\x12,\n\x04kind\x18\x05 \x01(\x0e2\x18.datasets.v1.DatasetKindR\x04kindJ\x04\x08\x03\x10\x04"\xa6\x01\n\x11TemporalDatapoint\x12.\n\x04time\x18\x01 \x01(\x0b2\x1a.google.protobuf.TimestampR\x04time\x12\x1e\n\x02id\x18\x02 \x01(\x0b2\x0e.tilebox.v1.IDR\x02id\x12A\n\x0eingestion_time\x18\x03 \x01(\x0b2\x1a.google.protobuf.TimestampR\ringestionTime"\xdf\x01\n\x17SpatioTemporalDatapoint\x12.\n\x04time\x18\x01 \x01(\x0b2\x1a.google.protobuf.TimestampR\x04time\x12\x1e\n\x02id\x18\x02 \x01(\x0b2\x0e.tilebox.v1.IDR\x02id\x12A\n\x0eingestion_time\x18\x03 \x01(\x0b2\x1a.google.protobuf.TimestampR\ringestionTime\x121\n\x08geometry\x18\x04 \x01(\x0b2\x15.datasets.v1.GeometryR\x08geometry*g\n\x0bDatasetKind\x12\x1c\n\x18DATASET_KIND_UNSPECIFIED\x10\x00\x12\x19\n\x15DATASET_KIND_TEMPORAL\x10\x01\x12\x1f\n\x1bDATASET_KIND_SPATIOTEMPORAL\x10\x02*E\n\tFieldRole\x12\x1a\n\x16FIELD_ROLE_UNSPECIFIED\x10\x00\x12\x1c\n\x18FIELD_ROLE_PRIMARY_TITLE\x10\x01B\x05\x92\x03\x02\x08\x02b\x08editionsp\xe8\x07',
     [
         well_known_types_pb.desc(),
         descriptor_pb.desc(),
@@ -411,6 +464,7 @@ class FieldRole(Enum):
     ],
     {
         "Field": Field,
+        "EnumValue": EnumValue,
         "FieldAnnotation": FieldAnnotation,
         "DatasetType": DatasetType,
         "AnnotatedType": AnnotatedType,