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,

pg_lake_table/src/ddl/alter_table.c

@@ -58,6 +58,7 @@
 #include "pg_lake/partitioning/partition_spec_catalog.h"
 #include "pg_lake/rest_catalog/rest_catalog.h"
 #include "pg_lake/object_store_catalog/object_store_catalog.h"
+#include "pg_lake/pgduck/numeric.h"
 #include "pg_lake/util/rel_utils.h"
 
 
@@ -119,6 +120,7 @@ typedef struct PgLakeDDL
 static bool Allowed(Node *arg, Oid relationId);
 static bool Disallowed(Node *arg, Oid relationId);
 static bool DisallowedAddColumnWithUnsupportedConstraints(Node *arg, Oid relationId);
+static bool AllowedAlterColumnTypeForIceberg(Node *arg, Oid relationId);
 static bool DisallowedForWritableRestRenameTable(Node *arg, Oid relationId);
 static bool DisallowedForWritableRestSetSchema(Node *arg, Oid relationId);
 
@@ -155,7 +157,7 @@ static const PgLakeDDL PgLakeDDLs[] = {
 #if PG_VERSION_NUM >= 170000
 	ALTER_TABLE_DDL(AT_SetExpression, Disallowed, Disallowed),
 #endif
-	ALTER_TABLE_DDL(AT_AlterColumnType, Disallowed, Disallowed),
+	ALTER_TABLE_DDL(AT_AlterColumnType, AllowedAlterColumnTypeForIceberg, Disallowed),
 
 	/* allowed for writable tables, not allowed for iceberg tables */
 	ALTER_TABLE_DDL(AT_AddIdentity, Disallowed, Allowed),
@@ -409,6 +411,32 @@ CreateDDLOperationsForAlterTable(AlterTableStmt *alterStmt)
 
 			ddlOperations = lappend(ddlOperations, ddlOperation);
 		}
+		else if (subcommand->subtype == AT_AlterColumnType)
+		{
+			char	   *columnName = subcommand->name;
+			AttrNumber	attrNo = get_attnum(relationId, columnName);
+
+			IcebergDDLOperation *ddlOperation = palloc0(sizeof(IcebergDDLOperation));
+
+			ddlOperation->type = DDL_COLUMN_ALTER_TYPE;
+			ddlOperation->attrNumber = attrNo;
+
+			/*
+			 * At this point, PgLakeCommonParentProcessUtility has already
+			 * been called so the column type in pg_attribute is updated. We
+			 * read the new type from the relation.
+			 */
+			Oid			newTypeOid = InvalidOid;
+			int32		newTypMod = -1;
+			Oid			newCollId = InvalidOid;
+
+			get_atttypetypmodcoll(relationId, attrNo,
+								  &newTypeOid, &newTypMod, &newCollId);
+
+			ddlOperation->newPgType = MakePGType(newTypeOid, newTypMod);
+
+			ddlOperations = lappend(ddlOperations, ddlOperation);
+		}
 		else if (subcommand->subtype == AT_DropNotNull)
 		{
 			IcebergDDLOperation *ddlOperation = palloc0(sizeof(IcebergDDLOperation));
@@ -841,10 +869,40 @@ ErrorIfUnsupportedAlterWritablePgLakeTableStmt(AlterTableStmt *alterStmt,
 
 			const char *cmdTypeStr = PgLakeUnsupportedAlterTableToString(cmd);
 
-			ereport(ERROR,
-					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-					 errmsg("ALTER TABLE %s command not supported for "
-							"%s tables", cmdTypeStr, tableTypeStr)));
+			if (cmd->subtype == AT_AlterColumnType &&
+				tableType == PG_LAKE_ICEBERG_TABLE_TYPE)
+			{
+				ColumnDef  *def = (ColumnDef *) cmd->def;
+
+				if (def->raw_default != NULL)
+				{
+					ereport(ERROR,
+							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+							 errmsg("ALTER TABLE %s command not supported for "
+									"%s tables", cmdTypeStr, tableTypeStr),
+							 errdetail("USING requires rewriting data files, "
+									   "but Iceberg schema evolution is a "
+									   "metadata-only operation.")));
+				}
+				else
+				{
+					ereport(ERROR,
+							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+							 errmsg("ALTER TABLE %s command not supported for "
+									"%s tables", cmdTypeStr, tableTypeStr),
+							 errdetail("Allowed type promotions for Iceberg tables "
+									   "are: int -> bigint, float -> double, and "
+									   "decimal(P,S) -> decimal(P',S) where P' > P "
+									   "(wider precision, same scale).")));
+				}
+			}
+			else
+			{
+				ereport(ERROR,
+						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+						 errmsg("ALTER TABLE %s command not supported for "
+								"%s tables", cmdTypeStr, tableTypeStr)));
+			}
 		}
 	}
 
