[chore] Update tutorials & bump version to 0.2.0

transfer_queue/version/version

@@ -1 +1 @@
-0.1.12.dev0
+0.2.0

tutorial/02_kv_interface.py

@@ -146,25 +146,39 @@ def demonstrate_kv_api():
 
     # Step 4: Only update tags through kv_put
     print("\n[Step 4] Update existing tags without providing value...")
+    print("  Tags hold application metadata; TQ tracks field readiness separately.")
     key_for_update_tags = "0_0"
     tag_update = {"global_steps": 0, "status": "finish", "model_version": 0}
     print(f"  Target Key: {key_for_update_tags}")
     print(f"  The updated tag is: {tag_update}")
     tq.kv_put(key=key_for_update_tags, partition_id=partition_id, fields=None, tag=tag_update)
     print(f"  ✓ Update success: Samples '0_0' now has tag as {tag_update}.")
 
-    # Step 5: Concatenate a field with kv_update, then clear only that field
-    print("\n[Step 5] Updating a field with kv_update (concat, then clear the field)...")
-    tq.kv_put(key=key, partition_id=partition_id, fields={"scratch": torch.tensor([1, 2])})
+    response_key = keys[0]
+    print(f"\n[Step 5] Building a response for '{response_key}' in chunks...")
+    print("  kv_update requires SimpleStorage; use one writer per key or coordinate writers with a KV lock.")
+    print("  A timed-out merge may have committed; retrying it blindly can append the same chunk twice.")
     tq.kv_update(
-        key=key,
+        key=response_key,
         partition_id=partition_id,
-        fields={"scratch": torch.tensor([3, 4])},
+        fields={"response": torch.tensor([1, 2])},
         merge_fn=lambda old, new: torch.cat([old, new]),
+        finalize=False,
     )
-    print("  ✓ kv_update concat: scratch of '0_0' is now [1, 2, 3, 4].")
-    tq.kv_clear(keys=key, partition_id=partition_id, fields="scratch")
-    print("  ✓ tq.kv_clear(fields=...): scratch of '0_0' is removed (key and other fields remain).")
+    print("  Stored fields:", tq.kv_describe(response_key, partition_id)[response_key])
+    print("  Ready fields:", tq.kv_exists(response_key, partition_id)[response_key])
+
+    tq.kv_update(
+        key=response_key,
+        partition_id=partition_id,
+        fields={"response": torch.tensor([3, 4])},
+        merge_fn=lambda old, new: torch.cat([old, new]),
+        finalize=True,
+    )
+    response = tq.kv_batch_get(response_key, partition_id, select_fields="response")
+    assert response["response"][0].tolist() == [1, 2, 3, 4]
+    print("  Published response:", response["response"][0])
+    tq.kv_put(response_key, partition_id, tag={"status": "finish"})
 
     # Step 6: List all keys and tags in a partition
     print("\n[Step 6] Listing all keys and tags in partition...")
@@ -183,17 +197,17 @@ def demonstrate_kv_api():
     retrieved_input_ids = tq.kv_batch_get(keys=all_keys, partition_id=partition_id, select_fields="input_ids")
     print(f"  ✓ Successfully retrieved only {list(retrieved_input_ids.keys())} field for all samples.")
 
-    # Step 8: Retrieve all fields using kv_batch_get
-    print("\n[Step 8] Retrieving all fields with kv_batch_get...")
+    print("\n[Step 8] Retrieving the fields ready in every requested key...")
     retrieved_all = tq.kv_batch_get(keys=all_keys, partition_id=partition_id)
-    print(f"  Retrieved all fields for {all_keys}:")
-    print(f"  Fields: {list(retrieved_all.keys())}")
-    print(
-        f"  Note: We cannot retrieve fields {list(response_batch.keys())}, since they only available in {append_keys}"
-    )
+    print(f"  Requested keys: {all_keys}")
+    print(f"  Common ready fields: {list(retrieved_all.keys())}")
+
+    print("\n[Step 9] Clearing selected fields while keeping keys...")
+    tq.kv_clear(all_keys, partition_id, fields="attention_mask")
+    print("  Keys and their other fields remain:", tq.kv_exists(all_keys, partition_id))
 
