test: producer-to-collector routing on pg_stat_ch.block_format

docker/docker-compose.arrow-route.yml

@@ -0,0 +1,23 @@
+# Standalone collector profile for t/037_arrow_routing.pl. Kept separate
+# from docker-compose.otel.yml (which serves t/024_otel_export.pl) so the
+# two tests can run concurrently — different ports + container names.
+#
+# Output JSONL files land in ./otel-routing/output/, bind-mounted into the
+# container at /output. The test cleans this directory before each run.
+
+services:
+  routing-otelcol:
+    image: otel/opentelemetry-collector-contrib:0.120.0
+    container_name: psch-routing-otelcol
+    command: ["--config=/etc/otelcol-contrib/config.yaml"]
+    volumes:
+      - ./otel-routing/collector-config.yaml:/etc/otelcol-contrib/config.yaml
+      - ./otel-routing/output:/output
+    ports:
+      - "14317:4317"   # gRPC OTLP receiver (host:14317 so it doesn't clash with otel-test's 4317)
+      - "23133:13133"  # Health check HTTP endpoint
+    healthcheck:
+      test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://localhost:13133/"]
+      interval: 5s
+      timeout: 5s
+      retries: 10

docker/otel-routing/collector-config.yaml

@@ -0,0 +1,69 @@
+# Minimum-faithful OTel collector config for the producer-to-collector
+# routing test (t/037_arrow_routing.pl).
+#
+# Mirrors the SHAPE of the prod datagres-otelcol pipeline (routing connector
+# fanning Arrow batches between the legacy query_logs_arrow target and the new
+# events_raw target based on the pg_stat_ch.block_format resource attribute)
+# but with stdout/file sinks instead of real receivers. The test asserts that
+# routing decisions match the producer-side marker emission contract — it does
+# NOT validate any downstream wire format. That's covered by:
+#   - t/036_unified_arrow_e2e.pl (producer Arrow IPC -> direct curl into CH)
+#   - clickgres-platform's receiver-faithful e2e test (producer -> real
+#     datagres-arrow-exporter -> CH events_raw)
+
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:4317
+
+connectors:
+  routing:
+    # Default pipeline catches anything that doesn't match a routing rule.
+    # In the test we treat hits on the default pipeline as routing failures
+    # (the producer should always emit one of the two known marker values).
+    default_pipelines: [logs/default]
+    error_mode: ignore
+    table:
+      - context: resource
+        statement: route() where attributes["pg_stat_ch.block_format"] == "arrow_events_raw"
+        pipelines: [logs/events_raw]
+      - context: resource
+        statement: route() where attributes["pg_stat_ch.block_format"] == "arrow_ipc"
+        pipelines: [logs/legacy]
+
+exporters:
+  # JSONL sinks — one file per route. Tests parse these directly to assert
+  # that batches landed on the expected pipeline.
+  file/legacy:
+    path: /output/legacy.jsonl
+  file/events_raw:
+    path: /output/events_raw.jsonl
+  file/default:
+    path: /output/default.jsonl
+  # Keep one stdout sink for easier debugging when a test fails locally.
+  debug:
+    verbosity: basic
+
+extensions:
+  health_check:
+    endpoint: 0.0.0.0:13133
+
+service:
+  extensions: [health_check]
+  pipelines:
+    # Ingress: receive OTLP gRPC and hand off to the routing connector.
+    logs/in:
+      receivers: [otlp]
+      exporters: [routing]
+    # Routed legs. Each pipeline takes its input from the connector and
+    # writes to a dedicated file plus the shared debug stdout.
+    logs/legacy:
+      receivers: [routing]
+      exporters: [file/legacy, debug]
+    logs/events_raw:
+      receivers: [routing]
+      exporters: [file/events_raw, debug]
+    logs/default:
+      receivers: [routing]
+      exporters: [file/default, debug]

docker/otel-routing/output/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

include/config/guc.h

@@ -33,6 +33,7 @@ extern int psch_min_duration_us;
 extern int psch_normalize_cache_max;
 extern double psch_sample_rate;
 extern bool psch_otel_arrow_passthrough;
+extern bool psch_use_unified_arrow_exporter;
 extern int psch_otel_max_block_bytes;
 extern char* psch_extra_attributes;
 extern char* psch_debug_arrow_dump_dir;

src/config/guc.c

@@ -35,6 +35,7 @@ int psch_min_duration_us = 0;
 int psch_normalize_cache_max = 32768;
 double psch_sample_rate = 1.0;
 bool psch_otel_arrow_passthrough = false;
+bool psch_use_unified_arrow_exporter = false;
 int psch_otel_max_block_bytes = 3 * 1024 * 1024;  // 3 MiB (max: 16 MiB)
 char* psch_extra_attributes = NULL;
 char* psch_debug_arrow_dump_dir = NULL;
@@ -356,6 +357,23 @@ void PschInitGuc(void) {
       0,
       NULL, NULL, NULL);
 
+  DefineCustomBoolVariable(
+      "pg_stat_ch.use_unified_arrow_exporter",
+      "Use the StatsExporter-implementing Arrow exporter instead of arrow_batch.cc.",
+      "When enabled together with use_otel and otel_arrow_passthrough, the bgworker "
+      "builds Arrow IPC via the typed StatsExporter column interface, writing the "
+      "events_raw schema (typed integer ids, no sprintf decimal-string encoding). "
+      "When off, the legacy arrow_batch.cc path runs and produces the query_logs_arrow "
+      "wire shape. Default off; flip on to opt a producer into the new exporter. "
+      "PGC_POSTMASTER: the bgworker picks the exporter implementation at init time, "
+      "so a runtime change would mismatch the per-cycle dispatcher (Arrow batches "
+      "would silently drop, since OTelArrowExporter does not implement SendArrowBatch).",
+      &psch_use_unified_arrow_exporter,
+      false,
+      PGC_POSTMASTER,
+      0,
+      NULL, NULL, NULL);
+
   DefineCustomIntVariable(
       "pg_stat_ch.otel_max_block_bytes",
       "Maximum Arrow batch size in bytes per OTLP request.",

src/export/exporter_interface.h

@@ -96,10 +96,20 @@ class StatsExporter {
   virtual int NumConsecutiveFailures() const = 0;
   virtual void ResetFailures() = 0;
   virtual int NumExported() const = 0;
-  virtual bool SendArrowBatch(const uint8_t* ipc_data, size_t ipc_len, int num_rows) {
+  // block_format is the discriminator emitted as the
+  // pg_stat_ch.block_format resource + log-record attribute on the OTLP
+  // request. The central OTel collector's routingconnector matches on
+  // it to fan to the right downstream receiver (e.g. legacy
+  // query_logs_arrow vs new events_raw). Callers pass an explicit value
+  // because Google Style forbids default arguments on virtual methods
+  // (the default is dispatched on the static type, not the dynamic one,
+  // which is a footgun for overriders).
+  virtual bool SendArrowBatch(const uint8_t* ipc_data, size_t ipc_len, int num_rows,
+                              string_view block_format) {
     (void)ipc_data;
     (void)ipc_len;
     (void)num_rows;
+    (void)block_format;
     return false;
   }