Add index clocks from nanoseconds to days

kumo/connector/connector.py

@@ -107,6 +107,7 @@ def _compute_col_stats_from_sample(
     *,
     approx: bool,
     data_version: Any,
+    timestamp_precision: str = 'seconds',
 ) -> Dict[BaseColumnConfig, Dict[ColStatType, Stats]]:
     r"""Reduce an already-bounded sample without another warehouse read."""
     out: Dict[BaseColumnConfig, Dict[ColStatType, Stats]] = {}
@@ -121,7 +122,8 @@ def _compute_col_stats_from_sample(
         # A ``timestamp``-stype column read raw (cast_response=False) can
         # arrive as a bare epoch integer/float; coerce it to datetime so
         # the MIN/MAX col-stats are typed ``pd.Timestamp`` values.
-        ser = coerce_timestamp_stype_series_for_stats(ser, col_cfg)
+        ser = coerce_timestamp_stype_series_for_stats(ser, col_cfg,
+                                                      timestamp_precision)
         # A warehouse decimal arrives as an object column of `decimal.Decimal`,
         # which the gate below reads as `string` and strips every numerical
         # stat from -- leaving a `numerical` column without the stats

kumo/connector/timestamp_convert.py

@@ -13,10 +13,11 @@
 interpreted.
 
 What remains on this side is the standardized output model:
-  * a timestamp collapses to an ``i64`` epoch-seconds value (UTC);
+  * a timestamp collapses to an ``i64`` epoch value (UTC) in ticks of an index
+    clock, seconds unless a snapshot's clock is given (:func:`to_epoch_ticks`);
   * a date collapses to an ``i32`` days-since-epoch value.
-Sub-second precision and timezone are never retained, and everything is computed
-in the integer epoch-seconds / epoch-days domain; only
+Timezone and precision finer than the clock are never retained, and everything
+is computed in the integer epoch-ticks / epoch-days domain; only
 :func:`epoch_seconds_to_timestamp` lifts the final scalar back into a
 ``pd.Timestamp`` for the stats value.
 
@@ -44,6 +45,15 @@
 
 _NATIVE_UNIT_DIVISOR = {'s': 1, 'ms': 1_000, 'us': 1_000_000, 'ns': 1_000_000_000}
 
+PANDAS_UNIT_OF_INDEX_CLOCK = {
+    'nanoseconds': 'ns',
+    'microseconds': 'us',
+    'milliseconds': 'ms',
+    'seconds': 's',
+    'hours': 'h',
+    'days': 'D',
+}
+
 
 # ---------------------------------------------------------------------------
 # Core integer converters.
@@ -94,15 +104,17 @@ def days_from_secs(secs: int) -> int:
 # ---------------------------------------------------------------------------
 # The DiskGraph crossing: one call per column, no per-value Python work.
 # ---------------------------------------------------------------------------
-def _resolve(series: pd.Series, hint: TemporalHint, *,
-             days: bool) -> np.ndarray:
-    r"""Apply ``hint`` to a whole column, returning epoch seconds or epoch days.
+def _resolve(series: pd.Series, hint: TemporalHint, *, days: bool,
+             timestamp_precision: str = 'seconds') -> np.ndarray:
+    r"""Apply ``hint`` to a whole column, returning epoch days or epoch ticks
+    of the ``timestamp_precision`` clock.
 
     An ``UNRESOLVED`` column declares seconds rather than declaring nothing: a
     declared hint is what stops DiskGraph re-detecting the column at apply time.
     """
     unit = hint.unit or ('s' if hint.kind is HintKind.UNRESOLVED else None)
-    epochs, _, _ = resolve_temporal(_as_arrow(series), hint.strftime, unit, days)
+    epochs, _, _ = resolve_temporal(_as_arrow(series), hint.strftime, unit,
+                                    days, timestamp_unit=timestamp_precision)
     out = np.empty(len(epochs), dtype=object)
     out[:] = epochs
     return out
@@ -146,13 +158,15 @@ def _native_divisor(dtype) -> int:
     return _NATIVE_UNIT_DIVISOR.get(unit, 1_000_000_000)
 
 
-def timestamp_secs_from_array(series: pd.Series) -> np.ndarray:
-    r"""Convert a native temporal series to epoch seconds.
+def timestamp_ticks_from_array(
+        series: pd.Series, timestamp_precision: str = 'seconds') -> np.ndarray:
+    r"""Convert a native temporal series to ticks of ``timestamp_precision``.
 
-    Reads the series' own native unit and ceiling-divides each value to seconds
-    (second resolution passes through, ms/us/ns go through the ceiling), so even a
-    native timestamp is ceiling-converted. Null (``NaT``) values map to ``None``.
-    Returns a numpy object array aligned with the input positions.
+    Reads the series' own native unit and ceiling-divides each value into a
+    coarser clock (the same unit passes through), so even a native timestamp is
+    ceiling-converted; a finer clock multiplies, saturating at the int64 range.
+    Null (``NaT``) values map to ``None``. Returns a numpy object array aligned
+    with the input positions.
 
     Fully vectorized: the ceiling is ``-((-a) // divisor)`` on the int64 view
     (floor-division floors toward negative infinity, so negating the floor of the
@@ -162,7 +176,9 @@ def timestamp_secs_from_array(series: pd.Series) -> np.ndarray:
     becomes a bogus epoch.
     """
     s = _as_naive_utc_datetime_series(series)
-    divisor = _native_divisor(s.dtype)
+    native_ns = 1_000_000_000 // _native_divisor(s.dtype)
+    tick_ns = pd.Timedelta(
+        1, unit=PANDAS_UNIT_OF_INDEX_CLOCK[timestamp_precision]).value
     ints = s.to_numpy().view('int64')
     isna = pd.isna(s).to_numpy()
     # An object dtype array initializes to None, which is the NaT/null result.
@@ -171,11 +187,17 @@ def timestamp_secs_from_array(series: pd.Series) -> np.ndarray:
     if not computable.any():
         return out
     sub = ints[computable]  # excludes NaT (the minimum-int64 sentinel)
-    if divisor == 1:
-        secs = sub  # Second resolution: already whole epoch seconds
+    if tick_ns == native_ns:
+        ticks = sub
+    elif tick_ns > native_ns:
+        ticks = -((-sub) // (tick_ns // native_ns))
     else:
-        secs = -((-sub) // divisor)
-    out[computable] = secs.tolist()  # store Python ints, leave nulls as None
+        factor = native_ns // tick_ns
+        limit = np.iinfo(np.int64).max // factor
+        ticks = np.where(
+            sub > limit, np.iinfo(np.int64).max,
+            np.where(sub < -limit, np.iinfo(np.int64).min, sub * factor))
+    out[computable] = ticks.tolist()  # store Python ints, leave nulls as None
     return out
 
 
@@ -190,9 +212,18 @@ def to_epoch_seconds(series: pd.Series, hint: TemporalHint) -> np.ndarray:
     unconvertible or null value becomes ``None``, never raising. Returns a numpy
     object array of ints and ``None`` aligned with the input.
     """
+    return to_epoch_ticks(series, hint, 'seconds')
+
+
+def to_epoch_ticks(series: pd.Series, hint: TemporalHint,
+                   timestamp_precision: str) -> np.ndarray:
+    r"""Like :func:`to_epoch_seconds`, in ticks of the ``timestamp_precision``
+    index clock.
+    """
     if hint.kind is HintKind.NATIVE or is_datetime64_any_dtype(series):
-        return timestamp_secs_from_array(series)
-    return _resolve(series, hint, days=False)
+        return timestamp_ticks_from_array(series, timestamp_precision)
+    return _resolve(series, hint, days=False,
+                    timestamp_precision=timestamp_precision)
 
 
 def to_epoch_days(series: pd.Series, hint: TemporalHint) -> np.ndarray:
@@ -201,7 +232,7 @@ def to_epoch_days(series: pd.Series, hint: TemporalHint) -> np.ndarray:
     Best-effort: unconvertible or null values become ``None``.
     """
     if hint.kind is HintKind.NATIVE or is_datetime64_any_dtype(series):
-        secs = timestamp_secs_from_array(series)
+        secs = timestamp_ticks_from_array(series)
         return np.array(
             [None if x is None else days_from_secs(int(x)) for x in secs],
             dtype=object)

kumo/connector/utils.py

@@ -26,7 +26,12 @@
 
 from io_utils import fsutil, kumo_dataclass, logging
 from kumo.config import BaseColumnConfig
-from kumo.connector.timestamp_convert import to_epoch_days, to_epoch_seconds
+from kumo.connector.timestamp_convert import (
+    PANDAS_UNIT_OF_INDEX_CLOCK,
+    to_epoch_days,
+    to_epoch_seconds,
+    to_epoch_ticks,
+)
 from kumo.connector.timestamp_infer import (
     HintKind,
     TemporalHint,
@@ -372,6 +377,7 @@ def name_inference_temporal_supported(
 def coerce_timestamp_stype_series_for_stats(
     ser: pd.Series,
     col_config: BaseColumnConfig,
+    timestamp_precision: str = 'seconds',
 ) -> pd.Series:
     r"""Coerce a ``timestamp``-stype series to a datetime series before the Python
     column-stat kernel reduces it.
@@ -385,11 +391,11 @@ def coerce_timestamp_stype_series_for_stats(
 
     The coercion resolves the column's :class:`TemporalHint`
     (:func:`_resolve_stats_temporal_hint`) and applies
-    :func:`~kumo.connector.timestamp_convert.to_epoch_seconds`, then wraps the
-    epoch-seconds back into a second-precision ``datetime64`` series. Because every
-    temporal column collapses to epoch-seconds with ceiling division, the resulting
-    MIN/MAX (and the other timestamp stats) are computed in the same epoch-seconds
-    domain the on-disk index uses and match it rather than drifting.
+    :func:`~kumo.connector.timestamp_convert.to_epoch_ticks` in the
+    ``timestamp_precision`` index clock, then wraps the ticks back into a
+    ``datetime64`` series of that precision. The resulting MIN/MAX (and the
+    other timestamp stats) are computed in the same domain the on-disk index
+    uses and match it rather than drifting.
 
     Best-effort and narrowly scoped:
       * non-``timestamp``-stype columns pass through untouched;
@@ -406,8 +412,9 @@ def coerce_timestamp_stype_series_for_stats(
         return ser
 
     hint = _resolve_stats_temporal_hint(ser, col_config)
-    epoch_seconds = to_epoch_seconds(ser, hint)
-    return pd.to_datetime(pd.Series(epoch_seconds, index=ser.index), unit='s',
+    ticks = to_epoch_ticks(ser, hint, timestamp_precision)
+    return pd.to_datetime(pd.Series(ticks, index=ser.index),
+                          unit=PANDAS_UNIT_OF_INDEX_CLOCK[timestamp_precision],
                           errors='coerce')
 
 

kumo/data/generic/diskgraph_batch_pred.py

@@ -110,7 +110,7 @@ def generate_batch_pred_table_via_diskgraph(
         raise RuntimeError('diskgraph index path is required')
 
     with kumo_timing("batch_pred_table.create_plan_compiler"):
-        compiler = DiskGraphPlanCompiler()
+        compiler = DiskGraphPlanCompiler(index_path=index_path)
 
     # Determine execution mode
     if task_type == core_types.TaskType.FORECASTING:
@@ -756,7 +756,7 @@ def generate_batch_pred_table_via_diskgraph_ttng(
     if not index_path:
         raise RuntimeError('diskgraph index path is required')
 
-    compiler = DiskGraphPlanCompiler()
+    compiler = DiskGraphPlanCompiler(index_path=index_path)
 
     if task_type == core_types.TaskType.FORECASTING:
         return _execute_forecasting_ttng(
@@ -804,6 +804,7 @@ def _execute_node_prediction_ttng(
     # Node-pred: no lp_resolution; tt-ingest synthesizes ENTITY1 -> entity_node_id.
     plan_json = build_ttng_predict_plan(
         query,
+        index_path=index_path,
         task_type=task_type,
         graph=graph,
         timestamps=timestamps,
@@ -927,6 +928,7 @@ def _execute_link_prediction_ttng(
 
     lhs_plan = build_ttng_predict_plan(
         original_query,
+        index_path=index_path,
         task_type=task_type,
         graph=graph,
         timestamps=timestamps,
@@ -1141,6 +1143,7 @@ def _rename_dg_to_kumo(batch: pa.RecordBatch) -> pa.RecordBatch:
 def build_ttng_predict_plan(
     query: ValidatedPredictiveQuery,
     *,
+    index_path: str,
     task_type: core_types.TaskType,
     graph: Graph,
     timestamps: List[int],
@@ -1157,7 +1160,7 @@ def build_ttng_predict_plan(
     from diskgraph import DiskGraphPlanCompiler
     from kumo_core.training.table import apply_reference_resolution
 
-    compiler = DiskGraphPlanCompiler()
+    compiler = DiskGraphPlanCompiler(index_path=index_path)
     tt_str = task_type.value
     is_link_pred = bool(getattr(task_type, 'is_link_pred', False))
 

kumo/data/generic/diskgraph_train_table.py

@@ -757,7 +757,7 @@ def generate_train_table_via_diskgraph(
     if not fsutil.exists(index_path):
         raise RuntimeError(f'diskgraph index path does not exist: {index_path}')
 
-    compiler = DiskGraphPlanCompiler()
+    compiler = DiskGraphPlanCompiler(index_path=index_path)
     split_spec = _build_split_spec(split_config, split_family, SplitSpec)
     if split_spec is None:
         raise RuntimeError(

kumo/diskgraph/ingest.py

@@ -18,6 +18,7 @@
 from kumo_core.diskgraph.converter import convert_diskgraph_config
 from kumo_core.diskgraph.ingest import (
     normalize_legacy_column_aliases,
+    set_timestamp_precision,
     validate_diskgraph_ingest_config,
 )
 from kumo_core.diskgraph.runner import (
@@ -159,6 +160,7 @@ def run_diskgraph_ingest(
     source_specs: Optional[dict] = None,
     subprocess_environment: Optional[dict[str, Optional[str]]] = None,
     heartbeat: Optional[Callable[[], None]] = None,
+    timestamp_precision: str = 'seconds',
 ) -> DiskgraphIngestResult:
     """Convert the kumo graph config and launch the DiskGraph supervisor.
 
@@ -190,6 +192,7 @@ def run_diskgraph_ingest(
             source_specs=source_specs,
             subprocess_environment=subprocess_environment,
         )
+        set_timestamp_precision(ingest_config, timestamp_precision)
         validate_diskgraph_ingest_config(ingest_config)
 
     # DiskGraph owns phase detection, concurrency reduction after OOM, and

kumo/diskgraph/service.py

@@ -660,6 +660,7 @@ def ingest(
         data_persist_dir: str,
         table_data_versions: dict,
         dataplane_root: str,
+        timestamp_precision: str = 'seconds',
         source_paths: Optional[dict] = None,
         source_specs: Optional[dict] = None,
         subprocess_environment: Optional[dict[str, Optional[str]]] = None,
@@ -689,6 +690,7 @@ def ingest(
             graph_config=graph_config,
             data_persist_dir=data_persist_dir,
             table_data_versions=table_data_versions,
+            timestamp_precision=timestamp_precision,
             output_dir=output_dir,
             source_paths=source_paths,
             source_specs=source_specs,
@@ -1721,6 +1723,7 @@ def compute_col_stats_for_parquet(
         parquet_paths: List[str],
         kumo_column_specs: List[dict],
         data_version: str,
+        timestamp_precision: str = 'seconds',
     ) -> Tuple[str, str]:
         """Compute Kumo-format column statistics over BP-output parquet.
 
@@ -1733,6 +1736,7 @@ def compute_col_stats_for_parquet(
                 ``name`` / ``stype`` / ``dtype`` (raw Kumo enum value strings);
                 pybind maps them via ``map_kumo_stype`` / ``map_kumo_dtype``.
             data_version: Stamped into every ``kumo_col_stats_v1`` entry.
+            timestamp_precision: Index clock that timestamp statistics are in.
 
         Returns:
             ``(full_json, min_json)`` — the full map and its min variant.
@@ -1742,6 +1746,7 @@ def compute_col_stats_for_parquet(
             parquet_paths=parquet_paths,
             kumo_column_specs_json=json.dumps(kumo_column_specs),
             data_version=data_version,
+            timestamp_unit=timestamp_precision,
         )
 
 

packages/diskgraph/diskgraph-core/src/metadata.rs

@@ -130,6 +130,7 @@ pub enum EdgeSortKeyTimestampKind {
     Int8,
     Int16,
     Int32,
+    Int64,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

packages/diskgraph/diskgraph-core/src/sortkey.rs

@@ -37,6 +37,7 @@ impl SortKeyDecoder {
                 EdgeSortKeyTimestampKind::Int8 => 8,
                 EdgeSortKeyTimestampKind::Int16 => 16,
                 EdgeSortKeyTimestampKind::Int32 => 32,
+                EdgeSortKeyTimestampKind::Int64 => 64,
             })
             .unwrap_or(0);
 
@@ -231,6 +232,7 @@ impl SortKeyEncoder {
                 EdgeSortKeyTimestampKind::Int8 => 8,
                 EdgeSortKeyTimestampKind::Int16 => 16,
                 EdgeSortKeyTimestampKind::Int32 => 32,
+                EdgeSortKeyTimestampKind::Int64 => 64,
             })
             .unwrap_or(0);
 
@@ -853,4 +855,36 @@ mod tests {
             "encode_upper(30) incorrectly includes ts=30"
         );
     }
+
+    #[test]
+    fn int64_timestamp_keys_round_trip_in_order() {
+        let meta = EdgeSortKeyMetadata {
+            width_bits: 64,
+            timestamp: Some(EdgeSortKeyTimestampMetadata {
+                table: "t".into(),
+                column: "c".into(),
+                encoding: EdgeSortKeyTimestampEncoding {
+                    kind: EdgeSortKeyTimestampKind::Int64,
+                    anchor: 0,
+                    step: 1,
+                },
+            }),
+            weight: None,
+        };
+        let encoder = SortKeyEncoder::new(&meta);
+        let decoder = SortKeyDecoder::new(&meta);
+        let times = [
+            -1_000_000_001,
+            -1,
+            0,
+            1,
+            1_700_000_000_123_456_789,
+            i64::MAX,
+        ];
+        let keys: Vec<u64> = times.iter().map(|&t| encoder.encode_lower(t)).collect();
+        assert!(keys.windows(2).all(|pair| pair[0] < pair[1]));
+        for (time, key) in times.iter().zip(keys) {
+            assert_eq!(decoder.decode(key).0, Some(*time));
+        }
+    }
 }