Ingestion¶

Records enter an HD instance through the Message Bus Ingester: one engine role that reads a source, maps each record onto the dataset schema, groups records into batches and streams them to a Storage Cluster Controller over the FermiHDI HD Wire Protocol (FHDWP). Everything downstream — the fan-out to every Storage Node, the hash chain, the filter node — starts from what the ingester sends. This page is what an operator and a data owner need to know about it: where it reads from, what it accepts, and how it decides where one batch ends and the next begins.

Sources¶

One ingester reads one source, chosen by ingression_type:

Source	Setting	Reads
Kafka	`kafka.brokers`, `kafka.topic`, `kafka.group_id`	a topic, as a consumer group member
RabbitMQ	`rabbitmq.url`, `rabbitmq.queue`	a queue
MQTT	`mqtt.address`, `mqtt.topic`	a topic
Local directory	`file.directory`, `file.sort_by`	every file in the directory, one at a time, deleting each once its records are acknowledged
S3 or a compatible object store	`s3.bucket`, `s3.prefix`, `s3.region`, `s3.endpoint`	every object under the prefix, one at a time

Kafka is what the installer wires. The others are configured through the ingester's own settings, which reach it as environment variables, a JSON file or command-line flags in that order of precedence; the full table is in the configuration reference of the ingester.

A message from a broker is one record, in JSON or in the RowBinary form (input_format), or a CSV row. A file or object is many records, and its extension says how they are encoded.

File and object formats¶

Every format decodes to one JSON record per row and then takes the same path a broker message takes, so the schema mapping, address parsing, batch rotation and time compression exist once. The extension is the format:

