pg_lake: clamp string and binary values to per-column byte limits on Iceberg writes

Some downstream consumers of Iceberg tables impose per-column byte
caps smaller than what PostgreSQL values in the source can carry.
For example, on Snowflake the column-type byte ceilings are:

  - STRING / VARCHAR : 16 MiB default, up to 128 MiB when declared
                      with an explicit larger length.
  - BINARY           : 8 MiB default, up to 64 MiB.
  - OBJECT / ARRAY /
    VARIANT          : 128 MiB.

Without a guard, rows whose individual values exceed the target
column's cap reach the consumer and surface as opaque "value too
long" errors when the consumer ingests them.  This adds an opt-in,
GUC-driven clamp at write time so values that would exceed a
configurable byte limit are normalized in place before they are
written to the Iceberg table.

Two new GUCs (PGC_USERSET, GUC_UNIT_BYTE, default 0 = disabled, no
behavior change unless set):

  - pg_lake_engine.iceberg_max_string_bytes governs text, varchar,
    bpchar, jsonb, json, and the aggregate JSON-serialized size of
    array, composite, and map values that land in STRING / OBJECT /
    ARRAY columns on the consumer side.
  - pg_lake_engine.iceberg_max_binary_bytes governs bytea.

Operators set the GUCs to the target column's actual ceiling, which
may be the default (16 MiB / 8 MiB on Snowflake) or a higher value
(up to 128 MiB / 64 MiB) when the destination column was declared
with an explicit larger length.

Per-leaf behavior, applied recursively through arrays, composites,
maps, and domains via IcebergSizeClampDatum:

  - text/varchar/bpchar:  truncate at a UTF-8 character boundary
                          (pg_mbcliplen).
  - bytea:                byte-truncate.
  - jsonb/json:           replace with NULL, since truncating the
                          serialized form would yield invalid JSON.

For jsonb the byte limit applies to the text-serialized form (what
the consumer ultimately receives), not the on-disk binary varlena, so
the size check serializes via jsonb_out before deciding.  For json
(stored as text) the varlena length matches what the consumer sees,
so VARSIZE_ANY_EXHDR is used directly.

In addition, an aggregate-size check NULLs an entire array or
composite when the sum of its leaf byte sizes exceeds
iceberg_max_string_bytes — the limit on the JSON-serialized form
that downstream consumers receive when an array or struct lands in
a STRING / OBJECT / ARRAY column.  Snowflake stores semi-structured
data without per-record headers, so sum-of-leaves is a close-enough
approximation of the serialized JSON length without paying for an
extra serialization pass.  Each recursion level reports its post-
clamp byte size up to its parent so the roll-up costs no extra walk.

Aggregate clamping covers all containers, not just those whose
elements/fields are themselves clampable.  For arrays of non-string
element types (e.g. int[]), the array's varlena content size
(VARSIZE_ANY_EXHDR) serves as a cheap upper-bound proxy for the JSON
serialization length; for composite fields of non-clampable types,
the field's typlen-based size is folded into the aggregate.

The new size clamp slots into the same FDW write path as the existing
out_of_range_values clamp.  ClampAndCheckConstraints invokes
IcebergSizeClampSlotInPlace after the temporal/numeric clamp and before
ExecConstraints, so a NOT NULL column whose value is replaced with NULL
fails the constraint (mirroring the existing numeric-NaN clamp
behavior).  PgLakeModifyState gains a needsSizeClamping flag computed
once at init via TupleDescNeedsIcebergSizeClamping, so the per-row work
is skipped entirely on tables whose columns cannot trigger size
clamping; the slot helper is also short-circuited unless at least one
GUC is non-zero.

Tests: pg_lake_table/tests/pytests/test_iceberg_size_clamping.py
covers UTF-8 boundary truncation, bytea byte truncation, jsonb/json →
NULL, NOT NULL constraint violation after clamp-to-NULL, recursion
into text[] and composite types, aggregate-size NULLing of array and
composite values whose leaves individually fit (including int[] and
all-non-clampable composite fields), the disabled-by-default no-op
path, and the only-one-GUC-set path.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Marco Slot <marco.slot@snowflake.com>

Author: sfc-gh-mslot

pg_lake_engine/include/pg_lake/pgduck/iceberg_datum_validation.h

@@ -41,3 +41,27 @@ extern PGDLLEXPORT Datum IcebergErrorOrClampDatum(Datum value, Oid typeOid,
 												  int32 typmod,
 												  IcebergOutOfRangePolicy policy,
 												  bool *isNull);
