Prevent Arrow offset overflow in DiskGraph temporal output

packages/diskgraph/diskgraph-pybind/README.md

@@ -170,6 +170,14 @@ value when a caller will immediately encode a batch. Sequence `batch` values
 are Arrow `RecordBatch` objects in both formats. Query semantics, ordering,
 limits, reductions, nulls, timings, and statistics are identical.
 
+Temporal row decoding uses Arrow `large_string`, `large_binary`, and `large_list`
+columns, including large string children inside lists. Their 64-bit offsets
+allow a query's combined values to exceed the 2 GiB limit of ordinary Arrow
+string and binary arrays without changing row values or order. This removes an
+offset limit; the complete query still needs enough memory for its output.
+See the [performance assessment](../docs/arrow-offset-performance.md) for
+measured costs and standalone reproduction commands.
+
 `SamplerEngine.query_temporal_lengths(seed_nodes, seed_timestamps, plan_json)`
 is the exact scheduling companion. It visits only frames named by sequences,
 shares a frame traversal between all of its sequence consumers, applies each

packages/diskgraph/diskgraph-pybind/python/examples/arrow_offset_query_bench.py

@@ -0,0 +1,170 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES.
+# All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+# Usage: python arrow_offset_query_bench.py FRESH_INDEX_ROOT LABEL [model|inspectable]
+# Reuse the generated index across baseline/candidate runs. Use a separate
+# installed wheel for each label; the label only identifies the JSON result.
+
+from pathlib import Path
+import json, time, sys, hashlib, statistics
+import diskgraph, pyarrow as pa, pyarrow.parquet as pq
+
+root = Path(sys.argv[1])
+variant = sys.argv[2]
+output_format = sys.argv[3] if len(sys.argv) > 3 else "model"
+root.mkdir(exist_ok=True)
+if not (root / "index").exists():
+    pq.write_table(pa.table({"id": list(range(100))}), root / "users.parquet")
+    n = 10000
+    pq.write_table(
+        pa.table(
+            {
+                "id": list(range(n)),
+                "owner": [i % 100 for i in range(n)],
+                "time": pa.array(
+                    [1700000000 + i // 100 for i in range(n)], type=pa.timestamp("s")
+                ),
+                "text": ["x" * 64 if i % 17 else None for i in range(n)],
+                "value": [float(i) for i in range(n)],
+                "blob": [b"abcdefgh" if i % 17 else None for i in range(n)],
+                "tags": [
+                    ["a", "bb", "ccc", "dddd"] if i % 17 else None for i in range(n)
+                ],
+            }
+        ),
+        root / "orders.parquet",
+    )
+    cfg = {
+        "dataset_name": "offset_perf",
+        "num_shards": 1,
+        "tables": [
+            {
+                "name": "users",
+                "primary_key": "id",
+                "columns": [{"name": "id", "ingested_dtype": "int64"}],
+                "source": {"type": "parquet", "paths": [str(root / "users.parquet")]},
+            },
+            {
+                "name": "orders",
+                "primary_key": "id",
+                "create_time": "time",
+                "columns": [
+                    {"name": k, "ingested_dtype": v}
+                    for k, v in [
+                        ("id", "int64"),
+                        ("owner", "int64"),
+                        ("time", "timestamp"),
+                        ("text", "string"),
+                        ("value", "float64"),
+                        ("blob", "bytes"),
+                        ("tags", "string_array"),
+                    ]
+                ],
+                "fkeys": [
+                    {
+                        "src_column": "owner",
+                        "dst_table": "users",
+                        "reverse": {
+                            "time": {"table": "orders", "column": "time", "unit": "s"}
+                        },
+                    }
+                ],
+                "source": {"type": "parquet", "paths": [str(root / "orders.parquet")]},
+            },
+        ],
+    }
+    diskgraph.ingest(
+        json.dumps(cfg),
+        str(root / "index"),
+        col_stats=False,
+        context_threads=1,
+        encode_threads=1,
+        pk_map_threads=1,
+        edges_map_threads=1,
+        index_threads=1,
+    )
+e = diskgraph.SamplerEngine(str(root / "index"))
+relation = next(
+    x["name"]
+    for x in json.loads(e.get_type_mapping())["edge_relations"]
+    if x["src_type"] == "users"
+)
+plan = {
+    "frames": [
+        {
+            "name": "f",
+            "root_type": "users",
+            "path": [{"relation": relation, "destination_type": "orders"}],
+            "time_column": "time",
+            "start_seconds": -102,
+            "stop_seconds": 0,
+            "distinct": True,
+            "columns": ["id", "text", "value", "blob", "tags"],
+        }
+    ],
+    "sequences": [
+        {
+            "name": "s",
+            "frame": "f",
+            "columns": ["id", "text", "value", "blob", "tags"],
+            "order": "descending",
+            "limit": 32,
+        }
+    ],
+    "aggregates": [],
+}
+ids = e.lookup_node_ids("users", pa.array([i % 100 for i in range(256)]))
+times = [1700000101] * 256
+p = json.dumps(plan)
+
+
+def run():
+    return e.query_temporal(
+        {"users": ids}, {"users": times}, p, output_format=output_format
+    )
+
+
+first = run()
+
+
+# Preserve a value-based output digest separately from timing.
+def normalize(x):
+    if isinstance(x, (pa.RecordBatch, pa.Table, pa.Array, pa.ChunkedArray)):
+        return x.to_pylist()
+    if isinstance(x, dict):
+        return {
+            k: normalize(v) for k, v in x.items() if k not in ("timings", "timings_ns")
+        }
+    if isinstance(x, (tuple, list)):
+        return [normalize(v) for v in x]
+    if hasattr(x, "tolist"):
+        return x.tolist()
+    return x
+
+
+digest = hashlib.sha256(
+    json.dumps(normalize(first), sort_keys=True, default=str).encode()
+).hexdigest()
+del first
+for _ in range(5):
+    run()
+start = time.perf_counter()
+latencies = []
+for _ in range(100):
+    t0 = time.perf_counter()
+    run()
+    latencies.append(time.perf_counter() - t0)
+print(
+    json.dumps(
+        {
+            "variant": variant,
+            "format": output_format,
+            "seconds": time.perf_counter() - start,
+            "queries": 100,
+            "median_query_seconds": statistics.median(latencies),
+            "p95_query_seconds": sorted(latencies)[94],
+            "digest": digest,
+        }
+    )
+)

packages/diskgraph/diskgraph-pybind/python/tests/test_temporal_sequence_limit.py

@@ -683,6 +683,11 @@ def test_shared_history_preserves_types_and_floating_point_order(
     extra = {
         "description": (pa.array([None, "alpha", "beta", "alpha"] * 25), "string"),
         "tags": (pa.array([None, [], ["a", "b"], ["c"]] * 25), "string_array"),
+        "blob": (pa.array([None, b"", b"\x00\xff", b"abc"] * 25), "bytes"),
+        "codes": (
+            pa.array([None, [], [-8, 7], [3]] * 25, type=pa.list_(pa.int8())),
+            "int8_array",
+        ),
         "enabled": (pa.array([True, False, None, True] * 25), "bool"),
     }
     for name, (values, dtype) in extra.items():
@@ -707,6 +712,8 @@ def test_shared_history_preserves_types_and_floating_point_order(
         plan["frames"][0]["columns"] += list(extra)
         plan["sequences"][0]["columns"] = [
             "tags",
+            "blob",
+            "codes",
             "value",
             "description",
             "enabled",
@@ -731,6 +738,13 @@ def test_shared_history_preserves_types_and_floating_point_order(
             for row in range(start, min(start + 3, stop))
         ]
         assert a["batch"].equals(b["batch"].take(pa.array(selected, type=pa.int64())))
+        assert a["batch"].schema.field("description").type == pa.large_string()
+        assert a["batch"].schema.field("blob").type == pa.large_binary()
+        assert a["batch"].schema.field("tags").type == pa.large_list(pa.large_string())
+        assert a["batch"].schema.field("codes").type == pa.large_list(pa.int8())
+        original_rows = {row["id"]: row for row in source.to_pylist()}
+        for row in a["batch"].to_pylist():
+            assert row == {name: original_rows[row["id"]][name] for name in row}
         assert actual["stats"]["values_decoded"] < full["stats"]["values_decoded"]
 
 

packages/diskgraph/diskgraph-pybind/src/temporal_query.rs

@@ -338,7 +338,7 @@ fn scalar_values_to_array(
                 })
                 .collect::<Result<Vec<_>, _>>()?,
         ))),
