code: def test_sparse_lengths_sum_rowwise_sparse(
      self,
      num_rows,
      blocksize,
      weighted,
      empty_indices,
      bit_rate,
      indices_64bit,
  ):
      net = core.Net("bench")
  
      input_data = np.random.rand(num_rows, blocksize).astype(
          np.float32
      )
      if empty_indices:
          lengths = np.zeros(num_rows, dtype=np.int32)
          num_indices = 0
      else:
          num_indices = np.random.randint(len(input_data))
          # the number of indices per sample
          lengths_split = np.clip(num_indices // 2, 1, 10)
          lengths = (
              np.ones([num_indices // lengths_split], dtype=np.int32)
              * lengths_split
          )
          # readjust num_indices when lengths_split doesn't divide num_indices
          num_indices = num_indices // lengths_split * lengths_split
      #  Use int32 here because int64 is covered by test_sparse_lengths_sum
      index_type = np.int64 if indices_64bit else np.int32
      indices = np.random.randint(
          low=0,
          high=len(input_data),
          size=[num_indices],
          dtype=index_type,
      )
      weights = np.random.uniform(size=[len(indices)]).astype(
          np.float32
      )
  
      op = core.CreateOperator(
          "FloatToFused" + str(bit_rate) + "BitRowwiseQuantized",
          "input_data",
          "quantized_data",
      )
      workspace.FeedBlob("input_data", input_data)
      workspace.RunOperatorOnce(op)
      quantized_data = workspace.FetchBlob("quantized_data")
  
      # Prune and generate mapping table
      sparsity = 0.7
      mapping_table = np.zeros(num_rows, dtype=np.int32)
      num_compressed_rows = 0
      unpruned_ids = []
      for i in range(num_rows):
          if np.random.uniform() < sparsity:
              mapping_table[i] = -1
              quantized_data[i, :] = 0
          else:
              mapping_table[i] = num_compressed_rows
              num_compressed_rows += 1
              unpruned_ids.append(i)
  
      pruned_quantized_data = quantized_data[unpruned_ids]
  
      inputs = (
          ["quantized_data"]
          + (["weights"] if weighted else [])
          + ["indices", "lengths"]
      )
      op = core.CreateOperator(
          "SparseLengths"
          + ("Weighted" if weighted else "")
          + "SumFused"
          + str(bit_rate)
          + "BitRowwise",
          inputs,
          "sum_reference",
      )
      net.Proto().op.extend([op])
  
      inputs[0] = "pruned_quantized_data"
      op = core.CreateOperator(
          "SparseLengths"
          + ("Weighted" if weighted else "")
          + "Sum"
          + str(bit_rate)
          + "BitRowwiseSparse",
          inputs + ["mapping_table"],
          "sum_pruned",
      )
      net.Proto().op.extend([op])
  
      op = core.CreateOperator(
          "SparseLengthsSumSparseLookup",
          ["indices", "lengths", "mapping_table"]
          + (["weights"] if weighted else []),
          ["new_indices", "new_lengths"]
          + (["new_weights"] if weighted else []),
      )
      net.Proto().op.extend([op])
      inputs = (
          ["pruned_quantized_data"]
          + (["new_weights"] if weighted else [])
          + ["new_indices", "new_lengths"]
      )
      op = core.CreateOperator(
          "SparseLengths"
          + ("Weighted" if weighted else "")
          + "SumFused"
          + str(bit_rate)
          + "BitRowwise",
          inputs,
          "sum_split",
      )
      net.Proto().op.extend([op])
  
      workspace.FeedBlob("quantized_data", quantized_data)
      workspace.FeedBlob("pruned_quantized_data", pruned_quantized_data)
      workspace.FeedBlob("weights", weights)
      workspace.FeedBlob("indices", indices)
      workspace.FeedBlob("lengths", lengths)
      workspace.FeedBlob("mapping_table", mapping_table)
  
      workspace.RunNetOnce(net)
  
      sum_reference = workspace.FetchBlob("sum_reference")
      sum_pruned = workspace.FetchBlob("sum_pruned")
      sum_split = workspace.FetchBlob("sum_split")
  
      np.testing.assert_array_equal(sum_reference, sum_pruned)
      np.testing.assert_array_equal(sum_reference, sum_split)
  
file_path: pytorch/caffe2/python/operator_test/lengths_reducer_fused_nbit_rowwise_ops_test.py
imports: import caffe2.python.hypothesis_test_util
  import hypothesis.strategies
  import numpy
  from caffe2.python import core, workspace
  from hypothesis import given
label_desc: Use image node_id 3 for calling the TestLengthsReducerOpsFusedNBitRowwise obj's underlying member method code with example usage: obj.test_sparse_lengths_sum_rowwise_sparse(num_rows, blocksize, weighted, empty_indices, bit_rate, indices_64bit) without return types