diff --git a/CHANGES.md b/CHANGES.md
index 365605c7065b..f0d5d06b9d9f 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -76,6 +76,9 @@
 * (Java/Python) `Watch` can bound its deduplication state by event time, retiring an output key once the greatest emitted timestamp has moved more than the allowed lateness past it. Java adds `Watch.growthOf(...).withTimestampCursor()`. Python adds `allowed_lateness` for the existing `timestamp_cursor` option ([#18459](https://github.com/apache/beam/issues/18459)).
 * (Java) Spark Structured Streaming runner: stateful ParDo with state, timers, `@RequiresTimeSortedInput` and tagged outputs is now supported in batch mode ([#39779](https://github.com/apache/beam/issues/39779)).
 * (Python) Added support for Vertex AI Model Monitoring V2 in RunInference ([#39738](https://github.com/apache/beam/issues/39738)).
+* [Flink Runner] Added opt-in static round-robin split assignment for small bounded sources via the new `sourceStaticSplitThresholdMb` pipeline option. The default of 0 keeps the existing lazy pull-based assignment ([#39873](https://github.com/apache/beam/issues/39873)).
+* Added automatic caching of bounded, single-pane side-input views for classic Java Flink DataStream execution ([#39866](https://github.com/apache/beam/issues/39866)).
+* (Python) Added `Sample.Any`, the Python equivalent of Java's `Sample.any`, which returns up to n arbitrary elements from a PCollection ([#18552](https://github.com/apache/beam/issues/18552)).
 
 ## Breaking Changes
 
@@ -89,7 +92,6 @@
 
 ## Bugfixes
 
-* (Java) Restored binary compatibility for `CoderTranslatorRegistrar` implementations compiled against Beam 2.76 and earlier ([#38714](https://github.com/apache/beam/issues/38714)).
 * (Java) Fixed the Spark runner firing processing-time timers in reverse timestamp order ([#39824](https://github.com/apache/beam/issues/39824)).
 * (Python) Fixed incorrect profiler options handling on portable runners ([#39613](https://github.com/apache/beam/issues/39613)).
 * (Java) KafkaIO dynamic reads no longer require the obsolete `beam_fn_api` experiment ([#29998](https://github.com/apache/beam/issues/29998)).
@@ -124,8 +126,6 @@
 
 ## New Features / Improvements
 
-* [Flink Runner] Added opt-in static round-robin split assignment for small bounded sources via the new `sourceStaticSplitThresholdMb` pipeline option. The default of 0 keeps the existing lazy pull-based assignment ([#39873](https://github.com/apache/beam/issues/39873)).
-* Added automatic caching of bounded, single-pane side-input views for classic Java Flink DataStream execution ([#39866](https://github.com/apache/beam/issues/39866)).
 * Added `GroupIntoBatches` transform and the standard
   `beam:coder:sharded_key:v1` coder to the Go SDK, along with
   `beam.Coder.IsDeterministic`, `beam.PCollection.WindowingStrategy`,
@@ -146,7 +146,6 @@
 * (Python) Added `Watch`, a transform that polls a growing set of outputs for each input element, deduplicates outputs across poll rounds, and stops per a user-supplied termination condition
   ([#21521](https://github.com/apache/beam/issues/21521)).
 * (Python) Added support to analyze core dumps created after python worker segmentation faults with `pystack` (or `gdb` if installed) using the `--profiler_agent=coredump` pipeline option. ([#39484](https://github.com/apache/beam/issues/39484)).
-* (Python) Added `Sample.Any`, the Python equivalent of Java's `Sample.any`, which returns up to n arbitrary elements from a PCollection ([#18552](https://github.com/apache/beam/issues/18552)).
 
 ## Breaking Changes
 
@@ -161,9 +160,6 @@
   Use pipeline option `--updateCompatibilityVersion=2.75.0` (or any older version) to keep the old behavior ([#39344](https://github.com/apache/beam/issues/39344)).
 * `DoFn.process` returning a `str`, `bytes`, or `dict` (instead of an iterable wrapping one) now raises a `TypeError` rather than silently iterating per-character/byte/key (Python) ([#18712](https://github.com/apache/beam/issues/18712)).
 * (Java) Added `DRAINING` and `DRAINED` states to `PipelineResult`, including runner state mappings and Dataflow update handling ([#39020](https://github.com/apache/beam/issues/39020)).
-* (Python) Typehints of dataclass fields are honored during type inferences. To restore the behavior of fallback-to-any,
-  use pipeline option `--exclude_infer_dataclass_field_type` ([#38797](https://github.com/apache/beam/issues/38797)).
-  However fixing forward is recommended.
 * (Java) IcebergIO and projects that use it must now be built with Java 17 or later as a result of Iceberg 1.11.0 upgrade ([#38925](https://github.com/apache/beam/issues/38925)).
 
 ## Bugfixes
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/MetricsAccumulator.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/MetricsAccumulator.java
index dbdfe11a585c..612d71b1aea1 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/MetricsAccumulator.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/MetricsAccumulator.java
@@ -17,6 +17,7 @@
  */
 package org.apache.beam.runners.spark.metrics;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.io.IOException;
 import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
 import org.apache.beam.runners.spark.SparkPipelineOptions;
@@ -82,6 +83,10 @@ public static void init(SparkPipelineOptions opts, JavaSparkContext jsc) {
     }
   }
 
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Spark merges only the accumulator instance the driver registered. A copy would collect metrics that nothing reports.")
   public static MetricsContainerStepMapAccumulator getInstance() {
     if (instance == null) {
       throw new IllegalStateException("Metrics accumulator has not been instantiated");
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java
index 63407b9f14d8..e8cbd895082a 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/metrics/MetricsAccumulator.java
@@ -17,6 +17,7 @@
  */
 package org.apache.beam.runners.spark.structuredstreaming.metrics;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import org.apache.beam.runners.core.metrics.MetricsContainerStepMap;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.spark.sql.SparkSession;
@@ -85,6 +86,10 @@ public MetricsContainerStepMap value() {
    * Get the {@link MetricsAccumulator} on this driver. If there's no such accumulator yet, it will
    * be created and registered using the provided {@link SparkSession}.
    */
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Spark merges only the accumulator instance the driver registered. A copy would collect metrics that nothing reports.")
   public static MetricsAccumulator getInstance(SparkSession session) {
     MetricsAccumulator current = instance;
     if (current != null) {
diff --git a/sdks/java/build-tools/src/main/resources/beam/spotbugs-filter.xml b/sdks/java/build-tools/src/main/resources/beam/spotbugs-filter.xml
index 4393ec6a624e..5f6f368228e5 100644
--- a/sdks/java/build-tools/src/main/resources/beam/spotbugs-filter.xml
+++ b/sdks/java/build-tools/src/main/resources/beam/spotbugs-filter.xml
@@ -57,7 +57,6 @@
 
   <!-- TODO(https://github.com/apache/beam/issues/35312) resolve findings-->
   <Bug pattern="CT_CONSTRUCTOR_THROW"/>
-  <Bug pattern="MS_EXPOSE_REP"/>
 
   <!--
     Many test classes are captured by lambdas and marked `implements Serializable`. They are not
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java
index 2cf79a723d63..c4f8e34d61ea 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java
@@ -52,6 +52,7 @@
 import org.apache.beam.sdk.values.TimestampedValue;
 import org.apache.beam.sdk.values.TypeDescriptor;
 import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.beam.sdk.values.ValueKind;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.HashMultimap;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
@@ -137,6 +138,8 @@ private CommonTypes() {
           TimestampedValue.class,
           CoderProviders.fromStaticMethods(
               TimestampedValue.class, TimestampedValue.TimestampedValueCoder.class));
+      builder.put(
+          ValueKind.class, CoderProviders.fromStaticMethods(ValueKind.class, ValueKindCoder.class));
       builder.put(Void.class, CoderProviders.fromStaticMethods(Void.class, VoidCoder.class));
       builder.put(
           byte[].class, CoderProviders.fromStaticMethods(byte[].class, ByteArrayCoder.class));
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/ValueKindCoder.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/ValueKindCoder.java
new file mode 100644
index 000000000000..24f1a119f429
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/ValueKindCoder.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.coders;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi.Elements;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.ValueKind;
+import org.apache.beam.sdk.values.ValueKindUtil;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * A {@link Coder} for {@link ValueKind}, encoded in 1 byte as the matching {@link
+ * Elements.ValueKind.Enum} number, so the wire format stays stable if the enum is reordered and
+ * matches the portability representation.
+ */
+public class ValueKindCoder extends AtomicCoder<ValueKind> {
+
+  public static ValueKindCoder of() {
+    return INSTANCE;
+  }
+
+  private static final ValueKindCoder INSTANCE = new ValueKindCoder();
+  private static final TypeDescriptor<ValueKind> TYPE_DESCRIPTOR =
+      TypeDescriptor.of(ValueKind.class);
+
+  private ValueKindCoder() {}
+
+  @Override
+  public void encode(ValueKind value, OutputStream outStream) throws IOException, CoderException {
+    if (value == null) {
+      throw new CoderException("cannot encode a null ValueKind");
+    }
+    outStream.write(ValueKindUtil.toProto(value).getNumber());
+  }
+
+  @Override
+  public ValueKind decode(InputStream inStream) throws IOException, CoderException {
+    int number = inStream.read();
+    if (number == -1) {
+      throw new CoderException(new EOFException("EOF encountered decoding a ValueKind"));
+    }
+    Elements.ValueKind.@Nullable Enum proto = Elements.ValueKind.Enum.forNumber(number);
+    if (proto == null) {
+      throw new CoderException("Unknown ValueKind number: " + number);
+    }
+
+    return ValueKindUtil.fromProto(proto);
+  }
+
+  @Override
+  public boolean consistentWithEquals() {
+    return true;
+  }
+
+  @Override
+  public boolean isRegisterByteSizeObserverCheap(ValueKind value) {
+    return true;
+  }
+
+  @Override
+  protected long getEncodedElementByteSize(ValueKind value) {
+    return 1;
+  }
+
+  @Override
+  public TypeDescriptor<ValueKind> getEncodedTypeDescriptor() {
+    return TYPE_DESCRIPTOR;
+  }
+}
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Lineage.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Lineage.java
index f845d8fde940..681f022c9042 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Lineage.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/Lineage.java
@@ -19,6 +19,7 @@
 
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -122,6 +123,10 @@ private static Lineage createLineage(PipelineOptions options, LineageDirection d
   }
 
   /** {@link Lineage} representing sources and optionally side inputs. */
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Every reporter writes into the same metric cell, so all callers need the one shared instance. A copy would drop the lineage it records.")
   public static Lineage getSources() {
     Lineage localSources = sources;
     if (localSources == null) {
@@ -131,6 +136,10 @@ public static Lineage getSources() {
   }
 
   /** {@link Lineage} representing sinks. */
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Every reporter writes into the same metric cell, so all callers need the one shared instance. A copy would drop the lineage it records.")
   public static Lineage getSinks() {
     Lineage localSinks = sinks;
     if (localSinks == null) {
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SourceMetrics.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SourceMetrics.java
index e9677626f88b..5d033d4a91aa 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SourceMetrics.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/metrics/SourceMetrics.java
@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.metrics;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Joiner;
 
 /** Standard {@link org.apache.beam.sdk.io.Source} Metrics. */
@@ -69,6 +70,10 @@ public static Counter bytesReadBySplit(String splitId) {
   }
 
   /** Gauge for source backlog in bytes. */
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "A Gauge is a handle onto one metric cell, not a value. A copy would send the reading nowhere.")
   public static Gauge backlogBytes() {
     return BACKLOG_BYTES_GAUGE;
   }
@@ -84,6 +89,10 @@ public static Gauge backlogBytesOfSplit(String splitId) {
   }
 
   /** Gauge for source backlog in elements. */
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "A Gauge is a handle onto one metric cell, not a value. A copy would send the reading nowhere.")
   public static Gauge backlogElements() {
     return BACKLOG_ELEMENTS_GAUGE;
   }
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/ModelCoders.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/ModelCoders.java
index 5059cc1c6b83..f625e376abcc 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/ModelCoders.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/ModelCoders.java
@@ -22,6 +22,7 @@
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
 import com.google.auto.value.AutoValue;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.Set;
 import org.apache.beam.model.pipeline.v1.RunnerApi.Coder;
 import org.apache.beam.model.pipeline.v1.RunnerApi.FunctionSpec;
@@ -97,6 +98,12 @@ private ModelCoders() {}
           SHARDED_KEY_CODER_URN,
           NULLABLE_CODER_URN);
 
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Returns a Guava ImmutableSet."
+              + " Spotbugs matches its known-immutable list by fully qualified name, so it cannot recognise collections relocated into org.apache.beam.vendor.guava."
+              + " See https://github.com/spotbugs/spotbugs/issues/1601.")
   public static Set<String> urns() {
     return MODEL_CODER_URNS;
   }
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/PTransformTranslation.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/PTransformTranslation.java
index 5f8782265f17..88bfb8b5b538 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/PTransformTranslation.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/PTransformTranslation.java
@@ -22,6 +22,7 @@
 import static org.apache.beam.sdk.util.construction.BeamUrns.getUrn;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
@@ -417,6 +418,12 @@ public RunnerApi.PTransform translate(
       knownPayloadTranslators;
 
   @Internal
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Returns a Guava ImmutableMap."
+              + " Spotbugs matches its known-immutable list by fully qualified name, so it cannot recognise collections relocated into org.apache.beam.vendor.guava."
+              + " See https://github.com/spotbugs/spotbugs/issues/1601.")
   public static Map<Class<? extends PTransform>, TransformPayloadTranslator>
       getKnownPayloadTranslators() {
     if (knownPayloadTranslators == null) {
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/ValueKindCoderTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/ValueKindCoderTest.java
new file mode 100644
index 000000000000..aeac2132c853
--- /dev/null
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/ValueKindCoderTest.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.coders;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.List;
+import org.apache.beam.sdk.testing.CoderProperties;
+import org.apache.beam.sdk.util.CoderUtils;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.ValueKind;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link ValueKindCoder}. */
+@RunWith(JUnit4.class)
+public class ValueKindCoderTest {
+
+  private static final Coder<ValueKind> TEST_CODER = ValueKindCoder.of();
+
+  private static final List<ValueKind> TEST_VALUES =
+      Arrays.asList(
+          ValueKind.INSERT, ValueKind.UPDATE_BEFORE, ValueKind.UPDATE_AFTER, ValueKind.DELETE);
+
+  /** One byte per value, holding the proto enum number. */
+  private static final List<String> TEST_ENCODINGS = Arrays.asList("AQ", "Ag", "Aw", "BA");
+
+  @Rule public ExpectedException thrown = ExpectedException.none();
+
+  @Test
+  public void testDecodeEncodeEqual() throws Exception {
+    for (ValueKind value : TEST_VALUES) {
+      CoderProperties.coderDecodeEncodeEqual(TEST_CODER, value);
+    }
+  }
+
+  @Test
+  public void testWireFormatEncode() throws Exception {
+    CoderProperties.coderEncodesBase64(TEST_CODER, TEST_VALUES, TEST_ENCODINGS);
+  }
+
+  /** VALUE_KIND_UNSPECIFIED (0) means INSERT, for backwards compatibility. */
+  @Test
+  public void testDecodeUnspecified() throws Exception {
+    assertEquals(ValueKind.INSERT, CoderUtils.decodeFromBase64(TEST_CODER, "AA"));
+  }
+
+  @Test
+  public void testDecodeUnknownNumberThrows() throws Exception {
+    thrown.expect(CoderException.class);
+    thrown.expectMessage("Unknown ValueKind number: 42");
+
+    CoderUtils.decodeFromBase64(TEST_CODER, "Kg");
+  }
+
+  @Test
+  public void testCoderRegistryResolvesValueKind() throws Exception {
+    assertEquals(
+        ValueKindCoder.of(),
+        CoderRegistry.createDefault().getCoder(TypeDescriptor.of(ValueKind.class)));
+  }
+}
diff --git a/sdks/java/io/iceberg/build.gradle b/sdks/java/io/iceberg/build.gradle
index e2e8a12d01eb..7340d1eeb411 100644
--- a/sdks/java/io/iceberg/build.gradle
+++ b/sdks/java/io/iceberg/build.gradle
@@ -49,6 +49,7 @@ dependencies {
     implementation library.java.avro
     implementation library.java.slf4j_api
     implementation library.java.joda_time
+    implementation library.java.guava
     implementation "org.apache.parquet:parquet-column:$parquet_version"
     implementation "org.apache.parquet:parquet-hadoop:$parquet_version"
     implementation "org.apache.parquet:parquet-common:$parquet_version"
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemas.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemas.java
index 1b81e008f19d..5c36967cfeba 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemas.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemas.java
@@ -17,87 +17,219 @@
  */
 package org.apache.beam.sdk.io.iceberg;
 
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.TreeMap;
+import java.util.TreeSet;
+import org.apache.beam.sdk.coders.AtomicCoder;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.CoderRegistry;
-import org.apache.beam.sdk.coders.KvCoder;
 import org.apache.beam.sdk.coders.ListCoder;
 import org.apache.beam.sdk.coders.MapCoder;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
 import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.NoSuchSchemaException;
+import org.apache.beam.sdk.schemas.SchemaCoder;
+import org.apache.beam.sdk.schemas.SchemaRegistry;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber;
 import org.apache.beam.sdk.transforms.Combine;
-import org.apache.beam.sdk.values.KV;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
- * Collects the distinct schemas among canonical file schema JSONs (see {@link FileSchemas}), with
- * the number of files per schema, most common first (ties broken by JSON). The commit side applies
- * schemas in this order, so the schema covering the most files wins a conflict.
- *
- * <p>Inputs are compared as strings, so they must already be canonical.
+ * One output entry per distinct schema: its file count and the columns EVERY file carrying it
+ * proved free of nulls; one file with a null in "name" forces "name" to relax, however many clean
+ * files sit next to it. Entries come out most common first (ties broken by the JSON text) because
+ * the commit side applies schemas in that order and the most common schema should win a conflict.
+ * Schemas are compared as strings, so inputs must already be canonical.
  */
 class CollectDistinctSchemas
-    extends Combine.CombineFn<String, Map<String, Long>, List<KV<String, Long>>> {
+    extends Combine.CombineFn<
+        CollectDistinctSchemas.SchemaGroup,
+        Map<String, CollectDistinctSchemas.Group>,
+        List<CollectDistinctSchemas.SchemaGroup>> {
+
+  /** Mutable accumulator counterpart of {@link SchemaGroup}. */
+  static final class Group {
+    long files;
+    TreeSet<String> nullFreeColumns;
+
+    Group(long files, TreeSet<String> nullFreeColumns) {
+      this.files = files;
+      this.nullFreeColumns = nullFreeColumns;
+    }
+
+    @Override
+    public boolean equals(@Nullable Object other) {
+      if (!(other instanceof Group)) {
+        return false;
+      }
+      Group that = (Group) other;
+      return files == that.files && nullFreeColumns.equals(that.nullFreeColumns);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(files, nullFreeColumns);
+    }
+  }
+
+  /**
+   * A schema, how many files carry it, and the columns all of them proved free of nulls.
+   * ReadFooterSchema emits one per file ({@code files} = 1); this combiner merges them.
+   */
+  @DefaultSchema(AutoValueSchema.class)
+  @AutoValue
+  abstract static class SchemaGroup {
+    private static @MonotonicNonNull SchemaCoder<SchemaGroup> coder;
+
+    static SchemaGroup of(String schemaJson, long files, List<String> nullFreeColumns) {
+      return new AutoValue_CollectDistinctSchemas_SchemaGroup(schemaJson, files, nullFreeColumns);
+    }
+
+    static SchemaCoder<SchemaGroup> getCoder() {
+      if (coder == null) {
+        try {
+          coder = SchemaRegistry.createDefault().getSchemaCoder(SchemaGroup.class);
+        } catch (NoSuchSchemaException e) {
+          throw new RuntimeException(e);
+        }
+      }
+      return coder;
+    }
+
+    @SchemaFieldNumber("0")
+    abstract String getSchemaJson();
+
+    @SchemaFieldNumber("1")
+    abstract long getFiles();
+
+    @SchemaFieldNumber("2")
+    abstract List<String> getNullFreeColumns();
+
+    @Override
+    public final String toString() {
+      return getFiles()
+          + " file(s), null-free in "
+          + getNullFreeColumns()
+          + ", schema "
+          + getSchemaJson();
+    }
+  }
 
   @Override
-  public Map<String, Long> createAccumulator() {
+  public Map<String, Group> createAccumulator() {
     return new TreeMap<>();
   }
 
   @Override
-  public Map<String, Long> addInput(Map<String, Long> accumulator, String schemaJson) {
-    add(accumulator, schemaJson, 1L);
+  public Map<String, Group> addInput(Map<String, Group> accumulator, SchemaGroup file) {
+    add(accumulator, file.getSchemaJson(), file.getFiles(), file.getNullFreeColumns());
     return accumulator;
   }
 
   @Override
-  public Map<String, Long> mergeAccumulators(Iterable<Map<String, Long>> accumulators) {
-    Map<String, Long> merged = createAccumulator();
-    for (Map<String, Long> accumulator : accumulators) {
-      for (Map.Entry<String, Long> entry : accumulator.entrySet()) {
-        add(merged, entry.getKey(), entry.getValue());
+  public Map<String, Group> mergeAccumulators(Iterable<Map<String, Group>> accumulators) {
+    Map<String, Group> merged = createAccumulator();
+    for (Map<String, Group> accumulator : accumulators) {
+      for (Map.Entry<String, Group> entry : accumulator.entrySet()) {
+        add(merged, entry.getKey(), entry.getValue().files, entry.getValue().nullFreeColumns);
       }
     }
     return merged;
   }
 
   @Override
-  public List<KV<String, Long>> extractOutput(Map<String, Long> accumulator) {
-    List<KV<String, Long>> schemas = new ArrayList<>();
-    for (Map.Entry<String, Long> entry : accumulator.entrySet()) {
-      schemas.add(KV.of(entry.getKey(), entry.getValue()));
+  public List<SchemaGroup> extractOutput(Map<String, Group> accumulator) {
+    List<SchemaGroup> schemas = new ArrayList<>();
+    for (Map.Entry<String, Group> entry : accumulator.entrySet()) {
+      schemas.add(
+          SchemaGroup.of(
+              entry.getKey(),
+              entry.getValue().files,
+              new ArrayList<>(entry.getValue().nullFreeColumns)));
     }
     schemas.sort(
         (a, b) -> {
-          int byCount = Long.compare(b.getValue(), a.getValue());
+          int byCount = Long.compare(b.getFiles(), a.getFiles());
           if (byCount != 0) {
             return byCount;
           }
-          return a.getKey().compareTo(b.getKey());
+          return a.getSchemaJson().compareTo(b.getSchemaJson());
         });
     return schemas;
   }
 
   @Override
-  public Coder<Map<String, Long>> getAccumulatorCoder(
-      CoderRegistry registry, Coder<String> inputCoder) {
-    return MapCoder.of(StringUtf8Coder.of(), VarLongCoder.of());
+  public Coder<Map<String, Group>> getAccumulatorCoder(
+      CoderRegistry registry, Coder<SchemaGroup> inputCoder) {
+    return MapCoder.of(StringUtf8Coder.of(), GroupCoder.INSTANCE);
   }
 
   @Override
-  public Coder<List<KV<String, Long>>> getDefaultOutputCoder(
-      CoderRegistry registry, Coder<String> inputCoder) {
-    return ListCoder.of(KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of()));
+  public Coder<List<SchemaGroup>> getDefaultOutputCoder(
+      CoderRegistry registry, Coder<SchemaGroup> inputCoder) {
+    return outputCoder();
+  }
+
+  static Coder<SchemaGroup> groupCoder() {
+    return SchemaGroup.getCoder();
+  }
+
+  static Coder<List<SchemaGroup>> outputCoder() {
+    return ListCoder.of(SchemaGroup.getCoder());
+  }
+
+  private static final Coder<List<String>> COLUMNS_CODER = ListCoder.of(StringUtf8Coder.of());
+
+  /** Sorted columns, so the encoding is deterministic. */
+  private static class GroupCoder extends AtomicCoder<Group> {
+    static final GroupCoder INSTANCE = new GroupCoder();
+
+    private GroupCoder() {}
+
+    @Override
+    public void encode(Group value, OutputStream out) throws IOException {
+      VarLongCoder.of().encode(value.files, out);
+      COLUMNS_CODER.encode(new ArrayList<>(value.nullFreeColumns), out);
+    }
+
+    @Override
+    public Group decode(InputStream in) throws IOException {
+      long files = VarLongCoder.of().decode(in);
+      return new Group(files, new TreeSet<>(COLUMNS_CODER.decode(in)));
+    }
   }
 
-  private static void add(Map<String, Long> accumulator, String schemaJson, long count) {
-    Long existing = accumulator.get(schemaJson);
+  private static void add(
+      Map<String, Group> accumulator,
+      String schemaJson,
+      long files,
+      Iterable<String> nullFreeColumns) {
+    Group existing = accumulator.get(schemaJson);
     if (existing == null) {
-      accumulator.put(schemaJson, count);
-    } else {
-      accumulator.put(schemaJson, existing + count);
+      TreeSet<String> copy = new TreeSet<>();
+      for (String column : nullFreeColumns) {
+        copy.add(column);
+      }
+      accumulator.put(schemaJson, new Group(files, copy));
+      return;
+    }
+    existing.files += files;
+    TreeSet<String> stillNullFree = new TreeSet<>();
+    for (String column : nullFreeColumns) {
+      if (existing.nullFreeColumns.contains(column)) {
+        stillNullFree.add(column);
+      }
     }
+    existing.nullFreeColumns = stillNullFree;
   }
 }
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileSchemas.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileSchemas.java
index 592e11e8c767..d95321153d5e 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileSchemas.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileSchemas.java
@@ -18,29 +18,222 @@
 package org.apache.beam.sdk.io.iceberg;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SchemaParser;
 import org.apache.iceberg.parquet.ParquetSchemaUtil;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
 import org.apache.parquet.hadoop.metadata.ParquetMetadata;
 
 /**
- * Derives the schema a file contributes to schema inference. The canonical form sorts struct fields
- * by name at every level and renumbers ids in deterministic order, so files that differ only in
- * column order produce identical JSON. Ids are positional and meaningless: the commit side
- * reconciles columns by name.
+ * What a file contributes to schema inference: the canonical form of the schema it declares, and
+ * the columns its footer proves free of nulls.
+ *
+ * <p>The schema half depends only on the declared schema, never on the data, so files written by
+ * the same job dedup to one entry no matter where their nulls fall. The null evidence is combined
+ * per schema by {@link CollectDistinctSchemas} and reapplied by the commit side via {@link
+ * #markRequired}.
+ *
+ * <p>The canonical form sorts struct fields by name at every level and renumbers ids. Ids are
+ * positional and meaningless (the commit side reconciles columns by name), so never diff two file
+ * schemas by id. Other field attributes (doc, defaults) are preserved, matching what SchemaDelta
+ * compares. Column paths are dotted, like pins.
  */
 final class FileSchemas {
   private FileSchemas() {}
 
+  /** Canonical JSON of the schema the file declares. */
   static String canonicalJson(ParquetMetadata footer) {
     Schema converted = ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema());
     return SchemaParser.toJson(canonical(converted));
   }
 
+  /** This file as a schema group of one: its declared schema and its null-free columns. */
+  static CollectDistinctSchemas.SchemaGroup schemaGroup(ParquetMetadata footer) {
+    Schema converted = ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema());
+    Schema tightened = tighten(converted, footer);
+    return CollectDistinctSchemas.SchemaGroup.of(
+        SchemaParser.toJson(canonical(converted)), 1, changedToRequired(converted, tightened));
+  }
+
+  /**
+   * Marks a declared-optional column required when every row group has a null count of zero for it,
+   * so the file does not request a relaxation it does not need; an absent count is not proof. A
+   * struct is null-free when any leaf under it is (a null struct nulls all its leaves). Nothing
+   * under lists or maps is tightened: a zero count there would be valid evidence too, but mapping
+   * physical chunk paths (writer-dependent names like {@code list.element}, {@code array}) onto the
+   * converted schema is not worth it. A file with no rows (no row groups, or only empty ones, as
+   * pyarrow writes an empty table) proves every column, matching how the pin check treats empty
+   * files.
+   */
+  static Schema tighten(Schema schema, ParquetMetadata footer) {
+    if (rowCount(footer) == 0) {
+      return new Schema(tightenAll(schema.asStruct()).fields());
+    }
+    Set<List<String>> zeroNullLeaves = leafPathsWithZeroNullCounts(footer);
+    if (zeroNullLeaves.isEmpty()) {
+      return schema;
+    }
+    return new Schema(
+        tightenStruct(schema.asStruct(), new ArrayList<>(), zeroNullLeaves).struct.fields());
+  }
+
+  /** With no rows, nothing can hold a null: every leaf and struct outside lists and maps. */
+  private static Types.StructType tightenAll(Types.StructType struct) {
+    List<Types.NestedField> fields = new ArrayList<>();
+    for (Types.NestedField field : struct.fields()) {
+      Type type = field.type();
+      if (type.isStructType()) {
+        fields.add(withOptionality(field, tightenAll(type.asStructType()), false));
+      } else if (type.isPrimitiveType()) {
+        fields.add(withOptionality(field, type, false));
+      } else {
+        fields.add(field);
+      }
+    }
+    return Types.StructType.of(fields);
+  }
+
+  /**
+   * Returns the schema with the given dotted column paths made required. The commit side parses a
+   * group's schema JSON (optionality as the writer declared it) and applies the group's null-free
+   * columns with this before classifying, so only relaxations some file actually needs remain.
+   */
+  static Schema markRequired(Schema declared, Collection<String> columns) {
+    if (columns.isEmpty()) {
+      return declared;
+    }
+    Types.StructType required = markRequiredStruct(declared.asStruct(), "", new HashSet<>(columns));
+    return new Schema(required.fields());
+  }
+
+  private static Types.StructType markRequiredStruct(
+      Types.StructType struct, String prefix, Set<String> columns) {
+    List<Types.NestedField> fields = new ArrayList<>();
+    for (Types.NestedField field : struct.fields()) {
+      String path = prefix + field.name();
+      Type type = field.type();
+      if (type.isStructType()) {
+        type = markRequiredStruct(type.asStructType(), path + ".", columns);
+      }
+      boolean required = !field.isRequired() && columns.contains(path);
+      fields.add(withOptionality(field, type, field.isOptional() && !required));
+    }
+    return Types.StructType.of(fields);
+  }
+
+  /** Dotted paths of fields the tightened schema made required, sorted. */
+  private static List<String> changedToRequired(Schema declared, Schema tightened) {
+    List<String> paths = new ArrayList<>();
+    collectChangedToRequired(declared.asStruct(), tightened.asStruct(), "", paths);
+    Collections.sort(paths);
+    return paths;
+  }
+
+  private static void collectChangedToRequired(
+      Types.StructType declared, Types.StructType tightened, String prefix, List<String> out) {
+    for (int i = 0; i < declared.fields().size(); i++) {
+      Types.NestedField before = declared.fields().get(i);
+      Types.NestedField after = tightened.fields().get(i);
+      String path = prefix + before.name();
+      if (before.isOptional() && after.isRequired()) {
+        out.add(path);
+      }
+      if (before.type().isStructType()) {
+        collectChangedToRequired(
+            before.type().asStructType(), after.type().asStructType(), path + ".", out);
+      }
+    }
+  }
+
+  private static long rowCount(ParquetMetadata footer) {
+    long rows = 0;
+    for (BlockMetaData block : footer.getBlocks()) {
+      rows += block.getRowCount();
+    }
+    return rows;
+  }
+
+  /**
+   * Leaf paths proven null-free in every row group that has rows (intersection over blocks). An
+   * empty row group holds no nulls whatever its statistics say, so it constrains nothing.
+   */
+  private static Set<List<String>> leafPathsWithZeroNullCounts(ParquetMetadata footer) {
+    Set<List<String>> proven = null;
+    for (BlockMetaData block : footer.getBlocks()) {
+      if (block.getRowCount() == 0) {
+        continue;
+      }
+      Set<List<String>> provenHere = new HashSet<>();
+      for (ColumnChunkMetaData chunk : block.getColumns()) {
+        Statistics<?> stats = chunk.getStatistics();
+        if (stats != null && stats.isNumNullsSet() && stats.getNumNulls() == 0) {
+          provenHere.add(Arrays.asList(chunk.getPath().toArray()));
+        }
+      }
+      if (proven == null) {
+        proven = provenHere;
+      } else {
+        proven.retainAll(provenHere);
+      }
+    }
+    if (proven == null) {
+      return new HashSet<>();
+    }
+    return proven;
+  }
+
+  private static final class Tightened {
+    final Types.StructType struct;
+
+    /** Some leaf below, not under a list or map, is proven: the struct itself was never null. */
+    final boolean hasNullFreeLeaf;
+
+    Tightened(Types.StructType struct, boolean hasNullFreeLeaf) {
+      this.struct = struct;
+      this.hasNullFreeLeaf = hasNullFreeLeaf;
+    }
+  }
+
+  private static Tightened tightenStruct(
+      Types.StructType struct, List<String> path, Set<List<String>> zeroNulls) {
+    List<Types.NestedField> fields = new ArrayList<>();
+    boolean hasNullFreeLeaf = false;
+    for (Types.NestedField field : struct.fields()) {
+      path.add(field.name());
+      if (field.type().isPrimitiveType()) {
+        boolean nullFreeLeaf = zeroNulls.contains(path);
+        fields.add(nullFreeLeaf ? withOptionality(field, field.type(), false) : field);
+        hasNullFreeLeaf |= nullFreeLeaf;
+      } else if (field.type().isStructType()) {
+        Tightened child = tightenStruct(field.type().asStructType(), path, zeroNulls);
+        boolean optional = field.isOptional() && !child.hasNullFreeLeaf;
+        fields.add(withOptionality(field, child.struct, optional));
+        hasNullFreeLeaf |= child.hasNullFreeLeaf;
+      } else {
+        fields.add(field);
+      }
+      path.remove(path.size() - 1);
+    }
+    return new Tightened(Types.StructType.of(fields), hasNullFreeLeaf);
+  }
+
+  /** Copies every attribute (id, name, doc, defaults), replacing only type and optionality. */
+  private static Types.NestedField withOptionality(
+      Types.NestedField field, Type type, boolean optional) {
+    return Types.NestedField.from(field).ofType(type).isOptional(optional).build();
+  }
+
   static Schema canonical(Schema schema) {
     Type sorted = TypeUtil.visit(schema.asStruct(), new SortFields());
     int[] nextId = {0};
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java
index 571c0b552441..e559feab3512 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java
@@ -35,10 +35,10 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * Emits the canonical schema (see {@link FileSchemas}) of every readable Parquet file as JSON.
- * Unreadable or non-Parquet files contribute nothing.
+ * Emits one {@link CollectDistinctSchemas.SchemaGroup} of one file per readable Parquet file: its
+ * declared schema and its null-free columns. Unreadable or non-Parquet files contribute nothing.
  */
-class ReadFooterSchema extends DoFn<String, String> {
+class ReadFooterSchema extends DoFn<String, CollectDistinctSchemas.SchemaGroup> {
   private static final Logger LOG = LoggerFactory.getLogger(ReadFooterSchema.class);
 
   static final int DEFAULT_THREAD_POOL_SIZE = 10;
@@ -65,24 +65,21 @@ class ReadFooterSchema extends DoFn<String, String> {
     this.maxInFlightTasks = maxInFlightTasks;
   }
 
-  /**
-   * {@code schemaJson} is null when the file contributes no schema. Counters are updated when the
-   * result is delivered, on the processing thread: metrics touched from the executor are lost.
-   */
+  /** Counters are updated on the processing thread: metrics touched from the executor are lost. */
   private static class ReadResult {
-    final @Nullable String schemaJson;
+    final CollectDistinctSchemas.@Nullable SchemaGroup schema;
     final boolean footerError;
     final Instant timestamp;
     final BoundedWindow window;
     final PaneInfo paneInfo;
 
     ReadResult(
-        @Nullable String schemaJson,
+        CollectDistinctSchemas.@Nullable SchemaGroup schema,
         boolean footerError,
         Instant timestamp,
         BoundedWindow window,
         PaneInfo paneInfo) {
-      this.schemaJson = schemaJson;
+      this.schema = schema;
       this.footerError = footerError;
       this.timestamp = timestamp;
       this.window = window;
@@ -114,7 +111,7 @@ public void process(
       @Timestamp Instant timestamp,
       BoundedWindow window,
       PaneInfo paneInfo,
-      OutputReceiver<String> output)
+      OutputReceiver<CollectDistinctSchemas.SchemaGroup> output)
       throws Exception {
     numFilesRead.inc();
     Callable<ReadResult> task = createReadTask(filePath, timestamp, window, paneInfo);
@@ -128,24 +125,22 @@ public void finishBundle(FinishBundleContext context) throws Exception {
 
   private static void outputAtFinish(ReadResult result, FinishBundleContext context) {
     count(result);
-    if (result.schemaJson != null) {
-      context.output(result.schemaJson, result.timestamp, result.window);
+    if (result.schema != null) {
+      context.output(result.schema, result.timestamp, result.window);
     }
   }
 
-  private static void outputResult(ReadResult result, OutputReceiver<String> output) {
+  private static void outputResult(
+      ReadResult result, OutputReceiver<CollectDistinctSchemas.SchemaGroup> output) {
     count(result);
-    if (result.schemaJson != null) {
+    if (result.schema != null) {
       output.outputWindowedValue(
-          result.schemaJson,
-          result.timestamp,
-          Collections.singleton(result.window),
-          result.paneInfo);
+          result.schema, result.timestamp, Collections.singleton(result.window), result.paneInfo);
     }
   }
 
   private static void count(ReadResult result) {
-    if (result.schemaJson != null) {
+    if (result.schema != null) {
       numSchemasEmitted.inc();
     }
     if (result.footerError) {
@@ -167,8 +162,7 @@ private static Callable<ReadResult> createReadTask(
       }
       try {
         ParquetMetadata footer = ParquetFooters.read(filePath);
-        return new ReadResult(
-            FileSchemas.canonicalJson(footer), false, timestamp, window, paneInfo);
+        return new ReadResult(FileSchemas.schemaGroup(footer), false, timestamp, window, paneInfo);
       } catch (Exception e) {
         LOG.warn(
             "Could not read the footer of {}; the file will not contribute to schema inference: {}",
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFile.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFile.java
index ceb96d50f8aa..779bc3753c7d 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFile.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFile.java
@@ -28,6 +28,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.sdk.schemas.AutoValueSchema;
 import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
 import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber;
@@ -37,11 +38,14 @@
 import org.apache.iceberg.FileMetadata;
 import org.apache.iceberg.Metrics;
 import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.SingleValueParser;
 import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.StructLike;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 @DefaultSchema(AutoValueSchema.class)
 @AutoValue
+@Internal
 public abstract class SerializableDeleteFile {
   public static SerializableDeleteFile.Builder builder() {
     return new AutoValue_SerializableDeleteFile.Builder();
@@ -62,7 +66,11 @@ public static SerializableDeleteFile.Builder builder() {
   @SchemaFieldNumber("4")
   public abstract long getFileSizeInBytes();
 
+  /**
+   * @deprecated Use {@link #getJsonPartition()} instead.
+   */
   @SchemaFieldNumber("5")
+  @Deprecated
   public abstract String getPartitionPath();
 
   @SchemaFieldNumber("6")
@@ -113,6 +121,9 @@ public static SerializableDeleteFile.Builder builder() {
   @SchemaFieldNumber("21")
   public abstract @Nullable Long getFileSequenceNumber();
 
+  @SchemaFieldNumber("22")
+  abstract @Nullable String getJsonPartition();
+
   @AutoValue.Builder
   abstract static class Builder {
     abstract Builder setContentType(FileContent content);
@@ -127,6 +138,8 @@ abstract static class Builder {
 
     abstract Builder setPartitionPath(String partitionPath);
 
+    abstract Builder setJsonPartition(String jsonPartition);
+
     abstract Builder setPartitionSpecId(int partitionSpec);
 
     abstract Builder setSortOrderId(@Nullable Integer sortOrderId);
@@ -163,7 +176,47 @@ abstract static class Builder {
   }
 
   public static SerializableDeleteFile from(
-      DeleteFile deleteFile, String partitionPath, boolean includeMetrics) {
+      DeleteFile deleteFile, Map<Integer, PartitionSpec> specs) {
+    return from(deleteFile, specs, true);
+  }
+
+  /**
+   * Creates a {@link SerializableDeleteFile}, resolving the file's {@link PartitionSpec} by its own
+   * spec id.
+   *
+   * <p>Delete files reached from a scan task may carry a spec id that differs from the spec of the
+   * data file they apply to, so the lookup has to be per delete file rather than against a single
+   * "current" spec.
+   */
+  public static SerializableDeleteFile from(
+      DeleteFile deleteFile, Map<Integer, PartitionSpec> specs, boolean includeMetrics) {
+    return from(
+        deleteFile,
+        checkStateNotNull(
+            specs.get(deleteFile.specId()),
+            "Could not create a SerializableDeleteFile because DeleteFile is written using a partition spec id '%s' that is not found in the provided specs: %s",
+            deleteFile.specId(),
+            specs.keySet()),
+        includeMetrics);
+  }
+
+  public static SerializableDeleteFile from(DeleteFile deleteFile, PartitionSpec spec) {
+    return from(deleteFile, spec, true);
+  }
+
+  public static SerializableDeleteFile from(
+      DeleteFile deleteFile, PartitionSpec spec, boolean includeMetrics) {
+    if (spec.specId() != deleteFile.specId()) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Cannot serialize DeleteFile: its partition spec id %s does not match the provided "
+                  + "spec id %s.",
+              deleteFile.specId(), spec.specId()));
+    }
+    // jsonPartition is the primary (handles evolved specs, special characters).
+    // partitionPath is the fallback for values that don't round-trip through JSON.
+    String jsonPartition = SingleValueParser.toJson(spec.partitionType(), deleteFile.partition());
+    String partitionPath = spec.partitionToPath(deleteFile.partition());
 
     SerializableDeleteFile.Builder builder =
         SerializableDeleteFile.builder()
@@ -171,6 +224,7 @@ public static SerializableDeleteFile from(
             .setFileFormat(deleteFile.format().name())
             .setFileSizeInBytes(deleteFile.fileSizeInBytes())
             .setPartitionPath(partitionPath)
+            .setJsonPartition(jsonPartition)
             .setPartitionSpecId(deleteFile.specId())
             .setRecordCount(deleteFile.recordCount())
             .setColumnSizes(deleteFile.columnSizes())
@@ -228,7 +282,21 @@ public DeleteFile createDeleteFile(
             .withMetrics(metrics)
             .withSplitOffsets(getSplitOffsets())
             .withEncryptionKeyMetadata(getKeyMetadata())
-            .withPartitionPath(getPartitionPath());
+            .withReferencedDataFile(getReferencedDataFile());
+
+    @Nullable String jsonPartition = getJsonPartition();
+    if (jsonPartition != null) {
+      try {
+        deleteFileBuilder = deleteFileBuilder.withPartition(partition(partitionSpec));
+      } catch (RuntimeException e) {
+        // Some partition values (e.g. NaN / Infinity floating-point) don't round-trip through the
+        // JSON representation; fall back to the partition-path string
+        deleteFileBuilder = deleteFileBuilder.withPartitionPath(getPartitionPath());
+      }
+    } else {
+      // Elements decoded from a pre-jsonPartition release carry only the partition path.
+      deleteFileBuilder = deleteFileBuilder.withPartitionPath(getPartitionPath());
+    }
 
     switch (getContentType()) {
       case POSITION_DELETES:
@@ -260,17 +328,22 @@ public DeleteFile createDeleteFile(
             "Unexpected content type for DeleteFile: " + getContentType());
     }
 
-    // needed for puffin files
+    // contentOffset / contentSizeInBytes really are Puffin-only: build() rejects a non-null value
+    // for either on any other format, and requires both (plus referencedDataFile) on Puffin.
     if (getFileFormat().equalsIgnoreCase(FileFormat.PUFFIN.name())) {
       deleteFileBuilder =
           deleteFileBuilder
               .withContentOffset(checkStateNotNull(getContentOffset()))
-              .withContentSizeInBytes(checkStateNotNull(getContentSizeInBytes()))
-              .withReferencedDataFile(checkStateNotNull(getReferencedDataFile()));
+              .withContentSizeInBytes(checkStateNotNull(getContentSizeInBytes()));
     }
     return deleteFileBuilder.build();
   }
 
+  private StructLike partition(PartitionSpec spec) {
+    return (StructLike)
+        SingleValueParser.fromJson(spec.partitionType(), checkStateNotNull(getJsonPartition()));
+  }
+
   @Override
   public final boolean equals(@Nullable Object o) {
     if (this == o) {
@@ -287,6 +360,7 @@ && getRecordCount() == that.getRecordCount()
         && getFileSizeInBytes() == that.getFileSizeInBytes()
         && getPartitionPath().equals(that.getPartitionPath())
         && getPartitionSpecId() == that.getPartitionSpecId()
+        && Objects.equals(getJsonPartition(), that.getJsonPartition())
         && Objects.equals(getSortOrderId(), that.getSortOrderId())
         && Objects.equals(getEqualityFieldIds(), that.getEqualityFieldIds())
         && Objects.equals(getKeyMetadata(), that.getKeyMetadata())
@@ -314,6 +388,7 @@ public final int hashCode() {
             getRecordCount(),
             getFileSizeInBytes(),
             getPartitionPath(),
+            getJsonPartition(),
             getPartitionSpecId(),
             getSortOrderId(),
             getEqualityFieldIds(),
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTask.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTask.java
index 3410c0a9d7ee..97bcbaa5bec2 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTask.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/SerializableChangelogTask.java
@@ -17,7 +17,6 @@
  */
 package org.apache.beam.sdk.io.iceberg.cdc;
 
-import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
 import com.google.auto.value.AutoValue;
@@ -267,13 +266,10 @@ static List<DeleteFile> getAddedDeleteFiles(ChangelogScanTask task) {
 
   private static List<SerializableDeleteFile> toSerializableDeletes(
       List<DeleteFile> dfs, Map<Integer, PartitionSpec> specs, boolean includeMetrics) {
+    // Serialize each delete file against its own spec (looked up by its spec id): a delete file may
+    // carry a different spec id than the data file it applies to.
     return dfs.stream()
-        .map(
-            df ->
-                SerializableDeleteFile.from(
-                    df,
-                    checkStateNotNull(specs.get(df.specId())).partitionToPath(df.partition()),
-                    includeMetrics))
+        .map(df -> SerializableDeleteFile.from(df, specs, includeMetrics))
         .collect(Collectors.toList());
   }
 }
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java
new file mode 100644
index 000000000000..1aa82914fde9
--- /dev/null
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.iceberg.cdc.sink;
+
+import com.google.common.base.MoreObjects;
+import java.util.Objects;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.ValueKind;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * One change record carried through the CDC sink's shuffle.
+ *
+ * <p>{@link ValueKind} is reified because it's not preserved across a {@code GroupByKey}.
+ */
+final class CdcRecord {
+
+  private final Row data;
+  private final ValueKind kind;
+  private final long sequenceNumber;
+
+  private CdcRecord(Row data, ValueKind kind, long sequenceNumber) {
+    this.data = data;
+    this.kind = kind;
+    this.sequenceNumber = sequenceNumber;
+  }
+
+  public static CdcRecord of(Row data, ValueKind kind, long sequenceNumber) {
+    return new CdcRecord(data, kind, sequenceNumber);
+  }
+
+  public Row getData() {
+    return data;
+  }
+
+  public ValueKind getKind() {
+    return kind;
+  }
+
+  public long getSequenceNumber() {
+    return sequenceNumber;
+  }
+
+  @Override
+  public boolean equals(@Nullable Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof CdcRecord)) {
+      return false;
+    }
+    CdcRecord that = (CdcRecord) o;
+    return sequenceNumber == that.sequenceNumber && kind == that.kind && data.equals(that.data);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(data, kind, sequenceNumber);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(CdcRecord.class)
+        .add("data", data)
+        .add("kind", kind)
+        .add("sequenceNumber", sequenceNumber)
+        .toString();
+  }
+}
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoder.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoder.java
new file mode 100644
index 000000000000..d477fe100509
--- /dev/null
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoder.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.iceberg.cdc.sink;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.coders.ValueKindCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.ValueKind;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * {@link CdcRecord} carries a {@link Row} field whose schema is known only at pipeline-construction
+ * time. We need a custom coder because {@code AutoValueSchema} only infers schemas at the class
+ * level and cannot infer a dynamic {@link Row} field, so {@code @DefaultSchema} alone cannot
+ * produce a working coder for {@link CdcRecord}.
+ */
+final class CdcRecordCoder extends CustomCoder<CdcRecord> {
+
+  private final RowCoder dataCoder;
+  private final ValueKindCoder kindCoder = ValueKindCoder.of();
+  private final VarLongCoder seqCoder = VarLongCoder.of();
+
+  private CdcRecordCoder(Schema dataSchema) {
+    this.dataCoder = RowCoder.of(dataSchema);
+  }
+
+  public static CdcRecordCoder of(Schema dataSchema) {
+    return new CdcRecordCoder(dataSchema);
+  }
+
+  public Schema getDataSchema() {
+    return dataCoder.getSchema();
+  }
+
+  @Override
+  public void encode(CdcRecord value, OutputStream outStream) throws IOException {
+    dataCoder.encode(value.getData(), outStream);
+    kindCoder.encode(value.getKind(), outStream);
+    seqCoder.encode(value.getSequenceNumber(), outStream);
+  }
+
+  @Override
+  public CdcRecord decode(InputStream inStream) throws IOException {
+    Row data = dataCoder.decode(inStream);
+    ValueKind kind = kindCoder.decode(inStream);
+    long seq = seqCoder.decode(inStream);
+    return CdcRecord.of(data, kind, seq);
+  }
+
+  @Override
+  public void verifyDeterministic() throws NonDeterministicException {
+    Coder.verifyDeterministic(this, "Data coder must be deterministic", dataCoder);
+  }
+
+  @Override
+  public boolean consistentWithEquals() {
+    // decode() rebuilds the row with this coder's schema object, which may differ from the
+    // original row's; false is always safe.
+    return false;
+  }
+
+  @Override
+  public boolean equals(@Nullable Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    return getDataSchema().equals(((CdcRecordCoder) o).getDataSchema());
+  }
+
+  @Override
+  public int hashCode() {
+    return getDataSchema().hashCode();
+  }
+}
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKey.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKey.java
new file mode 100644
index 000000000000..21a297b38240
--- /dev/null
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKey.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.iceberg.cdc.sink;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import org.apache.beam.sdk.values.ValueKind;
+
+/**
+ * Builds the byte-comparable secondary sort key used by {@code SortValues} (extensions/sorter) to
+ * order each (destination, shard, window) group: one primary key's records come out contiguous,
+ * ordered by sequence number then {@link #kindRank(ValueKind)} within the key.
+ *
+ * <p>The key is {@code [pkLen:4][pkBytes][seq ^ Long.MIN_VALUE:8][kindRank:1]}, big-endian.
+ */
+final class CdcSortKey {
+
+  private CdcSortKey() {}
+
+  /** Ranks change kinds so before-images sort before after-images at an equal {@code seq}. */
+  public static byte kindRank(ValueKind kind) {
+    switch (kind) {
+      case UPDATE_BEFORE:
+        return 0;
+      case DELETE:
+        return 1;
+      case UPDATE_AFTER:
+        return 2;
+      case INSERT:
+        return 3;
+      default:
+        throw new IllegalArgumentException("Unknown ValueKind: " + kind);
+    }
+  }
+
+  /**
+   * Encodes the deterministic, byte-comparable sort key {@code [pkLen:4][pkBytes][seq ^
+   * Long.MIN_VALUE:8][kindRank:1]} for one CDC record.
+   *
+   * <p>SortValues compares unsigned lexicographic byte order. The length prefix is needed to
+   * accurately compare two primary keys of varying byte-lengths. Flipping the sequence number's
+   * sign bit makes unsigned byte order match signed numeric order. kindRank breaks equal-seq ties.
+   */
+  public static byte[] encode(byte[] pkBytes, long seq, ValueKind kind) {
+    return ByteBuffer.allocate(4 + pkBytes.length + 9)
+        .putInt(pkBytes.length)
+        .put(pkBytes)
+        .putLong(seq ^ Long.MIN_VALUE)
+        .put(kindRank(kind))
+        .array();
+  }
+
+  /**
+   * Whether two encoded sort keys carry the same primary key, compared on the raw {@code
+   * [pkLen:4][pkBytes]} prefix.
+   */
+  public static boolean samePk(byte[] a, byte[] b) {
+    int aPkEnd = 4 + ByteBuffer.wrap(a).getInt(0);
+    int bPkEnd = 4 + ByteBuffer.wrap(b).getInt(0);
+    return Arrays.equals(a, 0, aPkEnd, b, 0, bPkEnd);
+  }
+}
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java
new file mode 100644
index 000000000000..2bcd8b5f6222
--- /dev/null
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/package-info.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * A change data capture (CDC) write sink for Apache Iceberg: applies inserts, updates, and deletes
+ * to a table, identifying rows by equality columns.
+ */
+package org.apache.beam.sdk.io.iceberg.cdc.sink;
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemasTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemasTest.java
index 206e13acd0a6..7b4a22b2618e 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemasTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemasTest.java
@@ -22,15 +22,17 @@
 import static org.junit.Assert.assertEquals;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.io.iceberg.CollectDistinctSchemas.Group;
+import org.apache.beam.sdk.io.iceberg.CollectDistinctSchemas.SchemaGroup;
 import org.apache.beam.sdk.testing.CoderProperties;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.transforms.Combine;
 import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SchemaParser;
@@ -60,38 +62,72 @@ public class CollectDistinctSchemasTest {
               required(1, "id", Types.LongType.get()),
               optional(2, "name", Types.StringType.get())));
 
+  private static final List<String> NONE = Collections.emptyList();
+
   private final CollectDistinctSchemas fn = new CollectDistinctSchemas();
 
   @Test
   public void testDedupsIdenticalSchemas() {
-    assertEquals(Arrays.asList(KV.of(ID_NAME, 3L)), combine(ID_NAME, ID_NAME, ID_NAME));
+    assertEquals(
+        Arrays.asList(group(ID_NAME, 3, NONE)),
+        combine(group(ID_NAME, 1, NONE), group(ID_NAME, 1, NONE), group(ID_NAME, 1, NONE)));
   }
 
-  /** Inputs are compared as strings; canonicalization is ReadFooterSchema's job. */
+  /** Schemas are compared as strings; canonicalization is ReadFooterSchema's job. */
   @Test
   public void testDifferentStringsAreDistinct() {
-    List<KV<String, Long>> out = combine(ID_NAME, NAME_ID, ID_LONG_NAME);
+    List<SchemaGroup> out =
+        combine(group(ID_NAME, 1, NONE), group(NAME_ID, 1, NONE), group(ID_LONG_NAME, 1, NONE));
     assertEquals(3, out.size());
-    for (KV<String, Long> entry : out) {
-      assertEquals(Long.valueOf(1L), entry.getValue());
+    for (SchemaGroup entry : out) {
+      assertEquals(1L, entry.getFiles());
     }
   }
 
   @Test
   public void testMostCommonFirstThenJson() {
-    List<KV<String, Long>> out = combine(NAME_ID, ID_LONG_NAME, ID_NAME, ID_LONG_NAME, NAME_ID);
+    List<SchemaGroup> out =
+        combine(
+            group(NAME_ID, 1, NONE),
+            group(ID_LONG_NAME, 1, NONE),
+            group(ID_NAME, 1, NONE),
+            group(ID_LONG_NAME, 1, NONE),
+            group(NAME_ID, 1, NONE));
+    assertEquals(
+        Arrays.asList(
+            group(ID_LONG_NAME, 2, NONE), group(NAME_ID, 2, NONE), group(ID_NAME, 1, NONE)),
+        out);
+  }
+
+  /**
+   * A column counts as proven for the group only if every file proved it: one file with nulls in a
+   * column is enough to make the table relax that column.
+   */
+  @Test
+  public void testNullFreeColumnsIntersect() {
+    List<SchemaGroup> out =
+        combine(
+            group(ID_NAME, 1, Arrays.asList("id", "name")),
+            group(ID_NAME, 1, Arrays.asList("id")),
+            group(NAME_ID, 1, Arrays.asList("name")));
     assertEquals(
-        Arrays.asList(KV.of(ID_LONG_NAME, 2L), KV.of(NAME_ID, 2L), KV.of(ID_NAME, 1L)), out);
+        Arrays.asList(
+            group(ID_NAME, 2, Arrays.asList("id")), group(NAME_ID, 1, Arrays.asList("name"))),
+        out);
   }
 
   @Test
-  public void testMergeSumsCounts() {
-    Map<String, Long> first = fn.addInput(fn.createAccumulator(), ID_NAME);
-    Map<String, Long> second = fn.addInput(fn.createAccumulator(), ID_NAME);
-    second = fn.addInput(second, NAME_ID);
-    List<KV<String, Long>> out =
-        fn.extractOutput(fn.mergeAccumulators(Arrays.asList(first, second)));
-    assertEquals(Arrays.asList(KV.of(ID_NAME, 2L), KV.of(NAME_ID, 1L)), out);
+  public void testNullFreeColumnsIntersectAcrossMergedAccumulators() {
+    Map<String, Group> first =
+        fn.addInput(fn.createAccumulator(), group(ID_NAME, 1, Arrays.asList("id", "name")));
+    Map<String, Group> second =
+        fn.addInput(fn.createAccumulator(), group(ID_NAME, 1, Arrays.asList("name")));
+    second = fn.addInput(second, group(NAME_ID, 1, Arrays.asList("id")));
+    List<SchemaGroup> out = fn.extractOutput(fn.mergeAccumulators(Arrays.asList(first, second)));
+    assertEquals(
+        Arrays.asList(
+            group(ID_NAME, 2, Arrays.asList("name")), group(NAME_ID, 1, Arrays.asList("id"))),
+        out);
   }
 
   @Test
@@ -101,30 +137,51 @@ public void testEmptyInput() {
 
   @Test
   public void testAccumulatorCoderRoundTrip() throws Exception {
-    Coder<Map<String, Long>> coder = fn.getAccumulatorCoder(null, null);
-    Map<String, Long> accumulator = fn.addInput(fn.createAccumulator(), ID_NAME);
-    accumulator = fn.addInput(accumulator, NAME_ID);
+    Coder<Map<String, Group>> coder = fn.getAccumulatorCoder(null, null);
+    Map<String, Group> accumulator =
+        fn.addInput(fn.createAccumulator(), group(ID_NAME, 1, Arrays.asList("id")));
+    accumulator = fn.addInput(accumulator, group(NAME_ID, 1, NONE));
     CoderProperties.coderDecodeEncodeEqual(coder, accumulator);
   }
 
+  /** Coders from separate calls must compare equal, or coder inference treats them as different. */
+  @Test
+  public void testCodersFromSeparateCallsAreEqual() throws Exception {
+    assertEquals(CollectDistinctSchemas.outputCoder(), CollectDistinctSchemas.outputCoder());
+    assertEquals(fn.getAccumulatorCoder(null, null), fn.getAccumulatorCoder(null, null));
+    // The output coder is deterministic; the accumulator coder is not required to be (MapCoder).
+    CollectDistinctSchemas.outputCoder().verifyDeterministic();
+  }
+
   @Test
   public void testPipeline() {
-    PCollection<List<KV<String, Long>>> out =
+    PCollection<List<SchemaGroup>> out =
         pipeline
-            .apply(Create.of(ID_NAME, NAME_ID, ID_NAME))
+            .apply(
+                Create.of(
+                        group(ID_NAME, 1, Arrays.asList("id", "name")),
+                        group(NAME_ID, 1, NONE),
+                        group(ID_NAME, 1, Arrays.asList("id")))
+                    .withCoder(CollectDistinctSchemas.groupCoder()))
             .apply(Combine.globally(new CollectDistinctSchemas()));
-    PAssert.that(out).containsInAnyOrder(Arrays.asList(KV.of(ID_NAME, 2L), KV.of(NAME_ID, 1L)));
+    PAssert.that(out)
+        .containsInAnyOrder(
+            Arrays.asList(group(ID_NAME, 2, Arrays.asList("id")), group(NAME_ID, 1, NONE)));
     pipeline.run();
   }
 
-  private List<KV<String, Long>> combine(String... schemaJsons) {
-    Map<String, Long> accumulator = fn.createAccumulator();
-    for (String schemaJson : schemaJsons) {
-      accumulator = fn.addInput(accumulator, schemaJson);
+  private List<SchemaGroup> combine(SchemaGroup... files) {
+    Map<String, Group> accumulator = fn.createAccumulator();
+    for (SchemaGroup file : files) {
+      accumulator = fn.addInput(accumulator, file);
     }
     return fn.extractOutput(accumulator);
   }
 
+  private static SchemaGroup group(String json, long files, List<String> proven) {
+    return SchemaGroup.of(json, files, proven);
+  }
+
   private static String json(Schema schema) {
     return SchemaParser.toJson(schema);
   }
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/FileSchemasTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/FileSchemasTest.java
index 9723695186be..6a6bef483b73 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/FileSchemasTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/FileSchemasTest.java
@@ -20,17 +20,362 @@
 import static org.apache.iceberg.types.Types.NestedField.optional;
 import static org.apache.iceberg.types.Types.NestedField.required;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.IntPredicate;
+import org.apache.hadoop.fs.Path;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.parquet.ParquetSchemaUtil;
 import org.apache.iceberg.types.Types;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.EncodingStats;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnPath;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Type.Repetition;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
 @RunWith(JUnit4.class)
 public class FileSchemasTest {
+  @Rule public final TemporaryFolder tmp = new TemporaryFolder();
+
+  // ---- tightening
+
+  // root: required id, optional name, optional struct address {city, zip}, optional list tags
+  private static final MessageType MIXED =
+      org.apache.parquet.schema.Types.buildMessage()
+          .required(PrimitiveTypeName.INT64)
+          .named("id")
+          .optional(PrimitiveTypeName.BINARY)
+          .as(LogicalTypeAnnotation.stringType())
+          .named("name")
+          .addField(
+              org.apache.parquet.schema.Types.buildGroup(Repetition.OPTIONAL)
+                  .optional(PrimitiveTypeName.BINARY)
+                  .as(LogicalTypeAnnotation.stringType())
+                  .named("city")
+                  .optional(PrimitiveTypeName.INT32)
+                  .named("zip")
+                  .named("address"))
+          .addField(
+              org.apache.parquet.schema.Types.buildGroup(Repetition.OPTIONAL)
+                  .as(LogicalTypeAnnotation.listType())
+                  .addField(
+                      org.apache.parquet.schema.Types.repeatedGroup()
+                          .optional(PrimitiveTypeName.BINARY)
+                          .as(LogicalTypeAnnotation.stringType())
+                          .named("element")
+                          .named("list"))
+                  .named("tags"))
+          .named("root");
+
+  private static final class Nulls {
+    final IntPredicate name;
+    final IntPredicate address;
+    final IntPredicate city;
+    final IntPredicate zip;
+
+    Nulls(IntPredicate name, IntPredicate address, IntPredicate city, IntPredicate zip) {
+      this.name = name;
+      this.address = address;
+      this.city = city;
+      this.zip = zip;
+    }
+
+    static final Nulls NONE = new Nulls(r -> false, r -> false, r -> false, r -> false);
+  }
+
+  private ParquetMetadata write(int rows, int rowGroups, boolean stats, Nulls nulls)
+      throws IOException {
+    File file = new File(tmp.getRoot(), "t" + System.nanoTime() + ".parquet");
+    ExampleParquetWriter.Builder builder =
+        ExampleParquetWriter.builder(new Path(file.getAbsolutePath()))
+            .withType(MIXED)
+            .withStatisticsEnabled(stats);
+    if (rowGroups > 1) {
+      int rowsPerGroup = rows / rowGroups;
+      builder =
+          builder
+              .withRowGroupSize(1L)
+              .withMinRowCountForPageSizeCheck(rowsPerGroup)
+              .withMaxRowCountForPageSizeCheck(rowsPerGroup);
+    }
+    SimpleGroupFactory factory = new SimpleGroupFactory(MIXED);
+    try (ParquetWriter<Group> writer = builder.build()) {
+      for (int row = 0; row < rows; row++) {
+        Group group = factory.newGroup();
+        group.add("id", (long) row);
+        if (!nulls.name.test(row)) {
+          group.add("name", "n" + row);
+        }
+        if (!nulls.address.test(row)) {
+          Group address = group.addGroup("address");
+          if (!nulls.city.test(row)) {
+            address.add("city", "c" + row);
+          }
+          if (!nulls.zip.test(row)) {
+            address.add("zip", row);
+          }
+        }
+        Group tags = group.addGroup("tags");
+        tags.addGroup("list").add("element", "t" + row);
+        writer.write(group);
+      }
+    }
+    return ParquetFooters.read(file.getAbsolutePath());
+  }
+
+  private static Schema tightened(ParquetMetadata footer) {
+    return FileSchemas.tighten(
+        ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema()), footer);
+  }
+
+  private static boolean isRequired(Schema schema, String path) {
+    return schema.findField(path).isRequired();
+  }
+
+  @Test
+  public void testProvenNullFreeColumnsBecomeRequired() throws IOException {
+    Schema schema = tightened(write(10, 1, true, Nulls.NONE));
+    assertTrue(isRequired(schema, "id"));
+    assertTrue(isRequired(schema, "name"));
+    assertTrue(isRequired(schema, "address"));
+    assertTrue(isRequired(schema, "address.city"));
+    assertTrue(isRequired(schema, "address.zip"));
+  }
+
+  @Test
+  public void testListAndElementStayAsDeclared() throws IOException {
+    Schema schema = tightened(write(10, 1, true, Nulls.NONE));
+    assertFalse(isRequired(schema, "tags"));
+    assertFalse(schema.findField("tags").type().asListType().isElementRequired());
+  }
+
+  @Test
+  public void testSomeNullsStayOptional() throws IOException {
+    Schema schema =
+        tightened(write(10, 1, true, new Nulls(r -> r == 3, r -> false, r -> false, r -> false)));
+    assertFalse(isRequired(schema, "name"));
+    assertTrue(isRequired(schema, "address.city"));
+  }
+
+  @Test
+  public void testAllNullsStayOptional() throws IOException {
+    Schema schema =
+        tightened(write(10, 1, true, new Nulls(r -> true, r -> false, r -> false, r -> false)));
+    assertFalse(isRequired(schema, "name"));
+  }
+
+  @Test
+  public void testStatsDisabledStaysOptional() throws IOException {
+    Schema schema = tightened(write(10, 1, false, Nulls.NONE));
+    assertTrue(isRequired(schema, "id"));
+    assertFalse(isRequired(schema, "name"));
+    assertFalse(isRequired(schema, "address"));
+    assertFalse(isRequired(schema, "address.city"));
+  }
+
+  @Test
+  public void testOneRowGroupWithNullsSpoilsTheProof() throws IOException {
+    ParquetMetadata footer =
+        write(100, 4, true, new Nulls(r -> r == 60, r -> false, r -> false, r -> false));
+    assertTrue("expected several row groups", footer.getBlocks().size() > 1);
+
+    Schema schema = tightened(footer);
+
+    assertFalse(isRequired(schema, "name"));
+    assertTrue(isRequired(schema, "address.zip"));
+  }
+
+  /** With no rows nothing can violate a required column, so every column counts as proven. */
+  @Test
+  public void testZeroRowsProveEverything() throws IOException {
+    ParquetMetadata footer = write(0, 1, true, Nulls.NONE);
+    assertEquals("parquet-mr writes no row group for zero rows", 0, footer.getBlocks().size());
+
+    Schema schema = tightened(footer);
+
+    assertTrue(isRequired(schema, "id"));
+    assertTrue(isRequired(schema, "name"));
+    assertTrue(isRequired(schema, "address"));
+    assertTrue(isRequired(schema, "address.city"));
+    assertFalse(isRequired(schema, "tags"));
+  }
+
+  /**
+   * pyarrow writes an empty table as one row group with zero rows and no statistics. parquet-mr
+   * never produces that shape, so the footer is assembled by hand.
+   */
+  @Test
+  public void testEmptyRowGroupWithoutStatsProvesEverything() throws IOException {
+    ParquetMetadata footer = withBlocks(write(0, 1, true, Nulls.NONE), emptyBlockWithoutStats());
+
+    Schema schema = tightened(footer);
+
+    assertTrue(isRequired(schema, "id"));
+    assertTrue(isRequired(schema, "name"));
+    assertTrue(isRequired(schema, "address"));
+    assertTrue(isRequired(schema, "address.city"));
+    assertFalse(isRequired(schema, "tags"));
+  }
+
+  @Test
+  public void testEmptyRowGroupDoesNotSpoilTheProof() throws IOException {
+    ParquetMetadata written = write(10, 1, true, Nulls.NONE);
+    ParquetMetadata footer =
+        withBlocks(written, written.getBlocks().get(0), emptyBlockWithoutStats());
+
+    Schema schema = tightened(footer);
+
+    assertTrue(isRequired(schema, "name"));
+    assertTrue(isRequired(schema, "address.city"));
+  }
+
+  @Test
+  public void testEmptyRowGroupDoesNotHideNullsElsewhere() throws IOException {
+    ParquetMetadata written =
+        write(10, 1, true, new Nulls(r -> r == 3, r -> false, r -> false, r -> false));
+    ParquetMetadata footer =
+        withBlocks(written, emptyBlockWithoutStats(), written.getBlocks().get(0));
+
+    Schema schema = tightened(footer);
+
+    assertFalse(isRequired(schema, "name"));
+    assertTrue(isRequired(schema, "address.city"));
+  }
+
+  private static ParquetMetadata withBlocks(ParquetMetadata footer, BlockMetaData... blocks) {
+    List<BlockMetaData> list = new ArrayList<>();
+    Collections.addAll(list, blocks);
+    return new ParquetMetadata(footer.getFileMetaData(), list);
+  }
+
+  /** One chunk per leaf of {@link #MIXED}, zero rows, statistics absent as a reader sees them. */
+  private static BlockMetaData emptyBlockWithoutStats() {
+    BlockMetaData block = new BlockMetaData();
+    block.setRowCount(0);
+    for (ColumnDescriptor column : MIXED.getColumns()) {
+      block.addColumn(
+          ColumnChunkMetaData.get(
+              ColumnPath.get(column.getPath()),
+              column.getPrimitiveType(),
+              CompressionCodecName.UNCOMPRESSED,
+              new EncodingStats.Builder().build(),
+              Collections.emptySet(),
+              Statistics.getBuilderForReading(column.getPrimitiveType()).build(),
+              0,
+              0,
+              0,
+              0,
+              0));
+    }
+    return block;
+  }
+
+  @Test
+  public void testCanonicalWithNullFreeColumnsReportsChangedOnly() throws IOException {
+    ParquetMetadata footer =
+        write(10, 1, true, new Nulls(r -> false, r -> false, r -> false, r -> r == 4));
+    CollectDistinctSchemas.SchemaGroup group = FileSchemas.schemaGroup(footer);
+    // id is declared required already; zip has a null; the rest flipped
+    assertEquals(
+        java.util.Arrays.asList("address", "address.city", "name"), group.getNullFreeColumns());
+    Schema declared = SchemaParser.fromJson(group.getSchemaJson());
+    assertFalse(isRequired(declared, "name"));
+  }
+
+  @Test
+  public void testMarkRequiredFlipsOnlyNamedColumns() {
+    Schema declared =
+        new Schema(
+            optional(1, "name", Types.StringType.get()),
+            optional(
+                2,
+                "address",
+                Types.StructType.of(
+                    optional(3, "city", Types.StringType.get()),
+                    optional(4, "zip", Types.IntegerType.get()))));
+    Schema required =
+        FileSchemas.markRequired(
+            declared, java.util.Arrays.asList("address", "address.city", "not_a_column"));
+    assertFalse(isRequired(required, "name"));
+    assertTrue(isRequired(required, "address"));
+    assertTrue(isRequired(required, "address.city"));
+    assertFalse(isRequired(required, "address.zip"));
+    assertEquals(
+        declared.asStruct(),
+        FileSchemas.markRequired(declared, java.util.Arrays.asList()).asStruct());
+  }
+
+  @Test
+  public void testNullStructKeepsStructAndLeavesOptional() throws IOException {
+    Schema schema =
+        tightened(write(10, 1, true, new Nulls(r -> false, r -> r == 5, r -> false, r -> false)));
+    assertFalse(isRequired(schema, "address"));
+    assertFalse(isRequired(schema, "address.city"));
+    assertFalse(isRequired(schema, "address.zip"));
+  }
+
+  @Test
+  public void testOneProvenLeafProvesTheStruct() throws IOException {
+    Schema schema =
+        tightened(write(10, 1, true, new Nulls(r -> false, r -> false, r -> r == 2, r -> false)));
+    assertTrue(isRequired(schema, "address"));
+    assertFalse(isRequired(schema, "address.city"));
+    assertTrue(isRequired(schema, "address.zip"));
+  }
+
+  @Test
+  public void testTightenPreservesIdsNamesAndTypes() throws IOException {
+    ParquetMetadata footer = write(10, 1, true, Nulls.NONE);
+    Schema converted = ParquetSchemaUtil.convert(footer.getFileMetaData().getSchema());
+    Schema schema = FileSchemas.tighten(converted, footer);
+    assertEquals(converted.columns().size(), schema.columns().size());
+    for (Types.NestedField field : converted.columns()) {
+      Types.NestedField after = schema.findField(field.fieldId());
+      assertEquals(field.name(), after.name());
+      assertEquals(field.type().typeId(), after.type().typeId());
+    }
+  }
+
+  @Test
+  public void testTightenAndCanonicalPreserveDocAndDefaults() {
+    Types.NestedField withAttributes =
+        Types.NestedField.optional("b")
+            .withId(2)
+            .ofType(Types.LongType.get())
+            .withDoc("the b")
+            .withWriteDefault(org.apache.iceberg.expressions.Literal.of(7L))
+            .build();
+    Schema schema = new Schema(withAttributes, required(1, "a", Types.StringType.get()));
+    Schema canonical = FileSchemas.canonical(schema);
+    Types.NestedField b = canonical.findField("b");
+    assertEquals("the b", b.doc());
+    assertEquals(7L, b.writeDefault());
+  }
+
+  // ---- canonicalization
 
   @Test
   public void testSortsTopLevelFieldsAndRenumbers() {
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java
index 6b3679913918..5cac095201db 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java
@@ -164,6 +164,38 @@ public void testNonParquetEmitsNothing() throws IOException {
     pipeline.run();
   }
 
+  /**
+   * The schema is emitted exactly as declared; which columns proved null-free travels beside it.
+   */
+  @Test
+  public void testNullFreeColumnsAreReported() throws IOException {
+    Record full = GenericRecord.create(FLAT_SCHEMA).copy("id", 1, "name", "a");
+    String clean = writeParquet("clean.parquet", FLAT_SCHEMA, full);
+    PCollection<CollectDistinctSchemas.SchemaGroup> out = run(clean);
+    assertSchemas(out, FLAT_SCHEMA);
+    assertNullFreeColumns(out, Arrays.asList("name"));
+    pipeline.run();
+  }
+
+  @Test
+  public void testColumnWithNullsIsNotReportedNullFree() throws IOException {
+    String withNull = writeParquet("null.parquet", FLAT_SCHEMA, record(FLAT_SCHEMA, "id", 1));
+    PCollection<CollectDistinctSchemas.SchemaGroup> out = run(withNull);
+    assertSchemas(out, FLAT_SCHEMA);
+    assertNullFreeColumns(out, Arrays.asList());
+    pipeline.run();
+  }
+
+  private static void assertNullFreeColumns(
+      PCollection<CollectDistinctSchemas.SchemaGroup> out, List<String> expected) {
+    PAssert.thatSingleton(out)
+        .satisfies(
+            group -> {
+              assertEquals(expected, group.getNullFreeColumns());
+              return null;
+            });
+  }
+
   @Test
   public void testPermutedColumnsProduceIdenticalSchema() throws IOException {
     Schema permuted =
@@ -177,7 +209,7 @@ public void testPermutedColumnsProduceIdenticalSchema() throws IOException {
         .satisfies(
             actual -> {
               List<String> jsons = new ArrayList<>();
-              actual.forEach(jsons::add);
+              actual.forEach(group -> jsons.add(group.getSchemaJson()));
               assertEquals(2, jsons.size());
               assertEquals(jsons.get(0), jsons.get(1));
               return null;
@@ -218,12 +250,16 @@ private static long counter(PipelineResult result, String name) {
     return total;
   }
 
-  private PCollection<String> run(String... paths) {
-    return pipeline.apply(Create.of(Arrays.asList(paths))).apply(ParDo.of(new ReadFooterSchema()));
+  private PCollection<CollectDistinctSchemas.SchemaGroup> run(String... paths) {
+    return pipeline
+        .apply(Create.of(Arrays.asList(paths)))
+        .apply(ParDo.of(new ReadFooterSchema()))
+        .setCoder(CollectDistinctSchemas.groupCoder());
   }
 
-  /** Asserts the emitted schemas equal the canonical forms of {@code expected}, in any order. */
-  private static void assertSchemas(PCollection<String> out, Schema... expected) {
+  /** Asserts the emitted declared schemas equal the canonical forms of {@code expected}. */
+  private static void assertSchemas(
+      PCollection<CollectDistinctSchemas.SchemaGroup> out, Schema... expected) {
     List<String> expectedJson = new ArrayList<>();
     for (Schema schema : expected) {
       expectedJson.add(SchemaParser.toJson(FileSchemas.canonical(schema)));
@@ -232,7 +268,8 @@ private static void assertSchemas(PCollection<String> out, Schema... expected) {
         .satisfies(
             actual -> {
               List<String> remaining = new ArrayList<>(expectedJson);
-              for (String json : actual) {
+              for (CollectDistinctSchemas.SchemaGroup group : actual) {
+                String json = group.getSchemaJson();
                 Schema schema = SchemaParser.fromJson(json);
                 boolean matched = false;
                 for (int i = 0; i < remaining.size(); i++) {
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFileTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFileTest.java
index 29ef30c97efb..02e6fec46578 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFileTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableDeleteFileTest.java
@@ -19,6 +19,12 @@
 
 import static java.util.Collections.emptyMap;
 import static java.util.Collections.singletonMap;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.nullValue;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
@@ -30,6 +36,9 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
+import org.apache.beam.sdk.schemas.SchemaCoder;
+import org.apache.beam.sdk.schemas.SchemaRegistry;
+import org.apache.beam.sdk.util.CoderUtils;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileContent;
 import org.apache.iceberg.FileFormat;
@@ -37,10 +46,16 @@
 import org.apache.iceberg.Metrics;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.types.Types;
+import org.checkerframework.checker.nullness.qual.Nullable;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
 
 /** Tests for {@link SerializableDeleteFile}. */
+@RunWith(JUnit4.class)
 public class SerializableDeleteFileTest {
   private static final org.apache.iceberg.Schema SCHEMA =
       new org.apache.iceberg.Schema(
@@ -85,7 +100,7 @@ public void testPositionDeleteRoundTripPreservesMetadataUsedByCdcReads() throws
             .build();
     setSequenceNumbers(deleteFile, 44L, 45L);
 
-    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, "category=A", true);
+    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, SPEC, true);
     DeleteFile reconstructed =
         serialized.createDeleteFile(
             singletonMap(SPEC.specId(), SPEC), singletonMap(0, SortOrder.unsorted()));
@@ -125,7 +140,7 @@ public void testEqualityDeleteRoundTripPreservesFieldIdsAndSortOrder() {
             .withRecordCount(2L)
             .build();
 
-    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, "category=A", true);
+    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, SPEC, true);
     DeleteFile reconstructed =
         serialized.createDeleteFile(singletonMap(SPEC.specId(), SPEC), singletonMap(7, sortOrder));
 
@@ -149,7 +164,7 @@ public void testPuffinDeleteRoundTripPreservesDeletionVectorMetadata() {
             .withReferencedDataFile("gs://bucket/data/category=A/data.parquet")
             .build();
 
-    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, "category=A", true);
+    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, SPEC, true);
     DeleteFile reconstructed =
         serialized.createDeleteFile(
             singletonMap(SPEC.specId(), SPEC), singletonMap(0, SortOrder.unsorted()));
@@ -160,9 +175,11 @@ public void testPuffinDeleteRoundTripPreservesDeletionVectorMetadata() {
     assertEquals("gs://bucket/data/category=A/data.parquet", reconstructed.referencedDataFile());
   }
 
+  /** Reconstruction fails clearly when the spec map or the sort-order map lacks the file's id. */
   @Test
-  public void testCreateDeleteFileFailsClearlyForMissingPartitionSpec() {
-    DeleteFile deleteFile =
+  public void testCreateDeleteFileFailsClearlyForMissingSpecOrSortOrder() {
+    // facet: missing partition spec.
+    DeleteFile positionDelete =
         FileMetadata.deleteFileBuilder(SPEC)
             .ofPositionDeletes()
             .withPath("gs://bucket/deletes/category=A/pos.parquet")
@@ -171,19 +188,17 @@ public void testCreateDeleteFileFailsClearlyForMissingPartitionSpec() {
             .withFileSizeInBytes(256L)
             .withRecordCount(2L)
             .build();
-    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, "category=A", true);
-
-    IllegalStateException thrown =
+    SerializableDeleteFile serializedPosition =
+        SerializableDeleteFile.from(positionDelete, SPEC, true);
+    IllegalStateException missingSpec =
         assertThrows(
-            IllegalStateException.class, () -> serialized.createDeleteFile(emptyMap(), null));
-
-    assertTrue(thrown.getMessage().contains("created with spec id '" + SPEC.specId() + "'"));
-  }
+            IllegalStateException.class,
+            () -> serializedPosition.createDeleteFile(emptyMap(), null));
+    assertTrue(missingSpec.getMessage().contains("created with spec id '" + SPEC.specId() + "'"));
 
-  @Test
-  public void testCreateEqualityDeleteFileFailsClearlyForMissingSortOrder() {
+    // facet: missing sort order (equality delete).
     SortOrder sortOrder = SortOrder.builderFor(SCHEMA).asc("id").withOrderId(7).build();
-    DeleteFile deleteFile =
+    DeleteFile equalityDelete =
         FileMetadata.deleteFileBuilder(SPEC)
             .ofEqualityDeletes(1)
             .withSortOrder(sortOrder)
@@ -193,14 +208,242 @@ public void testCreateEqualityDeleteFileFailsClearlyForMissingSortOrder() {
             .withFileSizeInBytes(256L)
             .withRecordCount(2L)
             .build();
-    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, "category=A", true);
+    SerializableDeleteFile serializedEquality =
+        SerializableDeleteFile.from(equalityDelete, SPEC, true);
+    IllegalStateException missingOrder =
+        assertThrows(
+            IllegalStateException.class,
+            () ->
+                serializedEquality.createDeleteFile(singletonMap(SPEC.specId(), SPEC), emptyMap()));
+    assertTrue(missingOrder.getMessage().contains("sort order id '7'"));
+  }
 
-    IllegalStateException thrown =
+  /**
+   * A {@link DeleteFile} must be serialized with the EXACT spec it was written with: a mismatched
+   * spec id and a spec id absent from the map each fail loudly.
+   */
+  @Test
+  public void fromRejectsMismatchedOrUnknownSpec() {
+    // facet: single-spec overload, wrong spec.
+    DeleteFile unpartitioned =
+        FileMetadata.deleteFileBuilder(PartitionSpec.unpartitioned())
+            .ofPositionDeletes()
+            .withPath("gs://bucket/deletes/pos.parquet")
+            .withFormat(FileFormat.PARQUET)
+            .withFileSizeInBytes(1L)
+            .withRecordCount(1L)
+            .build();
+    PartitionSpec otherSpec =
+        PartitionSpec.builderFor(SCHEMA).identity("category").withSpecId(1).build();
+    IllegalArgumentException mismatch =
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> SerializableDeleteFile.from(unpartitioned, otherSpec));
+    assertThat(mismatch.getMessage(), containsString("does not match"));
+
+    // facet: spec-map overload, id missing from the map.
+    DeleteFile deleteFile = positionDeletes(SPEC, partition(SPEC, "A"));
+    IllegalStateException unknown =
         assertThrows(
             IllegalStateException.class,
-            () -> serialized.createDeleteFile(singletonMap(SPEC.specId(), SPEC), emptyMap()));
+            () -> SerializableDeleteFile.from(deleteFile, emptyMap(), true));
+    assertThat(unknown.getMessage(), containsString("partition spec id '0'"));
+  }
+
+  /**
+   * All three delete-file kinds — equality, position, and a V3 deletion vector — round-trip through
+   * the schema coder with their partition tuples intact.
+   */
+  @Test
+  public void allDeleteFileKindsRoundTripThroughSchemaCoderWithPartition() throws Exception {
+    // facet: equality delete.
+    SortOrder sortOrder = SortOrder.builderFor(SCHEMA).asc("id").withOrderId(7).build();
+    DeleteFile equalityDelete =
+        FileMetadata.deleteFileBuilder(SPEC)
+            .ofEqualityDeletes(1, 2)
+            .withSortOrder(sortOrder)
+            .withPath("gs://bucket/deletes/category=A/eq.parquet")
+            .withFormat(FileFormat.PARQUET)
+            .withPartition(partition(SPEC, "A"))
+            .withFileSizeInBytes(256L)
+            .withRecordCount(2L)
+            .build();
+    DeleteFile equalityReconstructed =
+        encodeDecode(SerializableDeleteFile.from(equalityDelete, SPEC))
+            .createDeleteFile(singletonMap(SPEC.specId(), SPEC), singletonMap(7, sortOrder));
+    assertEquals(FileContent.EQUALITY_DELETES, equalityReconstructed.content());
+    assertEquals(equalityDelete.partition(), equalityReconstructed.partition());
+    assertEquals("category=A", SPEC.partitionToPath(equalityReconstructed.partition()));
+
+    // facet: position delete.
+    DeleteFile positionDelete = positionDeletes(SPEC, partition(SPEC, "A"));
+    DeleteFile positionReconstructed =
+        encodeDecode(SerializableDeleteFile.from(positionDelete, SPEC))
+            .createDeleteFile(singletonMap(SPEC.specId(), SPEC), null);
+    assertEquals(FileContent.POSITION_DELETES, positionReconstructed.content());
+    assertEquals(positionDelete.partition(), positionReconstructed.partition());
+    assertEquals("category=A", SPEC.partitionToPath(positionReconstructed.partition()));
+
+    // facet: V3 deletion vector (a Puffin blob with offset/size/referenced-data-file).
+    DeleteFile dv =
+        FileMetadata.deleteFileBuilder(SPEC)
+            .ofPositionDeletes()
+            .withPath("gs://bucket/deletes/category=A/dv.puffin")
+            .withFormat(FileFormat.PUFFIN)
+            .withPartition(partition(SPEC, "A"))
+            .withFileSizeInBytes(512L)
+            .withRecordCount(1L)
+            .withContentOffset(64L)
+            .withContentSizeInBytes(128L)
+            .withReferencedDataFile("gs://bucket/data/category=A/data.parquet")
+            .build();
+    DeleteFile dvReconstructed =
+        encodeDecode(SerializableDeleteFile.from(dv, SPEC))
+            .createDeleteFile(singletonMap(SPEC.specId(), SPEC), null);
+    assertEquals(FileFormat.PUFFIN, dvReconstructed.format());
+    assertEquals(Long.valueOf(64L), dvReconstructed.contentOffset());
+    assertEquals(Long.valueOf(128L), dvReconstructed.contentSizeInBytes());
+    assertEquals("gs://bucket/data/category=A/data.parquet", dvReconstructed.referencedDataFile());
+    assertEquals(dv.partition(), dvReconstructed.partition());
+    assertEquals("category=A", SPEC.partitionToPath(dvReconstructed.partition()));
+  }
+
+  /**
+   * An identity partition on a {@code timestamptz} column: {@link
+   * org.apache.iceberg.types.Conversions#fromPartitionString} has no case for TIMESTAMP at all, so
+   * the old partition-path round-trip blew up at reconstruct time. The JSON representation carries
+   * the raw micros and round-trips exactly.
+   */
+  @Test
+  public void timestampPartitionRoundTripsThroughJsonButNotThroughPartitionPath() {
+    org.apache.iceberg.Schema schema =
+        new org.apache.iceberg.Schema(
+            Types.NestedField.required(1, "id", Types.IntegerType.get()),
+            Types.NestedField.required(2, "event_time", Types.TimestampType.withZone()));
+    PartitionSpec spec = PartitionSpec.builderFor(schema).identity("event_time").build();
+    long micros = 1_709_618_828_000_009L;
+    GenericRecord partition = GenericRecord.create(spec.partitionType());
+    partition.setField("event_time", micros);
+    DeleteFile deleteFile = positionDeletes(spec, partition);
+
+    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, spec);
+    DeleteFile reconstructed = serialized.createDeleteFile(singletonMap(spec.specId(), spec), null);
+
+    assertEquals(Long.valueOf(micros), reconstructed.partition().get(0, Long.class));
+    assertEquals(deleteFile.partition(), reconstructed.partition());
+
+    // The pre-change wire shape (partition path only) cannot reconstruct this partition at all.
+    SerializableDeleteFile legacy = withoutJsonPartition(serialized);
+    assertThrows(
+        UnsupportedOperationException.class,
+        () -> legacy.createDeleteFile(singletonMap(spec.specId(), spec), null));
+  }
+
+  /**
+   * {@code PartitionSpec.partitionToPath} URL-encodes each value but {@code DataFiles.fillFromPath}
+   * never decodes it, so a string partition containing {@code / }, {@code &} or {@code =} used to
+   * come back SILENTLY WRONG — the delete would be registered under a {@code (specId, partition)}
+   * that no data file lives in, and would simply never apply. The JSON representation is exact.
+   */
+  @Test
+  public void stringPartitionWithSpecialCharactersRoundTripsExactly() {
+    String value = "a/b c&d=e";
+    DeleteFile deleteFile = positionDeletes(SPEC, partition(SPEC, value));
+
+    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, SPEC);
+    DeleteFile reconstructed = serialized.createDeleteFile(singletonMap(SPEC.specId(), SPEC), null);
+
+    assertEquals(value, reconstructed.partition().get(0, CharSequence.class).toString());
+    assertEquals(deleteFile.partition(), reconstructed.partition());
+
+    // Prove the old representation was silently lossy rather than merely throwing.
+    DeleteFile viaLegacyPath =
+        withoutJsonPartition(serialized).createDeleteFile(singletonMap(SPEC.specId(), SPEC), null);
+    assertThat(
+        viaLegacyPath.partition().get(0, CharSequence.class).toString(), not(equalTo(value)));
+  }
+
+  /**
+   * A null partition value is rendered as the literal text {@code null} in a partition path, which
+   * {@code fromPartitionString} hands back as the four-character string "null" for a string column
+   * — again silently wrong. JSON omits the field and it decodes back to a real null.
+   */
+  @Test
+  public void nullPartitionValueRoundTripsAsNull() {
+    DeleteFile deleteFile = positionDeletes(SPEC, partition(SPEC, null));
+
+    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, SPEC);
+    DeleteFile reconstructed = serialized.createDeleteFile(singletonMap(SPEC.specId(), SPEC), null);
+
+    assertThat(reconstructed.partition().get(0, CharSequence.class), nullValue());
+    assertEquals(deleteFile.partition(), reconstructed.partition());
+
+    // The old path turns the null into the literal string "null".
+    DeleteFile viaLegacyPath =
+        withoutJsonPartition(serialized).createDeleteFile(singletonMap(SPEC.specId(), SPEC), null);
+    assertEquals("null", viaLegacyPath.partition().get(0, CharSequence.class).toString());
+  }
+
+  /**
+   * NaN / Infinity floating-point partition values don't round-trip through the JSON partition
+   * representation ({@code SingleValueParser.fromJson} rejects the non-standard {@code NaN} token).
+   * Reconstruct must fall back to the partition-path string, which handles them, rather than
+   * crash-looping the sink at commit time.
+   */
+  @Test
+  public void nanFloatPartitionReconstructsViaPathFallback() {
+    org.apache.iceberg.Schema schema =
+        new org.apache.iceberg.Schema(
+            Types.NestedField.required(1, "f", Types.FloatType.get()),
+            Types.NestedField.optional(2, "data", Types.StringType.get()));
+    PartitionSpec spec = PartitionSpec.builderFor(schema).identity("f").build();
+    GenericRecord partition = GenericRecord.create(spec.partitionType());
+    partition.setField("f", Float.NaN);
+    DeleteFile deleteFile = positionDeletes(spec, partition);
+
+    SerializableDeleteFile serialized = SerializableDeleteFile.from(deleteFile, spec);
+    // Must reconstruct without throwing (JSON decode of NaN fails -> partition-path fallback).
+    DeleteFile reconstructed = serialized.createDeleteFile(singletonMap(spec.specId(), spec), null);
+
+    Object value = reconstructed.partition().get(0, Object.class);
+    assertThat(value, instanceOf(Float.class));
+    assertTrue("partition value must round-trip as NaN", Float.isNaN((Float) value));
+  }
+
+  private static GenericRecord partition(PartitionSpec spec, @Nullable Object value) {
+    GenericRecord record = GenericRecord.create(spec.partitionType());
+    record.set(0, value);
+    return record;
+  }
+
+  private static DeleteFile positionDeletes(PartitionSpec spec, StructLike partition) {
+    return FileMetadata.deleteFileBuilder(spec)
+        .ofPositionDeletes()
+        .withPath("gs://bucket/deletes/pos.parquet")
+        .withFormat(FileFormat.PARQUET)
+        .withPartition(partition)
+        .withFileSizeInBytes(256L)
+        .withRecordCount(2L)
+        .build();
+  }
+
+  /** Rebuilds the element as a pre-jsonPartition release would have encoded it. */
+  private static SerializableDeleteFile withoutJsonPartition(SerializableDeleteFile file) {
+    return SerializableDeleteFile.builder()
+        .setContentType(file.getContentType())
+        .setLocation(file.getLocation())
+        .setFileFormat(file.getFileFormat())
+        .setRecordCount(file.getRecordCount())
+        .setFileSizeInBytes(file.getFileSizeInBytes())
+        .setPartitionPath(file.getPartitionPath())
+        .setPartitionSpecId(file.getPartitionSpecId())
+        .build();
+  }
 
-    assertTrue(thrown.getMessage().contains("sort order id '7'"));
+  private static SerializableDeleteFile encodeDecode(SerializableDeleteFile file) throws Exception {
+    SchemaCoder<SerializableDeleteFile> coder =
+        SchemaRegistry.createDefault().getSchemaCoder(SerializableDeleteFile.class);
+    return CoderUtils.decodeFromByteArray(coder, CoderUtils.encodeToByteArray(coder, file));
   }
 
   private static void setSequenceNumbers(
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java
index 75bfaa1775da..8c23a844be6a 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtilsTest.java
@@ -360,10 +360,7 @@ private static SerializableChangelogTask task(
   private static List<SerializableDeleteFile> serializableDeletes(
       List<DeleteFile> deletes, Table table) {
     return deletes.stream()
-        .map(
-            delete ->
-                SerializableDeleteFile.from(
-                    delete, table.spec().partitionToPath(delete.partition()), true))
+        .map(delete -> SerializableDeleteFile.from(delete, table.specs(), true))
         .collect(Collectors.toList());
   }
 
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ReadFromChangelogsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ReadFromChangelogsTest.java
index 7a1e71d3d80b..2de2848f9363 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ReadFromChangelogsTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/ReadFromChangelogsTest.java
@@ -316,10 +316,7 @@ private static SerializableChangelogTask task(
   private static List<SerializableDeleteFile> serializableDeletes(
       List<DeleteFile> deletes, Table table) {
     return deletes.stream()
-        .map(
-            delete ->
-                SerializableDeleteFile.from(
-                    delete, table.spec().partitionToPath(delete.partition()), true))
+        .map(delete -> SerializableDeleteFile.from(delete, table.specs(), true))
         .collect(Collectors.toList());
   }
 
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoderTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoderTest.java
new file mode 100644
index 000000000000..a53b98a10aa9
--- /dev/null
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoderTest.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.iceberg.cdc.sink;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.RowCoder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.CoderProperties;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.ValueKind;
+import org.hamcrest.Matchers;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link CdcRecordCoder}. */
+@RunWith(JUnit4.class)
+public class CdcRecordCoderTest {
+
+  private static final Schema DATA_SCHEMA =
+      Schema.builder().addInt32Field("id").addStringField("name").addStringField("data").build();
+
+  @Test
+  public void roundTripsAllValueKinds() throws Exception {
+    for (ValueKind kind : ValueKind.values()) {
+      CdcRecord record =
+          CdcRecord.of(Row.withSchema(DATA_SCHEMA).addValues(1, "a", "x").build(), kind, 12L);
+      CoderProperties.coderDecodeEncodeEqual(CdcRecordCoder.of(DATA_SCHEMA), record);
+    }
+  }
+
+  /** The coder's own properties: deterministic, serializable, schema-keyed equality. */
+  @Test
+  public void coderIsDeterministicSerializableAndSchemaKeyed() throws Exception {
+    CdcRecordCoder.of(DATA_SCHEMA).verifyDeterministic();
+    CoderProperties.coderSerializable(CdcRecordCoder.of(DATA_SCHEMA));
+    // decode() rebuilds the row with the coder's schema object, which may differ from the
+    // original row's, so the coder must not claim consistency with equals.
+    assertFalse(CdcRecordCoder.of(DATA_SCHEMA).consistentWithEquals());
+    assertThat(CdcRecordCoder.of(DATA_SCHEMA).getDataSchema(), Matchers.equalTo(DATA_SCHEMA));
+
+    Schema otherSchema = Schema.builder().addInt32Field("other").build();
+    assertThat(CdcRecordCoder.of(DATA_SCHEMA), Matchers.equalTo(CdcRecordCoder.of(DATA_SCHEMA)));
+    assertThat(
+        CdcRecordCoder.of(DATA_SCHEMA).hashCode(),
+        Matchers.equalTo(CdcRecordCoder.of(DATA_SCHEMA).hashCode()));
+    assertThat(
+        CdcRecordCoder.of(DATA_SCHEMA),
+        Matchers.not(Matchers.equalTo(CdcRecordCoder.of(otherSchema))));
+  }
+
+  @Test
+  public void encodePinnedWireMapping() throws Exception {
+    // The stream layout is [row bytes][kind VarInt][seq VarLong]. Codes 0-3 each fit in a single
+    // VarInt byte equal to the code itself, so the byte immediately following the row bytes must
+    // be the pinned code for that ValueKind.
+    Row row = Row.withSchema(DATA_SCHEMA).addValues(1, "a", "x").build();
+    ByteArrayOutputStream rowOnly = new ByteArrayOutputStream();
+    RowCoder.of(DATA_SCHEMA).encode(row, rowOnly);
+    int rowLen = rowOnly.toByteArray().length;
+
+    assertPinnedKindCode(row, ValueKind.INSERT, rowLen, 0);
+    assertPinnedKindCode(row, ValueKind.UPDATE_BEFORE, rowLen, 1);
+    assertPinnedKindCode(row, ValueKind.UPDATE_AFTER, rowLen, 2);
+    assertPinnedKindCode(row, ValueKind.DELETE, rowLen, 3);
+  }
+
+  private static void assertPinnedKindCode(Row row, ValueKind kind, int rowLen, int expectedCode)
+      throws Exception {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    CdcRecordCoder.of(DATA_SCHEMA).encode(CdcRecord.of(row, kind, 1L), out);
+    byte[] bytes = out.toByteArray();
+    assertThat(bytes[rowLen] & 0xFF, Matchers.equalTo(expectedCode));
+  }
+
+  @Test
+  public void decodeRejectsUnknownKindCode() throws Exception {
+    // Hand-encode a stream with a valid data row and seq, but a ValueKind code (4) at the pinned
+    // mapping's boundary: one past the highest valid code (3).
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    RowCoder.of(DATA_SCHEMA)
+        .encode(Row.withSchema(DATA_SCHEMA).addValues(1, "a", "x").build(), out);
+    VarIntCoder.of().encode(4, out);
+    VarLongCoder.of().encode(1L, out);
+
+    CdcRecordCoder coder = CdcRecordCoder.of(DATA_SCHEMA);
+    assertThrows(
+        CoderException.class, () -> coder.decode(new ByteArrayInputStream(out.toByteArray())));
+  }
+}
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKeyTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKeyTest.java
new file mode 100644
index 000000000000..6e13f3858261
--- /dev/null
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSortKeyTest.java
@@ -0,0 +1,230 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.iceberg.cdc.sink;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.beam.sdk.values.ValueKind;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.UnsignedBytes;
+import org.hamcrest.Matchers;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link CdcSortKey}. */
+@RunWith(JUnit4.class)
+public class CdcSortKeyTest {
+
+  // Ascending sequence numbers, and ValueKinds in ascending kindRank order.
+  private static final long[] SEQS_ASCENDING = {-5L, -1L, 0L, 1L, 5L};
+  private static final ValueKind[] KINDS_BY_RANK_ASCENDING = {
+    ValueKind.UPDATE_BEFORE, ValueKind.DELETE, ValueKind.UPDATE_AFTER, ValueKind.INSERT
+  };
+
+  // Primary keys of assorted lengths and leading bytes.
+  private static final byte[][] PKS = {{0x02}, {0x01, 0x7F}, {(byte) 0xF0, 0x00, 0x10}};
+
+  @Test
+  public void kindRankOrdersBeforeImagesFirst() {
+    assertThat(CdcSortKey.kindRank(ValueKind.UPDATE_BEFORE), Matchers.equalTo((byte) 0));
+    assertThat(CdcSortKey.kindRank(ValueKind.DELETE), Matchers.equalTo((byte) 1));
+    assertThat(CdcSortKey.kindRank(ValueKind.UPDATE_AFTER), Matchers.equalTo((byte) 2));
+    assertThat(CdcSortKey.kindRank(ValueKind.INSERT), Matchers.equalTo((byte) 3));
+  }
+
+  /**
+   * The sorted keys must group each primary key's entries into one unbroken run, ordered by (seq,
+   * kind) within the run. Which primary key's run comes first is NOT part of the contract and is
+   * deliberately unasserted.
+   */
+  @Test
+  public void sortGroupsEachPkContiguouslyOrderedBySeqThenKindWithin() {
+    // Per pk, its keys in expected within-key order: seq ascending, kindRank breaking ties.
+    List<List<byte[]>> expectedByPk = new ArrayList<>();
+    for (byte[] pk : PKS) {
+      List<byte[]> expected = new ArrayList<>();
+      for (long seq : SEQS_ASCENDING) {
+        for (ValueKind kind : KINDS_BY_RANK_ASCENDING) {
+          expected.add(CdcSortKey.encode(pk, seq, kind));
+        }
+      }
+      expectedByPk.add(expected);
+    }
+
+    // Feed the sorter the same keys interleaved across pks (the shape a shard group arrives in);
+    // the sort must both regroup and reorder.
+    List<byte[]> sorted = new ArrayList<>();
+    for (int i = 0; i < expectedByPk.get(0).size(); i++) {
+      for (List<byte[]> keys : expectedByPk) {
+        sorted.add(keys.get(i));
+      }
+    }
+    sorted.sort(UnsignedBytes.lexicographicalComparator());
+
+    // Contiguity: each pk appears in exactly one run.
+    Set<Integer> seenRuns = new HashSet<>();
+    int previous = -1;
+    for (byte[] key : sorted) {
+      int pkIndex = pkIndexOf(key);
+      if (pkIndex != previous) {
+        assertTrue("entries of pk " + pkIndex + " are split across runs", seenRuns.add(pkIndex));
+        previous = pkIndex;
+      }
+    }
+
+    // Within each run, the (seq, kind) order pinned above.
+    for (int pkIndex = 0; pkIndex < PKS.length; pkIndex++) {
+      List<byte[]> run = new ArrayList<>();
+      for (byte[] key : sorted) {
+        if (pkIndexOf(key) == pkIndex) {
+          run.add(key);
+        }
+      }
+      List<byte[]> expected = expectedByPk.get(pkIndex);
+      assertThat(run, Matchers.hasSize(expected.size()));
+      for (int i = 0; i < expected.size(); i++) {
+        assertArrayEquals(expected.get(i), run.get(i));
+      }
+    }
+  }
+
+  /**
+   * Two primary keys where one's bytes are a strict prefix of the other's, with sequence numbers
+   * chosen so the unprefixed layout {@code [pkBytes][seq ^ Long.MIN_VALUE:8][kindRank:1]} WOULD
+   * interleave them; the length prefix must keep the short key's entries adjacent.
+   */
+  @Test
+  public void lengthPrefixKeepsAPrefixPkContiguous() {
+    byte[] shortPk = {0x0A};
+    byte[] longPk = {0x0A, (byte) 0x80};
+    byte[] shortLow = CdcSortKey.encode(shortPk, 0L, ValueKind.INSERT);
+    byte[] shortHigh = CdcSortKey.encode(shortPk, Long.MAX_VALUE, ValueKind.INSERT);
+    byte[] longMid = CdcSortKey.encode(longPk, 0L, ValueKind.INSERT);
+
+    // Fixture self-check: with the length prefix stripped, the long pk's entry lands BETWEEN the
+    // short pk's two entries (its second byte 0x80 ties the flipped seq 0 and loses to the
+    // flipped Long.MAX_VALUE).
+    List<byte[]> naive =
+        new ArrayList<>(Arrays.asList(stripLengthPrefix(shortHigh), stripLengthPrefix(longMid)));
+    naive.add(stripLengthPrefix(shortLow));
+    naive.sort(UnsignedBytes.lexicographicalComparator());
+    assertArrayEquals(stripLengthPrefix(shortLow), naive.get(0));
+    assertArrayEquals(stripLengthPrefix(longMid), naive.get(1));
+    assertArrayEquals(stripLengthPrefix(shortHigh), naive.get(2));
+
+    // The real keys: the short pk's entries stay adjacent, low seq first.
+    List<byte[]> sorted = new ArrayList<>(Arrays.asList(shortHigh, longMid, shortLow));
+    sorted.sort(UnsignedBytes.lexicographicalComparator());
+    int low = indexOfKey(sorted, shortLow);
+    int high = indexOfKey(sorted, shortHigh);
+    assertThat(high, Matchers.equalTo(low + 1));
+  }
+
+  @Test
+  public void encodeOrdersAcrossLongExtremes() {
+    byte[] pk = {0x01};
+    byte[] min = CdcSortKey.encode(pk, Long.MIN_VALUE, ValueKind.INSERT);
+    byte[] negOne = CdcSortKey.encode(pk, -1L, ValueKind.INSERT);
+    byte[] zero = CdcSortKey.encode(pk, 0L, ValueKind.INSERT);
+    byte[] one = CdcSortKey.encode(pk, 1L, ValueKind.INSERT);
+    byte[] max = CdcSortKey.encode(pk, Long.MAX_VALUE, ValueKind.INSERT);
+
+    List<byte[]> sorted = new ArrayList<>(Arrays.asList(max, zero, min, one, negOne));
+    sorted.sort(UnsignedBytes.lexicographicalComparator());
+
+    assertArrayEquals(min, sorted.get(0));
+    assertArrayEquals(negOne, sorted.get(1));
+    assertArrayEquals(zero, sorted.get(2));
+    assertArrayEquals(one, sorted.get(3));
+    assertArrayEquals(max, sorted.get(4));
+  }
+
+  /**
+   * The encoding is a frozen wire format: an in-place pipeline update replays in-flight groups
+   * through it, so changing any of these bytes silently re-orders live data. Update only with a
+   * migration story.
+   */
+  @Test
+  public void encodePinnedByteLayout() {
+    // pk {0x01, 0x02} -> pkLen 2; seq 5 -> flippedSeq 0x8000000000000005; INSERT -> kindRank 3.
+    assertArrayEquals(
+        new byte[] {0, 0, 0, 2, 1, 2, (byte) 0x80, 0, 0, 0, 0, 0, 0, 5, 3},
+        CdcSortKey.encode(new byte[] {1, 2}, 5L, ValueKind.INSERT));
+    // pk {0xAB} -> pkLen 1; seq 0 -> flippedSeq 0x8000000000000000; UPDATE_BEFORE -> kindRank 0.
+    assertArrayEquals(
+        new byte[] {0, 0, 0, 1, (byte) 0xAB, (byte) 0x80, 0, 0, 0, 0, 0, 0, 0, 0},
+        CdcSortKey.encode(new byte[] {(byte) 0xAB}, 0L, ValueKind.UPDATE_BEFORE));
+  }
+
+  /**
+   * {@code samePk} is the writer's block-boundary test: equal for any two keys of one pk whatever
+   * their seq/kind bytes, unequal across pks, including the prefix-pk pair whose seq bytes tie.
+   */
+  @Test
+  public void samePkComparesOnlyThePkPrefix() {
+    byte[] pk = {0x0A};
+    assertTrue(
+        CdcSortKey.samePk(
+            CdcSortKey.encode(pk, Long.MIN_VALUE, ValueKind.UPDATE_BEFORE),
+            CdcSortKey.encode(pk, Long.MAX_VALUE, ValueKind.INSERT)));
+    assertFalse(
+        CdcSortKey.samePk(
+            CdcSortKey.encode(new byte[] {0x01}, 5L, ValueKind.INSERT),
+            CdcSortKey.encode(new byte[] {0x02}, 5L, ValueKind.INSERT)));
+    // A strict-prefix pk with seq bytes continuing the longer pk's bytes must still differ.
+    assertFalse(
+        CdcSortKey.samePk(
+            CdcSortKey.encode(pk, 0L, ValueKind.INSERT),
+            CdcSortKey.encode(new byte[] {0x0A, (byte) 0x80}, 0L, ValueKind.INSERT)));
+  }
+
+  /** The index in {@link #PKS} of the pk carried in {@code key}'s length-prefixed prefix. */
+  private static int pkIndexOf(byte[] key) {
+    int pkLen = ByteBuffer.wrap(key).getInt();
+    byte[] pk = Arrays.copyOfRange(key, 4, 4 + pkLen);
+    for (int i = 0; i < PKS.length; i++) {
+      if (Arrays.equals(PKS[i], pk)) {
+        return i;
+      }
+    }
+    throw new AssertionError("unknown pk " + Arrays.toString(pk));
+  }
+
+  /** Drops the 4-byte length prefix, leaving the naive {@code [pkBytes][seq][kind]} layout. */
+  private static byte[] stripLengthPrefix(byte[] key) {
+    return Arrays.copyOfRange(key, 4, key.length);
+  }
+
+  private static int indexOfKey(List<byte[]> keys, byte[] key) {
+    for (int i = 0; i < keys.size(); i++) {
+      if (Arrays.equals(keys.get(i), key)) {
+        return i;
+      }
+    }
+    throw new AssertionError("key not found: " + Arrays.toString(key));
+  }
+}
diff --git a/sdks/java/io/thrift/src/main/java/org/apache/beam/sdk/io/thrift/ThriftSchema.java b/sdks/java/io/thrift/src/main/java/org/apache/beam/sdk/io/thrift/ThriftSchema.java
index e4e698faffa4..b24f132a0010 100644
--- a/sdks/java/io/thrift/src/main/java/org/apache/beam/sdk/io/thrift/ThriftSchema.java
+++ b/sdks/java/io/thrift/src/main/java/org/apache/beam/sdk/io/thrift/ThriftSchema.java
@@ -19,6 +19,7 @@
 
 import static java.util.Collections.unmodifiableMap;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
 import java.util.Arrays;
@@ -121,6 +122,10 @@ private ThriftSchema(Map<String, FieldType> typedefs) {
    *
    * @see #custom() for how to manually pass the beam type for container typedefs
    */
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "The default provider holds an empty typedef map that is never mutated. Callers needing typedefs go through custom(), which builds a separate instance.")
   public static @NonNull SchemaProvider provider() {
     return defaultProvider;
   }
diff --git a/sdks/java/testing/test-utils/src/main/java/org/apache/beam/sdk/testutils/NamedTestResult.java b/sdks/java/testing/test-utils/src/main/java/org/apache/beam/sdk/testutils/NamedTestResult.java
index a4f40598d5a3..112c29b30ee1 100644
--- a/sdks/java/testing/test-utils/src/main/java/org/apache/beam/sdk/testutils/NamedTestResult.java
+++ b/sdks/java/testing/test-utils/src/main/java/org/apache/beam/sdk/testutils/NamedTestResult.java
@@ -18,6 +18,7 @@
 package org.apache.beam.sdk.testutils;
 
 import com.google.cloud.bigquery.LegacySQLTypeName;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.Map;
 import org.apache.beam.sdk.testutils.publishing.InfluxDBPublisher;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
@@ -83,6 +84,12 @@ public Map<String, Object> toMap() {
         .build();
   }
 
+  @SuppressFBWarnings(
+      value = "MS_EXPOSE_REP",
+      justification =
+          "Returns a Guava ImmutableMap."
+              + " Spotbugs matches its known-immutable list by fully qualified name, so it cannot recognise collections relocated into org.apache.beam.vendor.guava."
+              + " See https://github.com/spotbugs/spotbugs/issues/1601.")
   public static Map<String, String> getSchema() {
     return schema;
   }
diff --git a/sdks/java/testing/tpcds/src/main/java/org/apache/beam/sdk/tpcds/TpcdsSchemas.java b/sdks/java/testing/tpcds/src/main/java/org/apache/beam/sdk/tpcds/TpcdsSchemas.java
index a22bab1c89e6..763e8164dee0 100644
--- a/sdks/java/testing/tpcds/src/main/java/org/apache/beam/sdk/tpcds/TpcdsSchemas.java
+++ b/sdks/java/testing/tpcds/src/main/java/org/apache/beam/sdk/tpcds/TpcdsSchemas.java
@@ -17,12 +17,18 @@
  */
 package org.apache.beam.sdk.tpcds;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
 
+@SuppressFBWarnings(
+    value = "MS_EXPOSE_REP",
+    justification =
+        "Fixed TPC-DS table definitions, built once at class load."
+            + " Schema is mutable only through setUUID, which no caller here uses, and copying twenty-four schemas on every accessor call would cost real time in a benchmark harness.")
 public class TpcdsSchemas {
   /**
    * Get all tpcds table schemas automatically by reading json files. In this case all field will be
diff --git a/sdks/python/build.gradle b/sdks/python/build.gradle
index 75afa5d7c968..339d87c76cbe 100644
--- a/sdks/python/build.gradle
+++ b/sdks/python/build.gradle
@@ -107,16 +107,24 @@ tasks.register("prepareExpansionServicesForYamlDocs") {
 
   dependsOn ":sdks:java:extensions:schemaio-expansion-service:shadowJar"
   dependsOn ":sdks:java:extensions:sql:expansion-service:shadowJar"
+  dependsOn ":sdks:java:io:amazon-web-services2:expansion-service:shadowJar"
+  dependsOn ":sdks:java:io:debezium:expansion-service:shadowJar"
   dependsOn ":sdks:java:io:expansion-service:shadowJar"
   dependsOn ":sdks:java:io:google-cloud-platform:expansion-service:shadowJar"
+  dependsOn ":sdks:java:io:messaging-expansion-service:shadowJar"
+  dependsOn ":sdks:java:io:snowflake:expansion-service:shadowJar"
 
   doLast {
     // Copy expansion service jar into cache path (.apache_beam/cache/jars/).
     copy {
       from project.tasks.findByPath(":sdks:java:extensions:schemaio-expansion-service:shadowJar")
       from project.tasks.findByPath(":sdks:java:extensions:sql:expansion-service:shadowJar")
+      from project.tasks.findByPath(":sdks:java:io:amazon-web-services2:expansion-service:shadowJar")
+      from project.tasks.findByPath(":sdks:java:io:debezium:expansion-service:shadowJar")
       from project.tasks.findByPath(":sdks:java:io:expansion-service:shadowJar")
       from project.tasks.findByPath(":sdks:java:io:google-cloud-platform:expansion-service:shadowJar")
+      from project.tasks.findByPath(":sdks:java:io:messaging-expansion-service:shadowJar")
+      from project.tasks.findByPath(":sdks:java:io:snowflake:expansion-service:shadowJar")
       into "${System.getProperty('user.home')}/.apache_beam/cache/jars/"
     }
   }
diff --git a/website/www/site/content/en/blog/beam-2.76.0.md b/website/www/site/content/en/blog/beam-2.76.0.md
index cea653eaade3..240c961cbd99 100644
--- a/website/www/site/content/en/blog/beam-2.76.0.md
+++ b/website/www/site/content/en/blog/beam-2.76.0.md
@@ -63,7 +63,6 @@ For more information on changes in 2.76.0, check out the [detailed release notes
 * (Python) Added `Watch`, a transform that polls a growing set of outputs for each input element, deduplicates outputs across poll rounds, and stops per a user-supplied termination condition
   ([#21521](https://github.com/apache/beam/issues/21521)).
 * (Python) Added support to analyze core dumps created after python worker segmentation faults with `pystack` (or `gdb` if installed) using the `--profiler_agent=coredump` pipeline option. ([#39484](https://github.com/apache/beam/issues/39484)).
-* (Python) Added `Sample.Any`, the Python equivalent of Java's `Sample.any`, which returns up to n arbitrary elements from a PCollection ([#18552](https://github.com/apache/beam/issues/18552)).
 
 ### Breaking Changes
 
@@ -78,9 +77,6 @@ For more information on changes in 2.76.0, check out the [detailed release notes
   Use pipeline option `--updateCompatibilityVersion=2.75.0` (or any older version) to keep the old behavior ([#39344](https://github.com/apache/beam/issues/39344)).
 * `DoFn.process` returning a `str`, `bytes`, or `dict` (instead of an iterable wrapping one) now raises a `TypeError` rather than silently iterating per-character/byte/key (Python) ([#18712](https://github.com/apache/beam/issues/18712)).
 * (Java) Added `DRAINING` and `DRAINED` states to `PipelineResult`, including runner state mappings and Dataflow update handling ([#39020](https://github.com/apache/beam/issues/39020)).
-* (Python) Typehints of dataclass fields are honored during type inferences. To restore the behavior of fallback-to-any,
-  use pipeline option `--exclude_infer_dataclass_field_type` ([#38797](https://github.com/apache/beam/issues/38797)).
-  However fixing forward is recommended.
 * (Java) IcebergIO and projects that use it must now be built with Java 17 or later as a result of Iceberg 1.11.0 upgrade ([#38925](https://github.com/apache/beam/issues/38925)).
 
 ### Bugfixes
diff --git a/website/www/site/content/en/documentation/io/connectors.md b/website/www/site/content/en/documentation/io/connectors.md
index 679b6bf1e0e3..f3078c6b378a 100644
--- a/website/www/site/content/en/documentation/io/connectors.md
+++ b/website/www/site/content/en/documentation/io/connectors.md
@@ -429,7 +429,10 @@ This table provides a consolidated, at-a-glance overview of the available built-
       ✔
       <a href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/io/jms/JmsIO.html">native</a>
     </td>
-    <td>Not available</td>
+    <td class="present">
+      ✔
+      <a href="https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.xlang.io.html#apache_beam.transforms.xlang.io.ReadFromJms">via X-language</a>
+    </td>
     <td>Not available</td>
     <td>Not available</td>
     <td>Not available</td>
@@ -790,7 +793,11 @@ This table provides a consolidated, at-a-glance overview of the available built-
       <a href="https://pkg.go.dev/github.com/apache/beam/sdks/v2/go/pkg/beam/io/xlang/debeziumio">via X-language</a>
     </td>
     <td>Not available</td>
-    <td>Not available</td>
+    <td class="present">
+      ✔
+      <br>
+      <a href="https://beam.apache.org/releases/yamldoc/current/#readfromdebezium">read</a>
+    </td>
     <td class="present">✔</td>
     <td class="present">✔</td>
     <td class="absent">✘</td>
@@ -1322,18 +1329,25 @@ This table provides a consolidated, at-a-glance overview of the available built-
   <tr>
     <td>Delta Lake</td>
     <td class="present">✔</td>
-    <td class="present">✘</td>
+    <td class="absent">✘</td>
     <td class="present">
       ✔
       <a href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/managed/Managed.html">native</a>
     </td>
+    <td class="present">
+      ✔
+      <a href="https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.managed.html#module-apache_beam.transforms.managed">via X-language</a>
+    </td>
     <td>Not available</td>
     <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
+    <td class="present">
+      ✔
+      <br>
+      <a href="https://beam.apache.org/releases/yamldoc/current/#readfromdelta">read</a>
+    </td>
     <td class="present">✔</td>
     <td class="absent">✘</td>
-    <td class="absent">✔</td>
+    <td class="present">✔</td>
   </tr>
 </table>
 </div>
@@ -1346,205 +1360,130 @@ This table provides a consolidated, at-a-glance overview of the available built-
     <th>Connector Name</th>
     <th>Source Supported</th>
     <th>Sink Supported</th>
-    <th>Java</th>
-    <th>Python</th>
-    <th>Go</th>
-    <th>Typescript</th>
-    <th>Yaml</th>
+    <th>Language</th>
     <th>Batch Supported</th>
     <th>Streaming Supported</th>
   </tr>
 
   <tr>
     <td>
-      <a href="https://github.com/google/hana-bq-beam-connector">SAP Hana to Google BigQuery</a>
+      <a href="https://cloud.google.com/bigtable/docs/hbase-dataflow-java">Cloud Bigtable (HBase based)</a>
     </td>
     <td class="present">✔</td>
     <td class="present">✔</td>
-    <td class="present">
-      ✔
-      native
-    </td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
+    <td>Java</td>
     <td class="present">✔</td>
     <td class="absent">✘</td>
   </tr>
   <tr>
     <td>
-      <a href="https://github.com/esakik/beam-mysql-connector">MySQL</a>
+      <a href="https://github.com/cognitedata/cdf-beam-connector-java">Cognite Data Fusion</a>
     </td>
     <td class="present">✔</td>
     <td class="present">✔</td>
-    <td>Not available</td>
-    <td class="present">
-      ✔
-      native
-    </td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td class="present">
-      ✔
-      <br>
-      <a href="https://beam.apache.org/releases/yamldoc/current/#readfrommysql">read</a> / <a href="https://beam.apache.org/releases/yamldoc/current/#writetomysql">write</a>
-    </td>
+    <td>Java</td>
+    <td class="present">✔</td>
     <td class="present">✔</td>
-    <td class="absent">✘</td>
   </tr>
   <tr>
     <td>
-      <a href="https://github.com/Refinitiv/trep-websockets-beam-io">TrepWsIO</a>
+      <a href="https://github.com/kineticadb/kinetica-connector-beam">KineticaDB</a>
     </td>
     <td class="present">✔</td>
-    <td class="absent">✘</td>
-    <td class="present">
-      ✔
-      native
-    </td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
     <td class="present">✔</td>
+    <td>Java</td>
     <td class="present">✔</td>
+    <td class="absent">✘</td>
   </tr>
   <tr>
     <td>
-      <a href="https://github.com/kineticadb/kinetica-connector-beam">KineticaDB</a>
+      <a href="https://github.com/google/hana-bq-beam-connector">SAP Hana to Google BigQuery</a>
     </td>
     <td class="present">✔</td>
     <td class="present">✔</td>
-    <td class="present">
-      ✔
-      native
-    </td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
+    <td>Java</td>
     <td class="present">✔</td>
     <td class="absent">✘</td>
   </tr>
   <tr>
     <td>
-      <a href="https://github.com/cognitedata/cdf-beam-connector-java">Cognite Data Fusion</a>
+      <a href="https://github.com/Refinitiv/trep-websockets-beam-io">TrepWsIO</a>
     </td>
     <td class="present">✔</td>
-    <td class="present">✔</td>
-    <td class="present">
-      ✔
-      native
-    </td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
+    <td class="absent">✘</td>
+    <td>Java</td>
     <td class="present">✔</td>
     <td class="present">✔</td>
   </tr>
   <tr>
     <td>
-      <a href="https://github.com/314e/beam-pyodbc-connector">Pyodbc</a>
+      <a href="https://github.com/beam-pyio">Beam PyIO (Collection of Python IO connectors)</a>
     </td>
     <td class="present">✔</td>
     <td class="present">✔</td>
-    <td>Not available</td>
-    <td class="present">
-      ✔
-      native
-    </td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
+    <td>Python</td>
+    <td class="present">✔</td>
     <td class="present">✔</td>
-    <td class="absent">✘</td>
   </tr>
   <tr>
     <td>
-      <a href="https://github.com/amient/goconnect">Go Connect</a>
+      <a href="https://github.com/jccatrinck/dataflow-cloud-sql-python">Cloud SQL</a>
     </td>
     <td class="present">✔</td>
-    <td class="present">✔</td>
     <td class="absent">✘</td>
-    <td class="absent">✘</td>
-    <td class="present">
-      ✔
-      native
-    </td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td class="present">✔</td>
+    <td>Python</td>
     <td class="present">✔</td>
+    <td class="absent">✘</td>
   </tr>
   <tr>
     <td>
-      <a href="https://github.com/tinybirdco/tinybird-beam">Tinybird</a>
+      <a href="https://github.com/esakik/beam-mysql-connector">MySQL</a>
     </td>
-    <td class="absent">✘</td>
     <td class="present">✔</td>
-    <td>Not available</td>
-    <td class="present">
-      ✔
-      native
-    </td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
     <td class="present">✔</td>
+    <td>Python</td>
     <td class="present">✔</td>
+    <td class="absent">✘</td>
   </tr>
   <tr>
     <td>
-      <a href="https://github.com/jccatrinck/dataflow-cloud-sql-python">Cloud SQL</a>
+      <a href="https://github.com/314e/beam-pyodbc-connector">Pyodbc</a>
     </td>
     <td class="present">✔</td>
-    <td class="absent">✘</td>
-    <td>Not available</td>
-    <td class="present">
-      ✔
-      native
-    </td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
+    <td class="present">✔</td>
+    <td>Python</td>
     <td class="present">✔</td>
     <td class="absent">✘</td>
   </tr>
   <tr>
     <td>
-      <a href="https://cloud.google.com/bigtable/docs/hbase-dataflow-java">Cloud Bigtable (HBase based)</a>
+      <a href="https://github.com/tinybirdco/tinybird-beam">Tinybird</a>
     </td>
+    <td class="absent">✘</td>
     <td class="present">✔</td>
+    <td>Python</td>
     <td class="present">✔</td>
-    <td class="present">
-      ✔
-      native
-    </td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
     <td class="present">✔</td>
-    <td class="absent">✘</td>
   </tr>
   <tr>
     <td>
-      <a href="https://github.com/beam-pyio">Beam PyIO (Collection of Python IO connectors)</a>
+      <a href="https://github.com/amient/goconnect">Go Connect</a>
     </td>
     <td class="present">✔</td>
     <td class="present">✔</td>
-    <td>Not available</td>
-    <td class="present">
-      ✔
-      native
+    <td>Go</td>
+    <td class="present">✔</td>
+    <td class="present">✔</td>
+  </tr>
+  <tr>
+    <td>
+      <a href="https://beam.apache.org/releases/yamldoc/current/#readfrommysql">MySQL, PostgreSQL, Oracle, SQL Server</a>
     </td>
-    <td>Not available</td>
-    <td>Not available</td>
-    <td>Not available</td>
     <td class="present">✔</td>
     <td class="present">✔</td>
+    <td>Yaml</td>
+    <td class="present">✔</td>
+    <td class="absent">✘</td>
   </tr>
 </table>
 </div>
diff --git a/website/www/site/data/capability_matrix.yaml b/website/www/site/data/capability_matrix.yaml
index e4db68c0cbeb..3f3bf39594c5 100644
--- a/website/www/site/data/capability_matrix.yaml
+++ b/website/www/site/data/capability_matrix.yaml
@@ -1736,7 +1736,6 @@ capability-matrix:
               l1: "Unverified"
               l2:
               l3:
-
             - class: kafka-streams
               l1: "No"
               l2: not implemented