code wiki / (root) / nx_shard_writer.nx

nx_shard_writer.nx

buildroot/runtime/nx_shard_writer.nx

5454 B172 linesdepth 5pulls 6 transitivereach 20 importersview sourcekind librarytopic shard
docsdependenciesstructsconstsfunctions

about

nx_shard_writer.nx -- size-rotating sharded output writer. When the current shard would exceed `rotate_bytes`, closes it and opens the next. Shards named <prefix>-shard-NNNN.jsonl (4-digit zero-padded for sortability). genealogy_id: log4j_size_rolling_policy lineage_id: output_rotation

dependencies 4 imports · 3 importers

nx_syscalls.nx nx_tier.nx nx_str.nx nx_strconv.nx nx_shard_writer.nx nx_ingest_runner.nx nx_ingest_runner_test.nx nx_shard_writer_test.nx

imports: nx_syscalls.nxnx_tier.nxnx_str.nxnx_strconv.nx

imported by: nx_ingest_runner.nxnx_ingest_runner_test.nxnx_shard_writer_test.nx

structs

29struct NxShardWriter

consts

21const NX_SHARDW_OK: nx_int = 0
22const NX_SHARDW_OPEN_FAIL: nx_int = -1
23const NX_SHARDW_WRITE_FAIL: nx_int = -2
24const NX_SHARDW_CLOSED: nx_int = -3
26const NX_SHARDW_PATH_BUF: nx_size = 512
27const NX_SHARDW_NEWLINE: nx_int = 10
41const NX_SHARDW_STRUCT_BYTES: nx_size = 80

functions

43func nx_shardw_format_4d(n: nx_int, buf: *u8) -> nx_int
called by 1: nx_shardw_path
54func nx_shardw_path(w: *NxShardWriter, idx: nx_int, out: *u8) -> nx_size
called by 1: nx_shardw_open_shard calls 3: nx_str_cpynx_str_lennx_shardw_format_4d
76func nx_shardw_open_shard(w: *NxShardWriter, idx: nx_int) -> nx_fd
called by 2: nx_shard_writer_opennx_shardw_rotate calls 2: sys_mmapnx_shardw_path
83func nx_shard_writer_open(prefix: *u8, rotate_bytes: nx_size) -> *NxShardWriter
called by 2: nx_ingest_run_newmain calls 4: sys_mmapnx_str_lennx_str_cpynx_shardw_open_shard
105func nx_shard_writer_set_raw(w: *NxShardWriter) -> nx_int
110func nx_shard_writer_close(w: *NxShardWriter) -> nx_int
called by 2: nx_ingest_run_closemain calls 1: sys_close
118func nx_shardw_rotate(w: *NxShardWriter) -> nx_int
called by 1: nx_shard_writer_emit calls 2: sys_closenx_shardw_open_shard
128func nx_shardw_write_all(fd: nx_fd, buf: *u8, len: nx_size) -> nx_size
called by 1: nx_shard_writer_emit calls 1: sys_write
139func nx_shard_writer_emit(w: *NxShardWriter, rec: *u8, len: nx_size) -> nx_int
called by 2: nx_ingest_run_offermain calls 3: nx_shardw_rotatenx_shardw_write_allsys_mmap
168func nx_shard_writer_shard_index(w: *NxShardWriter) -> nx_int { return w.shard_index }
called by 1: main
169func nx_shard_writer_total_records(w: *NxShardWriter) -> nx_int { return w.total_records }
called by 1: main
170func nx_shard_writer_total_bytes(w: *NxShardWriter) -> nx_size { return w.total_bytes }
called by 1: main
171func nx_shard_writer_current_bytes(w: *NxShardWriter) -> nx_size { return w.current_bytes }
called by 1: main
172func nx_shard_writer_n_shards(w: *NxShardWriter) -> nx_int { return w.shard_index + 1 }
called by 2: nx_ingest_run_n_shardsmain