polyc-query 2026.8.3

//! The `events` wide-table `TableProvider` stub.
//!
//! Fixed schema columns per docs/reference/datafusion-data-layer.md:205-215
//! ("The journal `TableProvider`"): `partition`, `position`, `kind`,
//! `kind_base`, `turn_id`, `trust`, `payload`, `payload_json`. `schema()`
//! reports those columns; `scan()` stays metadata-only, deferring real I/O
//! to a
//! `SendableRecordBatchStream` polled by the execution plan — the
//! three-layer split (planning, execution-plan, stream) `DataFusion`'s own
//! custom-provider guidance documents. Phase 1 (this crate today) drives
//! `schema()` off a `MemTable` built by [`crate::decode::events_batch`]
//! (docs/reference/datafusion-data-layer.md's Rollout phase 1: "a `MemTable`-
//! style batch decode"); `scan()` — the streaming provider dispatching over
//! `EventLogHost`'s shard channel — is phase 3 and stays unimplemented here.

use std::sync::Arc;

use arrow::datatypes::{DataType, Field, Schema, SchemaRef};

/// Fixed column names for the `events` wide table, in schema order.
///
/// docs/reference/datafusion-data-layer.md:212, plus `payload_json` — the
/// schema-less/freeform residual's "opaque and JSON-queryable now"
/// exposure (#1313, #1178): whether a payload is JSON is a property of the
/// bytes, not the kind, so the column is computed uniformly for every row
/// rather than gated per kind.
// Sealing this module to `pub(crate)` unmasked `EVENTS_COLUMNS` and
// `EventsTableProvider`'s `new`/`partition`/`partition` field as `dead_code`
// outside `#[cfg(test)]` builds: only `schema()` is called from production
// code (`crate::engine::QueryEngine::build`) today — the rest of this
// provider stub is exercised only by this module's own tests, pending the
// phase-3 real `TableProvider::scan` implementation this struct is scaffolding
// for (see the struct's own doc).
#[allow(dead_code)]
pub(crate) const EVENTS_COLUMNS: [&str; 8] = [
    "partition",
    "position",
    "kind",
    "kind_base",
    "turn_id",
    "trust",
    "payload",
    "payload_json",
];

/// The `events` wide-table `TableProvider`, scoped to one journal partition.
///
/// A real implementation implements `datafusion::catalog::TableProvider`:
/// `schema()` returns [`EVENTS_COLUMNS`]' Arrow schema; `scan()` dispatches
/// across `EventLogHost`'s shard/thread/`mpsc`-channel boundary rather than
/// reopening the journal per query (docs/reference/datafusion-data-layer.md:254-266).
/// This skeleton implements `schema()`; `scan()` — the streaming provider —
/// stays unimplemented (phase 1 wires `schema()` against `MemTable` batches
/// built by [`crate::decode::events_batch`], not a live `TableProvider::scan`).
#[derive(Debug)]
pub(crate) struct EventsTableProvider {
    /// Journal partition this provider reads (`conv-{id}`).
    #[allow(dead_code)] // see the `allow` above `EVENTS_COLUMNS`
    partition: String,
}

impl EventsTableProvider {
    /// Build a provider stub for `partition`. No journal handle is opened.
    #[must_use]
    #[allow(dead_code)] // see the `allow` above `EVENTS_COLUMNS`
    pub(crate) fn new(partition: impl Into<String>) -> Self {
        Self {
            partition: partition.into(),
        }
    }

    /// The journal partition this provider reads.
    #[must_use]
    #[allow(dead_code)] // see the `allow` above `EVENTS_COLUMNS`
    pub(crate) fn partition(&self) -> &str {
        &self.partition
    }