@@ -1043,6 +1101,113 @@ Disallowed(Node *arg, Oid relationId)
 }
 
 
+/*
+ * AllowedAlterColumnTypeForIceberg validates whether a column type change
+ * is allowed for an Iceberg table per the Iceberg spec v2 type promotion rules.
+ *
+ * The Iceberg spec allows the following type promotions:
+ *   - int -> long
+ *   - float -> double
+ *   - decimal(P, S) -> decimal(P', S) where P' > P (wider precision, same scale)
+ *
+ * See: https://iceberg.apache.org/spec/#schema-evolution
+ */
+static bool
+AllowedAlterColumnTypeForIceberg(Node *arg, Oid relationId)
+{
+	AlterTableCmd *cmd = (AlterTableCmd *) arg;
+
+	Assert(cmd->subtype == AT_AlterColumnType);
+
+	char	   *columnName = cmd->name;
+	ColumnDef  *def = (ColumnDef *) cmd->def;
+	TypeName   *newTypeName = def->typeName;
+
+	/*
+	 * Reject USING clause. USING requires rewriting data files, but Iceberg
+	 * schema evolution is metadata-only so data files are never rewritten.
+	 */
+	if (def->raw_default != NULL)
+		return false;
+
+	/* resolve the new type OID and typmod */
+	int32		newTypMod = 0;
+	Oid			newTypeOid = InvalidOid;
+
+	typenameTypeIdAndMod(NULL, newTypeName, &newTypeOid, &newTypMod);
+
+	/* get the current column type from the relation */
+	AttrNumber	attrNum = get_attnum(relationId, columnName);
+
+	if (attrNum == InvalidAttrNumber)
+		return false;
+
+	Oid			currentTypeOid = InvalidOid;
+	int32		currentTypMod = -1;
+	Oid			currentCollId = InvalidOid;
+
+	get_atttypetypmodcoll(relationId, attrNum,
+						  &currentTypeOid, &currentTypMod, &currentCollId);
+
+	/* same type is always allowed (no-op from Iceberg perspective) */
+	if (currentTypeOid == newTypeOid && currentTypMod == newTypMod)
+		return true;
+
+	/*
+	 * Iceberg type promotion: int -> long
+	 *
+	 * PostgreSQL int2 and int4 both map to Iceberg "int", and int8 maps to
+	 * Iceberg "long". So we allow int2/int4 -> int8.
+	 */
+	if ((currentTypeOid == INT4OID || currentTypeOid == INT2OID) &&
+		newTypeOid == INT8OID)
+		return true;
+
+	/*
+	 * Iceberg type promotion: float -> double
+	 *
+	 * PostgreSQL float4 maps to Iceberg "float" and float8 maps to Iceberg
+	 * "double". So we allow float4 -> float8.
+	 */
+	if (currentTypeOid == FLOAT4OID && newTypeOid == FLOAT8OID)
+		return true;
+
+	/*
+	 * Iceberg type promotion: decimal(P, S) -> decimal(P', S) where P' > P
+	 *
+	 * Both must be numeric, the new precision must be greater, and the scale
+	 * must remain the same.
+	 */
+	if (currentTypeOid == NUMERICOID && newTypeOid == NUMERICOID)
+	{
+		int			currentPrecision = -1;
+		int			currentScale = -1;
+		int			newPrecision = -1;
+		int			newScale = -1;
+
+		GetDuckdbAdjustedPrecisionAndScaleFromNumericTypeMod(currentTypMod,
+															 &currentPrecision,
+															 &currentScale);
+		GetDuckdbAdjustedPrecisionAndScaleFromNumericTypeMod(newTypMod,
+															 &newPrecision,
+															 &newScale);
+
+		/* new precision must be wider and scale must stay the same */
+		if (newPrecision > currentPrecision && newScale == currentScale)
+		{
+			/* also validate the new type is supported for Iceberg tables */
+			ErrorIfTypeUnsupportedNumericForIcebergTables(newTypMod, columnName);
+			return true;
+		}
+
+		return false;
+	}
+
+	/* all other type changes are disallowed */
+	return false;
+}
+
+
 /*
 * We have not yet implemented table renames in REST catalog.
 */