[default values] Spark catalog itest for ReadBridge initial-default overlay

integrations/spark/spark-3.1/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/ColumnDefaultSparkITest.java

@@ -0,0 +1,153 @@
+package com.linkedin.openhouse.spark.catalogtest;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import com.linkedin.openhouse.tablestest.OpenHouseSparkITest;
+import com.linkedin.openhouse.tablestest.SparkItestColumnDefaults;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+/**
+ * Catalog → ReadBridge overlay → Spark's Iceberg table. Iceberg 1.2.0.20's Spark readers do not
+ * fill {@code initial-default}; SQL fill/filter coverage is disabled until OpenHouse picks up
+ * linkedin/iceberg#267–269.
+ */
+public class ColumnDefaultSparkITest extends OpenHouseSparkITest {
+
+  private static final String DATABASE = SparkItestColumnDefaults.DATABASE;
+  private static final String ENABLED_PROP = "read-bridge.column-default.enabled";
+
+  @Test
+  public void readBridgeOverlaysInitialDefaultAfterSparkAddColumn() throws Exception {
+    String fqtn = "openhouse." + DATABASE + ".members_overlay";
+    try (SparkSession spark = getSparkSession()) {
+      spark.sql("DROP TABLE IF EXISTS " + fqtn);
+      spark.sql(
+          "CREATE TABLE "
+              + fqtn
+              + " (id bigint, name string) USING iceberg TBLPROPERTIES ('"
+              + ENABLED_PROP
+              + "'='true')");
+      spark.sql("INSERT INTO " + fqtn + " VALUES (1, 'Alice')");
+      spark.sql("ALTER TABLE " + fqtn + " ADD COLUMN country string");
+      spark.sql("ALTER TABLE " + fqtn + " ADD COLUMN tier int");
+      spark.sql("REFRESH TABLE " + fqtn);
+
+      Table table = Spark3Util.loadIcebergTable(spark, fqtn);
+      assertEquals(
+          "true",
+          table.properties().get(ENABLED_PROP),
+          "ramp property missing; props=" + table.properties());
+      assertEquals(
+          "US",
+          table.schema().findField("country").initialDefault(),
+          "ReadBridge overlay missing on country");
+      assertEquals(
+          1,
+          table.schema().findField("tier").initialDefault(),
+          "ReadBridge overlay missing on tier");
+
+      spark.sql("DROP TABLE " + fqtn);
+    }
+  }
+
+  /**
+   * Enable after bumping {@code iceberg_1_2_version} to a build that fills {@code initial-default}
+   * in Spark parquet/ORC readers (linkedin/iceberg#267–269).
+   */
+  @Disabled("Spark readers in iceberg 1.2.0.20 do not fill initial-default")
+  @ParameterizedTest(name = "format={0}, vectorized={1}")
+  @CsvSource({"parquet, false", "parquet, true", "orc, false", "orc, true"})
+  public void columnDefaultBackfillViaReadBridge(String fileFormat, boolean vectorized)
+      throws Exception {
+    String vectorizationProp =
+        "parquet".equals(fileFormat)
+            ? "read.parquet.vectorization.enabled"
+            : "read.orc.vectorization.enabled";
+    String tableName = String.format("members_%s_%s", fileFormat, vectorized);
+    String fqtn = String.format("openhouse.%s.%s", DATABASE, tableName);
+
+    try (SparkSession spark = getSparkSession()) {
+      spark.sql("DROP TABLE IF EXISTS " + fqtn);
+      spark.sql(
+          String.format(
+              "CREATE TABLE %s (id bigint, name string) USING iceberg TBLPROPERTIES ("
+                  + "'write.format.default'='%s', '%s'='%b', '%s'='true')",
+              fqtn, fileFormat, vectorizationProp, vectorized, ENABLED_PROP));
+      spark.sql("INSERT INTO " + fqtn + " VALUES (1, 'Alice')");
+      spark.sql("ALTER TABLE " + fqtn + " ADD COLUMN country string");
+      spark.sql("ALTER TABLE " + fqtn + " ADD COLUMN tier int");
+      spark.sql("INSERT INTO " + fqtn + " VALUES (2, 'Bob', 'CA', 5)");
+      spark.sql(
+          "INSERT INTO " + fqtn + " VALUES (3, 'Cleo', CAST(NULL AS string), CAST(NULL AS int))");
+
+      List<Object[]> full =
+          rows(spark, "SELECT id, name, country, tier FROM " + fqtn + " ORDER BY id");
+      assertRowsEqual(
+          Arrays.asList(
+              row(1L, "Alice", "US", 1), row(2L, "Bob", "CA", 5), row(3L, "Cleo", null, null)),
+          full);
+
+      assertIds(spark, fqtn, "country = 'US'", 1L);
+      assertIds(spark, fqtn, "country = 'CA'", 2L);
+      assertIds(spark, fqtn, "upper(country) = 'US'", 1L);
+      assertIds(spark, fqtn, "tier = 1", 1L);
+      assertIds(spark, fqtn, "country IS NOT NULL", 1L, 2L);
+      assertIds(spark, fqtn, "country IS NULL", 3L);
+
+      List<Object[]> projected =
+          rows(spark, "SELECT country FROM " + fqtn + " WHERE country = 'US'");
+      assertRowsEqual(Collections.singletonList(row("US")), projected);
+
+      spark.sql("DROP TABLE " + fqtn);
+    }
+  }
+
+  private static void assertRowsEqual(List<Object[]> expectedRows, List<Object[]> actualRows) {
+    assertEquals(expectedRows.size(), actualRows.size(), "row count");
+    for (int row = 0; row < expectedRows.size(); row++) {
+      Object[] expected = expectedRows.get(row);
+      Object[] actual = actualRows.get(row);
+      assertEquals(expected.length, actual.length, "column count at row " + row);
+      for (int col = 0; col < expected.length; col++) {
+        assertEquals(expected[col], actual[col], "row " + row + " col " + col);
+      }
+    }
+  }
+
+  private static List<Object[]> rows(SparkSession spark, String sql) {
+    return spark.sql(sql).collectAsList().stream()
+        .map(
+            row -> {
+              Object[] values = new Object[row.size()];
+              for (int i = 0; i < row.size(); i++) {
+                values[i] = row.isNullAt(i) ? null : row.get(i);
+              }
+              return values;
+            })
+        .collect(Collectors.toList());
+  }
+
+  private static Object[] row(Object... values) {
+    return values;
+  }
+
+  private static void assertIds(
+      SparkSession spark, String fqtn, String predicate, Object... expectedIds) {
+    List<Object[]> result =
+        rows(spark, "SELECT id FROM " + fqtn + " WHERE " + predicate + " ORDER BY id");
+    assertEquals(
+        Arrays.asList(expectedIds),
+        result.stream().map(r -> r[0]).collect(Collectors.toList()),
+        "filter: " + predicate);
+  }
+}

