iceberg: address PR #4427 follow-up review

Three review comments on PR #4427 that landed after merge:

- shredder: reject the schema/column mismatch where a Timestamp common
  reaches an Int32 column. coerceTemporalToNumeric returns int64
  UnixMilli/Micros/Nanos (~10^12), which the Int32Type arm cast to
  int32 with no bounds check, silently truncating into a garbage year.
  The arm is intended for Date / TimeOfDay coercions whose values do
  fit; bound-check post-coerce and fail loudly when they don't, with a
  message pointing the operator at BIGINT or a schema-metadata fix.

- output_iceberg: drop the redundant conf.Contains guard around
  require_schema_metadata parsing. The field declares Default(false)
  in the spec so FieldBool returns false-without-error on absence and
  the inline form fits cleanly.

- shredder: expose StrictTemporalMode as an exported struct field
  instead of the SetStrictTemporalMode setter. It's a single bool with
  no validation hook, so the setter added ceremony without value.
  Updates the sole writer.go caller to assign the field directly.

New regression test TestCoerceTemporalInt32OverflowGuard pins the
Int32 overflow rejection and confirms in-range Date / TimeOfDay
coercions still succeed.

Author: Jeffail

internal/impl/iceberg/output_iceberg.go

@@ -484,15 +484,14 @@ func parseSchemaEvolutionConfig(conf *service.ParsedConfig) (SchemaEvolutionConf
 		}
 	}
 
-	// Parse require_schema_metadata
-	if conf.Contains(ioFieldSchemaEvolution, ioFieldSchemaEvolutionRequireSchemaMetadata) {
-		cfg.RequireSchemaMetadata, err = conf.FieldBool(ioFieldSchemaEvolution, ioFieldSchemaEvolutionRequireSchemaMetadata)
-		if err != nil {
-			return cfg, err
-		}
-		if cfg.RequireSchemaMetadata && cfg.SchemaMetadata == "" {
-			return cfg, fmt.Errorf("%s.%s requires %s.%s to be set", ioFieldSchemaEvolution, ioFieldSchemaEvolutionRequireSchemaMetadata, ioFieldSchemaEvolution, ioFieldSchemaEvolutionSchemaMetadata)
-		}
+	// Parse require_schema_metadata. The field carries Default(false) in the
+	// spec so FieldBool returns false-without-error when absent, no Contains
+	// guard required.
+	if cfg.RequireSchemaMetadata, err = conf.FieldBool(ioFieldSchemaEvolution, ioFieldSchemaEvolutionRequireSchemaMetadata); err != nil {
+		return cfg, err
+	}
+	if cfg.RequireSchemaMetadata && cfg.SchemaMetadata == "" {
+		return cfg, fmt.Errorf("%s.%s requires %s.%s to be set", ioFieldSchemaEvolution, ioFieldSchemaEvolutionRequireSchemaMetadata, ioFieldSchemaEvolution, ioFieldSchemaEvolutionSchemaMetadata)
 	}
 
 	return cfg, nil

internal/impl/iceberg/shredder/shredder.go

@@ -87,13 +87,20 @@ type RecordShredder struct {
 	// typed columns instead of guessing. nil entries fall back to the
 	// pre-schema-metadata behavior — see [convertLeafValue].
 	fieldCommons map[int]*schema.Common
-	// strictTemporal causes [convertLeafValue] to refuse numeric inputs
-	// into time-typed columns when no schema metadata has been
-	// registered for that column. When false (the default), the value
-	// converter falls back to [bloblang.ValueAsTimestamp]'s seconds
-	// default — convenient but silently wrong if the upstream produced
-	// a different unit. When true, the writer fails the batch loudly.
-	strictTemporal bool
+	// StrictTemporalMode causes [convertLeafValue] to refuse numeric
+	// inputs into time-typed columns (TIMESTAMP / TIMESTAMPTZ / DATE /
+	// TIME) when no schema metadata has been registered for that
+	// column. When false (the default), the value converter falls back
+	// to [bloblang.ValueAsTimestamp]'s seconds default — convenient but
+	// silently wrong if the upstream produced a different unit. When
+	// true, the writer fails the batch loudly.
+	//
+	// No effect on time.Time / time.Duration values, which carry their
+	// own unit unambiguously, and no effect on non-time columns. Set
+	// directly on the shredder after construction; operators that
+	// cannot guarantee schema metadata flows end-to-end flip this on to
+	// fail loudly instead of silently corrupting dates by ~50,000 years.
+	StrictTemporalMode bool
 }
 
 // NewRecordShredder creates a new shredder for the given schema.
@@ -107,22 +114,6 @@ func NewRecordShredder(schema *iceberg.Schema, caseSensitive bool) *RecordShredd
 	}
 }
 