-    # Step 9: Clear specific keys
-    print("\n[Step 9] Clearing keys from partition...")
+    # Step 10: Clear specific keys
+    print("\n[Step 10] Clearing keys from partition...")
     keys_to_clear = all_keys[:2]  # Delete the first 2 keys
     tq.kv_clear(keys=keys_to_clear, partition_id=partition_id)
     print(f"  ✓ Cleared keys: {keys_to_clear}")
@@ -231,10 +245,6 @@ def main():
         - 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 must manually check status via tags externally.
-        - No built-in Sampler support: Must implement data dispatch by ReplayBuffer or single-controller externally.
-        - Not fully streaming: Consumers must wait for single-controller to dispatch `keys`.
         """
         )
     )
@@ -252,9 +262,11 @@ def main():
         print("=" * 80)
         print("\nKey Takeaways:")
         print("  1. KV API simplifies data access with Redis-style semantics")
-        print("  2. Use 'fields' parameter to get/put specific fields only")
-        print("  3. Tags enable custom metadata for production status, scores, etc.")
-        print("  4. Use kv_list to inspect partition contents")
+        print("  2. Read or write only the fields you need")
+        print("  3. Tags store custom metadata such as status and scores")
+        print("  4. Use `kv_list` to inspect keys and tags")
+        print("  5. Supports incremental updates and publishing completed responses")
+        print("  6. Supports field-level clearing while keeping keys")
 
         # Cleanup
         tq.close()

tutorial/03_metadata_concepts.py

@@ -317,6 +317,30 @@ def demonstrate_real_workflow():
         chunk_data = tq_client.get_data(chunk)
         print(f"  Chunk {i}: Retrieved chunk data: {chunk_data}")
 
+    print("[Step 7] Selecting rows and fields from BatchMeta with shared fields...")
+    video = torch.arange(12, dtype=torch.float32).reshape(3, 4)
+    public_meta = tq_client.put(
+        TensorDict({"video": video.unsqueeze(0)}, batch_size=1),
+        partition_id=tq.PUBLIC_PARTITION,
+    )
+    linked_meta = tq_client.link(batch_meta.select_samples([0, 1]), public_meta.select_samples([0, 0]), fields="video")
+    print("  Linked the first two samples to the same public video field.")
+    # Payloads stay on the public row; _links holds source indexes aligned with global_indexes.
+    # Selection filters these local references without changing the controller's links.
+    print(f"  Rows: {linked_meta.global_indexes}, fields: {linked_meta.field_names}")
+    print(f"  Internal links: {linked_meta._links}")
+
+    selected_meta = linked_meta.select_samples([1]).select_fields(["video"])
+    print("  Selected the second sample and kept only its video field.")
+    print(f"  Selected rows: {selected_meta.global_indexes}, fields: {selected_meta.field_names}")
+    print(f"  Links after selection: {selected_meta._links}")
+    torch.testing.assert_close(tq_client.get_data(selected_meta)["video"][0], video)
+    print("  Reading through the selected metadata returned the original public video.")
+
+    private_meta = linked_meta.select_fields(["input_ids"])
+    print("  Kept only the private input_ids field for both samples; video links are excluded from this metadata.")
+    print(f"  Private fields: {private_meta.field_names}, links: {private_meta._links}")
+
     # Cleanup
     tq_client.clear_partition(partition_id=partition_id)
     tq.close()
@@ -341,6 +365,7 @@ def main():
         - custom_meta: list[dict] aligned with global_indexes (one dict per sample)
         - Metadata operations: chunk, concat, union, select_fields, select_samples, reorder
         - batch.samples[i] returns a lazy view with .fields -> field_schema (read-only)
+        - client.link shares public fields across keyless rows on SimpleStorage
     """
         )
     )

tutorial/06_streaming_dataloader.py