tables-test-fixtures/tables-test-fixtures-iceberg-1.2/src/main/java/com/linkedin/openhouse/tablestest/SparkItestColumnDefaults.java

@@ -0,0 +1,64 @@
+package com.linkedin.openhouse.tablestest;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.IntNode;
+import com.fasterxml.jackson.databind.node.TextNode;
+import com.linkedin.openhouse.tables.model.TableDto;
+import com.linkedin.openhouse.tables.readbridge.ColumnDefaultsSource;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * Local-server stand-in for a deployment {@link ColumnDefaultsSource}. OSS OpenHouse has no
+ * encoder; Spark catalog itests need one so ReadBridge can stamp {@code initial-default} on load.
+ * Scoped to {@link #DATABASE} so other catalog tests stay on empty maps.
+ */
+@Configuration
+public class SparkItestColumnDefaults {
+
+  public static final String DATABASE = "d1_column_default";
+
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  @Bean
+  ColumnDefaultsSource sparkItestColumnDefaultsSource() {
+    return SparkItestColumnDefaults::defaults;
+  }
+
+  static Map<Integer, JsonNode> defaults(TableDto tableDto) {
+    if (tableDto == null || !DATABASE.equals(tableDto.getDatabaseId())) {
+      return Collections.emptyMap();
+    }
+    String schemaJson = tableDto.getSchema();
+    if (schemaJson == null || schemaJson.isEmpty()) {
+      return Collections.emptyMap();
+    }
+    JsonNode fields;
+    try {
+      fields = MAPPER.readTree(schemaJson).path("fields");
+    } catch (Exception e) {
+      return Collections.emptyMap();
+    }
+    if (!fields.isArray()) {
+      return Collections.emptyMap();
+    }
+    Map<Integer, JsonNode> out = new LinkedHashMap<>();
+    for (JsonNode field : fields) {
+      String name = field.path("name").asText();
+      if (!field.has("id")) {
+        continue;
+      }
+      int id = field.get("id").asInt();
+      if ("country".equals(name)) {
+        out.put(id, TextNode.valueOf("US"));
+      } else if ("tier".equals(name)) {
+        out.put(id, IntNode.valueOf(1));
+      }
+    }
+    return out;
+  }
+}

tables-test-fixtures/tables-test-fixtures-iceberg-1.2/src/main/java/com/linkedin/openhouse/tablestest/SpringH2TestApplication.java

@@ -11,6 +11,7 @@
 import org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Import;
 import org.springframework.context.annotation.Primary;
 
 @SpringBootApplication
@@ -42,6 +43,7 @@
     })
 @EnableAutoConfiguration(
     exclude = {SecurityAutoConfiguration.class, ManagementWebSecurityAutoConfiguration.class})
+@Import(SparkItestColumnDefaults.class)
 public class SpringH2TestApplication {
 
   public static void main(String[] args) {