    /// The fixed `events` wide-table Arrow schema, in [`EVENTS_COLUMNS`]
    /// order.
    ///
    /// Column types are chosen off `polyc_eventlog::Event`'s real fields
    /// (`crates/eventlog-model/src/event.rs`), not guessed:
    ///
    /// - `partition` (`Utf8`, non-null) — the journal partition name, not a
    ///   field on `Event` itself; supplied by the caller building the batch
    ///   (one partition per provider).
    /// - `position` (`UInt64`, non-null) — the journal's own monotonic
    ///   append position (`crate::decode::events_batch`'s `u64` input),
    ///   never carried on `Event` (see that struct's docs).
    /// - `kind` (`Utf8`, non-null) — `Event::kind` verbatim, the full
    ///   `base[:turn_uuid]` string.
    /// - `kind_base` (`Utf8`, non-null) — the `base` half of `kind`, split
    ///   via `polyc_proto::kinds::parse` (never re-derived locally).
    /// - `turn_id` (`Utf8`, NULLABLE) — the `:{turn_uuid}` suffix, present
    ///   only for per-turn kinds; a bare kind (memory/control events) has
    ///   none.
    /// - `trust` (`Utf8`, non-null) — `Event::trust` is `polyc_eventlog::TrustTag`,
    ///   a `#[repr(u8)]` enum with three variants (`Unspecified`,
    ///   `TrustedUser`, `QuarantinedContent`). This maps to `Utf8`, not the
    ///   raw discriminant `UInt8`, via `TrustTag::as_str()` — the tag's own
    ///   stable lowercase label (`"unspecified"` / `"trusted_user"` /
    ///   `"quarantined_content"`), already used for forensics rendering.
    ///   SQL callers get a self-documenting `WHERE trust = 'quarantined_content'`
    ///   predicate instead of a numeric code they have to look up.
    /// - `payload` (`Binary`, non-null) — `Event::payload`'s raw `Vec<u8>`,
    ///   opaque and byte-for-byte, exactly as the journal stores it.
    /// - `payload_json` (`Utf8`, NULLABLE) — `payload` re-exposed as text
    ///   when (and only when) it is valid UTF-8 that itself parses as JSON;
    ///   `NULL` for a binary or non-JSON payload. This is the "opaque and
    ///   JSON-queryable now" exposure for the schema-less/freeform kinds
    ///   (`approval_deferred`, `tool_input_rewrite`,
    ///   `tool_context_injection`, `tool_result_redaction`,
    ///   `taint_excision`, `admin_model_change`, and the freeform
    ///   approval/payment `args_json`-shaped payloads) (#1313, #1178). It is
    ///   uniform across every row rather than gated by
    ///   [`crate::decode::Decode`]'s typed/opaque split: whether a given
    ///   kind's payload happens to be JSON is a property of the bytes, not
    ///   of the kind, so branching this column on the decode registry would
    ///   only add a special case with no behavioral difference (a `Typed`
    ///   kind's payload is protobuf, not JSON, and decodes to `NULL` here
    ///   exactly as any other non-JSON payload would). See
    ///   [`crate::decode::events_batch`] for the decode step and why no
    ///   dedicated `DataFusion` JSON scalar function backs the column (none
    ///   ships in the pinned `datafusion = "=54.0.0"`).
    #[must_use]
    pub(crate) fn schema() -> SchemaRef {
        Arc::new(Schema::new(vec![
            Field::new("partition", DataType::Utf8, false),
            Field::new("position", DataType::UInt64, false),
            Field::new("kind", DataType::Utf8, false),
            Field::new("kind_base", DataType::Utf8, false),
            Field::new("turn_id", DataType::Utf8, true),
            Field::new("trust", DataType::Utf8, false),
            Field::new("payload", DataType::Binary, false),
            Field::new("payload_json", DataType::Utf8, true),
        ]))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// `schema()` reports exactly [`EVENTS_COLUMNS`], in order, with the
    /// documented types and nullability — the shape the design doc's
    /// journal `TableProvider` section fixes.
    #[test]
    fn schema_matches_events_columns() {
        let schema = EventsTableProvider::schema();
        let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(names, EVENTS_COLUMNS.to_vec());

        let expect = [
            ("partition", DataType::Utf8, false),
            ("position", DataType::UInt64, false),
            ("kind", DataType::Utf8, false),
            ("kind_base", DataType::Utf8, false),
            ("turn_id", DataType::Utf8, true),
            ("trust", DataType::Utf8, false),
            ("payload", DataType::Binary, false),
            ("payload_json", DataType::Utf8, true),
        ];
        for (field, (name, ty, nullable)) in schema.fields().iter().zip(expect) {
            assert_eq!(field.name(), name);
            assert_eq!(field.data_type(), &ty);
            assert_eq!(field.is_nullable(), nullable);
        }
    }

    #[test]
    fn provider_reports_its_partition() {
        let provider = EventsTableProvider::new("conv-abc");
        assert_eq!(provider.partition(), "conv-abc");
    }
}