Utils to differentiate internal and external tables

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

Author: sfc-gh-abozkurt

pg_lake_engine/include/pg_lake/pgduck/remote_storage.h

@@ -38,7 +38,6 @@ typedef struct RemoteFileDesc
 
 extern PGDLLEXPORT int64 GetRemoteFileSize(char *path);
 extern PGDLLEXPORT int64 GetRemoteParquetFileRowCount(char *path);
-extern PGDLLEXPORT List *GetRemoteParquetColumnStats(char *path, List *leafFields);
 extern PGDLLEXPORT List *ListRemoteFileDescriptions(char *pattern);
 extern PGDLLEXPORT List *ListRemoteFileNames(char *pattern);
 extern PGDLLEXPORT bool RemoteFileExists(char *path);

pg_lake_engine/include/pg_lake/util/rel_utils.h

@@ -67,6 +67,7 @@ struct PgLakeTableProperties;
 #define PG_LAKE_ICEBERG_SERVER_NAME "pg_lake_iceberg"
 
 extern PGDLLEXPORT bool IsAnyLakeForeignTableById(Oid foreignTableId);
+extern PGDLLEXPORT char *GetForeignTablePath(Oid foreignTableId);
 extern PGDLLEXPORT char *GetQualifiedRelationName(Oid relationId);
 extern PGDLLEXPORT const char *PgLakeTableTypeToName(PgLakeTableType tableType);
 extern PGDLLEXPORT PgLakeTableType GetPgLakeTableType(Oid foreignTableId);
@@ -80,7 +81,6 @@ extern PGDLLEXPORT bool IsPgLakeIcebergServerName(const char *serverName);
 extern PGDLLEXPORT char *GetWritableTableLocation(Oid relationId, char **queryArguments);
 extern PGDLLEXPORT void EnsureTableOwner(Oid relationId);
 extern PGDLLEXPORT struct PgLakeTableProperties GetPgLakeTableProperties(Oid relationId);
-extern PGDLLEXPORT bool IsInternalOrExternalIcebergTable(struct PgLakeTableProperties properties);
 
 /* range var help */
 extern PGDLLEXPORT List *MakeNameListFromRangeVar(const RangeVar *rel);

pg_lake_engine/src/utils/rel_utils.c

@@ -256,6 +256,32 @@ GetQualifiedRelationName(Oid relationId)
 	return quote_qualified_identifier(namespaceName, relationName);
 }
 
+
+/*
+* GetForeignTablePath - get the path option for the foreign table.
+*/
+char *
+GetForeignTablePath(Oid foreignTableId)
+{
+	ForeignTable *fTable = GetForeignTable(foreignTableId);
+	ListCell   *cell;
+
+	foreach(cell, fTable->options)
+	{
+		DefElem    *defel = (DefElem *) lfirst(cell);
+
+		if (strcmp(defel->defname, "path") == 0)
+		{
+			return defGetString(defel);
+		}
+	}
+
+	ereport(ERROR,
+			(errcode(ERRCODE_FDW_UNABLE_TO_CREATE_EXECUTION),
+			 errmsg("path option not found for foreign table %u", foreignTableId)));
+}
+
+
 /*
  * PgLakeTableTypeToName - convert the PgLakeTableType to a string.
  */
@@ -360,19 +386,6 @@ GetPgLakeTableProperties(Oid relationId)
 }
 
 
