Skip to content

Refactor IcebergPartitionTransform #141

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sfc-gh-abozkurt wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Refactor IcebergPartitionTransform #141

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
48d8555
Refactor IcebergPartitionTransform
sfc-gh-abozkurt Jan 9, 2026
a905847
separate structs for parse and analyze
sfc-gh-abozkurt Jan 12, 2026
d7db991
deep copy spec field
sfc-gh-abozkurt Jan 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions pg_lake_iceberg/include/pg_lake/iceberg/api/partitioning.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -38,7 +39,8 @@ typedef enum IcebergPartitionTransformType
PARTITION_TRANSFORM_VOID
} IcebergPartitionTransformType;

typedef struct IcebergPartitionTransform
/* Represents a parsed partition transform from table's partition_by string option. */
typedef struct ParsedIcebergPartitionTransform
{
IcebergPartitionTransformType type;

Expand All @@ -51,20 +53,22 @@ typedef struct IcebergPartitionTransform
size_t truncateLen;
};

/* partition field id */
int32_t partitionFieldId;
const char *columnName;
} ParsedIcebergPartitionTransform;

/* <columnName>_<transformName>, e.g. a_bucket */
const char *partitionFieldName;
/* Represents an analyzed partition transform with all necessary info. */
typedef struct IcebergPartitionTransform
{
/* parsed transform info */
ParsedIcebergPartitionTransform parsedTransform;

/* transform name, e.g. bucket[3] */
const char *transformName;
/* spec field info */
IcebergPartitionSpecField *specField;

/* source field of the column to which transform applies */
DataFileSchemaField *sourceField;

/* Postgres column info to which transform applies */
const char *columnName;
AttrNumber attnum;
PGType pgType;

Expand Down
1 change: 1 addition & 0 deletions pg_lake_iceberg/include/pg_lake/iceberg/metadata_spec.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -292,3 +292,4 @@ extern PGDLLEXPORT IcebergTableMetadata * ReadIcebergTableMetadata(const char *t
extern PGDLLEXPORT char *WriteIcebergTableMetadataToJson(IcebergTableMetadata * metadata);
extern PGDLLEXPORT void AppendIcebergTableSchemaForRestCatalog(StringInfo command, IcebergTableSchema * schemas, size_t schemas_length);
extern PGDLLEXPORT void AppendIcebergPartitionSpecFields(StringInfo command, IcebergPartitionSpecField * fields, size_t fields_length);
extern PGDLLEXPORT IcebergPartitionSpecField * DeepCopyIcebergPartitionSpecField(const IcebergPartitionSpecField * field);
2 changes: 1 addition & 1 deletion pg_lake_iceberg/src/iceberg/partitioning/partition.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}

Expand Down
22 changes: 1 addition & 21 deletions pg_lake_iceberg/src/iceberg/partitioning/spec_generation.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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] = *(DeepCopyIcebergPartitionSpecField(transform->specField));
fieldIndex++;
}

Expand Down
25 changes: 25 additions & 0 deletions pg_lake_iceberg/src/iceberg/write_table_metadata.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -764,6 +764,31 @@ AppendIcebergPartitionSpecFields(StringInfo command, IcebergPartitionSpecField *
appendStringInfoString(command, "]");
}

/*
* DeepCopyIcebergPartitionSpecField deep copies a IcebergPartitionSpecField.
*/
IcebergPartitionSpecField *
DeepCopyIcebergPartitionSpecField(const IcebergPartitionSpecField * field)
{
IcebergPartitionSpecField *copiedField = palloc0(sizeof(IcebergPartitionSpecField));

copiedField->field_id = field->field_id;
copiedField->name = pstrdup(field->name);
copiedField->name_length = field->name_length;
copiedField->transform = pstrdup(field->transform);
copiedField->transform_length = field->transform_length;

copiedField->source_id = field->source_id;
copiedField->source_ids_length = field->source_ids_length;
if (field->source_ids_length > 0)
{
copiedField->source_ids = palloc0(field->source_ids_length * sizeof(int32));
memcpy(copiedField->source_ids, field->source_ids, field->source_ids_length * sizeof(int32));
}

return copiedField;
}

