Do not generate a new schema if already exists (#89)

Author: sfc-gh-okalaci

pg_lake_engine/include/pg_lake/data_file/data_files.h

@@ -102,6 +102,24 @@ typedef enum TableMetadataOperationType
 struct IcebergPartitionSpec;
 struct Partition;
 
+
+/*
+* When a schema is set back to an existing schema in iceberg metadata,
+* we use DDL_EFFECT_SET_EXISTING_SCHEMA to indicate which effect the
+* DDL has. An example would be ALTER TABLE .. ADD COLUMN x; followed by
+* ALTER TABLE .. DROP COLUMN x; which results in the same schema as before.
+* In all other cases, we use DDL_EFFECT_ADD_SCHEMA as that'd mean there is
+* no existing schema to set to, and we should add a new schema.
+* ps: This is especially important for REST catalog/Polaris, where it rejects
+* identical schema addition.
+*/
+typedef enum DDLSchemaEffect
+{
+	DDL_EFFECT_NONE = 0,
+	DDL_EFFECT_ADD_SCHEMA = 1,
+	DDL_EFFECT_SET_EXISTING_SCHEMA = 2,
+}			DDLSchemaEffect;
+
 /*
  * TableMetadataOperation represents an operation on table metadata.
  */
@@ -125,8 +143,13 @@ typedef struct TableMetadataOperation
 	/* for a new deletion file, from which data file are we deleting? */
 	char	   *deletedFrom;
 
-	/* relevant to TABLE_DDL event, up-to-date schema */
-	DataFileSchema *schema;
+	/*
+	 * newSchema and existingSchemaId are mutually exclusive, set according to
+	 * ddlSchemaEffect.
+	 */
+	DDLSchemaEffect ddlSchemaEffect;
+	DataFileSchema *newSchema;
+	int32_t		existingSchemaId;
 
 	/* for multi-delete, which files we are deleting from */
 	List	   *deleteStats;

pg_lake_engine/include/pg_lake/parquet/leaf_field.h

@@ -50,6 +50,7 @@ typedef struct LeafField
 }			LeafField;
 
 extern PGDLLEXPORT int LeafFieldCompare(const ListCell *a, const ListCell *b);
+extern PGDLLEXPORT bool SchemaFieldsEquivalent(DataFileSchemaField * fieldA, DataFileSchemaField * fieldB);
 #if PG_VERSION_NUM < 170000
 extern PGDLLEXPORT int pg_cmp_s32(int32 a, int32 b);
 #endif

pg_lake_engine/include/pg_lake/util/string_utils.h

@@ -33,7 +33,7 @@ extern PGDLLEXPORT char *EscapedStringLiteral(const char *val);
 extern PGDLLEXPORT char *ReverseStringSearch(const char *haystack, const char *needle);
 extern PGDLLEXPORT int32_t AdjustAnyCharTypmod(int32_t typmod, int32_t newLength);
 extern PGDLLEXPORT int32_t GetAnyCharLengthFrom(int32_t typmod);
-
+extern PGDLLEXPORT bool PgStrcasecmpNullable(const char *a, const char *b);
 
 #define RangeVarQuoteIdentifier(rv) \
 	(((rv)->schemaname != NULL) ?							  \

pg_lake_engine/src/parquet/field.c

@@ -21,6 +21,7 @@
 
 #include "pg_lake/parquet/field.h"
 #include "pg_lake/parquet/leaf_field.h"
+#include "pg_lake/util/string_utils.h"
 
 static FieldStructElement * DeepCopyFieldStructElement(FieldStructElement * structElementField);
 static Field * DeepCopyField(const Field * field);
@@ -145,3 +146,40 @@ pg_cmp_s32(int32 a, int32 b)
 	return (a > b) - (a < b);
 }
 #endif
+
+
+
+/*
+* 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)
+{
+	if (fieldA->id != fieldB->id)
+		return false;
+
+	if (!PgStrcasecmpNullable(fieldA->name, fieldB->name))
+		return false;
+
+	if (fieldA->required != fieldB->required)
+		return false;
+
+	if (!PgStrcasecmpNullable(fieldA->doc, fieldB->doc))
+		return false;
+
+	if (!PgStrcasecmpNullable(fieldA->writeDefault, fieldB->writeDefault))
+		return false;
+
+	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.
+	 */
+	return true;
+}

pg_lake_engine/src/utils/string_utils.c

@@ -245,3 +245,19 @@ GetAnyCharLengthFrom(int32_t typmod)
 
 	return typmod - VARHDRSZ;
 }
+
+
+/*
+* PgStrcasecmpNullable compares two strings for equality, treating NULLs as equal.
+*/
+bool
+PgStrcasecmpNullable(const char *a, const char *b)
+{
+	if (a == NULL && b == NULL)
+		return true;
+
+	if (a == NULL || b == NULL)
+		return false;
+
+	return pg_strcasecmp(a, b) == 0;
+}

