Refactor IcebergPartitionTransform

IcebergPartitionTransform can embed a IcebergPartitionSpecField by removing
some individual fields, which makes it less verbose.

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

Author: sfc-gh-abozkurt

pg_lake_engine/src/data_file/data_file_stats.c

@@ -256,7 +256,7 @@ ExtractMinMaxForColumn(Datum map, char *colName, List **names, List **mins, List
 
 	if (minText != NULL && maxText != NULL)
 	{
-		*names = lappend(*names, colName);
+		*names = lappend(*names, pstrdup(colName));
 		*mins = lappend(*mins, minText);
 		*maxs = lappend(*maxs, maxText);
 	}

pg_lake_iceberg/include/pg_lake/iceberg/api/partitioning.h

@@ -19,6 +19,7 @@
 
 #include "postgres.h"
 
+#include "pg_lake/iceberg/metadata_spec.h"
 #include "pg_lake/parquet/field.h"
 #include "pg_lake/pgduck/type.h"
 #include "access/attnum.h"
@@ -51,14 +52,7 @@ typedef struct IcebergPartitionTransform
 		size_t		truncateLen;
 	};
 
-	/* partition field id */
-	int32_t		partitionFieldId;
-
-	/* <columnName>_<transformName>, e.g. a_bucket */
-	const char *partitionFieldName;
-
-	/* transform name, e.g. bucket[3] */
-	const char *transformName;
+	IcebergPartitionSpecField *specField;
 
 	/* source field of the column to which transform applies */
 	DataFileSchemaField *sourceField;

pg_lake_iceberg/src/iceberg/partitioning/partition.c

