add spark test

Signed-off-by: Aykut Bozkurt <aykut.bozkurt@snowflake.com>

Author: sfc-gh-abozkurt

pg_lake_table/tests/pytests/test_iceberg_ddl.py

@@ -1,6 +1,7 @@
 import pytest
 import psycopg2
 from utils_pytest import *
+from spark_utils import *
 
 import os
 import glob
@@ -2174,6 +2175,316 @@ def test_alter_column_type_partitioned_table(pg_conn, s3, with_default_location)
     pg_conn.commit()
 
 
+def test_alter_column_type_spark_comparison(
+    installcheck, spark_session, pg_conn, s3, with_default_location
+):
+    """Create the same table in Spark and pg_lake, perform identical type promotions,
+    verify both produce the same results, compare metadata schemas, and confirm
+    Spark can read pg_lake's metadata.json after type promotion."""
+    if installcheck:
+        return
+
+    pg_schema = "test_type_promo_pg"
+    spark_ns = "public"
+
+    run_command(f"CREATE SCHEMA {pg_schema};", pg_conn)
+
+    # ── 1. int → bigint ──
+
+    # Spark side
+    spark_session.sql(
+        f"CREATE TABLE {spark_ns}.spark_int_promo (a int, b int) USING iceberg"
+    )
+    spark_session.sql(f"INSERT INTO {spark_ns}.spark_int_promo VALUES (1, 10), (2, 20)")
+    spark_session.sql(
+        f"ALTER TABLE {spark_ns}.spark_int_promo ALTER COLUMN a TYPE bigint"
+    )
+    spark_session.sql(f"INSERT INTO {spark_ns}.spark_int_promo VALUES ({2**40}, 30)")
+
+    # pg_lake side — same operations
+    run_command(
+        f"CREATE TABLE {pg_schema}.int_tbl (a int, b int) USING iceberg;", pg_conn
+    )
+    run_command(f"INSERT INTO {pg_schema}.int_tbl VALUES (1, 10), (2, 20);", pg_conn)
+    pg_conn.commit()
+    run_command(f"ALTER TABLE {pg_schema}.int_tbl ALTER COLUMN a TYPE bigint;", pg_conn)
+    pg_conn.commit()
+    run_command(f"INSERT INTO {pg_schema}.int_tbl VALUES ({2**40}, 30);", pg_conn)
+    pg_conn.commit()
+
+    # Compare: query both natively
+    spark_query = f"SELECT a, b FROM {spark_ns}.spark_int_promo ORDER BY b ASC"
+    pg_query = f"SELECT a, b FROM {pg_schema}.int_tbl ORDER BY b ASC"
+
+    pg_lake_result = assert_query_result_on_spark_and_pg(
+        installcheck, spark_session, pg_conn, spark_query, pg_query
+    )
+
+    assert len(pg_lake_result) == 3
+    assert pg_lake_result == [[1, 10], [2, 20], [2**40, 30]]
+
+    # Compare full schemas JSON
+    spark_metadata_loc = (
+        spark_session.sql(
+            f"SELECT file FROM {spark_ns}.spark_int_promo.metadata_log_entries ORDER BY timestamp DESC"
+        )
+        .collect()[0]
+        .file
+    )
+    spark_json = normalize_json(read_s3_operations(s3, spark_metadata_loc))
+
+    pg_metadata_loc = run_query(
+        f"SELECT metadata_location FROM iceberg_tables "
+        f"WHERE table_name = 'int_tbl' AND table_namespace = '{pg_schema}'",
+        pg_conn,
+    )[0][0]
+    pg_json = normalize_json(read_s3_operations(s3, pg_metadata_loc))
+
+    assert_iceberg_schemas_equal(spark_json, pg_json, "int→bigint")
+
+    # Verify Spark can read pg_lake's metadata.json
+    # Disable vectorized reading: Spark 3.5 / Iceberg 1.4.3 vectorized reader
+    # crashes with ClassCastException when an old data file's Parquet physical
+    # type (int32) doesn't match the current schema type (int64) after promotion.
+    spark_session.conf.set("spark.sql.iceberg.vectorization.enabled", "false")
+    spark_register_table(
+        installcheck, spark_session, "int_tbl", pg_schema, pg_metadata_loc
+    )
+    spark_cross_query = f"SELECT a, b FROM {pg_schema}.int_tbl ORDER BY b ASC"
+    spark_cross = spark_session.sql(spark_cross_query).collect()
+    assert len(spark_cross) == 3
+    assert [spark_cross[0].a, spark_cross[0].b] == [1, 10]
+    assert [spark_cross[1].a, spark_cross[1].b] == [2, 20]
+    assert [spark_cross[2].a, spark_cross[2].b] == [2**40, 30]
+    spark_unregister_table(installcheck, spark_session, "int_tbl", pg_schema)
+    spark_session.conf.set("spark.sql.iceberg.vectorization.enabled", "true")
+
+    spark_session.sql(f"DROP TABLE {spark_ns}.spark_int_promo")
+
+    # ── 2. float → double ──
+
+    # Spark side
+    spark_session.sql(
+        f"CREATE TABLE {spark_ns}.spark_float_promo (a float, b int) USING iceberg"
+    )
+    spark_session.sql(f"INSERT INTO {spark_ns}.spark_float_promo VALUES (1.5, 1)")
+    spark_session.sql(
+        f"ALTER TABLE {spark_ns}.spark_float_promo ALTER COLUMN a TYPE double"
+    )
+    spark_session.sql(
+        f"INSERT INTO {spark_ns}.spark_float_promo VALUES (1.23456789012345, 2)"
+    )
+
+    # pg_lake side
+    run_command(
+        f"CREATE TABLE {pg_schema}.float_tbl (a real, b int) USING iceberg;", pg_conn
+    )
+    run_command(f"INSERT INTO {pg_schema}.float_tbl VALUES (1.5, 1);", pg_conn)
+    pg_conn.commit()
+    run_command(
+        f"ALTER TABLE {pg_schema}.float_tbl ALTER COLUMN a TYPE double precision;",
+        pg_conn,
+    )
+    pg_conn.commit()
+    run_command(
+        f"INSERT INTO {pg_schema}.float_tbl VALUES (1.23456789012345, 2);", pg_conn
+    )
+    pg_conn.commit()
+
+    spark_query = f"SELECT a, b FROM {spark_ns}.spark_float_promo ORDER BY b ASC"
+    pg_query = f"SELECT a, b FROM {pg_schema}.float_tbl ORDER BY b ASC"
+
+    pg_lake_result = assert_query_result_on_spark_and_pg(
+        installcheck, spark_session, pg_conn, spark_query, pg_query
+    )
+
+    assert len(pg_lake_result) == 2
+
+    # Compare full schemas JSON
+    spark_metadata_loc = (
+        spark_session.sql(
+            f"SELECT file FROM {spark_ns}.spark_float_promo.metadata_log_entries ORDER BY timestamp DESC"
+        )
+        .collect()[0]
+        .file
+    )
+    spark_json = normalize_json(read_s3_operations(s3, spark_metadata_loc))
+
+    pg_metadata_loc = run_query(
+        f"SELECT metadata_location FROM iceberg_tables "
+        f"WHERE table_name = 'float_tbl' AND table_namespace = '{pg_schema}'",
+        pg_conn,
+    )[0][0]
+    pg_json = normalize_json(read_s3_operations(s3, pg_metadata_loc))
+
+    assert_iceberg_schemas_equal(spark_json, pg_json, "float→double")
+
+    # Verify Spark can read pg_lake's metadata.json (vectorized off, same reason)
+    spark_session.conf.set("spark.sql.iceberg.vectorization.enabled", "false")
+    spark_register_table(
+        installcheck, spark_session, "float_tbl", pg_schema, pg_metadata_loc
+    )
+    spark_cross_query = f"SELECT a, b FROM {pg_schema}.float_tbl ORDER BY b ASC"
+    spark_cross = spark_session.sql(spark_cross_query).collect()
+    assert len(spark_cross) == 2
+    assert spark_cross[0].a == pytest.approx(1.5, abs=1e-6)
+    assert spark_cross[1].a == pytest.approx(1.23456789012345, abs=1e-10)
+    spark_unregister_table(installcheck, spark_session, "float_tbl", pg_schema)
+    spark_session.conf.set("spark.sql.iceberg.vectorization.enabled", "true")
+
+    spark_session.sql(f"DROP TABLE {spark_ns}.spark_float_promo")
+
+    # ── 3. decimal(P,S) → decimal(P',S) where P' > P ──
+
+    # Spark side
+    spark_session.sql(
+        f"CREATE TABLE {spark_ns}.spark_dec_promo (a decimal(10,2), b int) USING iceberg"
+    )
+    spark_session.sql(f"INSERT INTO {spark_ns}.spark_dec_promo VALUES (12345.67, 1)")
+    spark_session.sql(
+        f"ALTER TABLE {spark_ns}.spark_dec_promo ALTER COLUMN a TYPE decimal(20,2)"
+    )
+    spark_session.sql(
+        f"INSERT INTO {spark_ns}.spark_dec_promo VALUES (123456789012345.67, 2)"
+    )
+
+    # pg_lake side
+    run_command(
+        f"CREATE TABLE {pg_schema}.dec_tbl (a numeric(10,2), b int) USING iceberg;",
+        pg_conn,
+    )
+    run_command(f"INSERT INTO {pg_schema}.dec_tbl VALUES (12345.67, 1);", pg_conn)
+    pg_conn.commit()
+    run_command(
+        f"ALTER TABLE {pg_schema}.dec_tbl ALTER COLUMN a TYPE numeric(20,2);", pg_conn
+    )
+    pg_conn.commit()
+    run_command(
+        f"INSERT INTO {pg_schema}.dec_tbl VALUES (123456789012345.67, 2);", pg_conn
+    )
+    pg_conn.commit()
+
+    spark_query = f"SELECT a, b FROM {spark_ns}.spark_dec_promo ORDER BY b ASC"
+    pg_query = f"SELECT a, b FROM {pg_schema}.dec_tbl ORDER BY b ASC"
+
+    pg_lake_result = assert_query_result_on_spark_and_pg(
+        installcheck, spark_session, pg_conn, spark_query, pg_query
+    )
+
+    assert len(pg_lake_result) == 2
+
+    # Compare full schemas JSON
+    spark_metadata_loc = (
+        spark_session.sql(
+            f"SELECT file FROM {spark_ns}.spark_dec_promo.metadata_log_entries ORDER BY timestamp DESC"
+        )
+        .collect()[0]
+        .file
+    )
+    spark_json = normalize_json(read_s3_operations(s3, spark_metadata_loc))
+
+    pg_metadata_loc = run_query(
+        f"SELECT metadata_location FROM iceberg_tables "
+        f"WHERE table_name = 'dec_tbl' AND table_namespace = '{pg_schema}'",
+        pg_conn,
+    )[0][0]
+    pg_json = normalize_json(read_s3_operations(s3, pg_metadata_loc))
+
+    assert_iceberg_schemas_equal(spark_json, pg_json, "decimal widening")
+
+    # Verify Spark can read pg_lake's metadata.json
+    spark_register_table(
+        installcheck, spark_session, "dec_tbl", pg_schema, pg_metadata_loc
+    )
+    spark_cross_query = f"SELECT a, b FROM {pg_schema}.dec_tbl ORDER BY b ASC"
+    spark_cross = spark_session.sql(spark_cross_query).collect()
+    assert len(spark_cross) == 2
+    assert str(spark_cross[0].a) == "12345.67"
+    assert str(spark_cross[1].a) == "123456789012345.67"
+    spark_unregister_table(installcheck, spark_session, "dec_tbl", pg_schema)
+
+    spark_session.sql(f"DROP TABLE {spark_ns}.spark_dec_promo")
+
+    # ── 4. partitioned table: int → bigint on partition column ──
+
+    # Spark side
+    spark_session.sql(
+        f"CREATE TABLE {spark_ns}.spark_part_promo (a int, b int) "
+        f"USING iceberg PARTITIONED BY (b)"
+    )
+    spark_session.sql(
+        f"INSERT INTO {spark_ns}.spark_part_promo VALUES (1, 10), (2, 20)"
+    )
+    spark_session.sql(
+        f"ALTER TABLE {spark_ns}.spark_part_promo ALTER COLUMN b TYPE bigint"
+    )
+    spark_session.sql(f"INSERT INTO {spark_ns}.spark_part_promo VALUES (3, 30)")
+
+    # pg_lake side
+    run_command(
+        f"CREATE TABLE {pg_schema}.part_tbl (a int, b int) "
+        f"USING iceberg WITH (partition_by = 'b');",
+        pg_conn,
+    )
+    run_command(f"INSERT INTO {pg_schema}.part_tbl VALUES (1, 10), (2, 20);", pg_conn)
+    pg_conn.commit()
+    run_command(
+        f"ALTER TABLE {pg_schema}.part_tbl ALTER COLUMN b TYPE bigint;", pg_conn
+    )
+    pg_conn.commit()
+    run_command(f"INSERT INTO {pg_schema}.part_tbl VALUES (3, 30);", pg_conn)
+    pg_conn.commit()
+
+    spark_query = f"SELECT a, b FROM {spark_ns}.spark_part_promo ORDER BY a ASC"
+    pg_query = f"SELECT a, b FROM {pg_schema}.part_tbl ORDER BY a ASC"
+
+    pg_lake_result = assert_query_result_on_spark_and_pg(
+        installcheck, spark_session, pg_conn, spark_query, pg_query
+    )
+
+    assert len(pg_lake_result) == 3
+    assert pg_lake_result == [[1, 10], [2, 20], [3, 30]]
+
+    # Compare full schemas JSON
+    spark_metadata_loc = (
+        spark_session.sql(
+            f"SELECT file FROM {spark_ns}.spark_part_promo.metadata_log_entries ORDER BY timestamp DESC"
+        )
+        .collect()[0]
+        .file
+    )
+    spark_json = normalize_json(read_s3_operations(s3, spark_metadata_loc))
+
+    pg_metadata_loc = run_query(
+        f"SELECT metadata_location FROM iceberg_tables "
+        f"WHERE table_name = 'part_tbl' AND table_namespace = '{pg_schema}'",
+        pg_conn,
+    )[0][0]
+    pg_json = normalize_json(read_s3_operations(s3, pg_metadata_loc))
+
+    assert_iceberg_schemas_equal(spark_json, pg_json, "partitioned int→bigint")
+
+    # Verify Spark can read pg_lake's metadata.json (vectorized off, same reason)
+    spark_session.conf.set("spark.sql.iceberg.vectorization.enabled", "false")
+    spark_register_table(
+        installcheck, spark_session, "part_tbl", pg_schema, pg_metadata_loc
+    )
+    spark_cross_query = f"SELECT a, b FROM {pg_schema}.part_tbl ORDER BY a ASC"
+    spark_cross = spark_session.sql(spark_cross_query).collect()
+    assert len(spark_cross) == 3
+    assert [spark_cross[0].a, spark_cross[0].b] == [1, 10]
+    assert [spark_cross[1].a, spark_cross[1].b] == [2, 20]
+    assert [spark_cross[2].a, spark_cross[2].b] == [3, 30]
+    spark_unregister_table(installcheck, spark_session, "part_tbl", pg_schema)
+    spark_session.conf.set("spark.sql.iceberg.vectorization.enabled", "true")
+
+    spark_session.sql(f"DROP TABLE {spark_ns}.spark_part_promo")
+
+    # cleanup
+    run_command(f"DROP SCHEMA {pg_schema} CASCADE;", pg_conn)
+    pg_conn.commit()
+
+
 def get_current_schema_id(pg_conn, s3, namespace, name):
 
     metadata_location = run_query(

test_common/helpers/iceberg.py

@@ -798,3 +798,36 @@ def adjust_object_store_settings(superuser_conn):
 
     run_command("SELECT pg_reload_conf()", superuser_conn)
     superuser_conn.commit()
+
+
+def assert_iceberg_schemas_equal(left_json, right_json, label=""):
+    """Compare the full schemas array from two Iceberg metadata JSONs.
+
+    Both engines should produce the same number of schema versions with
+    matching field definitions.  Field IDs and schema-ids are ignored
+    because they may be assigned differently by each engine.
+    """
+    def _norm_fields(fields):
+        return [
+            (f["name"],
+             re.sub(r"\s+", "", f["type"]) if isinstance(f["type"], str) else f["type"],
+             f.get("required", False))
+            for f in fields
+        ]
+
+    left_schemas = left_json["schemas"]
+    right_schemas = right_json["schemas"]
+
+    assert len(left_schemas) == len(right_schemas), (
+        f"[{label}] schema count mismatch: "
+        f"left has {len(left_schemas)}, right has {len(right_schemas)}"
+    )
+
+    for idx, (ls, rs) in enumerate(zip(left_schemas, right_schemas)):
+        left_fields = _norm_fields(ls["fields"])
+        right_fields = _norm_fields(rs["fields"])
+        assert left_fields == right_fields, (
+            f"[{label}] schema #{idx} field mismatch:\n"
+            f"  left:  {left_fields}\n"
+            f"  right: {right_fields}"
+        )