+
+/*
+ * IcebergSizeClampDatum truncates or NULLs a Datum so that string and
+ * binary values fit the byte limits expressed by
+ * pg_lake_engine.iceberg_max_string_bytes and
+ * pg_lake_engine.iceberg_max_binary_bytes (0 = no limit).
+ *
+ * Lossless types are truncated:
+ *   - text/varchar/bpchar  -> trimmed at a UTF-8 character boundary to
+ *                             iceberg_max_string_bytes.
+ *   - bytea                -> byte-truncated to iceberg_max_binary_bytes.
+ *
+ * Structured-string types are replaced with NULL via *isNull = true,
+ * since truncation would corrupt them:
+ *   - jsonb/json
+ *
+ * Recurses through arrays, composites, maps, and domains.  Nested values
+ * that would be NULLed are absorbed as NULL within the reconstructed
+ * container.
+ *
+ * If both GUCs are 0, the value is returned unchanged regardless of type.
+ */
+extern PGDLLEXPORT Datum IcebergSizeClampDatum(Datum value, Oid typeOid,
+											   int32 typmod, bool *isNull);

pg_lake_engine/include/pg_lake/pgduck/iceberg_validation.h

@@ -72,3 +72,22 @@ extern PGDLLEXPORT bool TypeNeedsIcebergValidation(Oid typeOid, int32 typmod,
 #define TEMPORAL_DATE_MIN_YEAR		(-4712)
 #define TEMPORAL_TIMESTAMP_MIN_YEAR	1
 #define TEMPORAL_MAX_YEAR			9999
+
+/*
+ * Downstream byte limits for values written to Iceberg tables, set via the
+ * pg_lake_engine.iceberg_max_string_bytes and
+ * pg_lake_engine.iceberg_max_binary_bytes GUCs.  0 means no limit.  These
+ * caps are imposed by some downstream consumers (e.g. Snowflake VARCHAR
+ * 16 MiB / BINARY 8 MiB) and applied via IcebergSizeClampDatum.
+ */
+extern PGDLLEXPORT int IcebergMaxStringBytes;
+extern PGDLLEXPORT int IcebergMaxBinaryBytes;
+
+/*
+ * TypeNeedsIcebergSizeClamping returns true if a Datum of typeOid (or any
+ * lossless string / structured-string / bytea component nested within it)
+ * could potentially be size-clamped by IcebergSizeClampDatum.  Recurses
+ * through arrays, composites, maps, and domains.  Independent of the
+ * current GUC values.
+ */
+extern PGDLLEXPORT bool TypeNeedsIcebergSizeClamping(Oid typeOid);

pg_lake_engine/src/init.c

@@ -44,6 +44,7 @@
 #include "pg_extension_base/pg_extension_base_ids.h"
 #include "pg_lake/pgduck/cache_worker.h"
 #include "pg_lake/pgduck/client.h"
+#include "pg_lake/pgduck/iceberg_validation.h"
 #include "pg_lake/util/s3_writer_utils.h"
 #include "utils/guc.h"
 
@@ -186,6 +187,33 @@ _PG_init(void)
 							0,
 							NULL, NULL, NULL);
 
+	DefineCustomIntVariable("pg_lake_engine.iceberg_max_string_bytes",
+							gettext_noop("Maximum bytes for string values written to "
+										 "Iceberg tables. Values of text/varchar/bpchar "
+										 "exceeding this size are truncated at a UTF-8 "
+										 "character boundary; values of jsonb/json are "
+										 "replaced with NULL since truncation would "
+										 "corrupt the structure. 0 disables the limit. "
+										 "Intended for downstream consumers (e.g. "
+										 "Snowflake) that impose per-column byte caps."),
+							NULL,
+							&IcebergMaxStringBytes,
+							0, 0, INT_MAX,
+							PGC_USERSET,
+							GUC_UNIT_BYTE,
+							NULL, NULL, NULL);
+
+	DefineCustomIntVariable("pg_lake_engine.iceberg_max_binary_bytes",
+							gettext_noop("Maximum bytes for bytea values written to "
+										 "Iceberg tables. Values exceeding this size are "
+										 "byte-truncated. 0 disables the limit."),
+							NULL,
+							&IcebergMaxBinaryBytes,
+							0, 0, INT_MAX,
+							PGC_USERSET,
+							GUC_UNIT_BYTE,
+							NULL, NULL, NULL);
+
 	DefineCustomStringVariable(
 							   "pg_lake.stage_location",
 							   gettext_noop("Base URL for @STAGE/ resolution in paths"),