From e88b168c665199f623588fa914e8665344633d13 Mon Sep 17 00:00:00 2001
From: tomtongue <tanaka1201t@gmail.com>
Date: Tue, 18 Feb 2025 19:38:25 +0900
Subject: [PATCH] Fix setting row-lineage from table properties

---
 .../org/apache/iceberg/TableMetadata.java     |  6 ++++++
 .../iceberg/TestRowLineageMetadata.java       | 15 ++++++++++++++
 .../java/org/apache/iceberg/TestTables.java   | 20 +++++++++++++++++++
 3 files changed, 41 insertions(+)

diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index 25a6cc96ce39..ea61f300b666 100644
--- a/core/src/main/java/org/apache/iceberg/TableMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -133,6 +133,11 @@ static TableMetadata newTableMetadata(
     int freshSortOrderId = sortOrder.isUnsorted() ? sortOrder.orderId() : INITIAL_SORT_ORDER_ID;
     SortOrder freshSortOrder = freshSortOrder(freshSortOrderId, freshSchema, sortOrder);
 
+    // configure row lineage using table properties
+    Boolean rowLineage =
+        PropertyUtil.propertyAsBoolean(
+            properties, TableProperties.ROW_LINEAGE, DEFAULT_ROW_LINEAGE);
+
     // Validate the metrics configuration. Note: we only do this on new tables to we don't
     // break existing tables.
     MetricsConfig.fromProperties(properties).validateReferencedColumns(schema);
@@ -146,6 +151,7 @@ static TableMetadata newTableMetadata(
         .setDefaultSortOrder(freshSortOrder)
         .setLocation(location)
         .setProperties(properties)
+        .setRowLineage(rowLineage)
         .build();
   }
 
diff --git a/core/src/test/java/org/apache/iceberg/TestRowLineageMetadata.java b/core/src/test/java/org/apache/iceberg/TestRowLineageMetadata.java
index 0fb9ee880f6b..3a72f654038a 100644
--- a/core/src/test/java/org/apache/iceberg/TestRowLineageMetadata.java
+++ b/core/src/test/java/org/apache/iceberg/TestRowLineageMetadata.java
@@ -26,6 +26,7 @@
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.primitives.Ints;
 import org.apache.iceberg.types.Types;
 import org.junit.jupiter.api.AfterEach;
@@ -306,6 +307,20 @@ public void testEnableRowLineageViaProperty() {
     assertThat(table.ops().current().rowLineageEnabled()).isTrue();
   }
 
+  @TestTemplate
+  public void testEnableRowLineageViaPropertyAtTableCreation() {
+    assumeThat(formatVersion).isGreaterThanOrEqualTo(TableMetadata.MIN_FORMAT_VERSION_ROW_LINEAGE);
+
+    TestTables.TestTable table =
+        TestTables.create(
+            tableDir,
+            "test",
+            TEST_SCHEMA,
+            ImmutableMap.of(TableProperties.ROW_LINEAGE, "true"),
+            formatVersion);
+    assertThat(table.ops().current().rowLineageEnabled()).isTrue();
+  }
+
   private final AtomicInteger fileNum = new AtomicInteger(0);
 
   private DataFile fileWithRows(long numRows) {
diff --git a/core/src/test/java/org/apache/iceberg/TestTables.java b/core/src/test/java/org/apache/iceberg/TestTables.java
index eeff5db8e5a6..de4aff8142b9 100644
--- a/core/src/test/java/org/apache/iceberg/TestTables.java
+++ b/core/src/test/java/org/apache/iceberg/TestTables.java
@@ -52,6 +52,26 @@ public static TestTable create(
     return create(temp, name, schema, spec, SortOrder.unsorted(), formatVersion);
   }
 
+  public static TestTable create(
+      File temp, String name, Schema schema, Map<String, String> properties, int formatVersion) {
+    TestTableOperations ops = new TestTableOperations(name, temp);
+    if (ops.current() != null) {
+      throw new AlreadyExistsException("Table %s already exists at location: %s", name, temp);
+    }
+
+    ops.commit(
+        null,
+        newTableMetadata(
+            schema,
+            PartitionSpec.unpartitioned(),
+            SortOrder.unsorted(),
+            temp.toString(),
+            properties,
+            formatVersion));
+
+    return new TestTable(ops, name);
+  }
+
   public static TestTable create(
       File temp,
       String name,