-// SetStrictTemporalMode toggles whether numeric inputs into time-typed
-// columns require registered schema metadata. With strict mode on, a bare
-// int64 / float64 value reaching a TIMESTAMP / TIMESTAMPTZ / DATE / TIME
-// column with no [schema.Common] in the field map is rejected with a
-// per-field error rather than guessed-as-Unix-seconds.
-//
-// Strict mode has no effect on time.Time / time.Duration values, which
-// carry their own unit unambiguously, and no effect on non-time columns.
-//
-// Defaults to off (back-compat). Operators that cannot guarantee schema
-// metadata flows end-to-end can flip this on to fail loudly instead of
-// silently corrupting dates by ~50,000 years.
-func (rs *RecordShredder) SetStrictTemporalMode(on bool) {
-	rs.strictTemporal = on
-}
-
 // SetFieldSchemaMetadata supplies a field-ID → schema.Common map that the
 // leaf value converter consults when it sees a numeric input destined for a
 // time-typed Iceberg column (TIMESTAMP, TIMESTAMPTZ, TIMESTAMP_NS,
@@ -264,7 +255,7 @@ func (rs *RecordShredder) shredValue(
 
 	default:
 		// Leaf/primitive type.
-		pqVal, err := convertLeafValue(value, typ, rs.commonForField(fieldID), rs.strictTemporal)
+		pqVal, err := convertLeafValue(value, typ, rs.commonForField(fieldID), rs.StrictTemporalMode)
 		if err != nil {
 			return err
 		}
@@ -478,6 +469,14 @@ func convertLeafValue(value any, typ iceberg.Type, common *schema.Common, strict
 			if strictTemporal {
 				return parquet.NullValue(), fmt.Errorf("int column received %T while schema metadata declares type %v; require_schema_metadata=true demands the existing column type match the schema metadata — recreate the table to migrate", value, common.Type)
 			}
+			// A Timestamp common's UnixMilli/Micros/Nanos far exceeds the
+			// int32 range, but the Int32Type arm is intended for Date /
+			// TimeOfDay coercions whose values do fit. Reject the
+			// schema/column mismatch loudly rather than silently
+			// truncating into a garbage year.
+			if n > math.MaxInt32 || n < math.MinInt32 {
+				return parquet.NullValue(), fmt.Errorf("int column received %T with schema metadata type %v; coerced value %d overflows int32 — the column should be BIGINT or the schema metadata is wrong", value, common.Type, n)
+			}
 			return parquet.Int32Value(int32(n)), nil
 		}
 		i, err := bloblang.ValueAsInt64(value)

internal/impl/iceberg/shredder/temporal_test.go

@@ -498,6 +498,50 @@ func TestCoerceTemporalIntoNumericColumn(t *testing.T) {
 	})
 }
 
