Core, Spark: Add row lineage metadata columns, and surface them in SparkTable metadata columns (apache#12596)

Author: amogh-jahagirdar

core/src/main/java/org/apache/iceberg/MetadataColumns.java

@@ -93,21 +93,37 @@ private MetadataColumns() {}
           "_commit_snapshot_id",
           Types.LongType.get(),
           "Commit snapshot ID");
+  public static final NestedField ROW_ID =
+      NestedField.optional(
+          Integer.MAX_VALUE - 107,
+          "_row_id",
+          Types.LongType.get(),
+          "Implicit row ID that is automatically assigned");
+  public static final NestedField LAST_UPDATED_SEQUENCE_NUMBER =
+      NestedField.optional(
+          Integer.MAX_VALUE - 108,
+          "_last_updated_sequence_number",
+          Types.LongType.get(),
+          "Sequence number when the row was last updated");
 
   private static final Map<String, NestedField> META_COLUMNS =
       ImmutableMap.of(
           FILE_PATH.name(), FILE_PATH,
           ROW_POSITION.name(), ROW_POSITION,
           IS_DELETED.name(), IS_DELETED,
-          SPEC_ID.name(), SPEC_ID);
+          SPEC_ID.name(), SPEC_ID,
+          ROW_ID.name(), ROW_ID,
+          LAST_UPDATED_SEQUENCE_NUMBER.name(), LAST_UPDATED_SEQUENCE_NUMBER);
 
   private static final Set<Integer> META_IDS =
       ImmutableSet.of(
           FILE_PATH.fieldId(),
           ROW_POSITION.fieldId(),
           IS_DELETED.fieldId(),
           SPEC_ID.fieldId(),
-          PARTITION_COLUMN_ID);
+          PARTITION_COLUMN_ID,
+          ROW_ID.fieldId(),
+          LAST_UPDATED_SEQUENCE_NUMBER.fieldId());
 
   public static Set<Integer> metadataFieldIds() {
     return META_IDS;

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java

@@ -39,6 +39,7 @@
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.TableScan;
+import org.apache.iceberg.TableUtil;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.Evaluator;
 import org.apache.iceberg.expressions.Expression;
@@ -48,6 +49,7 @@
 import org.apache.iceberg.expressions.StrictMetricsEvaluator;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
@@ -256,13 +258,23 @@ public Set<TableCapability> capabilities() {
   @Override
   public MetadataColumn[] metadataColumns() {
     DataType sparkPartitionType = SparkSchemaUtil.convert(Partitioning.partitionType(table()));
-    return new MetadataColumn[] {
-      new SparkMetadataColumn(MetadataColumns.SPEC_ID.name(), DataTypes.IntegerType, false),
-      new SparkMetadataColumn(MetadataColumns.PARTITION_COLUMN_NAME, sparkPartitionType, true),
-      new SparkMetadataColumn(MetadataColumns.FILE_PATH.name(), DataTypes.StringType, false),
-      new SparkMetadataColumn(MetadataColumns.ROW_POSITION.name(), DataTypes.LongType, false),
-      new SparkMetadataColumn(MetadataColumns.IS_DELETED.name(), DataTypes.BooleanType, false)
-    };
+    ImmutableList.Builder<SparkMetadataColumn> metadataColumns = ImmutableList.builder();
+    metadataColumns.add(
+        new SparkMetadataColumn(MetadataColumns.SPEC_ID.name(), DataTypes.IntegerType, false),
+        new SparkMetadataColumn(MetadataColumns.PARTITION_COLUMN_NAME, sparkPartitionType, true),
+        new SparkMetadataColumn(MetadataColumns.FILE_PATH.name(), DataTypes.StringType, false),
+        new SparkMetadataColumn(MetadataColumns.ROW_POSITION.name(), DataTypes.LongType, false),
+        new SparkMetadataColumn(MetadataColumns.IS_DELETED.name(), DataTypes.BooleanType, false));
+
+    if (TableUtil.formatVersion(table()) >= 3) {
+      metadataColumns.add(
+          new SparkMetadataColumn(MetadataColumns.ROW_ID.name(), DataTypes.LongType, true));
+      metadataColumns.add(
+          new SparkMetadataColumn(
+              MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.name(), DataTypes.LongType, true));
+    }
+
+    return metadataColumns.build().toArray(SparkMetadataColumn[]::new);
   }
 
   @Override

spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java

@@ -58,6 +58,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.spark.TestBase;
 import org.apache.iceberg.types.Types;
+import org.apache.spark.sql.AnalysisException;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Encoders;
 import org.apache.spark.sql.Row;
@@ -94,12 +95,17 @@ public static Object[][] parameters() {
       {FileFormat.PARQUET, true, 1},
       {FileFormat.PARQUET, false, 2},
       {FileFormat.PARQUET, true, 2},
+      {FileFormat.PARQUET, false, 3},
+      {FileFormat.PARQUET, true, 3},
       {FileFormat.AVRO, false, 1},
       {FileFormat.AVRO, false, 2},
+      {FileFormat.AVRO, false, 3},
       {FileFormat.ORC, false, 1},
       {FileFormat.ORC, true, 1},
       {FileFormat.ORC, false, 2},
       {FileFormat.ORC, true, 2},
+      {FileFormat.ORC, false, 3},
+      {FileFormat.ORC, true, 3},
     };
   }
 
@@ -311,6 +317,28 @@ public void testConflictingColumns() {
         sql("SELECT _spec_id, _partition, _renamed_spec_id FROM %s", TABLE_NAME));
   }
 
+  @TestTemplate
+  public void testRowLineageColumnsResolvedInV3OrHigher() {
+    if (formatVersion >= 3) {
+      // Test against an empty table to ensure column resolution in formats supporting row lineage
+      // and so that the test doesn't have to change with inheritance
+      assertEquals(
+          "Rows must match",
+          ImmutableList.of(),
+          sql("SELECT _row_id, _last_updated_sequence_number, id FROM %s", TABLE_NAME));
+    } else {
+      // Should fail to resolve row lineage metadata columns in V1/V2 tables
+      assertThatThrownBy(() -> sql("SELECT _row_id FROM %s", TABLE_NAME))
+          .isInstanceOf(AnalysisException.class)
+          .hasMessageContaining(
+              "A column or function parameter with name `_row_id` cannot be resolved");
+      assertThatThrownBy(() -> sql("SELECT _last_updated_sequence_number FROM %s", TABLE_NAME))
+          .isInstanceOf(AnalysisException.class)
+          .hasMessageContaining(
+              "A column or function parameter with name `_last_updated_sequence_number` cannot be resolved");
+    }
+  }
+
   private void createAndInitTable() throws IOException {
     Map<String, String> properties = Maps.newHashMap();
     properties.put(FORMAT_VERSION, String.valueOf(formatVersion));