Fix NOT NULL bypass when clamp mode converts values to NULL (#316)

ExecConstraints was running before IcebergErrorOrClampSlotInPlace, so
values like bounded numeric NaN and multidimensional arrays passed the
NOT NULL check in their original (non-null) form, then got clamped to
NULL and silently stored in NOT NULL columns.

Move clamping before ExecConstraints in both postgresExecForeignInsert
and postgresExecForeignUpdate so constraint checks see post-clamp values.

Factor the clamping + ExecConstraints sequence into a shared
ClampAndCheckConstraints helper used by both INSERT and UPDATE paths.

---------

Signed-off-by: sfc-gh-npuka <naisila.puka@snowflake.com>

Author: sfc-gh-npuka

pg_lake_table/src/fdw/pg_lake_table.c

@@ -502,6 +502,11 @@ static void merge_fdw_options(PgLakeRelationInfo * fpinfo,
 							  const PgLakeRelationInfo * fpinfo_o,
 							  const PgLakeRelationInfo * fpinfo_i);
 
+static void IcebergErrorOrClampSlotInPlace(TupleTableSlot *slot, TupleDesc tupleDesc,
+										   IcebergOutOfRangePolicy policy);
+static void ClampAndCheckConstraints(PgLakeModifyState * fmstate,
+									 ResultRelInfo *resultRelInfo,
+									 TupleTableSlot *slot, EState *estate);
 static void WriteInsertRecord(PgLakeModifyState * modifyState, TupleTableSlot *slot);
 static void PrepareDeletionSlot(PgLakeFileModifyState * fileModifyState,
 								uint64 fileRowNumber,
@@ -2220,14 +2225,7 @@ postgresExecForeignInsert(EState *estate,
 {
 	PgLakeModifyState *fmstate = (PgLakeModifyState *) resultRelInfo->ri_FdwState;
 
-	/*
-	 * Constraint checks are skipped by PostgreSQL itself, since it assumes
-	 * them to be unenforceable as data can change underneath.
-	 */
-	Relation	resultRelationDesc = resultRelInfo->ri_RelationDesc;
-
-	if (resultRelationDesc->rd_att->constr)
-		ExecConstraints(resultRelInfo, slot, estate);
+	ClampAndCheckConstraints(fmstate, resultRelInfo, slot, estate);
 
 	WriteInsertRecord(fmstate, slot);
 
@@ -2268,12 +2266,7 @@ postgresExecForeignUpdate(EState *estate,
 		return NULL;
 	}
 
-	/*
-	 * Constraint checks are skipped by PostgreSQL itself, since it assumes
-	 * them to be unenforceable as data can change underneath.
-	 */
-	if (resultRelation->rd_att->constr)
-		ExecConstraints(resultRelInfo, slot, estate);
+	ClampAndCheckConstraints(fmstate, resultRelInfo, slot, estate);
 
 	/*
 	 * Also check the tuple against the partition constraint, since PostgreSQL
@@ -2643,19 +2636,40 @@ IcebergErrorOrClampSlotInPlace(TupleTableSlot *slot, TupleDesc tupleDesc,
 
 
 /*
- * WriteInsertRecord validates the tuple against Iceberg write constraints
- * (when applicable) and forwards it to the insert destination.
+ * ClampAndCheckConstraints normalizes the slot for Iceberg write and then
+ * runs PostgreSQL constraint checks (NOT NULL, CHECK, etc.).
+ *
+ * Clamping must happen first so that ExecConstraints sees post-clamp values,
+ * e.g. bounded numeric NaN clamped to NULL is caught by NOT NULL.
+ *
+ * PostgreSQL skips constraint checks on foreign tables, so we run them
+ * ourselves.
+ */
+static void
+ClampAndCheckConstraints(PgLakeModifyState * fmstate,
+						 ResultRelInfo *resultRelInfo,
+						 TupleTableSlot *slot, EState *estate)
+{
+	if (fmstate->outOfRangePolicy != ICEBERG_OOR_NONE &&
+		fmstate->needsOutOfRangeValidation)
+		IcebergErrorOrClampSlotInPlace(slot, fmstate->tupleDesc,
+									   fmstate->outOfRangePolicy);
+
+	Relation	rel = resultRelInfo->ri_RelationDesc;
+
+	if (rel->rd_att->constr)
+		ExecConstraints(resultRelInfo, slot, estate);
+}
+
+
+/*
+ * WriteInsertRecord forwards the tuple to the insert destination.
  */
 static void
 WriteInsertRecord(PgLakeModifyState * modifyState, TupleTableSlot *slot)
 {
 	DestReceiver *insertDest = modifyState->insertDest;
 
-	if (modifyState->outOfRangePolicy != ICEBERG_OOR_NONE &&
-		modifyState->needsOutOfRangeValidation)
-		IcebergErrorOrClampSlotInPlace(slot, modifyState->tupleDesc,
-									   modifyState->outOfRangePolicy);
-
 	if (modifyState->insertedRowCount == 0)
 	{
 		/* incoming inserts have the tuple descriptor of the table */

pg_lake_table/tests/pytests/test_iceberg_validation.py

@@ -2844,3 +2844,241 @@ def test_explain_shows_pg_nullify_nested_list_in_list_transform(
         run_command("RESET search_path;", pg_conn)
         run_command(f"DROP SCHEMA IF EXISTS {schema} CASCADE;", pg_conn)
         pg_conn.commit()
+
+
+# =====================================================================
+# NOT NULL constraint enforcement after clamping
+#
+# When out_of_range_values = 'clamp', values like bounded numeric NaN
+# and multidimensional arrays are clamped to NULL.  The NOT NULL
+# constraint must still be enforced on the post-clamp value.
+# =====================================================================
+
+
+def test_clamp_to_null_enforces_not_null(pg_conn, extension, s3, with_default_location):
+    """NOT NULL constraint is enforced on values clamped to NULL.
+
+    Covers both value types that clamp to NULL (bounded numeric NaN and
+    multidimensional arrays) and both DML paths (INSERT and UPDATE).
+
+    +-------+---------------------+--------+----------------------------+
+    | case  | value               | op     | expected                   |
+    +-------+---------------------+--------+----------------------------+
+    | NaN   | 'NaN'::numeric      | INSERT | not-null constraint error  |
+    | NaN   | 'NaN'::numeric      | UPDATE | not-null constraint error  |
+    | mdim  | ARRAY[ARRAY[1,2,3]] | INSERT | not-null constraint error  |
+    | mdim  | ARRAY[ARRAY[1,2,3]] | UPDATE | not-null constraint error  |
+    +-------+---------------------+--------+----------------------------+
+    """
+    schema = "test_clamp_null_nn"
+
+    run_command(f"CREATE SCHEMA {schema};", pg_conn)
+    run_command(f"SET search_path TO {schema};", pg_conn)
+
+    try:
+        run_command(
+            "CREATE TABLE num_target ("
+            "  id int,"
+            "  n numeric(18,6) NOT NULL"
+            ") USING iceberg WITH (out_of_range_values = 'clamp');",
+            pg_conn,
+        )
+        run_command(
+            "CREATE TABLE arr_target ("
+            "  id int,"
+            "  vals int[] NOT NULL"
+            ") USING iceberg WITH (out_of_range_values = 'clamp');",
+            pg_conn,
+        )
+        pg_conn.commit()
+
+        # -- NaN INSERT into NOT NULL numeric --
+        err = run_command(
+            "INSERT INTO num_target VALUES (1, 'NaN'::numeric(18,6));",
+            pg_conn,
+            raise_error=False,
+        )
+        assert "not-null constraint" in str(err)
+        pg_conn.rollback()
+
+        # -- NaN UPDATE on NOT NULL numeric --
+        run_command("INSERT INTO num_target VALUES (1, 42.0);", pg_conn)
+        pg_conn.commit()
+
+        err = run_command(
+            "UPDATE num_target SET n = 'NaN'::numeric(18,6) WHERE id = 1;",
+            pg_conn,
+            raise_error=False,
+        )
+        assert "not-null constraint" in str(err)
+        pg_conn.rollback()
+
+        result = run_query("SELECT n FROM num_target WHERE id = 1;", pg_conn)
+        assert result[0][0] is not None
+
+        # -- Multidimensional array INSERT into NOT NULL column --
+        err = run_command(
+            "INSERT INTO arr_target VALUES (1, ARRAY[ARRAY[1,2,3]]);",
+            pg_conn,
+            raise_error=False,
+        )
+        assert "not-null constraint" in str(err)
+        pg_conn.rollback()
+
+        # -- Multidimensional array UPDATE on NOT NULL column --
+        run_command("INSERT INTO arr_target VALUES (1, ARRAY[10,20,30]);", pg_conn)
+        pg_conn.commit()
+
+        err = run_command(
+            "UPDATE arr_target SET vals = ARRAY[ARRAY[1,2],ARRAY[3,4]] WHERE id = 1;",
+            pg_conn,
+            raise_error=False,
+        )
+        assert "not-null constraint" in str(err)
+        pg_conn.rollback()
+
+        result = run_query("SELECT vals FROM arr_target WHERE id = 1;", pg_conn)
+        assert result[0][0] == [10, 20, 30]
+    finally:
+        pg_conn.rollback()
+        run_command("RESET search_path;", pg_conn)
+        run_command(f"DROP SCHEMA IF EXISTS {schema} CASCADE;", pg_conn)
+        pg_conn.commit()
+
+
+# =====================================================================
+# CHECK constraint interaction with clamping
+#
+# When clamping runs before constraint checks, CHECK constraints
+# evaluate the post-clamp value.  This means clamping can "rescue"
+# values that would otherwise violate a CHECK.  These tests document
+# the expected behavior: the CHECK validates what is actually stored.
+# =====================================================================
+
+
+def test_temporal_clamp_rescues_check_constraint(
+    pg_conn, extension, s3, with_default_location
+):
+    """Temporal clamping can satisfy a CHECK that the original value would violate.
+
+    Year 10000 violates CHECK (d <= '9999-12-31'), but clamping brings it
+    to exactly 9999-12-31, which satisfies the CHECK.
+    """
+    schema = "test_clamp_check_temporal"
+
+    run_command(f"CREATE SCHEMA {schema};", pg_conn)
+    run_command(f"SET search_path TO {schema};", pg_conn)
+    run_command("SET TIME ZONE 'UTC';", pg_conn)
+
+    try:
+        run_command(
+            "CREATE TABLE target (d date CHECK (d <= '9999-12-31'::date))"
+            " USING iceberg WITH (out_of_range_values = 'clamp');",
+            pg_conn,
+        )
+        pg_conn.commit()
+
+        run_command(
+            "INSERT INTO target VALUES ('infinity'::date);",
+            pg_conn,
+        )
+        pg_conn.commit()
+
+        result = run_query("SELECT d::text FROM target;", pg_conn)
+        assert result[0][0] == "9999-12-31"
+    finally:
+        pg_conn.rollback()
+        run_command("RESET TIME ZONE;", pg_conn)
+        run_command("RESET search_path;", pg_conn)
+        run_command(f"DROP SCHEMA IF EXISTS {schema} CASCADE;", pg_conn)
+        pg_conn.commit()
+
+
+def test_temporal_clamp_still_fails_strict_check_constraint(
+    pg_conn, extension, s3, with_default_location
+):
+    """Temporal clamping cannot rescue a CHECK that the clamped value still violates.
+
+    infinity is clamped to 9999-12-31, but CHECK (d < '9999-12-31') uses
+    strict less-than, so the clamped value still fails.
+    """
+    schema = "test_clamp_check_strict"
+
+    run_command(f"CREATE SCHEMA {schema};", pg_conn)
+    run_command(f"SET search_path TO {schema};", pg_conn)
+    run_command("SET TIME ZONE 'UTC';", pg_conn)
+
+    try:
+        run_command(
+            "CREATE TABLE target (d date CHECK (d < '9999-12-31'::date))"
+            " USING iceberg WITH (out_of_range_values = 'clamp');",
+            pg_conn,
+        )
+        pg_conn.commit()
+
+        err = run_command(
+            "INSERT INTO target VALUES ('infinity'::date);",
+            pg_conn,
+            raise_error=False,
+        )
+        assert "check constraint" in str(err).lower()
+        pg_conn.rollback()
+    finally:
+        pg_conn.rollback()
+        run_command("RESET TIME ZONE;", pg_conn)
+        run_command("RESET search_path;", pg_conn)
+        run_command(f"DROP SCHEMA IF EXISTS {schema} CASCADE;", pg_conn)
+        pg_conn.commit()
+
+
+def test_numeric_nan_clamp_succeeds_on_nullable_column(
+    pg_conn, extension, s3, with_default_location
+):
+    """Bounded numeric NaN is clamped to NULL and stored when the column is nullable.
+
+    This is the happy-path counterpart of the NOT NULL tests: clamping
+    converts NaN to NULL, which is valid for a nullable column.  Covers
+    both INSERT and UPDATE.
+    """
+    schema = "test_clamp_nan_nullable"
+
+    run_command(f"CREATE SCHEMA {schema};", pg_conn)
+    run_command(f"SET search_path TO {schema};", pg_conn)
+
+    try:
+        run_command(
+            "CREATE TABLE target ("
+            "  id int,"
+            "  n numeric(18,6)"
+            ") USING iceberg WITH (out_of_range_values = 'clamp');",
+            pg_conn,
+        )
+        pg_conn.commit()
+
+        # INSERT: NaN is clamped to NULL and succeeds
+        run_command(
+            "INSERT INTO target VALUES (1, 'NaN'::numeric(18,6));",
+            pg_conn,
+        )
+        pg_conn.commit()
+
+        result = run_query("SELECT n FROM target WHERE id = 1;", pg_conn)
+        assert result[0][0] is None
+
+        # UPDATE: NaN is clamped to NULL and succeeds
+        run_command("INSERT INTO target VALUES (2, 99.0);", pg_conn)
+        pg_conn.commit()
+
+        run_command(
+            "UPDATE target SET n = 'NaN'::numeric(18,6) WHERE id = 2;",
+            pg_conn,
+        )
+        pg_conn.commit()
+
+        result = run_query("SELECT n FROM target WHERE id = 2;", pg_conn)
+        assert result[0][0] is None
+    finally:
+        pg_conn.rollback()
+        run_command("RESET search_path;", pg_conn)
+        run_command(f"DROP SCHEMA IF EXISTS {schema} CASCADE;", pg_conn)
+        pg_conn.commit()