@@ -14,7 +14,7 @@
 # limitations under the License.
 
 """
-Tutorial 5: Streaming DataLoader for Distributed Training
+Tutorial 6: Streaming DataLoader for Distributed Training
 
 This script demonstrates how to use StreamingDataset and StreamingDataLoader
 for efficient streaming data loading in distributed training scenarios.

tutorial/07_shared_fields.py

@@ -0,0 +1,108 @@
+# Copyright 2025 The TransferQueue Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tutorial 7: Shared Fields
+
+Share stored fields across samples through PUBLIC_PARTITION and kv_link. This
+example uses a tensor field named features and a private score for each sample.
+
+Run: python tutorial/07_shared_fields.py
+
+Requires SimpleStorage. Public keys are shared by applications on the same TQ;
+use an application-specific prefix. Linked fields and referenced public rows
+cannot be overwritten.
+"""
+
+import sys
+from pathlib import Path
+
+import ray
+import torch
+from tensordict import TensorDict
+
+sys.path.append(str(Path(__file__).resolve().parent.parent))
+
+import transfer_queue as tq  # noqa: E402
+
+
+def main():
+    """Demonstrate linking, reading and reclaiming shared fields."""
+    print("=" * 80)
+    print("TransferQueue Tutorial 7: Shared Fields")
+    print("=" * 80)
+    print("A field can be stored once and referenced by multiple samples.")
+    print("Each sample can also have its own private fields.")
+    print("Setting up TransferQueue with the default SimpleStorage backend...")
+    ray.init(namespace="TransferQueueTutorial", include_dashboard=False)
+    try:
+        tq.init()
+        partition_id = "train"
+
+        print("\n[Step 1] Storing shared fields in the public partition...")
+        public_values = {
+            "tutorial07/shared_0": torch.arange(24, dtype=torch.float32).reshape(6, 4),
+            "tutorial07/shared_1": torch.arange(24, 48, dtype=torch.float32).reshape(6, 4),
+        }
+        public_keys = list(public_values)
+        for public_key, value in public_values.items():
+            tq.kv_put(public_key, tq.PUBLIC_PARTITION, fields={"features": value})
+            print(f"  Stored {public_key}: features shape={tuple(value.shape)}")
+
+        print("\n[Step 2] Linking sample fields to the shared values...")
+        keys = [f"sample_{i}" for i in range(4)]
+        sources = [public_keys[0], public_keys[0], public_keys[1], public_keys[1]]
+        tq.kv_link(keys, partition_id, public_key=sources, fields="features")
+        descriptions = tq.kv_describe(keys, partition_id)
+        for key in keys:
+            print(f"  {key} reads features from {descriptions[key]['links']['features']}")
+        print("  The links reference the public rows; no features payload is copied into the sample rows.")
+
+        print("\n[Step 3] Adding a private field to each sample...")
+        scores = torch.tensor([0.1, 0.2, 0.3, 0.4])
+        tq.kv_batch_put(keys, partition_id, fields=TensorDict({"score": scores}, batch_size=len(keys)))
+        print("  Added score values:", scores.tolist())
+        print("  Each sample now has a shared features field and its own score field.")
+
+        print("\n[Step 4] Reading shared and private fields together...")
+        batch = tq.kv_batch_get(keys, partition_id)
+        print("  Returned fields:", list(batch.keys()))
+        for i, (key, source) in enumerate(zip(keys, sources, strict=True)):
+            torch.testing.assert_close(batch["features"][i], public_values[source])
+            print(f"  {key}: features shape={tuple(batch['features'][i].shape)}, score={batch['score'][i].item():.1f}")
+
+        print("\n[Step 5] Clearing sample keys and their links...")
+        tq.kv_clear(keys, partition_id)
+        print("  Cleared sample keys:", keys)
+        print("  Public keys still available:", list(tq.kv_exists(public_keys, tq.PUBLIC_PARTITION)))
+
+        print("\n[Step 6] Reclaiming shared data with no remaining links...")
+        freed = tq.kv_clear_unlinked_public()
+        assert set(freed) == set(public_keys)
+        print("  Reclaimed public keys:", freed)
+
+        print("\n" + "=" * 80)
+        print("Tutorial Complete!")
+        print("=" * 80)
+        print("\nKey Takeaways:")
+        print("  1. Multiple samples can share a stored field through links")
+        print("  2. Reads combine shared fields with each sample's private fields")
+        print("  3. Shared data can be reclaimed after its last link is removed")
+    finally:
+        tq.close()
+        ray.shutdown()
+        print("\nCleanup complete")
+
+
+if __name__ == "__main__":
+    main()