Compress operation-ready plaintexts without changing CKKS semantics

Example source: examples/16_compressed_plaintext.py

This example converts a periodic operation-ready plaintext to exact CompressedPlaintext, verifies dense-equivalent addition and multiplication, and measures storage and evaluator cost. The evaluator reads the compact operand directly rather than expanding a dense plaintext first.

Use this representation when a repeatedly used plaintext has exact repetition or exact sparse structure after CKKS encoding and arithmetic preparation. Keep using Plaintext when the encoded tensor is not exactly representable by a supported layout or when compact storage does not improve the measured workload.

Run the complete example

Start with the default period on the 8,192-slot, 40-bit-scale baseline:

python examples/16_compressed_plaintext.py \
  --device cpu \
  --preset slots8192-scale40-levels7-int64 \
  --period 256 \
  --iterations 20

Use --device cuda:0 to run the same example through CUDA. The command reports:

• the slot period and exact encoded unique count;
• dense and compact tensor bytes;
• bit-exact ciphertext equality against dense addition and multiplication;
• maximum cleartext error after compressed multiplication;
• synchronized dense and compressed evaluator medians.

Try several powers of two that divide the slot count:

python examples/16_compressed_plaintext.py --preset slots8192-scale40-levels7-int64 --period 64
python examples/16_compressed_plaintext.py --preset slots8192-scale40-levels7-int64 --period 512

A smaller period usually stores fewer unique encoded values, but storage reduction alone does not predict evaluator latency. Measure the operation, level, batch shape, device, and period used by the deployed workload.

1. Identify the representation requirement

Plaintext remains the general CKKS value. CompressedPlaintext is a separate exact value for an operation-ready RNS plaintext whose encoded last axis can be reconstructed without loss.

The two physical layouts are:

dense Plaintext:      [*batch, limb, coefficient_or_ntt_index]
CompressedPlaintext:  [*batch, limb, unique_encoded_value]
strided implicit data:[*batch, limb]

The compression layout describes the encoded coefficient or NTT axis. It does not describe semantic CKKS slot order. CKKS embedding permutes slots, and integer coefficient rounding can destroy repetition that is visible in the source message.

For example, a semantic slot vector with power-of-two period r has a specific property under the current codec:

• its prepared coefficient representation is exactly strided sparse;
• its prepared NTT representation has 2 * r exact values in contiguous repeated blocks.

The checked constructor verifies those claims against the actual dense tensor. Do not infer compressibility from source-message appearance alone.

2. Understand the three encoded-axis layouts

Let the ring dimension be N, the compact width be U, and repeat_count = N // U. For N = 8, U = 2, and compact data [a, b], the exact expansions are:

cyclic:          [a, b, a, b, a, b, a, b]
contiguous:      [a, a, a, a, b, b, b, b]
strided_sparse:  [a, z, z, z, b, z, z, z]

For strided_sparse, z is one exact implicit_data value per batch member and RNS limb. It is not assumed to be zero. The compact entries occupy indices u * repeat_count; every other dense position uses that row's stored implicit value.

The supported arithmetic is:

Layout	Coefficient-domain addition	NTT-domain multiplication
cyclic	Yes	Yes
contiguous	Yes	Yes
strided_sparse	Yes	No

strided_sparse is coefficient-domain only. Multiplication requires a cyclic or contiguous NTT-domain value. An application that needs both addition and multiplication prepares and retains two separate exact compressed values.

For every layout:

• N and U must be powers of two;
• 0 < U < N;
• U must divide N;
• data is integral and uses Montgomery residues;
• the value records its format version, ring dimension, context, level, actual scale, domain, basis, residue form, and exact prime_ids.

3. Build the dense operation-ready values first

The maintained example creates one periodic complex factor:

