Fix nested timetz reads from Iceberg

[marknefedov]

Summary

Reading a timetz value stored inside a composite, array, or map in an Iceberg table returned a different instant depending on the session timezone: '12:30:00+04' stored as 08:30:00 UTC came back as 08:30:00-04 under America/New_York. Top-level timetz columns were unaffected.

The read projection only promoted top-level TIMETZ back to TIMETZ (and had interval-only helpers for structs and maps), so nested leaves reached PostgreSQL as plain TIME digits and timetz_in attached the session offset.

Changes

• Replace the interval-only BuildStructWithIntervalProjection/BuildMapWithIntervalProjection helpers with one recursive conversion (AppendIcebergReadConversion) that mirrors the write-side AppendRewriteExpression traversal and handles both INTERVAL and TIMETZ at any nesting depth — arrays, composites, maps, and domains — and composes with the compatibility-mode storage casts in BuildStorageToSurfaceProjection.
• Remove the now-redundant inner CAST(... AS TIMETZ) from the write-side AppendTimeTzUtcCast: it existed precisely because nested TIMETZ leaves read back from Iceberg arrived as plain TIME — the case this PR fixes at the source — so every source feeding an Iceberg write now exposes genuine TIMETZ.
• Add regression coverage that reads nested timetz (composite, array-of-composite, map, and a compatibility_mode = 'snowflake' mix with uuid) under a non-UTC session timezone — the condition that masked the bug in the existing tests.

Testing

Built the working tree and ran the timetz, interval, uuid-compat, and reserved-keyword suites in the repo's PG18 development image: 17 passed. The changed engine objects also compile against PG16 and PG17 from the same image.

Fixes #332.

[sfc-gh-dachristensen]

Were things being written out correctly? Like did this result in data drift in updates?

[marknefedov]

Initial writes were always correct. All write paths UTC-normalize from the authoritative PostgreSQL value (row-by-row via TimeTzOutForPGDuck, pushdown via the AT TIME ZONE 'UTC' rewrite), regardless of the session timezone. Pushed-down statements that read Iceberg data and wrote it back were also safe: the inner defensive CAST(... AS TIMETZ) this PR removes made that round-trip a no-op on the stored digits. Compaction rewrites values in their storage shape, so it couldn't drift either.

But yes, updates could drift persisted data. The row-by-row UPDATE path (scan → ExecForeignUpdate → rewrite) pulls the old row up into PostgreSQL through the broken read projection, so under a non-UTC session a nested timetz arrived as the wrong instant, and the write path then faithfully persisted that wrong instant. Empirically, on the base commit an unrelated UPDATE ... SET note = 'after' under America/New_York shifted every nested stored value by +4h (08:30 → 12:30, 01:30 → 05:30, …); the same test on this branch preserves the stored digits exactly. UTC sessions and top-level timetz columns were unaffected.

I've added a regression test that locks this in at the raw-storage level: it writes identical nested values from a UTC and a non-UTC session, verifies the stored digits match, then updates each row from its session and re-reads the new snapshot's raw values (test_nested_timetz_writes_and_updates_are_session_timezone_independent).

[marknefedov]

Found a regression in this PR, AppendIcebergReadConversion unwraps container domains, but ReadEmptyDataSource/ChooseCompatibleDuckDBType does not. An empty table column using a domain over a composite, array, or interval is synthesized as NULL::VARCHAR, the new conversion then applies struct-field access, list_transform, or .months to that VARCHAR, causing a DuckDB binder error.

[marknefedov]

Added tests and rerun full suite, should be fine now.

[sfc-gh-mslot]

The diagnosis is right and the shape of the fix is the right one: replacing the two
hand-rolled special cases (BuildStructWithIntervalProjection,
BuildMapWithIntervalProjection) with a single recursive
AppendIcebergReadConversion that mirrors the write path's AppendRewriteExpression
is a real simplification, and it makes depths that were previously broken or
unreachable work. I verified the underlying bug and the fix by hand:
COPY-ing a composite with a timetz leaf out of Iceberg returns 05:00-04 on
main (wrong instant) and the correct 05:00+00 with this branch.

One blocking regression, then documentation drift, then cleanups.