pg_lake_iceberg/src/iceberg/api/table_metadata.c

@@ -76,6 +76,7 @@ static void SetSnapshotReference(IcebergTableMetadata * metadata, uint64_t snaps
 static void GroupExpiredSnapshots(IcebergTableMetadata * metadata, IcebergSnapshot * expiredSnapshots, int *expiredSnapshotCount,
 								  IcebergSnapshot * nonExpiredSnapshots, int *nonExpiredSnapshotCount);
 static bool SnapshotAgeExceeded(IcebergSnapshot * snapshot, int64_t currentTimeMs, int maxAge);
+static int32_t MaxSchemaId(IcebergTableSchema * schemas, size_t schemasLength);
 
 /*
 * GenerateEmptyTableMetadata generates an empty iceberg table metadata
@@ -544,7 +545,6 @@ void
 AppendCurrentPostgresSchema(Oid relationId, IcebergTableMetadata * metadata,
 							DataFileSchema * schema)
 {
-	int			currentSchemaId = metadata->current_schema_id;
 	int			currentSchemaLength = metadata->schemas_length;
 
 	IcebergTableSchema *newSchema = RebuildIcebergSchemaFromDataFileSchema(relationId, schema, &metadata->last_column_id);
@@ -560,13 +560,33 @@ AppendCurrentPostgresSchema(Oid relationId, IcebergTableMetadata * metadata,
 	}
 
 	/* first schema should always start with id=0 */
-	newSchema->schema_id = (currentSchemaLength == 0) ? 0 : currentSchemaId + 1;
+	newSchema->schema_id = (currentSchemaLength == 0) ? 0 : MaxSchemaId(metadata->schemas, currentSchemaLength) + 1;
 	metadata->schemas[currentSchemaLength] = *newSchema;
 	metadata->schemas_length = currentSchemaLength + 1;
 	metadata->current_schema_id = newSchema->schema_id;
 }
 
 
+/*
+* MaxSchemaId finds the maximum schema ID from the given schemas.
+*/
+static int32_t
+MaxSchemaId(IcebergTableSchema * schemas, size_t schemasLength)
+{
+	int32_t		maxSchemaId = -1;
+
+	for (size_t i = 0; i < schemasLength; i++)
+	{
+		if (schemas[i].schema_id > maxSchemaId)
+		{
+			maxSchemaId = schemas[i].schema_id;
+		}
+	}
+
+	Assert(maxSchemaId != -1);
+	return maxSchemaId;
+}
+
 /*
 * AppendPartitionSpec appends given partition spec to the metadata.
 */

pg_lake_iceberg/src/iceberg/metadata_operations.c

@@ -82,7 +82,7 @@ typedef struct IcebergSnapshotBuilder
 	/* whether to apply manifest compaction */
 	bool		applyManifestCompaction;
 
-	/* a DDL has changed the iceberg schema */
+	/* a DDL has changed the iceberg schema or set to an existing schema */
 	bool		regenerateSchema;
 
 	/* a DDL has changed partition specs */
@@ -96,8 +96,10 @@ typedef struct IcebergSnapshotBuilder
 	/* whether to expire old snapshots */
 	bool		expireOldSnapshots;
 
-	/* up-to-date schema for table */
+	/* new schema */
 	DataFileSchema *schema;
+	/* a DDL has set to an existing schema */
+	int32_t		schemaId;
 
 	/* snapshot operation */
 	SnapshotOperation operation;