period = 256
unique_index = torch.arange(period, dtype=torch.float64)
unique_slots = torch.complex(
    0.03 * torch.cos(unique_index * 0.07)
    + 0.001 * unique_index / period,
    0.02 * torch.sin(unique_index * 0.05),
)
factor = unique_slots.repeat(engine.num_slots // period)

Encoding alone does not select the arithmetic state. Prepare independently for multiplication and addition:

dense_multiply = engine.prepare_plaintext_for_multiplication(
    engine.encode(factor)
)
dense_add = engine.prepare_plaintext_for_addition(
    engine.encode(factor)
)

The multiplication value is NTT-domain Montgomery RNS. The addition value is coefficient-domain Montgomery RNS. Both retain the level, actual scale, context, basis, and active prime rows chosen by the engine.

4. Convert with bit-exact validation

For the periodic factor above, create the two compressed values as follows:

compressed_multiply = fh.CompressedPlaintext.from_plaintext(
    dense_multiply,
    unique_count=2 * period,
    compression_layout="contiguous",
)

compressed_add = fh.CompressedPlaintext.from_plaintext(
    dense_add,
    unique_count=2 * period,
    compression_layout="strided_sparse",
)

from_plaintext checks the complete encoded last axis bit for bit. It raises ValueError instead of approximating unequal values, changing encoding semantics, or silently choosing another layout. On success it clones the compact slice and, for strided_sparse, the implicit values. The compressed value therefore does not retain the dense input's backing storage.

The conversion lifecycle has these operations:

Expanded diagram

100%

Use decompression when a consumer requires an uncompressed dense value:

restored_dense = compressed_multiply.to_plaintext()
assert torch.equal(restored_dense.data, dense_multiply.data)

to_plaintext() allocates the full N-element encoded axis. Evaluator kernels do not call it.

5. Use the compressed operands directly

Multiplication accepts a compatible cyclic or contiguous NTT compressed plaintext:

ciphertext = engine.encrypt_message(message)
ciphertext_ntt = engine.coefficient_domain_to_ntt_domain(ciphertext)
compressed_result = engine.multiply_plaintext(
    ciphertext_ntt,
    compressed_multiply,
)

The operation preserves the ciphertext level and records the scale product:

$${\mathrm{\Delta }}_{\mathrm{out}}={\mathrm{\Delta }}_{\mathrm{ciphertext}}{\mathrm{\Delta }}_{\mathrm{plaintext}}.$$

It does not rescale implicitly. Apply the same rescale schedule you would use with a dense prepared plaintext.

Addition accepts a compatible coefficient-domain compressed plaintext:

compressed_sum = engine.add_plaintext(ciphertext, compressed_add)

Addition requires exact scale equality and preserves that scale. It modifies only the c0 component mathematically. The in-place form makes the storage mutation visible:

work = ciphertext.clone()
engine.add_plaintext_(work, compressed_add)

The homogeneous-batch shape requirement is unchanged. A genuinely unbatched compressed plaintext broadcasts over a ciphertext batch. A compressed plaintext with a nonempty batch prefix must match ciphertext.batch_shape exactly.

6. Verify equivalence against dense arithmetic

Compression is an exact storage and execution representation, not a numerical approximation. Compare the resulting ciphertext tensors against the same operation with the dense prepared plaintext:

dense_product = engine.multiply_plaintext(ciphertext_ntt, dense_multiply)
compact_product = engine.multiply_plaintext(ciphertext_ntt, compressed_multiply)
assert torch.equal(compact_product.data, dense_product.data)


dense_sum = engine.add_plaintext(ciphertext, dense_add)
compact_sum = engine.add_plaintext(ciphertext, compressed_add)
assert torch.equal(compact_sum.data, dense_sum.data)

Then decrypt a representative result and compare it with the cleartext operation:

decoded = engine.decrypt_message(
    engine.ntt_domain_to_coefficient_domain(compact_product)
)
expected = message * factor
max_error = torch.max(torch.abs(decoded.cpu() - expected)).item()

Tensor equality verifies the compressed kernel against dense CKKS arithmetic. The cleartext comparison separately checks the expected CKKS approximation error.

7. Measure storage and evaluator cost separately

For cyclic and contiguous layouts, compact tensor storage scales with U instead of N:

dense_bytes = dense_multiply.data.numel() * dense_multiply.data.element_size()
compact_bytes = compressed_multiply.nbytes
storage_reduction = dense_bytes / compact_bytes

A strided_sparse value also stores one implicit value per batch member and limb, so its payload is proportional to U + 1 rather than only U. Serialization metadata adds a small fixed overhead in either case.

The compact evaluator kernels read right-hand-side values directly, but they still produce every ciphertext coefficient or NTT position. Compression can reduce plaintext storage and right-hand-side memory traffic; it does not reduce the ciphertext size or guarantee a speedup. Benchmark with synchronization and report both storage and latency, as the maintained example does.

Keep lifecycle policy separate from representation. If both arithmetic states are reused, retain compressed_add and compressed_multiply as two exact values. FHElium does not hide one state behind an engine-owned conversion cache.

8. Serialize and move the exact compressed value

CompressedPlaintext participates in the core exact-value interfaces. For example:

compressed_cpu = compressed_multiply.to("cpu")
fh.save_value(
    compressed_cpu,
    "factor.safetensors",
    overwrite=True,
)

restored = fh.load_value(
    "factor.safetensors",
    expected_type=fh.CompressedPlaintext,
    device=engine.device,
)

The file preserves the compression-format version, compact and implicit tensor metadata, cryptographic state, and exact encoded layout. Typed distributed transport, residency helpers, execution signatures, and CUDA Graph validation likewise treat the compressed value as an exact value rather than as a recipe to re-encode semantic slots.

9. Recognize rejected layouts

Expect conversion or evaluation to fail in these cases:

• the source is a slots or approximate-coefficient Plaintext, not an operation-ready RNS plaintext;
• unique_count is nonpositive, not a power of two, equal to or larger than N, or does not divide N;
• the dense encoded axis is not bit-exactly representable by the requested layout;
• strided_sparse is requested for an NTT value or used for multiplication;
• the compressed value and ciphertext differ in context, level, basis, prime_ids, ring dimension, dtype, device, or required domain;
• a batched compressed plaintext has a different nonempty batch shape;
• addition scales are not exactly equal;
• an incompatible compression-format version is loaded.

A source vector can be semantically short, constant over blocks, or generated from a low-dimensional formula and still fail exact encoded-axis validation. That failure preserves the exact representation rule: use the dense Plaintext, or change the application's packing and validate the resulting operation-ready value again. Do not weaken the equality check or choose a larger unique_count unless the new representation is still smaller than N and passes exact validation.

Complete runnable source

"""Compress a periodic operation-ready plaintext without changing Plaintext.

The example starts from a standard periodic CKKS slot message, encodes it with
the ordinary codec, verifies the exact repeated NTT structure, and converts it
to a separate CompressedPlaintext. The evaluator kernel reads the compact
operand directly; it does not materialize a dense plaintext during multiply.
"""

from __future__ import annotations

import argparse
import time

import torch
from common import add_engine_args, make_engine

import fhelium as fh


def _synchronize(device: torch.device) -> None:
    if device.type == "cuda":
        torch.cuda.synchronize(device)


def _median_ms(
    operation,
    *,
    iterations: int,
    device: torch.device,
) -> float:
    for _ in range(3):
        operation()
    _synchronize(device)
    samples = []
    for _ in range(iterations):
        start = time.perf_counter()
        operation()
        _synchronize(device)
        samples.append((time.perf_counter() - start) * 1e3)
    return float(torch.tensor(samples).median())


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    add_engine_args(
        parser,
        default_preset=fh.Preset.slots8192_scale40_levels7_int64.value,
    )
    parser.add_argument("--period", type=int, default=256)
    parser.add_argument("--iterations", type=int, default=20)
    args = parser.parse_args()

    engine = make_engine(args)
    period = args.period
    if period <= 0 or period & (period - 1):
        raise ValueError("--period must be a positive power of two")
    if engine.num_slots % period:
        raise ValueError("--period must divide the CKKS slot count")

    # Repetition modes describe the encoded tensor's last axis. Their compact
    # payload is identical; only index expansion differs.
    compact_example = torch.tensor([1, 2])
    print("encoded cyclic:    ", compact_example.repeat(4).tolist())
    print(
        "encoded contiguous:",
        compact_example.repeat_interleave(4).tolist(),
    )

    unique_index = torch.arange(period, dtype=torch.float64)
    unique_slots = torch.complex(
        0.03 * torch.cos(unique_index * 0.07) + 0.001 * unique_index / period,
        0.02 * torch.sin(unique_index * 0.05),
    )
    factor = unique_slots.repeat(engine.num_slots // period)

    # A period-r semantic slot vector yields 2r exact encoded NTT values,
    # stored as contiguous repeated blocks for the current CKKS codec.
    dense = engine.prepare_plaintext_for_multiplication(engine.encode(factor))
    compressed = fh.CompressedPlaintext.from_plaintext(
        dense,
        unique_count=2 * period,
        compression_layout="contiguous",
    )
    dense_addend = engine.prepare_plaintext_for_addition(engine.encode(factor))
    sparse_addend = fh.CompressedPlaintext.from_plaintext(
        dense_addend,
        unique_count=2 * period,
        compression_layout="strided_sparse",
    )
    dense_data = dense.data
    compressed_data = compressed.data
    if dense_data is None or compressed_data is None:
        raise RuntimeError("prepared plaintext payloads must be materialized")
    dense_bytes = dense_data.numel() * dense_data.element_size()
    compressed_bytes = compressed_data.numel() * compressed_data.element_size()
    print(f"ring dimension:       {engine.config.N}")
    print(f"periodic slots:       {period}")
    print(f"encoded unique count: {compressed.unique_count}")
    print(f"dense bytes:          {dense_bytes:,}")
    print(f"compressed bytes:     {compressed_bytes:,}")
    print(f"storage reduction:    {dense_bytes / compressed_bytes:.1f}x")
    print(f"sparse-add bytes:     {sparse_addend.nbytes:,}")

    slot_index = torch.arange(engine.num_slots, dtype=torch.float64)
    message = torch.complex(
        0.01 * torch.sin(slot_index * 0.013),
        0.008 * torch.cos(slot_index * 0.011),
    )
    ciphertext = engine.encrypt_message(message)
    ciphertext_ntt = engine.coefficient_domain_to_ntt_domain(ciphertext)
    dense_result = engine.multiply_plaintext(ciphertext_ntt, dense)
    compressed_result = engine.multiply_plaintext(ciphertext_ntt, compressed)
    if not torch.equal(compressed_result.data, dense_result.data):
        raise AssertionError("Compressed and dense ciphertexts differ")
    decoded = engine.decrypt_message(
        engine.ntt_domain_to_coefficient_domain(compressed_result)
    ).cpu()
    max_error = torch.max(torch.abs(decoded - message * factor)).item()
    print(f"maximum cleartext error: {max_error:.3e}")

    dense_sum = engine.add_plaintext(ciphertext, dense_addend)
    sparse_sum = engine.add_plaintext(ciphertext, sparse_addend)
    if not torch.equal(sparse_sum.data, dense_sum.data):
        raise AssertionError("Sparse and dense addition ciphertexts differ")

    dense_ms = _median_ms(
        lambda: engine.multiply_plaintext(ciphertext_ntt, dense),
        iterations=args.iterations,
        device=engine.device,
    )
    compressed_ms = _median_ms(
        lambda: engine.multiply_plaintext(ciphertext_ntt, compressed),
        iterations=args.iterations,
        device=engine.device,
    )
    print(f"dense evaluator median:      {dense_ms:.3f} ms")
    print(f"compressed evaluator median: {compressed_ms:.3f} ms")
    print(f"evaluator speedup:           {dense_ms / compressed_ms:.2f}x")

    dense_add_work = ciphertext.clone()
    sparse_add_work = ciphertext.clone()
    dense_add_ms = _median_ms(
        lambda: engine.add_plaintext_(dense_add_work, dense_addend),
        iterations=args.iterations,
        device=engine.device,
    )
    sparse_add_ms = _median_ms(
        lambda: engine.add_plaintext_(sparse_add_work, sparse_addend),
        iterations=args.iterations,
        device=engine.device,
    )
    print(f"dense in-place addition median:  {dense_add_ms:.3f} ms")
    print(f"sparse in-place addition median: {sparse_add_ms:.3f} ms")
    print(
        f"in-place addition speedup:       {dense_add_ms / sparse_add_ms:.2f}x"
    )


if __name__ == "__main__":
    main()

Related API and implementation detail

• Values and state API
• Serialization API
• Value model and identity
• CompressedPlaintext internals
• CKKS workload cost model