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 that fires consistently across both the per-tuple
FDW path and the DuckDB-pushdown paths (INSERT..SELECT, snapshot
loads via postgres_scan, COPY FROM, etc.).

Three 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 (per-leaf clamp).  text/varchar/bpchar are
    truncated at a UTF-8 character boundary; jsonb/json are replaced
    with NULL since truncating the serialized form would yield invalid
    JSON.
  - pg_lake_engine.iceberg_max_binary_bytes governs bytea (byte-truncate).
  - pg_lake_engine.iceberg_max_aggregate_bytes governs the on-disk
    size of array, composite, and map values that land in OBJECT/
    ARRAY/VARIANT columns.  The whole container is replaced with NULL
    when its varlena exceeds the limit; we deliberately do NOT recurse
    into elements/fields to per-leaf-clamp inner strings, because (a)
    inner leaves don't have their own column cap on the consumer side
    — they're just JSON inside the parent OBJECT/ARRAY/VARIANT — and
    (b) when iceberg_max_string_bytes <= iceberg_max_aggregate_bytes,
    no inner leaf can exceed the per-leaf limit while staying inside
    an under-cap container.  Distinct from the string GUC because
    consumers usually cap semi-structured types more loosely than
    STRING (e.g. on Snowflake: 128 MiB vs. 16 MiB).

Operators set the GUCs to the target column's actual ceiling.

Two complementary code paths cover all writers:

1. **Per-tuple FDW path** (INSERT .. VALUES, single-row UPDATE/DELETE
   pipelines): IcebergSizeClampSlotInPlace runs in-process after the
   existing temporal/numeric clamp and before ExecConstraints.
   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.

   Hot-path optimizations: IcebergSizeClampDatum short-circuits
   varlena values whose total on-disk size is comfortably under every
   active limit (toast_raw_datum_size against the smallest active
   GUC, halved to cover jsonb's binary-vs-text gap), avoiding even
   the leaf scalar dispatch on small values.  IcebergSizeClampString-
   Scalar additionally pre-checks jsonb binary size against half the
   string limit before falling back to jsonb_out, since text
   serialization of typical jsonb data is bounded by a small constant
   factor over the binary representation.  For arrays/composites/
   maps, no deconstruct/deform happens: a single toast_raw_datum_size
   compared against iceberg_max_aggregate_bytes decides pass-through
   vs. NULL.

2. **DuckDB-pushdown path** (INSERT..SELECT, snapshot/initial-copy
   via postgres_scan, COPY FROM, compaction): a new
   IcebergWrapQueryWithSizeClampChecks rewriter wraps the inner
   SELECT with an outer projection that calls two new DuckDB scalar
   UDFs registered in duckdb_pglake (iceberg_size_clamp_text and
   iceberg_size_clamp_blob) for lossless leaf truncation, expresses
   jsonb/json NULL-on-overflow inline via strlen(::VARCHAR), and
   applies aggregate-NULL inline as
   `CASE WHEN strlen(<container>::VARCHAR) > <aggregate_max> THEN NULL
   ELSE <container> END`.  No list_transform / struct_pack inside
   containers, mirroring the per-tuple side.  The wrapper is hooked
   into pg_lake_engine/src/pgduck/write_data.c alongside the existing
   out_of_range_values clamp wrapper, so all pushdown writers
   (INSERT..SELECT, snapshot via postgres_scan, COPY FROM,
   vacuum/compaction) flow through it without per-callsite plumbing.

Per-leaf truncation output is byte-identical between the two paths.
The aggregate threshold uses different proxies — varlena size on the
PG side, JSON-serialized text length on the SQL side — but both are
within a small constant factor of the consumer-visible size.

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, container
NULLing on aggregate overflow (text[], composite-of-text, int[],
all-bigint composite), the disabled-by-default no-op path, the
only-one-GUC-set path, INSERT..SELECT pushdown clamping across
text/bytea/jsonb, and INSERT..SELECT pushdown aggregate NULLing of
int[].

🤖 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

duckdb_pglake/src/duckdb_pglake_extension.cpp

@@ -389,6 +389,174 @@ PgErrorNestedListFun(DataChunk &args, ExpressionState &state, Vector &result)
 }
 
 