-        Some(ScalarValue::String(_)) => Ok(Arc::new(StringArray::from(
+        Some(ScalarValue::String(_)) => Ok(Arc::new(LargeStringArray::from(
             values
                 .iter()
                 .map(|value| match value {
@@ -470,7 +470,7 @@ fn table_rows(
         );
     }
     decoder
-        .decode_batch_to_arrow(&rows, projection_indices)
+        .decode_batch_to_arrow_large(&rows, projection_indices)
         .map_err(|error| error.to_string())
 }
 
@@ -1853,6 +1853,22 @@ pub(crate) fn config_error(error: String) -> PyErr {
 mod tests {
     use super::*;
 
+    #[test]
+    fn model_string_aggregates_keep_large_offsets_and_nulls() {
+        let values = [
+            ScalarValue::String("café".into()),
+            ScalarValue::Null,
+            ScalarValue::String(String::new()),
+        ];
+        let actual = scalar_values_to_array(&values, None).unwrap();
+        let strings = actual.as_any().downcast_ref::<LargeStringArray>().unwrap();
+        assert_eq!(
+            strings.iter().collect::<Vec<_>>(),
+            vec![Some("café"), None, Some("")]
+        );
+        assert!(scalar_values_to_array(&[values[0].clone(), ScalarValue::I64(1)], None).is_err());
+    }
+
     #[test]
     fn frame_sort_keeps_existing_tie_order_for_every_scalar_kind() {
         use rand::{seq::SliceRandom, SeedableRng};

packages/diskgraph/diskgraph-rowpack/examples/arrow_offset_bench.rs

@@ -0,0 +1,68 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES.
+// All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use anyhow::Result;
+use diskgraph_rowpack::*;
+use std::{hint::black_box, time::Instant};
+fn main() -> Result<()> {
+    let args: Vec<_> = std::env::args().collect();
+    let kind = &args[1];
+    let n: usize = args[2].parse()?;
+    let repeats: usize = args[3].parse()?;
+    let text = "abcdefgh".repeat(if kind == "long" { 128 } else { 1 });
+    let tokens = vec!["a", "bb", "ccc", "dddd"];
+    let (dtype, value) = match kind.as_str() {
+        "numeric" => (IngestedType::Int64, ValueRef::Int64(42)),
+        "list" => (IngestedType::StringArray, ValueRef::StringArray(&tokens)),
+        "binary" => (IngestedType::Bytes, ValueRef::Bytes(text.as_bytes())),
+        _ => (IngestedType::String, ValueRef::String(&text)),
+    };
+    let columns = vec![RowPackColumn {
+        name: "value".into(),
+        logical_type: dtype,
+        fixed_len: None,
+        precision: None,
+        scale: None,
+    }];
+    let mut cb = RowPackContextBuilder::new(columns, RowPackConfig::default());
+    cb.observe_row(&[value.clone()])?;
+    cb.observe_row(&[ValueRef::Null])?;
+    let ctx = cb.finish()?;
+    let encoder = RowPackEncoder::new(ctx.clone())?;
+    let mut scratch = RowPackScratch::new(&ctx);
+    let mut encoded = Vec::new();
+    encoder.encode_row_into(&[value], &mut encoded, &mut scratch)?;
+    let mut null = Vec::new();
+    encoder.encode_row_into(&[ValueRef::Null], &mut null, &mut scratch)?;
+    let rows: Vec<&[u8]> = (0..n)
+        .map(|i| {
+            if i % 17 == 0 {
+                null.as_slice()
+            } else {
+                encoded.as_slice()
+            }
+        })
+        .collect();
+    let decoder = RowPackDecoder::new(ctx)?;
+    for _ in 0..3 {
+        black_box(decoder.decode_batch_to_arrow_large(&rows, &[0])?);
+    }
+    let start = Instant::now();
+    let mut bytes = 0;
+    for _ in 0..repeats {
+        let batch = decoder.decode_batch_to_arrow_large(black_box(&rows), &[0])?;
+        assert_eq!(batch.num_rows(), n);
+        bytes = batch.column(0).get_buffer_memory_size();
+        black_box(batch);
+    }
+    println!(
+        "{{\"case\":\"{}\",\"rows\":{},\"repeats\":{},\"seconds\":{},\"output_buffer_bytes\":{}}}",
+        kind,
+        n,
+        repeats,
+        start.elapsed().as_secs_f64(),
+        bytes
+    );
+    Ok(())
+}