static void
AppendIcebergSortOrderFields(StringInfo command, IcebergSortOrderField * fields, size_t fields_length)
{
Expand Down
1 change: 0 additions & 1 deletion pg_lake_table/include/pg_lake/partitioning/partition_spec_catalog.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
14 changes: 7 additions & 7 deletions pg_lake_table/src/fdw/data_file_pruning.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -699,7 +699,7 @@ static Expr *
PartitionFieldBoundConstraint(PartitionField * partitionField, IcebergPartitionTransform * partitionTransform,
ColumnToFieldIdMapping * entry)
{
IcebergPartitionTransformType type = partitionTransform->type;
IcebergPartitionTransformType type = partitionTransform->parsedTransform.type;

if (type != PARTITION_TRANSFORM_IDENTITY &&
partitionField->value == NULL)
Expand Down Expand Up @@ -751,7 +751,7 @@ IdentityPartitionFieldBoundConstraint(PartitionField * partitionField,
{
bool isNull = false;
Datum partitionDatum =
PartitionValueToDatum(partitionTransform->type, partitionField->value, partitionField->value_length,
PartitionValueToDatum(partitionTransform->parsedTransform.type, partitionField->value, partitionField->value_length,
partitionTransform->resultPgType, &isNull);

OpExpr *columnBoundEquality = copyObject(entry->equalityOperatorExpression);
Expand Down Expand Up @@ -780,7 +780,7 @@ TruncatePartitionFieldBoundConstraint(PartitionField * partitionField,
if (pgType.postgresTypeOid == INT4OID || pgType.postgresTypeOid == INT2OID)
{
int32 partitionValue = *(int32_t *) partitionField->value;
int truncateLen = partitionTransform->truncateLen;
int truncateLen = partitionTransform->parsedTransform.truncateLen;

int32 upperBound;

Expand All @@ -798,7 +798,7 @@ TruncatePartitionFieldBoundConstraint(PartitionField * partitionField,
else if (pgType.postgresTypeOid == INT8OID)
{
int64 partitionValue = *(int64_t *) partitionField->value;
int truncateLen = partitionTransform->truncateLen;
int truncateLen = partitionTransform->parsedTransform.truncateLen;

int64 upperBound;

Expand All @@ -825,7 +825,7 @@ TruncatePartitionFieldBoundConstraint(PartitionField * partitionField,
return NULL;
}

int truncateLen = partitionTransform->truncateLen;
int truncateLen = partitionTransform->parsedTransform.truncateLen;
char *truncatedUpperBound = TruncateUpperBoundForText(pstrdup(partitionValue), truncateLen);

if (truncatedUpperBound == NULL)
Expand All @@ -848,7 +848,7 @@ TruncatePartitionFieldBoundConstraint(PartitionField * partitionField,
memcpy(VARDATA_ANY(partitionValue), partitionField->value, partitionField->value_length);

bytea *partitionValueCopy = (bytea *) pg_detoast_datum_copy((struct varlena *) partitionValue);
int truncateLen = partitionTransform->truncateLen;
int truncateLen = partitionTransform->parsedTransform.truncateLen;

/* increment the last byte of the upper bound, which does not overflow */
partitionValueCopy = TruncateUpperBoundForBytea(partitionValueCopy, truncateLen);
Expand Down Expand Up @@ -1813,7 +1813,7 @@ ExtendClausesForBucketPartitioning(Partition * partition, List *partitionTransfo
if (partitionTransform == NULL)
continue;

if (partitionTransform->type != PARTITION_TRANSFORM_BUCKET)
if (partitionTransform->parsedTransform.type != PARTITION_TRANSFORM_BUCKET)
{
/* only extend restrict info for bucket transform */
continue;
Expand Down
50 changes: 24 additions & 26 deletions pg_lake_table/src/fdw/partition_transform.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down Expand Up @@ -251,13 +251,11 @@ 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 = DeepCopyIcebergPartitionSpecField(specField);

transform->attnum =
GetAttributeForFieldId(relationId, specField->source_id);
transform->columnName = get_attname(relationId, transform->attnum, false);
transform->parsedTransform.columnName = get_attname(relationId, transform->attnum, false);
transform->pgType = GetAttributePGType(relationId, transform->attnum);

if (IsInternalIcebergTable(relationId))
Expand All @@ -274,10 +272,10 @@ GetPartitionTransformFromSpecField(Oid relationId, IcebergPartitionSpecField * s
}

/* parse transform name */
ParseTransformName(transform->transformName,
&transform->type,
&transform->bucketCount,
&transform->truncateLen);
ParseTransformName(transform->specField->transform,
&transform->parsedTransform.type,
&transform->parsedTransform.bucketCount,
&transform->parsedTransform.truncateLen);

/* set transform's postgres type */
transform->resultPgType = GetTransformResultPGType(transform);
Expand Down Expand Up @@ -413,13 +411,13 @@ 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);

switch (transform->type)
switch (transform->parsedTransform.type)
{
case PARTITION_TRANSFORM_IDENTITY:
field->value = ApplyIdentityTransformToColumn(transform, columnValue, isNull,
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -498,7 +496,7 @@ ApplyTruncateTransformToColumn(IcebergPartitionTransform * transform, Datum colu

PGType sourceType = transform->pgType;
PGType resultType = transform->resultPgType;
int64_t truncateLen = (int64_t) transform->truncateLen;
int64_t truncateLen = (int64_t) transform->parsedTransform.truncateLen;
Datum truncatedColumnValue = 0;

if (sourceType.postgresTypeOid == INT2OID)
Expand Down Expand Up @@ -767,7 +765,7 @@ ApplyBucketTransformToColumn(IcebergPartitionTransform * transform, Datum column
{
int64_t value = (int64_t) DatumGetInt16(columnValue);

*bucketValue = (MurmurHash3_32_Long(value) & INT32_MAX) % transform->bucketCount;
*bucketValue = (MurmurHash3_32_Long(value) & INT32_MAX) % transform->parsedTransform.bucketCount;
}
else if (transform->pgType.postgresTypeOid == INT4OID)
{
Expand All @@ -777,27 +775,27 @@ ApplyBucketTransformToColumn(IcebergPartitionTransform * transform, Datum column
*/
int64_t value = (int64_t) DatumGetInt32(columnValue);

*bucketValue = (MurmurHash3_32_Long(value) & INT32_MAX) % transform->bucketCount;
*bucketValue = (MurmurHash3_32_Long(value) & INT32_MAX) % transform->parsedTransform.bucketCount;
}
else if (transform->pgType.postgresTypeOid == INT8OID)
{
int64_t value = DatumGetInt64(columnValue);

*bucketValue = (MurmurHash3_32_Long(value) & INT32_MAX) % transform->bucketCount;
*bucketValue = (MurmurHash3_32_Long(value) & INT32_MAX) % transform->parsedTransform.bucketCount;
}
else if (transform->pgType.postgresTypeOid == TEXTOID ||
transform->pgType.postgresTypeOid == VARCHAROID ||
transform->pgType.postgresTypeOid == BPCHAROID)
{
const char *value = TextDatumGetCString(columnValue);

*bucketValue = (MurmurHash3_32_Bytes(value, strlen(value)) & INT32_MAX) % transform->bucketCount;
*bucketValue = (MurmurHash3_32_Bytes(value, strlen(value)) & INT32_MAX) % transform->parsedTransform.bucketCount;
}
else if (transform->pgType.postgresTypeOid == BYTEAOID)
{
bytea *value = DatumGetByteaP(columnValue);

*bucketValue = (MurmurHash3_32_Bytes(VARDATA_ANY(value), VARSIZE_ANY_EXHDR(value)) & INT32_MAX) % transform->bucketCount;
*bucketValue = (MurmurHash3_32_Bytes(VARDATA_ANY(value), VARSIZE_ANY_EXHDR(value)) & INT32_MAX) % transform->parsedTransform.bucketCount;
}
else if (transform->pgType.postgresTypeOid == DATEOID)
{
Expand All @@ -809,47 +807,47 @@ ApplyBucketTransformToColumn(IcebergPartitionTransform * transform, Datum column
* spec normally hashes int bytes for date type but spark hashes long
* bytes of date. We follow spark here.
*/
*bucketValue = (MurmurHash3_32_Long(daysFromEpoch) & INT32_MAX) % transform->bucketCount;
*bucketValue = (MurmurHash3_32_Long(daysFromEpoch) & INT32_MAX) % transform->parsedTransform.bucketCount;
}
else if (transform->pgType.postgresTypeOid == TIMESTAMPOID)
{
Timestamp value = DatumGetTimestamp(columnValue);

int64_t microsecsFromEpoch = AdjustTimestampFromPostgresToUnix(value);

*bucketValue = (MurmurHash3_32_Long(microsecsFromEpoch) & INT32_MAX) % transform->bucketCount;
*bucketValue = (MurmurHash3_32_Long(microsecsFromEpoch) & INT32_MAX) % transform->parsedTransform.bucketCount;
}
else if (transform->pgType.postgresTypeOid == TIMESTAMPTZOID)
{
TimestampTz value = DatumGetTimestampTz(columnValue);

int64_t microsecsFromEpoch = AdjustTimestampFromPostgresToUnix(value);

*bucketValue = (MurmurHash3_32_Long(microsecsFromEpoch) & INT32_MAX) % transform->bucketCount;
*bucketValue = (MurmurHash3_32_Long(microsecsFromEpoch) & INT32_MAX) % transform->parsedTransform.bucketCount;
}
else if (transform->pgType.postgresTypeOid == TIMEOID)
{
TimeADT value = DatumGetTimeADT(columnValue);

int64_t microsecsFromMidnight = value;

*bucketValue = (MurmurHash3_32_Long(microsecsFromMidnight) & INT32_MAX) % transform->bucketCount;
*bucketValue = (MurmurHash3_32_Long(microsecsFromMidnight) & INT32_MAX) % transform->parsedTransform.bucketCount;
}
else if (transform->pgType.postgresTypeOid == UUIDOID)
{
size_t valueSize = 0;
unsigned char *value = PGIcebergBinarySerializePartitionFieldValue(columnValue, transform->sourceField->type,
transform->pgType, &valueSize);

*bucketValue = (MurmurHash3_32_Bytes(value, valueSize) & INT32_MAX) % transform->bucketCount;
*bucketValue = (MurmurHash3_32_Bytes(value, valueSize) & INT32_MAX) % transform->parsedTransform.bucketCount;
}
else if (transform->pgType.postgresTypeOid == NUMERICOID)
{
size_t valueSize = 0;
unsigned char *value = PGIcebergBinarySerializePartitionFieldValue(columnValue, transform->sourceField->type,
transform->pgType, &valueSize);

*bucketValue = (MurmurHash3_32_Bytes(value, valueSize) & INT32_MAX) % transform->bucketCount;
*bucketValue = (MurmurHash3_32_Bytes(value, valueSize) & INT32_MAX) % transform->parsedTransform.bucketCount;
}
else
{
Expand Down Expand Up @@ -977,7 +975,7 @@ SerializePartitionValueToPGText(void *value, size_t valueLength, IcebergPartitio
/* First, deserialize back */
bool isNull = false;
Datum partitionDatum =
PartitionValueToDatum(transform->type, value, valueLength,
PartitionValueToDatum(transform->parsedTransform.type, value, valueLength,
transform->resultPgType, &isNull);

if (isNull)
Expand Down
Loading