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 *inCurrentSchema);
 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

@@ -59,7 +59,7 @@ static void *ApplyDayTransformToColumn(IcebergPartitionTransform * transform,
 static void *ApplyHourTransformToColumn(IcebergPartitionTransform * transform,
 										Datum columnValue, bool isNull,
 										size_t *valueSize);
-static IcebergPartitionTransform * GetPartitionTransformFromSpecField(Oid relationId,
+static IcebergPartitionTransform * GetPartitionTransformFromSpecField(Oid relationId, AttrNumber attNum,
 																	  IcebergPartitionSpecField * specField);
 static void ParseTransformName(const char *name, IcebergPartitionTransformType * type,
 							   size_t *bucketCount, size_t *truncateLen);
@@ -196,7 +196,19 @@ GetPartitionTransformsFromSpecFields(Oid relationId, List *specFields)
 	{
 		IcebergPartitionSpecField *specField = lfirst(lc);
 
-		IcebergPartitionTransform *transform = GetPartitionTransformFromSpecField(relationId, specField);
+		bool		inCurrentSchema = false;
+
+		AttrNumber	attNum =
+			GetAttributeForFieldId(relationId, specField->source_id, &inCurrentSchema);
+
+		/*
+		 * skip building transform if there is no corresponding attribute in
+		 * the current postgres schema, which means no pruning is possible
+		 */
+		if (!inCurrentSchema)
+			continue;
+
+		IcebergPartitionTransform *transform = GetPartitionTransformFromSpecField(relationId, attNum, specField);
 
 		Assert(transform != NULL);
 		transformList = lappend(transformList, transform);
@@ -213,16 +225,14 @@ GetPartitionTransformsFromSpecFields(Oid relationId, List *specFields)
 * bucket count, and truncate length for ease of use.
 */
 static IcebergPartitionTransform *
-GetPartitionTransformFromSpecField(Oid relationId, IcebergPartitionSpecField * specField)
+GetPartitionTransformFromSpecField(Oid relationId, AttrNumber attNum, IcebergPartitionSpecField * specField)
 {
 	IcebergPartitionTransform *transform = palloc0(sizeof(IcebergPartitionTransform));
 
 	transform->partitionFieldId = specField->field_id;
 	transform->partitionFieldName = pstrdup(specField->name);
 	transform->transformName = pstrdup(specField->transform);
-
-	transform->attnum =
-		GetAttributeForFieldId(relationId, specField->source_id);
+	transform->attnum = attNum;
 	transform->columnName = get_attname(relationId, transform->attnum, false);
 	transform->pgType = GetAttributePGType(relationId, transform->attnum);
 

pg_lake_table/src/fdw/schema_operations/field_id_mapping_catalog.c

@@ -56,8 +56,8 @@ static void InsertFieldMapping(Oid relationId, int attrIcebergFieldId,
 							   AttrNumber pg_attnum, PGType pgType,
 							   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 GetAttributeForFieldIdForInternalIcebergTable(Oid relationId, int fieldId, bool *inCurrentSchema);
+static AttrNumber GetAttributeForFieldIdForExternalIcebergTable(char *metadataPath, Oid relationId, int fieldId, bool *inCurrentSchema);
 
 #ifdef USE_ASSERT_CHECKING
 static List *GetAllRegisteredAttnumsForTopLevelColumns(Oid relationId);
@@ -244,10 +244,12 @@ 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 inCurrentSchema if there is no attribute in current postgres schema
+ * which corresponds to the field.
+ */
 AttrNumber
-GetAttributeForFieldId(Oid relationId, int fieldId)
+GetAttributeForFieldId(Oid relationId, int fieldId, bool *inCurrentSchema)
 {
 	if (IsInternalIcebergTable(relationId))
 	{
@@ -259,17 +261,18 @@ GetAttributeForFieldId(Oid relationId, int fieldId)
 
 		char	   *currentMetadataPath = GetIcebergMetadataLocation(relationId, false);
 
-		return GetAttributeForFieldIdForExternalIcebergTable(currentMetadataPath, relationId, fieldId);
+		return GetAttributeForFieldIdForExternalIcebergTable(currentMetadataPath, relationId, fieldId, inCurrentSchema);
 	}
 }
 
 
 /*
 * GetAttributeForFieldIdForInternalIcebergTable gets the attribute number for a given field ID
-* for internal Iceberg tables from catalog.
+* for internal Iceberg tables from catalog. It also sets if the corresponding attribute is in the
+* current postgres schema.
 */
 static AttrNumber
-GetAttributeForFieldIdForInternalIcebergTable(Oid relationId, int fieldId)
+GetAttributeForFieldIdForInternalIcebergTable(Oid relationId, int fieldId, bool *inCurrentSchema)
 {
 	DECLARE_SPI_ARGS(2);
 
@@ -288,17 +291,26 @@ GetAttributeForFieldIdForInternalIcebergTable(Oid relationId, int fieldId)
 	 */
 	bool		readOnly = false;
 
-	SPI_EXECUTE("SELECT pg_attnum FROM " MAPPING_TABLE_NAME
+	SPI_EXECUTE("SELECT m.pg_attnum, not attisdropped as in_current_schema "
+				"FROM " MAPPING_TABLE_NAME " m JOIN pg_attribute p ON "
+				" m.table_name OPERATOR(pg_catalog.=) p.attrelid "
+				" AND m.pg_attnum OPERATOR(pg_catalog.=) p.attnum"
 				" WHERE table_name OPERATOR(pg_catalog.=) $1"
 				" AND field_id OPERATOR(pg_catalog.=) $2", readOnly);
 
 	/* there is a primary key on these filters */
 	Assert(SPI_processed == 1);
 
-	bool		isNull = false;
-	AttrNumber	attrNo = GET_SPI_VALUE(INT2OID, 0, 1, &isNull);
+	bool		isAttrNumNull = false;
+	AttrNumber	attrNo = GET_SPI_VALUE(INT2OID, 0, 1, &isAttrNumNull);
+
+	Assert(!isAttrNumNull);
 
-	Assert(!isNull);
+	bool		inCurrentSchemaNull = false;
+
+	*inCurrentSchema = GET_SPI_VALUE(BOOLOID, 0, 2, &inCurrentSchemaNull);
+
+	Assert(!inCurrentSchemaNull);
 
 	SPI_END();
 
@@ -311,7 +323,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 *inCurrentSchema)
 {
 	DataFileSchema *schema = GetDataFileSchemaForExternalIcebergTable(metadataPath);
 
@@ -338,16 +350,20 @@ GetAttributeForFieldIdForExternalIcebergTable(char *metadataPath, Oid relationId
 
 	SPI_EXECUTE("SELECT attnum FROM pg_attribute "
 				"WHERE attrelid OPERATOR(pg_catalog.=) $1 "
-				"AND attname OPERATOR(pg_catalog.=) $2 "
-				"AND NOT attisdropped", readOnly);
+				"AND attname OPERATOR(pg_catalog.=) $2", readOnly);
 
-	/* there is a primary key on these filters */
-	Assert(SPI_processed == 1);
+	AttrNumber	attrNo = 0;
 
-	bool		isNull = false;
-	AttrNumber	attrNo = GET_SPI_VALUE(INT2OID, 0, 1, &isNull);
-
-	Assert(!isNull);
+	if (SPI_processed == 0)
+	{
+		*inCurrentSchema = false;
+		attrNo = -1;
+	}
+	else
+	{
+		Assert(SPI_processed == 1);
+		attrNo = GET_SPI_VALUE(INT2OID, 0, 1, inCurrentSchema);
+	}
 
 	SPI_END();