-/*
-* IsInternalOrExternalIcebergTable - check if the table is an internal or
-* external iceberg table.
-*/
-bool
-IsInternalOrExternalIcebergTable(PgLakeTableProperties properties)
-{
-	PgLakeTableType tableType = properties.tableType;
-
-	return tableType == PG_LAKE_ICEBERG_TABLE_TYPE ||
-		(tableType == PG_LAKE_TABLE_TYPE && properties.format == DATA_FORMAT_ICEBERG);
-}
-
 /*
  * MakeNameListFromRangeVar makes a namelist from a RangeVar. Its behaviour
  * should be the exact opposite of postgres' makeRangeVarFromNameList.

pg_lake_iceberg/include/pg_lake/iceberg/catalog.h

@@ -19,6 +19,8 @@
 
 #include "nodes/primnodes.h"
 
+#include "pg_lake/util/rel_utils.h"
+
 /*
  * Default location prefix for pg_lake_iceberg tables. Used when the location
  * option is not specified at "CREATE FOREIGN SERVER pg_lake_iceberg OPTIONS ()".
@@ -53,6 +55,8 @@ extern PGDLLEXPORT void DeleteInternalIcebergCatalogTable(Oid relationId);
 
 extern PGDLLEXPORT List *GetAllInternalIcebergRelationIds(void);
 extern PGDLLEXPORT char *GetIcebergCatalogMetadataLocation(Oid relationId, bool forUpdate);
+extern PGDLLEXPORT char *GetIcebergExternalMetadataLocation(Oid relationId);
+extern PGDLLEXPORT IcebergCatalogType GetIcebergCatalogType(Oid relationId);
 extern PGDLLEXPORT char *GetIcebergCatalogPreviousMetadataLocation(Oid relationId, bool forUpdate);
 extern PGDLLEXPORT void UpdateExternalCatalogMetadataLocation(char *catalogName, char *schemaName, char *tableName, const char *metadataLocation,
 															  const char *previousMetadataLocation);

pg_lake_iceberg/include/pg_lake/rest_catalog/rest_catalog.h

@@ -36,7 +36,6 @@ extern char *RestCatalogClientSecret;
 extern PGDLLEXPORT char *RestCatalogFetchAccessToken(void);
 extern PGDLLEXPORT void RegisterNamespaceToRestCatalog(const char *catalogName, const char *namespaceName);
 extern PGDLLEXPORT void ErrorIfRestNamespaceDoesNotExist(const char *catalogName, const char *namespaceName);
-extern PGDLLEXPORT IcebergCatalogType GetIcebergCatalogType(Oid relationId);
 extern PGDLLEXPORT char *GetRestCatalogName(Oid relationId);
 extern PGDLLEXPORT char *GetRestCatalogNamespace(Oid relationId);
 extern PGDLLEXPORT char *GetRestCatalogTableName(Oid relationId);

pg_lake_iceberg/src/iceberg/catalog.c

@@ -18,13 +18,15 @@
 #include "postgres.h"
 #include "miscadmin.h"
 
-#include "catalog/namespace.h"
+#include "pg_lake/extensions/pg_lake_iceberg.h"
 #include "pg_lake/iceberg/catalog.h"
+#include "pg_lake/object_store_catalog/object_store_catalog.h"
 #include "pg_lake/rest_catalog/rest_catalog.h"
-#include "pg_lake/extensions/pg_lake_iceberg.h"
 #include "pg_lake/util/rel_utils.h"
 #include "pg_lake/util/spi_helpers.h"
+#include "catalog/namespace.h"
 #include "commands/dbcommands.h"
+#include "foreign/foreign.h"
 #include "utils/lsyscache.h"
 #include "utils/guc.h"
 
@@ -354,6 +356,70 @@ GetIcebergCatalogMetadataLocation(Oid relationId, bool forUpdate)
 	return GetIcebergCatalogMetadataLocationInternal(relationId, false, forUpdate);
 }
 
+
+/*
+ * GetIcebergExternalMetadataLocation returns the metadata location for an external iceberg table.
+ */
+char *
+GetIcebergExternalMetadataLocation(Oid relationId)
+{
+	IcebergCatalogType icebergCatalogType = GetIcebergCatalogType(relationId);
+
+	char	   *currentMetadataPath = NULL;
+
+	if (icebergCatalogType == REST_CATALOG_READ_ONLY)
+	{
+		currentMetadataPath = GetMetadataLocationForRestCatalogForIcebergTable(relationId);
+	}
+	else if (icebergCatalogType == OBJECT_STORE_READ_ONLY)
+	{
+		currentMetadataPath = GetMetadataLocationFromExternalObjectStoreCatalogForTable(relationId);
+	}
+	else
+	{
+		currentMetadataPath = GetForeignTablePath(relationId);
+	}
+
+	return currentMetadataPath;
+}
+
+
+IcebergCatalogType
+GetIcebergCatalogType(Oid relationId)
+{
+	if (!IsPgLakeIcebergForeignTableById(relationId))
+		return NOT_ICEBERG_TABLE;
+
+	ForeignTable *foreignTable = GetForeignTable(relationId);
+	List	   *options = foreignTable->options;
+
+	bool		hasRestCatalogOption = HasRestCatalogTableOption(options);
+	bool		hasObjectStoreCatalogOption = HasObjectStoreCatalogTableOption(options);
+	bool		hasReadOnlyOption = HasReadOnlyOption(options);
+
+	if (hasRestCatalogOption && hasReadOnlyOption)
+	{
+		return REST_CATALOG_READ_ONLY;
+	}
+	else if (hasRestCatalogOption && !hasReadOnlyOption)
+	{
+		return REST_CATALOG_READ_WRITE;
+	}
+	else if (hasObjectStoreCatalogOption && hasReadOnlyOption)
+	{
+		return OBJECT_STORE_READ_ONLY;
+	}
+	else if (hasObjectStoreCatalogOption && !hasReadOnlyOption)
+	{
+		return OBJECT_STORE_READ_WRITE;
+	}
+	else
+	{
+		return POSTGRES_CATALOG;
+	}
+}
+
+
 /*
 * GetIcebergCatalogPreviousMetadataLocation returns the previous metadata location for a table
 * in the iceberg catalog table. Returns NULL if the record is not found.

pg_lake_iceberg/src/rest_catalog/rest_catalog.c

@@ -449,42 +449,6 @@ EncodeBasicAuth(const char *clientId, const char *clientSecret)
 }
 
 
-IcebergCatalogType
-GetIcebergCatalogType(Oid relationId)
-{
-	if (!IsPgLakeIcebergForeignTableById(relationId))
-		return NOT_ICEBERG_TABLE;
-
-	ForeignTable *foreignTable = GetForeignTable(relationId);
-	List	   *options = foreignTable->options;
-
-	bool		hasRestCatalogOption = HasRestCatalogTableOption(options);
-	bool		hasObjectStoreCatalogOption = HasObjectStoreCatalogTableOption(options);
-	bool		hasReadOnlyOption = HasReadOnlyOption(options);
-
-	if (hasRestCatalogOption && hasReadOnlyOption)
-	{
-		return REST_CATALOG_READ_ONLY;
-	}
-	else if (hasRestCatalogOption && !hasReadOnlyOption)
-	{
-		return REST_CATALOG_READ_WRITE;
-	}
-	else if (hasObjectStoreCatalogOption && hasReadOnlyOption)
-	{
-		return OBJECT_STORE_READ_ONLY;
-	}
-	else if (hasObjectStoreCatalogOption && !hasReadOnlyOption)
-	{
-		return OBJECT_STORE_READ_WRITE;
-	}
-	else
-	{
-		return POSTGRES_CATALOG;
-	}
-}
-
-
 /*
 * Readable rest catalog tables always use the catalog_table_name option
 * as the table name in the external catalog. Writable rest catalog tables

pg_lake_table/include/pg_lake/fdw/utils.h

@@ -33,8 +33,10 @@
 #include "pg_lake/copy/copy_format.h"
 
 extern PGDLLEXPORT bool IsAnyLakeForeignTable(RangeTblEntry *rte);
-extern PGDLLEXPORT char *GetForeignTablePath(Oid foreignTableId);
 extern PGDLLEXPORT CopyDataFormat GetForeignTableFormat(Oid foreignTableId);
+extern PGDLLEXPORT bool IsAnyIcebergTable(Oid relationId);
+extern PGDLLEXPORT bool IsAnyInternalIcebergTable(Oid relationId);
+extern PGDLLEXPORT bool IsAnyExternalIcebergTable(Oid relationId);
 extern PGDLLEXPORT bool IsWritablePgLakeTable(Oid relationId);
 extern PGDLLEXPORT void ErrorIfTypeUnsupportedForIcebergTables(Oid typeOid, int32 typmod, char *columnName);
 extern PGDLLEXPORT void ErrorIfTypeUnsupportedNumericForIcebergTables(int32 typmod, char *columnName);

pg_lake_table/src/fdw/data_file_pruning.c

@@ -50,10 +50,12 @@
 #include "pg_lake/extensions/postgis.h"
 #include "pg_lake/fdw/schema_operations/field_id_mapping_catalog.h"
 #include "pg_lake/fdw/schema_operations/register_field_ids.h"
+#include "pg_lake/fdw/utils.h"
 #include "pg_lake/fdw/data_file_pruning.h"
 #include "pg_lake/fdw/partition_transform.h"
 #include "pg_lake/fdw/writable_table.h"
 #include "pg_lake/iceberg/api/table_schema.h"
+#include "pg_lake/iceberg/catalog.h"
 #include "pg_lake/iceberg/data_file_stats.h"
 #include "pg_lake/iceberg/partitioning/partition.h"
 #include "pg_lake/iceberg/temporal_utils.h"
@@ -183,11 +185,9 @@ PruneDataFiles(Oid relationId, List *dataFiles, List *baseRestrictInfoList, Prun
 	List	   *columnsUsedInFilters = ColumnsUsedInRestrictions(relationId, baseRestrictInfoList);
 	List	   *partitionTransforms = AllPartitionTransformList(relationId);
 	PgLakeTableProperties tableProperties = GetPgLakeTableProperties(relationId);
-	PgLakeTableType tableType = tableProperties.tableType;
-	IcebergCatalogType icebergCatalogType = GetIcebergCatalogType(relationId);
 
 	if ((!EnableDataFilePruning && !EnablePartitionPruning) ||
-		!IsInternalOrExternalIcebergTable(tableProperties))
+		!IsAnyIcebergTable(relationId))
 	{
 		/*
 		 * User disabled or no columns used in filters, so we cannot prune any
@@ -239,20 +239,15 @@ PruneDataFiles(Oid relationId, List *dataFiles, List *baseRestrictInfoList, Prun
 		List	   *columnStats = NIL;
 		Partition  *partition = NULL;
 
-		if (tableType == PG_LAKE_ICEBERG_TABLE_TYPE &&
-			(icebergCatalogType == POSTGRES_CATALOG || icebergCatalogType == REST_CATALOG_READ_WRITE ||
-			 icebergCatalogType == OBJECT_STORE_READ_WRITE))
+		if (IsAnyInternalIcebergTable(relationId))
 		{
 			TableDataFile *tableDataFile = (TableDataFile *) list_nth(dataFiles, dataFileIndex);
 
 			Assert(tableDataFile->content == CONTENT_DATA);
 			columnStats = tableDataFile->stats.columnStats;
 			partition = tableDataFile->partition;
 		}
-		else if ((tableType == PG_LAKE_TABLE_TYPE && tableProperties.format == DATA_FORMAT_ICEBERG) ||
-				 (tableType == PG_LAKE_ICEBERG_TABLE_TYPE &&
-				  (icebergCatalogType == REST_CATALOG_READ_ONLY ||
-				   icebergCatalogType == OBJECT_STORE_READ_ONLY)))
+		else if (IsAnyExternalIcebergTable(relationId))
 		{
 			DataFile   *dataFile = (DataFile *) list_nth(dataFiles, dataFileIndex);
 
@@ -334,7 +329,6 @@ AddFieldIdsUsedInQuery(HTAB *fieldIdsUsedInQuery, Oid relationId, PgLakeTablePro
 					   List *columnsUsedInFilters)
 {
 	PgLakeTableType tableType = tableProperties.tableType;
-	IcebergCatalogType icebergCatalogType = GetIcebergCatalogType(relationId);
 
 	List	   *attrNos = NIL;
 	ListCell   *columnCell = NULL;
@@ -350,9 +344,7 @@ AddFieldIdsUsedInQuery(HTAB *fieldIdsUsedInQuery, Oid relationId, PgLakeTablePro
 	/* fetch the field mappings for all columns in a single catalog lookup */
 	List	   *fields = NIL;
 