+/*
+ * IcebergByteSize returns the in-memory byte footprint of any Datum.
+ *
+ * Walks the value at the vector level (no text serialization), summing
+ * VARCHAR/BLOB string_t sizes, recursing through LIST/MAP children, and
+ * adding fixed-width sizes for scalar types.  Used PG-side as a cheap
+ * proxy for "size that lands in the consumer's OBJECT/ARRAY/VARIANT
+ * column", in lieu of casting the whole container to VARCHAR (which
+ * would trigger DuckDB's text serialization).
+ *
+ * STRUCT field byte sums are computed by recursing into each field's
+ * vector at the same row index; LIST/MAP recurse over `entry.length`
+ * children starting at `entry.offset` in the child vector.
+ */
+static int64_t
+IcebergComputeByteSize(Vector &v, idx_t row, idx_t row_count)
+{
+	if (v.GetVectorType() != VectorType::FLAT_VECTOR)
+		v.Flatten(row_count);
+
+	if (FlatVector::IsNull(v, row))
+		return 0;
+
+	auto &type = v.GetType();
+
+	switch (type.id())
+	{
+		case LogicalTypeId::VARCHAR:
+		case LogicalTypeId::BLOB:
+		{
+			auto data = FlatVector::GetData<string_t>(v);
+			return (int64_t) data[row].GetSize();
+		}
+
+		case LogicalTypeId::LIST:
+		case LogicalTypeId::MAP:
+		{
+			auto list_data = FlatVector::GetData<list_entry_t>(v);
+			auto entry = list_data[row];
+			auto &child = ListVector::GetEntry(v);
+			idx_t child_size = ListVector::GetListSize(v);
+
+			int64_t total = 0;
+
+			for (idx_t i = 0; i < entry.length; i++)
+				total += IcebergComputeByteSize(child, entry.offset + i,
+												child_size);
+
+			return total;
+		}
+
+		case LogicalTypeId::STRUCT:
+		{
+			auto &children = StructVector::GetEntries(v);
+			int64_t total = 0;
+
+			for (auto &child : children)
+				total += IcebergComputeByteSize(*child, row, row_count);
+
+			return total;
+		}
+
+		default:
+			return (int64_t) GetTypeIdSize(type.InternalType());
+	}
+}
+
+
+static void
+IcebergByteSizeFun(DataChunk &args, ExpressionState &state, Vector &result)
+{
+	auto &input = args.data[0];
+	idx_t count = args.size();
+
+	if (input.GetVectorType() != VectorType::FLAT_VECTOR)
+		input.Flatten(count);
+
+	auto out = FlatVector::GetData<int64_t>(result);
+	auto &out_validity = FlatVector::Validity(result);
+
+	for (idx_t i = 0; i < count; i++)
+	{
+		if (FlatVector::IsNull(input, i))
+		{
+			out_validity.SetInvalid(i);
+			continue;
+		}
+		out[i] = IcebergComputeByteSize(input, i, count);
+	}
+}
+
+
+static unique_ptr<FunctionData>
+IcebergByteSizeBind(ClientContext &context, ScalarFunction &bound_function,
+					vector<unique_ptr<Expression>> &arguments)
+{
+	bound_function.return_type = LogicalType::BIGINT;
+	return nullptr;
+}
+
+
+/*
+ * IcebergSizeClampTextFun truncates a VARCHAR value at a UTF-8 character
+ * boundary so its byte length does not exceed the second argument.  If the
+ * limit is <= 0 the value is returned unchanged so callers can encode
+ * "disabled" as 0.  Used to enforce Snowflake STRING/VARCHAR per-column
+ * caps on the pushdown write path.
+ *
+ * Algorithm: if the input fits, return it.  Otherwise, walk back from
+ * `limit` to the nearest UTF-8 leading byte (continuation bytes have
+ * the bit pattern 10xxxxxx).  Worst case backs up at most 3 bytes,
+ * since UTF-8 codepoints are at most 4 bytes long.
+ */
+static void
+IcebergSizeClampTextFun(DataChunk &args, ExpressionState &state, Vector &result)
+{
+	BinaryExecutor::Execute<string_t, int32_t, string_t>(
+		args.data[0], args.data[1], result, args.size(),
+		[&](string_t input, int32_t limit) {
+			if (limit <= 0)
+				return input;
+
+			auto data = input.GetData();
+			auto size = (int64_t) input.GetSize();
+			int64_t lim = (int64_t) limit;
+
+			if (size <= lim)
+				return input;
+
+			int64_t trim = lim;
+			while (trim > 0 &&
+				   (((unsigned char) data[trim]) & 0xC0) == 0x80)
+			{
+				trim--;
+			}
+
+			return StringVector::AddString(result, data, trim);
+		});
+}
+
+
+/*
+ * IcebergSizeClampBlobFun byte-truncates a BLOB value to the second
+ * argument.  If the limit is <= 0 the value is returned unchanged.
+ * Used to enforce Snowflake BINARY per-column caps on the pushdown
+ * write path.
+ */
+static void
+IcebergSizeClampBlobFun(DataChunk &args, ExpressionState &state, Vector &result)
+{
+	BinaryExecutor::Execute<string_t, int32_t, string_t>(
+		args.data[0], args.data[1], result, args.size(),
+		[&](string_t input, int32_t limit) {
+			if (limit <= 0)
+				return input;
+
+			auto size = (int64_t) input.GetSize();
+			int64_t lim = (int64_t) limit;
+
+			if (size <= lim)
+				return input;
+
+			return StringVector::AddStringOrBlob(result, input.GetData(),
+												 (idx_t) lim);
+		});
+}
+
+
 
 static void LoadInternal(ExtensionLoader &loader) {
 
@@ -438,6 +606,30 @@ static void LoadInternal(ExtensionLoader &loader) {
 		loader.RegisterFunction(pg_error_nested);
 	}
 
+	{
+		ScalarFunction iceberg_size_clamp_text(
+			"iceberg_size_clamp_text",
+			{LogicalType::VARCHAR, LogicalType::INTEGER},
+			LogicalType::VARCHAR,
+			IcebergSizeClampTextFun);
+		loader.RegisterFunction(iceberg_size_clamp_text);
+
+		ScalarFunction iceberg_size_clamp_blob(
+			"iceberg_size_clamp_blob",
+			{LogicalType::BLOB, LogicalType::INTEGER},
+			LogicalType::BLOB,
+			IcebergSizeClampBlobFun);
+		loader.RegisterFunction(iceberg_size_clamp_blob);
+
+		ScalarFunction iceberg_byte_size(
+			"iceberg_byte_size",
+			{LogicalType::ANY},
+			LogicalType::BIGINT,
+			IcebergByteSizeFun,
+			IcebergByteSizeBind);
+		loader.RegisterFunction(iceberg_byte_size);
+	}
+
 	PgLakeUtilityFunctions::RegisterFunctions(loader);
 	PgLakeFileSystemFunctions::RegisterFunctions(loader);
 

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_query_validation.h

@@ -37,6 +37,21 @@ extern PGDLLEXPORT char *IcebergWrapQueryWithErrorOrClampChecks(char *query,
 																IcebergOutOfRangePolicy policy,
 																bool queryHasRowId);
 
+/*
+ * IcebergWrapQueryWithSizeClampChecks wraps a query with size-clamp
+ * expressions for columns whose values may exceed downstream byte caps:
+ * text/varchar/bpchar truncated at a UTF-8 character boundary, bytea
+ * byte-truncated, jsonb/json NULLed when too long, and arrays/structs/
+ * maps NULLed when their leaf-byte sum exceeds
+ * pg_lake_engine.iceberg_max_string_bytes.
+ *
+ * Returns the original query unchanged when both GUCs are zero or no
+ * column carries a clampable type.
+ */
+extern PGDLLEXPORT char *IcebergWrapQueryWithSizeClampChecks(char *query,
+															 TupleDesc tupleDesc,
+															 bool queryHasRowId);
+
 /*
  * IcebergWrapQueryWithNativeTypeConversion wraps a query to rewrite
  * columns whose native DuckDB shape does not match Iceberg's.  See the

pg_lake_engine/include/pg_lake/pgduck/iceberg_validation.h

@@ -72,3 +72,23 @@ 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;
+extern PGDLLEXPORT int IcebergMaxAggregateBytes;
+
+/*
+ * 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,50 @@ _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);
+
+	DefineCustomIntVariable("pg_lake_engine.iceberg_max_aggregate_bytes",
+							gettext_noop("Maximum bytes for the JSON-serialized form of "
+										 "array, composite, and map values written to "
+										 "Iceberg tables. The whole container is replaced "
+										 "with NULL when the sum of its leaf byte sizes "
+										 "exceeds this size. 0 disables the limit. Distinct "
+										 "from iceberg_max_string_bytes because downstream "
+										 "consumers' OBJECT/ARRAY/VARIANT columns typically "
+										 "have a much larger ceiling than STRING/VARCHAR "
+										 "(e.g. on Snowflake: 128 MiB vs. 16 MiB)."),
+							NULL,
+							&IcebergMaxAggregateBytes,
+							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"),