1. Blocking: COPY ... FROM '<iceberg metadata>' hard-fails on nested timetz

Binder Error: Column "s" referenced that exists in the SELECT clause -
but this column cannot be referenced before it is defined

Reproducer:

CREATE TYPE slot AS (label text, at_time timetz);
CREATE TABLE src (id int, s slot) USING iceberg;
CREATE TABLE dst (id int, s slot) USING iceberg;
INSERT INTO src VALUES (1, ROW('open', '09:00:00+04')::slot);
COPY dst FROM '<src metadata_location>' WITH (FORMAT 'iceberg');  -- errors

Cause: the new generic block in TupleDescToProjectionList inherits the legacy
alias suppression from the code it replaced:

char *columnAliasString =
    !addCast ? psprintf(" AS %s", duckdb_quote_identifier(columnName)) : "";

Under addCast (the COPY FROM pushdown path) the pre-PR expression was
"s"::STRUCT(label VARCHAR, at_time TIME) — a cast of a bare column reference, which
DuckDB implicitly still names s, so the write-side rewrite wrapper could bind it.
The new CASE WHEN s IS NOT NULL THEN struct_pack(...) ELSE NULL END gets an
auto-generated name instead, and the wrapper's reference to s no longer resolves.
The emitted SQL shows it plainly — id carries its alias, the struct column does not:

SELECT id, ... struct_pack(label := s.label, at_time := CAST(...)) AS s
FROM (SELECT id::INTEGER AS id,
             CASE WHEN s IS NOT NULL THEN struct_pack(...) ELSE NULL END
      FROM read_parquet(...))

The !addCast comment on BuildColumnProjection justifies the suppression with "the
caller is going to do its own casting or alias", but in this path nobody does.

This is not purely new: an interval leaf hits the same wall on main today (I
confirmed it), because BuildStructWithIntervalProjection had the identical alias
pattern. So the PR widens a pre-existing bug from interval to any nested
timetz — but since it unifies both into one block, one fix covers both.

Suggested fix — do what BuildStorageToSurfaceProjection already does ~100 lines
above (cast to the surface type, always alias), which also restores the type
enforcement addCast exists to provide:

 			if (nativeProjection != NULL)
 			{
-				char	   *columnAliasString =
-					!addCast ? psprintf(" AS %s", duckdb_quote_identifier(columnName)) : "";
+				const char *castTargetType =
+					GetFullDuckDBTypeNameForPGType(MakePGType(columnTypeId,
+															 columnTypeMod),
+												   DATA_FORMAT_PARQUET);
 
 				if (hasColumns)
 					appendStringInfoString(&projection, ", ");
 
-				appendStringInfo(&projection, "%s%s",
-								 nativeProjection, columnAliasString);
+				appendStringInfo(&projection, "CAST(%s AS %s) AS %s",
+								 nativeProjection, castTargetType,
+								 duckdb_quote_identifier(columnName));
 
 				hasColumns = true;
 				continue;
 			}

With this applied the timetz COPY returns the correct 05:00:00+00, the
pre-existing interval COPY failure is fixed as well, and 134 tests across
test_iceberg_timetz_type, test_iceberg_interval_type, test_iceberg_uuid_compat,
test_domain, test_iceberg_empty_table, test_compatibility_mode,
test_duckdb_reserved_keywords, test_iceberg_types and test_complex_types pass.

Worth a regression test: COPY dst FROM '<iceberg metadata>' WITH (FORMAT 'iceberg')
into a target holding both a composite-with-timetz and a composite-with-interval.

2. The PR description and a test docstring describe code that was reverted

Commit b1dfa54 ("Preserve timetz typing in Iceberg write rewrites") restored
CAST(CAST((%s) AS TIMETZ) AT TIME ZONE 'UTC' AS TIME) on the write side, but:

• the PR description still says the write path drops the inner CAST(... AS TIMETZ);
• _assert_timetz_utc_cast_in_explain's docstring (rewritten in 9ec998d) still
  claims the wrapper emits CAST((<expr>) AT TIME ZONE 'UTC' AS TIME).

