skip building transform if external table spec field is from old schema

Signed-off-by: Aykut Bozkurt <[email protected]>

Author: sfc-gh-abozkurt

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

@@ -31,7 +31,7 @@ extern PGDLLEXPORT DataFileSchema * GetDataFileSchemaForInternalIcebergTable(Oid
 extern PGDLLEXPORT List *GetLeafFieldsForInternalIcebergTable(Oid relationId);
 extern PGDLLEXPORT List *GetRegisteredFieldForAttributes(Oid relationId, List *attrNos);
 extern PGDLLEXPORT DataFileSchemaField * GetRegisteredFieldForAttribute(Oid relationId, AttrNumber attrNo);
-extern PGDLLEXPORT AttrNumber GetAttributeForFieldId(Oid relationId, int fieldId);
+extern PGDLLEXPORT AttrNumber GetAttributeForFieldId(Oid relationId, int fieldId, bool *attributeExists);
 extern PGDLLEXPORT void UpdateRegisteredFieldWriteDefaultForAttribute(Oid relationId, AttrNumber attNum, const char *writeDefault);
 extern PGDLLEXPORT int GetLargestRegisteredFieldId(Oid relationId);
 extern PGDLLEXPORT void RegisterIcebergColumnMapping(Oid relationId, Field * field,

pg_lake_table/src/fdw/partition_transform.c

@@ -198,8 +198,8 @@ GetPartitionTransformsFromSpecFields(Oid relationId, List *specFields)
 
 		IcebergPartitionTransform *transform = GetPartitionTransformFromSpecField(relationId, specField);
 
-		Assert(transform != NULL);
-		transformList = lappend(transformList, transform);
+		if (transform != NULL)
+			transformList = lappend(transformList, transform);
 	}
 
 	return transformList;
@@ -221,19 +221,32 @@ GetPartitionTransformFromSpecField(Oid relationId, IcebergPartitionSpecField * s
 	transform->partitionFieldName = pstrdup(specField->name);
 	transform->transformName = pstrdup(specField->transform);
 
+	bool		attributeExists = false;
+
 	transform->attnum =
-		GetAttributeForFieldId(relationId, specField->source_id);
+		GetAttributeForFieldId(relationId, specField->source_id, &attributeExists);
+
+	if (!attributeExists)
+	{
+		/*
+		 * skip this transform. Pruning wont apply anyway because postgres
+		 * schema (current iceberg schema) misses it.
+		 */
+		Assert(IsAnyExternalIcebergTable(relationId));
+		return NULL;
+	}
+
 	transform->columnName = get_attname(relationId, transform->attnum, false);
 	transform->pgType = GetAttributePGType(relationId, transform->attnum);
 
-	Assert(IsAnyIcebergTable(relationId));
-
 	if (IsAnyInternalIcebergTable(relationId))
 	{
 		transform->sourceField = GetRegisteredFieldForAttribute(relationId, transform->attnum);
 	}
 	else
 	{
+		Assert(IsAnyExternalIcebergTable(relationId));
+
 		DataFileSchema *schema = GetDataFileSchemaForTable(relationId);
 
 		transform->sourceField = GetDataFileSchemaFieldById(schema, specField->source_id);

pg_lake_table/src/fdw/schema_operations/field_id_mapping_catalog.c

@@ -57,7 +57,7 @@ static void InsertFieldMapping(Oid relationId, int attrIcebergFieldId,
 							   const char *writeDefault, const char *initialDefault,
 							   int parentFieldId);
 static AttrNumber GetAttributeForFieldIdForInternalIcebergTable(Oid relationId, int fieldId);
-static AttrNumber GetAttributeForFieldIdForExternalIcebergTable(char *metadataPath, Oid relationId, int fieldId);
+static AttrNumber GetAttributeForFieldIdForExternalIcebergTable(char *metadataPath, Oid relationId, int fieldId, bool *attributeExists);
 
 #ifdef USE_ASSERT_CHECKING
 static List *GetAllRegisteredAttnumsForTopLevelColumns(Oid relationId);
@@ -244,20 +244,25 @@ GetRegisteredFieldForAttribute(Oid relationId, AttrNumber attrNo)
 }
 
 /*
-* GetAttributeForFieldId gets the attribute number for a given field ID.
-*/
+ * GetAttributeForFieldId gets the attribute number for a given field ID.
+ * It sets attributeExists if there is no attribute which corresponds to
+ * given field id.
+ */
 AttrNumber
-GetAttributeForFieldId(Oid relationId, int fieldId)
+GetAttributeForFieldId(Oid relationId, int fieldId, bool *attributeExists)
 {
 	Assert(IsAnyIcebergTable(relationId));
 
 	if (IsAnyInternalIcebergTable(relationId))
+	{
+		*attributeExists = true;
 		return GetAttributeForFieldIdForInternalIcebergTable(relationId, fieldId);
+	}
 	else
 	{
 		char	   *currentMetadataPath = GetIcebergMetadataLocation(relationId, false);
 
-		return GetAttributeForFieldIdForExternalIcebergTable(currentMetadataPath, relationId, fieldId);
+		return GetAttributeForFieldIdForExternalIcebergTable(currentMetadataPath, relationId, fieldId, attributeExists);
 	}
 }
 
@@ -309,7 +314,7 @@ GetAttributeForFieldIdForInternalIcebergTable(Oid relationId, int fieldId)
 * for external Iceberg tables from iceberg metadata.
  */
 static AttrNumber
-GetAttributeForFieldIdForExternalIcebergTable(char *metadataPath, Oid relationId, int fieldId)
+GetAttributeForFieldIdForExternalIcebergTable(char *metadataPath, Oid relationId, int fieldId, bool *attributeExists)
 {
 	DataFileSchema *schema = GetDataFileSchemaForExternalIcebergTable(metadataPath);
 
@@ -342,10 +347,9 @@ GetAttributeForFieldIdForExternalIcebergTable(char *metadataPath, Oid relationId
 	/* there is a primary key on these filters */
 	Assert(SPI_processed == 1);
 
-	bool		isNull = false;
-	AttrNumber	attrNo = GET_SPI_VALUE(INT2OID, 0, 1, &isNull);
+	AttrNumber	attrNo = GET_SPI_VALUE(INT2OID, 0, 1, attributeExists);
 
-	Assert(!isNull);
+	Assert(attributeExists || IsAnyExternalIcebergTable(relationId));
 
 	SPI_END();