syntax = "proto3";

package nominal.iceberg.v1;

import "buf/validate/validate.proto";
import "google/protobuf/timestamp.proto";
import "nominal/protos/iceberg/v1/table.proto";

option java_multiple_files = true;
option java_package = "io.nominal.iceberg.v1";

// One Parquet file just uploaded to object storage by the streaming Iceberg
// writer. The downstream committer reads these from a Kafka "WAL" topic,
// groups them per Iceberg TableIdentifier, and registers them with Iceberg in
// a single AppendFiles per (table, batch) snapshot.
//
// The writer encodes the full table target (namespace + table name) AND the
// resolved partition value the file belongs in.
//
// File-level statistics (row count, column min/max, split offsets, file size)
// are derived at commit time from the Parquet file footer so the WAL schema stays independent of Iceberg internals.
//
// The Kafka record key is DatasetRid so all files for one dataset land on the same partition.
// The producer is the authority on table location — the consumer never re-derives namespaces.
message IcebergFileWrittenEvent {
  // Full Iceberg table location (namespace + table name).
  TableTarget table = 1 [(buf.validate.field).required = true];

  // Pre-computed Iceberg partition value. For series tables (identity partition
  // on int month) this is raw yyyymm; for data tables (month(timestamp)
  // transform) this is months-since-epoch. The committer passes this through
  // unchanged to PartitionData.set(0, ...).
  int32 partition_value = 2;

  // Storage-agnostic object key, e.g. s3://… URIs for S3/MinIO or gs://… URIs for GCS.
  // The Iceberg catalog's FileIO resolves the appropriate cheme at commit time.
  // Also used as the dedupe key for the snapshot-summary ledger (nominal.appended-paths), so producers must
  // emit deterministic keys across writer retries.
  string object_key = 3 [(buf.validate.field).string.min_len = 1];

  // Wall-clock ingest timestamp captured by the writer, primarily for
  // observability (lag, end-to-end latency monitoring).
  google.protobuf.Timestamp ingested_at = 4;

  // Provenance of the upstream channel-writer Kafka batch this file was written
  // from. The committer dedupes by recording, per (source_topic,
  // source_partition), the highest source_end_offset it has committed into each
  // table (snapshot summary property nominal.source-watermarks) and skipping any
  // file whose source_end_offset is at or below that watermark.
  //
  // This is robust where a WAL-offset watermark would not be: the writer
  // re-emits the same file (same provenance) at a NEW WAL offset on reprocessing
  // (see IcebergStreamWriter.writeParquetToleratingAlreadyExists), so dedupe must
  // key on the upstream offset, which is stable across re-emits.
  //
  // These fields are additive, but the streaming committer assumes they are
  // present: the WAL is drained before cutover (see the watermark rollout), so
  // every streaming event produced after deploy carries provenance. An event that
  // nonetheless arrives without source_topic is treated as a new file and appended
  // (never dropped); it is NOT deduped against the file-name ledger
  // (nominal.appended-paths), which is used only by the bulk-ingest path.
  string source_topic = 5;
  int32 source_partition = 6;
  int64 source_end_offset = 7;
}