Extension	Format	Notes
`.json`	one JSON object, or an array of them
`.jsonl`, `.ndjson`	one JSON object per line
`.csv`	a header row, then one record per row	cells that read as numbers become numbers
`.parquet`, `.pq`	Apache Parquet	a list column is an array (what an array field takes), a dictionary-encoded column is its value; a timestamp or date column is converted to the unit the schema declares for the field (`time.unit`, or the batch field's `time_index_interval_unit`), so a PostgreSQL export in microseconds lands in a dataset keyed on seconds as seconds
`.orc`	Apache ORC
`.avro`	Avro object container file	the writer's schema travels in the file; a union is its branch, so a nullable `long` is `5` or `null`
`.pb`, `.protobuf`	Protocol Buffers	needs a descriptor set (`formats.protobuf_descriptor_file`, written by `protoc --descriptor_set_out --include_imports`) and the message type (`formats.protobuf_message_type`). Messages are length-delimited unless `formats.protobuf_delimited` is false. 64-bit integers stay numbers
`.xml`	XML	each child of the root is a record, or every element named `formats.xml_record_element` wherever it sits; attributes become `@name`, repeated children become arrays, text that reads as a number or boolean becomes one

Files are taken in sort_by order — by name, or by modification time — and a file with any other extension is left alone. A file is removed only after the storage cluster has acknowledged its records.

Mapping records onto the schema¶

The dataset schema names every field, its type and its byte position in the fixed-width record. A JSON key that matches a field name is written there; keys the schema does not name are ignored, and fields no key supplies are zero. Four things deserve a word:

Addresses. A field the schema types ipv4 is a 32-bit unsigned integer and one typed ipv6 a 128-bit one, both big-endian on the wire. The ingester accepts either the integer or the address written as text — 10.0.0.1, 2001:db8::1 — and stores the integer. An IPv4 address given for an ipv6 field is stored IPv4-mapped (::ffff:10.0.0.1). Every query service renders the integer back as an address; the SDK does neither and sees only the integer.
Time. A timestamp field is the integer it is, in the unit the dataset declares (time_index_interval_unit: seconds by default, or milliseconds, nanoseconds, minutes). It is not parsed from a date string.
Text. A string field is stored UTF-32, four bytes per character, whatever string_encoding names as the source, so that a byte-offset filter is always character aligned. The exception is a field whose string_encoding is ascii: it is stored one byte per character, since every ASCII character already is one, and holds four times as many characters in the same bytes. A character outside ASCII arriving for such a field is stored as ?, and the ingester says so.
Enumerations. A dataset may declare an enums table: named lists of labels, each assigned an unsigned integer code, with one label as the default. A uint8, uint16, uint32 or uint64 field that names one in its enum property stores the code, four bytes at most for a uint32 rather than the sixty-four a sixteen-character string would occupy. The ingester accepts the label, matched exactly and case-sensitively, or the decimal text of a defined code. Anything else stores the default's code and is counted, because a producer that spells a label differently from the schema maps every record to the default, and that should be visible, not silent. Every query service renders the code back as its label.
Time fields and intra-record time compression. An integer field may declare time with a unit (nanoseconds through days): its value is a count of that unit since the Unix epoch, and the unit is what lets another field be stored as a delta from it. A field carrying time_reference names such a source field and a delta unit, and stores not a time of its own but the signed difference from the source's value in the same record: an event a few seconds after the record's own timestamp fits an int16 of seconds or an int32 of milliseconds rather than eight bytes. The value arrives absolute, in the field's own time unit or, when it declares none, the source's; the ingester computes and stores the delta. A delta the field cannot hold is clamped to its nearest bound and counted, and a record that carries no value for the source stores zero and is counted, since either is a schema-sizing or producer problem someone should hear about. On an array of integers both declarations apply to each element, measured from the same source. Every query service hands the field back expanded, as the absolute time; a filter on it, however, compares with the stored delta.
Chess notation. A chess_move field is a 16-bit code that stands for one Standard Algebraic Notation move (Nf3, exd5, e8=Q, O-O, with a + or #): the value arrives as the SAN text and the code is computed from it, with no table to ship, so every service reads the same move. Text the code cannot carry -- a move disambiguated by both file and rank, or not a move at all -- stores zero, "no move", and is counted. An array of chess_move may be given a whole PGN movetext (1. e4 e5 2. Nf3 ...) instead of a list: move numbers, comments, variations and the result are dropped and the moves are stored in order. A chess_fen field is a Forsyth-Edwards position packed into 32 bytes: an 8-byte occupancy bitmask (a1 is bit 0), 20 bytes of 5-bit piece tokens in occupancy order, and 4 bytes of game state (side to move, castling rights, en passant file, the halfmove clock in 8 bits, up to 255, and the fullmove number in 15 bits, up to 32,767; a larger value is stored as the largest). The value arrives as the FEN text; a string that is not a position stores zero, "no position", and is counted. A chess_move_cc field is chess.com's own real per-ply move code: two raw bytes, exactly as chess.com's own Parquet export stores them. Unlike chess_move there is no formula to compute the code from -- the value arrives as that same 2-character code and is stored verbatim, unchanged, so it never fails to represent a move the way chess_move can. An array of chess_move_cc may be given the source string directly instead of a list: it is split into 2-character chunks in order, not tokenized as PGN movetext the way chess_move is, since chess.com's own export has no PGN structure to drop. Whether a chess_move or chess_fen field is handed back on the query path as its notation or as its raw stored form is the field's own render property in the schema ("raw" or "decoded", default "decoded") -- not a setting the ingester has any part in. chess_move_cc is different: every query service always hands it back raw, its own 2-character code, whatever render says (or whether the field declares one at all) -- decoding it to long algebraic notation is an explicit capability a consumer calls itself, not something rendered on the query path automatically. See querying.md.
Arrays. A bytes field may declare an array: an element type, the most elements the field holds, and for string or numeric elements the width of each. The value arrives as a JSON array ("tags": ["vip", "eu"]), and the ingester writes the count followed by the elements, each exactly as a field of that type would be written, with the rest of the field zero. A scalar is taken as a one-element list. Elements past the field's capacity are dropped from the end and counted, because a producer whose lists outgrow the field is a schema-sizing problem someone should hear about, and a dropped record would be worse than a shortened list. An enum on the field applies to each element.

How batches are formed¶

A batch is the unit the storage cluster stores, hashes, indexes and serves. It carries a 64-byte id, and the ingester decides the id of every record it sends. When the id changes between one record and the next the open batch is closed with an end-of-batch marker and the next one opened — and a closed batch is never reopened. There are three ways the id is chosen, and the dataset schema picks among them.

A time field, rounded to the time index. The schema's batch_id_field names a time field and sets round_batch_id_field; every record's time is rounded down to time_index_interval from the Unix epoch. With an interval of 60 seconds every batch is the top of a minute, whichever second of the minute the first record arrived in, and the batch rotates each time the minute changes. This is how a time-series dataset is stored, and it is what lets a query name its batches from a time window alone. The time field and the interval are in the same unit. The schema's flag is enough on its own; the ingester's own round_batch_id_field setting only forces rounding on.

Any field's value, as it is. Without round_batch_id_field the field's value is the batch id verbatim: a string field is the id text itself, up to 64 bytes; an integer field is its decimal text; a uint64 field is written as eight big-endian bytes. Tenant, customer, region and device ids all work this way. Two things follow. Records have to arrive grouped by that key, or the ingester closes and reopens the same batch over and over. And the query side has to be able to name the ids — a time range enumerates itself, a key set does not, so a client passes each key it wants through the SDK's string or number batch-id generator.

The ingester's clock. With no batch_id_field, or for a record that lacks the field, the id is the ingest time rounded to the interval, as a uint64. The ingester's batch_interval_seconds overrides the interval. This is what a Kafka ingester does by default.

Record size and the data-plane frame¶

A record travels whole inside one FHDWP frame. The standard 1200-byte frame carries 1133 bytes of records, and a jumbo frame (FERMIHDI_JUMBO=1 on the SCC, every Storage Node and the ingester) carries 8433. A record wider than the payload travels alone in a frame of its own, so a wide dataset works in both modes; jumbo frames only pack more records into each frame. The widest record a dataset may declare is 8433 bytes, and the ingester refuses to start with a wider one.

Closing the last batch¶

A batch closes when the first record of a later batch arrives. When records stop, the ingester closes the open batch after a timeout instead (batch_close_timeout_mode):

Mode	Closes the open batch after
`dynamic` (default)	the longest time any one file or object has spanned so far, measured on its batch field, and never less than one batch interval; `batch_close_timeout_sec` until a file has been read
`manual`	`batch_close_timeout_sec` (default 300) with no record
`off`	never; only a later batch closes it

A closed batch is never reopened. Records that arrive for one, from a file delivered out of time order or the second half of a batch split across files that arrived further apart than the timeout, are refused by the storage cluster, and the ingester logs them as an error.

Files that span several batches¶

A file's rows are rarely in time order, and a closed batch is never reopened, so whenever the batch id is a rounded time field a file or S3 source automatically orders each file's records by it before handing them over — no setting to turn on. A Parquet file holding five minutes of records yields five batches, one per minute, in order, however its rows were written. The ordering applies to numeric keys under time rounding; a file keyed by an unrounded string or integer arrives in file order and should be written grouped by that key — the ingester does not reorder it for you.

A custom ingester¶

The ingester is one program built on the SDK's ingest client, and a customer can build another: FermiHDIIngestClient registers with the DMS as an HD_CLIENT node (a registration token, a certificate from the DMS-CA, the SCC targets in return), opens the FHDWP data plane on the interface it is given, and sends its batches straight to the Storage Cluster Controllers — the same path the ingester uses, without any of its sources. It draws on the HD_CLIENT element of the licence, or the HD_ANY pool. The ingest client is available from C++ and Python; the SDK overview has the modes and the bindings that support each.

Verifying a source before it feeds a deployment¶

--dry-run (or FERMIHDI_DRY_RUN=1) runs the same published ingester image with the same source and format handling described above, but registers with no DMS, connects to no SCC and needs no licence. It reads, decodes and maps records exactly as a real ingest would — the same field-name matching, the same batching, the same per-field warnings — and reports a summary instead of sending anything anywhere:

docker run --rm \
  -v /path/to/corpus:/data:ro -v /path/to/schema.json:/schema.json:ro \
  cr.fermihdi.io/hd/ingester:<tag> \
  --dry-run --ingression-type file --file-dir /data --schema /schema.json

Every source and format flag documented above works the same way under --dry-run. Two flags are specific to it: --dry-run-idle-exit-sec (default 5; stop after this many seconds with no new record, including when none ever arrived, so a misconfigured source fails fast rather than waiting forever — 0 runs until Ctrl-C, for watching a live stream as long as wanted) and --dry-run-max-records (default unlimited; stop after this many records, for a quick check against a large corpus).

The run ends with a summary — records read, batches formed, and the same mapping-warning counters What to watch describes below (enum defaults, array truncation, clamped time deltas, unmapped chess or uuid values) — and exits non-zero if nothing was read at all. A clean run, with records read and every warning counter at zero, is the strongest evidence this pipeline has that a source and a dataset schema are actually compatible, without touching a real deployment.

What to watch¶

The ingester registers with the DMS like every node and serves /healthz and /metrics on its health port (8081 by default). The figures that matter are records in, batches closed, and the acknowledgement lag from the storage cluster; a file source that stops consuming files has usually stopped receiving acknowledgements, and the SCC's log says why. The engine sizing — two cores at the floor, four allocated, 300 × 2 MB hugepages — is in the Administrator's Guide.