Spark: Fix row lineage inheritance for distributed planning

Author: amogh-jahagirdar

spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java

@@ -196,6 +196,18 @@ public static Object[][] parameters() {
         LOCAL,
         3
       },
+      {
+        "testhadoop",
+        SparkCatalog.class.getName(),
+        ImmutableMap.of("type", "hadoop"),
+        FileFormat.PARQUET,
+        false,
+        WRITE_DISTRIBUTION_MODE_HASH,
+        true,
+        null,
+        DISTRIBUTED,
+        3
+      },
       {
         "spark_catalog",
         SparkSessionCatalog.class.getName(),

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/ManifestFileBean.java

@@ -36,6 +36,7 @@ public class ManifestFileBean implements ManifestFile, Serializable {
   private Long addedSnapshotId = null;
   private Integer content = null;
   private Long sequenceNumber = null;
+  private Long firstRowId = null;
 
   public static ManifestFileBean fromManifest(ManifestFile manifest) {
     ManifestFileBean bean = new ManifestFileBean();
@@ -46,6 +47,7 @@ public static ManifestFileBean fromManifest(ManifestFile manifest) {
     bean.setAddedSnapshotId(manifest.snapshotId());
     bean.setContent(manifest.content().id());
     bean.setSequenceNumber(manifest.sequenceNumber());
+    bean.setFirstRowId(manifest.firstRowId());
 
     return bean;
   }
@@ -98,6 +100,10 @@ public void setSequenceNumber(Long sequenceNumber) {
     this.sequenceNumber = sequenceNumber;
   }
 
+  public void setFirstRowId(Long firstRowId) {
+    this.firstRowId = firstRowId;
+  }
+
   @Override
   public String path() {
     return path;
@@ -173,6 +179,11 @@ public ByteBuffer keyMetadata() {
     return null;
   }
 
+  @Override
+  public Long firstRowId() {
+    return firstRowId;
+  }
+
   @Override
   public ManifestFile copy() {
     throw new UnsupportedOperationException("Cannot copy");