Add high-level KV interface for TransferQueue

[Copilot]

TransferQueue's existing APIs expose low-level metadata concepts (BatchMeta, SampleMeta, FieldMeta) directly to users. This PR adds a Redis-style KV layer that requires zero knowledge of internals.

Three API Tiers

Tier	Style	Streaming	Sampler	Use Case
KV Interface (this PR)	Put/Get/List/Clear	✗	✗	Fine-grained access with external dispatch
StreamingDataLoader (#23)	PyTorch DataLoader	✓	✓	Fully-streamed training pipelines
Low-level Client	Metadata-based	✓	✓	Maximum flexibility

Changes

• transfer_queue/client.py — KV methods on AsyncTransferQueueClient + sync wrappers on TransferQueueClient. String keys map to global_index internally; tags stored as custom_meta.
• transfer_queue/interface.py — Top-level tq.kv_* functions
• transfer_queue/__init__.py — Exports
• tutorial/02_kv_interface.py — Tutorial with partition isolation, batch ops, selective clear examples
• tests/e2e/test_kv_interface_e2e.py — E2E coverage for put/get/list/clear, field selection, tags, partition isolation, error paths

API

import transfer_queue as tq

tq.kv_batch_put(
    kv_pairs={
        "s0": {"input_ids": torch.tensor([1, 2]), "reward": torch.tensor([0.5])},
        "s1": {"input_ids": torch.tensor([3, 4]), "reward": torch.tensor([0.8])},
    },
    partition_id="rollout_v1",
    tags={"s0": {"status": "done"}, "s1": {"status": "done"}},
)

results = tq.kv_batch_get(keys=["s0"], fields=["reward"], partition_id="rollout_v1")
entries = tq.kv_list(partition_id="rollout_v1")  # [{"key": "s0", "tag": {"status": "done"}}, ...]
tq.kv_clear(keys=["s0"], partition_id="rollout_v1")

All methods have async_ variants. See tutorial/02_kv_interface.py for full examples.

Limitations vs native APIs

• No built-in production/consumption tracking — use tags
• No built-in sampler — dispatch keys externally
• Not streaming — consumers poll for dispatched keys

Warning

Firewall rules blocked me from connecting to one or more addresses (expand for details)

I tried to connect to the following addresses, but was blocked by firewall rules:

• metadata.google.internal
  • Triggering command: /usr/bin/python /usr/bin/python /home/REDACTED/.local/lib/python3.12/site-packages/ray/dashboard/dashboard.py --host=127.0.0.1 --port=8265 --port-retries=50 --temp-dir=/tmp/ray --log-dir=/tmp/ray/session_2026-02-08_06-44-47_525468_4342/logs --session-dir=/tmp/ray/session_2026-02-08_06-44-47_525468_4342 --logging-rotate-bytes=536870912 --logging-rotate-backup-count=5 --gcs-address=127.0.0.1:64208 --cluster-id-hex=969f0778aa8bb88015485d86f90294f79f53d14b67a48359577abf00 --node-ip-address=127.0.0.1 --stdout-filepath=/tmp/ray/session_2026-02-08_06-44-47_525468_4342/logs/dashboard.out --stderr-filepath=/tmp/ray/session_2026-02-08_06-44-47_525468_4342/logs/dashboard.err (dns block)
  • Triggering command: /usr/bin/python /usr/bin/python /home/REDACTED/.local/lib/python3.12/site-packages/ray/dashboard/dashboard.py --host=127.0.0.1 --port=8265 --port-retries=50 --temp-dir=/tmp/ray --log-dir=/tmp/ray/session_2026-02-08_06-46-19_007332_5049/logs --session-dir=/tmp/ray/session_2026-02-08_06-46-19_007332_5049 --logging-rotate-bytes=536870912 --logging-rotate-backup-count=5 --gcs-address=127.0.0.1:64463 --cluster-id-hex=469e69e4ac12d34d2b3dcd8908377809815623cd27afb5cc925c9727 --node-ip-address=127.0.0.1 --stdout-filepath=/tmp/ray/session_2026-02-08_06-46-19_007332_5049/logs/dashboard.out --stderr-filepath=/tmp/ray/session_2026-02-08_06-46-19_007332_5049/logs/dashboard.err (dns block)

If you need me to access, download, or install something from one of these locations, you can either:

• Configure Actions setup steps to set up my environment, which run before the firewall is enabled
• Add the appropriate URLs or hosts to the custom allowlist in this repository's Copilot coding agent settings (admins only)

Original prompt

请根据PR内容，帮我撰写一个PR的说明。我的初稿是：## Background

In previous versions of TransferQueue, the learning curve is a little bit sharp. Users have to undersand the BatchMeta SampleMeta and FieldMeta design in tutorial/02_metadat_concepts.py, and have to investigate all the flexible usage in TransferQueueClient. Although in previous PR #26, we simplifies the initialization and eliminates the necessary to manually create a TransferQueueClient object, the API is still directly exposed to users.

High-Level Key-Value based API

    Key Methods:
    1. (async_)kv_put          - Insert/Update a multi-column sample by key, with optional metadata tag
    2. (async_)kv_batch_put    - Put multiple key-value pairs efficiently in batch
    3. (async_)kv_batch_get          - Retrieve samples (by keys), supporting column selection (by fields)
    4. (async_)kv_list         - List keys and tags (metadata) in a partition
    5. (async_)kv_clear        - Remove key-value pairs from storage

    Key Features:
    ✓ Redis-style Semantics  - Familiar KV interface (Put/Get/List) for zero learning curve
    ✓ Fine-grained Access    - Update or retrieve specific fields (columns) within a key (row) without full op.
    ✓ Partition Isolation    - Logical separation of storage namespaces
    ✓ Metadata Tags          - Lightweight metadata for status tracking
    ✓ Pluggable Backends     - Supports multiple backends

    Use Cases:
    - Focusing on fine-grained data access where extreme streaming performance is non-essential
    - Integration with external ReplayBuffer/single-controller that manage sample dispatching
    
    Limitations (vs low-level native APIs):
    - No built-in production/consumption tracking: Users have to manually check status through tags.
    - No built-in Sampler support: Must implement data dispatch by ReplayBuffer or single-controller externally.
    - No fully streaming: Consumers must wait for single-controller to dispatch `keys`.

Usage Example

Please refer to tutorial/02_kv_interface.py and tests/e2e/test_kv_interface_e2e.py for details.

请参考这个初稿，结合tutorial/02_kv_interface.py、tests/e2e/test_kv_interface_e2e.py来完善。现在TransferQueue有3套API，高阶API包括：KV接口（本PR），StreamingDataLoader（比如https://github.com/Ascend/TransferQueue/pull/23），低阶API是之前我们在TransferQueueClient里提供的。请在这个PR描述里体现这一特点。

---

💡 You can make Copilot smarter by setting up custom instructions, customizing its development environment and configuring Model Context Protocol (MCP) servers. Learn more Copilot coding agent tips in the docs.