#179 Optimize pickle storage based on object size Pull request opened by pingsutw ## Summary • Add size-based optimization for pickle serialization • Small objects (≤10KB) are now stored inline as binary data • Large objects continue to be stored as blob files • Reduces storage overhead and improves performance for small objects ## Changes • Add

DEFAULT_PICKLE_BYTES_LIMIT

constant (10KB) • Modify

to_literal

to use

sys.getsizeof

for size-based storage decision • Update

to_python_value

to handle both binary and blob storage formats • Add comprehensive unit tests covering all scenarios including edge cases ## Test Plan • All existing tests pass • New comprehensive test suite with 11 test cases covering: • Size threshold behavior (small vs large objects) • Boundary conditions (exactly at limit) • Binary and blob storage/retrieval • Error handling for invalid literals • Complete roundtrip testing • Mocked file operations to avoid filesystem dependencies import typing import flyte # TaskEnvironments provide a simple way of grouping configuration used by tasks (more later). env = flyte.TaskEnvironment(name="hello_world", resources=flyte.Resources(memory="250Mi")) # use TaskEnvironments to define tasks, which are regular Python functions. @env.task def fn(x: int) -> typing.Any: # type annotations are recommended. slope, intercept = 2, 5 return slope * x + intercept # tasks can also call other tasks, which will be manifested in different containers. @env.task def main(x_list: list[int]) -> typing.Any: x_len = len(x_list) if x_len < 10: raise ValueError(f"x_list doesn't have a larger enough sample size, found: {x_len}") y_list = list(flyte.map(fn, x_list)) # flyte.map is like Python map, but runs in parallel. print("y_list:", y_list) y_mean = sum(y_list) / len(y_list) return y_mean if name == "__main__": flyte.init_from_config() # establish remote connection from within your script. run = flyte.run(main, x_list=list(range(10))) # run remotely inline and pass data. # print various attributes of the run. print(run.name) print(run.url) run.wait(run) # stream the logs from the root action to the terminal. flyteorg/flyte-sdk