The assertion itself only greps for AT TIME ZONE 'UTC' AS TIME, so it passes under
either form and doesn't catch the drift. Please re-sync both before merge — the
description is what reviewers and git log will read later.

3. No new interval test, though the interval read path was rewritten wholesale

Interval no longer goes through BuildStructWithIntervalProjection /
BuildMapWithIntervalProjection (both deleted) but through the new recursion, and all
519 new test lines are timetz. test_iceberg_interval_type.py has no
array-of-composite-with-interval and no map-with-composite-value case — exactly the
depths the deleted code got wrong (it emitted {field: "col"."field"} against a
list). I checked those depths do work now, so this is about locking in the newly
gained coverage rather than a suspected defect.

4. GetEmptyIcebergInputTypeName does discarded work at every node

DuckDBTypeInfo duckdbType = ChooseCompatibleDuckDBType(typeOid, typmod, DATA_FORMAT_ICEBERG, false);
DuckDBTypeInfo storageType = GuessStorageType(duckdbType, DATA_FORMAT_ICEBERG);

if (!TypeNeedsIcebergReadConversion(typeOid))
    return storageType.typeName;

Both calls render the full nested type name and are thrown away for every node that
does need conversion; they belong below the early-out. Related:
AppendIcebergReadConversion re-runs TypeNeedsIcebergReadConversion — itself a full
subtree walk — at each level, so a wide composite costs O(depth x subtree) syscache
lookups per planned query. Harmless at depth 2, worth a thought.

5. QuoteDuckDBStructKeySQL is now dead code

It was introduced by #297 specifically for the struct-literal projection builder this
PR deletes, and has no callers left. Switching to duckdb_quote_identifier is the
correct call, since struct_pack(name := ...) takes identifiers rather than string
literals — I verified a composite with field names containing '', "" and the
DuckDB reserved word pivot round-trips correctly. Please drop the function, its
declaration in struct_conversion.h, and the cross-reference in
QuoteDuckDBStructKey's comment, so the next reader isn't sent to a dead helper.

6. Nits

• duckdb_quote_identifier((char *) columnName) (4 sites) — the parameter is already
  const char *; the mirrored write-side code omits the cast.
• Assert(converted); (void) converted; three times is noisier than the write side,
  which just ignores the return value.
• The new ReadEmptyDataSource comment about preferVarchar describes an unreachable
  combination: preferVarchar is only set for CSV/JSON reads in copy.c, and
  compaction reads use DATA_FORMAT_PARQUET, so it is never true alongside
  DATA_FORMAT_ICEBERG.

Things I checked that are fine

• Lambda variable naming (_x%d keyed on depth) cannot shadow: the composite
  branch is the only one that keeps depth, and it never emits a lambda.
• Map detection correctly precedes the generic domain unwrap — pg_map is a domain
  over an array of key/value composites, and unwrapping first would lose the MAP shape.
• The CASE WHEN ... IS NOT NULL guard around struct_pack is necessary (struct_pack
  resets validity) and mirrors the write side.
• Unbounded mutual recursion isn't reachable: PostgreSQL rejects recursive composites
  ("composite type node cannot be made a member of itself").
• Rebases cleanly onto current main (18 commits, including "Support iceberg schema
  type change" #223); neither new commit touches the projection code.
• Builds warning-free under the repo's flags.

Out of scope, but found while probing

• A timetz MAP key can't be looked up after a round-trip:
  '09:00:00+04' comes back as '05:00:00+00', and PostgreSQL's timetz = is
  offset-sensitive ('09:00+04' = '05:00+00' is FALSE), so
  map_type.extract(m, '09:00:00+04'::timetz) returns NULL. Inherent to Iceberg's
  UTC-normalized TIMETZ storage and true for top-level timetz too, so it predates
  this PR — but nested keys make it easier to hit.
• Two pre-existing write-path failures, unrelated to this PR: INSERT into an
  Iceberg table fails for a domain-over-array column
  (columns={'v':'LIST'} in the CSV reader) and for a composite that has had an
  attribute dropped (field_ids still contains ........pg.dropped.1........).
  Notably GetEmptyIcebergInputTypeName in this PR does skip attisdropped.

(removing changes requested status, since I'm traveling)