@@ -225,7 +225,7 @@ FindPartitionTransformById(List *transforms, int32_t partitionFieldId, bool erro
 	{
 		IcebergPartitionTransform *transform = (IcebergPartitionTransform *) lfirst(cell);
 
-		if (transform->partitionFieldId == partitionFieldId)
+		if (transform->specField->field_id == partitionFieldId)
 			return transform;
 	}
 

pg_lake_iceberg/src/iceberg/partitioning/spec_generation.c

@@ -60,27 +60,7 @@ BuildPartitionSpecFromPartitionTransforms(Oid relationId, List *partitionTransfo
 	{
 		IcebergPartitionTransform *transform = lfirst(transformCell);
 
-		IcebergPartitionSpecField *field = palloc0(sizeof(IcebergPartitionSpecField));
-
-		field->source_id = transform->sourceField->id;
-
-		/*
-		 * We do not support partition transforms on multi columns (v3
-		 * feature), and to comply with the iceberg spec/reference
-		 * implementation for v2, we still fill the source_ids array.
-		 */
-		field->source_ids_length = 1;
-		field->source_ids = palloc0(sizeof(int) * field->source_ids_length);
-		field->source_ids[0] = transform->sourceField->id;
-
-		field->field_id = transform->partitionFieldId;
-
-		field->name = pstrdup(transform->partitionFieldName);
-		field->name_length = strlen(transform->partitionFieldName);
-		field->transform = pstrdup(transform->transformName);
-		field->transform_length = strlen(transform->transformName);
-
-		spec->fields[fieldIndex] = *field;
+		spec->fields[fieldIndex] = *(transform->specField);
 		fieldIndex++;
 	}
 

pg_lake_table/include/pg_lake/partitioning/partition_spec_catalog.h

@@ -43,7 +43,6 @@ typedef struct IcebergPartitionSpecHashEntry
 extern void UpdateDefaultPartitionSpecId(Oid relationId, int specId);
 extern void InsertPartitionSpecAndPartitionFields(Oid relationId, IcebergPartitionSpec * spec);
 extern int	GetLargestSpecId(Oid relationId);
-extern List *GetAllIcebergPartitionSpecIds(Oid relationId);
 extern PGDLLEXPORT int GetCurrentSpecId(Oid relationId);
 extern int	GetLargestPartitionFieldId(Oid relationId);
 extern IcebergPartitionSpecField * GetIcebergPartitionFieldFromCatalog(Oid relationId, int fieldId);

pg_lake_table/src/fdw/partition_transform.c

@@ -162,7 +162,7 @@ PartitionTransformsEqual(IcebergPartitionSpec * spec, List *partitionTransforms)
 		 * Iceberg does here:
 		 * https://github.com/apache/iceberg/blob/8b55ac834015ce664f879ecfe1e80a941a994420/api/src/main/java/org/apache/iceberg/PartitionSpec.java#L239-L259
 		 */
-		if (strcasecmp(specField->name, transform->partitionFieldName) != 0)
+		if (strcasecmp(specField->name, transform->specField->name) != 0)
 		{
 			return false;
 		}
@@ -251,9 +251,7 @@ GetPartitionTransformFromSpecField(Oid relationId, IcebergPartitionSpecField * s
 {
 	IcebergPartitionTransform *transform = palloc0(sizeof(IcebergPartitionTransform));
 
-	transform->partitionFieldId = specField->field_id;
-	transform->partitionFieldName = pstrdup(specField->name);
-	transform->transformName = pstrdup(specField->transform);
+	transform->specField = specField;
 
 	transform->attnum =
 		GetAttributeForFieldId(relationId, specField->source_id);
@@ -274,7 +272,7 @@ GetPartitionTransformFromSpecField(Oid relationId, IcebergPartitionSpecField * s
 	}
 
 	/* parse transform name */
-	ParseTransformName(transform->transformName,
+	ParseTransformName(transform->specField->transform,
 					   &transform->type,
 					   &transform->bucketCount,
 					   &transform->truncateLen);
@@ -413,8 +411,8 @@ ApplyPartitionTransformToTuple(IcebergPartitionTransform * transform, TupleTable
 {
 	PartitionField *field = palloc0(sizeof(PartitionField));
 
-	field->field_name = pstrdup(transform->partitionFieldName);
-	field->field_id = transform->partitionFieldId;
+	field->field_name = pstrdup(transform->specField->name);
+	field->field_id = transform->specField->field_id;
 
 	bool		isNull = false;
 	Datum		columnValue = slot_getattr(slot, transform->attnum, &isNull);
@@ -453,7 +451,7 @@ ApplyPartitionTransformToTuple(IcebergPartitionTransform * transform, TupleTable
 			ereport(ERROR,
 					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 					 errmsg("applying transform %s is not yet support ",
-							transform->transformName)));
+							transform->specField->transform)));
 	}
 
 	field->value_type = GetTransformResultAvroType(transform);

pg_lake_table/src/fdw/partitioning/partition_by_parser.c

@@ -529,14 +529,24 @@ AnalyzeIcebergTablePartitionBy(Oid relationId, List *transforms)
 
 		transform->sourceField = sourceField;
 
+		transform->specField = palloc0(sizeof(IcebergPartitionSpecField));
+
 		/* set transform name */
-		transform->transformName = GenerateTransformName(transform);
+		transform->specField->transform = GenerateTransformName(transform);
+		transform->specField->transform_length = strlen(transform->specField->transform);
 
 		/* set partition field name */
-		transform->partitionFieldName = GeneratePartitionFieldName(transform, relationId);
+		transform->specField->name = GeneratePartitionFieldName(transform, relationId);
+		transform->specField->name_length = strlen(transform->specField->name);
 
 		/* set partition field id */
-		transform->partitionFieldId = ++largestPartitionFieldId;
+		transform->specField->field_id = ++largestPartitionFieldId;
+
+		/* set source field id */
+		transform->specField->source_id = sourceField->id;
+		transform->specField->source_ids_length = 1;
+		transform->specField->source_ids = palloc0(sizeof(int) * transform->specField->source_ids_length);
+		transform->specField->source_ids[0] = transform->specField->source_id;
 
 		/* 3) Check column type compatibility. */
 		EnsureValidTypeForTransform(transform->type, transform->pgType.postgresTypeOid);
@@ -822,7 +832,7 @@ EnsureNoDuplicateTransforms(List *transforms)
 				ereport(ERROR,
 						(errcode(ERRCODE_DUPLICATE_OBJECT),
 						 errmsg("\"%s\" transform on column \"%s\" appears multiple times in partition spec",
-								transform->transformName, transform->columnName)));
+								transform->specField->transform, transform->columnName)));
 		}
 	}
 }