Support iceberg schema type change

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

Author: sfc-gh-abozkurt

pg_lake_engine/src/parquet/field.c

@@ -25,6 +25,7 @@
 #include "pg_lake/util/string_utils.h"
 
 static FieldStructElement * DeepCopyFieldStructElement(FieldStructElement * structElementField);
+static bool FieldTypesEqual(const Field * fieldA, const Field * fieldB);
 
 /*
  * DeepCopyField deep copies a Field.
@@ -149,11 +150,49 @@ pg_cmp_s32(int32 a, int32 b)
 
 
 
+/*
+ * FieldTypesEqual recursively compares two Field structs for type equality.
+ */
+static bool
+FieldTypesEqual(const Field * fieldA, const Field * fieldB)
+{
+	if (fieldA->type != fieldB->type)
+		return false;
+
+	switch (fieldA->type)
+	{
+		case FIELD_TYPE_SCALAR:
+			return strcmp(fieldA->field.scalar.typeName,
+						  fieldB->field.scalar.typeName) == 0;
+		case FIELD_TYPE_LIST:
+			return FieldTypesEqual(fieldA->field.list.element,
+								   fieldB->field.list.element);
+		case FIELD_TYPE_MAP:
+			return FieldTypesEqual(fieldA->field.map.key,
+								   fieldB->field.map.key) &&
+				FieldTypesEqual(fieldA->field.map.value,
+								fieldB->field.map.value);
+		case FIELD_TYPE_STRUCT:
+			{
+				if (fieldA->field.structType.nfields != fieldB->field.structType.nfields)
+					return false;
+				for (size_t i = 0; i < fieldA->field.structType.nfields; i++)
+				{
+					if (!FieldTypesEqual(fieldA->field.structType.fields[i].type,
+										 fieldB->field.structType.fields[i].type))
+						return false;
+				}
+				return true;
+			}
+		default:
+			return false;
+	}
+}
+
+
 /*
 * SchemaFieldsEquivalent compares two DataFileSchemaField structs for equivalence.
 * It returns true if they are equivalent, false otherwise.
-* Note that we do not compare the field->type here, as we do not allow changing
-* the type of any field in the schema, including nested types.
 */
 bool
 SchemaFieldsEquivalent(DataFileSchemaField * fieldA, DataFileSchemaField * fieldB)
@@ -176,11 +215,9 @@ SchemaFieldsEquivalent(DataFileSchemaField * fieldA, DataFileSchemaField * field
 	if (!PgStrcasecmpNullable(fieldA->initialDefault, fieldB->initialDefault))
 		return false;
 
-	/*
-	 * We don't allow changing any of the types of the fields in the schema,
-	 * including the fields of nested types. So we don't need to compare
-	 * anything about the field->type here.
-	 */
+	if (!FieldTypesEqual(fieldA->type, fieldB->type))
+		return false;
+
 	return true;
 }
 

pg_lake_iceberg/src/iceberg/iceberg_type_binary_serde.c

@@ -371,10 +371,29 @@ PGIcebergBinaryDeserialize(unsigned char *binaryValue, size_t binaryLen, Field *
 	}
 	else if (pgType.postgresTypeOid == INT8OID)
 	{
-		binaryValue = FromLittleEndian64(binaryValue);
-		int64		longValue = *((int64 *) binaryValue);
+		/*
+		 * Per the Iceberg spec, column metrics are serialized using the type
+		 * at the time the data file was written. After an int -> long type
+		 * promotion, existing data files retain 4-byte int bounds while the
+		 * current schema expects 8-byte long. Widen on read.
+		 *
+		 * See:
+		 * https://iceberg.apache.org/spec/#binary-single-value-serialization
+		 */
+		if (binaryLen == sizeof(int32))
+		{
+			binaryValue = FromLittleEndian32(binaryValue);
+			int32		intValue = *((int32 *) binaryValue);
 
-		datum = Int64GetDatum(longValue);
+			datum = Int64GetDatum((int64) intValue);
+		}
+		else
+		{
+			binaryValue = FromLittleEndian64(binaryValue);
+			int64		longValue = *((int64 *) binaryValue);
+
+			datum = Int64GetDatum(longValue);
+		}
 	}
 	else if (pgType.postgresTypeOid == FLOAT4OID)
 	{
@@ -385,10 +404,25 @@ PGIcebergBinaryDeserialize(unsigned char *binaryValue, size_t binaryLen, Field *
 	}
 	else if (pgType.postgresTypeOid == FLOAT8OID)
 	{
-		binaryValue = FromLittleEndian64(binaryValue);
-		float8		doubleValue = *((float8 *) binaryValue);
+		/*
+		 * Same as int -> long above: after a float -> double type promotion,
+		 * existing data files retain 4-byte float bounds while the current
+		 * schema expects 8-byte double. Widen on read.
+		 */
+		if (binaryLen == sizeof(float4))
+		{
+			binaryValue = FromLittleEndian32(binaryValue);
+			float4		floatValue = *((float4 *) binaryValue);
 
-		datum = Float8GetDatum(doubleValue);
+			datum = Float8GetDatum((float8) floatValue);
+		}
+		else
+		{
+			binaryValue = FromLittleEndian64(binaryValue);
+			float8		doubleValue = *((float8 *) binaryValue);
+
+			datum = Float8GetDatum(doubleValue);
+		}
 	}
 	else if (pgType.postgresTypeOid == DATEOID)
 	{

pg_lake_table/include/pg_lake/ddl/ddl_changes.h

@@ -21,6 +21,8 @@
 #include "nodes/pg_list.h"
 #include "nodes/parsenodes.h"
 
+#include "pg_lake/pgduck/type.h"
+
 /*
  * DDLOperationType describes a type of ddl operation, which
  * cause catalog modification and possibly iceberg metadata modification.
@@ -37,6 +39,7 @@ typedef enum DDLOperationType
 	DDL_COLUMN_DROP_NOT_NULL = 8,
 	DDL_TABLE_SET_PARTITION_BY = 9,
 	DDL_TABLE_DROP_PARTITION_BY = 10,
+	DDL_COLUMN_ALTER_TYPE = 11,
 }			DDLOperationType;
 
 /*
@@ -55,6 +58,9 @@ typedef struct IcebergDDLOperation
 	/* applicable for set/drop default */
 	const char *writeDefault;
 
+	/* applicable for alter column type */
+	PGType		newPgType;
+
 	/* applicable for create iceberg table */
 	bool		hasCustomLocation;
 

pg_lake_table/include/pg_lake/fdw/schema_operations/field_id_mapping_catalog.h

@@ -33,6 +33,7 @@ extern PGDLLEXPORT List *GetRegisteredFieldForAttributes(Oid relationId, List *a
 extern PGDLLEXPORT DataFileSchemaField * GetRegisteredFieldForAttribute(Oid relationId, AttrNumber attrNo);
 extern PGDLLEXPORT AttrNumber GetAttributeForFieldId(Oid relationId, int fieldId);
 extern PGDLLEXPORT void UpdateRegisteredFieldWriteDefaultForAttribute(Oid relationId, AttrNumber attNum, const char *writeDefault);
+extern PGDLLEXPORT void UpdateRegisteredFieldTypeForAttribute(Oid relationId, AttrNumber attNum, PGType newPgType);
 extern PGDLLEXPORT int GetLargestRegisteredFieldId(Oid relationId);
 extern PGDLLEXPORT void RegisterIcebergColumnMapping(Oid relationId, Field * field,
 													 AttrNumber attNo, int parentFieldId, PGType pgType,