+// TestCoerceTemporalInt32OverflowGuard covers the schema/column mismatch
+// where a TIMESTAMP schema common reaches an Int32 column — coerceTemporal-
+// ToNumeric returns a UnixMilli value (~10^12) that vastly exceeds int32
+// range. The Int32 arm is intended for Date / TimeOfDay coercions whose
+// values fit in int32; a Timestamp value silently truncating into a garbage
+// year is the failure mode this guard exists to prevent.
+//
+// The complementary Int64 arm has no bounds problem (Timestamp values fit
+// comfortably in int64) and is verified separately by
+// TestCoerceTemporalIntoNumericColumn.
+func TestCoerceTemporalInt32OverflowGuard(t *testing.T) {
+	const tsMillis = int64(1_700_000_000_000) // 2023-11-14, ~10^12, well beyond int32
+
+	t.Run("Timestamp(Millis) into Int32 errors with overflow message", func(t *testing.T) {
+		v := time.UnixMilli(tsMillis).UTC()
+		_, err := convertLeafValue(v, iceberg.Int32Type{}, tsCommon(schema.TimeUnitMillis, true), false)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "overflows int32", "must reject the schema/column mismatch loudly, not truncate to a garbage year")
+	})
+
+	t.Run("Timestamp(Micros) into Int32 errors with overflow message", func(t *testing.T) {
+		v := time.UnixMilli(tsMillis).UTC()
+		_, err := convertLeafValue(v, iceberg.Int32Type{}, tsCommon(schema.TimeUnitMicros, true), false)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "overflows int32")
+	})
+
+	// Sanity check: the in-range coercions the Int32 arm was designed for
+	// still succeed.
+	t.Run("Date into Int32 still succeeds", func(t *testing.T) {
+		v := time.Date(2024, 1, 15, 0, 0, 0, 0, time.UTC)
+		pq, err := convertLeafValue(v, iceberg.Int32Type{}, &schema.Common{Type: schema.Date}, false)
+		require.NoError(t, err)
+		assert.Equal(t, int32(19737), pq.Int32())
+	})
+
+	t.Run("TimeOfDay into Int32 still succeeds", func(t *testing.T) {
+		d := 8*time.Hour + 30*time.Minute
+		pq, err := convertLeafValue(d, iceberg.Int32Type{}, todCommon(schema.TimeUnitMillis), false)
+		require.NoError(t, err)
+		assert.Equal(t, int32(8*3600+30*60)*1000, pq.Int32())
+	})
+}
+
 // TestCoerceTemporalRejectedInStrictMode confirms that the temporal->numeric
 // coerce path is disabled when require_schema_metadata=true. In strict mode
 // a type disagreement between the existing column and the schema metadata

internal/impl/iceberg/writer.go

@@ -55,7 +55,7 @@ type writer struct {
 // resolver supplies optional per-message schema metadata used by the shredder
 // to interpret numeric inputs into time-typed columns; pass nil to disable.
 // requireSchemaMetadata enables shredder strict mode — see
-// [shredder.RecordShredder.SetStrictTemporalMode].
+// [shredder.RecordShredder.StrictTemporalMode].
 func NewWriter(tbl *table.Table, comm *committer, caseSensitive bool, writerOpts []parquet.WriterOption, resolver *typeResolver, requireSchemaMetadata bool, logger *service.Logger) *writer {
 	return &writer{
 		table:                 tbl,
@@ -222,9 +222,7 @@ func (w *writer) messagesToParquet(batch service.MessageBatch) ([]partitionFile,
 	// silently for messages 1..N; in that case the writer must be
 	// extended to per-message metadata lookup with a small cache.
 	rs := shredder.NewRecordShredder(schema, w.caseSensitive)
-	if w.requireSchemaMetadata {
-		rs.SetStrictTemporalMode(true)
-	}
+	rs.StrictTemporalMode = w.requireSchemaMetadata
 	if w.resolver != nil && len(batch) > 0 {
 		if common, err := w.resolver.parseSchemaMetadata(batch[0]); err != nil {
 			w.logger.Warnf("parsing schema metadata for shredder: %v (falling back to schema-agnostic conversion)", err)