@@ -183,7 +185,10 @@ ApplyIcebergMetadataChanges(Oid relationId, List *metadataOperations, List *allT
 
 	if (builder->createTable || builder->regenerateSchema)
 	{
-		AppendCurrentPostgresSchema(relationId, metadata, builder->schema);
+		if (builder->schema != NULL)
+			AppendCurrentPostgresSchema(relationId, metadata, builder->schema);
+		else
+			metadata->current_schema_id = builder->schemaId;
 	}
 
 	if (builder->createTable || builder->regeneratePartitionSpec)
@@ -588,7 +593,7 @@ ProcessIcebergMetadataOperations(Oid relationId, List *metadataOperations,
 					Assert(!builder->createTable);
 
 					builder->createTable = true;
-					builder->schema = operation->schema;
+					builder->schema = operation->newSchema;
 					builder->partitionSpecs = operation->partitionSpecs;
 					builder->defaultSpecId = operation->defaultSpecId;
 
@@ -604,14 +609,33 @@ ProcessIcebergMetadataOperations(Oid relationId, List *metadataOperations,
 					 */
 					Assert(!builder->regenerateSchema);
 
+					builder->regenerateSchema = true;
+
 					/*
 					 * We are requested to update the table schema, and we'll
 					 * handle this operation in
 					 * TrackIcebergMetadataChangesInTx().
 					 */
 
-					builder->regenerateSchema = true;
-					builder->schema = operation->schema;
+					if (operation->ddlSchemaEffect == DDL_EFFECT_ADD_SCHEMA)
+					{
+						/* these two are mutually exclusive */
+						Assert(operation->existingSchemaId == -1);
+						Assert(operation->newSchema != NULL);
+
+						builder->schema = operation->newSchema;
+					}
+					else if (operation->ddlSchemaEffect == DDL_EFFECT_SET_EXISTING_SCHEMA)
+					{
+						/* these two are mutually exclusive */
+						Assert(operation->existingSchemaId != -1);
+						Assert(operation->newSchema == NULL);
+						builder->schemaId = operation->existingSchemaId;
+					}
+					else
+					{
+						ereport(ERROR, (errmsg("Unsupported DDL schema effect: %d", operation->ddlSchemaEffect)));
+					}
 
 					break;
 				}

pg_lake_table/src/transaction/track_iceberg_metadata_changes.c

@@ -50,6 +50,8 @@ static List *GetDataFileMetadataOperations(const TableMetadataOperationTracker *
 										   List *allTransforms);
 static List *GetDDLMetadataOperations(const TableMetadataOperationTracker * opTracker);
 static void DeleteInProgressAddedFiles(Oid relationId, List *addedFiles);
+static bool AreSchemasEqual(IcebergTableSchema * existingSchema, DataFileSchema * newSchema);
+static int32_t GetSchemaIdForIcebergTableIfExists(const TableMetadataOperationTracker * opTracker, DataFileSchema * schema);
 static int	ComparePartitionSpecsById(const ListCell *a, const ListCell *b);
 
 
@@ -596,7 +598,7 @@ GetDDLMetadataOperations(const TableMetadataOperationTracker * opTracker)
 		TableMetadataOperation *createOp = palloc0(sizeof(TableMetadataOperation));
 
 		createOp->type = TABLE_CREATE;
-		createOp->schema = schema;
+		createOp->newSchema = schema;
 		createOp->partitionSpecs = newPartitionSpecs;
 		createOp->defaultSpecId = defaultSpecId;
 
@@ -611,10 +613,30 @@ GetDDLMetadataOperations(const TableMetadataOperationTracker * opTracker)
 
 	if (opTracker->relationAltered)
 	{
+		/*
+		 * When a table is altered, its schema has changed. However, it might
+		 * have been set to an existing schema in the iceberg metadata, if so,
+		 * use that schemaId.
+		 */
 		TableMetadataOperation *ddlOp = palloc0(sizeof(TableMetadataOperation));
 
 		ddlOp->type = TABLE_DDL;
-		ddlOp->schema = schema;
+
+		int32_t		existingSchemaId =
+			GetSchemaIdForIcebergTableIfExists(opTracker, schema);
+
+		if (existingSchemaId != -1)
+		{
+			ddlOp->ddlSchemaEffect = DDL_EFFECT_SET_EXISTING_SCHEMA;
+			ddlOp->existingSchemaId = existingSchemaId;
+			ddlOp->newSchema = NULL;
+		}
+		else
+		{
+			ddlOp->ddlSchemaEffect = DDL_EFFECT_ADD_SCHEMA;
+			ddlOp->newSchema = schema;
+			ddlOp->existingSchemaId = -1;
+		}
 
 		operations = lappend(operations, ddlOp);
 	}
@@ -645,3 +667,52 @@ ComparePartitionSpecsById(const ListCell *a, const ListCell *b)
 
 	return pg_cmp_s32(specA->spec_id, specB->spec_id);
 }
+
+
+/*
+* GetSchemaIdForIcebergTableIfExists checks if the given schema already exists
+ * in the iceberg table metadata. If it exists, it returns the schema ID, otherwise -1.
+*/
+static int32_t
+GetSchemaIdForIcebergTableIfExists(const TableMetadataOperationTracker * opTracker, DataFileSchema * schema)
+{
+	IcebergTableMetadata *metadata = GetLastPushedIcebergMetadata(opTracker);
+
+	if (metadata == NULL)
+		return -1;
+
+	IcebergTableSchema *schemas = metadata->schemas;
+
+	for (int schemaIndex = 0; schemaIndex < metadata->schemas_length; schemaIndex++)
+	{
+		IcebergTableSchema *existingSchema = &schemas[schemaIndex];
+
+		if (AreSchemasEqual(existingSchema, schema))
+			return schemas[schemaIndex].schema_id;
+
+	}
+
+	return -1;
+}
+
+
+/*
+* AreSchemasEqual compares two schemas for equality.
+*/
+static bool
+AreSchemasEqual(IcebergTableSchema * existingSchema, DataFileSchema * newSchema)
+{
+	if (existingSchema->fields_length != newSchema->nfields)
+		return false;
+
+	for (size_t i = 0; i < existingSchema->fields_length; i++)
+	{
+		DataFileSchemaField *existingField = &existingSchema->fields[i];
+		DataFileSchemaField *newField = &newSchema->fields[i];
+
+		if (!SchemaFieldsEquivalent(existingField, newField))
+			return false;
+	}
+
+	return true;
+}