Update KJT stride calculation logic to be based off of inverse_indices for VBE KJTs. (#3119)

Summary:

For VBE KJTs, ppdate the `_maybe_compute_stride_kjt` logic to calculate stride based off of `inverse_indices` when its set.

Currently, stride of VBE KJT with `stride_per_key_per_rank` is calculated as the max "stride per key". This is different from the batch size of the EBC output KeyedTensor which is based off of inverse_indices. This causes issues in IR module serialization.

Reviewed By: TroyGarden

Differential Revision: D76997485

Author: jd7-tr

torchrec/sparse/jagged_tensor.py

@@ -1129,13 +1129,18 @@ def _maybe_compute_stride_kjt(
     lengths: Optional[torch.Tensor],
     offsets: Optional[torch.Tensor],
     stride_per_key_per_rank: Optional[torch.IntTensor],
+    inverse_indices: Optional[Tuple[List[str], torch.Tensor]] = None,
 ) -> int:
     if stride is None:
         if len(keys) == 0:
             stride = 0
         elif (
             stride_per_key_per_rank is not None and stride_per_key_per_rank.numel() > 0
         ):
+            # For VBE KJT, batch size should be based on inverse_indices when set.
+            if inverse_indices is not None:
+                return inverse_indices[1].shape[-1]
+
             s = stride_per_key_per_rank.sum(dim=1).max().item()
             if not torch.jit.is_scripting() and is_non_strict_exporting():
                 stride = torch.sym_int(s)
@@ -1882,7 +1887,9 @@ def from_lengths_sync(
         lengths: torch.Tensor,
         weights: Optional[torch.Tensor] = None,
         stride: Optional[int] = None,
-        stride_per_key_per_rank: Optional[List[List[int]]] = None,
+        stride_per_key_per_rank: Optional[
+            Union[torch.IntTensor, List[List[int]]]
+        ] = None,
         inverse_indices: Optional[Tuple[List[str], torch.Tensor]] = None,
     ) -> "KeyedJaggedTensor":
         """
@@ -2180,6 +2187,7 @@ def stride(self) -> int:
             self._lengths,
             self._offsets,
             self._stride_per_key_per_rank,
+            self._inverse_indices,
         )
         self._stride = stride
         return stride

torchrec/sparse/tests/test_keyed_jagged_tensor.py

@@ -1017,6 +1017,18 @@ def test_meta_device_compatibility(self) -> None:
             lengths=torch.tensor([], device=torch.device("meta")),
         )
 
+    def test_vbe_kjt_stride(self) -> None:
+        inverse_indices = torch.tensor([[0, 1, 0], [0, 0, 0]])
+        kjt = KeyedJaggedTensor(
+            keys=["f1", "f2", "f3"],
+            values=torch.tensor([5, 6, 7, 1, 2, 3, 0, 1]),
+            lengths=torch.tensor([3, 3, 2]),
+            stride_per_key_per_rank=[[2], [1]],
+            inverse_indices=(["f1", "f2"], inverse_indices),
+        )
+
+        self.assertEqual(kjt.stride(), inverse_indices.shape[-1])
+
 
 class TestKeyedJaggedTensorScripting(unittest.TestCase):
     def test_scriptable_forward(self) -> None: