hubcio commented on code in PR #4505: URL: https://github.com/apache/iggy/pull/4505#discussion_r4271951601
########## core/connectors/sinks/questdb_sink/README.md: ########## @@ -0,0 +1,312 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. +--> + +# QuestDB Sink Connector + +Writes Apache Iggy messages to [QuestDB](https://questdb.com) over QWP, +QuestDB's native binary protocol, on a WebSocket transport. The same build +works with QuestDB open source and QuestDB Enterprise. Enterprise features are +enabled through the connect string. + +Requires QuestDB 10.0 or newer. + +## Configuration + +```toml +[plugin_config] +connection_string = "ws::addr=localhost:9000;" +table = "iggy_events" +timestamp_source = "message" +timestamp_unit = "auto" +symbol_columns = ["region", "device"] +uuid_columns = [] +integer_columns = [] +include_stream_column = true +include_topic_column = true +include_partition_column = false +include_offset_column = false +include_headers = false +ack_level = "ok" +flush_timeout = "30s" +batch_size = 1000 +max_flush_bytes = 1000000 +log_rejected_payload = false +numbers_as_double = true +verbose_logging = false +``` + +| Field | Default | Description | +| ----- | ------- | ----------- | +| `connection_string` | required | QuestDB connect string, `ws::` or `wss::`. Treated as a secret. | +| `table` | required | Target table. | +| `timestamp_source` | `message` | `message`, `origin`, `payload`, or `server`. | +| `timestamp_field` | none | Payload field holding the timestamp. Required when `timestamp_source = "payload"`. | +| `timestamp_unit` | `auto` | `auto`, `seconds`, `millis`, `micros`, `nanos`. `auto` infers the unit from the magnitude. | +| `symbol_columns` | `[]` | Payload fields stored as `SYMBOL`. A record with an object or array in one of these fields is rejected. | +| `uuid_columns` | `[]` | Payload fields holding RFC-4122 strings, stored as `UUID`. A record with a non-string value there is rejected. | +| `integer_columns` | `[]` | Payload fields stored as `LONG` while `numbers_as_double` is on. A fractional value there is rejected. | +| `include_stream_column` | `true` | Write the stream name as a `SYMBOL` column named `stream`. | +| `include_topic_column` | `true` | Write the topic name as a `SYMBOL` column named `topic`. | +| `include_partition_column` | `false` | Write `partition_id` as a `LONG`. | +| `include_offset_column` | `false` | Write `offset` as a `LONG`. | +| `include_headers` | `false` | Write each header as a `VARCHAR` column named `header_<key>`. Binary header values are base64-encoded. | +| `ack_level` | `ok` | `ok` waits for the server to accept a flush. `durable` waits for the Enterprise durable ack. | +| `flush_timeout` | `30s` | How long a flush waits for its ack without progress. `0s` with `ack_level = "ok"` means fire and forget. `0s` is not allowed with `durable`. | +| `batch_size` | `1000` | Maximum rows per flush. | +| `max_flush_bytes` | `1000000` | Flush when the buffer reaches this size, even before `batch_size` rows. | +| `numbers_as_double` | `true` | Write every JSON number as a `DOUBLE`. See [Numbers](#numbers). | +| `log_rejected_payload` | `false` | Add a truncated payload to the log line of a rejected record. | +| `verbose_logging` | `false` | Log per-batch details at `info` instead of `debug`. | + +Field names in the column lists match without regard to case, as in QuestDB. + +Credentials, TLS, store-and-forward, reconnect and multi-host failover are set +in the connect string. See the +[connect string reference](https://questdb.com/docs/connect/clients/connect-string/). + +### Batch sizing + +A QWP frame has a size cap, about 4 MiB by default, and a frame over the cap is +rejected with all of its rows. `max_flush_bytes` flushes before a batch of wide +rows reaches it. The 1 MB default leaves room, because the buffer size is an +estimate and a frame can also carry the symbol dictionary. If you raise it, also +raise `sf_max_segment_bytes` in the connect string. + +## Store-and-forward + +Set `sf_dir` in the connect string to persist outgoing frames on disk. Frames +the server has not acknowledged are sent again after a disconnect or a runtime +restart. + +```toml +connection_string = "ws::addr=localhost:9000;sf_dir=/var/lib/iggy/questdb-sf;sender_id=iggy;" +``` + +Use it in production. The connectors runtime commits offsets before +`consume()` runs ([#2928](https://github.com/apache/iggy/issues/2928)) and does +not replay a failed batch ([#2927](https://github.com/apache/iggy/issues/2927)), +so rows lost in the sink cannot be read again from Iggy. + +- The parent directory of `sf_dir` must exist. `~` is not expanded. +- Give each connector instance its own `sender_id` or its own `sf_dir`. + Otherwise one instance can replay frames the other queued. +- A frame that QuestDB rejects terminally, for example on a schema mismatch, + stays on disk and is sent again after a restart. To recover, delete the slot + directory under `<sf_dir>/<sender_id>-ingest-*/`. +- If a symbol dictionary fills up, the slot carries it to the next connection, + and flushes keep failing until the slot drains. This usually means a + `symbol_columns` field with unbounded cardinality. Store it as a `VARCHAR`. + +## Connection pool + +The connector borrows a pooled connection for each chunk it writes, and the +runtime writes each topic in its own task. `sender_pool_max` caps the pool and +defaults to 4. When the pool is full, a borrow waits `acquire_timeout_ms` (5 s by +default) and then fails, and that batch is lost. Set `sender_pool_max` to at +least the number of topics the connector consumes: + +```toml +connection_string = "ws::addr=localhost:9000;sender_pool_max=16;" +``` + +## Enterprise + +- **Multi-host failover**: list several endpoints, for example + `addr=node-a:9000,node-b:9000`. The client moves to another endpoint on + reconnect and keeps buffering while a replica is promoted. +- **TLS**: use `wss::`, with `tls_roots` and `tls_roots_password` for a private + CA. `tls_verify=unsafe_off` turns off certificate checks. A default build + refuses it. Build with the `insecure-skip-verify` feature to use it against a + test server. +- **Durable ack**: set `request_durable_ack=on` in the connect string and + `ack_level = "durable"`. The connector does not start with `durable` alone, + because QuestDB would refuse every wait. The server also needs replication + configured. Without it no flush is ever acknowledged. + +### Choosing an ack level + +`ok` returns when the server accepts a flush. `durable` also waits until the +write-ahead log is shipped. With `ok`, a primary that fails before shipping can +lose rows it already acknowledged. Use `durable` when losing that tail matters. + +`durable` costs one round trip per flush, so larger batches reduce the cost. In +one test that flushed 10 rows every 250 ms, `durable` reached about 8 rows/s and +`ok` about 40 rows/s. + +## Types + +QuestDB creates the table and its columns from the first rows it receives. +Create the table yourself if you need partitioning, `TTL`, `DEDUP UPSERT KEYS`, +symbol capacities or a specific designated timestamp name. + +| JSON | QuestDB | +| ---- | ------- | +| string | `VARCHAR`, or `SYMBOL` / `UUID` when listed in `symbol_columns` / `uuid_columns` | +| number | `DOUBLE`, or `LONG` when listed in `integer_columns` | +| boolean | `BOOLEAN` | +| array of numbers | `DOUBLE[]`, up to 3 dimensions. An empty array is rejected. | +| object | `VARCHAR` holding the JSON text | +| `null` | column omitted | + +Integer arrays are stored as `DOUBLE[]`, because QuestDB has no other array +type. `BOOLEAN` has no null, so an omitted boolean reads back as `false`. + +A payload that is not a JSON object goes into one `VARCHAR` column named +`payload`. That covers `Raw`, `Text`, `FlatBuffer` and `Avro` payloads, proto +text that is not JSON, and JSON arrays or scalars. + +### Numbers + +JSON has one number type, so a field can arrive as `20` in one record and +`20.5` in the next. With `numbers_as_double = true`, the default, every number +is written as a `DOUBLE`, and the column type does not depend on which value +came first. + +A `DOUBLE` holds integers exactly only up to 2^53 (9007199254740992). Values +such as nanosecond epochs, snowflake ids and hashes are larger. List those +fields in `integer_columns` to store them as `LONG`: + +```toml +integer_columns = ["trade_id", "event_nanos"] +``` + +A whole value written as `20.0` is accepted in an `integer_columns` field. A +fractional value or a non-number there is rejected. When the connector rounds a +number past 2^53 to a `DOUBLE`, it logs one warning per kind of value with the +column name. + +With `numbers_as_double = false`, each value takes its type from its JSON shape: +integers become `LONG` and fractions `DOUBLE`. Two problems follow: + +- The client fixes a column's type from the first row of each flush. Later rows + in that flush with the other number shape are rejected. +- If the first rows of an auto-created column are all whole numbers, QuestDB + creates it as `LONG` for good. + +Turn it off only if every numeric field is a true integer and you create the +table with explicit column types. + +### Timestamps + +An auto-created table names its designated timestamp `timestamp`. A payload +field with the same name collides with it, and the server rejects the whole +flush. Rename the field with a transform, or create the table with +`timestamp(<name>)`. The field named by `timestamp_field` is used only as the +designated timestamp. + +`message` and `origin` timestamps are in microseconds, so an auto-created table +gets a `TIMESTAMP` column. `payload` timestamps are converted to nanoseconds and +give a `TIMESTAMP_NS` column. With `server`, QuestDB stamps each row on arrival. +Use one source per table. + +## Delivery semantics + +Delivery is at-least-once. A reconnect, a failover or a store-and-forward replay +can send rows again. Declare +[`DEDUP UPSERT KEYS(...)`](https://questdb.com/docs/concepts/deduplication/) on +tables that must not have duplicates. + +A batch fails when: + +- QuestDB rejects a frame terminally. +- A chunk fails to write. If the cause is the connection, the connector skips + the remaining chunks of the batch. It logs the offsets and message ids of the + messages it did not write. +- No flush is acknowledged for 10 batches in a row on a topic. A single late + ack is logged as a warning, because the frames stay queued and are delivered + later. + +The runtime counts a failed batch in `iggy_connector_errors_total`. It does not Review Comment: lets leave it as is. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