-	if (tableType == PG_LAKE_ICEBERG_TABLE_TYPE &&
-		(icebergCatalogType == POSTGRES_CATALOG || icebergCatalogType == REST_CATALOG_READ_WRITE ||
-		 icebergCatalogType == OBJECT_STORE_READ_WRITE))
+	if (IsAnyInternalIcebergTable(relationId))
 	{
 		fields = GetRegisteredFieldForAttributes(relationId, attrNos);
 
@@ -362,9 +354,7 @@ AddFieldIdsUsedInQuery(HTAB *fieldIdsUsedInQuery, Oid relationId, PgLakeTablePro
 		 */
 		Assert(list_length(fields) == list_length(columnsUsedInFilters));
 	}
-	else if ((tableType == PG_LAKE_TABLE_TYPE && tableProperties.format == DATA_FORMAT_ICEBERG) ||
-			 (tableType == PG_LAKE_ICEBERG_TABLE_TYPE && (icebergCatalogType == REST_CATALOG_READ_ONLY ||
-														  icebergCatalogType == OBJECT_STORE_READ_ONLY)))
+	else if (IsAnyExternalIcebergTable(relationId))
 	{
 		fields = GetExternalIcebergFieldsForAttributes(relationId, columnsUsedInFilters);
 

pg_lake_table/src/fdw/schema_operations/register_field_ids.c

@@ -237,21 +237,9 @@ CreatePostgresColumnMappingsForColumnDefs(Oid relationId, List *columnDefList, b
 List *
 CreatePostgresColumnMappingsForIcebergTableFromExternalMetadata(Oid relationId)
 {
-	char	   *currentMetadataPath = NULL;
 	IcebergCatalogType icebergCatalogType = GetIcebergCatalogType(relationId);
 
-	if (icebergCatalogType == REST_CATALOG_READ_ONLY)
-	{
-		currentMetadataPath = GetMetadataLocationForRestCatalogForIcebergTable(relationId);
-	}
-	else if (icebergCatalogType == OBJECT_STORE_READ_ONLY)
-	{
-		currentMetadataPath = GetMetadataLocationFromExternalObjectStoreCatalogForTable(relationId);
-	}
-	else
-	{
-		currentMetadataPath = GetIcebergCatalogMetadataLocation(relationId, true);
-	}
+	char	   *currentMetadataPath = GetIcebergExternalMetadataLocation(relationId);
 
 	DataFileSchema *schema = GetDataFileSchemaForExternalIcebergTable(currentMetadataPath);